// Leader-side minimal roles for the token ring. Per the authoritative design: // the leader only (1) initiates the first token, and (2) judges token loss. // All other processing (adopt cluster picture, append own info, run commands, // forward) is IDENTICAL between leader and ordinary nodes — no phase flips, // no passed-accounting, no cycle bookkeeping in the leader. package cluster import ( "context" "log" "sync" "time" ) // ringHopDelay paces each token hop (real-world cadence, avoids busy-loop). // 500ms keeps a 3-node ring well under the LossTimeout floor (2500ms) so a // healthy round is never misjudged lost, while not idly burning CPU/HTTP at // 20Hz like the old 50ms did. const ringHopDelay = 500 * time.Millisecond // LossTimeout is the token-loss threshold per design: roundDelay/2 + 20ms, // floored so a healthy fast ring is never misjudged. func LossTimeout(roundDelay time.Duration) time.Duration { t := roundDelay/2 + 20*time.Millisecond if t < 2500*time.Millisecond { return 2500 * time.Millisecond } return t } // inFlight tracks token-in-flight state (leader only, for loss judging). type inFlight struct { mu sync.Mutex active bool sentAt time.Time delay time.Duration } func (f *inFlight) mark(delay time.Duration) { f.mu.Lock() f.active = true f.sentAt = time.Now() f.delay = delay f.mu.Unlock() } func (f *inFlight) clear() { f.mu.Lock() f.active = false f.mu.Unlock() } func (f *inFlight) inflight() bool { f.mu.Lock() defer f.mu.Unlock() return f.active } func (f *inFlight) age() time.Duration { f.mu.Lock() defer f.mu.Unlock() if !f.active { return 0 } return time.Since(f.sentAt) } func (f *inFlight) lastDelay() time.Duration { f.mu.Lock() defer f.mu.Unlock() return f.delay } // Send is called after OnToken for the LEADER. The token has come full // circle (one round complete), so the leader bumps the cycle, stamps a fresh // SentAt (so any stale in-flight older token is dropped downstream), and // forwards to the successor to start the next round. This is what makes // `cycle` advance under the perpetual-flow model — without it cycle stuck // at the StartRing value forever. func (e *Engine) Send(ctx context.Context, tk *Token) error { if !e.selfRemoved { e.state.Cycle = tk.Cycle + 1 tk.Cycle = e.state.Cycle // Keep State.Cycle in lockstep with Token.Cycle so the snapshot // (which reads e.state.Cycle) reflects the real round number. tk.State = e.state tk.SentAt = time.Now().UnixMilli() if tk.SentAt > e.lastTokenAt { e.lastTokenAt = tk.SentAt } } return e.forwardToNext(ctx, tk) } // forwardToNext sends the token to the next alive successor; on failure it // marks that node offline, reattaches its tasks, and tries the next hop. func (e *Engine) forwardToNext(ctx context.Context, tk *Token) error { for hops := 0; hops < len(e.state.Nodes); hops++ { next, ok := e.nextRecipient() if !ok { e.inflight.clear() return nil } if e.send == nil { return nil } err := e.send(ctx, next, tk) if err == nil { if e.state.LeaderID == e.ID { e.inflight.mark(e.state.RoundDelay) } // Pace the ring so tokens circulate at a realistic cadence. // Pacing handled by the parallel rhythm timer in OnToken. return nil } log.Printf("ring[%s] send to %s failed: %v", e.ID, next, err) // Do not declare a neighbor offline on a single timeout: transient // send failures (network jitter, busy handler) must not break the // ring. Only after consecutive failures do we evict the node. e.failMu.Lock() e.failCount[next]++ if e.failCount[next] >= 2 { e.state.MarkOffline(next) e.state.OfflineReassign(next) delete(e.failCount, next) } } e.inflight.clear() return nil } // becomeLeader promotes this node (used when the monitored leader dies). func (e *Engine) becomeLeader() { e.state.LeaderID = e.ID for i := range e.state.Nodes { e.state.Nodes[i].IsLeader = e.state.Nodes[i].ID == e.ID } log.Printf("ring[%s] promoted to leader", e.ID) } // WatchLeader runs the leader liveness monitor: the predecessor of the leader // pings it; on failure it marks the leader offline and promotes itself. func (e *Engine) WatchLeader(ctx context.Context) { tick := time.NewTicker(2 * time.Second) defer tick.Stop() for { select { case <-ctx.Done(): return case <-tick.C: if e.state.LeaderID == e.ID { continue // we are the leader; predecessor monitors us } leaderPred, ok := e.state.AlivePredecessor(e.state.LeaderID) if !ok || leaderPred != e.ID { continue // only the leader's predecessor pings it } if e.send != nil { hbCtx, cancel := context.WithTimeout(ctx, 1500*time.Millisecond) err := e.send(hbCtx, e.state.LeaderID, nil) // nil = heartbeat cancel() if err != nil { log.Printf("ring[%s] heartbeat to leader %s failed: %v", e.ID, e.state.LeaderID, err) e.state.MarkOffline(e.state.LeaderID) e.state.OfflineReassign(e.state.LeaderID) e.becomeLeader() } } } } } // WatchTokenLoss runs the leader's token-loss judge: if a token was sent and // does not return within LossTimeout, the leader issues a fresh token (all // nodes drop older stamps via the SentAt guard, so at most one circulates). func (e *Engine) WatchTokenLoss(ctx context.Context) { tick := time.NewTicker(200 * time.Millisecond) defer tick.Stop() for { select { case <-ctx.Done(): return case <-tick.C: if e.ID != e.state.LeaderID { continue } if !e.inflight.inflight() { continue } delay := e.inflight.lastDelay() if delay <= 0 { delay = 200 * time.Millisecond } if e.inflight.age() > LossTimeout(delay) { log.Printf("ring[%s] token lost, resending cycle %d", e.ID, e.state.Cycle) e.inflight.clear() e.StartRing(ctx) } } } }