From 9df83163051390e41b07332d54f719b8745af32b Mon Sep 17 00:00:00 2001 From: dimgigov Date: Thu, 30 Jul 2026 21:31:05 +0300 Subject: [PATCH] feat(raft): transparent leader write/DDL forwarding MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - BARADB_RAFT_CLIENT_PEERS maps node id → SQL client host:port - Followers proxy DML/DDL to the known leader over the wire protocol (falls back to "not leader" when the map is missing) - E2E: follower CREATE/INSERT succeed via forward; docs updated --- docs/bg/distributed.md | 3 +- docs/en/distributed.md | 3 +- src/barabadb/core/config.nim | 27 +++++ src/barabadb/core/server.nim | 207 +++++++++++++++++++++++++-------- tests/bugfix_test.nim | 9 ++ tests/raft_writes_e2e_test.nim | 51 ++++---- 6 files changed, 228 insertions(+), 72 deletions(-) diff --git a/docs/bg/distributed.md b/docs/bg/distributed.md index 50f857b..52f2e5d 100644 --- a/docs/bg/distributed.md +++ b/docs/bg/distributed.md @@ -16,8 +16,9 @@ Leader election и log репликация през TCP. Включване: | `BARADB_RAFT_PORT` | Raft 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 | -Когато Raft е активен, SQL DML (`INSERT`/`UPDATE`/`DELETE`/`MERGE` и транзакционен `COMMIT`) и schema DDL (`CREATE`/`DROP`/`ALTER` table, index, view, graph, …) се приемат само от лидера на **`default`** базата. DML отива като put/delete; DDL — като `ddl` запис с оригиналния SQL, преизпълнен на всеки възел при apply. Followers отказват и двете с `not leader; leader is '…'`. Записи към друга database name се отказват. `CREATE`/`DROP DATABASE` не се репликират през raft (multi-DB е извън v1). Приложен DML обновява и secondary B-tree/FTS/HNSW индекси и in-memory графи. +Когато 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 индекси/графи. ```nim import barabadb/core/raft diff --git a/docs/en/distributed.md b/docs/en/distributed.md index bc0b865..4075b57 100644 --- a/docs/en/distributed.md +++ b/docs/en/distributed.md @@ -16,8 +16,9 @@ Leader election and log replication over TCP. Enable with: | `BARADB_RAFT_PORT` | Raft TCP port | | `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 | -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 reject both with `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. +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. ```nim import barabadb/core/raft diff --git a/src/barabadb/core/config.nim b/src/barabadb/core/config.nim index 5ca577b..1629bc6 100644 --- a/src/barabadb/core/config.nim +++ b/src/barabadb/core/config.nim @@ -40,6 +40,8 @@ type raftPeers*: seq[string] raftNodeId*: string raftPeerAddrs*: Table[string, tuple[host: string, port: int]] + ## SQL client ports for leader forwarding (id@host:clientPort). + raftPeerClientAddrs*: Table[string, tuple[host: string, port: int]] raftWriteTimeoutMs*: int CompactionStrategy* = enum @@ -80,6 +82,7 @@ proc defaultConfig*(): BaraConfig = raftPeers: @[], raftNodeId: "", raftPeerAddrs: initTable[string, tuple[host: string, port: int]](), + raftPeerClientAddrs: initTable[string, tuple[host: string, port: int]](), raftWriteTimeoutMs: 5_000, ) @@ -207,6 +210,30 @@ 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) + # 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", "") + if clientPeersEnv.len > 0: + cfg.raftPeerClientAddrs = initTable[string, tuple[host: string, port: int]]() + for raw in clientPeersEnv.split(","): + let entry = raw.strip() + if entry.len == 0: continue + let atPos = entry.rfind('@') + if atPos < 0: + raise newException(ValueError, + "Invalid BARADB_RAFT_CLIENT_PEERS entry '" & entry & "': expected id@host:port") + let id = entry[0 ..< atPos] + let hostPort = entry[atPos + 1 .. ^1] + let colonPos = hostPort.rfind(':') + let host = if colonPos >= 0: hostPort[0 ..< colonPos] else: "" + let portStr = if colonPos >= 0: hostPort[colonPos + 1 .. ^1] else: "" + var port = 0 + try: port = parseInt(portStr) + except ValueError: discard + if id.len == 0 or host.len == 0 or port < 1 or port > 65535: + raise newException(ValueError, + "Invalid BARADB_RAFT_CLIENT_PEERS entry '" & entry & "': expected id@host:port with port 1-65535") + cfg.raftPeerClientAddrs[id] = (host, port) # ---------------------------------------------------------------------- # Master Loader diff --git a/src/barabadb/core/server.nim b/src/barabadb/core/server.nim index da318d4..c23ec75 100644 --- a/src/barabadb/core/server.nim +++ b/src/barabadb/core/server.nim @@ -207,6 +207,90 @@ proc valueToWire(val: string, colType: string): WireValue = return WireValue(kind: fkJson, jsonVal: val) return WireValue(kind: fkString, strVal: val) +proc forwardRecvExact(sock: AsyncSocket, size: int): Future[string] {.async.} = + var buf = "" + while buf.len < size: + let chunk = await sock.recv(size - buf.len) + if chunk.len == 0: break + buf.add(chunk) + return buf + +proc forwardQueryToLeader*(host: string, port: int, query: string, + params: seq[WireValue] = @[], + timeoutMs: int = 5000): Future[(bool, QueryResult, string)] {.async.} = + ## Proxy a write/DDL to the known leader's SQL port. Used by followers when + ## BARADB_RAFT_CLIENT_PEERS maps leader id → host:clientPort. + var sock: AsyncSocket = nil + try: + sock = newAsyncSocket() + let okConn = await withTimeout(sock.connect(host, Port(port)), min(timeoutMs, 2000)) + if not okConn: + return (false, QueryResult(), "leader forward connect timeout") + let reqId = 1'u32 + let msg = if params.len > 0: + makeQueryParamsMessage(reqId, query, params) + else: + makeQueryMessage(reqId, query) + await sock.send(cast[string](msg)) + + var qr = QueryResult() + var gotComplete = false + while true: + let headerData = await forwardRecvExact(sock, 12) + if headerData.len < 12: + break + var hbytes = newSeq[byte](headerData.len) + for i, c in headerData: hbytes[i] = byte(c) + var pos = 0 + let kind = MsgKind(readUint32(hbytes, pos)) + let length = int(readUint32(hbytes, pos)) + discard readUint32(hbytes, pos) # requestId + let payloadStr = if length > 0: await forwardRecvExact(sock, length) else: "" + if payloadStr.len < length: + break + var payload = newSeq[byte](payloadStr.len) + for i, c in payloadStr: payload[i] = byte(c) + case kind + of mkError: + var epos = 0 + discard readUint32(payload, epos) + let emsg = readString(payload, epos) + return (false, QueryResult(), emsg) + of mkData: + var dpos = 0 + let colCount = int(readUint32(payload, dpos)) + qr.columns = @[] + for i in 0 ..< colCount: + qr.columns.add(readString(payload, dpos)) + qr.columnTypes = @[] + for i in 0 ..< colCount: + qr.columnTypes.add(FieldKind(payload[dpos])) + inc dpos + let rowCount = int(readUint32(payload, dpos)) + qr.rowCount = rowCount + qr.rows = @[] + for r in 0 ..< rowCount: + var row: seq[WireValue] = @[] + for c in 0 ..< colCount: + row.add(deserializeValue(payload, dpos)) + qr.rows.add(row) + of mkComplete: + var cpos = 0 + if payload.len >= 4: + qr.affectedRows = int(readUint32(payload, cpos)) + gotComplete = true + break + else: + discard + if gotComplete: + return (true, qr, "") + return (false, QueryResult(), "leader forward incomplete response") + except CatchableError as e: + return (false, QueryResult(), "leader forward failed: " & e.msg) + finally: + if sock != nil: + try: sock.close() except CatchableError: discard + proc waitRaftCommit(node: RaftNode, lastIdx: uint64, timeoutMs: int): Future[(bool, string)] {.async.} = let deadline = getMonoTime() + initDuration(milliseconds = timeoutMs) while node.commitIndex < lastIdx and getMonoTime() < deadline: @@ -249,7 +333,9 @@ proc appendDdlToRaft*(node: RaftNode, sql: string, proc executeQuery(db: LSMTree, ctx: ExecutionContext, query: string, params: seq[WireValue] = @[], replication: ReplicationManager = nil, raftNode: RaftNode = nil, - raftWriteTimeoutMs: int = 5000): Future[(bool, QueryResult, string)] {.async.} = + raftWriteTimeoutMs: int = 5000, + raftPeerClientAddrs: Table[string, tuple[host: string, port: int]] = + initTable[string, tuple[host: string, port: int]]()): 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. @@ -258,6 +344,9 @@ proc executeQuery(db: LSMTree, ctx: ExecutionContext, query: string, params: seq var msg = "" var kvPairs: seq[(string, seq[byte])] = @[] var needsRaftDdl = false + var needsForward = false + var forwardHost = "" + var forwardPort = 0 withStorageGate: try: let tokens = tokenize(query) @@ -282,59 +371,71 @@ proc executeQuery(db: LSMTree, ctx: ExecutionContext, query: string, params: seq dbName & "'") if raftNode.state != rsLeader: let who = if raftNode.leaderId.len > 0: raftNode.leaderId else: "none elected" - return (false, QueryResult(), "not leader; leader is '" & who & "'") + # Transparent leader forwarding when client SQL addresses are known. + if who != "none elected" and who in raftPeerClientAddrs: + let peerAddr = raftPeerClientAddrs[who] + needsForward = true + forwardHost = peerAddr.host + forwardPort = peerAddr.port + else: + return (false, QueryResult(), "not leader; leader is '" & who & "'") - let res = executor.executeQuery(ctx, astNode, params) - if res.success: - # 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) - qr = QueryResult(affectedRows: res.affectedRows, rowCount: res.rows.len) - qr.columns = res.columns + if not needsForward: + let res = executor.executeQuery(ctx, astNode, params) + if res.success: + # 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) + qr = QueryResult(affectedRows: res.affectedRows, rowCount: res.rows.len) + qr.columns = res.columns - var colTypes: seq[string] = @[] - var tableName = "" - if astNode.stmts[0].kind == nkSelect and astNode.stmts[0].selFrom != nil: - tableName = astNode.stmts[0].selFrom.fromTable - elif astNode.stmts[0].kind == nkInsert: - tableName = astNode.stmts[0].insTarget - elif astNode.stmts[0].kind == nkUpdate: - tableName = astNode.stmts[0].updTarget + var colTypes: seq[string] = @[] + var tableName = "" + if astNode.stmts[0].kind == nkSelect and astNode.stmts[0].selFrom != nil: + tableName = astNode.stmts[0].selFrom.fromTable + elif astNode.stmts[0].kind == nkInsert: + tableName = astNode.stmts[0].insTarget + elif astNode.stmts[0].kind == nkUpdate: + tableName = astNode.stmts[0].updTarget - if tableName.len > 0 and tableName in ctx.tables: - let tbl = ctx.tables[tableName] - for col in res.columns: - var found = "" - for c in tbl.columns: - if c.name.toLower() == col.toLower(): - found = c.colType - break - colTypes.add(found) + if tableName.len > 0 and tableName in ctx.tables: + let tbl = ctx.tables[tableName] + for col in res.columns: + var found = "" + for c in tbl.columns: + if c.name.toLower() == col.toLower(): + found = c.colType + break + colTypes.add(found) + else: + colTypes = newSeq[string](res.columns.len) + + qr.columnTypes = colTypes.mapIt(typeToFieldKind(it)) + qr.rows = @[] + for row in res.rows: + var wireRow: seq[WireValue] = @[] + for i, col in res.columns: + let val = if col in row: valueToString(row[col]) else: "\\N" + let cType = if i < colTypes.len: colTypes[i] else: "" + wireRow.add(valueToWire(val, cType)) + qr.rows.add(wireRow) + ok = true + msg = res.message + kvPairs = res.keyValuePairs else: - colTypes = newSeq[string](res.columns.len) - - qr.columnTypes = colTypes.mapIt(typeToFieldKind(it)) - qr.rows = @[] - for row in res.rows: - var wireRow: seq[WireValue] = @[] - for i, col in res.columns: - let val = if col in row: valueToString(row[col]) else: "\\N" - let cType = if i < colTypes.len: colTypes[i] else: "" - wireRow.add(valueToWire(val, cType)) - qr.rows.add(wireRow) - ok = true - msg = res.message - kvPairs = res.keyValuePairs - else: - return (false, QueryResult(), res.message) + return (false, QueryResult(), res.message) except Exception as e: return (false, QueryResult(), e.msg) + # Follower write/DDL: proxy to leader SQL port (outside the storage gate). + if needsForward: + return await forwardQueryToLeader(forwardHost, forwardPort, query, params, + raftWriteTimeoutMs) # Raft log append + majority wait (outside the storage gate). # DDL batches ship the original SQL once (re-executed on apply). Pure DML # ships KV pairs. Mixed DDL+DML in one query uses the DDL path only so the @@ -650,7 +751,10 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.} if shardCheck: let startTicks = getMonoTime().ticks() - let (success, result, errorMsg) = await executeQuery(connCtx.db, connCtx, queryStr, replication=server.replicationManager, raftNode=server.raftNode, raftWriteTimeoutMs=server.config.raftWriteTimeoutMs) + let (success, result, errorMsg) = await executeQuery(connCtx.db, connCtx, queryStr, + replication=server.replicationManager, raftNode=server.raftNode, + raftWriteTimeoutMs=server.config.raftWriteTimeoutMs, + raftPeerClientAddrs=server.config.raftPeerClientAddrs) let durationMs = int((getMonoTime().ticks() - startTicks) div 1_000_000) if durationMs >= slowThreshold: @@ -670,7 +774,10 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.} info("[" & $clientId & "] QueryParams: " & queryStr & " (" & $params.len & " params)") let startTicks = getMonoTime().ticks() - let (success, result, errorMsg) = await executeQuery(connCtx.db, connCtx, queryStr, params, replication=server.replicationManager, raftNode=server.raftNode, raftWriteTimeoutMs=server.config.raftWriteTimeoutMs) + let (success, result, errorMsg) = await executeQuery(connCtx.db, connCtx, queryStr, params, + replication=server.replicationManager, raftNode=server.raftNode, + raftWriteTimeoutMs=server.config.raftWriteTimeoutMs, + raftPeerClientAddrs=server.config.raftPeerClientAddrs) let durationMs = int((getMonoTime().ticks() - startTicks) div 1_000_000) if durationMs >= slowThreshold: diff --git a/tests/bugfix_test.nim b/tests/bugfix_test.nim index 032b603..45eadfd 100644 --- a/tests/bugfix_test.nim +++ b/tests/bugfix_test.nim @@ -362,6 +362,15 @@ suite "Bug fixes — UNIQUE index enforcement": suite "Raft peer address parsing": + test "BARADB_RAFT_CLIENT_PEERS populates raftPeerClientAddrs": + putEnv("BARADB_RAFT_CLIENT_PEERS", "n1@10.0.0.1:9472,n2@10.0.0.2:9472") + defer: delEnv("BARADB_RAFT_CLIENT_PEERS") + var cfg = defaultConfig() + loadConfigFromEnv(cfg) + check cfg.raftPeerClientAddrs.len == 2 + check cfg.raftPeerClientAddrs["n1"] == ("10.0.0.1", 9472) + check cfg.raftPeerClientAddrs["n2"] == ("10.0.0.2", 9472) + test "id@host:port entries populate raftPeerAddrs": putEnv("BARADB_RAFT_PEERS", "n1@127.0.0.1:9473,n2@10.0.0.5:9474,n3") defer: delEnv("BARADB_RAFT_PEERS") diff --git a/tests/raft_writes_e2e_test.nim b/tests/raft_writes_e2e_test.nim index 219f41f..aba97b5 100644 --- a/tests/raft_writes_e2e_test.nim +++ b/tests/raft_writes_e2e_test.nim @@ -142,6 +142,10 @@ proc runWritesScenario() = let peers = "n1@127.0.0.1:" & $(rbase + 1) & ",n2@127.0.0.1:" & $(rbase + 2) & ",n3@127.0.0.1:" & $(rbase + 3) + # SQL client ports for transparent leader write forwarding. + let clientPeers = "n1@127.0.0.1:" & $(cbase + 10) & + ",n2@127.0.0.1:" & $(cbase + 20) & + ",n3@127.0.0.1:" & $(cbase + 30) var nodes: seq[NodeProc] for i in 1 .. 3: @@ -156,6 +160,7 @@ proc runWritesScenario() = env["BARADB_RAFT_PORT"] = $(rbase + i) env["BARADB_RAFT_NODE_ID"] = id env["BARADB_RAFT_PEERS"] = peers + env["BARADB_RAFT_CLIENT_PEERS"] = clientPeers env["BARADB_DATA_DIR"] = dataDir env["BARADB_LOG_LEVEL"] = "info" let p = startProcess(BinaryPath, env = env, @@ -228,23 +233,29 @@ proc runWritesScenario() = db.close() echo "leader schema committed via raft ddl" - # Follower rejection of DDL (not just DML). + # Follower CREATE is forwarded to the leader (BARADB_RAFT_CLIENT_PEERS). block: let db = openClient(nodes[followerIdx].clientPort) - var rejected = false try: - db.exec(sql"CREATE TABLE should_fail (id INT)") + db.exec(sql"CREATE TABLE fwd_from_follower (id INT PRIMARY KEY)") except CatchableError as e: - rejected = "not leader" in e.msg - if not rejected: - echo "follower CREATE failed without 'not leader': ", e.msg - db.close() - if not rejected: - echo "follower CREATE was not rejected with 'not leader'" + echo "follower CREATE (forward) failed: ", e.msg dumpAll(nodes) fail() return - echo "follower CREATE rejected with 'not leader'" + db.close() + # Leader must see the table (forward applied on leader, then raft ddl). + block: + let db = openClient(nodes[leaderIdx].clientPort) + try: + discard db.getAllRows(sql"SELECT * FROM fwd_from_follower") + except CatchableError as e: + echo "leader never saw forwarded CREATE: ", e.msg + dumpAll(nodes) + fail() + return + db.close() + echo "follower CREATE forwarded to leader" # Wait until the follower has applied CREATE TABLE (SELECT no longer # errors with unknown table). Deadline 5s. @@ -317,23 +328,23 @@ proc runWritesScenario() = return echo "follower index-backed SELECT saw the row" - # Follower rejection: DML on a follower must fail with "not leader". + # Follower DML is forwarded to the leader and replicated to the cluster. block: let db = openClient(nodes[followerIdx].clientPort) - var rejected = false try: - db.exec(sql"INSERT INTO rw_test (id, name) VALUES (99, 'nope')") + db.exec(sql"INSERT INTO rw_test (id, name) VALUES (99, 'via-forward')") except CatchableError as e: - rejected = "not leader" in e.msg - if not rejected: - echo "follower INSERT failed but without 'not leader': ", e.msg - db.close() - if not rejected: - echo "follower INSERT was not rejected with a 'not leader' error" + echo "follower INSERT (forward) failed: ", e.msg dumpAll(nodes) fail() return - echo "follower INSERT rejected with 'not leader'" + db.close() + if not waitForRow(nodes[leaderIdx].clientPort, "via-forward", 5): + echo "leader never saw forwarded INSERT row" + dumpAll(nodes) + fail() + return + echo "follower INSERT forwarded to leader" # Failover: kill the leader. A survivor must accept a write once it wins # a new term (majority of the remaining 2-of-3). Log lines can thrash