feat(M6): token ring data plane + engine — pending adds disappear on claim, topology written back for full cluster view (per authoritative design)

internal/cluster/ring_leader.go

@@ -0,0 +1,213 @@
+// Leader-side cycle control + fault tolerance for the token ring:
+//   - phase flip (round1 -> round2 -> next cycle) decided by the leader when
+//     the token returns with all alive nodes passed;
+//   - token-loss resend when it does not come back within roundDelay/2 + 20ms;
+//   - predecessor monitors the leader by heartbeat and promotes itself when the
+//     leader is gone;
+//   - send failures (successor down) mark the neighbor offline, reattach its
+//     tasks, and hop to the next alive successor.
+package cluster
+
+import (
+	"context"
+	"log"
+	"sync"
+	"time"
+)
+
+// LossTimeout returns the token-loss threshold per the authoritative design:
+// roundDelay/2 + 20ms.
+func LossTimeout(roundDelay time.Duration) time.Duration {
+	return roundDelay/2 + 20*time.Millisecond
+}
+
+// inFlight tracks token-in-flight state (leader only).
+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
+}
+
+// flipIfComplete promotes the collect round to the sync round when every alive
+// node has been stamped; at the end of sync it starts a fresh cycle. Only the
+// leader flips.
+func (e *Engine) flipIfComplete(tk *Token) {
+	if e.state.LeaderID != e.ID {
+		return
+	}
+	switch tk.Phase {
+	case PhaseCollect:
+		if allAlivePassed(&e.state, tk.Passed) {
+			tk.Phase = PhaseSync
+			tk.Passed = nil
+			e.state.RoundDelay = tkDelaySince(tk)
+		}
+	case PhaseSync:
+		if allAlivePassed(&e.state, tk.Passed) {
+			// cycle done: start next collect round.
+			e.state.Cycle++
+			e.state.RoundDelay = tkDelaySince(tk)
+		}
+	}
+}
+
+func allAlivePassed(s *State, passed []string) bool {
+	for _, n := range s.Nodes {
+		if !n.Alive {
+			continue
+		}
+		if !contains(passed, n.ID) {
+			return false
+		}
+	}
+	return true
+}
+
+// Send moves the token onward (or completes the cycle at the leader). It is
+// called by the HTTP handler after OnToken. A full cycle stops at the leader.
+func (e *Engine) Send(ctx context.Context, tk *Token) error {
+	if e.state.LeaderID == e.ID {
+		e.flipIfComplete(tk)
+		// If the sync round completed, the leader holds the token; watchdog
+		// StartRing launches the next cycle. Clear in-flight marker.
+		if tk.Phase == PhaseSync && allAlivePassed(&e.state, tk.Passed) {
+			e.inflight.clear()
+			return nil
+		}
+	}
+	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 {
+			e.inflight.mark(e.state.RoundDelay)
+			return nil
+		}
+		log.Printf("ring[%s] send to %s failed: %v", e.ID, next, err)
+		e.state.MarkOffline(next)
+		e.state.OfflineReassign(next)
+		if e.state.LeaderID == next {
+			e.becomeLeader()
+		}
+	}
+	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 heartbeat monitor: if our successor is the leader, ping
+// it; on failure mark it offline and promote ourselves. Stops on ctx cancel.
+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:
+			succ, ok := e.state.AliveSuccessor(e.ID)
+			if !ok {
+				continue
+			}
+			if e.state.LeaderID != succ {
+				continue
+			}
+			if e.send != nil {
+				hbCtx, cancel := context.WithTimeout(ctx, 1500*time.Millisecond)
+				err := e.send(hbCtx, succ, nil) // nil token = heartbeat ping
+				cancel()
+				if err != nil {
+					log.Printf("ring[%s] heartbeat to leader %s failed: %v", e.ID, succ, err)
+					e.state.MarkOffline(succ)
+					e.state.OfflineReassign(succ)
+					e.becomeLeader()
+				}
+			}
+		}
+	}
+}
+
+// WatchTokenLoss runs on the leader: if a token was sent and does not return
+// within LossTimeout, resend it. Stops on ctx cancel.
+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)
+			}
+		}
+	}
+}