fix: close consistency and rollback gaps

Author: gitlayzer

.github/workflows/release.yml

@@ -52,6 +52,18 @@ jobs:
       - name: Run vet
         run: go vet ./...
 
+      - name: Run race tests
+        run: go test -race ./cmd ./pkg/session ./pkg/accesspolicy ./pkg/publicauth ./pkg/k8s ./pkg/clusterconnect ./pkg/tunnel ./pkg/mesh ./pkg/daemon
+
+      - name: Install gosec
+        run: go install github.com/securego/gosec/v2/cmd/gosec@v2.22.8
+
+      - name: Run gosec
+        run: $(go env GOPATH)/bin/gosec ./cmd ./pkg/... ./assets ./
+
+      - name: Verify module files are tidy
+        run: go mod tidy -diff
+
       - name: Check npm package builder syntax
         run: node --check scripts/build-npm-packages.mjs
 

.goreleaser.yaml

@@ -3,7 +3,7 @@ version: 2
 
 before:
   hooks:
-    - go mod tidy
+    - go mod tidy -diff
 
 builds:
   - env:

cmd/apply.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"os"
@@ -143,28 +144,35 @@ func runApplyConfig(ctx context.Context, config *applyFile, dryRun bool) ([]appl
 		return nil, fmt.Errorf("failed to init k8s client: %w", err)
 	}
 
-	results := make([]applyResult, 0, len(config.Tunnels))
+	tunnelIDs := make([]string, 0, len(config.Tunnels))
 	for _, item := range config.Tunnels {
-		result, err := applyOneTunnel(ctx, item, authData, client, kubeconfig, dryRun)
+		tunnelID, err := applyTunnelID(item.Name)
 		if err != nil {
-			rollbackApplyResults(client, results)
-			return results, err
+			return nil, err
 		}
-		results = append(results, result)
+		tunnelIDs = append(tunnelIDs, tunnelID)
 	}
