webui4frpc/internal/cluster/ring_engine.go

// Package cluster implements the token-ring cooperative network.
// Authoritative design: plan.md §M6 (令牌环网拓扑).
package cluster

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"strconv"
	"sync"
	"time"

	"webui4frpc/internal/store"
)

// Handler is what the engine calls when the node must act on a task
// (create the frpc worker / forward). Injected to avoid import cycle.
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
}

// Engine drives one member of the token ring.
type Engine struct {
	ID      string
	Addr    string
	User    string
	Pass    string
	Version string
	Cache   []string
	Handler Handler

	// nodeKey is this node's cluster admission key. A newcomer must present
	// the sponsor's nodeKey (as JoinInfo.JoinKey) to join via it. Persisted
	// in store.Settings; stable across restarts so -join-key stays valid.
	// Generated lazily: on CreateCluster (creator) or AdoptState (joiner),
	// NOT at startup — a fresh node that hasn't created/joined has no key.
	// Cleared on detachAsStandalone (leaving the cluster invalidates the key).
	nodeKey    string
	keyPersist func(key string) error // persists nodeKey to store (nil in tests)
	// peerPersist saves the cached peer list (JSON of [{addr,key},...]) so a
	// crashed node can auto-rejoin on restart via any cached peer. Called on
	// 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()
	// localDisabled records this node's own enabled/disabled decisions that have
	// not yet been confirmed by the ring. It is what makes the RING authoritative
	// without losing a decision that was made locally and has not yet had a
	// chance to reach the other members.
	//
	// The problem it solves: OnToken/AdoptState do `e.state = tk.State`, a
	// wholesale replacement. A stop decided between two token cycles therefore
	// disappears on the very next adoption if that token was captured before the
	// decision — the flag never reaches the rest of the ring, and the forward
	// keeps running. (Verified with a probe before writing this: adopting a
	// stale token flipped disabled back to false.)
	//
	// A key stays in this map until an adoption reports it disabled, at which
	// point the whole cluster agrees and the key is dropped. Keys for forwards
	// that vanish from the topology are swept on every update, so the map cannot
	// grow without bound. The value false is what keeps a RE-ENABLE alive for the
	// same reason a stop needs protecting.
	localDisabled map[string]bool

	state State
	// myAddr maps our Node ID to the address peers dial.
	myAddr string

	// send moves the token to the next node (injected transport).
	send func(ctx context.Context, next string, tk *Token) error

	// inflight tracks token-in-flight state (leader only).
	inflight    inFlight
	Log         *ClusterLog
	lastLogSent int64
	// lastRingStart time of the previous cycle launch (leader throttle).
	lastRingStart time.Time
	// lastTokenAt: leader-stamped token timestamp of the newest valid token
	// this node accepted. Older tokens (multi-token conflict leftovers) are
	// dropped so only one token effectively circulates.
	lastTokenAt int64
	// lastSyncAt unix-seconds when this node last completed a sync round.
	lastSyncAt int64
	// failCount counts consecutive send failures per neighbor; a node is
	// declared offline only after repeated failures (transient jitter must
	// not break the ring).
	failCount map[string]int
	// failMu guards failCount (HTTP handlers run concurrently).
	failMu sync.Mutex
	// pendingJoin: newcomers accepted via JoinNode but NOT YET injected into
	// the token. Per the authoritative design (plan §新节点加入): a sponsor
	// injects the newcomer into the token only when the token reaches it
	// ("令牌发到自己时，先转发给新节点，并把新节点加入令牌中的集群信息").
	// Until then the newcomer is NOT in e.state.Nodes — so the OnToken
	// state-merge (incoming authoritative) cannot wash it back out.
	pendingJoin []JoinInfo
	// selfRemoved is set when a node-remove command targeted THIS node and
	// it has run SelfRemove; the OnToken "append own node info" step is then
	// skipped (otherwise UpsertNode(self) would re-add the removed node).
	selfRemoved bool
	// removedNext: the original successor captured right before SelfRemove,
	// so Forward can hand the token to it even though this node is no longer
	// in state.Nodes (plan §移除节点 step 3: "令牌传递给自身原本的下一家").
	removedNext string
	// published records task IDs this node placed into the token it last
	// forwarded. When the next token returns WITHOUT one of those IDs, the
	// task was consumed downstream (claimed/revoked) — the localPending
	// re-merge must NOT resurrect it, or the task rides forever (a re-claim /
	// re-revoke storm centered on the submitter). Reset each round to the
	// tasks actually leaving on this token.
	published map[string]struct{}
}

// NewEngine builds the engine; state holds this node as initial leader unless
// a peer list says otherwise (creation node starts the ring).
func NewEngine(id, addr, user, pass, version string, cache []string, h Handler, send func(ctx context.Context, next string, tk *Token) error, selfAddr string, isLeader bool, nodeKey string) *Engine {
	e := &Engine{
		ID: id, Addr: addr, User: user, Pass: pass,
		Version: version, Cache: cache, Handler: h,
		nodeKey: nodeKey,
		state: State{
			LeaderID:     "",
			Cycle:        0,
			PendingTasks: map[string]*Task{},
			Topology:     map[string]*TopoEntry{},
			RoundDelay:   2 * time.Second,
		},
		myAddr:    selfAddr,
		send:      send,
		Log:       NewClusterLog(),
		failCount: map[string]int{},
		published: map[string]struct{}{},
	}
	n := Node{ID: id, Addr: selfAddr, Alive: true, IsLeader: isLeader,
		Load: Load{MemPct: 10, NetPct: 10}, Version: version, Cache: cache, NodeKey: nodeKey}
	e.state.UpsertNode(n)
	return e
}

