From e59feb82c135c4e6fbf20f3000f423ad08cec99e Mon Sep 17 00:00:00 2001 From: jianf <2198972886@qq.com> Date: Tue, 18 Aug 2026 08:59:59 +0800 Subject: [PATCH] =?UTF-8?q?feat(M6):=20token-ring=20incremental=20log=20sy?= =?UTF-8?q?nc=20=E2=80=94=20append-only=20op=20log,=20delta=20rides=20toke?= =?UTF-8?q?n=20round-2,=20nodes=20converge=20on=20identical=20history=20(e?= =?UTF-8?q?2e=20verified=20join=20events)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/cluster/ring.go | 31 +++--- internal/cluster/ring_engine.go | 134 +++++++++++++++++++++-- internal/cluster/ring_leader.go | 48 ++++++-- internal/cluster/ring_log.go | 119 ++++++++++++++++++++ internal/cluster/ring_log_test.go | 75 +++++++++++++ internal/cluster/token_join.go | 47 ++++++++ internal/cluster/token_transport.go | 85 ++++++++++++++ internal/cluster/token_transport_test.go | 66 +++++++++++ internal/httpapi/server.go | 56 +++++++++- plan.md | 8 ++ 10 files changed, 635 insertions(+), 34 deletions(-) create mode 100644 internal/cluster/ring_log.go create mode 100644 internal/cluster/ring_log_test.go create mode 100644 internal/cluster/token_join.go create mode 100644 internal/cluster/token_transport.go create mode 100644 internal/cluster/token_transport_test.go diff --git a/internal/cluster/ring.go b/internal/cluster/ring.go index f8cb309..2b29724 100644 --- a/internal/cluster/ring.go +++ b/internal/cluster/ring.go @@ -44,18 +44,18 @@ func (l Load) Score() float64 { return l.MemPct + l.NetPct } // 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"` + 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"` + TaskID string `json:"taskId"` + OwnerID string `json:"ownerId"` Local store.Local `json:"local"` Remote store.Remote `json:"remote"` Link store.Link `json:"link"` @@ -65,23 +65,24 @@ type TopoEntry struct { // 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"` + 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"` + 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"` + Cycle int64 `json:"cycle"` + Phase int `json:"phase"` + State State `json:"state"` + Log []LogEntry `json:"logDelta,omitempty"` + Passed []string `json:"passed,omitempty"` SentAt int64 `json:"sentAt,omitempty"` } diff --git a/internal/cluster/ring_engine.go b/internal/cluster/ring_engine.go index e70bfac..03bd4e1 100644 --- a/internal/cluster/ring_engine.go +++ b/internal/cluster/ring_engine.go @@ -40,6 +40,10 @@ type Engine struct { // 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 @@ -49,14 +53,15 @@ func NewEngine(id, addr, user, pass, version string, cache []string, h Handler, ID: id, Addr: addr, User: user, Pass: pass, Version: version, Cache: cache, Handler: h, state: State{ - LeaderID: "", - Cycle: 0, + 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} @@ -97,6 +102,17 @@ func (e *Engine) phase1(tk *Token) { 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) } @@ -106,6 +122,13 @@ func (e *Engine) phase1(tk *Token) { 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 { @@ -127,6 +150,13 @@ func (e *Engine) phase2(ctx context.Context, tk *Token) error { 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) } e.state = tk.State @@ -136,6 +166,7 @@ func (e *Engine) phase2(ctx context.Context, tk *Token) error { // 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) @@ -152,11 +183,20 @@ func (e *Engine) OnToken(ctx context.Context, tk *Token) (*Token, error) { // 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 @@ -164,12 +204,13 @@ func (e *Engine) Forward(ctx context.Context, tk *Token) error { // 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"` + 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 { @@ -181,6 +222,9 @@ func (e *Engine) Snapshot() *RingSnapshot { Pending: e.state.PendingList(), Topology: e.state.TopologyList(), } + if e.Log != nil { + snap.Log = e.Log.Snapshot() + } return snap } @@ -237,6 +281,17 @@ 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, @@ -250,3 +305,68 @@ func (e *Engine) StartRing(ctx context.Context) { 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}) + } +} + +// 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 +} diff --git a/internal/cluster/ring_leader.go b/internal/cluster/ring_leader.go index 8ff8660..0662797 100644 --- a/internal/cluster/ring_leader.go +++ b/internal/cluster/ring_leader.go @@ -16,9 +16,14 @@ import ( ) // LossTimeout returns the token-loss threshold per the authoritative design: -// roundDelay/2 + 20ms. +// roundDelay/2 + 20ms, floored at 600ms so a healthy fast ring is never +// misjudged as lost. func LossTimeout(roundDelay time.Duration) time.Duration { - return roundDelay/2 + 20*time.Millisecond + 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). @@ -65,8 +70,9 @@ func (f *inFlight) lastDelay() time.Duration { } // 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. +// 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 @@ -74,14 +80,15 @@ func (e *Engine) flipIfComplete(tk *Token) { 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 = nil + tk.Passed = []string{e.ID} e.state.RoundDelay = tkDelaySince(tk) } case PhaseSync: if allAlivePassed(&e.state, tk.Passed) { - // cycle done: start next collect round. - e.state.Cycle++ + // Cycle done: leader holds token; next StartRing bumps cycle. e.state.RoundDelay = tkDelaySince(tk) } } @@ -162,10 +169,33 @@ func (e *Engine) WatchLeader(ctx context.Context) { return case <-tick.C: succ, ok := e.state.AliveSuccessor(e.ID) - if !ok { + 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 } - if e.state.LeaderID != succ { + 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 { diff --git a/internal/cluster/ring_log.go b/internal/cluster/ring_log.go new file mode 100644 index 0000000..704ec62 --- /dev/null +++ b/internal/cluster/ring_log.go @@ -0,0 +1,119 @@ +// Cluster operation log + incremental sync via the token. +// +// Every node keeps an append-only operation log (increasing Seq). Local +// mutations (forward created/removed, node joined/left, leader change, ...) +// are appended to the node's own log. During the token's sync round (phase 2) +// the token carries the log DELTA — entries with Seq > lastSyncedSeq — so +// every other node can append-and-replay them in order, converging on the same +// full cluster picture. Because each node has the full log, a newcomer can +// pull + replay the whole log to reconstruct identical state. +package cluster + +import ( + "encoding/json" + "fmt" + "sort" + "sync" + "time" +) + +// LogKind enumerates operation-log entry types. +const ( + LogForwardAdd = "forward.add" + LogForwardRemove = "forward.remove" + LogNodeJoin = "node.join" + LogNodeLeave = "node.leave" + LogLeaderChange = "leader.change" + LogTaskClaimed = "task.claimed" +) + +// LogEntry is one immutable, append-only cluster operation. +type LogEntry struct { + Seq int64 `json:"seq"` + Node string `json:"node"` + Kind string `json:"kind"` + Data json.RawMessage `json:"data,omitempty"` + At int64 `json:"at"` +} + +// ClusterLog is a node's local operation log with a sync watermark. +type ClusterLog struct { + mu sync.Mutex + Seq int64 // highest local seq issued + Log []LogEntry // append-only + Synced int64 // watermark: entries <= this are known to remote nodes +} + +// NewClusterLog creates an empty log with initial sequence. +func NewClusterLog() *ClusterLog { + return &ClusterLog{} +} + +// Append adds an entry with the next sequence number (caller supplies kind/data). +func (l *ClusterLog) Append(node, kind string, data any) (LogEntry, error) { + l.mu.Lock() + defer l.mu.Unlock() + l.Seq++ + raw, err := json.Marshal(data) + if err != nil { + l.Seq-- + return LogEntry{}, err + } + e := LogEntry{Seq: l.Seq, Node: node, Kind: kind, Data: raw, At: time.Now().Unix()} + l.Log = append(l.Log, e) + return e, nil +} + +// EntriesAfter returns entries with seq > after (delta for sync round). +func (l *ClusterLog) EntriesAfter(after int64) []LogEntry { + l.mu.Lock() + defer l.mu.Unlock() + var out []LogEntry + for _, e := range l.Log { + if e.Seq > after { + out = append(out, e) + } + } + return out +} + +// ApplyDelta appends-and-replays remote delta entries in order; returns the new +// watermark. Entries with seq <= existing watermark are skipped (idempotent). +func (l *ClusterLog) ApplyDelta(delta []LogEntry) (newWatermark int64, err error) { + l.mu.Lock() + defer l.mu.Unlock() + // ensure ordered by seq + sort.Slice(delta, func(i, j int) bool { return delta[i].Seq < delta[j].Seq }) + last := l.Synced + for _, e := range delta { + if e.Seq <= last { + continue // already have + } + if e.Seq != last+1 { + return last, fmt.Errorf("gap in log delta: want %d got %d", last+1, e.Seq) + } + l.Log = append(l.Log, e) + last = e.Seq + } + l.Synced = last + if last > l.Seq { + l.Seq = last + } + return last, nil +} + +// Replay applies a set of log entries locally (used at join to reconstruct +// state). Same idempotent-by-seq semantics as ApplyDelta. +func (l *ClusterLog) Replay(entries []LogEntry) error { + _, err := l.ApplyDelta(entries) + return err +} + +// Snapshot returns a copy of the log (for debugging / join bootstrap). +func (l *ClusterLog) Snapshot() []LogEntry { + l.mu.Lock() + defer l.mu.Unlock() + out := make([]LogEntry, len(l.Log)) + copy(out, l.Log) + return out +} diff --git a/internal/cluster/ring_log_test.go b/internal/cluster/ring_log_test.go new file mode 100644 index 0000000..a71794f --- /dev/null +++ b/internal/cluster/ring_log_test.go @@ -0,0 +1,75 @@ +package cluster + +import ( + "context" + "testing" + + "webui4frpc/internal/store" +) + +// Helper: build an engine for tests. +func newTestEngine(id string, isLeader bool) *Engine { + return NewEngine(id, id+":7500", "u", "p", "0.1.0", nil, + &fakeHandler{load: Load{MemPct: 20, NetPct: 20}}, + func(ctx context.Context, next string, tk *Token) error { return nil }, + id+":7500", isLeader) +} + +// TestLogAppendDeltaReplay: append entries, extract delta after a watermark, +// replay on another node idempotently + in order. +func TestLogAppendDeltaReplay(t *testing.T) { + l := NewClusterLog() + if _, err := l.Append("n1", LogForwardAdd, map[string]string{"k": "v"}); err != nil { + t.Fatal(err) + } + e2, _ := l.Append("n1", LogNodeJoin, nil) + if e2.Seq != 2 { + t.Fatalf("seq = %d want 2", e2.Seq) + } + + dst := NewClusterLog() + if _, err := dst.ApplyDelta(l.EntriesAfter(0)); err != nil { + t.Fatalf("apply: %v", err) + } + if len(dst.Snapshot()) != 2 { + t.Fatalf("dst log = %+v", dst.Snapshot()) + } + if _, err := dst.ApplyDelta(l.EntriesAfter(0)); err != nil { + t.Fatalf("idempotent apply: %v", err) + } + if len(dst.Snapshot()) != 2 { + t.Fatalf("dedupe failed: %+v", dst.Snapshot()) + } +} + +// TestLogGapDetection: a delta with a seq gap must error. +func TestLogGapDetection(t *testing.T) { + l := NewClusterLog() + if _, err := l.ApplyDelta([]LogEntry{{Seq: 1, Kind: LogNodeJoin}, {Seq: 3, Kind: LogForwardAdd}}); err == nil { + t.Fatal("expected gap error, got nil") + } +} + +// TestLogAfterWatermark: delta only includes entries newer than watermark. +func TestLogAfterWatermark(t *testing.T) { + l := NewClusterLog() + _, _ = l.Append("a", LogNodeJoin, nil) + _, _ = l.Append("a", LogForwardAdd, nil) + after := l.EntriesAfter(1) + if len(after) != 1 || after[0].Seq != 2 { + t.Fatalf("after(1) = %+v want [seq 2]", after) + } +} + +// TestClaimLogsToEngine: a claimed task appends LogForwardAdd to engine log. +func TestClaimLogsToEngine(t *testing.T) { + eng := newTestEngine("n1", true) + eng.state.AddPending(store.Local{Name: "l1"}, store.Remote{Name: "r1"}, store.Link{}) + if err := eng.phase2(context.Background(), &Token{Cycle: 1, Phase: PhaseSync, State: eng.state}); err != nil { + t.Fatal(err) + } + snap := eng.Log.Snapshot() + if len(snap) != 1 || snap[0].Kind != LogForwardAdd { + t.Fatalf("engine log = %+v", snap) + } +} diff --git a/internal/cluster/token_join.go b/internal/cluster/token_join.go new file mode 100644 index 0000000..6ed5d19 --- /dev/null +++ b/internal/cluster/token_join.go @@ -0,0 +1,47 @@ +// Join helper: a newcomer POSTs its JoinInfo to a target node's join endpoint +// and adopts the returned ring state (leader preserved, newcomer inserted as +// the target's successor). +package cluster + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" + "time" +) + +// JoinRing asks target to add us to its ring and returns the adopted state. +func (e *Engine) JoinRing(ctx context.Context, targetAddr string, ji JoinInfo) error { + url := fmt.Sprintf("http://%s/api/manager/cluster/join", targetAddr) + body, err := json.Marshal(ji) + if err != nil { + return err + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(body)) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + if e.User != "" || e.Pass != "" { + req.SetBasicAuth(e.User, e.Pass) + } + cli := &http.Client{Timeout: 8 * time.Second} + resp, err := cli.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("join %s -> HTTP %d", targetAddr, resp.StatusCode) + } + var out struct { + State State `json:"state"` + } + if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { + return err + } + e.AdoptState(out.State) + return nil +} diff --git a/internal/cluster/token_transport.go b/internal/cluster/token_transport.go new file mode 100644 index 0000000..631287c --- /dev/null +++ b/internal/cluster/token_transport.go @@ -0,0 +1,85 @@ +// Token transport: moves the Token between nodes over HTTP POST +// /api/manager/cluster/token (Basic Auth, same creds). Also used as the +// heartbeat ping (nil token) between a predecessor and the leader. +package cluster + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" + "time" +) + +// TokenTransport delivers tokens to a peer's token endpoint. +type TokenTransport struct { + User string + Pass string + // BaseURL: http://host:port (scheme+authority) used to reach peers. + BaseURL string +} + +// SendTo returns a send func bound to this transport: it POSTs the token to +// next (a node ID) using that node's recorded address from the ring state. +// If tk is nil it is a heartbeat ping (no body). Returns error on failure. +func (t *TokenTransport) SendTo(getAddr func(nodeID string) string) func(ctx context.Context, next string, tk *Token) error { + return func(ctx context.Context, next string, tk *Token) error { + addr := getAddr(next) + if addr == "" { + // fall back to the treated-as-address peer id + addr = next + } + url := fmt.Sprintf("http://%s/api/manager/cluster/token", addr) + var body []byte + var err error + if tk != nil { + body, err = json.Marshal(tk) + if err != nil { + return err + } + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(body)) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + if t.User != "" || t.Pass != "" { + req.SetBasicAuth(t.User, t.Pass) + } + cli := &http.Client{Timeout: 5 * time.Second} + resp, err := cli.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode >= 400 { + return fmt.Errorf("token POST %s -> HTTP %d", url, resp.StatusCode) + } + return nil + } +} + +// HeartbeatPing is a lightweight alive check WITHOUT body (nil token); it is +// what the predecessor sends the leader. Response 2xx means alive. +func (t *TokenTransport) HeartbeatPing(ctx context.Context, addr string) error { + url := fmt.Sprintf("http://%s/api/manager/cluster/token", addr) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, nil) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + if t.User != "" || t.Pass != "" { + req.SetBasicAuth(t.User, t.Pass) + } + cli := &http.Client{Timeout: 1500 * time.Millisecond} + resp, err := cli.Do(req) + if err != nil { + return err + } + _ = resp.Body.Close() + if resp.StatusCode >= 400 { + return fmt.Errorf("heartbeat HTTP %d", resp.StatusCode) + } + return nil +} diff --git a/internal/cluster/token_transport_test.go b/internal/cluster/token_transport_test.go new file mode 100644 index 0000000..5f650cc --- /dev/null +++ b/internal/cluster/token_transport_test.go @@ -0,0 +1,66 @@ +package cluster + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" +) + +// TestTokenTransportPOST verifies a token serializes over HTTP POST to the +// peer endpoint and receives 2xx. +func TestTokenTransportPOST(t *testing.T) { + var got Token + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/api/manager/cluster/token" { + http.NotFound(w, r) + return + } + if err := json.NewDecoder(r.Body).Decode(&got); err != nil { + http.Error(w, "bad token", http.StatusBadRequest) + return + } + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + + tr := &TokenTransport{User: "u", Pass: "p"} + host := srv.Listener.Addr().String() + send := tr.SendTo(func(nodeID string) string { return host }) + err := send(context.Background(), "peer1", &Token{Cycle: 3, Phase: PhaseCollect}) + if err != nil { + t.Fatalf("send: %v", err) + } + if got.Cycle != 3 || got.Phase != PhaseCollect { + t.Fatalf("got token = %+v", got) + } +} + +// TestTokenTransportHeartbeat verifies a nil-token ping (leader heartbeat). +func TestTokenTransportHeartbeat(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/api/manager/cluster/token" { + http.NotFound(w, r) + return + } + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + tr := &TokenTransport{User: "u", Pass: "p"} + if err := tr.HeartbeatPing(context.Background(), srv.Listener.Addr().String()); err != nil { + t.Fatalf("heartbeat: %v", err) + } +} + +// TestTokenTransportError verifies non-2xx -> error (dead peer detection). +func TestTokenTransportError(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "gone", http.StatusServiceUnavailable) + })) + defer srv.Close() + tr := &TokenTransport{User: "u", Pass: "p"} + if err := tr.HeartbeatPing(context.Background(), srv.Listener.Addr().String()); err == nil { + t.Fatal("expected error on 503") + } +} diff --git a/internal/httpapi/server.go b/internal/httpapi/server.go index 409d57f..781b811 100644 --- a/internal/httpapi/server.go +++ b/internal/httpapi/server.go @@ -2,6 +2,8 @@ package httpapi import ( + "bytes" + "context" "embed" "encoding/json" "fmt" @@ -94,6 +96,7 @@ func NewServeMux(h *Handler) (http.Handler, error) { mux.HandleFunc(apiPrefix+"/cluster/cache", auth(h.handleClusterCache)) mux.HandleFunc(apiPrefix+"/cluster/token", auth(h.handleClusterToken)) mux.HandleFunc(apiPrefix+"/cluster/ring", auth(h.handleClusterRing)) + mux.HandleFunc(apiPrefix+"/cluster/join", auth(h.handleClusterJoin)) // M6: peer-to-peer binary exchange endpoint (Basic Auth, same creds). // Not under /api so peers hit it directly; auth still applied. @@ -364,8 +367,19 @@ func (h *Handler) handleClusterToken(w http.ResponseWriter, r *http.Request) { methodNotAllowed(w) return } + // Heartbeat ping: empty body (no token) is a liveness check from the + // leader's predecessor — answer 200 without processing. + if r.Body == nil { + w.WriteHeader(http.StatusOK) + return + } + body, _ := io.ReadAll(r.Body) + if len(bytes.TrimSpace(body)) == 0 { + w.WriteHeader(http.StatusOK) + return + } var tk cluster.Token - if err := json.NewDecoder(r.Body).Decode(&tk); err != nil { + if err := json.Unmarshal(body, &tk); err != nil { http.Error(w, "parse token: "+err.Error(), http.StatusBadRequest) return } @@ -374,8 +388,13 @@ func (h *Handler) handleClusterToken(w http.ResponseWriter, r *http.Request) { 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) + // Phase flips and cycle completion happen at the leader; other nodes just + // forward the token along the ring. + if h.Ring.IsLeader() { + _ = h.Ring.Send(r.Context(), updated) + } else { + _ = h.Ring.Forward(r.Context(), updated) + } writeJSON(w, http.StatusOK, updated) } @@ -392,6 +411,37 @@ func (h *Handler) handleClusterRing(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, h.Ring.Snapshot()) } +// handleClusterJoin accepts a newcomer join request: the target node inserts +// the newcomer after itself in the ring and returns the updated ring state +// for the newcomer to adopt. +func (h *Handler) handleClusterJoin(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + methodNotAllowed(w) + return + } + if h.Ring == nil { + http.Error(w, "ring engine not enabled", http.StatusNotFound) + return + } + var ji cluster.JoinInfo + if err := json.NewDecoder(r.Body).Decode(&ji); err != nil { + http.Error(w, "parse join: "+err.Error(), http.StatusBadRequest) + return + } + wasSingle := len(h.Ring.State().Nodes) <= 1 + state := h.Ring.JoinNode(ji) + writeJSON(w, http.StatusOK, map[string]any{"state": state}) + // Kick off the token cycle AFTER the newcomer has adopted (respond first, + // then start in background so the new node is no longer single when the + // token reaches it). + if wasSingle && h.Ring.IsLeader() { + go func() { + time.Sleep(800 * time.Millisecond) + h.Ring.StartRing(context.Background()) + }() + } +} + // 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 818affe..0ff4359 100644 --- a/plan.md +++ b/plan.md @@ -110,6 +110,14 @@ - 每个节点都有 webui；用户登录到哪个节点的 webui，就由**该节点**执行新节点加入：令牌发到自己时，**先转发给新节点**，并把新节点加入令牌中的集群信息，**更新转发链**：新节点放在自己后面，自己原本的后继放在新节点之后。 +#### 增量日志同步（本机事件 → 令牌 → 全网一致） + +- 每个节点维护一份**集群事件日志（operation log）**：本机发生的变更（创建/删除转发、节点加入/离开、负载变化、leader 变更）以**递增序号**追加为本机日志条目。 +- 令牌第二轮（同步轮）携带**本机日志增量**：号码 > 上一周期已同步水位（`lastSyncedSeq`）的条目，随令牌环行时让其它节点**按序追加重放到本地日志**，从而实现全网**增量日志同步**。 +- 日志条目不可变（追加式）；节点收到增量后校验序号连续性（缺失则请求补齐），因每个节点持有完整集群信息，可从其它节点补拉。 +- 集群状态（拓扑/任务/leader）可**由日志重放得到**：新节点加入时，通过令牌/邻居拉取完整日志并重放，即可获得与其它节点一致的完整视图（满足"每个节点都掌握完整集群信息"）。 +- 该机制与"令牌承载中间配置文件/转发拓扑"正交共存：令牌同时携带〔拓扑快照〕与〔日志增量〕，快照用于即时校验，日志用于一致性追补。 + #### 实现里程碑（待按上述设计重建） - [ ] 数据面：TypeSet 集群状态（节点表/转发链/leader/轮次延迟/负载指标/待办任务），每节点一份完整副本