From 9ff9c2f6bef95914624275be0c8e51f2ec57f696 Mon Sep 17 00:00:00 2001 From: dimgigov Date: Fri, 31 Jul 2026 03:12:06 +0300 Subject: [PATCH] feat(raft): compaction unpinned from stale peers (snapshot fallback) --- src/barabadb/core/config.nim | 3 +++ src/barabadb/core/raft.nim | 29 ++++++++++++++++++++-- src/baradadb.nim | 2 ++ tests/test_all.nim | 47 ++++++++++++++++++++++++++++++++++++ 4 files changed, 79 insertions(+), 2 deletions(-) diff --git a/src/barabadb/core/config.nim b/src/barabadb/core/config.nim index 744ac01..068cd27 100644 --- a/src/barabadb/core/config.nim +++ b/src/barabadb/core/config.nim @@ -45,6 +45,7 @@ type raftWriteTimeoutMs*: int raftLogMaxEntries*: int raftSnapChunkKb*: int + raftPeerStaleMs*: int raftTlsEnabled*: bool raftTlsCertFile*: string raftTlsKeyFile*: string @@ -93,6 +94,7 @@ proc defaultConfig*(): BaraConfig = raftWriteTimeoutMs: 5_000, raftLogMaxEntries: 256, raftSnapChunkKb: 256, + raftPeerStaleMs: 30000, raftTlsEnabled: false, raftTlsCertFile: "", raftTlsKeyFile: "", @@ -226,6 +228,7 @@ proc loadConfigFromEnv*(cfg: var BaraConfig) = cfg.raftWriteTimeoutMs = parseEnvInt(getEnv("BARADB_RAFT_WRITE_TIMEOUT_MS", ""), cfg.raftWriteTimeoutMs) cfg.raftLogMaxEntries = parseEnvInt(getEnv("BARADB_RAFT_LOG_MAX_ENTRIES", ""), cfg.raftLogMaxEntries) cfg.raftSnapChunkKb = parseEnvInt(getEnv("BARADB_RAFT_SNAP_CHUNK_KB", ""), cfg.raftSnapChunkKb) + cfg.raftPeerStaleMs = parseEnvInt(getEnv("BARADB_RAFT_PEER_STALE_MS", ""), cfg.raftPeerStaleMs) cfg.raftTlsEnabled = parseEnvBool(getEnv("BARADB_RAFT_TLS_ENABLED", ""), cfg.raftTlsEnabled) cfg.raftTlsCertFile = getEnv("BARADB_RAFT_TLS_CERT_FILE", cfg.raftTlsCertFile) cfg.raftTlsKeyFile = getEnv("BARADB_RAFT_TLS_KEY_FILE", cfg.raftTlsKeyFile) diff --git a/src/barabadb/core/raft.nim b/src/barabadb/core/raft.nim index 7f1119d..755f3d7 100644 --- a/src/barabadb/core/raft.nim +++ b/src/barabadb/core/raft.nim @@ -67,6 +67,15 @@ type # Leader state nextIndex*: Table[string, uint64] matchIndex*: Table[string, uint64] + ## Monotonic ms of the last successful AppendEntries reply per peer, + ## initialized to "now" in becomeLeader (grace window) and bumped in + ## handleAppendReply. Leader compaction excludes peers silent longer than + ## raftPeerStaleMs from its minMatch — a stale peer no longer pins the log + ## forever; it is caught up via InstallSnapshot on return (T9). + matchIndexSeenMs*: Table[string, int64] + ## Stale window in ms (BARADB_RAFT_PEER_STALE_MS, default 30000; + ## 0 = default). + raftPeerStaleMs*: int # Cluster peers*: seq[string] leaderId*: string @@ -219,6 +228,8 @@ proc newRaftNode*(id: string, peers: seq[string], raftPort: int = 0, metrics: RaftMetrics(), nextIndex: initTable[string, uint64](), matchIndex: initTable[string, uint64](), + matchIndexSeenMs: initTable[string, int64](), + raftPeerStaleMs: 30000, peers: peers, leaderId: "", electionTimeout: 150 + rand(150), @@ -277,15 +288,22 @@ proc termAtIndex(node: RaftNode, index: uint64): uint64 = 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. + ## bounded. Leader: never discard past any responsive peer's matchIndex + ## (catch-up via AppendEntries remains possible); peers silent longer than + ## raftPeerStaleMs are excluded and catch up via InstallSnapshot instead. + ## 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: + let staleMs = if node.raftPeerStaleMs > 0: node.raftPeerStaleMs else: 30000 + let nowMs = getMonoTime().ticks() div 1_000_000 var minMatch = through for peer in node.peers: + let seenMs = node.matchIndexSeenMs.getOrDefault(peer, 0) + if seenMs <= 0 or nowMs - seenMs > staleMs.int64: + continue # stale peer — unpinned, snapshotted on return (T9) let m = node.matchIndex.getOrDefault(peer, 0'u64) if m < minMatch: minMatch = m through = minMatch @@ -346,6 +364,7 @@ proc becomeFollower*(node: RaftNode, term: uint64) = node.votesReceived.clear() node.nextIndex.clear() node.matchIndex.clear() + node.matchIndexSeenMs.clear() # Leader-only snapshot-send state is meaningless once we step down node.snapRejectStreak.clear() node.snapPending.clear() @@ -367,9 +386,14 @@ proc becomeLeader*(node: RaftNode) = if node.metrics != nil: inc node.metrics.electionsTotal info("Raft node " & node.id & " became leader for term " & $node.currentTerm) + let nowMs = getMonoTime().ticks() div 1_000_000 + node.matchIndexSeenMs.clear() for peer in node.peers: node.nextIndex[peer] = node.lastLogIndex + 1 node.matchIndex[peer] = 0 + # Grace window: an unreplied peer still pins compaction until it has been + # silent for raftPeerStaleMs since this leadership began. + node.matchIndexSeenMs[peer] = nowMs node.snapRejectStreak.clear() node.snapPending.clear() @@ -618,6 +642,7 @@ proc handleAppendReply*(node: RaftNode, peerId: string, reply: RaftMessage) = if reply.success: node.matchIndex[peerId] = reply.matchIdx node.nextIndex[peerId] = reply.matchIdx + 1 + node.matchIndexSeenMs[peerId] = getMonoTime().ticks() div 1_000_000 node.snapRejectStreak.del(peerId) node.snapPending.excl(peerId) diff --git a/src/baradadb.nim b/src/baradadb.nim index 5bdb894..209ab3f 100644 --- a/src/baradadb.nim +++ b/src/baradadb.nim @@ -368,6 +368,8 @@ proc main() = raftNode.logMaxEntries = config.raftLogMaxEntries if config.raftSnapChunkKb > 0: raftNode.snapChunkBytes = config.raftSnapChunkKb * 1024 + if config.raftPeerStaleMs > 0: + raftNode.raftPeerStaleMs = config.raftPeerStaleMs tcpServer.raftNode = raftNode # C3b: executeQuery rejects writes on followers httpServer.raftNode = raftNode # /metrics + /health raft gauges # Wire state machine: committed entries update LSM + secondary indexes diff --git a/tests/test_all.nim b/tests/test_all.nim index 5d758c5..738acc2 100644 --- a/tests/test_all.nim +++ b/tests/test_all.nim @@ -3338,6 +3338,53 @@ suite "Raft SQL Write Path": check n.lastSnapshotIndex == 0 check n.log.len == 15 + test "leader compactLog unpins from a stale peer (never replied, stale window exceeded)": + var n = newRaftNode("n1", @["n2"], raftPort = 29123) + n.logMaxEntries = 5 + n.raftPeerStaleMs = 1000 + n.becomeLeader() + n.matchIndex["n2"] = 0 # peer never caught up + # Last successful reply is long past the stale window. + n.matchIndexSeenMs["n2"] = getMonoTime().ticks() div 1_000_000 - 60_000 + for i in 1 .. 15: + discard n.appendLog("put", cast[seq[byte]]("x")) + n.commitIndex = uint64(i) + n.lastApplied = uint64(i) + n.compactLog() + # Stale peer excluded from minMatch — compaction runs through lastApplied. + check n.lastSnapshotIndex == 15 + check n.log.len == 0 + + test "leader compactLog still pins at a recently-responsive peer matchIndex": + var n = newRaftNode("n1", @["n2"], raftPort = 29124) + n.logMaxEntries = 5 + n.raftPeerStaleMs = 1000 + n.becomeLeader() + n.matchIndex["n2"] = 3 + n.matchIndexSeenMs["n2"] = getMonoTime().ticks() div 1_000_000 # just replied + for i in 1 .. 15: + discard n.appendLog("put", cast[seq[byte]]("x")) + n.commitIndex = uint64(i) + n.lastApplied = uint64(i) + n.compactLog() + check n.lastSnapshotIndex == 3 + check n.log.len == 12 + + test "leader compactLog respects grace window for an unreplied peer": + var n = newRaftNode("n1", @["n2"], raftPort = 29125) + n.logMaxEntries = 5 + n.raftPeerStaleMs = 30000 + n.becomeLeader() # matchIndexSeenMs initialized to now — within grace + n.matchIndex["n2"] = 0 # peer has not replied yet + for i in 1 .. 15: + discard n.appendLog("put", cast[seq[byte]]("x")) + n.commitIndex = uint64(i) + n.lastApplied = uint64(i) + n.compactLog() + # Grace window still active — peer pins compaction at matchIndex 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,