// State returns the engine's current cluster picture.
func (e *Engine) State() *State { return &e.state }

// nextRecipient picks the successor to hand the token to, skipping dead nodes.
func (e *Engine) nextRecipient() (string, bool) {
	return e.state.AliveSuccessor(e.ID)
}

// myNode returns this node's entry from state.
func (e *Engine) myNode() Node {
	i := e.state.Find(e.ID)
	if i < 0 {
		return Node{ID: e.ID, Addr: e.myAddr, Alive: true}
	}
	return e.state.Nodes[i]
}

// loadSnapshot reads our runtime load. The primary signal is the count of
// forwards this node currently owns (Forwards) — this is what makes the
// lowest-load claim actually distribute tasks across nodes instead of the
// leader hogging every claimable task in a single token pass (its stored
// load was never refreshed between claims, so it stayed "lowest"). mem/net
// from the handler only break ties at equal forward count.
func (e *Engine) loadSnapshot() Load {
	l := Load{Forwards: len(e.state.ForwardsOwnedBy(e.ID))}
	if e.Handler != nil {
		b := e.Handler.RuntimeLoad()
		l.MemPct, l.NetPct = b.MemPct, b.NetPct
	} else {
		l.MemPct, l.NetPct = 20, 20
	}
	return l
}

// OnToken is the SINGLE-ROUND token handler. Per the authoritative design
// (plan §令牌环协议): on receiving the token a node simultaneously
//
//	(a) ADOPTS the carried cluster picture — incoming state is authoritative
//	    for membership + per-node fields. This is safe because structural
//	    changes (join/remove) are NOT kept in local state waiting to survive
//	    a merge: joins ride a separate pendingJoin channel injected INTO the
//	    token here, and a self-remove writes the node out of state so the
//	    downstream merge naturally drops it.
//	(b) APPLIES the incremental log delta, then re-attaches own fresh entries.
//	(c) APPENDS own node info (load/version) — skipped if we self-removed.
//	(d) INJECTS pending newcomers right after ourselves + forwards to them.
//	(e) Paces via a parallel rhythm timer (max(ops, timer)).
func (e *Engine) OnToken(ctx context.Context, tk *Token) (*Token, error) {
	// Multi-token guard: only the newest leader-stamped token is kept.
	if tk.SentAt > 0 && e.lastTokenAt > 0 && tk.SentAt < e.lastTokenAt {
		log.Printf("ring[%s] drop stale token sentAt=%d (last=%d)", e.ID, tk.SentAt, e.lastTokenAt)
		return nil, nil
	}
	if tk.SentAt > e.lastTokenAt {
		e.lastTokenAt = tk.SentAt
	}
	// Per-round heartbeat — steady state, no information. See debug.go: this
	// fired ~2x/second and dominated the journal on every node.
	debugToken("ring[%s] OnToken cycle=%d", e.ID, tk.Cycle)

	// Parallel rhythm timer: operations run while the pace clock ticks.
	// Delay scales with alive node count (more nodes → lower per-hop delay,
	// keeping the round time ~constant for real-time sync).
	alive := 0
	for i := range tk.State.Nodes {
		if tk.State.Nodes[i].Alive {
			alive++
		}
	}
	rhythm := time.NewTimer(hopDelayFor(alive))
	defer rhythm.Stop()

	// (a) ADOPT the cluster picture. Incoming state is authoritative: joins
	// ride pendingJoin (injected below), removes write the node out of state
	// so a downstream merge drops it — so a plain assignment is correct and
	// does NOT wash out local structural changes.
	// EXCEPTION: pending COMMANDS (SubmitTask / RemoveNode / AddRevoke) are
	// injected into e.state.PendingTasks by HTTP handlers OUTSIDE OnToken, so
	// a blanket overwrite would drop them before the token carries them. Keep
	// local-only pending commands and re-merge them after the adoption — BUT
	// only those never yet published into a departing token. A task we already
	// published that is now absent from the incoming token was consumed
	// downstream (claimed/revoked); resurrecting it would make it ride forever
	// (a re-claim / re-revoke storm centered on the submitter node).
	localPending := map[string]*Task{}
	for id, t := range e.state.PendingTasks {
		if _, sent := e.published[id]; sent {
			continue
		}
		if _, inToken := tk.State.PendingTasks[id]; inToken {
			continue
		}
		localPending[id] = t
	}
	rd := e.state.RoundDelay // preserve if incoming carries none (zero-guard)
	e.state = tk.State
	if e.state.RoundDelay == 0 {
		e.state.RoundDelay = rd
	}
	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.
	// Re-assert decisions this node made locally that the ring has not yet
	// confirmed, BEFORE the host's topology sync runs: the sync reads
	// TopologyDisabled to learn the cluster's view, so the local decision must
	// already be applied or a stop could be reported as "enabled" and dropped.
	e.reconcileLocalDisabled()
	if e.topologySync != nil {
		e.topologySync()
	}
	for id, t := range localPending {
		e.state.PendingTasks[id] = t
	}

	// (b) apply incremental log delta; trim consumed entries off the token.
	if e.Log != nil && len(tk.Log) > 0 {
		// Raise the local seq allocator above the ring's high-water mark so a
		// restarted node (whose log was rebuilt from scratch at Seq=0) never
		// re-issues sequence numbers that already exist in the shared history —
		// duplicate seqs would make ApplyDelta silently drop the new entries as
		// "already have" and pollute full-log attachments with ambiguous ids.
		for _, en := range tk.Log {
			if en.Seq > e.Log.Seq {
				e.Log.Seq = en.Seq
			}
		}
		wm, err := e.Log.ApplyDelta(tk.Log)
		if err != nil {
			log.Printf("ring[%s] log delta gap: %v (request full sync later)", e.ID, err)
		}
		keep := tk.Log[:0]
		for _, en := range tk.Log {
			if en.Seq > wm {
				keep = append(keep, en)
			}
		}
		tk.Log = keep
	}

	// execute pending commands addressed to us or claimable by lowest load.
	// runCommands may set e.selfRemoved + e.removedNext on a self-remove.
	if err := e.runCommands(ctx, tk); err != nil {
		return tk, err
	}

	// (c) append own node info (refresh load/lastSeen) — skipped when we
	// just self-removed, otherwise UpsertNode(self) would resurrect us and
	// undo the removal the command just performed.
	if !e.selfRemoved {
		e.state.UpsertNode(Node{
			ID: e.ID, Addr: e.myAddr, Alive: true,
			IsLeader: e.state.LeaderID == e.ID,
			Load:     e.loadSnapshot(),
			Version:  e.Version, Cache: e.Cache, NodeKey: e.nodeKey,
		})
	}

	// (d) inject pending newcomers right after ourselves + log node.join.
	// This is the plan's "令牌发到自己时把新节点加入令牌" step: the
	// sponsor writes the newcomer into the token's state and the token then
	// flows to the newcomer (its successor) so it can participate.
	e.injectPendingJoin()

	// re-attach OWN full log so peers converge even after gaps. A node that
	// was offline when seq=N circulated never gets seq=N from the delta (it
	// was trimmed off by peers whose watermark advanced past N). Attaching the
	// FULL local log every cycle lets any peer missing entries backfill them
	// next round — the cluster log is small (tens of entries) so the cost is
	// negligible. ApplyDelta dedupes by seq so re-sent entries are a no-op
	// for peers that already have them.
	if !e.selfRemoved && e.Log != nil {
		mine := e.Log.Snapshot()
		if len(mine) > 0 {
			tk.Log = append(tk.Log, mine...)
			if last := mine[len(mine)-1]; last.Seq > e.lastLogSent {
				e.lastLogSent = last.Seq
			}
		}
	}

	// Publish our consolidated state back into the token.
	tk.State = e.state
	e.lastSyncAt = time.Now().Unix()
	// Persist the cached peer list so a crash/restart can auto-rejoin.
	e.persistPeers()
	// Record the tasks leaving on this token so their absence from the next
	// incoming token is recognized as "consumed downstream" rather than
	// "never sent" — otherwise the localPending re-merge above would
	// resurrect them and they would ride the ring forever.
	e.published = make(map[string]struct{}, len(e.state.PendingTasks))
	for id := range e.state.PendingTasks {
		e.published[id] = struct{}{}
	}
	// Forward only after BOTH the operations and the rhythm timer are done.
	select {
	case <-rhythm.C:
	case <-ctx.Done():
		return tk, ctx.Err()
	}
	return tk, nil
}

