webui4frpc/internal/cluster/ring_engine.go

// Package cluster implements the token-ring cooperative network.
// Authoritative design: plan.md §M6 (令牌环网拓扑).
package cluster

import (
	"context"
	"log"
	"time"

	"webui4frpc/internal/store"
)

// Phase constants for the two-round cycle.
const (
	PhaseCollect = 1 // round 1: append own info to token
	PhaseSync    = 2 // round 2: sync cluster state, claim tasks
)

// Handler is what the engine calls when the node must act on a task
// (create the frpc worker / forward). Injected to avoid import cycle.
type Handler interface {
	Claim(ctx context.Context, tk *Task) error
	RuntimeLoad() Load
}

// Engine drives one member of the token ring.
type Engine struct {
	ID      string
	Addr    string
	User    string
	Pass    string
	Version string
	Cache   []string
	Handler Handler

	state State
	// myAddr maps our Node ID to the address peers dial.
	myAddr string

	// send moves the token to the next node (injected transport).
	send func(ctx context.Context, next string, tk *Token) error

	// inflight tracks token-in-flight state (leader only).
	inflight    inFlight
	Log         *ClusterLog
	lastLogSent int64
	// lastRingStart time of the previous cycle launch (leader throttle).
	lastRingStart time.Time
}

// NewEngine builds the engine; state holds this node as initial leader unless
// a peer list says otherwise (creation node starts the ring).
func NewEngine(id, addr, user, pass, version string, cache []string, h Handler, send func(ctx context.Context, next string, tk *Token) error, selfAddr string, isLeader bool) *Engine {
	e := &Engine{
		ID: id, Addr: addr, User: user, Pass: pass,
		Version: version, Cache: cache, Handler: h,
		state: State{
			LeaderID:     "",
			Cycle:        0,
			PendingTasks: map[string]*Task{},
			Topology:     map[string]*TopoEntry{},
			RoundDelay:   200 * time.Millisecond,
		},
		myAddr: selfAddr,
		send:   send,
		Log:    NewClusterLog(),
	}
	n := Node{ID: id, Addr: selfAddr, Alive: true, IsLeader: isLeader,
		Load: Load{MemPct: 10, NetPct: 10}, Version: version, Cache: cache}
	e.state.UpsertNode(n)
	return e
}

// State returns the engine's current cluster picture.
func (e *Engine) State() *State { return &e.state }

// nextRecipient picks the successor to hand the token to, skipping dead nodes.
func (e *Engine) nextRecipient() (string, bool) {
	return e.state.AliveSuccessor(e.ID)
}

// myNode returns this node's entry from state.
func (e *Engine) myNode() Node {
	i := e.state.Find(e.ID)
	if i < 0 {
		return Node{ID: e.ID, Addr: e.myAddr, Alive: true}
	}
	return e.state.Nodes[i]
}

// loadSnapshot reads our runtime load (mem+net) from the handler.
func (e *Engine) loadSnapshot() Load {
	if e.Handler != nil {
		return e.Handler.RuntimeLoad()
	}
	return Load{MemPct: 20, NetPct: 20}
}

// phase1 appends this node's info to the token (round 1).
func (e *Engine) phase1(tk *Token) {
	e.state.UpsertNode(Node{
		ID: e.ID, Addr: e.myAddr, Alive: true,
		IsLeader: e.state.LeaderID == e.ID || tk.State.LeaderID == e.ID,
		Load:     e.loadSnapshot(),
		Version:  e.Version, Cache: e.Cache,
	})
	// Attach our own log entries not yet seen by the ring (incremental sync):
	// entries after the last forwarded watermark ride the token for others.
	if e.Log != nil {
		mine := e.Log.EntriesAfter(e.lastLogSent)
		if len(mine) > 0 {
			tk.Log = append(tk.Log, mine...)
			if last := mine[len(mine)-1]; last.Seq > e.lastLogSent {
				e.lastLogSent = last.Seq
			}
		}
	}
	tk.Passed = append(tk.Passed, e.ID)
}

