fix(cluster): 重入成功后若自己是 leader 必须启动令牌环,否则集群冻结

上一提交(50c7729)部署后环**仍卡死**,实测 cycle 停在 458198、topology=0、
pending=3。查 .30 日志定位到第二个、也是真正让集群不动的原因:

    ring[.30] rejoined cluster via cached peer 192.168.2.60:7500
    ring[.30] startup reconcile: reclaimed=0 submitted=3
    (此后再无任何 forward cycle / token-send —— 令牌从未转过)

## 根因

retryRejoinCached 在 JoinRing 成功后**直接 return**,把"环会继续转"当作前提。
但 StartRing 只对 leader 生效,而滚动重启恰恰让 leader 自己重入(leader 最后
重启)。此时没有任何节点会转令牌 —— 非 leader 的 StartRing 是 no-op。

结果:leader 重入成功却从不发令牌 → cycle 冻结 → 那三条 pending 转发永远无人
认领 → topology 一直是 0。这与设计文稿 §容错行为不符(环应继续推进)。

## 修法

重入成功后调用 `go ring.StartRing(ctx)`。对非 leader 是 no-op,且内部按
RoundDelay 自节流,多调不会重发,故可无条件调用。

## 测试

TestRejoinThenStartRingCoversTheRejoinPath:用 httptest 起一个假 peer,端到端跑
真正的 retryRejoinCached,断言重入成功后 leader 确实发出了令牌。

★ 这条测试重写过两次才有效:
1. 第一版直接调 `eng.StartRing(...)` —— **假绿**,把修复自己的 helper 当成被测
   对象,回退修复行后仍然全绿(已实测确认)。
2. 第二版改走 retryRejoinCached,但传入 httptest 的完整 URL;JoinRing 自己拼
   "http://<addr>/api/manager/cluster/join",于是地址变成
   `http://http://127.0.0.1:PORT/...` 而永远打不通。改为裸 host:port。

双向验证:移除修复行后如期变红
("after a successful rejoin the leader emitted NO token: the ring is frozen"),
还原后变绿。

go build / go vet / go test ./... 全绿,gofmt 干净。
This commit is contained in:
JianFeeeee
2026-10-05 17:34:57 +08:00
parent 2dd32648e6
commit bfd7265405
2 changed files with 159 additions and 0 deletions

View File