// runCommands executes pending commands carried by the token that this node
// must handle. Per the authoritative design (plan §M6), commands are DIRECTED
// to their executor — they are NOT blindly claimed by the lowest-load node:
//   - node-removal: rides the token until it reaches the TARGET node, which
//     self-removes (plan §移除节点: "令牌传递到被移除节点自身时，该节点执行
//     自移除"). A remove command for ANOTHER node is left in pending so it keeps
//     riding; if its target is already gone/offline, the lowest node consumes
//     it as a no-op so it cannot ride forever.
//   - revocation: rides to the OWNING node (plan §任务撤销: "持有该转发的节点
//     收到撤销任务后取消"); a revoke whose forward is already absent from the
//     topology is an idempotent no-op, consumed by the lowest-load node.
//   - forward creation: claimed by the lowest-load node (plan §负载摘取).
func (e *Engine) runCommands(ctx context.Context, tk *Token) error {
	for {
		pending := e.state.PendingList()
		if len(pending) == 0 {
			break
		}
		selfIsLowest := false
		if low := e.state.LowestAlive(); low != nil && low.ID == e.ID {
			selfIsLowest = true
		}
		// Pick the first command this node may act on this pass.
		var target *Task
		for _, t := range pending {
			if t.RemoveNode == e.ID {
				target = t // directed at us — execute regardless of load
				break
			}
			if t.RemoveNode != "" {
				// directed at another node; let it ride unless the target is
				// already gone (not in ring / offline) — then the lowest node
				// consumes the stale command as a no-op so it cannot loop.
				if selfIsLowest {
					if idx := e.state.Find(t.RemoveNode); idx < 0 || !e.state.Nodes[idx].Alive {
						target = t
						break
					}
				}
				continue
			}
			if t.Revoke {
				if owner := e.state.TopologyOwner(t); owner == e.ID {
					target = t // we own the forward — execute the revoke
					break
				} else if owner == "" && selfIsLowest {
					target = t // forward already gone — idempotent no-op
					break
				}
				continue // owned elsewhere — ride to the owner
			}
			if t.Restart {
				// Owner-directed, exactly like a revoke: only the node holding the
				// forward may act. Selecting it anywhere else would let a non-owner
				// spawn a worker for somebody else's forward (observed live: a
				// non-owner logged "restarted t4" and ran the forward itself).
				//
				// A restart whose forward is already gone is a no-op consumed by the
				// lowest node, so a task can never ride forever if its owner
				// departed before seeing it.
				if owner := e.state.TopologyOwner(t); owner == e.ID {
					target = t
					break
				} else if owner == "" && selfIsLowest {
					target = t // forward gone — consume and re-enable nothing
					break
				}
				continue // owned elsewhere — ride to the owner
			}
			if selfIsLowest {
				target = t // generic forward create — lowest-load claim
				break
			}
		}
		if target == nil {
			break
		}
		claimed := e.state.ClaimPending(target.ID)
		if claimed == nil {
			break
		}
		if claimed.Revoke {
			// 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 marked && e.Log != nil {
				_, _ = e.Log.Append(e.ID, LogForwardStop, map[string]any{
					"taskId": claimed.ID, "local": claimed.Local.Name, "remote": claimed.Remote.Name,
				})
			}
			continue
		}
		if claimed.RemoveNode != "" && claimed.RemoveNode == e.ID {
			// Plan §移除节点: self-remove re-queues own forwards, drops ring
			// position, and the token continues to the original successor.
			// Stop the local workers for every forward we own FIRST (plan:
			// "完全取消一切集群远程转发，停其 worker"), then SelfRemove
			// moves them back to pending for another member to claim.
			if e.Handler != nil {
				for _, te := range e.state.ForwardsOwnedBy(e.ID) {
					t := &Task{ID: te.TaskID, Local: te.Local, Remote: te.Remote, Link: te.Link}
					if err := e.Handler.Revoke(ctx, t); err != nil {
						log.Printf("ring[%s] self-remove revoke %s: %v", e.ID, te.TaskID, err)
					}
				}
			}
			wasLeader := e.state.LeaderID == e.ID
			if succ, ok := e.state.AliveSuccessor(e.ID); ok && succ != e.ID {
				e.removedNext = succ
			}
			e.state.SelfRemove(e.ID)
			e.selfRemoved = true
			// Leader failover: if we were the leader, the ring would run
			// leaderless after our departure — LeaderID would be "" (SelfRemove
			// clears it), no node would call Send (cycle never advances), and
			// WatchLeader can't find AlivePredecessor("") to promote a
			// successor. Designate the captured successor as the new leader
			// so the token carries a valid LeaderID downstream; the successor
			// then calls Send on its turn and the ring keeps cycling.
			if wasLeader && e.removedNext != "" {
				e.state.LeaderID = e.removedNext
				for i := range e.state.Nodes {
					e.state.Nodes[i].IsLeader = e.state.Nodes[i].ID == e.removedNext
				}
				if e.Log != nil {
					_, _ = e.Log.Append(e.ID, LogLeaderChange, map[string]string{"leader": e.removedNext})
				}
			}
			if e.Log != nil {
				_, _ = e.Log.Append(e.ID, LogNodeLeave, map[string]string{"node": e.ID})
			}
			log.Printf("ring[%s] self-removed from cluster (token command %s); next=%s leader=%s",
				e.ID, claimed.ID, e.removedNext, e.state.LeaderID)
			continue
		}
		if claimed.RemoveNode != "" {
			// A remove command whose target is no longer in the ring: the
			// target already self-removed (or was never a member). Drop it
			// here — it MUST NOT fall through to the forward-create path,
			// which would spawn a phantom empty forward (the remove task
			// carries no local/remote/link). This branch also absorbs the
			// command after the injector's localPending merge resurrects a
			// copy once the original has been consumed downstream.
			log.Printf("ring[%s] drop fulfilled remove %s (target %s gone)",
				e.ID, claimed.ID, claimed.RemoveNode)
			continue
		}
		// Defense-in-depth against duplicate claims: if an active topology
		// entry for this forward already exists (owned by us or another
		// node), this task is a stale resurrected copy or a multi-token
		// collision — drop it WITHOUT spawning, so we never end up with an
		// orphaned worker running a forward the topology attributes to a
		// different node. Safe for OfflineReassign: that path deletes the
		// topology entry BEFORE re-queueing, so TopologyOwner returns "" and
		// the legitimate re-claim passes through.
		//
		// 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 {
			// Only reached when this node owns the forward (or it is already gone
			// and we are the fallback consumer — see the selection loop).
			owner := e.state.TopologyOwner(claimed)
			e.state.UpdateTopologyDisabled(claimed.Local.Name, claimed.Remote.Name, claimed.Link.RemotePort, false)
			if owner != e.ID {
				// Forward vanished before we got here: nothing to re-enable.
				log.Printf("ring[%s] discard restart %s: no owner", e.ID, claimed.ID)
				continue
			}
			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,
				})
			}
			// Logging is the app's job (RestartFn reports the actual worker
			// outcome); the ring layer stays quiet so one restart is one line
			// rather than two identical ones.
			continue
		}
		if e.Handler != nil {
			if err := e.Handler.Claim(ctx, claimed); err != nil {
				e.state.PendingTasks[claimed.ID] = claimed
				log.Printf("ring[%s] claim %s failed: %v", e.ID, claimed.ID, err)
				break
			}
		}
		if e.Log != nil {
			_, _ = e.Log.Append(e.ID, LogForwardAdd, map[string]any{
				"taskId": claimed.ID, "local": claimed.Local.Name, "remote": claimed.Remote.Name,
			})
		}
		e.state.AddTopology(claimed, e.ID)
		// Refresh our own stored load so the next selfIsLowest check in this
		// same pass sees the incremented Forwards count — otherwise we'd keep
		// claiming (stored load stays stale until we forward the token) and
		// hog every claimable task, defeating lowest-load distribution.
		if i := e.state.Find(e.ID); i >= 0 {
			e.state.Nodes[i].Load = e.loadSnapshot()
		}
	}
	return nil
}