// phase2 syncs cluster info from the token and claims pending tasks if we are
// the lowest-load node. A claimed task DISAPPEARS from pending and is
// written into the active topology so every member knows who runs what.
func (e *Engine) phase2(ctx context.Context, tk *Token) error {
	e.state = tk.State
	e.state.LeaderID = tk.State.LeaderID
	// Incremental log sync: adopt deltas carried by the token, then attach
	// our own new entries so peers can converge.
	if e.Log != nil && len(tk.Log) > 0 {
		if _, err := e.Log.ApplyDelta(tk.Log); err != nil {
			log.Printf("ring[%s] log delta gap: %v (request full sync later)", e.ID, err)
		}
	}
	for {
		pending := e.state.PendingList()
		if len(pending) == 0 {
			break
		}
		low := e.state.LowestAlive()
		if low == nil || low.ID != e.ID {
			break
		}
		tk0 := pending[0]
		claimed := e.state.ClaimPending(tk0.ID)
		if claimed == nil {
			break
		}
		if e.Handler != nil {
			if err := e.Handler.Claim(ctx, claimed); err != nil {
				e.state.PendingTasks[claimed.ID] = claimed
				log.Printf("ring[%s] claim %s failed: %v", e.ID, claimed.ID, err)
				break
			}
		}
		// Record the claim in the operation log so all peers converge on who
		// owns which forward (incremental log sync).
		if e.Log != nil {
			_, _ = e.Log.Append(e.ID, LogForwardAdd, map[string]any{
				"taskId": claimed.ID, "local": claimed.Local.Name, "remote": claimed.Remote.Name,
			})
		}
		e.state.AddTopology(claimed, e.ID)
	}
	// Publish our updated state back into the token so the next node carries
	// the fresh topology + pending set (do NOT revert local state to the
	// incoming snapshot — that would discard the claim we just made).
	tk.State = e.state
	tk.Passed = append(tk.Passed, e.ID)
	return nil
}

// OnToken receives the token: process by phase, return updated token.
func (e *Engine) OnToken(ctx context.Context, tk *Token) (*Token, error) {
	log.Printf("ring[%s] OnToken cycle=%d phase=%d passed=%v", e.ID, tk.Cycle, tk.Phase, tk.Passed)
	switch tk.Phase {
	case PhaseCollect:
		e.phase1(tk)
	case PhaseSync:
		if err := e.phase2(ctx, tk); err != nil {
			return tk, err
		}
	default:
		return tk, nil
	}
	return tk, nil
}

// Forward hands the token to this node's successor over the injected send.
// It is the transport hook used by the HTTP handler after OnToken.
func (e *Engine) Forward(ctx context.Context, tk *Token) error {
	nodes := make([]string, 0, len(e.state.Nodes))
	for _, n := range e.state.Nodes {
		nodes = append(nodes, n.ID)
	}
	log.Printf("ring[%s] fwd-debug id=%s nodes=%v", e.ID, e.ID, nodes)
	next, ok := e.state.AliveSuccessor(e.ID)
	if !ok {
		return nil // single-node ring
	}
	if next == e.ID {
		return nil // never forward to ourselves
	}
	if e.send != nil {
		log.Printf("ring[%s] forward cycle=%d phase=%d to %s", e.ID, tk.Cycle, tk.Phase, next)
		return e.send(ctx, next, tk)
	}
	return nil
}

// Snapshot returns a serializable view of the ring for the frontend.
type RingSnapshot struct {
	LeaderID   string       `json:"leaderId"`
	Cycle      int64        `json:"cycle"`
	RoundDelay int64        `json:"roundDelayMs"`
	Nodes      []Node       `json:"nodes"`
	Pending    []*Task      `json:"pending"`
	Topology   []*TopoEntry `json:"topology"`
	Log        []LogEntry   `json:"log,omitempty"`
}

func (e *Engine) Snapshot() *RingSnapshot {
	snap := &RingSnapshot{
		LeaderID:   e.state.LeaderID,
		Cycle:      e.state.Cycle,
		RoundDelay: e.state.RoundDelay.Milliseconds(),
		Nodes:      e.state.Nodes,
		Pending:    e.state.PendingList(),
		Topology:   e.state.TopologyList(),
	}
	if e.Log != nil {
		snap.Log = e.Log.Snapshot()
	}
	return snap
}

// advanceToken decides the next recipient: normal successor; if we are leader
// and everyone passed, flip phase for the second round.
func (e *Engine) advanceToken(ctx context.Context, tk *Token) error {
	if tk.Phase == PhaseCollect {
		all := true
		for _, n := range e.state.Nodes {
			if !n.Alive {
				continue
			}
			if !contains(tk.Passed, n.ID) {
				all = false
				break
			}
		}
		if all && e.state.LeaderID == e.ID {
			tk.Phase = PhaseSync
			tk.Passed = nil
			e.state.RoundDelay = tkDelaySince(tk)
		}
	}
	next, ok := e.state.AliveSuccessor(e.ID)
	if !ok {
		return nil
	}
	if e.send != nil {
		return e.send(ctx, next, tk)
	}
	return nil
}

