From 53704e10361dc245c52a34617b4ea97410180544 Mon Sep 17 00:00:00 2001 From: dimgigov Date: Thu, 30 Jul 2026 21:35:41 +0300 Subject: [PATCH] feat(raft): safe log compaction with snapshot metadata - lastSnapshotIndex/Term bound the compacted prefix; lastLogIndex/Term and AppendEntries prevLog checks respect the snapshot base - compactLog drops entries only through min(matchIndex, lastApplied) on the leader so lagging peers can still catch up via AppendEntries - Persist snapshot fields in raft_state.bin; BARADB_RAFT_LOG_MAX_ENTRIES - Fix commit-index scan to use findLogEntryByIndex (works after compact) --- docs/bg/distributed.md | 3 + docs/en/distributed.md | 3 + src/barabadb/core/config.nim | 3 + src/barabadb/core/raft.nim | 121 +++++++++++++++++++++++++++++------ src/baradadb.nim | 2 + tests/test_all.nim | 35 ++++++++++ 6 files changed, 146 insertions(+), 21 deletions(-) diff --git a/docs/bg/distributed.md b/docs/bg/distributed.md index 52f2e5d..2fe135f 100644 --- a/docs/bg/distributed.md +++ b/docs/bg/distributed.md @@ -17,9 +17,12 @@ Leader election и log репликация през TCP. Включване: | `BARADB_RAFT_PEERS` | Списък `id@host:port` (вкл. себе си) | | `BARADB_RAFT_WRITE_TIMEOUT_MS` | Макс. изчакване за majority commit при SQL записи (по подразбиране 5000) | | `BARADB_RAFT_CLIENT_PEERS` | Опционален `id@host:clientPort` map за leader write forwarding | +| `BARADB_RAFT_LOG_MAX_ENTRIES` | Лимит на in-memory raft log (по подразбиране 256); safe prefix compact | Когато Raft е активен, SQL DML и schema DDL се приемат само от лидера на **`default`**. DML отива като put/delete; DDL — като `ddl` запис. Followers **препращат** write/DDL към лидера, ако е зададен `BARADB_RAFT_CLIENT_PEERS`; иначе връщат `not leader; leader is '…'`. Записи към друга database name се отказват. `CREATE`/`DROP DATABASE` не се репликират. Приложен DML обновява и secondary индекси/графи. +**Log compaction (v1):** след apply node-ът може да изреже safe prefix, когато log-ът надхвърли `BARADB_RAFT_LOG_MAX_ENTRIES`. Leader не реже след matchIndex на peer (catch-up с AppendEntries). Snapshot metadata се пази в `raft_state.bin`. + ```nim import barabadb/core/raft diff --git a/docs/en/distributed.md b/docs/en/distributed.md index 4075b57..8d43fe8 100644 --- a/docs/en/distributed.md +++ b/docs/en/distributed.md @@ -17,9 +17,12 @@ Leader election and log replication over TCP. Enable with: | `BARADB_RAFT_PEERS` | Comma-separated `id@host:port` (include self) | | `BARADB_RAFT_WRITE_TIMEOUT_MS` | Max wait for majority commit on SQL writes (default 5000) | | `BARADB_RAFT_CLIENT_PEERS` | Optional `id@host:clientPort` map for leader write forwarding | +| `BARADB_RAFT_LOG_MAX_ENTRIES` | Soft cap on in-memory raft log length (default 256); safe prefix compact | When Raft is enabled, SQL DML (`INSERT`/`UPDATE`/`DELETE`/`MERGE` and transactional `COMMIT`) and schema DDL (`CREATE`/`DROP`/`ALTER` table, index, view, graph, …) are accepted only on the leader of the **`default`** database. DML ships as put/delete log entries; DDL ships as a `ddl` entry with the original SQL and is re-executed on every node at apply. Followers that receive a write/DDL **forward** it to the leader when `BARADB_RAFT_CLIENT_PEERS` maps the leader id to a SQL client address; otherwise they return `not leader; leader is '…'`. Writes against any other database name are rejected (`raft writes only supported on the 'default' database`). `CREATE`/`DROP DATABASE` are not raft-replicated (multi-DB is out of scope for v1). Committed DML also updates secondary B-tree/FTS/HNSW indexes and in-memory graphs. +**Log compaction (v1):** after apply, each node may drop a fully-safe log prefix once `log.len` exceeds `BARADB_RAFT_LOG_MAX_ENTRIES`. The leader never discards past any peer's `matchIndex` (so lagging followers still catch up via AppendEntries). Snapshot metadata (`lastSnapshotIndex`/`Term`) is persisted in `raft_state.bin`; full InstallSnapshot state-machine payloads are not required while this safe-prefix policy holds. + ```nim import barabadb/core/raft diff --git a/src/barabadb/core/config.nim b/src/barabadb/core/config.nim index 1629bc6..278a528 100644 --- a/src/barabadb/core/config.nim +++ b/src/barabadb/core/config.nim @@ -43,6 +43,7 @@ type ## SQL client ports for leader forwarding (id@host:clientPort). raftPeerClientAddrs*: Table[string, tuple[host: string, port: int]] raftWriteTimeoutMs*: int + raftLogMaxEntries*: int CompactionStrategy* = enum csSizeTiered = "size_tiered" @@ -84,6 +85,7 @@ proc defaultConfig*(): BaraConfig = raftPeerAddrs: initTable[string, tuple[host: string, port: int]](), raftPeerClientAddrs: initTable[string, tuple[host: string, port: int]](), raftWriteTimeoutMs: 5_000, + raftLogMaxEntries: 256, ) # ---------------------------------------------------------------------- @@ -210,6 +212,7 @@ proc loadConfigFromEnv*(cfg: var BaraConfig) = cfg.raftPeerAddrs[id] = (host, port) cfg.raftNodeId = getEnv("BARADB_RAFT_NODE_ID", cfg.raftNodeId) cfg.raftWriteTimeoutMs = parseEnvInt(getEnv("BARADB_RAFT_WRITE_TIMEOUT_MS", ""), cfg.raftWriteTimeoutMs) + cfg.raftLogMaxEntries = parseEnvInt(getEnv("BARADB_RAFT_LOG_MAX_ENTRIES", ""), cfg.raftLogMaxEntries) # Optional: client (SQL) addresses for leader write forwarding. # Same id@host:port shape as BARADB_RAFT_PEERS, but ports are BARADB_PORT values. let clientPeersEnv = getEnv("BARADB_RAFT_CLIENT_PEERS", "") diff --git a/src/barabadb/core/raft.nim b/src/barabadb/core/raft.nim index 2cc67ed..30a6c73 100644 --- a/src/barabadb/core/raft.nim +++ b/src/barabadb/core/raft.nim @@ -33,6 +33,14 @@ type log*: seq[LogEntry] commitIndex*: uint64 lastApplied*: uint64 + ## Compacted prefix: log entries with index <= lastSnapshotIndex are gone. + ## Safe compaction only discards entries every peer has already matched + ## (leader) or that this node has applied (follower), so catch-up via + ## AppendEntries still works without InstallSnapshot payloads. + lastSnapshotIndex*: uint64 + lastSnapshotTerm*: uint64 + ## Trigger compaction when log.len exceeds this (0 = default 256). + logMaxEntries*: int # State machine callback applyCommand*: proc(cmd: string, data: seq[byte]) {.gcsafe.} # Distributed transaction callbacks (for raft→disttxn integration) @@ -100,6 +108,9 @@ proc saveState(node: RaftNode) = s.write(uint32(entry.data.len)) if entry.data.len > 0: s.writeData(addr entry.data[0], entry.data.len) + # Snapshot base (appended for backward-compatible load of older files) + s.write(node.lastSnapshotIndex) + s.write(node.lastSnapshotTerm) s.close() moveFile(tmpPath, path) @@ -133,6 +144,16 @@ proc loadState(node: RaftNode) = if s.readData(addr data[0], dataLen) != dataLen: raise newException(IOError, "Incomplete Raft log data read") node.log[i] = LogEntry(term: term, index: index, command: cmd, data: data) + # Optional trailing snapshot fields (absent in pre-compaction state files) + if not s.atEnd: + node.lastSnapshotIndex = s.readUint64() + if not s.atEnd: + node.lastSnapshotTerm = s.readUint64() + # lastApplied/commitIndex must not sit below the compacted base + if node.lastApplied < node.lastSnapshotIndex: + node.lastApplied = node.lastSnapshotIndex + if node.commitIndex < node.lastSnapshotIndex: + node.commitIndex = node.lastSnapshotIndex except IOError, OSError: echo "[WARN] Failed to load Raft state from ", path, ": ", getCurrentExceptionMsg() s.close() @@ -148,6 +169,9 @@ proc newRaftNode*(id: string, peers: seq[string], raftPort: int = 0, log: @[], commitIndex: 0, lastApplied: 0, + lastSnapshotIndex: 0, + lastSnapshotTerm: 0, + logMaxEntries: 256, nextIndex: initTable[string, uint64](), matchIndex: initTable[string, uint64](), peers: peers, @@ -176,12 +200,12 @@ proc addNode*(cluster: RaftCluster, id: string) = proc lastLogIndex*(node: RaftNode): uint64 = if node.log.len == 0: - return 0 + return node.lastSnapshotIndex return node.log[^1].index proc lastLogTerm*(node: RaftNode): uint64 = if node.log.len == 0: - return 0 + return node.lastSnapshotTerm return node.log[^1].term proc findLogEntryByIndex(node: RaftNode, index: uint64): int = @@ -192,9 +216,52 @@ proc findLogEntryByIndex(node: RaftNode, index: uint64): int = return i return -1 +proc termAtIndex(node: RaftNode, index: uint64): uint64 = + ## Term of the log entry (or snapshot base) at `index`, or 0 if unknown. + if index == 0: return 0 + if index == node.lastSnapshotIndex: return node.lastSnapshotTerm + let pos = node.findLogEntryByIndex(index) + if pos >= 0: return node.log[pos].term + return 0 + +proc compactLog*(node: RaftNode) = + ## Drop a fully-replicated / applied log prefix so the in-memory log stays + ## bounded. Leader: never discard past any peer's matchIndex (catch-up via + ## AppendEntries remains possible). Follower: discard through lastApplied. + let maxEntries = if node.logMaxEntries > 0: node.logMaxEntries else: 256 + if node.log.len <= maxEntries: + return + var through = node.lastApplied + if node.state == rsLeader and node.peers.len > 0: + var minMatch = through + for peer in node.peers: + let m = node.matchIndex.getOrDefault(peer, 0'u64) + if m < minMatch: minMatch = m + through = minMatch + if through <= node.lastSnapshotIndex: + return + let pos = node.findLogEntryByIndex(through) + if pos < 0: + return + node.lastSnapshotTerm = node.log[pos].term + node.lastSnapshotIndex = through + if pos + 1 < node.log.len: + node.log = node.log[(pos + 1) .. ^1] + else: + node.log = @[] + # Keep lastApplied/commit at least at the snapshot base + if node.lastApplied < node.lastSnapshotIndex: + node.lastApplied = node.lastSnapshotIndex + if node.commitIndex < node.lastSnapshotIndex: + node.commitIndex = node.lastSnapshotIndex + node.saveState() + proc applyCommitted(node: RaftNode) = while node.lastApplied < node.commitIndex: inc node.lastApplied + # Entries at/below the snapshot base were already applied before compact. + if node.lastApplied <= node.lastSnapshotIndex: + continue let pos = node.findLogEntryByIndex(node.lastApplied) if pos >= 0: let entry = node.log[pos] @@ -213,6 +280,7 @@ proc applyCommitted(node: RaftNode) = else: if node.applyCommand != nil: node.applyCommand(entry.command, entry.data) + node.compactLog() proc becomeFollower*(node: RaftNode, term: uint64) = node.state = rsFollower @@ -283,13 +351,20 @@ proc handleAppendEntries*(node: RaftNode, msg: RaftMessage): RaftMessage = # Check if log contains entry at prevLogIndex with prevLogTerm if msg.prevLogIndex > 0: - let prevPos = node.findLogEntryByIndex(msg.prevLogIndex) - if prevPos < 0: - return reply - if node.log[prevPos].term != msg.prevLogTerm: - # Delete conflicting entries - node.log.setLen(prevPos) + if msg.prevLogIndex < node.lastSnapshotIndex: + # Leader is behind our snapshot base — reject return reply + if msg.prevLogIndex == node.lastSnapshotIndex: + if msg.prevLogTerm != node.lastSnapshotTerm: + return reply + else: + let prevPos = node.findLogEntryByIndex(msg.prevLogIndex) + if prevPos < 0: + return reply + if node.log[prevPos].term != msg.prevLogTerm: + # Delete conflicting entries + node.log.setLen(prevPos) + return reply # Append new entries var logChanged = false @@ -328,13 +403,13 @@ proc requestVote*(node: RaftNode): seq[RaftMessage] = )) proc appendEntries*(node: RaftNode, peerId: string): RaftMessage = - let nextIdx = node.nextIndex.getOrDefault(peerId, node.lastLogIndex + 1) + var nextIdx = node.nextIndex.getOrDefault(peerId, node.lastLogIndex + 1) + # Never try to send entries already discarded by our snapshot base. + if nextIdx <= node.lastSnapshotIndex: + nextIdx = node.lastSnapshotIndex + 1 + node.nextIndex[peerId] = nextIdx let prevIdx = nextIdx - 1 - var prevTerm: uint64 = 0 - if prevIdx > 0: - let prevPos = node.findLogEntryByIndex(prevIdx) - if prevPos >= 0: - prevTerm = node.log[prevPos].term + let prevTerm = node.termAtIndex(prevIdx) var entries: seq[LogEntry] = @[] let startPos = node.findLogEntryByIndex(nextIdx) @@ -398,29 +473,33 @@ proc handleAppendReply*(node: RaftNode, peerId: string, reply: RaftMessage) = # Update commit index using true majority calculation let majority = (node.peers.len + 1 + 1) div 2 # majority of cluster (peers + leader) var newCommitIdx = node.commitIndex - - # Check each index from highest to current commitIndex+1 + + # Walk logical indices high→low via findLogEntryByIndex (log may be compacted). for idx in countdown(int(node.lastLogIndex), int(node.commitIndex) + 1): if idx <= 0: break + let pos = node.findLogEntryByIndex(uint64(idx)) + if pos < 0: + continue # Only commit entries from current term (Raft safety property) - if uint64(idx) <= node.lastLogIndex and node.log[idx - 1].term == node.currentTerm: - # Count how many nodes have replicated this index + if node.log[pos].term == node.currentTerm: var count = 1 # Leader itself for peerId2, mIdx in node.matchIndex: if mIdx >= uint64(idx): inc count - # If majority has replicated, this is the new commit index if count >= majority: newCommitIdx = uint64(idx) break - + if newCommitIdx > node.commitIndex: node.commitIndex = newCommitIdx node.applyCommitted() else: - if node.nextIndex[peerId] > 1: + let floor = node.lastSnapshotIndex + 1 + if node.nextIndex.getOrDefault(peerId, 1) > floor: dec node.nextIndex[peerId] + else: + node.nextIndex[peerId] = floor proc state*(node: RaftNode): RaftState = node.state proc isLeader*(node: RaftNode): bool = node.state == rsLeader diff --git a/src/baradadb.nim b/src/baradadb.nim index bcf9cb3..c5534b0 100644 --- a/src/baradadb.nim +++ b/src/baradadb.nim @@ -343,6 +343,8 @@ proc main() = var raftNode = newRaftNode(config.raftNodeId, raftPeers, config.raftPort, dataDir = raftDataDir) raftNode.peerAddrs = config.raftPeerAddrs + if config.raftLogMaxEntries > 0: + raftNode.logMaxEntries = config.raftLogMaxEntries tcpServer.raftNode = raftNode # C3b: executeQuery rejects writes on followers # Wire state machine: committed entries update LSM + secondary indexes # (B-tree/FTS/HNSW/graphs) and re-execute schema DDL on every node. diff --git a/tests/test_all.nim b/tests/test_all.nim index b5abd7b..a694222 100644 --- a/tests/test_all.nim +++ b/tests/test_all.nim @@ -2706,6 +2706,41 @@ suite "Raft SQL Write Path": check res.keyValuePairs.len == 1 check res.keyValuePairs[0][1].len == 0 + test "compactLog discards applied prefix and preserves lastSnapshot base": + var n = newRaftNode("n1", @[], raftPort = 29120) + n.logMaxEntries = 8 + n.becomeLeader() + # Single-node: simulate applied entries then compact. + for i in 1 .. 20: + let e = n.appendLog("put", cast[seq[byte]]("k" & $i & "\x00v")) + check e.index == uint64(i) + n.commitIndex = e.index + n.lastApplied = e.index + n.compactLog() + check n.log.len <= n.logMaxEntries + check n.lastSnapshotIndex > 0 + check n.lastLogIndex == 20 + let e21 = n.appendLog("put", cast[seq[byte]]("k21\x00v")) + check e21.index == 21 + n.commitIndex = 21 + n.lastApplied = 21 + n.compactLog() + check n.lastLogIndex == 21 + + test "leader compactLog never discards past a lagging peer matchIndex": + var n = newRaftNode("n1", @["n2"], raftPort = 29121) + n.logMaxEntries = 5 + n.becomeLeader() + n.matchIndex["n2"] = 0 # peer never caught up + for i in 1 .. 15: + discard n.appendLog("put", cast[seq[byte]]("x")) + n.commitIndex = uint64(i) + n.lastApplied = uint64(i) + n.compactLog() + # With matchIndex[n2]=0, through=0 — no compaction past snapshot 0 + check n.lastSnapshotIndex == 0 + check n.log.len == 15 + test "appendDdlToRaft fails when node is not leader": var n = newRaftNode("n1", @["n2"], raftPort = 29113) let (ok, err) = waitFor appendDdlToRaft(n,