// Forward hands the token to this node's successor over the injected send.
// It is the transport hook used by the HTTP handler after OnToken. If this
// node just self-removed, its ID is no longer in state.Nodes so
// AliveSuccessor would fail — use the original successor captured before
// removal (plan §移除节点 step 3: "令牌传递给自身原本的下一家"). Otherwise
// delegates to forwardToNext which handles send-failure → mark offline →
// reassign → try next hop (plan §故障自幽).
func (e *Engine) Forward(ctx context.Context, tk *Token) error {
	if e.removedNext != "" {
		next := e.removedNext
		e.removedNext = ""
		var err error
		if e.send != nil {
			log.Printf("ring[%s] forward (self-removed) cycle=%d to %s", e.ID, tk.Cycle, next)
			err = e.send(ctx, next, tk)
		}
		// After handing off the token to the old successor, detach to a
		// fresh standalone state so Snapshot() no longer serves the old
		// cluster picture (members, topology, log) after self-leave. The
		// token already carries the published state (with the new leader if
		// we designated one); the local reset does not affect the sent token.
		if e.selfRemoved {
			e.detachAsStandalone()
		}
		return err
	}
	return e.forwardToNext(ctx, tk)
}

// detachAsStandalone resets the engine to a fresh standalone state after a
// self-leave has completed (the token was handed to the old successor). This
// prevents Snapshot() from serving the old cluster picture — other members,
// the full topology, pending tasks, and the cluster log — after the node has
// permanently left the ring. Equivalent to CreateCluster minus the IsMember
// guard (we are already detached) plus a fresh log (old cluster events stale).
func (e *Engine) detachAsStandalone() {
	// Leaving the cluster invalidates this node's admission key — a
	// standalone node has no key until it creates/joins again. Clear both
	// the in-memory key and the persisted copy (so a restart doesn't
	// resurrect a stale key for a node that's no longer in any cluster).
	e.nodeKey = ""
	if e.keyPersist != nil {
		_ = e.keyPersist("")
	}
	// Clear the cached peer list so this node does NOT auto-rejoin on
	// restart — it explicitly left the cluster.
	e.clearPeers()
	e.state = State{
		LeaderID:     e.ID,
		PendingTasks: map[string]*Task{},
		Topology:     map[string]*TopoEntry{},
		RoundDelay:   2 * time.Second,
	}
	e.state.UpsertNode(Node{
		ID: e.ID, Addr: e.myAddr, Alive: true, IsLeader: true,
		Load: e.loadSnapshot(), Version: e.Version, Cache: e.Cache, NodeKey: e.nodeKey,
	})
	e.selfRemoved = false
	e.removedNext = ""
	e.lastRingStart = time.Time{}
	e.lastTokenAt = 0
	e.lastSyncAt = 0
	e.inflight.clear()
	e.failCount = map[string]int{}
	e.published = map[string]struct{}{}
	if e.Log != nil {
		e.Log = NewClusterLog()
		_, _ = e.Log.Append(e.ID, LogLeaderChange, map[string]string{"leader": e.ID})
	}
	log.Printf("ring[%s] detached to standalone after self-leave", e.ID)
}

