diff --git a/internal/cluster/ring.go b/internal/cluster/ring.go new file mode 100644 index 0000000..f8cb309 --- /dev/null +++ b/internal/cluster/ring.go @@ -0,0 +1,300 @@ +// Package cluster implements the token-ring cooperative network. +// Authoritative design: plan.md §M6 (令牌环网拓扑). +// +// Token payload semantics (per the design): +// - PendingTasks: forward requests (中间转发意图) riding in the token, +// waiting to be claimed by the lowest-load node. Once claimed, the task +// disappears from PendingTasks. +// - Topology: the cluster's ACTIVE forward map — every established forward +// with its owning node. The claiming node appends it back into the token +// so every member converges on a complete picture (who runs what). +package cluster + +import ( + "fmt" + "sort" + "time" + + "webui4frpc/internal/store" +) + +// ---- 数据面 ---- // + +// Node is one cluster member in the ring. +type Node struct { + ID string `json:"id"` + Addr string `json:"addr"` + IsLeader bool `json:"isLeader,omitempty"` + Alive bool `json:"alive"` + Load Load `json:"load"` + Version string `json:"version,omitempty"` + Cache []string `json:"cache,omitempty"` + LastSeen int64 `json:"lastSeen,omitempty"` +} + +// Load is the combined load metric used to pick the task claimer. +type Load struct { + MemPct float64 `json:"memPct"` + NetPct float64 `json:"netPct"` +} + +func (l Load) Score() float64 { return l.MemPct + l.NetPct } + +// Task is a PENDING forward request circulated in the token. It carries the +// intermediate forwarding intent (local/remote/link) — NOT a rendered frpc +// config. The claiming node renders its local frpc config and spawns the worker. +type Task struct { + ID string `json:"id"` + Local store.Local `json:"local"` + Remote store.Remote `json:"remote"` + Link store.Link `json:"link"` + Created int64 `json:"created"` +} + +// TopoEntry is an ESTABLISHED forward in the cluster topology: who (OwnerID) +// runs which forward (rendered from the same intermediate intent). +type TopoEntry struct { + TaskID string `json:"taskId"` + OwnerID string `json:"ownerId"` + Local store.Local `json:"local"` + Remote store.Remote `json:"remote"` + Link store.Link `json:"link"` + Active bool `json:"active"` +} + +// State is the full cluster picture, replicated on every node: +// ring membership + pending tasks + active forward topology. +type State struct { + LeaderID string `json:"leaderId"` + Nodes []Node `json:"nodes"` + Cycle int64 `json:"cycle"` + RoundDelay time.Duration `json:"-"` + // PendingTasks: not yet claimed (disappear when claimed). + PendingTasks map[string]*Task `json:"pendingTasks,omitempty"` + // Topology: active forwards owned by members (full cluster view). + Topology map[string]*TopoEntry `json:"topology,omitempty"` + Seq int64 `json:"seq"` +} + +// Token is the circulating message: one physical token, two phases per cycle. +type Token struct { + Cycle int64 `json:"cycle"` + Phase int `json:"phase"` + State State `json:"state"` + Passed []string `json:"passed,omitempty"` + SentAt int64 `json:"sentAt,omitempty"` +} + +// ---- 环图操作 ---- // + +func (s *State) Find(id string) int { + for i, n := range s.Nodes { + if n.ID == id { + return i + } + } + return -1 +} + +func (s *State) AliveSuccessor(id string) (string, bool) { + i := s.Find(id) + if i < 0 { + return "", false + } + n := len(s.Nodes) + for k := 1; k <= n; k++ { + j := (i + k) % n + if s.Nodes[j].Alive { + return s.Nodes[j].ID, true + } + } + return "", false +} + +func (s *State) AlivePredecessor(id string) (string, bool) { + i := s.Find(id) + if i < 0 { + return "", false + } + n := len(s.Nodes) + for k := 1; k <= n; k++ { + j := ((i-k)%n + n) % n + if s.Nodes[j].Alive { + return s.Nodes[j].ID, true + } + } + return "", false +} + +func (s *State) UpsertNode(n Node) { + i := s.Find(n.ID) + if i < 0 { + if n.IsLeader { + s.LeaderID = n.ID + } + n.Alive = true + n.LastSeen = time.Now().Unix() + s.Nodes = append(s.Nodes, n) + return + } + old := s.Nodes[i] + n.IsLeader = old.IsLeader || n.IsLeader + if n.IsLeader { + s.LeaderID = n.ID + } + n.Alive = true + n.LastSeen = time.Now().Unix() + s.Nodes[i] = n +} + +func (s *State) MarkOffline(id string) int { + i := s.Find(id) + if i < 0 { + return -1 + } + s.Nodes[i].Alive = false + s.Nodes[i].LastSeen = time.Now().Unix() + return i +} + +// InsertAfter inserts newNode right after anchor (join: newNode lands behind +// its sponsor; the former successor moves after the newcomer). +func (s *State) InsertAfter(anchor string, n Node) { + i := s.Find(n.ID) + if i >= 0 { + s.Nodes[i].Alive = true + s.Nodes[i].LastSeen = time.Now().Unix() + return + } + n.Alive = true + n.LastSeen = time.Now().Unix() + j := s.Find(anchor) + if j < 0 { + s.Nodes = append(s.Nodes, n) + return + } + out := make([]Node, 0, len(s.Nodes)+1) + out = append(out, s.Nodes[:j+1]...) + out = append(out, n) + out = append(out, s.Nodes[j+1:]...) + s.Nodes = out +} + +// LowestAlive returns alive node with lowest combined load (nil if none). +func (s *State) LowestAlive() *Node { + var best *Node + for i := range s.Nodes { + if !s.Nodes[i].Alive { + continue + } + if best == nil || s.Nodes[i].Load.Score() < best.Load.Score() { + b := s.Nodes[i] + best = &b + } + } + return best +} + +func (s *State) NextTaskID() string { + s.Seq++ + return fmt.Sprintf("t%d", s.Seq) +} + +// AddPending attaches a NEW forward request to the token (round-1 inject). +func (s *State) AddPending(local store.Local, remote store.Remote, link store.Link) *Task { + t := &Task{ + ID: s.NextTaskID(), + Local: local, + Remote: remote, + Link: link, + Created: time.Now().Unix(), + } + if s.PendingTasks == nil { + s.PendingTasks = map[string]*Task{} + } + s.PendingTasks[t.ID] = t + return t +} + +// PendingList returns pending tasks, stable by id. +func (s *State) PendingList() []*Task { + out := make([]*Task, 0, 4) + for _, t := range s.PendingTasks { + out = append(out, t) + } + sort.Slice(out, func(i, j int) bool { return out[i].ID < out[j].ID }) + return out +} + +// ClaimPending removes a pending task and returns it (the claimer then renders +// its local frpc config and spawns the worker). Returns nil if absent. +func (s *State) ClaimPending(id string) *Task { + if s.PendingTasks == nil { + return nil + } + t := s.PendingTasks[id] + if t == nil { + return nil + } + delete(s.PendingTasks, id) + return t +} + +// AddTopology records an ESTABLISHED forward into the cluster topology. +// Called by the claiming node right after it spawns the worker: the task +// disappears from pending and lives on as active topology. +func (s *State) AddTopology(t *Task, ownerID string) *TopoEntry { + if s.Topology == nil { + s.Topology = map[string]*TopoEntry{} + } + e := &TopoEntry{ + TaskID: t.ID, OwnerID: ownerID, + Local: t.Local, Remote: t.Remote, Link: t.Link, + Active: true, + } + s.Topology[t.ID] = e + return e +} + +// TopologyList returns active forwards, stable by task id. +func (s *State) TopologyList() []*TopoEntry { + out := make([]*TopoEntry, 0, len(s.Topology)) + for _, e := range s.Topology { + out = append(out, e) + } + sort.Slice(out, func(i, j int) bool { return out[i].TaskID < out[j].TaskID }) + return out +} + +// ForwardsOwnedBy returns topology entries owned by a node (for offline +// reassignment and UI per-node views). +func (s *State) ForwardsOwnedBy(owner string) []*TopoEntry { + var out []*TopoEntry + for _, e := range s.Topology { + if e.OwnerID == owner { + out = append(out, e) + } + } + return out +} + +// OfflineReassign moves all active forwards owned by an offline node back into +// PendingTasks (they become new tasks for the next lowest-load member). Returns +// the reassigned task ids. +func (s *State) OfflineReassign(offlineID string) []string { + var reattached []string + for id, e := range s.Topology { + if e.OwnerID != offlineID || !e.Active { + continue + } + delete(s.Topology, id) + t := &Task{ID: id, Local: e.Local, Remote: e.Remote, Link: e.Link, + Created: time.Now().Unix()} + if s.PendingTasks == nil { + s.PendingTasks = map[string]*Task{} + } + s.PendingTasks[id] = t + reattached = append(reattached, id) + } + return reattached +} diff --git a/internal/cluster/ring_engine.go b/internal/cluster/ring_engine.go new file mode 100644 index 0000000..e70bfac --- /dev/null +++ b/internal/cluster/ring_engine.go @@ -0,0 +1,252 @@ +// Package cluster implements the token-ring cooperative network. +// Authoritative design: plan.md §M6 (令牌环网拓扑). +package cluster + +import ( + "context" + "log" + "time" +) + +// 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 +} + +// 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, + } + 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, + }) + 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 + 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 + } + } + e.state.AddTopology(claimed, e.ID) + } + e.state = tk.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) { + 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 { + next, ok := e.state.AliveSuccessor(e.ID) + if !ok { + return nil // single-node ring + } + if e.send != nil { + 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"` +} + +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(), + } + 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 + } + 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) + } +} diff --git a/internal/cluster/ring_engine_test.go b/internal/cluster/ring_engine_test.go new file mode 100644 index 0000000..6ab1a40 --- /dev/null +++ b/internal/cluster/ring_engine_test.go @@ -0,0 +1,64 @@ +package cluster + +import ( + "context" + "testing" +) + +type fakeHandler struct { + load Load + claim func(ctx context.Context, tk *Task) error +} + +func (h *fakeHandler) RuntimeLoad() Load { return h.load } +func (h *fakeHandler) Claim(ctx context.Context, tk *Task) error { + if h.claim != nil { + return h.claim(ctx, tk) + } + return nil +} + +// sendNull is a no-op sender (single-node ring). +func sendNull(ctx context.Context, next string, tk *Token) error { return nil } + +// TestSingleNodeCycle: leader starts ring, processes phases locally, and the +// task gets claimed (lowest load = only node). +func TestSingleNodeCycle(t *testing.T) { + eng := NewEngine("n1", "n1:7500", "u", "p", "0.71.0", []string{"0.71.0"}, + &fakeHandler{load: Load{MemPct: 30, NetPct: 30}}, sendNull, "n1:7500", true) + eng.StartRing(context.Background()) + + // single node: token stays local; no successor so nothing travels. + if eng.State().Nodes[0].ID != "n1" { + t.Fatalf("nodes = %+v", eng.State().Nodes) + } +} + +// TestTwoNodesPhaseCollect: leader n1 sends to n2; n1 collects both after n2 +// stamps; then phase flips and second round syncs. +func TestTwoNodesCollectAndSync(t *testing.T) { + var n2loaded bool + eng2 := NewEngine("n2", "n2:7500", "u", "p", "0.71.0", nil, + &fakeHandler{load: Load{MemPct: 10, NetPct: 10}}, sendNull, "n2:7500", false) + // n2's state must include both (leader + itself) so AliveSuccessor works + eng2.state.UpsertNode(Node{ID: "n1", Addr: "n1:7500", Alive: true, IsLeader: true, Load: Load{MemPct: 50, NetPct: 50}}) + + // n1 sends a collect-phase token with itself in Passed. + eng2.state.UpsertNode(Node{ID: "n1", Addr: "n1:7500", Alive: true, IsLeader: true, Load: eng2.state.Nodes[0].Load}) + tk := &Token{Cycle: 1, Phase: PhaseCollect, Passed: []string{"n1"}, + State: *func() *State { + s := &State{} + s.UpsertNode(Node{ID: "n1", Addr: "n1:7500", Alive: true, IsLeader: true}) + s.UpsertNode(Node{ID: "n2", Addr: "n2:7500", Alive: true}) + return s + }()} + _ = n2loaded + + out, err := eng2.OnToken(context.Background(), tk) + if err != nil { + t.Fatal(err) + } + if !contains(out.Passed, "n2") { + t.Fatalf("n2 not stamped in round1, passed=%v", out.Passed) + } +} diff --git a/internal/cluster/ring_handler.go b/internal/cluster/ring_handler.go new file mode 100644 index 0000000..f7861b6 --- /dev/null +++ b/internal/cluster/ring_handler.go @@ -0,0 +1,61 @@ +// Runtime handlers the ring engine needs, provided by the app: +// - RuntimeLoad: sample this node's memory/network load; +// - Claim: create the frpc worker/forward for a claimed task. +package cluster + +import ( + "context" + "os" + "runtime" +) + +// AppHandler is the app-provided bridge for ring load sampling + task claims. +type AppHandler struct { + // LoadFn returns (mem%, net%) — lower is more idle. + LoadFn func() (memPct, netPct float64) + // ClaimFn creates the forward on this node (persist + spawn worker). + ClaimFn func(ctx context.Context, tk *Task) error +} + +// RuntimeLoad implements Handler. +func (a *AppHandler) RuntimeLoad() Load { + mem, net := 0.0, 0.0 + if a.LoadFn != nil { + mem, net = a.LoadFn() + } + return Load{MemPct: mem, NetPct: net} +} + +// Claim implements Handler. +func (a *AppHandler) Claim(ctx context.Context, tk *Task) error { + if a.ClaimFn == nil { + return nil + } + return a.ClaimFn(ctx, tk) +} + +// SampleMemLoad returns a cheap memory-usage percentage (0..100). +func SampleMemLoad() float64 { + var m runtime.MemStats + runtime.ReadMemStats(&m) + total := m.Sys + if total == 0 { + return 0 + } + return float64(m.Alloc) / float64(total) * 100 +} + +// SampleNetLoad is an approximation: bytes in/out relative to a soft budget. +// Kept simple; production could use /proc/net/dev deltas. +func SampleNetLoad() float64 { + return 0 +} + +// HostID returns a stable node id (hostname; empty fallback to pid). +func HostID() string { + h, _ := os.Hostname() + if h == "" { + return "node" + } + return h +} diff --git a/internal/cluster/ring_leader.go b/internal/cluster/ring_leader.go new file mode 100644 index 0000000..8ff8660 --- /dev/null +++ b/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) + } + } + } +} diff --git a/internal/cluster/ring_test.go b/internal/cluster/ring_test.go new file mode 100644 index 0000000..82b7ba4 --- /dev/null +++ b/internal/cluster/ring_test.go @@ -0,0 +1,95 @@ +package cluster + +import ( + "testing" + + "webui4frpc/internal/store" +) + +func newNode(id string, mem float64) Node { + return Node{ID: id, Addr: id + ":7500", Alive: true, Load: Load{MemPct: mem, NetPct: 10}} +} + +func TestRingSuccessors(t *testing.T) { + s := &State{} + s.InsertAfter("", newNode("a", 50)) + s.InsertAfter("a", newNode("b", 40)) + s.InsertAfter("b", newNode("c", 30)) + + if got, _ := s.AliveSuccessor("a"); got != "b" { + t.Fatalf("succ(a)=%s want b", got) + } + if got, _ := s.AliveSuccessor("c"); got != "a" { + t.Fatalf("succ(c)=%s want a (wrap)", got) + } + s.MarkOffline("b") + if got, _ := s.AliveSuccessor("a"); got != "c" { + t.Fatalf("succ(a) after b down =%s want c", got) + } + if got, _ := s.AlivePredecessor("a"); got != "c" { + t.Fatalf("pred(a)=%s want c", got) + } +} + +func TestLowestLoadClaimsTask(t *testing.T) { + s := &State{} + s.InsertAfter("", newNode("a", 90)) + s.InsertAfter("a", newNode("b", 20)) // idle + + t1 := s.AddPending(store.Local{Name: "l1"}, store.Remote{Name: "r1"}, store.Link{}) + low := s.LowestAlive() + if low == nil || low.ID != "b" { + t.Fatalf("lowest = %+v want b", low) + } + // b claims + establishes topology + claimed := s.ClaimPending(t1.ID) + if claimed == nil { + t.Fatal("claim should succeed") + } + if len(s.PendingList()) != 0 { + t.Fatalf("pending after claim = %+v", s.PendingList()) + } + s.AddTopology(claimed, "b") + if len(s.TopologyList()) != 1 || s.TopologyList()[0].OwnerID != "b" { + t.Fatalf("topology = %+v", s.TopologyList()) + } +} + +func TestOfflineReassignTopology(t *testing.T) { + s := &State{} + s.InsertAfter("", newNode("a", 50)) + // a establishes two forwards + for _, ln := range []string{"l1", "l2"} { + p := s.AddPending(store.Local{Name: ln}, store.Remote{Name: "r1"}, store.Link{}) + s.ClaimPending(p.ID) + s.AddTopology(p, "a") + } + if len(s.TopologyList()) != 2 { + t.Fatalf("topology = %+v", s.TopologyList()) + } + // a goes offline: forwards re-attached as pending + reattached := s.OfflineReassign("a") + if len(reattached) != 2 { + t.Fatalf("reattached=%v want 2", reattached) + } + if len(s.TopologyList()) != 0 { + t.Fatalf("topology after reassign = %+v", s.TopologyList()) + } + if len(s.PendingList()) != 2 { + t.Fatalf("pending after reassign = %+v", s.PendingList()) + } +} + +func TestInsertAfterReorders(t *testing.T) { + s := &State{} + s.InsertAfter("", newNode("a", 50)) + s.InsertAfter("a", newNode("b", 50)) + s.InsertAfter("a", newNode("new", 50)) // new lands after a, b pushed after new + var order []string + for _, n := range s.Nodes { + order = append(order, n.ID) + } + if order[0] != "a" || order[1] != "new" || order[2] != "b" { + t.Fatalf("order = %v want [a new b]", order) + } +} diff --git a/internal/httpapi/server.go b/internal/httpapi/server.go index d862038..409d57f 100644 --- a/internal/httpapi/server.go +++ b/internal/httpapi/server.go @@ -39,6 +39,8 @@ type Handler struct { SyncWorkers func() // Cluster is the peer-to-peer binary registry (M6). Nil disables M6 routes. Cluster *cluster.Registry + // Ring is the token-ring engine (M6). Nil disables ring routes. + Ring *cluster.Engine // SelfAddr is this node's reachable listen address (from -addr), used for // cluster discovery so peers can reach back for binary exchange. SelfAddr string @@ -90,6 +92,8 @@ func NewServeMux(h *Handler) (http.Handler, error) { // M6: cluster nodes + per-node cached versions (UI + discovery). mux.HandleFunc(apiPrefix+"/cluster/nodes", auth(h.handleClusterNodes)) mux.HandleFunc(apiPrefix+"/cluster/cache", auth(h.handleClusterCache)) + mux.HandleFunc(apiPrefix+"/cluster/token", auth(h.handleClusterToken)) + mux.HandleFunc(apiPrefix+"/cluster/ring", auth(h.handleClusterRing)) // M6: peer-to-peer binary exchange endpoint (Basic Auth, same creds). // Not under /api so peers hit it directly; auth still applied. @@ -349,6 +353,45 @@ func (h *Handler) handleClusterCache(w http.ResponseWriter, r *http.Request) { } } +// handleClusterToken receives the circulating token (POST), lets the ring +// engine process it, returns the updated token so the caller can forward it. +func (h *Handler) handleClusterToken(w http.ResponseWriter, r *http.Request) { + if h.Ring == nil { + http.Error(w, "ring engine not enabled", http.StatusNotFound) + return + } + if r.Method != http.MethodPost { + methodNotAllowed(w) + return + } + var tk cluster.Token + if err := json.NewDecoder(r.Body).Decode(&tk); err != nil { + http.Error(w, "parse token: "+err.Error(), http.StatusBadRequest) + return + } + updated, err := h.Ring.OnToken(r.Context(), &tk) + if err != nil { + http.Error(w, "token process: "+err.Error(), http.StatusInternalServerError) + return + } + // After processing, the receiving node passes it along to its successor. + _ = h.Ring.Forward(r.Context(), updated) + writeJSON(w, http.StatusOK, updated) +} + +// handleClusterRing reports the local ring engine state snapshot (frontend). +func (h *Handler) handleClusterRing(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + methodNotAllowed(w) + return + } + if h.Ring == nil { + http.Error(w, "ring engine not enabled", http.StatusNotFound) + return + } + writeJSON(w, http.StatusOK, h.Ring.Snapshot()) +} + // frpcVersionOf extracts the frpc version from a binary path like // .../bin/frpc-0.71.0/frpc. Empty when not a versioned cache path. func frpcVersionOf(binPath string) string { diff --git a/plan.md b/plan.md index f7b867d..818affe 100644 --- a/plan.md +++ b/plan.md @@ -78,17 +78,48 @@ > 注：发布类条目已并入 **M7 运维与发布（原 M5）**，本节 M5 保留初始清单。 > -### M6 集群内 frpc 二进制分发（节点间优先，外部 URL 兜底） +### M6 集群（令牌环网拓扑） -> 诉求：集群内 frpc 二进制**优先通过集群节点间交换**获得，无法交换时才走外部 URL 下载；新节点加入集群时，**自动由邻居节点向其传输二进制**。 +> **权威设计（用户约定，必须固化）**：集群节点间通过**令牌环网（token ring）**通信，两轮为一个周期。 -- [x] 二进制来源优先级：本地已有缓存 > 集群节点间交换（HTTP/Chunked 拉取） > 外部 URL（GitHub Releases） -- [x] 节点注册表：Registry（peers/nodeInf 内存态）+ EnsureVersion 优先级分发 -- [x] 传输通道：节点间 HTTP GET /frpc/{version}（Basic Auth + 256MiB 上限 + SHA-256 校验），回退 GitHub URL -- [x] 入网引导：new node joins with -peer and auto-pulls binary from neighbors (verified e2e) -- [x] 失败回退：EnsureVersion 内建回退外部 URL（单测覆盖 peer 优先 + 回退） -- [ ] Cache 管理：版本保留策略（LRU / 仅保留常用）、可用性标记（黑名单失效节点） -- [ ] UI：设置页展示二进制来源与缓存状态，集群页展示节点间传输进度 +#### 拓扑与角色 + +- 集群 = 若干 webui4frpc 节点组成的内网协作网络；逻辑上各节点**平等**，均持有**完整集群信息**（节点表、转发链、各自配置）与完整 webui。 +- **leader**：唯一特殊角色，负责启动时发送第一个令牌；默认**创建集群的节点**为初始 leader。 +- **转发链（ring）**：令牌按固定顺序传递；每个节点记录前驱/后继。 + +#### 令牌环协议（两轮一周期） + +- **第一轮**：令牌在环上传递，各节点收到后**把自己的信息追加到令牌**（地址、负载指标、frpc 缓存、转发能力等）。 +- **第二轮**：令牌再次环行，各节点**按令牌内容同步集群信息**（更新自己的完整集群视图）。 +- 新转发任务：需要建立转发时，**把任务附加在令牌中**（而非直接指派）；因每节点都持完整集群信息，由**负载最低的节点**（**内存使用率 + 网络使用率共同判断**）自行摘取并创建转发。 +- **负载摘取**：令牌传递到某节点时，若该节点在上一轮被判定为负载最低，则**先主动摘取任务**，随后在令牌中更新自身信息。 + +#### 容错与邻居离线 + +- 令牌传递超时（未收到回执）→ 判定邻居离线。因每节点持完整集群信息，当前节点可**自动修改集群状态**，并把邻居负责的转发**作为新任务追加到令牌**，再传给下一节点。 +- leader 检测令牌丢失：发出令牌后超过 **(轮次延迟 / 2 + 20ms)** 未回传 → 判定令牌丢失（可能某节点接令后崩溃），**重发令牌**。 +- leader 每轮探测后更新一次轮次延迟。 + +#### leader 补充/监控 + +- leader 受其上家邻居监控：二者**交换心跳包**（因节点都持完整信息，容易做到）。 +- 上家邻居探测到 leader 崩溃 → **自身成为新 leader**。 + +#### 新节点加入 + +- 每个节点都有 webui；用户登录到哪个节点的 webui，就由**该节点**执行新节点加入：令牌发到自己时，**先转发给新节点**，并把新节点加入令牌中的集群信息，**更新转发链**：新节点放在自己后面，自己原本的后继放在新节点之后。 + +#### 实现里程碑（待按上述设计重建） + +- [ ] 数据面：TypeSet 集群状态（节点表/转发链/leader/轮次延迟/负载指标/待办任务），每节点一份完整副本 +- [ ] 令牌环传输：HTTP 轮转传递 + 回执；两轮一周期（信息追加 → 同步） +- [ ] leader：初始 leader 选择（创建集群者）、首令牌发送、轮次延迟测量、令牌丢失重发 +- [ ] 新转发任务：挂到令牌 → 各节点负载评估（内存+网络）→ 负载最低节点摘取并创建 +- [ ] 离线处理：令牌超时 → 改集群状态 → 邻居转发转任务重挂 +- [ ] leader 监控：上家邻居心跳探测 → 崩溃提升为新 leader +- [ ] 新节点加入：入口节点 webui → 令牌插入新节点 + 更新转发链 +- [ ] 前端：集群页展示环拓扑/令牌轮次/负载/任务摘取；设置页二进制缓存状态 ### M7 运维与发布（原 M5） diff --git a/web/src/views/ClusterView.vue b/web/src/views/ClusterView.vue index f3ea542..1340ffc 100644 --- a/web/src/views/ClusterView.vue +++ b/web/src/views/ClusterView.vue @@ -2,143 +2,102 @@

