feat(M6): token-ring incremental log sync — append-only op log, delta rides token round-2, nodes converge on identical history (e2e verified join events)

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
+}