// Snapshot returns a serializable view of the ring for the frontend.
type RingSnapshot struct {
	SelfID     string       `json:"selfId"`
	LeaderID   string       `json:"leaderId"`
	Cycle      int64        `json:"cycle"`
	LastSync   int64        `json:"lastSync"`
	RoundDelay int64        `json:"roundDelayMs"`
	Nodes      []Node       `json:"nodes"`
	Pending    []*Task      `json:"pending"`
	Topology   []*TopoEntry `json:"topology"`
	Log        []LogEntry   `json:"log,omitempty"`
	// NodeKey: this node's cluster admission key. The cluster page displays it
	// so the operator can copy it for newcomers joining via this node.
	NodeKey string `json:"nodeKey,omitempty"`
}

func (e *Engine) Snapshot() *RingSnapshot {
	snap := &RingSnapshot{
		SelfID:     e.ID,
		LeaderID:   e.state.LeaderID,
		Cycle:      e.state.Cycle,
		LastSync:   e.lastSyncAt,
		RoundDelay: e.state.RoundDelay.Milliseconds(),
		Nodes:      e.state.Nodes,
		Pending:    e.state.PendingList(),
		Topology:   e.state.TopologyList(),
		NodeKey:    e.nodeKey,
	}
	if e.Log != nil {
		snap.Log = e.Log.Snapshot()
	}
	return snap
}

