From 333941ab6515f11c6aeb7b9b10dce383a46d5136 Mon Sep 17 00:00:00 2001 From: dimgigov Date: Thu, 30 Jul 2026 19:36:16 +0300 Subject: [PATCH] feat(raft): leader appends writes to raft log and waits for commit --- src/barabadb/core/config.nim | 3 ++ src/barabadb/core/server.nim | 57 ++++++++++++++++++--- tests/test_all.nim | 99 ++++++++++++++++++++++++++++++++++++ 3 files changed, 152 insertions(+), 7 deletions(-) diff --git a/src/barabadb/core/config.nim b/src/barabadb/core/config.nim index 0438f2d..5ca577b 100644 --- a/src/barabadb/core/config.nim +++ b/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 diff --git a/src/barabadb/core/server.nim b/src/barabadb/core/server.nim index 3613e76..40a760d 100644 --- a/src/barabadb/core/server.nim +++ b/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 - if replication != nil and res.keyValuePairs.len > 0: + # Ship written key-value pairs to replicas (legacy path; skipped when + # 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: diff --git a/tests/test_all.nim b/tests/test_all.nim index d55b290..b24132f 100644 --- a/tests/test_all.nim +++ b/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")