@ -752,6 +752,20 @@ func retryRejoinCached(ctx context.Context, ring *cluster.Engine, peersJSON stri
cancel()
if err == nil {
log.Printf("ring[%s] rejoined cluster via cached peer %s", ring.ID, p.Addr)
// A rejoined node must be able to drive the ring itself. The
// normal case is that the existing leader keeps circulating the
// token, but if WE are the leader (typical after a rolling
// restart where the leader came back last, or the leader died
// while we were down) nobody else will: StartRing is a no-op on
// non-leaders, and the ring sits still with every task stranded
// in pending. Observed live: the leader restarted, rejoined fine,
// and then never forwarded a token — cycle frozen, topology
// empty, pending=3.
//
// StartRing is safe to call unconditionally: it returns
// immediately unless this node leads, and it throttles itself
// against RoundDelay so a redundant call cannot double-send.
go ring.StartRing(ctx)
return
}
log.Printf("ring[%s] rejoin %s failed: %v", ring.ID, p.Addr, err)

View File

@ -0,0 +1,145 @@
package main
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"strings"
"sync"
"testing"
"time"
"webui4frpc/internal/cluster"
"webui4frpc/internal/store"
)
// fakeJoinPeer stands in for a healthy cluster member: it answers the rejoin
// POST and records it, so the rejoin path can be exercised end to end.
type fakeJoinPeer struct {
mu sync.Mutex
joined int
stateFn func() cluster.State
}
func (p *fakeJoinPeer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
p.mu.Lock()
p.joined++
p.mu.Unlock()
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(map[string]any{"state": p.stateFn()})
}
func (p *fakeJoinPeer) joins() int {
p.mu.Lock()
defer p.mu.Unlock()
return p.joined
}
// TestRejoinThenStartRingCoversTheRejoinPath is the regression test for the
// frozen ring, and it deliberately drives retryRejoinCached itself rather than
// calling StartRing directly — a test that calls the fix's own helper proves
// nothing about whether the fix is wired up (verified: such a test stayed green
// with the fix reverted).
//
// The failure, observed live on the three-node cluster: the leader restarted,
// rejoined successfully, logged "startup reconcile: reclaimed=0 submitted=3",
// and then never forwarded a token. cycle froze, topology stayed empty and the
// three forwards sat in pending forever, because rejoin returned immediately
// and only a leader may call StartRing.
func TestRejoinThenStartRingCoversTheRejoinPath(t *testing.T) {
selfID := "192.168.2.30:7500"
// The peer hands back a ring in which SELF is the leader — the situation a
// rolling restart produces when the leader comes back last.
var peerState cluster.State
peer := &fakeJoinPeer{stateFn: func() cluster.State { return peerState }}
// Bind on loopback so the engine's hardcoded peer address is reachable.
// We therefore advertise 127.0.0.1 in the ring while keeping the same IDs.
ln := httptest.NewServer(peer)
defer ln.Close()
peerState = cluster.State{
LeaderID: selfID,
Nodes: []cluster.Node{
{ID: selfID, Addr: selfID, Alive: true, IsLeader: true},
{ID: "peer:7500", Addr: "peer:7500", Alive: true},
},
Topology: map[string]*cluster.TopoEntry{},
PendingTasks: map[string]*cluster.Task{
"t1": {ID: "t1", Local: store.Local{Name: "web"},
Remote: store.Remote{Name: "frps1", IP: "1.2.3.4", Port: 7000, Enabled: true},
Link: store.Link{RemotePort: 18081}},
},
}
// Redirect the peer's advertised address at the transport level: the engine
// dials whatever the ring says, so give the ring the test server's address.
// JoinRing builds "http://<addr>/api/manager/cluster/join" itself, so the
// cached entry must be a bare host:port, not a URL.
realAddr := strings.TrimPrefix(ln.URL, "http://")
for i := range peerState.Nodes {
if peerState.Nodes[i].ID == "peer:7500" {
peerState.Nodes[i].Addr = realAddr
}
}
var mu sync.Mutex
var emitted int
eng := cluster.NewEngine(selfID, selfID, "u", "p", "0.1.2", nil,
&cluster.AppHandler{ClaimFn: func(ctx context.Context, tk *cluster.Task) error { return nil }},
func(ctx context.Context, next string, tk *cluster.Token) error {
if tk != nil {
mu.Lock()
emitted++
mu.Unlock()
}
return nil
},
selfID, true, "")
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
// The rejoin path, as main() runs it. Cached peers point at the test server.
peersJSON, err := json.Marshal([]map[string]string{{"addr": realAddr, "key": "peerkey"}})
if err != nil {
t.Fatal(err)
}
done := make(chan struct{})
go func() {
defer close(done)
retryRejoinCached(ctx, eng, string(peersJSON))
}()
// retryRejoinCached returns once rejoin succeeds; give the goroutine a moment.
for i := 0; i < 100; i++ {
if peer.joins() > 0 {
break
}
sleepShort()
}
if peer.joins() == 0 {
t.Fatal("the rejoin never reached the peer — the test setup is wrong")
}
<-done
// Rejoin succeeded. The engine must now consider itself the leader...
if !eng.IsLeader() {
t.Fatalf("precondition broken: rejoined node does not consider itself leader")
}
sleepShort()
// ...and it must have restarted the ring: a token has to be emitted, or the
// cluster sits frozen with its forwards stranded in pending.
mu.Lock()
got := emitted
mu.Unlock()
if got == 0 {
t.Fatal("after a successful rejoin the leader emitted NO token: the ring is frozen " +
"and pending forwards are never claimed (the live failure this guards)")
}
}
func sleepShort() { time.Sleep(10 * time.Millisecond) }