// tkDelaySince measures elapsed since token SentAt (leader round delay).
func tkDelaySince(tk *Token) time.Duration {
	if tk.SentAt == 0 {
		return 2 * time.Second
	}
	return time.Duration(time.Now().UnixMilli()-tk.SentAt) * time.Millisecond
}

func contains(xs []string, v string) bool {
	for _, x := range xs {
		if x == v {
			return true
		}
	}
	return false
}

// StartRing kicks off a fresh token from the leader. The leader stamps a new
// SentAt (so older in-flight tokens are dropped downstream) and marks inflight
// so WatchTokenLoss can detect a lost first hop — previously StartRing sent
// without marking, so a first-round loss was never noticed.
func (e *Engine) StartRing(ctx context.Context) {
	if e.state.LeaderID != e.ID {
		return
	}
	next, ok := e.state.AliveSuccessor(e.ID)
	if !ok || next == e.ID {
		return // single-node ring; resumes once a newcomer joins
	}
	// Throttle: do not start a new cycle until RoundDelay has elapsed.
	if !e.lastRingStart.IsZero() && time.Since(e.lastRingStart) < e.state.RoundDelay {
		return
	}
	e.lastRingStart = time.Now()
	e.state.Cycle++
	tk := &Token{
		Cycle:  e.state.Cycle,
		State:  e.state,
		SentAt: time.Now().UnixMilli(),
	}
	if e.send != nil {
		if err := e.send(ctx, next, tk); err != nil {
			log.Printf("ring[%s] start cycle %d send: %v", e.ID, tk.Cycle, err)
			return
		}
		// Mark inflight ONLY on a successful first send so token-loss
		// detection covers the very first hop too.
		e.inflight.mark(e.state.RoundDelay)
	}
}

// JoinInfo is a node's self-description sent when requesting to join a ring.
type JoinInfo struct {
	ID      string   `json:"id"`
	Addr    string   `json:"addr"`
	Version string   `json:"version,omitempty"`
	Cache   []string `json:"cache,omitempty"`
	// JoinKey is the sponsor's nodeKey — the newcomer must present it to
	// prove it is authorized to join via the sponsor. The sponsor verifies
	// ji.JoinKey == e.nodeKey; mismatch → 403.
	JoinKey string `json:"joinKey,omitempty"`
}

// injectPendingJoin writes every queued newcomer into state right after
// ourselves (so the newcomer becomes our successor) and logs node.join. Used
// both by OnToken (plan: "令牌发到自己时把新节点加入令牌") and by JoinNode
// when the leader is a single node with no token circulating.
func (e *Engine) injectPendingJoin() {
	for _, ji := range e.pendingJoin {
		nn := Node{ID: ji.ID, Addr: ji.Addr, Alive: true,
			Load: Load{MemPct: 50, NetPct: 50}, Version: ji.Version, Cache: ji.Cache}
		e.state.InsertAfter(e.ID, nn)
		if e.Log != nil {
			_, _ = e.Log.Append(e.ID, LogNodeJoin, map[string]string{"node": nn.ID, "addr": nn.Addr})
		}
		log.Printf("ring[%s] injected newcomer %s after self", e.ID, nn.ID)
	}
	e.pendingJoin = nil
}

// JoinNode accepts a newcomer's join request. Per plan §新节点加入 the
// sponsor does NOT mutate its own state immediately (it would be washed out
// by the next incoming token's authoritative merge). Instead it queues the
// newcomer on pendingJoin; OnToken injects it into the token right after
// the sponsor. EXCEPTION: a single-node leader has no token circulating
// (StartRing refuses for lack of a successor), so the newcomer would never
// be injected — in that case we inject immediately and kick off the ring.
func (e *Engine) JoinNode(j JoinInfo) *State {
	e.pendingJoin = append(e.pendingJoin, j)
	if e.state.LeaderID == "" {
		e.state.LeaderID = e.ID
	}
	// Single-node leader: no successor, no circulating token → the pending
	// newcomer would never be injected. Inject now (safe: no foreign token
	// can overwrite a single-node leader) and start the ring.
	if e.state.LeaderID == e.ID && len(e.state.Nodes) == 1 {
		e.injectPendingJoin()
		go e.StartRing(context.Background())
	}
	return &e.state
}

