webui4frpc/internal/cluster/ring_leader.go

// 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, floored at 600ms so a healthy fast ring is never
// misjudged as lost.
func LossTimeout(roundDelay time.Duration) time.Duration {
	t := roundDelay/2 + 20*time.Millisecond
	if t < 600*time.Millisecond {
		return 600 * time.Millisecond
	}
	return t
}

// 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 completes the cycle. Only the
// leader flips. Passed carries the set of nodes that saw this token this round;
// it is NOT reset between phases (only between cycles).
func (e *Engine) flipIfComplete(tk *Token) {
	if e.state.LeaderID != e.ID {
		return
	}
	switch tk.Phase {
	case PhaseCollect:
		if allAlivePassed(&e.state, tk.Passed) {
			// Round 1 done: go to round 2 (sync). Round-2 stamping restarts
			// from the leader so the token walks the ring once more.
			tk.Phase = PhaseSync
			tk.Passed = []string{e.ID}
			e.state.RoundDelay = tkDelaySince(tk)
		}
	case PhaseSync:
		if allAlivePassed(&e.state, tk.Passed) {
			// Cycle done: leader holds token; next StartRing bumps 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 || succ == e.ID {
				continue // single-node ring: nothing to monitor
			}
			// Only the leader's predecessor (the previous alive node) monitors
			// it; a node whose successor is the leader is the predecessor.
			if e.state.LeaderID != succ {
				continue
			}
			pred, ok := e.state.AlivePredecessor(e.ID)
			if !ok || pred == e.ID {
				continue
			}
			// We monitor the leader only if we ARE its predecessor.
			if succ != e.state.LeaderID {
				continue
			}
			// pred must be us: leader's predecessor sends the heartbeat.
			if e.state.Nodes[e.state.Find(succ)].ID == e.ID {
				continue
			}
			if e.state.Find(e.ID) == -1 {
				continue
			}
			// This node is the leader's predecessor iff leader's predecessor
			// (alive) equals our ID.
			leaderPred, ok := e.state.AlivePredecessor(e.state.LeaderID)
			if !ok || leaderPred != e.ID {
				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)
			}
		}
	}
}