feat(raft): leader appends writes to raft log and waits for commit

src/barabadb/core/config.nim

@@ -40,6 +40,7 @@ type
     raftPeers*: seq[string]
     raftNodeId*: string
     raftPeerAddrs*: Table[string, tuple[host: string, port: int]]
+    raftWriteTimeoutMs*: int
 
   CompactionStrategy* = enum
     csSizeTiered = "size_tiered"
@@ -79,6 +80,7 @@ proc defaultConfig*(): BaraConfig =
     raftPeers: @[],
     raftNodeId: "",
     raftPeerAddrs: initTable[string, tuple[host: string, port: int]](),
+    raftWriteTimeoutMs: 5_000,
   )
 
 # ----------------------------------------------------------------------
@@ -204,6 +206,7 @@ proc loadConfigFromEnv*(cfg: var BaraConfig) =
         cfg.raftPeers.add(id)
         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)
 
 # ----------------------------------------------------------------------
 # Master Loader

src/barabadb/core/server.nim

@@ -6,6 +6,7 @@ import std/sequtils
 import std/tables
 import std/endians
 import std/monotimes
+import std/times
 import std/locks
 import std/nativesockets
 when defined(windows):
@@ -206,11 +207,43 @@ proc valueToWire(val: string, colType: string): WireValue =
     return WireValue(kind: fkJson, jsonVal: val)
   return WireValue(kind: fkString, strVal: val)
 
+proc appendWriteToRaft*(node: RaftNode, kvPairs: seq[(string, seq[byte])],
+                        timeoutMs: int): Future[(bool, string)] {.async.} =
+  ## C3b leader write path: append each written KV pair to the Raft log and
+  ## wait for majority commit. An empty value encodes a delete; the entry
+  ## format matches applyCommand ("put": key \x00 value, "delete": key).
+  ##
+  ## MUST be called from the async event-loop thread that owns `node` and
+  ## WITHOUT holding the storage gate: commitIndex advances via
+  ## handleAppendReply on the same loop, and applyCommand re-enters the
+  ## (non-reentrant) gate — waiting under the gate would deadlock the loop.
+  var lastIdx = 0'u64
+  for (key, value) in kvPairs:
+    let entry = if value.len > 0:
+        node.appendLog("put", cast[seq[byte]](key & "\x00" & cast[string](value)))
+      else:
+        node.appendLog("delete", cast[seq[byte]](key))
+    if entry.index == 0:
+      return (false, "lost leadership during raft append")
+    lastIdx = entry.index
+  let deadline = getMonoTime() + initDuration(milliseconds = timeoutMs)
+  while node.commitIndex < lastIdx and getMonoTime() < deadline:
+    await sleepAsync(10)
+  if node.commitIndex < lastIdx:
+    return (false, "raft commit timeout")
+  return (true, "")
+
 proc executeQuery(db: LSMTree, ctx: ExecutionContext, query: string, params: seq[WireValue] = @[],
                    replication: ReplicationManager = nil,
-                   raftNode: RaftNode = nil): (bool, QueryResult, string) =
+                   raftNode: RaftNode = nil,
+                   raftWriteTimeoutMs: int = 5000): Future[(bool, QueryResult, string)] {.async.} =
   ## All storage access is under the global StorageGate so HTTP worker threads
   ## and the TCP event loop never touch ORC-managed LSM/executor state concurrently.
+  ## The gate is released BEFORE the Raft commit wait — see appendWriteToRaft.
+  var ok = false
+  var qr = QueryResult()
+  var msg = ""
+  var kvPairs: seq[(string, seq[byte])] = @[]
   withStorageGate:
     try:
       let tokens = tokenize(query)
