feat(raft): compaction unpinned from stale peers (snapshot fallback)

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)

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)
 

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

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,