// 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). // The entry is treated as already-synced (Synced advanced with Seq): a locally // appended entry exists on this node and will be broadcast; the next full-log // attachment from a peer must NOT re-apply it as a duplicate. Without this, // a node that appends seq=N while its Synced lags behind would later receive // its own seq=N in a peer's full attachment and ApplyDelta would re-add it // (duplicate rows in the audit log). 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) if e.Seq > l.Synced { l.Synced = e.Seq } 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 } // DetailOf produces a human-readable one-line summary of a log entry's // payload, matching the frontend's detailOf formatting. Used by the audit CSV // export and anywhere a flat text rendering of an entry is needed. func DetailOf(e LogEntry) string { if len(e.Data) == 0 { return "" } var d map[string]any if err := json.Unmarshal(e.Data, &d); err != nil { return string(e.Data) } str := func(key string) string { s, _ := d[key].(string); return s } switch e.Kind { case LogForwardAdd, LogForwardRemove: s := str("local") + " → " + str("remote") if id := str("taskId"); id != "" { if len(id) > 8 { id = id[len(id)-8:] } s += " · " + id } return s case LogNodeJoin: if addr := str("addr"); addr != "" { return str("node") + " @ " + addr } return str("node") case LogNodeLeave: return str("node") case LogLeaderChange: return "→ " + str("leader") case LogTaskClaimed: return fmt.Sprintf("%s→%s:%v", str("local"), str("remote"), d["port"]) default: if len(d) > 0 { b, _ := json.Marshal(d) return string(b) } return "" } }