Fix nftables node updates after panel restart (#525)

## Summary
- Skip agent protocol SetProtocol commands when updating nftables nodes.
- Preserve existing nftables SSH credentials when edit forms omit secret
fields.
- Add regression coverage for nftables updates and SSH config
persistence across repository reopen.

## Test Plan
- rtk go test ./...

go-backend/internal/http/handler/control_plane.go

@@ -14,6 +14,7 @@ import (
 	"time"
 
 	"go-backend/internal/http/client"
+	runtimenft "go-backend/internal/runtime/nftables"
 	"go-backend/internal/store/model"
 	"go-backend/internal/ws"
 )
@@ -772,6 +773,9 @@ func (h *Handler) diagnoseForwardRuntime(ctx context.Context, forward *forwardRe
 	if ctx == nil {
 		ctx = context.Background()
 	}
+	if payload, handled, err := h.diagnoseNftablesForwardRuntime(forward); handled || err != nil {
+		return payload, err
+	}
 	forwardName, workItems, err := h.prepareForwardDiagnosis(forward)
 	if err != nil {
 		return nil, err
@@ -787,6 +791,111 @@ func (h *Handler) diagnoseForwardRuntime(ctx context.Context, forward *forwardRe
 	return payload, nil
 }
 
+func (h *Handler) diagnoseNftablesForwardRuntime(forward *forwardRecord) (map[string]interface{}, bool, error) {
+	if forward == nil {
+		return nil, false, errForwardNotFound
+	}
+	nftMode, entryNodeIDs, err := h.tunnelUsesNftables(forward.TunnelID)
+	if err != nil {
+		return nil, false, err
+	}
+	if !nftMode {
+		return nil, false, nil
+	}
+	if len(entryNodeIDs) == 0 {
+		return nil, true, errors.New("nftables 转发缺少入口节点")
+	}
+	targets, err := resolveDiagnosisTargets(forward.RemoteAddr)
+	if err != nil {
+		return nil, true, err
+	}
+	results, err := h.buildNftablesForwardDiagnosisResults(forward, entryNodeIDs[0], targets)
+	if err != nil {
+		return nil, true, err
+	}
+	payload := map[string]interface{}{
+		"forwardName": forward.Name,
+		"timestamp":   time.Now().UnixMilli(),
+		"results":     results,
+	}
+	return payload, true, nil
+}
+
+func (h *Handler) buildNftablesForwardDiagnosisResults(forward *forwardRecord, nodeID int64, targets []diagnosisTarget) ([]map[string]interface{}, error) {
+	if h == nil || h.repo == nil {
+		return nil, errors.New("handler not initialized")
+	}
+	node, err := h.getNodeRecord(nodeID)
+	if err != nil {
+		return nil, err
+	}
+	bindings, err := h.repo.ListNftRuleBindingsByNode(nodeID)
+	if err != nil {
+		return nil, err
+	}
+	var binding *model.NftRuleBinding
+	for i := range bindings {
+		if bindings[i].ForwardID == forward.ID {
+			binding = &bindings[i]
+			break
+		}
+	}
+	target := diagnosisTarget{}
+	if len(targets) > 0 {
+		target = targets[0]
+	}
+
+	status := "missing"
+	message := "nftables 规则未下发"
+	success := false
+	inPort := 0
+	protocols := ""
+	targetAddr := strings.TrimSpace(forward.RemoteAddr)
+	ruleHash := ""
+	if binding != nil {
+		status = strings.ToLower(strings.TrimSpace(binding.Status))
+		inPort = binding.InPort
+		protocols = strings.TrimSpace(binding.Protocols)
+		targetAddr = strings.TrimSpace(binding.TargetAddr)
+		ruleHash = strings.TrimSpace(binding.RuleHash)
+		if status == "" {
+			status = "pending"
+		}
+		if status == runtimenft.StatusApplied {
+			success = true
+			message = "nftables 规则已下发"
+		} else if strings.TrimSpace(binding.LastError) != "" {
+			message = binding.LastError
+		} else {
+			message = "nftables 规则未完成下发"
+		}
+	}
+
+	packetLoss := 100
+	if success {
+		packetLoss = 0
+	}
+	result := map[string]interface{}{
+		"success":       success,
+		"nodeName":      node.Name,
+		"nodeId":        strconv.FormatInt(nodeID, 10),
+		"targetIp":      target.IP,
+		"targetPort":    target.Port,
+		"description":   fmt.Sprintf("nftables规则(%s)->目标(%s)", node.Name, defaultString(target.Address, targetAddr)),
+		"averageTime":   0,
+		"packetLoss":    packetLoss,
+		"message":       message,
+		"fromChainType": 1,
+		"forwardMode":   "nftables",
+		"nftRuleStatus": status,
+		"nftRuleHash":   ruleHash,
+		"inPort":        inPort,
+		"protocols":     protocols,
+		"targetAddr":    targetAddr,
+	}
+	return []map[string]interface{}{result}, nil
+}
+
 func (h *Handler) prepareForwardDiagnosis(forward *forwardRecord) (string, []diagnosisWorkItem, error) {
 	if forward == nil {
 		return "", nil, errForwardNotFound

go-backend/internal/http/handler/jobs.go

@@ -2,11 +2,14 @@ package handler
 
 import (
 	"context"
+	"log"
 	"time"
 
 	"go-backend/internal/license"
 )
 
+var nftablesTrafficCollectInterval = 30 * time.Second
+
 func (h *Handler) StartBackgroundJobs() {
 	if h == nil || h.repo == nil {
 		return
@@ -131,7 +134,19 @@ func (h *Handler) runTunnelQualityProber(ctx context.Context) {
 
 func (h *Handler) runNftablesTrafficCollectLoop(ctx context.Context) {
 	defer h.jobsWG.Done()
-	ticker := time.NewTicker(time.Minute)
+	h.runNftablesStartupReconcile(ctx)
+	select {
+	case <-ctx.Done():
+		return
+	default:
+		h.runNftablesTrafficCollectJob(time.Now())
+	}
+
+	interval := nftablesTrafficCollectInterval
+	if interval <= 0 {
+		interval = 30 * time.Second
+	}
+	ticker := time.NewTicker(interval)
 	defer ticker.Stop()
 
 	for {
@@ -144,6 +159,27 @@ func (h *Handler) runNftablesTrafficCollectLoop(ctx context.Context) {
 	}
 }
 
+func (h *Handler) runNftablesStartupReconcile(ctx context.Context) {
+	if h == nil || h.repo == nil {
+		return
+	}
+	nodes, err := h.repo.ListNftablesNodesForCollection()
+	if err != nil {
+		log.Printf("nftables startup reconcile failed op=list_nodes err=%v", err)
+		return
+	}
+	for _, node := range nodes {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+		}
+		if err := h.syncNftablesNode(node.NodeID); err != nil {
+			log.Printf("nftables startup reconcile failed node_id=%d err=%v", node.NodeID, err)
+		}
+	}
+}
+
 func (h *Handler) runHourlyStatsLoop(ctx context.Context) {
 	defer h.jobsWG.Done()
 

go-backend/internal/http/handler/mutations.go

@@ -417,6 +417,12 @@ func (h *Handler) nodeUpdate(w http.ResponseWriter, r *http.Request) {
 	newHTTP := asInt(req["http"], currentHTTP)
 	newTLS := asInt(req["tls"], currentTLS)
 	newSocks := asInt(req["socks"], currentSocks)
+	forwardMode := defaultNodeForwardMode(strings.TrimSpace(asString(req["forwardMode"])))
+	currentForwardMode, err := h.repo.GetNodeForwardMode(id)
+	if err != nil {
+		response.WriteJSON(w, response.Err(-2, err.Error()))
+		return
+	}
 	serverIP := asString(req["serverIp"])
 	if serverIP != "" {
 		if err := IsValidNodeAddress(serverIP); err != nil {
@@ -424,7 +430,8 @@ func (h *Handler) nodeUpdate(w http.ResponseWriter, r *http.Request) {
 			return
 		}
 	}
-	if currentStatus == 1 && (newHTTP != currentHTTP || newTLS != currentTLS || newSocks != currentSocks) {
+	usesNftablesRuntime := forwardMode == "nftables" || defaultNodeForwardMode(currentForwardMode) == "nftables"
+	if currentStatus == 1 && !usesNftablesRuntime && (newHTTP != currentHTTP || newTLS != currentTLS || newSocks != currentSocks) {
 		if err := h.applyNodeProtocolChange(id, newHTTP, newTLS, newSocks); err != nil {
 			response.WriteJSON(w, response.ErrDefault(err.Error()))
 			return
@@ -432,7 +439,6 @@ func (h *Handler) nodeUpdate(w http.ResponseWriter, r *http.Request) {
 	}
 
 	now := time.Now().UnixMilli()
-	forwardMode := defaultNodeForwardMode(strings.TrimSpace(asString(req["forwardMode"])))
 	if err := h.repo.UpdateNode(id,
 		asString(req["name"]),
 		serverIP,

go-backend/internal/http/handler/nftables_runtime_test.go

@@ -10,6 +10,7 @@ import (
 	"net/http/httptest"
 	"path/filepath"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -20,6 +21,7 @@ import (
 )
 
 type fakeNftablesManager struct {
+	mu             sync.Mutex
 	testErr        error
 	reconcileErr   error
 	reconcileHit   int
@@ -33,11 +35,15 @@ type fakeNftablesManager struct {
 }
 
 func (f *fakeNftablesManager) Test(_ context.Context, cfg runtimenft.SSHConfig) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
 	f.lastConfig = cfg
 	return f.testErr
 }
 
 func (f *fakeNftablesManager) Reconcile(_ context.Context, cfg runtimenft.SSHConfig, plan runtimenft.NodePlan) (runtimenft.ApplyResult, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
 	f.reconcileHit++
 	f.lastConfig = cfg
 	f.lastPlan = plan
@@ -47,16 +53,20 @@ func (f *fakeNftablesManager) Reconcile(_ context.Context, cfg runtimenft.SSHCon
 	return runtimenft.ApplyResult{
 		NodeID: plan.NodeID,
 		Script: "table inet flvx {}",
-		Hashes: map[int64]string{plan.NodeID: "hash"},
+		Hashes: runtimenft.PlanHashes(plan),
 	}, nil
 }
 
 func (f *fakeNftablesManager) Clear(context.Context, runtimenft.SSHConfig) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
 	f.clearHit++
 	return f.clearErr
 }
 
 func (f *fakeNftablesManager) CollectCounters(_ context.Context, cfg runtimenft.SSHConfig) ([]runtimenft.CounterSample, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
 	f.collectHit++
 	f.lastConfig = cfg
 	if f.collectErr != nil {
@@ -65,6 +75,18 @@ func (f *fakeNftablesManager) CollectCounters(_ context.Context, cfg runtimenft.
 	return f.counterSamples, nil
 }
 
+func (f *fakeNftablesManager) reconcileCount() int {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return f.reconcileHit
+}
+
+func (f *fakeNftablesManager) collectCount() int {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return f.collectHit
+}
+
 type nftablesTestFixture struct {
 	handler *Handler
 	nodeID  int64
@@ -158,6 +180,49 @@ func TestNodeNftablesReconcileEndpointPersistsBindings(t *testing.T) {
 	}
 }
 
+func TestStartBackgroundJobsReconcilesNftablesRulesAtStartup(t *testing.T) {
+	fixture := setupNftablesHandler(t)
+	h := fixture.handler
+	seedNftablesSSHConfig(t, h, fixture.nodeID)
+	tunnelID := seedTunnelForNftables(t, h, "nft-startup-tunnel", fixture.nodeID)
+	seedForwardForNftables(t, h, tunnelID, fixture.nodeID, "203.0.113.9:8080")
+	manager := &fakeNftablesManager{}
+	h.nftablesManager = manager
+
+	h.StartBackgroundJobs()
+	t.Cleanup(h.StopBackgroundJobs)
+
+	waitForCondition(t, time.Second, func() bool {
+		return manager.reconcileCount() > 0
+	}, "nftables startup reconcile")
+}
+
+func TestStartBackgroundJobsCollectsNftablesTrafficImmediatelyAndUsesFastInterval(t *testing.T) {
+	fixture := setupNftablesCollectionFixture(t)
+	h := fixture.handler
+	manager := &fakeNftablesManager{counterSamples: []runtimenft.CounterSample{
+		{ForwardID: fixture.forwardID, Protocol: "tcp", Direction: runtimenft.CounterDirectionToTarget, Bytes: 1000, Packets: 10},
+	}}
+	h.nftablesManager = manager
+
+	oldInterval := nftablesTrafficCollectInterval
+	nftablesTrafficCollectInterval = 20 * time.Millisecond
+	t.Cleanup(func() { nftablesTrafficCollectInterval = oldInterval })
+
+	h.StartBackgroundJobs()
+	t.Cleanup(h.StopBackgroundJobs)
+
+	waitForCondition(t, time.Second, func() bool {
+		return manager.collectCount() >= 2
+	}, "immediate and repeated nftables traffic collection")
+}
+
+func TestNftablesTrafficCollectIntervalDefaultsToThirtySeconds(t *testing.T) {
+	if nftablesTrafficCollectInterval != 30*time.Second {
+		t.Fatalf("expected default nftables traffic collection interval 30s, got %s", nftablesTrafficCollectInterval)
+	}
+}
+
 func TestNodeNftablesClearEndpointClearsBindings(t *testing.T) {
 	fixture := setupNftablesHandler(t)
 	seedNftablesSSHConfig(t, fixture.handler, fixture.nodeID)
@@ -280,6 +345,41 @@ func TestNodeUpdatePreservesExistingNftablesSecretsWhenFieldsOmitted(t *testing.
 	}
 }
 
+func TestNodeUpdateSkipsAgentProtocolCommandForNftablesNode(t *testing.T) {
+	fixture := setupNftablesHandler(t)
+	seedNftablesSSHConfig(t, fixture.handler, fixture.nodeID)
+
+	req := newAuthenticatedJSONRequest(t, map[string]interface{}{
+		"id":          fixture.nodeID,
+		"name":        "nft-node-updated",
+		"serverIp":    "198.51.100.10",
+		"serverIpV4":  "198.51.100.10",
+		"port":        "1000-65535",
+		"forwardMode": "nftables",
+		"http":        1,
+		"tls":         1,
+		"socks":       1,
+		"sshConfig": map[string]interface{}{
+			"host":     "203.0.113.30",
+			"port":     22,
+			"username": "admin",
+			"authType": "password",
+			"sudoMode": "none",
+		},
+	})
+	res := httptest.NewRecorder()
+	fixture.handler.nodeUpdate(res, req)
+	assertNftablesSuccessWithBody(t, res)
+
+	cfg, err := fixture.handler.repo.GetNodeSSHConfig(fixture.nodeID)
+	if err != nil {
+		t.Fatalf("load ssh config: %v", err)
+	}
+	if !cfg.Password.Valid || cfg.Password.String != "secret" {
+		t.Fatalf("expected password secret to be preserved, got %+v", cfg)
+	}
+}
+
 func TestValidateNftablesForwardRequestRejectsHostnameTarget(t *testing.T) {
 	fixture := setupNftablesHandler(t)
 	h := fixture.handler
@@ -421,6 +521,61 @@ func TestTunnelBatchRedeployUsesNftablesReconcile(t *testing.T) {
 	}
 }
 
+func TestDiagnoseForwardRuntimeReturnsNftablesRuleStatus(t *testing.T) {
+	fixture := setupNftablesHandler(t)
+	h := fixture.handler
+	tunnelID := seedTunnelForNftables(t, h, "nft-diagnose-tunnel", fixture.nodeID)
+	forward := seedForwardForNftables(t, h, tunnelID, fixture.nodeID, "203.0.113.9:8080")
+	now := time.Now().UnixMilli()
+	if err := h.repo.UpsertNftRuleBinding(repo.NftRuleBindingInput{
+		ForwardID:  forward.ID,
+		NodeID:     fixture.nodeID,
+		InPort:     20000,
+		Protocols:  "tcp,udp",
+		TargetAddr: "203.0.113.9:8080",
+		RuleHash:   "hash-a",
+		Status:     runtimenft.StatusApplied,
+	}, now); err != nil {
+		t.Fatalf("seed nft binding: %v", err)
+	}
+
+	payload, err := h.diagnoseForwardRuntime(context.Background(), &forwardRecord{
+		ID:         forward.ID,
+		Name:       forward.Name,
+		TunnelID:   tunnelID,
+		RemoteAddr: "203.0.113.9:8080",
+	})
+	if err != nil {
+		t.Fatalf("diagnose forward: %v", err)
+	}
+	results, ok := payload["results"].([]map[string]interface{})
+	if !ok || len(results) != 1 {
+		t.Fatalf("expected one nftables diagnosis result, got %#v", payload["results"])
+	}
+	result := results[0]
+	if result["forwardMode"] != "nftables" || result["nftRuleStatus"] != runtimenft.StatusApplied {
+		t.Fatalf("expected nftables applied result, got %#v", result)
+	}
+	if result["success"] != true {
+		t.Fatalf("expected nftables diagnosis success, got %#v", result)
+	}
+	if !strings.Contains(asString(result["message"]), "已下发") {
+		t.Fatalf("expected applied message, got %#v", result["message"])
+	}
+}
+
+func waitForCondition(t *testing.T, timeout time.Duration, condition func() bool, description string) {
+	t.Helper()
+	deadline := time.Now().Add(timeout)
+	for time.Now().Before(deadline) {
+		if condition() {
+			return
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	t.Fatalf("timed out waiting for %s", description)
+}
+
 func setupNftablesHandler(t *testing.T) nftablesTestFixture {
 	t.Helper()
 

go-backend/internal/store/repo/repository_nftables_test.go

@@ -1,6 +1,7 @@
 package repo
 
 import (
+	"path/filepath"
 	"strings"
 	"testing"
 	"time"
@@ -123,6 +124,114 @@ func TestNftablesNodeModeSSHConfigAndBindingPersistence(t *testing.T) {
 	}
 }
 
+func TestNftablesNodeSSHConfigSurvivesRepositoryReopen(t *testing.T) {
+	tests := []struct {
+		name string
+		cfg  NftSSHConfigInput
+	}{
+		{
+			name: "password",
+			cfg: NftSSHConfigInput{
+				Host:       "203.0.113.10",
+				Port:       2222,
+				Username:   "root",
+				AuthType:   "password",
+				Password:   "ssh-password",
+				Passphrase: "key-passphrase",
+				SudoMode:   "sudo",
+			},
+		},
+		{
+			name: "private-key",
+			cfg: NftSSHConfigInput{
+				Host:       "203.0.113.11",
+				Port:       2223,
+				Username:   "admin",
+				AuthType:   "private_key",
+				PrivateKey: "PRIVATE-KEY-SHOULD-PERSIST",
+				Passphrase: "key-passphrase",
+				SudoMode:   "sudo_su",
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			dbPath := filepath.Join(t.TempDir(), "nftables-ssh.sqlite")
+			r, err := Open(dbPath)
+			if err != nil {
+				t.Fatalf("open repo: %v", err)
+			}
+
+			now := time.Now().UnixMilli()
+			if err := r.CreateNode(
+				"nft-node",
+				"secret",
+				tt.cfg.Host,
+				nil,
+				nil,
+				"10000-20000",
+				nil,
+				nil,
+				nil,
+				nil,
+				nil,
+				0,
+				0,
+				0,
+				now,
+				1,
+				"[::]",
+				"[::]",
+				1,
+				0,
+				nil,
+				nil,
+				nil,
+				nil,
+				"nftables",
+			); err != nil {
+				t.Fatalf("CreateNode: %v", err)
+			}
+
+			nodes, err := r.ListNodes()
+			if err != nil {
+				t.Fatalf("ListNodes: %v", err)
+			}
+			nodeID := nodes[0]["id"].(int64)
+			if err := r.UpsertNodeSSHConfig(nodeID, tt.cfg, now); err != nil {
+				t.Fatalf("UpsertNodeSSHConfig: %v", err)
+			}
+			if err := r.Close(); err != nil {
+				t.Fatalf("close repo: %v", err)
+			}
+
+			reopened, err := Open(dbPath)
+			if err != nil {
+				t.Fatalf("reopen repo: %v", err)
+			}
+			defer reopened.Close()
+
+			loaded, err := reopened.GetNodeSSHConfig(nodeID)
+			if err != nil {
+				t.Fatalf("GetNodeSSHConfig after reopen: %v", err)
+			}
+			if loaded.Host != tt.cfg.Host || loaded.Port != tt.cfg.Port || loaded.Username != tt.cfg.Username || loaded.AuthType != tt.cfg.AuthType || loaded.SudoMode != tt.cfg.SudoMode {
+				t.Fatalf("unexpected ssh config after reopen: %+v", loaded)
+			}
+			if tt.cfg.Password != "" && (!loaded.Password.Valid || loaded.Password.String != tt.cfg.Password) {
+				t.Fatalf("expected password to persist after reopen, got %+v", loaded.Password)
+			}
+			if tt.cfg.PrivateKey != "" && (!loaded.PrivateKey.Valid || loaded.PrivateKey.String != tt.cfg.PrivateKey) {
+				t.Fatalf("expected private key to persist after reopen, got %+v", loaded.PrivateKey)
+			}
+			if tt.cfg.Passphrase != "" && (!loaded.Passphrase.Valid || loaded.Passphrase.String != tt.cfg.Passphrase) {
+				t.Fatalf("expected passphrase to persist after reopen, got %+v", loaded.Passphrase)
+			}
+		})
+	}
+}
+
 func TestUpdateNodeWithoutForwardModePreservesExistingMode(t *testing.T) {
 	r, err := Open(":memory:")
 	if err != nil {