fix(cluster): startup reconcile 不该等 leader，否则非 leader 永远不重建自己的 worker

前两个提交修好分裂与冻结后，滚动重启实测又暴露第三个缺口：

    .106 重入成功 → adopt-state（topology 里 .106 是 gitea_ssh 的 owner）
                  → 但 .106 没有任何 frpc worker，且日志里从无 reconcile

拓扑说 .106 拥有该转发，进程却不存在 —— 拓扑与实际不一致。

## 根因：代码与自己注释矛盾

启动 reconcile 的注释原文就是「every node (leader or not) checks its OWN topology
entries and re-claims any whose per-forward frpc worker is absent」，但代码第一行是

    for { if ring.IsLeader() { break }; sleep 2s }

即必须先当上 leader 才执行。reconcile 做的是**本节点**的活（重建自己拥有的
worker），而三分之一个环里每台都拥有转发 —— 于是所有非 leader 节点永远等不到，
worker 永远不重建。节点可以「已回到集群、拓扑里写着它 own 某转发、却没有任何进程」
这种自相矛盾的状态。

## 修法

新增 Engine.HasSelfInTopology()：当前拓扑是否指名本节点为某条 active 转发
的 owner。reconcile 改等这个条件（30s 上限后无论如何继续，保证孤立节点也能推进），
不再等 leadership。领导者与 reconcile 本就无关：follower 也 own 转发。

## 测试

TestHasSelfInTopology：用**两节点**环，转发由非 leader 的 n2 持有，断言非 leader
仍然报告 true。

★ 这条测试第一版是**假绿**：单节点环里唯一的成员必然是 leader，两种实现结果相同，
加回 IsLeader 门控后测试仍全绿（已实测）。改成两节点、明确断言
`precondition: n2 must not be the leader` 之后，加回门控如期变红：

    a non-leader that owns a forward reports false — the startup reconcile would
    skip it and leave its worker down

go build / go vet / go test ./... 全绿，gofmt 干净。

cmd/webui4frpc/main.go

@@ -561,17 +561,32 @@ func main() {
 	// LocalOnly forwards are never in the topology and are handled by the
 	// auto-start loop above.
 	go func() {
-		for {
-			if ring.IsLeader() {
-				break
-			}
+		// No leader gate here. The reconcile body is per-NODE work: it re-claims
+		// the forwards THIS node owns, whose worker processes died with the
+		// restart. Every member owns forwards (a 3-node ring splits them), so
+		// gating on leadership meant non-leaders waited forever and never
+		// rebuilt their own workers — a node could come back up "in the cluster"
+		// while owning a forward with no process behind it. Observed live: after a
+		// rolling restart a non-leader rejoined, adopted state listing itself as
+		// the owner of a forward, and never started it, because it sat waiting to
+		// become leader.
+		//
+		// What actually needs settling is the ring, not the leadership: wait for
+		// the topology to show us (with a bounded fallback so a lone node that
+		// rejoined a ring whose leader is unreachable still makes progress).
+		deadline := time.After(30 * time.Second)
+		for !ring.HasSelfInTopology() {
 			select {
-			case <-time.After(2 * time.Second):
+			case <-deadline:
+				log.Printf("ring[%s] startup reconcile: self not in topology after 30s, proceeding anyway", ring.ID)
+			case <-time.After(500 * time.Millisecond):
+				continue
 			case <-clusterCtx.Done():
 				return
 			}
+			break
 		}
-		// Let the ring stabilize after leader election / rejoin.
+		// Let the ring stabilize after election / rejoin before acting on it.
 		time.Sleep(3 * time.Second)
 		snap := ring.Snapshot()
 		reclaimed := 0

internal/cluster/ring_disabled_test.go

@@ -155,3 +155,41 @@ func TestLocalDecisionsDoNotAccumulate(t *testing.T) {
 		t.Fatalf("decisions for forwards outside the topology must be swept, got %v", eng.localDisabled)
 	}
 }
+
+// TestHasSelfInTopology is the wait condition for the startup reconcile: the
+// reconcile rebuilds the workers for forwards THIS node owns, so a node can only
+// act once the adopted state actually names it as an owner.
+//
+// It is deliberately independent of leadership — the bug it guards against was a
+// reconcile that waited for leadership and therefore never ran on followers, even
+// though followers own forwards.
+func TestHasSelfInTopology(t *testing.T) {
+	// Two nodes; the forward is owned by n2, which is NOT the leader. A single-node
+	// ring would make this test pass under either implementation, because the only
+	// member is necessarily the leader — which is exactly how the first version of
+	// this test came to be a false green.
+	eng := newTestEngine("n2", false)
+	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, SentAt: 1}); err != nil {
+		t.Fatal(err)
+	}
+	if eng.IsLeader() {
+		t.Fatal("precondition: n2 must not be the leader in this test")
+	}
+	// n2 claimed it, so it owns a forward — despite not leading.
+	if !eng.HasSelfInTopology() {
+		t.Error("a non-leader that owns a forward reports false — the startup reconcile " +
+			"would skip it and leave its worker down")
+	}
+
+	// A node that owns nothing must report false (so it does not act early on a
+	// stale or empty topology).
+	other := newTestEngine("n9", true)
+	if other.HasSelfInTopology() {
+		t.Error("a node owning nothing must not claim it appears in the topology")
+	}
+	// Guard against a regression where leadership is folded into the predicate.
+	if !eng.HasSelfInTopology() && eng.IsLeader() {
+		t.Error("predicate must not be gated on leadership")
+	}
+}

internal/cluster/ring_engine.go

@@ -1206,3 +1206,20 @@ func (e *Engine) LeaderAddr() string {
 	}
 	return e.state.Nodes[i].Addr
 }
+
+// HasSelfInTopology reports whether the current topology names this node as the
+// owner of at least one active forward.
+//
+// The startup reconcile uses it as its wait condition: a node can only rebuild
+// the workers for forwards it owns once the adopted state says it owns them.
+// Leadership is deliberately NOT part of the test — followers own forwards too,
+// and gating the reconcile on leadership left them permanently unable to restore
+// their own workers.
+func (e *Engine) HasSelfInTopology() bool {
+	for _, te := range e.state.TopologyList() {
+		if te.OwnerID == e.ID && te.Active {
+			return true
+		}
+	}
+	return false
+}