-	if !dryRun {
+
+	results := make([]applyResult, 0, len(config.Tunnels))
+	err = withTunnelOperationLocks(tunnelIDs, func() error {
+		for _, item := range config.Tunnels {
+			result, applyErr := applyOneTunnel(ctx, item, authData, client, kubeconfig, false)
+			if applyErr != nil {
+				return applyErrorWithRollback(applyErr, rollbackApplyResults(client, results))
+			}
+			results = append(results, result)
+		}
 		if err := ensureDaemonRunningFn(); err != nil {
-			rollbackApplyResults(client, results)
-			return results, fmt.Errorf("failed to start local daemon: %w", err)
+			return applyErrorWithRollback(fmt.Errorf("failed to start local daemon: %w", err), rollbackApplyResults(client, results))
 		}
 		for _, result := range results {
 			if err := waitForDaemonSession(result.TunnelID, daemonConnectTimeout); err != nil {
-				rollbackApplyResults(client, results)
-				return results, err
+				return applyErrorWithRollback(err, rollbackApplyResults(client, results))
 			}
 		}
-	}
-	return results, nil
+		return nil
+	})
+	return results, err
 }
 
 func loadApplyFile(path string) (*applyFile, error) {
@@ -253,14 +261,22 @@ func applyOneTunnel(ctx context.Context, item applyTunnel, authData *auth.AuthDa
 		existing, err := session.Get(normalized.TunnelID)
 		if err == nil {
 			alreadyExisted = true
-			existingSession = existing
 			currentNamespace := ""
 			if client != nil {
 				currentNamespace = client.Namespace()
 			}
 			if err := validateExistingApplySessionScope(*existing, authData, currentNamespace); err != nil {
 				return result, err
 			}
+			if client != nil {
+				if err := refreshSessionFromRemoteLocked(ctx, existing); err != nil {
+					return result, fmt.Errorf("tunnel %s: sync existing remote state: %w", normalized.TunnelID, err)
+				}
+			}
+			if strings.TrimSpace(existing.Secret) == "" {
+				return result, fmt.Errorf("tunnel %s already exists but its local secret is unavailable; stop or cleanup the old session before apply", existing.TunnelID)
+			}
+			existingSession = existing
 			if existing.Secret != "" {
 				secret = existing.Secret
 			}
@@ -447,9 +463,6 @@ func validateExistingApplySessionScope(existing session.TunnelSession, authData
 			return fmt.Errorf("tunnel %s already belongs to namespace %s; current namespace is %s", existing.TunnelID, existing.Namespace, currentNamespace)
 		}
 	}
-	if strings.TrimSpace(existing.Secret) == "" {
-		return fmt.Errorf("tunnel %s already exists but its local secret is unavailable; stop or cleanup the old session before apply", existing.TunnelID)
-	}
 	return nil
 }
 
@@ -488,7 +501,8 @@ func rollbackExistingApplyTunnel(client *k8s.Client, previous session.TunnelSess
 	return firstErr
 }
 
-func rollbackApplyResults(client *k8s.Client, results []applyResult) {
+func rollbackApplyResults(client *k8s.Client, results []applyResult) error {
+	var rollbackErrors []error
 	for i := len(results) - 1; i >= 0; i-- {
 		result := results[i]
 		if result.TunnelID == "" {
@@ -497,16 +511,30 @@ func rollbackApplyResults(client *k8s.Client, results []applyResult) {
 		if result.NewTunnel {
 			if client != nil {
 				cleanupCtx, cancel := context.WithTimeout(context.Background(), tunnelCleanupTimeout)
-				_ = client.CleanupTunnel(cleanupCtx, result.TunnelID)
+				if err := client.CleanupTunnel(cleanupCtx, result.TunnelID); err != nil {
+					rollbackErrors = append(rollbackErrors, fmt.Errorf("cleanup tunnel %s: %w", result.TunnelID, err))
+				}
 				cancel()
 			}
-			_ = session.Delete(result.TunnelID)
+			if err := session.Delete(result.TunnelID); err != nil {
+				rollbackErrors = append(rollbackErrors, fmt.Errorf("delete local session %s: %w", result.TunnelID, err))
+			}
 			continue
 		}
 		if result.Previous != nil {
-			_ = rollbackExistingApplyTunnel(client, *result.Previous)
+			if err := rollbackExistingApplyTunnel(client, *result.Previous); err != nil {
+				rollbackErrors = append(rollbackErrors, fmt.Errorf("restore tunnel %s: %w", result.TunnelID, err))
+			}
 		}
 	}
+	return errors.Join(rollbackErrors...)
+}
+
+func applyErrorWithRollback(applyErr, rollbackErr error) error {
+	if rollbackErr == nil {
+		return applyErr
+	}
+	return errors.Join(applyErr, fmt.Errorf("rollback failed: %w", rollbackErr))
 }
 
 func normalizeApplyTunnel(item applyTunnel) (normalizedApplyTunnel, error) {
@@ -763,6 +791,9 @@ func applyTunnelID(name string) (string, error) {
 	if name != strings.ToLower(name) || !applyNamePattern.MatchString(name) {
 		return "", fmt.Errorf("invalid tunnel name %q: use lowercase DNS-compatible names, e.g. web or api-dev", name)
 	}
+	if strings.HasPrefix(name, "mesh-") {
+		return "", fmt.Errorf("invalid tunnel name %q: the mesh- prefix is reserved for Sealtun Mesh resources", name)
+	}
 	return name, nil
 }
 

cmd/apply_test.go

@@ -2,6 +2,7 @@ package cmd
 
 import (
 	"context"
+	"errors"
 	"os"
 	"path/filepath"
 	"strings"
@@ -135,6 +136,15 @@ func TestNormalizeApplyTunnelRejectsUnsafeNames(t *testing.T) {
 	}
 }
 
+func TestApplyTunnelIDRejectsMeshReservedPrefix(t *testing.T) {
+	t.Parallel()
+
+	_, err := applyTunnelID("mesh-global")
+	if err == nil || !strings.Contains(err.Error(), "reserved") {
+		t.Fatalf("expected reserved mesh prefix error, got %v", err)
+	}
+}
+
 func TestNormalizeApplyTunnelDefaultsProtocol(t *testing.T) {
 	t.Parallel()
 
@@ -903,7 +913,7 @@ func TestRollbackApplyResultsRestoresExistingLocalSession(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	rollbackApplyResults(nil, []applyResult{{
+	if err := rollbackApplyResults(nil, []applyResult{{
 		TunnelID: "web",
 		Previous: &session.TunnelSession{
 			TunnelID:     previous.TunnelID,
@@ -917,7 +927,9 @@ func TestRollbackApplyResultsRestoresExistingLocalSession(t *testing.T) {
 			Secret:       previous.Secret,
 			Mode:         previous.Mode,
 		},
-	}})
+	}}); err != nil {
+		t.Fatalf("rollbackApplyResults returned error: %v", err)
+	}
 
 	current, err := session.Get("web")
 	if err != nil {
@@ -940,3 +952,52 @@ func TestRollbackApplyResultsRestoresExistingLocalSession(t *testing.T) {
 		}
 	}
 }
+
+func TestApplyOneTunnelPropagatesRemoteRefreshFailure(t *testing.T) {
+	t.Setenv("SEALTUN_HOME", t.TempDir())
+	if err := session.Save(session.TunnelSession{
+		TunnelID:  "web",
+		Region:    "https://gzg.sealos.run",
+		Namespace: "default",
+		LocalPort: "3000",
+		Protocol:  "https",
+		Secret:    "secret",
+	}); err != nil {
+		t.Fatal(err)
+	}
+
+	want := errors.New("remote state unavailable")
+	previousCollect := collectSessionRemoteState
+	collectSessionRemoteState = func(context.Context, session.TunnelSession) (*k8s.TunnelRemoteState, error) {
+		return nil, want
+	}
+	t.Cleanup(func() { collectSessionRemoteState = previousCollect })
+
+	_, err := applyOneTunnel(context.Background(), applyTunnel{Name: "web", LocalPort: 3000}, &auth.AuthData{Region: "https://gzg.sealos.run"}, &k8s.Client{}, "", false)
+	if !errors.Is(err, want) {
+		t.Fatalf("apply error = %v, want %v", err, want)
+	}
+}
+
+func TestRollbackApplyResultsReturnsLocalRestoreFailure(t *testing.T) {
+	home := t.TempDir()
+	t.Setenv("SEALTUN_HOME", home)
+	if err := os.WriteFile(filepath.Join(home, ".sealtun"), []byte("not a directory"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+
+	err := rollbackApplyResults(nil, []applyResult{{
+		TunnelID: "web",
+		Previous: &session.TunnelSession{
+			TunnelID:  "web",
+			Region:    "https://gzg.sealos.run",
+			Namespace: "default",
+			LocalPort: "3000",
+			Protocol:  "https",
+			Secret:    "secret",
+		},
+	}})
+	if err == nil || !strings.Contains(err.Error(), "restore tunnel web") {
+		t.Fatalf("expected rollback failure to be reported, got %v", err)
+	}
+}

cmd/daemon.go

@@ -62,18 +62,22 @@ var daemonCmd = &cobra.Command{
 					continue
 				}
 				if sessionExpired(sess, time.Now()) {
-					fmt.Printf("[+] tunnel %s expired; cleaning up...\n", sess.TunnelID)
-					cleanupCtx, cancel := context.WithTimeout(ctx, tunnelCleanupTimeout)
-					err := cleanupSessionResources(cleanupCtx, sess)
-					cancel()
+					latest, removed, err := cleanupExpiredDaemonSession(ctx, sess.TunnelID)
 					if err != nil {
 						fmt.Printf("[!] expired tunnel %s cleanup failed: %v\n", sess.TunnelID, err)
 						continue
 					}
-					if err := session.Delete(sess.TunnelID); err != nil && !os.IsNotExist(err) {
-						fmt.Printf("[!] expired tunnel %s local cleanup failed: %v\n", sess.TunnelID, err)
+					if removed {
+						fmt.Printf("[+] expired tunnel %s cleaned up\n", sess.TunnelID)
+						continue
+					}
+					if latest == nil {
+						continue
+					}
+					sess = *latest
+					if sess.Mode != "daemon" {
+						continue
 					}
-					continue
 				}
 				if sess.ConnectionState == session.ConnectionStateStopped {
 					continue
@@ -108,6 +112,34 @@ var daemonCmd = &cobra.Command{
 	},
 }
 
+func cleanupExpiredDaemonSession(ctx context.Context, tunnelID string) (latest *session.TunnelSession, removed bool, err error) {
+	err = withTunnelOperationLockContext(ctx, tunnelID, func() error {
+		current, getErr := session.Get(tunnelID)
+		if os.IsNotExist(getErr) {
+			return nil
+		}
+		if getErr != nil {
+			return getErr
+		}
+		if current.Mode != "daemon" || !sessionExpired(*current, time.Now()) {
+			latest = current
+			return nil
+		}
+
+		cleanupCtx, cancel := context.WithTimeout(ctx, tunnelCleanupTimeout)
+		defer cancel()
+		if cleanupErr := cleanupSessionResources(cleanupCtx, *current); cleanupErr != nil {
+			return cleanupErr
+		}
+		if deleteErr := session.Delete(tunnelID); deleteErr != nil && !os.IsNotExist(deleteErr) {
+			return fmt.Errorf("delete local session: %w", deleteErr)
+		}
+		removed = true
+		return nil
+	})
+	return latest, removed, err
+}
+
 func reconcileDaemonWorkers(
 	ctx context.Context,
 	mu *sync.Mutex,