// AdoptState replaces this node's cluster picture with the state provided by
// the join target, then re-inserts us (in case we were absent).
func (e *Engine) AdoptState(s State) {
	ns := make([]string, 0, len(s.Nodes))
	for _, n := range s.Nodes {
		ns = append(ns, n.ID)
	}
	log.Printf("ring[%s] adopt-state nodes=%v", e.ID, ns)
	kept := map[string]*Task{}
	for id, t := range e.state.PendingTasks {
		if _, ok := s.PendingTasks[id]; !ok {
			kept[id] = t
		}
	}
	e.state = s
	if e.state.PendingTasks == nil {
		e.state.PendingTasks = map[string]*Task{}
	}
	for id, t := range kept {
		e.state.PendingTasks[id] = t
	}
	// Re-assert decisions this node made locally that the ring has not yet
	// confirmed, BEFORE the host's topology sync runs: the sync reads
	// TopologyDisabled to learn the cluster's view, so the local decision must
	// already be applied or a stop could be reported as "enabled" and dropped.
	e.reconcileLocalDisabled()
	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 {
		e.Log.Append(e.ID, LogNodeJoin, map[string]string{"node": e.ID, "addr": e.myAddr})
	}
	// Newcomer generates its own admission key after joining, so future
	// nodes can join via it. Per the user's design: "每个节点加入集群后
	// 生成自身密钥". A node that already has a persisted key (restart
	// re-join) keeps it.
	e.ensureNodeKey()
	// Persist the cached peer list from the adopted ring state.
	e.persistPeers()
}

// CreateCluster reseeds this node as a fresh standalone leader (single-node
// ring). Used after a self-leave left the ring empty, or to (re)affirm seed
// state on a standalone node. Refuses if this node is still a multi-node
// member — reseeding mid-cluster would split the ring (split-brain). Safe to
// call when standalone/empty: no token circulates to a non-member, so no
// concurrent OnToken can overwrite the reset.
func (e *Engine) CreateCluster() error {
	if e.IsMember() {
		return fmt.Errorf("node is a multi-node cluster member; leave first")
	}
	e.ensureNodeKey()
	e.state = State{
		LeaderID:     e.ID,
		Cycle:        0,
		PendingTasks: map[string]*Task{},
		Topology:     map[string]*TopoEntry{},
		RoundDelay:   2 * time.Second,
	}
	e.state.UpsertNode(Node{
		ID: e.ID, Addr: e.myAddr, Alive: true, IsLeader: true,
		Load: e.loadSnapshot(), Version: e.Version, Cache: e.Cache, NodeKey: e.nodeKey,
	})
	e.selfRemoved = false
	e.removedNext = ""
	e.lastRingStart = time.Time{}
	e.lastTokenAt = 0
	e.inflight.clear()
	if e.Log != nil {
		_, _ = e.Log.Append(e.ID, LogLeaderChange, map[string]string{"leader": e.ID})
	}
	log.Printf("ring[%s] created/seeded fresh standalone cluster as leader", e.ID)
	return nil
}

// ResetTopology clears all established topology entries (pending tasks are
// kept). Used by the leader's startup reconcile: the recovered topology
// records forwards whose worker processes died with the restart, and HasTask
// would short-circuit re-submission — so the leader wipes topology first,
// then re-submits every forward to ride the token fresh.
func (e *Engine) ResetTopology() {
	e.state.Topology = map[string]*TopoEntry{}
}

// SubmitTask adds a new forward request to pending; it rides the next token
// round and is claimed by the lowest-load member.
// RemoveNode publishes a node-removal command via the token; the target
// node self-removes when the command reaches it.
func (e *Engine) RemoveNode(nodeID string) *Task {
	return e.state.AddRemoveNode(nodeID)
}

// RemoveTopologyEntry drops the active topology entry for a forward identified
// by its natural key. Exposed so the app's claim path can retire an entry it
// refuses to serve (see ClaimFn: a disabled forward must stop looking active,
// otherwise the startup reconcile keeps re-claiming it on every restart).
func (e *Engine) RemoveTopologyEntry(local, remote string, port int) bool {
	return e.state.RemoveTopology(local, remote, port)
}

// RevokeTask publishes a revocation for an established forward through the
// same token channel; the owning node stops the worker and drops topology.
func (e *Engine) RevokeTask(local store.Local, remote store.Remote, link store.Link) *Task {
	return e.state.AddRevoke(local, remote, link)
}

func (e *Engine) SubmitTask(local store.Local, remote store.Remote, link store.Link) *Task {
	// 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)
}

// 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 {
			return true
		}
	}
	for _, t := range e.state.TopologyList() {
		if t.Local.Name == local && t.Remote.Name == remote && t.Link.RemotePort == port {
			return true
		}
	}
	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)
}

// 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.
//
// The decision is also remembered locally until the ring confirms it, because
// adoption replaces the whole state: a token captured before this call would
// otherwise wash the decision out on the next cycle and it would never reach
// the other members. See the localDisabled field.
func (e *Engine) UpdateTopologyDisabled(local, remote string, port int, disabled bool) bool {
	if e.localDisabled == nil {
		e.localDisabled = map[string]bool{}
	}
	e.localDisabled[disableKey(local, remote, port)] = disabled
	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).
//
// A pending local decision takes precedence over the adopted state: it has not
// had a chance to reach the other members yet, and reporting the adopted value
// would make the local store (and the status page) disagree with the user's
// most recent action.
func (e *Engine) TopologyDisabled(local, remote string, port int) (bool, bool) {
	if d, pending := e.localDisabled[disableKey(local, remote, port)]; pending {
		return d, true
	}
	return e.state.TopologyDisabled(local, remote, port)
}

