feat(cluster): 停用改为「标记」语义，让 disabled 真正随令牌环跨节点传播

承接用户提问「设计上停用不是本来就会跨节点传输吗」——核实结论：结构上确实如此（TopoEntry.Link 是完整 store.Link，整个 State 随 token 每轮广播），但实际路径断了。断点正是「撤销会删掉 topology 条目」：条目是 flag 的载体，删了就无处传播，于是停用只能靠一次性 revoke 任务投递给 owner，**owner 当时不在线就收不到**（实测 .60 记 disabled=1 / .106 记 0，就是这么来的）。 ## 改为标记而非移除撤销不再 RemoveTopology，而是 UpdateTopologyDisabled(true)，条目保留、 Link.Disabled=true、Active=false。Active 正是为此存在：OfflineReassign() 只处理 Active 条目，所以停用的转发在 owner 掉线时不会被重新排队。 - 新增 UpdateTopologyDisabled / TopologyDisabled（照 UpdateTopologyGroup 的桥） - 新增 store.ReconcileLinkDisabled 作接收端：adoption 时把环上的 flag 落进本地 store；本节点没有该转发时补一条 disabled 占位行（否则日后在本节点被 claim 会复活），enable 则不建行 - SetTopologySync 由单向（store→环）扩为双向：群组仍上行，disabled 下行 - AddTopology 的 Active 跟随 Link.Disabled（原本硬编码 true，认领一个停用转发就会复活它） - 审计日志细分 forward.stop / forward.start，与 forward.remove 区分 ## 语义变更带出的两个新问题（都已修） 1. **「启动」这条路断了**。条目保留 ⇒ SubmitTask 被去重挡下，而认领路径的 duplicate-claim 防御又会丢弃「已有 owner」的任务 ⇒ 重启任务发不出去，owner 永远收不到，转发**能停不能起**。修：新增 Task.Restart 这一独立任务类型 + SubmitRestart + Handler.RestartFn，显式绕过 duplicate-claim 防御并原地复活（不重复建条目、不重跑 claim 簿记）。 SubmitTask 的守卫同时从 HasTask 收窄为新的 HasActiveTask（跳过 disabled 条目与撤销任务）；saveCanvas 的判断相应改用 HasActiveTask，避免每次保存都对已标记的转发重复发撤销。 2. 原本两处 RemoveTopologyEntry 调用（ClaimFn/RevokeFn 的 disabled 分支）在新语义下会把本该保留的条目删掉，改为 UpdateTopologyDisabled。 ## 测试（每个都做了「回退修复行→必须变红→还原变绿」双向验证） - TestStoppedTopologyEntrySurvivesAdoption —— 离线成员也能学到停用，一次性 revoke 任务永远做不到这一点 - TestStoppedForwardNotRequeuedOnNodeDeparture / TestAddTopologyRespectsDisabledFlag —— 标记而非删除为何安全 - TestSubmitTaskNotBlockedByStoppedEntry / TestSubmitTaskStillDedupesActiveForward - TestRestartTaskBypassesDuplicateClaimGuard / TestRestartFlagSurvivesTokenSerialization - TestStopThenStartPublishesRestartTask（HTTP 端到端，断言**任务真的发出**） - TestReconcileLinkDisabled*（store 侧三条） ★ 两次踩到**假绿**：第一版只断言 store 层（newTestHandler 的 Ring 为 nil，坏掉的路根本没执行）；第二版在 re-enable **之后**才调 SubmitTask，此时新旧谓词结果相同，测不出差异。都是靠「回退修复行看是否变红」抓出来的 —— 这个双向验证已经是本项目的固定动作。 go build / go vet / go test ./... 全绿，gofmt 干净。
2026-10-03 15:43:59 +00:00 · 2026-09-26 10:44:04 +08:00
parent 041cc04dd6
commit 1c835425de
12 changed files with 830 additions and 48 deletions
--- a/internal/cluster/ring.go
+++ b/internal/cluster/ring.go
@ -62,6 +62,12 @@ type Task struct {
 	// owning node cancels it (stop worker, drop from topology). Reuses the
 	// same publish channel as creation (round-1 inject, round-2 apply).
 	Revoke bool `json:"revoke,omitempty"`
+	// Restart marks a RE-ENABLE task: the forward is already claimed and its
+	// topology entry still exists (a stop keeps the entry as the flag's
+	// carrier), so the creation path would dedupe the submission and the
+	// duplicate-claim guard would discard it. The owner applies this one
+	// unconditionally: re-mark the entry enabled and (re)spawn the worker.
+	Restart bool `json:"restart,omitempty"`
 	// RemoveNode: node ID to remove from the ring; the node self-removes when
 	// the command reaches it via the token.
 	RemoveNode string `json:"removeNode,omitempty"`
@ -333,10 +339,15 @@ func (s *State) AddTopology(t *Task, ownerID string) *TopoEntry {
 	if s.Topology == nil {
 		s.Topology = map[string]*TopoEntry{}
 	}
+	// Active mirrors the task's disabled flag rather than being unconditionally
+	// true. A claim can legitimately carry a stopped forward (e.g. a node
+	// re-claiming from a departed peer), and forcing Active=true there would
+	// resurrect it: OfflineReassign only re-queues Active entries, and every
+	// reader that filters on Active would treat it as running again.
 	e := &TopoEntry{
 		TaskID: t.ID, OwnerID: ownerID,
 		Local: t.Local, Remote: t.Remote, Link: t.Link,
-		Active: true,
+		Active: !t.Link.Disabled,
 	}
 	s.Topology[t.ID] = e
 	return e
@ -408,6 +419,39 @@ func (s *State) ForwardsOwnedBy(owner string) []*TopoEntry {
 // (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.
+// UpdateTopologyDisabled sets the disabled flag on the topology entry matching
+// the forward's natural key. Mirror of UpdateTopologyGroup: it makes a locally
+// decided stop/start part of the topology so the next token cycle carries it to
+// every other member (the alternative — a one-shot revoke task addressed at the
+// owner — only converges if that owner happens to be online right then).
+//
+// Returns false when the forward is not in the topology, which is not an error:
+// a stopped forward may legitimately have no entry yet.
+func (s *State) UpdateTopologyDisabled(local, remote string, port int, disabled bool) bool {
+	for _, e := range s.Topology {
+		if e.Local.Name == local && e.Remote.Name == remote && e.Link.RemotePort == port {
+			e.Link.Disabled = disabled
+			// Keep Active consistent with the flag so readers that key off it
+			// (status page, load accounting) agree with Link.Disabled.
+			e.Active = !disabled
+			return true
+		}
+	}
+	return false
+}
+
+// TopologyDisabled returns the disabled flag the cluster currently holds for a
+// forward, and whether such an entry exists. Used by the adoption path to learn
+// a peer's decision into the local store.
+func (s *State) TopologyDisabled(local, remote string, port int) (bool, bool) {
+	for _, e := range s.Topology {
+		if e.Local.Name == local && e.Remote.Name == remote && e.Link.RemotePort == port {
+			return e.Link.Disabled, true
+		}
+	}
+	return false, false
+}
+
 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 {
--- a/internal/cluster/ring_engine.go
+++ b/internal/cluster/ring_engine.go
@ -18,6 +18,11 @@ import (
 type Handler interface {
 	Claim(ctx context.Context, tk *Task) error
 	Revoke(ctx context.Context, tk *Task) error
+	// Restart re-enables an already-claimed forward and brings its worker back.
+	// Needed because a stop keeps the topology entry, which makes both the
+	// creation path (idempotency guard) and the duplicate-claim guard refuse to
+	// re-create an existing forward.
+	Restart(ctx context.Context, tk *Task) error
 	RuntimeLoad() Load
 }

@ -390,23 +395,34 @@ func (e *Engine) runCommands(ctx context.Context, tk *Token) error {
 			break
 		}
 		if claimed.Revoke {
-			// The revoke handler must run REGARDLESS of whether a topology entry
-			// existed. RemoveTopology() returns false when the forward is not in
-			// the topology, and gating the handler on it made the revoke a silent
-			// no-op in exactly the case that matters: a forward the user stopped
-			// while it was already absent from the topology kept its worker
-			// running forever (observed live — a disabled forward kept dialing a
-			// local service that was intentionally down, thousands of
-			// connection-refused lines). Stopping the worker and retiring the
-			// entry are independent duties; the entry may simply not be there.
-			removed := e.state.RemoveTopology(claimed.Local.Name, claimed.Remote.Name, claimed.Link.RemotePort)
+			// A user-requested stop is recorded as a FLAG on the topology entry,
+			// not as a removal. The entry is the carrier that makes the decision
+			// travel: TopoEntry.Link is a full store.Link and State rides the
+			// token every cycle, so keeping the entry is what lets a stop reach
+			// every member — including a node that was offline when the stop was
+			// issued. Removing the entry instead left the flag nowhere to live,
+			// which is why a stop used to converge only if the owner happened to
+			// be online to receive a one-shot revoke task.
+			//
+			// Marking Active=false is also what makes the entry inert for the
+			// rebalancing paths: OfflineReassign() skips inactive entries, and
+			// the startup reconcile skips them too, so a stopped forward is
+			// neither re-queued on a node departure nor re-claimed on a restart.
+			//
+			// Handler.Revoke still runs unconditionally — it is what actually
+			// stops the per-forward worker, and it must run even when there was no
+			// entry to mark (gating it on RemoveTopology()'s result made stops of
+			// already-absent forwards silently do nothing, so their workers ran
+			// forever: observed live as thousands of connection-refused lines
+			// against a local service that was intentionally down).
+			marked := e.state.UpdateTopologyDisabled(claimed.Local.Name, claimed.Remote.Name, claimed.Link.RemotePort, true)
 			if e.Handler != nil {
 				if err := e.Handler.Revoke(ctx, claimed); err != nil {
 					log.Printf("ring[%s] revoke %s: %v", e.ID, claimed.ID, err)
 				}
 			}
-			if removed && e.Log != nil {
-				_, _ = e.Log.Append(e.ID, LogForwardRemove, map[string]any{
+			if marked && e.Log != nil {
+				_, _ = e.Log.Append(e.ID, LogForwardStop, map[string]any{
 					"taskId": claimed.ID, "local": claimed.Local.Name, "remote": claimed.Remote.Name,
 				})
 			}
@ -475,10 +491,38 @@ func (e *Engine) runCommands(ctx context.Context, tk *Token) error {
 		// different node. Safe for OfflineReassign: that path deletes the
 		// topology entry BEFORE re-queueing, so TopologyOwner returns "" and
 		// the legitimate re-claim passes through.
-		if owner := e.state.TopologyOwner(claimed); owner != "" {
-			log.Printf("ring[%s] drop duplicate task %s: %s→%s:%d already owned by %s",
-				e.ID, claimed.ID, claimed.Local.Name, claimed.Remote.Name,
-				claimed.Link.RemotePort, owner)
+		//
+		// A RESTART task is the deliberate exception this guard must not eat: it
+		// exists precisely to re-activate a forward whose entry is still present
+		// (a stop keeps the entry as the flag's carrier), so applying the guard
+		// would swallow every re-enable and leave the forward stopped forever.
+		if !claimed.Restart {
+			if owner := e.state.TopologyOwner(claimed); owner != "" {
+				log.Printf("ring[%s] drop duplicate task %s: %s→%s:%d already owned by %s",
+					e.ID, claimed.ID, claimed.Local.Name, claimed.Remote.Name,
+					claimed.Link.RemotePort, owner)
+				continue
+			}
+		}
+		if claimed.Restart {
+			// Re-enable in place and hand the worker back to this node (the
+			// entry's owner). This must NOT create a second topology entry, which
+			// is why it skips the claim bookkeeping below.
+			e.state.UpdateTopologyDisabled(claimed.Local.Name, claimed.Remote.Name, claimed.Link.RemotePort, false)
+			if e.Handler != nil {
+				if err := e.Handler.Restart(ctx, claimed); err != nil {
+					e.state.PendingTasks[claimed.ID] = claimed
+					log.Printf("ring[%s] restart %s failed: %v", e.ID, claimed.ID, err)
+					break
+				}
+			}
+			if e.Log != nil {
+				_, _ = e.Log.Append(e.ID, LogForwardStart, map[string]any{
+					"taskId": claimed.ID, "local": claimed.Local.Name, "remote": claimed.Remote.Name,
+				})
+			}
+			log.Printf("ring[%s] restarted %s: %s→%s:%d", e.ID, claimed.ID,
+				claimed.Local.Name, claimed.Remote.Name, claimed.Link.RemotePort)
 			continue
 		}
 		if e.Handler != nil {
@ -817,14 +861,87 @@ func (e *Engine) RevokeTask(local store.Local, remote store.Remote, link store.L
 }

 func (e *Engine) SubmitTask(local store.Local, remote store.Remote, link store.Link) *Task {
-	if e.HasTask(local.Name, remote.Name, link.RemotePort) {
+	// Guard on an ACTIVE task only. A topology entry that is merely marked
+	// disabled is kept around as the carrier for the stop flag, so counting it
+	// here would make this a permanent no-op and a stopped forward could never
+	// be started again.
+	if e.HasActiveTask(local.Name, remote.Name, link.RemotePort) {
 		return nil
 	}
 	return e.state.AddPending(local, remote, link)
 }

-// HasTask reports whether a forward with the same local/remote/remotePort is
-// already pending or active in the topology (idempotency guard for resaves).
+// HasTopologyEntry reports whether the forward has a topology entry (whether or
+// not it is marked disabled). A stop keeps the entry as the flag's carrier, so
+// "has an entry" is what distinguishes "already claimed by someone" from "never
+// submitted" — the distinction startForward needs.
+func (e *Engine) HasTopologyEntry(local, remote string, port int) bool {
+	_, found := e.state.TopologyDisabled(local, remote, port)
+	return found
+}
+
+// SubmitRestart publishes a task asking the current owner of an already-claimed
+// forward to bring its worker back up.
+//
+// Neither of the existing channels can express "restart what already exists":
+// SubmitTask is guarded by HasActiveTask and the entry is still present, so it
+// dedupes; and the claim path drops any task for a forward that already has an
+// owner (its defence against duplicate-claim collisions). Re-enabling a stopped
+// forward therefore needs its own task kind, which the owner applies without
+// re-running the claim bookkeeping.
+func (e *Engine) SubmitRestart(local store.Local, remote store.Remote, link store.Link) *Task {
+	// The task must not carry the stop: the whole point is to re-enable.
+	link.Disabled = false
+	t := &Task{
+		ID:      e.state.NextTaskID(),
+		Local:   local,
+		Remote:  remote,
+		Link:    link,
+		Created: time.Now().Unix(),
+		Restart: true,
+	}
+	if e.state.PendingTasks == nil {
+		e.state.PendingTasks = map[string]*Task{}
+	}
+	e.state.PendingTasks[t.ID] = t
+	return t
+}
+
+// HasActiveTask reports whether a forward with this key is genuinely in flight
+// or running: a non-revocation pending task, or a topology entry that is not
+// marked disabled.
+//
+// It is deliberately narrower than HasTask. Under mark-don't-remove a stopped
+// forward keeps its topology entry (that entry is what carries the flag between
+// nodes), so "an entry exists" no longer means "this forward is active".
+// SubmitTask must use THIS predicate, otherwise starting a stopped forward —
+// startForward() publishes the enable and then calls SubmitTask — would be
+// swallowed by its own idempotency guard.
+func (e *Engine) HasActiveTask(local, remote string, port int) bool {
+	for _, t := range e.state.PendingList() {
+		if t.Revoke {
+			continue // a revocation is the opposite of an active forward
+		}
+		if t.Local.Name == local && t.Remote.Name == remote && t.Link.RemotePort == port {
+			return true
+		}
+	}
+	for _, t := range e.state.TopologyList() {
+		if t.Link.Disabled {
+			continue // stopped; the entry is only a flag carrier
+		}
+		if t.Local.Name == local && t.Remote.Name == remote && t.Link.RemotePort == port {
+			return true
+		}
+	}
+	return false
+}
+
+// HasTask reports whether ANY record for this forward exists — a pending task
+// (including an in-flight revocation) or a topology entry (including one marked
+// disabled). This is the "have we already handled this key" question, used to
+// avoid re-issuing work on repeated canvas saves. For "is it actually running"
+// use HasActiveTask instead.
 func (e *Engine) HasTask(local, remote string, port int) bool {
 	for _, t := range e.state.PendingList() {
 		if t.Local.Name == local && t.Remote.Name == remote && t.Link.RemotePort == port {
@ -846,6 +963,19 @@ func (e *Engine) UpdateTopologyGroup(local, remote string, port int, group strin
 	return e.state.UpdateTopologyGroup(local, remote, port, group)
 }

+// UpdateTopologyDisabled marks a forward enabled/disabled in the topology so
+// the decision rides the next token cycle to every member. Paired with
+// TopologyDisabled, which the adoption hook uses to learn peers' decisions.
+func (e *Engine) UpdateTopologyDisabled(local, remote string, port int, disabled bool) bool {
+	return e.state.UpdateTopologyDisabled(local, remote, port, disabled)
+}
+
+// TopologyDisabled reports the cluster's view of a forward's disabled flag.
+// The bool is false when the forward has no topology entry (nothing to learn).
+func (e *Engine) TopologyDisabled(local, remote string, port int) (bool, bool) {
+	return e.state.TopologyDisabled(local, remote, port)
+}
+
 // IsLeader reports whether this node is the current ring leader.
 func (e *Engine) IsLeader() bool { return e.state.LeaderID == e.ID }

--- a/internal/cluster/ring_engine_test.go
+++ b/internal/cluster/ring_engine_test.go
@ -8,9 +8,10 @@ import (
 )

 type fakeHandler struct {
-	load   Load
-	claim  func(ctx context.Context, tk *Task) error
-	revoke func(ctx context.Context, tk *Task) error
+	load    Load
+	claim   func(ctx context.Context, tk *Task) error
+	revoke  func(ctx context.Context, tk *Task) error
+	restart func(ctx context.Context, tk *Task) error
 }

 func (h *fakeHandler) Revoke(ctx context.Context, tk *Task) error {
@ -20,6 +21,13 @@ func (h *fakeHandler) Revoke(ctx context.Context, tk *Task) error {
 	return nil
 }

+func (h *fakeHandler) Restart(ctx context.Context, tk *Task) error {
+	if h.restart != nil {
+		return h.restart(ctx, tk)
+	}
+	return nil
+}
+
 func (h *fakeHandler) RuntimeLoad() Load { return h.load }
 func (h *fakeHandler) Claim(ctx context.Context, tk *Task) error {
 	if h.claim != nil {
--- a/internal/cluster/ring_handler.go
+++ b/internal/cluster/ring_handler.go
@ -17,6 +17,10 @@ type AppHandler struct {
 	ClaimFn func(ctx context.Context, tk *Task) error
 	// RevokeFn cancels the forward on this node (stop worker + drop from store).
 	RevokeFn func(ctx context.Context, tk *Task) error
+	// RestartFn re-enables an already-claimed forward on this node (clear the
+	// stopped flag + bring the worker back). Required because a stop keeps the
+	// topology entry, so the normal claim path will not re-create it.
+	RestartFn func(ctx context.Context, tk *Task) error
 }

 // RuntimeLoad implements Handler.
@ -44,6 +48,14 @@ func (a *AppHandler) Revoke(ctx context.Context, tk *Task) error {
 	return a.RevokeFn(ctx, tk)
 }

+// Restart implements Handler.
+func (a *AppHandler) Restart(ctx context.Context, tk *Task) error {
+	if a.RestartFn == nil {
+		return nil
+	}
+	return a.RestartFn(ctx, tk)
+}
+
 // SampleMemLoad returns a cheap memory-usage percentage (0..100).
 func SampleMemLoad() float64 {
 	var m runtime.MemStats
--- a/internal/cluster/ring_log.go
+++ b/internal/cluster/ring_log.go
@ -20,6 +20,8 @@ import (
 // LogKind enumerates operation-log entry types.
 const (
 	LogForwardAdd    = "forward.add"
+	LogForwardStop   = "forward.stop"
+	LogForwardStart  = "forward.start"
 	LogForwardRemove = "forward.remove"
 	LogNodeJoin      = "node.join"
 	LogNodeLeave     = "node.leave"
@ -140,7 +142,7 @@ func DetailOf(e LogEntry) string {
 	}
 	str := func(key string) string { s, _ := d[key].(string); return s }
 	switch e.Kind {
-	case LogForwardAdd, LogForwardRemove:
+	case LogForwardAdd, LogForwardStop, LogForwardStart, LogForwardRemove:
 		s := str("local") + " → " + str("remote")
 		if id := str("taskId"); id != "" {
 			if len(id) > 8 {
--- a/internal/cluster/ring_revoke_test.go
+++ b/internal/cluster/ring_revoke_test.go
@ -2,14 +2,22 @@ package cluster

 import (
 	"context"
+	"encoding/json"
 	"testing"

 	"webui4frpc/internal/store"
 )

-// TestRevokeTaskRemovesTopology: a revoke task published to the ring removes
-// the forward from topology, calls Handler.Revoke, and logs forward.remove.
-func TestRevokeTaskRemovesTopology(t *testing.T) {
+// TestRevokeTaskMarksTopologyDisabled: a stop delivered through the ring marks
+// the topology entry disabled instead of deleting it, calls Handler.Revoke, and
+// logs forward.stop.
+//
+// The entry is deliberately KEPT: TopoEntry.Link is the carrier that makes the
+// stop travel, and State rides the token every cycle. Keeping it is what lets a
+// stop reach a member that was offline when the stop was issued — deleting the
+// entry left the flag nowhere to live, so a stop converged only if the owning
+// node happened to be online for a one-shot revoke task.
+func TestRevokeTaskMarksTopologyDisabled(t *testing.T) {
 	revoked := false
 	eng := NewEngine("n1", "n1:7500", "u", "p", "0.71.0", nil,
 		&fakeHandler{load: Load{MemPct: 5, NetPct: 5},
@ -26,27 +34,73 @@ func TestRevokeTaskRemovesTopology(t *testing.T) {
 	if len(eng.state.TopologyList()) != 1 {
 		t.Fatalf("topology after claim = %+v", eng.state.TopologyList())
 	}
+	if d, _ := eng.state.TopologyDisabled("web", "frps1", 18081); d {
+		t.Fatal("a freshly claimed forward must not start out disabled")
+	}

 	// publish a revoke task pointing at the same forward
 	eng.state.AddRevoke(store.Local{Name: "web"}, store.Remote{Name: "frps1"}, store.Link{RemotePort: 18081})
 	if _, err := eng.OnToken(context.Background(), &Token{Cycle: 2, State: eng.state}); err != nil {
 		t.Fatal(err)
 	}
-	if len(eng.state.TopologyList()) != 0 {
-		t.Fatalf("topology after revoke = %+v", eng.state.TopologyList())
+
+	// The entry must SURVIVE, marked disabled, so the flag keeps travelling.
+	if len(eng.state.TopologyList()) != 1 {
+		t.Fatalf("topology entry was removed; the disabled flag would have no carrier: %+v",
+			eng.state.TopologyList())
+	}
+	disabled, known := eng.state.TopologyDisabled("web", "frps1", 18081)
+	if !known {
+		t.Fatal("topology entry missing after revoke")
+	}
+	if !disabled {
+		t.Fatal("revoke did not mark the topology entry disabled")
+	}
+	// Active must follow the flag, since OfflineReassign only re-queues Active
+	// entries — a stopped forward must not be resurrected by a node departure.
+	if eng.state.TopologyList()[0].Active {
+		t.Fatal("a stopped forward must not remain Active (OfflineReassign would re-queue it)")
 	}
 	if !revoked {
 		t.Fatal("Handler.Revoke was not called")
 	}
-	// log should contain forward.remove
-	var sawRemove bool
+	// log should contain forward.stop (not forward.remove — nothing was removed)
+	var sawStop bool
 	for _, e := range eng.Log.Snapshot() {
-		if e.Kind == LogForwardRemove {
-			sawRemove = true
+		if e.Kind == LogForwardStop {
+			sawStop = true
 		}
 	}
-	if !sawRemove {
-		t.Fatalf("log missing forward.remove: %+v", eng.Log.Snapshot())
+	if !sawStop {
+		t.Fatalf("log missing forward.stop: %+v", eng.Log.Snapshot())
+	}
+}
+
+// TestStoppedTopologyEntrySurvivesAdoption is the property that makes the whole
+// mark-don't-remove design work: because the entry persists, a node that adopts
+// ring state learns the stop. This is the offline-member case a one-shot revoke
+// task could never cover.
+func TestStoppedTopologyEntrySurvivesAdoption(t *testing.T) {
+	src := newTestEngine("n1", true)
+	src.state.AddPending(store.Local{Name: "web"}, store.Remote{Name: "frps1"}, store.Link{RemotePort: 18081})
+	if _, err := src.OnToken(context.Background(), &Token{Cycle: 1, State: src.state}); err != nil {
+		t.Fatal(err)
+	}
+	src.state.AddRevoke(store.Local{Name: "web"}, store.Remote{Name: "frps1"}, store.Link{RemotePort: 18081})
+	if _, err := src.OnToken(context.Background(), &Token{Cycle: 2, State: src.state}); err != nil {
+		t.Fatal(err)
+	}
+
+	// A different node adopts the ring state (what e.state = tk.State does).
+	peer := newTestEngine("n2", false)
+	peer.AdoptState(src.state)
+
+	disabled, known := peer.TopologyDisabled("web", "frps1", 18081)
+	if !known {
+		t.Fatal("peer did not receive the topology entry for the stopped forward")
+	}
+	if !disabled {
+		t.Fatal("peer adopted the entry but no longer sees it as disabled — the stop did not propagate")
 	}
 }

@ -80,3 +134,213 @@ func TestRevokeIdempotent(t *testing.T) {
 			"must still stop the worker, otherwise stopped forwards keep running", revoked)
 	}
 }
+
+// TestStoppedForwardNotRequeuedOnNodeDeparture pins why marking (rather than
+// deleting) is safe: OfflineReassign only re-queues ACTIVE entries, so a
+// stopped forward is not silently handed to another node when its owner goes
+// away. Deleting the entry, or leaving it Active, would both resurrect it.
+func TestStoppedForwardNotRequeuedOnNodeDeparture(t *testing.T) {
+	eng := newTestEngine("n1", true)
+	eng.state.AddPending(store.Local{Name: "web"}, store.Remote{Name: "frps1"}, store.Link{RemotePort: 18081})
+	if _, err := eng.OnToken(context.Background(), &Token{Cycle: 1, State: eng.state}); err != nil {
+		t.Fatal(err)
+	}
+	eng.state.AddRevoke(store.Local{Name: "web"}, store.Remote{Name: "frps1"}, store.Link{RemotePort: 18081})
+	if _, err := eng.OnToken(context.Background(), &Token{Cycle: 2, State: eng.state}); err != nil {
+		t.Fatal(err)
+	}
+
+	owner := eng.state.TopologyList()[0].OwnerID
+	eng.state.OfflineReassign(owner)
+
+	if n := len(eng.state.PendingList()); n != 0 {
+		t.Fatalf("a stopped forward was re-queued on the owner's departure (pending=%d): %+v",
+			n, eng.state.PendingList())
+	}
+}
+
+// TestSubmitTaskNotBlockedByStoppedEntry is the regression test for the bug the
+// mark-don't-remove change introduced: SubmitTask's idempotency guard used
+// HasTask, which matches a disabled topology entry. Since a stop now KEEPS the
+// entry, any submission for that forward while it is still stopped is swallowed
+// by its own guard — so a stopped forward can never be started again.
+//
+// The probe deliberately submits WITHOUT re-enabling first. Re-enabling would
+// clear the flag and make HasTask and HasActiveTask agree, hiding the defect;
+// the guard has to be exercised at the moment the entry is still stopped.
+func TestSubmitTaskNotBlockedByStoppedEntry(t *testing.T) {
+	eng := newTestEngine("n1", true)
+	eng.state.AddPending(store.Local{Name: "web"}, store.Remote{Name: "frps1"}, store.Link{RemotePort: 18081})
+	if _, err := eng.OnToken(context.Background(), &Token{Cycle: 1, State: eng.state}); err != nil {
+		t.Fatal(err)
+	}
+
+	// stop it -> entry is kept, marked disabled
+	eng.state.AddRevoke(store.Local{Name: "web"}, store.Remote{Name: "frps1"}, store.Link{RemotePort: 18081})
+	if _, err := eng.OnToken(context.Background(), &Token{Cycle: 2, State: eng.state}); err != nil {
+		t.Fatal(err)
+	}
+	if d, _ := eng.state.TopologyDisabled("web", "frps1", 18081); !d {
+		t.Fatal("precondition: forward should be disabled")
+	}
+	// The stopped entry IS findable by the broad predicate — which is exactly why
+	// the narrow one has to exist.
+	if !eng.HasTask("web", "frps1", 18081) {
+		t.Fatal("precondition: the stopped entry should still be findable by HasTask")
+	}
+	if eng.HasActiveTask("web", "frps1", 18081) {
+		t.Fatal("precondition: a stopped forward must NOT count as active")
+	}
+
+	// The regression: a submission must be published for a stopped forward.
+	// With a HasTask guard this returns nil and the forward is stuck forever.
+	tk := eng.SubmitTask(store.Local{Name: "web"}, store.Remote{Name: "frps1"},
+		store.Link{RemotePort: 18081})
+	if tk == nil {
+		t.Fatal("SubmitTask was swallowed by the stopped topology entry — " +
+			"a stopped forward could never be started again")
+	}
+	if tk.Link.RemotePort != 18081 {
+		t.Fatalf("unexpected task: %+v", tk)
+	}
+}
+
+// TestSubmitTaskStillDedupesActiveForward is the other half of the guard: the
+// narrower predicate must not turn SubmitTask into a duplicate-task generator.
+func TestSubmitTaskStillDedupesActiveForward(t *testing.T) {
+	eng := newTestEngine("n1", true)
+	eng.state.AddPending(store.Local{Name: "web"}, store.Remote{Name: "frps1"}, store.Link{RemotePort: 18081})
+	if _, err := eng.OnToken(context.Background(), &Token{Cycle: 1, State: eng.state}); err != nil {
+		t.Fatal(err)
+	}
+	if !eng.HasActiveTask("web", "frps1", 18081) {
+		t.Fatal("precondition: a claimed forward must count as active")
+	}
+	if tk := eng.SubmitTask(store.Local{Name: "web"}, store.Remote{Name: "frps1"},
+		store.Link{RemotePort: 18081}); tk != nil {
+		t.Fatalf("SubmitTask must stay idempotent for an active forward, got %+v", tk)
+	}
+}
+
+// TestStoppedForwardNotRequeuedOnNodeDeparture pins why marking (rather than
+// deleting) is safe: OfflineReassign only re-queues ACTIVE entries, so a
+// stopped forward is not silently handed to another node when its owner goes
+
+// TestAddTopologyRespectsDisabledFlag: a claim that carries a stopped forward
+// must not mark the new entry Active, or it would come back to life.
+func TestAddTopologyRespectsDisabledFlag(t *testing.T) {
+	s := &State{Topology: map[string]*TopoEntry{}}
+	e := s.AddTopology(&Task{
+		ID: "t1", Local: store.Local{Name: "web"}, Remote: store.Remote{Name: "frps1"},
+		Link: store.Link{RemotePort: 18081, Disabled: true},
+	}, "n1")
+	if e.Active {
+		t.Fatal("AddTopology forced Active=true for a disabled claim — the forward would resurrect")
+	}
+	if !e.Link.Disabled {
+		t.Fatal("AddTopology dropped the disabled flag from the link")
+	}
+}
+
+// TestRestartTaskBypassesDuplicateClaimGuard is the engine half of the
+// re-enable path: a restart task for a forward that still HAS a topology entry
+// must reach Handler.Restart.
+//
+// This is precisely what the duplicate-claim guard refuses — it exists to stop
+// a stale task from spawning a second worker for an owned forward, and a restart
+// is indistinguishable from that unless it is an explicit task kind. Marking
+// the entry stopped (instead of deleting it) is what made this necessary: the
+// entry the guard keys on now outlives a stop.
+func TestRestartTaskBypassesDuplicateClaimGuard(t *testing.T) {
+	restarts := 0
+	claims := 0
+	eng := NewEngine("n1", "n1:7500", "u", "p", "0.71.0", nil,
+		&fakeHandler{load: Load{MemPct: 5, NetPct: 5},
+			claim:   func(ctx context.Context, tk *Task) error { claims++; return nil },
+			restart: func(ctx context.Context, tk *Task) error { restarts++; return nil }},
+		func(ctx context.Context, next string, tk *Token) error { return nil },
+		"n1:7500", true, "")
+
+	// Establish + stop: entry survives, marked disabled.
+	eng.state.AddPending(store.Local{Name: "web"}, store.Remote{Name: "frps1"}, store.Link{RemotePort: 18081})
+	if _, err := eng.OnToken(context.Background(), &Token{Cycle: 1, State: eng.state}); err != nil {
+		t.Fatal(err)
+	}
+	eng.state.AddRevoke(store.Local{Name: "web"}, store.Remote{Name: "frps1"}, store.Link{RemotePort: 18081})
+	if _, err := eng.OnToken(context.Background(), &Token{Cycle: 2, State: eng.state}); err != nil {
+		t.Fatal(err)
+	}
+	if d, _ := eng.state.TopologyDisabled("web", "frps1", 18081); !d {
+		t.Fatal("precondition: should be disabled")
+	}
+	entryCount := len(eng.state.TopologyList())
+
+	// Start: publish a restart and run a cycle.
+	eng.state.UpdateTopologyDisabled("web", "frps1", 18081, false)
+	eng.SubmitRestart(store.Local{Name: "web"}, store.Remote{Name: "frps1"},
+		store.Link{RemotePort: 18081})
+	if _, err := eng.OnToken(context.Background(), &Token{Cycle: 3, State: eng.state}); err != nil {
+		t.Fatal(err)
+	}
+
+	if restarts != 1 {
+		t.Fatalf("Handler.Restart called %d times, want 1 — the restart task was swallowed "+
+			"by the duplicate-claim guard, so the forward would never come back", restarts)
+	}
+	if claims != 1 {
+		t.Fatalf("Handler.Claim called %d times, want 1 (the original claim only); "+
+			"a restart must not re-run the claim bookkeeping", claims)
+	}
+	// Re-enabling must not create a second entry.
+	if n := len(eng.state.TopologyList()); n != entryCount {
+		t.Fatalf("restart created a duplicate topology entry: %d -> %d", entryCount, n)
+	}
+	if d, _ := eng.state.TopologyDisabled("web", "frps1", 18081); d {
+		t.Fatal("entry is still disabled after the restart task was applied")
+	}
+	var sawStart bool
+	for _, e := range eng.Log.Snapshot() {
+		if e.Kind == LogForwardStart {
+			sawStart = true
+		}
+	}
+	if !sawStart {
+		t.Fatalf("log missing forward.start: %+v", eng.Log.Snapshot())
+	}
+}
+
+// TestRestartFlagSurvivesTokenSerialization: the restart task travels to the
+// owner inside the token, so the flag must round-trip through JSON. Without the
+// struct tag it would silently deserialize as false and the owner would treat
+// it as an ordinary claim — which the duplicate guard then discards.
+func TestRestartFlagSurvivesTokenSerialization(t *testing.T) {
+	eng := newTestEngine("n1", true)
+	eng.SubmitRestart(store.Local{Name: "web"}, store.Remote{Name: "frps1"},
+		store.Link{RemotePort: 18081})
+
+	blob, err := json.Marshal(&Token{Cycle: 7, State: eng.state})
+	if err != nil {
+		t.Fatal(err)
+	}
+	var back Token
+	if err := json.Unmarshal(blob, &back); err != nil {
+		t.Fatal(err)
+	}
+	var found *Task
+	for _, tk := range back.State.PendingList() {
+		if tk.Local.Name == "web" && tk.Link.RemotePort == 18081 {
+			found = tk
+		}
+	}
+	if found == nil {
+		t.Fatal("restart task did not survive token serialization")
+	}
+	if !found.Restart {
+		t.Fatal("the restart flag was lost in JSON — the owner would see a plain claim " +
+			"and the duplicate guard would discard it")
+	}
+	// Revoke and Restart must stay distinguishable.
+	if found.Revoke {
+		t.Fatal("a restart task must not also read as a revocation")
+	}
+}