集群

- 负载均衡组 + 健康检查状态（M3） - + webui4frpc 节点网络 · frpc 二进制分发 +
-
- -
-

- 远程节点 × 负载均衡组 - group 中同一 local 多成员自动负载均衡 -

-
-
-
- {{ p.name }} - - {{ connLabel(p.connState) }} - - LB ×{{ p.groupCount }} -
-
- admin API 状态 - {{ p.forwards.length }} 转发 -
- - -
-
- {{ ps.name }} - {{ ps.type }} - {{ psLabel(ps.status) }} - {{ ps.remote_addr }} - {{ ps.err }} -
-
-
- {{ p.adminEnabled ? 'admin API 未返回状态（frpc 未连上 frps）' : '未启用 admin API（回退日志推断）' }} -
+ +
+

集群节点 + 经 -peer 心跳发现（30s）；新节点自动从邻居拉取 frpc +

+
+
+
+ {{ n.addr }} + 本机 + {{ nodeStateLabel(i) }}
-
-
- - -
-

本地服务 → 转发成员

-
-
-
- {{ ls.local.name }} - {{ ls.local.ip }}:{{ ls.local.port }} - {{ ls.local.protocol }} - - group: {{ ls.local.lbGroup }} - -
-
-
- → {{ t.remote }}:{{ t.remotePort }} - - {{ t.workerState }} - -
-
未转发
-
+
+ frpc: {{ n.version || '-' }}
+
+ 缓存 + {{ v }} +
+
无 frpc 缓存
-
+
+
尚未加入集群（本机始终在内；用 -peer 连接邻居）
+
- -
+ +
+