// reconcileLocalDisabled re-asserts this node's not-yet-confirmed disabled
// decisions onto the freshly adopted state. Called right after e.state is
// replaced, and paired with reconcileTopologySync (which pushes those decisions
// out to the ring).
//
// A decision is dropped once the adopted state reports the SAME value: at that
// point every member agrees and there is nothing left to protect. Entries whose
// forward no longer exists in the topology are dropped too, so the map tracks
// only live disagreements.
func (e *Engine) reconcileLocalDisabled() {
	if len(e.localDisabled) == 0 {
		return
	}
	for id, te := range e.state.Topology {
		_ = id
		k := disableKey(te.Local.Name, te.Remote.Name, te.Link.RemotePort)
		want, pending := e.localDisabled[k]
		if !pending {
			continue
		}
		if te.Link.Disabled == want {
			// The ring caught up (or agreed independently): stop tracking.
			delete(e.localDisabled, k)
			continue
		}
		// Still divergent: keep asserting our decision onto the adopted state.
		te.Link.Disabled = want
		te.Active = !want
	}
	// Forget decisions for forwards that left the topology entirely — otherwise
	// a long-lived cluster would accumulate dead keys.
	live := make(map[string]struct{}, len(e.state.Topology))
	for _, te := range e.state.Topology {
		live[disableKey(te.Local.Name, te.Remote.Name, te.Link.RemotePort)] = struct{}{}
	}
	for k := range e.localDisabled {
		if _, ok := live[k]; !ok {
			delete(e.localDisabled, k)
		}
	}
}

// IsLeader reports whether this node is the current ring leader.
func (e *Engine) IsLeader() bool { return e.state.LeaderID == e.ID }

// NodeKey returns this node's cluster admission key (for the frontend to
// display so the operator can copy it for newcomers).
func (e *Engine) NodeKey() string { return e.nodeKey }

// SetKeyPersist installs the callback used to persist the nodeKey to durable
// storage (store.SetNodeKey). Called once from main.go after NewEngine. Tests
// leave it nil — ensureNodeKey still generates the key in-memory.
func (e *Engine) SetKeyPersist(fn func(key string) error) { e.keyPersist = fn }

// SetPeerPersist installs the callback used to persist the cached peer list
// 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 }

// disableKey builds the map key for a forward's disabled decision: the natural
// triple (local, remote, remotePort), which is how every other part of the code
// identifies a forward. A NUL separator keeps it unambiguous for names that
// could otherwise collide across the boundaries.
func disableKey(local, remote string, port int) string {
	return local + "\x00" + remote + "\x00" + strconv.Itoa(port)
}

// 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
// always has the latest peer list to rejoin through. Skipped for standalone
// (single-node) rings — a standalone node has no peers to cache.
func (e *Engine) persistPeers() {
	if e.peerPersist == nil {
		return
	}
	type peerEntry struct {
		Addr string `json:"addr"`
		Key  string `json:"key"`
	}
	var peers []peerEntry
	for _, n := range e.state.Nodes {
		if n.ID == e.ID || !n.Alive {
			continue
		}
		if n.Addr == "" || n.NodeKey == "" {
			continue
		}
		peers = append(peers, peerEntry{Addr: n.Addr, Key: n.NodeKey})
	}
	if len(peers) == 0 {
		return // standalone or all-offline: don't overwrite a good cache
	}
	data, err := json.Marshal(peers)
	if err != nil {
		return
	}
	if err := e.peerPersist(string(data)); err != nil {
		log.Printf("ring[%s] persist cluster peers failed: %v", e.ID, err)
	}
}

// clearPeers wipes the cached peer list (called from detachAsStandalone so
// an explicit leave does NOT auto-rejoin on restart).
func (e *Engine) clearPeers() {
	if e.peerPersist != nil {
		_ = e.peerPersist("")
	}
}

// ensureNodeKey generates a random admission key if this node doesn't have one
// yet, and persists it via the keyPersist callback (so it survives restarts).
// Called from CreateCluster (the creator generates a key so others can join
// via it) and AdoptState (a newcomer generates its own key after joining, so
// future nodes can join via it). Per the user's design: "每个节点加入集群后
// 生成自身密钥" — the key is born with cluster membership, not at startup.
func (e *Engine) ensureNodeKey() {
	if e.nodeKey != "" {
		return
	}
	e.nodeKey = GenerateNodeKey()
	if e.keyPersist != nil {
		if err := e.keyPersist(e.nodeKey); err != nil {
			log.Printf("ring[%s] persist nodeKey failed: %v", e.ID, err)
		}
	}
}

// IsMember reports whether this node is currently an active multi-node member
// (self is in the ring alongside others). Used by the create/join gates to
// refuse actions that would split an active ring. A detached node (self not
// in ring — e.g. after a self-leave) or a standalone node returns false and
// may create/join freely. NOTE: after a self-leave the engine keeps the other
// members in state.Nodes (it only dropped self), so a plain len>1 check would
// wrongly block a detached node — the self-in-ring test is essential.
func (e *Engine) IsMember() bool {
	if len(e.state.Nodes) <= 1 {
		return false
	}
	for _, n := range e.state.Nodes {
		if n.ID == e.ID {
			return true
		}
	}
	return false
}

// LeaderAddr returns the current leader's address.
func (e *Engine) LeaderAddr() string {
	i := e.state.Find(e.state.LeaderID)
	if i < 0 {
		return ""
	}
	return e.state.Nodes[i].Addr
}