@@ -227,15 +260,16 @@ proc executeQuery(db: LSMTree, ctx: ExecutionContext, query: string, params: seq
 
       let res = executor.executeQuery(ctx, astNode, params)
       if res.success:
-        # Ship written key-value pairs to replicas
+        # Ship written key-value pairs to replicas (legacy path; skipped when
-        if replication != nil and res.keyValuePairs.len > 0:
+        # the raft path below handles the statement).
+        if raftNode == nil and replication != nil and res.keyValuePairs.len > 0:
           for (key, value) in res.keyValuePairs:
             var data = newSeq[byte](key.len + 1 + value.len)
             for i, c in key: data[i] = byte(c)
             data[key.len] = byte(0)
             for i, c in value: data[key.len + 1 + i] = c
             discard replication.writeLsn(data)
-        var qr = QueryResult(affectedRows: res.affectedRows, rowCount: res.rows.len)
+        qr = QueryResult(affectedRows: res.affectedRows, rowCount: res.rows.len)
         qr.columns = res.columns
 
         var colTypes: seq[string] = @[]
@@ -268,11 +302,20 @@ proc executeQuery(db: LSMTree, ctx: ExecutionContext, query: string, params: seq
             let cType = if i < colTypes.len: colTypes[i] else: ""
             wireRow.add(valueToWire(val, cType))
           qr.rows.add(wireRow)
-        return (true, qr, res.message)
+        ok = true
+        msg = res.message
+        kvPairs = res.keyValuePairs
       else:
         return (false, QueryResult(), res.message)
     except Exception as e:
       return (false, QueryResult(), e.msg)
+  # C3b: leader appends writes to the Raft log and waits for majority commit
+  # (outside the storage gate — see appendWriteToRaft).
+  if ok and raftNode != nil and kvPairs.len > 0:
+    let (raftOk, raftErr) = await appendWriteToRaft(raftNode, kvPairs, raftWriteTimeoutMs)
+    if not raftOk:
+      return (false, QueryResult(), raftErr)
+  return (ok, qr, msg)
 
 # ----------------------------------------------------------------------
 # Response Serialization
@@ -574,7 +617,7 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.}
 
         if shardCheck:
           let startTicks = getMonoTime().ticks()
-          let (success, result, errorMsg) = executeQuery(connCtx.db, connCtx, queryStr, replication=server.replicationManager, raftNode=server.raftNode)
+          let (success, result, errorMsg) = await executeQuery(connCtx.db, connCtx, queryStr, replication=server.replicationManager, raftNode=server.raftNode, raftWriteTimeoutMs=server.config.raftWriteTimeoutMs)
           let durationMs = int((getMonoTime().ticks() - startTicks) div 1_000_000)
 
           if durationMs >= slowThreshold:
@@ -594,7 +637,7 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.}
         info("[" & $clientId & "] QueryParams: " & queryStr & " (" & $params.len & " params)")
 
         let startTicks = getMonoTime().ticks()
-        let (success, result, errorMsg) = executeQuery(connCtx.db, connCtx, queryStr, params, replication=server.replicationManager, raftNode=server.raftNode)
+        let (success, result, errorMsg) = await executeQuery(connCtx.db, connCtx, queryStr, params, replication=server.replicationManager, raftNode=server.raftNode, raftWriteTimeoutMs=server.config.raftWriteTimeoutMs)
         let durationMs = int((getMonoTime().ticks() - startTicks) div 1_000_000)
 
         if durationMs >= slowThreshold:

tests/test_all.nim

@@ -2607,6 +2607,105 @@ suite "Raft Network Transport":
       check replyMsg.kind == rmkRequestVoteReply
       check replyMsg.success
 