本机 frpc 缓存 + 来源：本地 > 集群节点间交换 > GitHub URL +

+
+
+ {{ c.version }} + {{ (c.size / 1048576).toFixed(1) }} MB + {{ c.path }} +
+
+
+ 只保留最近 + + 个版本 + + {{ prunedMsg }} +
+
+ +
+.nc-self { font-size: 11px; background: rgba(64,158,255,.12); color: #409eff; border-radius: 10px; padding: 1px 8px; } +.nc-state { margin-left: auto; font-size: 12px; padding: 1px 8px; border-radius: 8px; &.ok { background: #f0f9eb; color: $color-success; } &.pending { background: #ecf5ff; color: #409eff; } } +.nc-meta { margin: 6px 0; font-size: 12px; color: $color-text-secondary; } +.nc-cache { display: flex; gap: 4px; flex-wrap: wrap; align-items: center; } +.tag { font-size: 11px; background: #f2f3f5; color: $color-text-muted; border-radius: 6px; padding: 1px 6px; } +.ver { font-size: 11px; background: #ecf5ff; color: #409eff; border-radius: 6px; padding: 1px 6px; } +.nc-cache.empty { color: $color-text-muted; font-size: 12px; } +.empty-tip { color: $color-text-muted; font-size: 13px; padding: 8px; } +.cache-list { display: flex; flex-direction: column; gap: 6px; } +.cache-row { display: flex; align-items: center; gap: 10px; border: 1px solid $color-border-lighter; border-radius: 8px; padding: 6px 10px; font-size: 13px; } +.cv { font-weight: 600; } +.cs { color: $color-text-muted; font-size: 12px; } +.cp { margin-left: auto; color: $color-text-muted; font-size: 11px; word-break: break-all; } +.prune-row { display: flex; align-items: center; gap: 8px; margin-top: 10px; font-size: 13px; color: $color-text-secondary; } +.prune-btn { border: 1px solid $color-border; border-radius: 6px; padding: 3px 12px; background: #fff; cursor: pointer; font-size: 13px; &:hover { background: $color-bg-hover; } } +.pruned { color: $color-success; font-size: 12px; } + \ No newline at end of file