diff --git a/cmd/webui4frpc/main.go b/cmd/webui4frpc/main.go index 55be602..a014254 100644 --- a/cmd/webui4frpc/main.go +++ b/cmd/webui4frpc/main.go @@ -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) diff --git a/cmd/webui4frpc/rejoin_ring_test.go b/cmd/webui4frpc/rejoin_ring_test.go new file mode 100644 index 0000000..9864170 --- /dev/null +++ b/cmd/webui4frpc/rejoin_ring_test.go @@ -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:///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) }