feat(raft): transparent leader write/DDL forwarding

- 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

@@ -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

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

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

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,8 +371,16 @@ 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"
+          # 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 & "'")
 
+      if not needsForward:
         let res = executor.executeQuery(ctx, astNode, params)
         if res.success:
           # Ship written key-value pairs to replicas (legacy path; skipped when
@@ -335,6 +432,10 @@ proc executeQuery(db: LSMTree, ctx: ExecutionContext, query: string, params: seq
           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:

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")

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
+        echo "follower CREATE (forward) failed: ", 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'"
         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
+        echo "follower INSERT (forward) failed: ", 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"
         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