From 3c0e833cfcac473edc2a1be05ddcca73bd4261ee Mon Sep 17 00:00:00 2001 From: sagit <36596628+Sagit-chu@users.noreply.github.com> Date: Thu, 30 Apr 2026 14:36:26 +0800 Subject: [PATCH] fix: reduce runtime sync disruptions (#484) --- .../internal/http/handler/control_plane.go | 57 +++++- .../http/handler/control_plane_test.go | 46 +++++ go-backend/internal/http/handler/mutations.go | 176 ++++++++++++++---- go-gost/x/config/config.go | 2 +- go-gost/x/config/persist.go | 17 +- go-gost/x/socket/chain.go | 12 +- go-gost/x/socket/limiter.go | 30 ++- go-gost/x/socket/limiter_test.go | 52 ++++++ go-gost/x/socket/service.go | 58 ++++-- go-gost/x/socket/service_test.go | 51 +++++ go-gost/x/socket/websocket_reporter.go | 30 --- 11 files changed, 409 insertions(+), 122 deletions(-) create mode 100644 go-gost/x/socket/limiter_test.go create mode 100644 go-gost/x/socket/service_test.go diff --git a/go-backend/internal/http/handler/control_plane.go b/go-backend/internal/http/handler/control_plane.go index 75af395..ef5ab60 100644 --- a/go-backend/internal/http/handler/control_plane.go +++ b/go-backend/internal/http/handler/control_plane.go @@ -484,13 +484,39 @@ func (h *Handler) forwardServiceBaseCandidates(forward *forwardRecord) ([]string } func (h *Handler) deleteForwardServiceBasesOnNode(nodeID int64, bases []string) error { - return deleteForwardServiceCandidates(bases, func(name string) error { - payload := map[string]interface{}{ - "services": []string{name}, + names := buildForwardServiceDeleteNames(bases) + if len(names) == 0 { + return nil + } + payload := map[string]interface{}{"services": names} + _, err := h.sendNodeCommand(nodeID, "DeleteService", payload, false, true) + return err +} + +func buildForwardServiceDeleteNames(bases []string) []string { + names := make([]string, 0, len(bases)*3) + seen := make(map[string]struct{}, len(bases)*3) + appendName := func(name string) { + name = strings.TrimSpace(name) + if name == "" { + return } - _, err := h.sendNodeCommand(nodeID, "DeleteService", payload, false, false) - return err - }) + if _, ok := seen[name]; ok { + return + } + seen[name] = struct{}{} + names = append(names, name) + } + for _, base := range bases { + base = strings.TrimSpace(base) + if base == "" { + continue + } + appendName(base + "_tcp") + appendName(base + "_udp") + appendName(base) + } + return names } func (h *Handler) controlForwardServices(forward *forwardRecord, commandType string, tolerateNotFound bool) error { @@ -520,6 +546,25 @@ func (h *Handler) controlForwardServices(forward *forwardRecord, commandType str candidateTunnelIDs = append(candidateTunnelIDs, userTunnelIDs...) candidateTunnelIDs = append(candidateTunnelIDs, allUserTunnelIDs...) bases := buildForwardServiceBaseCandidates(forward.ID, forward.UserID, userTunnelID, candidateTunnelIDs) + if strings.EqualFold(strings.TrimSpace(commandType), "DeleteService") { + seen := map[int64]struct{}{} + for _, fp := range ports { + if _, ok := seen[fp.NodeID]; ok { + continue + } + seen[fp.NodeID] = struct{}{} + if err := h.deleteForwardServiceBasesOnNode(fp.NodeID, bases); err != nil { + if isNodeOfflineOrTimeoutError(err) { + continue + } + if tolerateNotFound && isNotFoundError(err) { + continue + } + return err + } + } + return nil + } seen := map[int64]struct{}{} healed := false for _, fp := range ports { diff --git a/go-backend/internal/http/handler/control_plane_test.go b/go-backend/internal/http/handler/control_plane_test.go index 86c6433..9c5132c 100644 --- a/go-backend/internal/http/handler/control_plane_test.go +++ b/go-backend/internal/http/handler/control_plane_test.go @@ -201,6 +201,52 @@ func TestDeleteForwardServiceCandidatesDeletesAllMatchingVariants(t *testing.T) } } +func TestBuildForwardServiceDeleteNamesBatchesAndDeduplicatesVariants(t *testing.T) { + bases := []string{"57_7_7", "57_7_0", "57_7_7"} + got := buildForwardServiceDeleteNames(bases) + want := []string{"57_7_7_tcp", "57_7_7_udp", "57_7_7", "57_7_0_tcp", "57_7_0_udp", "57_7_0"} + if !reflect.DeepEqual(got, want) { + t.Fatalf("expected %v, got %v", want, got) + } +} + +func TestRemovedTunnelRuntimeNodeIDsSeparatesChainAndServiceRoles(t *testing.T) { + oldRows := []chainNodeRecord{ + {NodeID: 1, ChainType: 1}, + {NodeID: 2, ChainType: 2}, + {NodeID: 3, ChainType: 3}, + {NodeID: 5, ChainType: 2}, + {NodeID: 6, ChainType: 3}, + } + newRows := []chainNodeRecord{ + {NodeID: 2, ChainType: 3}, + {NodeID: 3, ChainType: 3}, + {NodeID: 5, ChainType: 1}, + } + + removedChains := removedTunnelRuntimeNodeIDs(oldRows, newRows, tunnelRuntimeNeedsChain) + if want := []int64{1, 2}; !reflect.DeepEqual(removedChains, want) { + t.Fatalf("expected removed chains %v, got %v", want, removedChains) + } + + removedServices := removedTunnelRuntimeNodeIDs(oldRows, newRows, tunnelRuntimeNeedsService) + if want := []int64{5, 6}; !reflect.DeepEqual(removedServices, want) { + t.Fatalf("expected removed services %v, got %v", want, removedServices) + } +} + +func TestTunnelForwardRuntimeNeedsSyncOnlyWhenTypeOrEntriesChange(t *testing.T) { + if tunnelForwardRuntimeNeedsSync(2, 2, []int64{1, 2}, []int64{2, 1}) { + t.Fatalf("same tunnel type and same entry set should not resync forwards") + } + if !tunnelForwardRuntimeNeedsSync(1, 2, []int64{1}, []int64{1}) { + t.Fatalf("type change should resync forwards") + } + if !tunnelForwardRuntimeNeedsSync(2, 2, []int64{1}, []int64{1, 2}) { + t.Fatalf("entry set change should resync forwards") + } +} + func TestValidateForwardPortAvailabilityRejectsOtherForwardOccupancy(t *testing.T) { h := &Handler{repo: nil} node := &nodeRecord{ID: 9, Name: "test-node"} diff --git a/go-backend/internal/http/handler/mutations.go b/go-backend/internal/http/handler/mutations.go index af3a159..a52c777 100644 --- a/go-backend/internal/http/handler/mutations.go +++ b/go-backend/internal/http/handler/mutations.go @@ -748,21 +748,8 @@ func (h *Handler) cleanupTunnelRuntime(tunnelID int64) { return } - protocol := strings.TrimSpace(tunnel.Protocol) - if protocol == "" { - protocol = "tls" - } chainName := fmt.Sprintf("chains_%d", tunnelID) - serviceNames := []string{ - fmt.Sprintf("tunnel_%d", tunnelID), - fmt.Sprintf("%d_tls", tunnelID), - fmt.Sprintf("%d_kcp", tunnelID), - fmt.Sprintf("%d_wss", tunnelID), - fmt.Sprintf("%d_mtls", tunnelID), - fmt.Sprintf("%d_mwss", tunnelID), - fmt.Sprintf("%d_tcp", tunnelID), - fmt.Sprintf("%d_mtcp", tunnelID), - } + serviceNames := tunnelRuntimeServiceNames(tunnelID) for _, row := range chainRows { if row.ChainType == 1 { @@ -776,6 +763,70 @@ func (h *Handler) cleanupTunnelRuntime(tunnelID int64) { } } +func tunnelRuntimeServiceNames(tunnelID int64) []string { + return []string{ + fmt.Sprintf("tunnel_%d", tunnelID), + fmt.Sprintf("%d_tls", tunnelID), + fmt.Sprintf("%d_kcp", tunnelID), + fmt.Sprintf("%d_wss", tunnelID), + fmt.Sprintf("%d_mtls", tunnelID), + fmt.Sprintf("%d_mwss", tunnelID), + fmt.Sprintf("%d_tcp", tunnelID), + fmt.Sprintf("%d_mtcp", tunnelID), + } +} + +func tunnelRuntimeNeedsChain(row chainNodeRecord) bool { + return row.ChainType == 1 || row.ChainType == 2 +} + +func tunnelRuntimeNeedsService(row chainNodeRecord) bool { + return row.ChainType == 2 || row.ChainType == 3 +} + +func removedTunnelRuntimeNodeIDs(oldRows, newRows []chainNodeRecord, needsRuntime func(chainNodeRecord) bool) []int64 { + if len(oldRows) == 0 || needsRuntime == nil { + return nil + } + newRuntimeNodes := make(map[int64]struct{}, len(newRows)) + for _, row := range newRows { + if row.NodeID <= 0 || !needsRuntime(row) { + continue + } + newRuntimeNodes[row.NodeID] = struct{}{} + } + seen := make(map[int64]struct{}, len(oldRows)) + removed := make([]int64, 0) + for _, row := range oldRows { + if row.NodeID <= 0 || !needsRuntime(row) { + continue + } + if _, ok := seen[row.NodeID]; ok { + continue + } + seen[row.NodeID] = struct{}{} + if _, stillNeeded := newRuntimeNodes[row.NodeID]; stillNeeded { + continue + } + removed = append(removed, row.NodeID) + } + return removed +} + +func (h *Handler) cleanupObsoleteTunnelRuntime(tunnelID int64, oldRows, newRows []chainNodeRecord) { + if h == nil || tunnelID <= 0 || len(oldRows) == 0 { + return + } + chainName := fmt.Sprintf("chains_%d", tunnelID) + for _, nodeID := range removedTunnelRuntimeNodeIDs(oldRows, newRows, tunnelRuntimeNeedsChain) { + _, _ = h.sendNodeCommand(nodeID, "DeleteChains", map[string]interface{}{"chain": chainName}, false, true) + } + serviceNames := tunnelRuntimeServiceNames(tunnelID) + for _, nodeID := range removedTunnelRuntimeNodeIDs(oldRows, newRows, tunnelRuntimeNeedsService) { + _, _ = h.sendNodeCommand(nodeID, "DeleteService", map[string]interface{}{"services": serviceNames}, false, true) + } +} + func (h *Handler) tunnelGet(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { response.WriteJSON(w, response.ErrDefault("请求失败")) @@ -819,12 +870,15 @@ func (h *Handler) tunnelUpdate(w http.ResponseWriter, r *http.Request) { return } oldEntryNodeIDs, _ := h.tunnelEntryNodeIDs(id) - - h.cleanupTunnelRuntime(id) + typeVal := asInt(req["type"], 1) + oldTunnel, _ := h.getTunnelRecord(id) + oldChainRows, _ := h.listChainNodesForTunnel(id) + if oldTunnel != nil && oldTunnel.Type == 2 && typeVal != 2 { + h.cleanupTunnelRuntime(id) + } h.cleanupFederationRuntime(id) now := time.Now().UnixMilli() - typeVal := asInt(req["type"], 1) ipPreference := asString(req["ipPreference"]) localDomain := h.federationLocalDomain() @@ -917,13 +971,19 @@ func (h *Handler) tunnelUpdate(w http.ResponseWriter, r *http.Request) { } if typeVal == 2 { - createdChains, createdServices, applyErr := h.applyTunnelRuntime(runtimeState) + applyRuntime := h.applyTunnelRuntime + if oldTunnel != nil && oldTunnel.Type == 2 { + applyRuntime = h.applyTunnelRuntimeUpsert + } + createdChains, createdServices, applyErr := applyRuntime(runtimeState) if applyErr != nil { updateProtocol := "tls" if len(runtimeState.InNodes) > 0 && strings.TrimSpace(runtimeState.InNodes[0].Protocol) != "" { updateProtocol = strings.TrimSpace(runtimeState.InNodes[0].Protocol) } - h.rollbackTunnelRuntime(createdChains, createdServices, id, updateProtocol) + if oldTunnel == nil || oldTunnel.Type != 2 { + h.rollbackTunnelRuntime(createdChains, createdServices, id, updateProtocol) + } h.releaseFederationRuntimeRefs(federationReleaseRefs) _ = h.repo.DeleteFederationTunnelBindingsByTunnel(id) if len(federationReleaseRefs) == 0 && shouldDeferTunnelRuntimeApplyError(applyErr) { @@ -933,9 +993,20 @@ func (h *Handler) tunnelUpdate(w http.ResponseWriter, r *http.Request) { response.WriteJSON(w, response.ErrDefault(applyErr.Error())) return } + newChainRows, _ := h.listChainNodesForTunnel(id) + h.cleanupObsoleteTunnelRuntime(id, oldChainRows, newChainRows) } - if forwards, fwdErr := h.listForwardsByTunnel(id); fwdErr == nil { + oldType := 0 + if oldTunnel != nil { + oldType = oldTunnel.Type + } + if tunnelForwardRuntimeNeedsSync(oldType, typeVal, oldEntryNodeIDs, newEntryNodeIDs) { + forwards, fwdErr := h.listForwardsByTunnel(id) + if fwdErr != nil { + response.WriteJSON(w, response.OKEmpty()) + return + } for i := range forwards { _ = h.syncForwardServices(&forwards[i], "UpdateService", true) } @@ -944,6 +1015,13 @@ func (h *Handler) tunnelUpdate(w http.ResponseWriter, r *http.Request) { response.WriteJSON(w, response.OKEmpty()) } +func tunnelForwardRuntimeNeedsSync(oldType, newType int, oldEntryNodeIDs, newEntryNodeIDs []int64) bool { + if oldType != newType { + return true + } + return !sameInt64Set(oldEntryNodeIDs, newEntryNodeIDs) +} + func sameInt64Set(a, b []int64) bool { if len(a) != len(b) { return false @@ -3358,6 +3436,14 @@ func (h *Handler) cleanupFederationRuntime(tunnelID int64) { } func (h *Handler) applyTunnelRuntime(state *tunnelCreateState) ([]int64, []int64, error) { + return h.applyTunnelRuntimeWithMode(state, false) +} + +func (h *Handler) applyTunnelRuntimeUpsert(state *tunnelCreateState) ([]int64, []int64, error) { + return h.applyTunnelRuntimeWithMode(state, true) +} + +func (h *Handler) applyTunnelRuntimeWithMode(state *tunnelCreateState, upsert bool) ([]int64, []int64, error) { if h == nil || state == nil { return nil, nil, errors.New("invalid tunnel runtime state") } @@ -3376,7 +3462,7 @@ func (h *Handler) applyTunnelRuntime(state *tunnelCreateState) ([]int64, []int64 if err != nil { return createdChains, createdServices, err } - if _, err := h.sendNodeCommand(inNode.NodeID, "AddChains", chainData, true, false); err != nil { + if err := h.applyTunnelChainOnNode(inNode.NodeID, chainData, upsert); err != nil { if shouldDeferTunnelRuntimeApplyError(err) { continue } @@ -3399,7 +3485,7 @@ func (h *Handler) applyTunnelRuntime(state *tunnelCreateState) ([]int64, []int64 if err != nil { return createdChains, createdServices, err } - if _, err := h.sendNodeCommand(chainNode.NodeID, "AddChains", chainData, true, false); err != nil { + if err := h.applyTunnelChainOnNode(chainNode.NodeID, chainData, upsert); err != nil { if shouldDeferTunnelRuntimeApplyError(err) { continue } @@ -3408,7 +3494,7 @@ func (h *Handler) applyTunnelRuntime(state *tunnelCreateState) ([]int64, []int64 createdChains = append(createdChains, chainNode.NodeID) serviceData := buildTunnelChainServiceConfig(state.TunnelID, chainNode, state.Nodes[chainNode.NodeID], len(nextTargets)) - if err := h.addTunnelServiceOnNode(chainNode.NodeID, state.TunnelID, serviceData); err != nil { + if err := h.addTunnelServiceOnNodeWithMode(chainNode.NodeID, state.TunnelID, serviceData, upsert); err != nil { if shouldDeferTunnelRuntimeApplyError(err) { continue } @@ -3424,7 +3510,7 @@ func (h *Handler) applyTunnelRuntime(state *tunnelCreateState) ([]int64, []int64 continue } serviceData := buildTunnelChainServiceConfig(state.TunnelID, outNode, state.Nodes[outNode.NodeID], 1) - if err := h.addTunnelServiceOnNode(outNode.NodeID, state.TunnelID, serviceData); err != nil { + if err := h.addTunnelServiceOnNodeWithMode(outNode.NodeID, state.TunnelID, serviceData, upsert); err != nil { if shouldDeferTunnelRuntimeApplyError(err) { continue } @@ -3436,6 +3522,27 @@ func (h *Handler) applyTunnelRuntime(state *tunnelCreateState) ([]int64, []int64 return createdChains, createdServices, nil } +func (h *Handler) applyTunnelChainOnNode(nodeID int64, chainData map[string]interface{}, upsert bool) error { + if upsert { + return h.upsertTunnelChainOnNode(nodeID, chainData) + } + _, err := h.sendNodeCommand(nodeID, "AddChains", chainData, true, false) + return err +} + +func (h *Handler) upsertTunnelChainOnNode(nodeID int64, chainData map[string]interface{}) error { + if h == nil { + return errors.New("invalid tunnel chain context") + } + chainName := asString(chainData["name"]) + if strings.TrimSpace(chainName) == "" { + return errors.New("转发链名称不能为空") + } + payload := map[string]interface{}{"chain": chainName, "data": chainData} + _, err := h.sendNodeCommand(nodeID, "UpdateChains", payload, true, false) + return err +} + func retryTunnelServiceAddWithCleanup(add func() error, cleanup func() error, wait time.Duration) error { if add == nil { return errors.New("invalid tunnel service add callback") @@ -3457,6 +3564,10 @@ func retryTunnelServiceAddWithCleanup(add func() error, cleanup func() error, wa } func (h *Handler) addTunnelServiceOnNode(nodeID, tunnelID int64, serviceData []map[string]interface{}) error { + return h.addTunnelServiceOnNodeWithMode(nodeID, tunnelID, serviceData, false) +} + +func (h *Handler) addTunnelServiceOnNodeWithMode(nodeID, tunnelID int64, serviceData []map[string]interface{}, upsert bool) error { if h == nil { return errors.New("invalid tunnel service context") } @@ -3466,9 +3577,13 @@ func (h *Handler) addTunnelServiceOnNode(nodeID, tunnelID int64, serviceData []m serviceName = strings.TrimSpace(name) } } + command := "AddService" + if upsert { + command = "UpdateService" + } return retryTunnelServiceAddWithCleanup( func() error { - _, err := h.sendNodeCommand(nodeID, "AddService", serviceData, true, false) + _, err := h.sendNodeCommand(nodeID, command, serviceData, true, false) return err }, func() error { @@ -3487,16 +3602,7 @@ func (h *Handler) rollbackTunnelRuntime(chainNodeIDs, serviceNodeIDs []int64, tu protocol = "tls" } seenServices := make(map[int64]struct{}) - serviceNames := []string{ - fmt.Sprintf("tunnel_%d", tunnelID), - fmt.Sprintf("%d_tls", tunnelID), - fmt.Sprintf("%d_kcp", tunnelID), - fmt.Sprintf("%d_wss", tunnelID), - fmt.Sprintf("%d_mtls", tunnelID), - fmt.Sprintf("%d_mwss", tunnelID), - fmt.Sprintf("%d_tcp", tunnelID), - fmt.Sprintf("%d_mtcp", tunnelID), - } + serviceNames := tunnelRuntimeServiceNames(tunnelID) for i := len(serviceNodeIDs) - 1; i >= 0; i-- { nodeID := serviceNodeIDs[i] if _, ok := seenServices[nodeID]; ok { diff --git a/go-gost/x/config/config.go b/go-gost/x/config/config.go index c7cb4f8..0d88dbc 100644 --- a/go-gost/x/config/config.go +++ b/go-gost/x/config/config.go @@ -48,7 +48,7 @@ func OnUpdate(f func(c *Config) error) error { globalMux.Unlock() if err == nil { - persist() + err = persist() } return err diff --git a/go-gost/x/config/persist.go b/go-gost/x/config/persist.go index 17c36f3..99b8746 100644 --- a/go-gost/x/config/persist.go +++ b/go-gost/x/config/persist.go @@ -40,19 +40,19 @@ func EnablePersist() { } // persist writes the current global config to the configured file atomically. -func persist() { +func persist() error { persistMu.Lock() path := persistPath enabled := persistEnable persistMu.Unlock() if !enabled || path == "" { - return + return nil } cfg := Global() if cfg == nil { - return + return nil } var buf bytes.Buffer @@ -60,7 +60,7 @@ func persist() { enc.SetIndent("", " ") if err := enc.Encode(cfg); err != nil { fmt.Printf("⚠️ config persist: marshal failed: %v\n", err) - return + return fmt.Errorf("config persist: marshal failed: %w", err) } // Atomic write: write to temp file then rename @@ -68,7 +68,7 @@ func persist() { tmp, err := os.CreateTemp(dir, ".gost-*.tmp") if err != nil { fmt.Printf("⚠️ config persist: create temp file failed: %v\n", err) - return + return fmt.Errorf("config persist: create temp file failed: %w", err) } tmpName := tmp.Name() @@ -76,19 +76,20 @@ func persist() { tmp.Close() os.Remove(tmpName) fmt.Printf("⚠️ config persist: write failed: %v\n", err) - return + return fmt.Errorf("config persist: write failed: %w", err) } if err := tmp.Close(); err != nil { os.Remove(tmpName) fmt.Printf("⚠️ config persist: close temp file failed: %v\n", err) - return + return fmt.Errorf("config persist: close temp file failed: %w", err) } if err := os.Rename(tmpName, path); err != nil { os.Remove(tmpName) fmt.Printf("⚠️ config persist: rename failed: %v\n", err) - return + return fmt.Errorf("config persist: rename failed: %w", err) } fmt.Printf("💾 节点配置已持久化到 %s\n", path) + return nil } diff --git a/go-gost/x/socket/chain.go b/go-gost/x/socket/chain.go index 74b6325..4a49a1d 100644 --- a/go-gost/x/socket/chain.go +++ b/go-gost/x/socket/chain.go @@ -31,12 +31,10 @@ func createChain(req createChainRequest) error { return errors.New("chain " + name + " already exists") } - config.OnUpdate(func(c *config.Config) error { + return config.OnUpdate(func(c *config.Config) error { c.Chains = append(c.Chains, &req.Data) return nil }) - - return nil } func updateChain(req updateChainRequest) error { @@ -58,7 +56,7 @@ func updateChain(req updateChainRequest) error { return errors.New("chain " + name + " already exists") } - config.OnUpdate(func(c *config.Config) error { + return config.OnUpdate(func(c *config.Config) error { found := false for i := range c.Chains { if c.Chains[i].Name == name { @@ -72,8 +70,6 @@ func updateChain(req updateChainRequest) error { } return nil }) - - return nil } func deleteChain(req deleteChainRequest) error { @@ -84,7 +80,7 @@ func deleteChain(req deleteChainRequest) error { registry.ChainRegistry().Unregister(name) } - config.OnUpdate(func(c *config.Config) error { + return config.OnUpdate(func(c *config.Config) error { chains := c.Chains c.Chains = nil for _, s := range chains { @@ -95,8 +91,6 @@ func deleteChain(req deleteChainRequest) error { } return nil }) - - return nil } type createChainRequest struct { diff --git a/go-gost/x/socket/limiter.go b/go-gost/x/socket/limiter.go index e623f4e..dd613a9 100644 --- a/go-gost/x/socket/limiter.go +++ b/go-gost/x/socket/limiter.go @@ -25,12 +25,10 @@ func createLimiter(req createLimiterRequest) error { return errors.New("limiter " + name + " already exists") } - config.OnUpdate(func(c *config.Config) error { + return config.OnUpdate(func(c *config.Config) error { c.Limiters = append(c.Limiters, &req.Data) return nil }) - - return nil } func updateLimiter(req updateLimiterRequest) error { @@ -49,7 +47,7 @@ func updateLimiter(req updateLimiterRequest) error { return errors.New("limiter " + name + " already exists") } - config.OnUpdate(func(c *config.Config) error { + return config.OnUpdate(func(c *config.Config) error { found := false for i := range c.Limiters { if c.Limiters[i].Name == name { @@ -63,8 +61,6 @@ func updateLimiter(req updateLimiterRequest) error { } return nil }) - - return nil } func deleteLimiter(req deleteLimiterRequest) error { @@ -75,7 +71,7 @@ func deleteLimiter(req deleteLimiterRequest) error { registry.TrafficLimiterRegistry().Unregister(name) } - config.OnUpdate(func(c *config.Config) error { + return config.OnUpdate(func(c *config.Config) error { limiteres := c.Limiters c.Limiters = nil for _, s := range limiteres { @@ -86,8 +82,6 @@ func deleteLimiter(req deleteLimiterRequest) error { } return nil }) - - return nil } type createLimiterRequest struct { @@ -120,10 +114,10 @@ func createConnLimiter(req createLimiterRequest) error { return errors.New("conn limiter " + name + " already exists") } - if c := config.Global(); c != nil { + return config.OnUpdate(func(c *config.Config) error { c.CLimiters = append(c.CLimiters, &req.Data) - } - return nil + return nil + }) } func updateConnLimiter(req updateLimiterRequest) error { @@ -139,7 +133,7 @@ func updateConnLimiter(req updateLimiterRequest) error { return errors.New("conn limiter " + name + " already exists") } - if c := config.Global(); c != nil { + return config.OnUpdate(func(c *config.Config) error { for i := range c.CLimiters { if c.CLimiters[i].Name == name { c.CLimiters[i] = &req.Data @@ -147,8 +141,8 @@ func updateConnLimiter(req updateLimiterRequest) error { } } c.CLimiters = append(c.CLimiters, &req.Data) - } - return nil + return nil + }) } func deleteConnLimiter(req deleteLimiterRequest) error { @@ -158,7 +152,7 @@ func deleteConnLimiter(req deleteLimiterRequest) error { registry.ConnLimiterRegistry().Unregister(name) } - if c := config.Global(); c != nil { + return config.OnUpdate(func(c *config.Config) error { limiteres := c.CLimiters c.CLimiters = nil for _, s := range limiteres { @@ -167,6 +161,6 @@ func deleteConnLimiter(req deleteLimiterRequest) error { } c.CLimiters = append(c.CLimiters, s) } - } - return nil + return nil + }) } diff --git a/go-gost/x/socket/limiter_test.go b/go-gost/x/socket/limiter_test.go new file mode 100644 index 0000000..c8ede26 --- /dev/null +++ b/go-gost/x/socket/limiter_test.go @@ -0,0 +1,52 @@ +package socket + +import ( + "path/filepath" + "testing" + + corelogger "github.com/go-gost/core/logger" + "github.com/go-gost/x/config" + xlogger "github.com/go-gost/x/logger" + "github.com/go-gost/x/registry" +) + +func TestCreateConnLimiterUpdatesGlobalConfig(t *testing.T) { + corelogger.SetDefault(xlogger.Nop()) + + name := "conn_limiter_tdd" + originalConfig := config.Global() + defer config.Set(originalConfig) + registry.ConnLimiterRegistry().Unregister(name) + defer registry.ConnLimiterRegistry().Unregister(name) + config.Set(&config.Config{}) + + err := createConnLimiter(createLimiterRequest{Data: config.LimiterConfig{Name: name, Limits: []string{"$ 1"}}}) + if err != nil { + t.Fatalf("create conn limiter: %v", err) + } + + cfg := config.Global() + if len(cfg.CLimiters) != 1 || cfg.CLimiters[0] == nil || cfg.CLimiters[0].Name != name { + t.Fatalf("expected conn limiter in global config, got %#v", cfg.CLimiters) + } +} + +func TestCreateLimiterReportsPersistFailure(t *testing.T) { + corelogger.SetDefault(xlogger.Nop()) + + name := "traffic_limiter_persist_tdd" + originalConfig := config.Global() + originalPersistPath := config.PersistPath() + defer config.Set(originalConfig) + defer config.SetPersistPath(originalPersistPath) + registry.TrafficLimiterRegistry().Unregister(name) + defer registry.TrafficLimiterRegistry().Unregister(name) + config.Set(&config.Config{}) + config.SetPersistPath(filepath.Join(t.TempDir(), "missing", "gost.json")) + config.EnablePersist() + + err := createLimiter(createLimiterRequest{Data: config.LimiterConfig{Name: name, Limits: []string{"$ 1"}}}) + if err == nil { + t.Fatalf("expected persist failure to be returned") + } +} diff --git a/go-gost/x/socket/service.go b/go-gost/x/socket/service.go index 731ea19..217a2c1 100644 --- a/go-gost/x/socket/service.go +++ b/go-gost/x/socket/service.go @@ -3,6 +3,7 @@ package socket import ( "errors" "fmt" + "reflect" "strings" "time" @@ -54,9 +55,8 @@ func createServices(req createServicesRequest) error { if err := registry.ServiceRegistry().Register(ps.config.Name, ps.service); err != nil { // 如果注册失败，回滚已注册的服务 for _, regName := range registeredServices { - if svc := registry.ServiceRegistry().Get(regName); svc != nil { + if registry.ServiceRegistry().Get(regName) != nil { registry.ServiceRegistry().Unregister(regName) - svc.Close() } } return errors.New("service " + ps.config.Name + " already exists") @@ -72,14 +72,12 @@ func createServices(req createServicesRequest) error { } // 第四阶段：更新配置 - config.OnUpdate(func(c *config.Config) error { + return config.OnUpdate(func(c *config.Config) error { for _, ps := range parsedServices { c.Services = append(c.Services, &ps.config) } return nil }) - - return nil } func updateServices(req updateServicesRequest) error { @@ -98,17 +96,23 @@ func updateServices(req updateServicesRequest) error { } // 第二阶段：逐个更新服务（Upsert模式：存在则更新，不存在则创建） + changedServices := make([]struct { + config config.ServiceConfig + service service.Service + }, 0, len(req.Data)) for i := range req.Data { serviceConfig := &req.Data[i] name := serviceConfig.Name + if registry.ServiceRegistry().Get(name) != nil && serviceConfigUnchanged(name, *serviceConfig) { + continue + } // 1. 获取旧服务 old := registry.ServiceRegistry().Get(name) // 2. 关闭旧服务 (如果存在) if old != nil { - old.Close() - // 3. 从注册表移除旧服务 + // 3. 从注册表移除旧服务；registry 会负责关闭旧服务。 registry.ServiceRegistry().Unregister(name) } @@ -117,6 +121,10 @@ func updateServices(req updateServicesRequest) error { if err != nil { return errors.New("create service " + name + " failed: " + err.Error()) } + changedServices = append(changedServices, struct { + config config.ServiceConfig + service service.Service + }{*serviceConfig, svc}) // 5. 注册新服务 if err := registry.ServiceRegistry().Register(name, svc); err != nil { @@ -127,12 +135,15 @@ func updateServices(req updateServicesRequest) error { // 6. 启动新服务 go svc.Serve() } + if len(changedServices) == 0 { + return nil + } // 第三阶段：更新配置 - config.OnUpdate(func(c *config.Config) error { - for i := range req.Data { + if err := config.OnUpdate(func(c *config.Config) error { + for i := range changedServices { // 创建副本以确保指针安全 - cfgCopy := req.Data[i] + cfgCopy := changedServices[i].config found := false for j := range c.Services { if c.Services[j].Name == cfgCopy.Name { @@ -146,11 +157,30 @@ func updateServices(req updateServicesRequest) error { } } return nil - }) + }); err != nil { + return err + } return nil } +func serviceConfigUnchanged(name string, next config.ServiceConfig) bool { + cfg := config.Global() + if cfg == nil { + return false + } + next.Status = nil + for _, current := range cfg.Services { + if current == nil || strings.TrimSpace(current.Name) != name { + continue + } + currentCopy := *current + currentCopy.Status = nil + return reflect.DeepEqual(currentCopy, next) + } + return false +} + func deleteServices(req deleteServicesRequest) error { if len(req.Services) == 0 { @@ -183,7 +213,6 @@ func deleteServices(req deleteServicesRequest) error { // 第二阶段：删除所有服务 for _, std := range servicesToDelete { registry.ServiceRegistry().Unregister(std.name) - std.service.Close() } // 确保所有请求删除的服务都从注册表中移除（即使之前未找到实例） for _, name := range namesToRemove { @@ -193,7 +222,7 @@ func deleteServices(req deleteServicesRequest) error { } // 第三阶段：更新配置 - config.OnUpdate(func(c *config.Config) error { + err := config.OnUpdate(func(c *config.Config) error { services := c.Services c.Services = nil for _, s := range services { @@ -211,8 +240,7 @@ func deleteServices(req deleteServicesRequest) error { return nil }) xservice.GetGlobalTrafficManager().RemoveServices(namesToRemove...) - - return nil + return err } func pauseServices(req pauseServicesRequest) error { diff --git a/go-gost/x/socket/service_test.go b/go-gost/x/socket/service_test.go new file mode 100644 index 0000000..ee68322 --- /dev/null +++ b/go-gost/x/socket/service_test.go @@ -0,0 +1,51 @@ +package socket + +import ( + "net" + "testing" + + corelogger "github.com/go-gost/core/logger" + "github.com/go-gost/core/service" + "github.com/go-gost/x/config" + xlogger "github.com/go-gost/x/logger" + "github.com/go-gost/x/registry" +) + +type recordingService struct { + closed int +} + +func (s *recordingService) Serve() error { return nil } +func (s *recordingService) Addr() net.Addr { return nil } +func (s *recordingService) Close() error { + s.closed++ + return nil +} + +func TestUpdateServicesSkipsUnchangedServiceWithoutRestart(t *testing.T) { + corelogger.SetDefault(xlogger.Nop()) + + name := "unchanged_service_tdd" + existing := &recordingService{} + + registry.ServiceRegistry().Unregister(name) + defer registry.ServiceRegistry().Unregister(name) + if err := registry.ServiceRegistry().Register(name, service.Service(existing)); err != nil { + t.Fatalf("register existing service: %v", err) + } + + originalConfig := config.Global() + defer config.Set(originalConfig) + serviceConfig := config.ServiceConfig{Name: name, Addr: "127.0.0.1:0"} + config.Set(&config.Config{Services: []*config.ServiceConfig{&serviceConfig}}) + + if err := updateServices(updateServicesRequest{Data: []config.ServiceConfig{serviceConfig}}); err != nil { + t.Fatalf("unchanged update should succeed without parsing/restarting: %v", err) + } + if existing.closed != 0 { + t.Fatalf("unchanged service was restarted, closed %d times", existing.closed) + } + if got := registry.ServiceRegistry().Get(name); got != service.Service(existing) { + t.Fatalf("expected existing service to remain registered") + } +} diff --git a/go-gost/x/socket/websocket_reporter.go b/go-gost/x/socket/websocket_reporter.go index c213eca..23c5e54 100644 --- a/go-gost/x/socket/websocket_reporter.go +++ b/go-gost/x/socket/websocket_reporter.go @@ -782,7 +782,6 @@ func (w *WebSocketReporter) routeCommand(cmd CommandMessage) { fmt.Println("🔔 收到命令: ", string(jsonBytes)) var err error var response CommandResponse - var needSaveConfig bool // 标记是否需要保存配置（只有状态变更命令才需要） // 传递 requestId response.RequestId = cmd.RequestId @@ -792,63 +791,49 @@ func (w *WebSocketReporter) routeCommand(cmd CommandMessage) { case "AddService": err = w.handleAddService(cmd.Data) response.Type = "AddServiceResponse" - needSaveConfig = true case "UpdateService": err = w.handleUpdateService(cmd.Data) response.Type = "UpdateServiceResponse" - needSaveConfig = true case "DeleteService": err = w.handleDeleteService(cmd.Data) response.Type = "DeleteServiceResponse" - needSaveConfig = true case "PauseService": err = w.handlePauseService(cmd.Data) response.Type = "PauseServiceResponse" - needSaveConfig = true case "ResumeService": err = w.handleResumeService(cmd.Data) response.Type = "ResumeServiceResponse" - needSaveConfig = true // Chain 相关命令 case "AddChains": err = w.handleAddChain(cmd.Data) response.Type = "AddChainsResponse" - needSaveConfig = true case "UpdateChains": err = w.handleUpdateChain(cmd.Data) response.Type = "UpdateChainsResponse" - needSaveConfig = true case "DeleteChains": err = w.handleDeleteChain(cmd.Data) response.Type = "DeleteChainsResponse" - needSaveConfig = true // Limiter 相关命令 case "AddLimiters": err = w.handleAddLimiter(cmd.Data) response.Type = "AddLimitersResponse" - needSaveConfig = true case "UpdateLimiters": err = w.handleUpdateLimiter(cmd.Data) response.Type = "UpdateLimitersResponse" - needSaveConfig = true case "DeleteLimiters": err = w.handleDeleteLimiter(cmd.Data) response.Type = "DeleteLimitersResponse" - needSaveConfig = true case "AddCLimiters": err = w.handleAddCLimiter(cmd.Data) response.Type = "AddCLimitersResponse" - needSaveConfig = true case "UpdateCLimiters": err = w.handleUpdateCLimiter(cmd.Data) response.Type = "UpdateCLimitersResponse" - needSaveConfig = true case "DeleteCLimiters": err = w.handleDeleteCLimiter(cmd.Data) response.Type = "DeleteCLimitersResponse" - needSaveConfig = true // TCP Ping 诊断命令（只读，不需要保存配置） case "TcpPing": @@ -876,7 +861,6 @@ func (w *WebSocketReporter) routeCommand(cmd CommandMessage) { case "SetProtocol": err = w.handleSetProtocol(cmd.Data) response.Type = "SetProtocolResponse" - needSaveConfig = true // 升级 Agent 命令（异步执行，不需要保存配置） case "UpgradeAgent": @@ -895,20 +879,6 @@ func (w *WebSocketReporter) routeCommand(cmd CommandMessage) { response.Type = "UnknownCommandResponse" } - // 只有状态变更命令才保存配置 - if needSaveConfig { - if saveErr := saveConfig(); saveErr != nil { - fmt.Printf("❌ 保存配置失败: %v\n", saveErr) - if err == nil { - err = fmt.Errorf("保存配置失败: %v", saveErr) - } else { - err = fmt.Errorf("%v; 保存配置失败: %v", err, saveErr) - } - } else { - fmt.Println("✅ 配置已保存到 gost.json") - } - } - // 发送响应 if err != nil { response.Success = false