+suite "Raft SQL Write Path":
+  test "leader append+commit wait round-trips through applyCommand":
+    proc scenario() =
+      var n1 = newRaftNode("n1", @["n2", "n3"], raftPort = 29031)
+      var n2 = newRaftNode("n2", @["n1", "n3"], raftPort = 29032)
+      var n3 = newRaftNode("n3", @["n1", "n2"], raftPort = 29033)
+
+      # Wide deterministic stagger: n1 wins the first election and 50ms
+      # heartbeats keep n2/n3 timers reset for the rest of the test
+      # (a tight 150/250/350 stagger lets n2's timer fire before the new
+      # leader's first heartbeat lands, causing a term-2/term-3 cascade).
+      n1.electionTimeout = 150
+      n2.electionTimeout = 1500
+      n3.electionTimeout = 3000
+
+      n1.peerAddrs["n2"] = ("127.0.0.1", 29032)
+      n1.peerAddrs["n3"] = ("127.0.0.1", 29033)
+      n2.peerAddrs["n1"] = ("127.0.0.1", 29031)
+      n2.peerAddrs["n3"] = ("127.0.0.1", 29033)
+      n3.peerAddrs["n1"] = ("127.0.0.1", 29031)
+      n3.peerAddrs["n2"] = ("127.0.0.1", 29032)
+
+      # Record every applyCommand invocation per node. NOTE: the recorder is
+      # built in a helper proc because Nim closures capture a for-loop `let`
+      # binding by reference — all iterations would share the final value.
+      var nodes = [n1, n2, n3]
+      var applied: array[3, seq[string]]
+      proc makeRecorder(i: int): proc(cmd: string, data: seq[byte]) {.gcsafe.} =
+        result = proc(cmd: string, data: seq[byte]) {.gcsafe.} =
+          applied[i].add(cmd & "|" & cast[string](data))
+      for i, n in nodes:
+        n.applyCommand = makeRecorder(i)
+
+      let net1 = newRaftNetwork(n1)
+      let net2 = newRaftNetwork(n2)
+      let net3 = newRaftNetwork(n3)
+
+      asyncCheck net1.run()
+      asyncCheck net2.run()
+      asyncCheck net3.run()
+      waitFor sleepAsync(50)
+
+      # Wait for the production timerLoop to elect exactly one leader
+      var leaderIdx = -1
+      var waited = 0
+      while waited < 3000:
+        leaderIdx = -1
+        var leaderCount = 0
+        for i, n in nodes:
+          if n.isLeader:
+            leaderIdx = i
+            inc leaderCount
+        if leaderCount == 1: break
+        waitFor sleepAsync(100)
+        waited += 100
+      check leaderIdx >= 0
+
+      if leaderIdx >= 0:
+        let leader = nodes[leaderIdx]
+        let followerIdx = (leaderIdx + 1) mod 3
+
+        # Server-side leader write path: append + wait for majority commit
+        let (ok, errMsg) = waitFor appendWriteToRaft(leader,
+          @[("users.1", cast[seq[byte]]("alice"))], timeoutMs = 3000)
+        check ok
+        if not ok: echo "appendWriteToRaft failed: ", errMsg
+
+        # Follower applies the committed entry shortly after (next heartbeat)
+        waited = 0
+        while applied[followerIdx].len == 0 and waited < 2000:
+          waitFor sleepAsync(50)
+          waited += 50
+        check applied[followerIdx].len >= 1
+        check "put|users.1\x00alice" in applied[followerIdx]
+
+      net1.stop()
+      net2.stop()
+      net3.stop()
+      waitFor sleepAsync(50)
+    scenario()
+
+  test "txn COMMIT delete kvPairs are empty-valued":
+    var testDir = getTempDir() / "baradb_raft_commit_del_" & $getCurrentProcessId() & "_" & $getMonoTime().ticks
+    createDir(testDir)
+    var db = newLSMTree(testDir)
+    var ctx = qexec.newExecutionContext(db)
+    discard qexec.executeQuery(ctx, parse("CREATE TABLE t (id INTEGER, v TEXT)"))
+    # Inserted OUTSIDE the txn: execDelete reads ctx.db directly and does not
+    # see rows buffered in the pending txn's writeSet.
+    discard qexec.executeQuery(ctx, parse("INSERT INTO t (id, v) VALUES (1, 'x')"))
+    discard qexec.executeQuery(ctx, parse("BEGIN"))
+    discard qexec.executeQuery(ctx, parse("DELETE FROM t WHERE id = 1"))
+    let res = qexec.executeQuery(ctx, parse("COMMIT"))
+    check res.success
+    # The deleted row's kvPair must carry an empty value so followers delete,
+    # not resurrect, the key.
+    check res.keyValuePairs.len == 1
+    check res.keyValuePairs[0][1].len == 0
+
 suite "CLI Autocomplete":
   test "Autocomplete commands":
     let res = autocomplete("he")