// tkDelaySince measures elapsed ms since token SentAt (leader round delay).
func tkDelaySince(tk *Token) time.Duration {
	if tk.SentAt == 0 {
		return 200 * time.Millisecond
	}
	return time.Duration(time.Now().UnixMilli()-tk.SentAt) * time.Millisecond
}

func contains(xs []string, v string) bool {
	for _, x := range xs {
		if x == v {
			return true
		}
	}
	return false
}

// StartRing kicks off the cycle from the leader by sending the first token
// (phase 1) to the next node. Called once at leader boot.
func (e *Engine) StartRing(ctx context.Context) {
	if e.state.LeaderID != e.ID {
		return
	}
	// Single-node ring has no successor to hand the token to; do not POST to
	// ourselves. The cycle resumes once a newcomer joins (see JoinNode).
	if next, ok := e.state.AliveSuccessor(e.ID); !ok || next == e.ID {
		return
	}
	// Throttle: do not start a new cycle until roundDelay has elapsed since
	// the last one, so a healthy ring cycles at a deliberate pace.
	if !e.lastRingStart.IsZero() && time.Since(e.lastRingStart) < e.state.RoundDelay {
		return
	}
	e.lastRingStart = time.Now()
	e.state.Cycle++
	tk := &Token{
		Cycle:  e.state.Cycle,
		Phase:  PhaseCollect,
		State:  e.state,
		Passed: []string{e.ID},
		SentAt: time.Now().UnixMilli(),
	}
	e.phase1(tk)
	if err := e.advanceToken(ctx, tk); err != nil {
		log.Printf("ring[%s] start cycle %d: %v", e.ID, tk.Cycle, err)
	}
}

// JoinInfo is a node's self-description sent when requesting to join a ring.
type JoinInfo struct {
	ID      string   `json:"id"`
	Addr    string   `json:"addr"`
	Version string   `json:"version,omitempty"`
	Cache   []string `json:"cache,omitempty"`
}

// JoinNode handles an incoming join request from a new node: it inserts the
// newcomer right after this node (so the newcomer becomes our successor),
// keeps this node the leader, and returns the updated ring state for the
// newcomer to adopt.
func (e *Engine) JoinNode(j JoinInfo) *State {
	n := Node{ID: j.ID, Addr: j.Addr, Alive: true,
		Load: Load{MemPct: 50, NetPct: 50}, Version: j.Version, Cache: j.Cache}
	e.state.InsertAfter(e.ID, n)
	if e.Log != nil {
		_, _ = e.Log.Append(e.ID, LogNodeJoin, map[string]string{"node": n.ID, "addr": n.Addr})
	}
	if e.state.LeaderID == "" {
		e.state.LeaderID = e.ID
	}
	return &e.state
}

// AdoptState replaces this node's cluster picture with the state provided by
// the join target, then re-inserts us (in case we were absent).
func (e *Engine) AdoptState(s State) {
	ns := make([]string, 0, len(s.Nodes))
	for _, n := range s.Nodes {
		ns = append(ns, n.ID)
	}
	log.Printf("ring[%s] adopt-state nodes=%v", e.ID, ns)
	kept := map[string]*Task{}
	for id, t := range e.state.PendingTasks {
		if _, ok := s.PendingTasks[id]; !ok {
			kept[id] = t
		}
	}
	e.state = s
	if e.state.PendingTasks == nil {
		e.state.PendingTasks = map[string]*Task{}
	}
	for id, t := range kept {
		e.state.PendingTasks[id] = t
	}
	e.state.UpsertNode(Node{ID: e.ID, Addr: e.myAddr, Alive: true,
		Load: e.loadSnapshot(), Version: e.Version, Cache: e.Cache})
	if e.Log != nil {
		_, _ = e.Log.Append(e.ID, LogNodeJoin, map[string]string{"node": e.ID, "addr": e.myAddr})
	}
}

// SubmitTask adds a new forward request to pending; it rides the next token
// round and is claimed by the lowest-load member.
func (e *Engine) SubmitTask(local store.Local, remote store.Remote, link store.Link) *Task {
	return e.state.AddPending(local, remote, link)
}

// IsLeader reports whether this node is the current ring leader.
func (e *Engine) IsLeader() bool { return e.state.LeaderID == e.ID }

// LeaderAddr returns the current leader's address.
func (e *Engine) LeaderAddr() string {
	i := e.state.Find(e.state.LeaderID)
	if i < 0 {
		return ""
	}
	return e.state.Nodes[i].Addr
}