From 2552c14faf97c1c6986bd57774ba7fb65722c64d Mon Sep 17 00:00:00 2001 From: jianf <2198972886@qq.com> Date: Wed, 19 Aug 2026 21:52:28 +0800 Subject: [PATCH] fix: start/stop for ring-only forwards + group sync via ring topology MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three issues fixed: 1. Start/stop returned 404 for forwards owned by other nodes: findLinkByTriple only checked local SQLite store. Added findLinkFromTopology fallback — startForward/stopForward now look up ring topology entries when the local store doesn't have the link. 2. Group labels didn't sync through the ring: handleForwardsAssign only updated local SQLite, never the ring topology. Added State.UpdateTopologyGroup + Engine.UpdateTopologyGroup — handleForwardsAssign now updates both. Added topologySync callback called after every e.state = tk.State (OnToken) and e.state = s (AdoptState) to re-apply local store group overrides onto the freshly adopted topology, so they survive state adoption and propagate via the next token forward. handleForwardsGroupDelete also clears ring topology entries. 3. Status-merge branch omitted Group field: ring-only forwards always showed '未分组'. Added Group: t.Link.Group to the topology-merge forwardStatus. Verified: 4-node cluster, 5 forwards, group assigned on node-a propagates to all nodes within one token cycle; stop from non-owning node succeeds (HTTP 200) and worker stops on the owning node. --- internal/cluster/ring.go | 14 ++++ internal/cluster/ring_engine.go | 28 +++++++ internal/httpapi/handlers_forwards.go | 106 ++++++++++++++++---------- internal/httpapi/server.go | 1 + 4 files changed, 110 insertions(+), 39 deletions(-) diff --git a/internal/cluster/ring.go b/internal/cluster/ring.go index 99647f5..9cabbb5 100644 --- a/internal/cluster/ring.go +++ b/internal/cluster/ring.go @@ -404,6 +404,20 @@ func (s *State) ForwardsOwnedBy(owner string) []*TopoEntry { return out } +// UpdateTopologyGroup sets the group label on the topology entry matching the +// (local, remote, port) triple. Returns true if found. The updated entry +// propagates to all nodes via the next token cycle — group changes sync +// through the ring without a dedicated command. +func (s *State) UpdateTopologyGroup(local, remote string, port int, group string) bool { + for _, e := range s.Topology { + if e.Local.Name == local && e.Remote.Name == remote && e.Link.RemotePort == port { + e.Link.Group = group + return true + } + } + return false +} + // OfflineReassign moves all active forwards owned by an offline node back into // PendingTasks (they become new tasks for the next lowest-load member). Returns // the reassigned task ids. diff --git a/internal/cluster/ring_engine.go b/internal/cluster/ring_engine.go index 8cdb3d0..d57df3b 100644 --- a/internal/cluster/ring_engine.go +++ b/internal/cluster/ring_engine.go @@ -44,6 +44,12 @@ type Engine struct { // every token cycle (OnToken) and on AdoptState. Cleared (pass "") on // detachAsStandalone — an explicit leave must NOT auto-rejoin. peerPersist func(peersJSON string) error + // topologySync is called right after e.state = tk.State (OnToken) or + // e.state = s (AdoptState) so the host can re-apply local store overrides + // (e.g. group labels) onto the freshly adopted topology. Without this, + // group changes made via HTTP handlers between token cycles are overwritten + // by the next state adoption and never propagate to other nodes. + topologySync func() state State // myAddr maps our Node ID to the address peers dial. @@ -219,6 +225,12 @@ func (e *Engine) OnToken(ctx context.Context, tk *Token) (*Token, error) { if e.state.PendingTasks == nil { e.state.PendingTasks = map[string]*Task{} } + // Re-apply local store overrides (group labels, disabled flags) onto + // the freshly adopted topology so they survive state adoption and + // propagate to all nodes via the next token forward. + if e.topologySync != nil { + e.topologySync() + } for id, t := range localPending { e.state.PendingTasks[id] = t } @@ -695,6 +707,9 @@ func (e *Engine) AdoptState(s State) { for id, t := range kept { e.state.PendingTasks[id] = t } + if e.topologySync != nil { + e.topologySync() + } e.state.UpsertNode(Node{ID: e.ID, Addr: e.myAddr, Alive: true, Load: e.loadSnapshot(), Version: e.Version, Cache: e.Cache, NodeKey: e.nodeKey}) if e.Log != nil { @@ -780,6 +795,13 @@ func (e *Engine) HasTask(local, remote string, port int) bool { return false } +// UpdateTopologyGroup sets the group label on a topology entry. The change +// propagates to all nodes via the next token cycle. Used by the forwards +// assign handler so group labels sync through the ring. +func (e *Engine) UpdateTopologyGroup(local, remote string, port int, group string) bool { + return e.state.UpdateTopologyGroup(local, remote, port, group) +} + // IsLeader reports whether this node is the current ring leader. func (e *Engine) IsLeader() bool { return e.state.LeaderID == e.ID } @@ -796,6 +818,12 @@ func (e *Engine) SetKeyPersist(fn func(key string) error) { e.keyPersist = fn } // to durable storage (store.SetClusterPeers). Called once from main.go. func (e *Engine) SetPeerPersist(fn func(peersJSON string) error) { e.peerPersist = fn } +// SetTopologySync installs the callback used to re-apply local store overrides +// (group labels, disabled flags) onto the ring topology after each state +// adoption. Called once from main.go. Without this, group changes made via +// HTTP handlers are overwritten by the next e.state = tk.State. +func (e *Engine) SetTopologySync(fn func()) { e.topologySync = fn } + // persistPeers extracts all alive peers (addr + nodeKey, excluding self) // from the current ring state and persists them via the peerPersist callback. // Called on every token cycle (OnToken) and on AdoptState so a crashed node diff --git a/internal/httpapi/handlers_forwards.go b/internal/httpapi/handlers_forwards.go index 834f12b..7807381 100644 --- a/internal/httpapi/handlers_forwards.go +++ b/internal/httpapi/handlers_forwards.go @@ -36,34 +36,44 @@ func findLinkByTriple(s *store.Store, local, remote string, port int) (store.Lin return store.Link{}, false } +// findLinkFromTopology looks up a forward's local/remote/link data from the +// ring topology when it's not in the local SQLite store (forward owned by +// another node). Returns the store.Local, store.Remote, store.Link, and ok. +func (h *Handler) findLinkFromTopology(local, remote string, port int) (store.Local, store.Remote, store.Link, bool) { + if h.Ring == nil { + return store.Local{}, store.Remote{}, store.Link{}, false + } + snap := h.Ring.Snapshot() + for _, t := range snap.Topology { + if t.Local.Name == local && t.Remote.Name == remote && t.Link.RemotePort == port { + return t.Local, t.Remote, t.Link, true + } + } + return store.Local{}, store.Remote{}, store.Link{}, false +} + // startForward flips a forward to enabled and brings it up. A local-only // forward restarts/starts its local frpc worker so the re-enabled proxy is // rendered back in; a cluster (non-localOnly) forward is submitted to the ring // so the lowest-load member claims and spawns it (plan §令牌环协议). SubmitTask -// is idempotent via HasTask. Returns store.ErrNotFound if the forward is gone. +// is idempotent via HasTask. Falls back to ring topology when the forward +// exists only in the cluster (owned by another node, not in local store). func (h *Handler) startForward(local, remote string, port int) error { ln, ok := findLinkByTriple(h.Store, local, remote, port) - if !ok { - return store.ErrNotFound - } - loc, ok := h.Store.GetLocal(local) - if !ok { - return store.ErrNotFound - } - rem, ok := h.Store.GetRemote(remote) - if !ok { - return store.ErrNotFound + loc, lok := h.Store.GetLocal(local) + rem, rok := h.Store.GetRemote(remote) + if !ok || !lok || !rok { + // Fall back to ring topology for forwards owned by other nodes. + tLoc, tRem, tLn, tok := h.findLinkFromTopology(local, remote, port) + if !tok { + return store.ErrNotFound + } + ln, loc, rem = tLn, tLoc, tRem } _ = h.Store.SetLinkDisabled(local, remote, port, false) - // Reflect the post-start (enabled) state in the snapshot handed to the - // ring: ln was fetched before SetLinkDisabled, so for a re-start of a - // previously stopped forward it still carries disabled=true. Without this - // the topology entry would embed a stale disabled flag (cosmetically - // wrong, and confusing if anything reads topology's link snapshot). ln.Disabled = false if loc.LocalOnly { if h.Process != nil { - // Re-render (proxy re-added) on a running worker, or start it. if _, has := h.Process.Status(remote); has { _ = h.Process.Restart(remote) } else { @@ -82,23 +92,22 @@ func (h *Handler) startForward(local, remote string, port int) error { // forward restarts its worker so renderRemote omits the proxy (sibling forwards // on the same remote keep running — true per-forward stop); a cluster forward // is revoked so the owning node cancels its worker and drops it from the -// topology (plan §任务撤销). RevokeTask is idempotent. +// topology (plan §任务撤销). RevokeTask is idempotent. Falls back to ring +// topology when the forward exists only in the cluster (owned by another node). func (h *Handler) stopForward(local, remote string, port int) error { ln, ok := findLinkByTriple(h.Store, local, remote, port) - if !ok { - return store.ErrNotFound + loc, lok := h.Store.GetLocal(local) + rem, rok := h.Store.GetRemote(remote) + if !ok || !lok || !rok { + tLoc, tRem, tLn, tok := h.findLinkFromTopology(local, remote, port) + if !tok { + return store.ErrNotFound + } + ln, loc, rem = tLn, tLoc, tRem + } else { + _ = h.Store.SetLinkDisabled(local, remote, port, true) } - loc, ok := h.Store.GetLocal(local) - if !ok { - return store.ErrNotFound - } - rem, ok := h.Store.GetRemote(remote) - if !ok { - return store.ErrNotFound - } - _ = h.Store.SetLinkDisabled(local, remote, port, true) if loc.LocalOnly { - // Re-render without this proxy; only meaningful while a worker runs. if h.Process != nil { if _, has := h.Process.Status(remote); has { _ = h.Process.Restart(remote) @@ -217,9 +226,9 @@ type assignReq struct { // handleForwardsAssign changes the group label of a single forward. The // status page group chip is the quick entry; the canvas port editor also -// carries a group field (both persist via the same store column). The -// change is purely a DB update — no worker/ring action is needed (group is -// a management label, not a runtime knob). +// carries a group field (both persist via the same store column). The local +// DB row is updated if present; the ring topology entry is always updated so +// the group label propagates to all nodes via the next token cycle. func (h *Handler) handleForwardsAssign(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { methodNotAllowed(w) @@ -234,12 +243,21 @@ func (h *Handler) handleForwardsAssign(w http.ResponseWriter, r *http.Request) { http.Error(w, "local/remote/remotePort required", http.StatusBadRequest) return } - if _, ok := findLinkByTriple(h.Store, req.Local, req.Remote, req.RemotePort); !ok { - http.Error(w, store.ErrNotFound.Error(), http.StatusNotFound) - return + found := false + if _, ok := findLinkByTriple(h.Store, req.Local, req.Remote, req.RemotePort); ok { + found = true + if err := h.Store.SetLinkGroup(req.Local, req.Remote, req.RemotePort, req.Group); err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } } - if err := h.Store.SetLinkGroup(req.Local, req.Remote, req.RemotePort, req.Group); err != nil { - http.Error(w, err.Error(), http.StatusInternalServerError) + if h.Ring != nil { + if h.Ring.UpdateTopologyGroup(req.Local, req.Remote, req.RemotePort, req.Group) { + found = true + } + } + if !found { + http.Error(w, store.ErrNotFound.Error(), http.StatusNotFound) return } writeJSON(w, http.StatusOK, map[string]any{"ok": true}) @@ -248,7 +266,8 @@ func (h *Handler) handleForwardsAssign(w http.ResponseWriter, r *http.Request) { // handleForwardsGroupDelete dissolves a group: every forward in the named // group is moved to 未分组 (grp=""). The group itself is not a stored entity // — it exists only as a label on links — so clearing all members is the -// complete "delete". No worker/ring action (group is a management label). +// complete "delete". Clears both local store links and ring topology entries +// so the change syncs to all nodes. func (h *Handler) handleForwardsGroupDelete(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { methodNotAllowed(w) @@ -274,5 +293,14 @@ func (h *Handler) handleForwardsGroupDelete(w http.ResponseWriter, r *http.Reque } _ = h.Store.SetLinkGroup(ln.Local, ln.Remote, ln.RemotePort, "") } + // Also clear group on ring topology entries (forwards owned by other nodes). + if h.Ring != nil { + snap := h.Ring.Snapshot() + for _, t := range snap.Topology { + if t.Link.Group == req.Group { + h.Ring.UpdateTopologyGroup(t.Local.Name, t.Remote.Name, t.Link.RemotePort, "") + } + } + } writeJSON(w, http.StatusOK, map[string]any{"ok": true}) } diff --git a/internal/httpapi/server.go b/internal/httpapi/server.go index 666c3d9..f2c9d37 100644 --- a/internal/httpapi/server.go +++ b/internal/httpapi/server.go @@ -370,6 +370,7 @@ func (h *Handler) handleStatus(w http.ResponseWriter, r *http.Request) { Local: t.Local.Name, Remote: t.Remote.Name, RemotePort: t.Link.RemotePort, LocalOnly: false, Kind: "remote", OwnerID: t.OwnerID, Active: t.Active, LocalIP: t.Local.IP, LocalPort: t.Local.Port, LocalProto: t.Local.Protocol, + Group: t.Link.Group, }) seenFwd[k] = true }