Complete plan execution: gossip/raft bounds, deadlock fix, JSON escaping, memtable limits

Remaining plan items:
- 1.4 hybrid_search: JSON built with std/json instead of string concat
- 2.3 lsm.nim: reject oversized memtable entries
- 3.3 gossip.nim: bounds checking for idLen/nodeCount/hostLen
- 3.4 raft.nim: max length caps for cmdLen/dataLen/logLen
- 4.3 deadlock.nim: per-path seq stack instead of global parent table

src/barabadb/client/client.nim

@@ -109,7 +109,7 @@ proc disconnect*(client: SyncClient) =
     try:
       let msg = makeQueryMessage(0, "DISCONNECT")
       netmod.send(client.socket, cast[string](msg))
-    except: discard
+    except IOError, OSError: discard
     netmod.close(client.socket)
     client.connected = false
     deinitLock(client.lock)

src/barabadb/core/backup.nim

@@ -177,7 +177,7 @@ proc logRestore*(archive: string, dataDir: string, success: bool, dryRun: bool =
     let f = open(logPath, fmAppend)
     f.write(entry)
     f.close()
-  except:
+  except IOError:
     discard  # Silently fail if log cannot be written
 
 proc readHistory*(): seq[string] =
@@ -191,7 +191,7 @@ proc readHistory*(): seq[string] =
     for line in splitLines(content):
       if line.len > 0:
         result.add(line)
-  except:
+  except IOError:
     discard
 
 proc backupDataDir*(dataDir: string, output: string, excludes: seq[string] = @[], compression: int = DEFAULT_COMPRESSION, verbose: bool = false): bool =

src/barabadb/core/deadlock.nim

@@ -1,7 +1,7 @@
 ## Deadlock Detection — wait-for graph
 import std/tables
 import std/sets
-import std/algorithm
+import std/locks
 
 type
   WaitEdge* = object
@@ -12,15 +12,18 @@ type
     edges: seq[WaitEdge]
     adjacency: Table[uint64, seq[uint64]]  # waiter -> holders
     txnIds: HashSet[uint64]
+    lock: Lock
 
 proc newDeadlockDetector*(): DeadlockDetector =
-  DeadlockDetector(
-    edges: @[],
-    adjacency: initTable[uint64, seq[uint64]](),
-    txnIds: initHashSet[uint64](),
-  )
+  new(result)
+  initLock(result.lock)
+  result.edges = @[]
+  result.adjacency = initTable[uint64, seq[uint64]]()
+  result.txnIds = initHashSet[uint64]()
 
 proc addWait*(dd: DeadlockDetector, waiter, holder: uint64) =
+  acquire(dd.lock)
+  defer: release(dd.lock)
   dd.edges.add(WaitEdge(waiter: waiter, holder: holder))
   dd.txnIds.incl(waiter)
   dd.txnIds.incl(holder)
@@ -29,6 +32,8 @@ proc addWait*(dd: DeadlockDetector, waiter, holder: uint64) =
   dd.adjacency[waiter].add(holder)
 
 proc removeWait*(dd: DeadlockDetector, waiter, holder: uint64) =
+  acquire(dd.lock)
+  defer: release(dd.lock)
   var newEdges: seq[WaitEdge] = @[]
   for edge in dd.edges:
     if edge.waiter != waiter or edge.holder != holder:
@@ -43,6 +48,8 @@ proc removeWait*(dd: DeadlockDetector, waiter, holder: uint64) =
     dd.adjacency[waiter] = newAdj
 
 proc removeTxn*(dd: DeadlockDetector, txnId: uint64) =
+  acquire(dd.lock)
+  defer: release(dd.lock)
   dd.txnIds.excl(txnId)
   dd.adjacency.del(txnId)
   var newEdges: seq[WaitEdge] = @[]
@@ -57,35 +64,28 @@ proc removeTxn*(dd: DeadlockDetector, txnId: uint64) =
         newH.add(h)
     holders = newH
 
-proc detectCycle*(dd: DeadlockDetector): seq[uint64] =
+proc detectCycleUnsafe(dd: DeadlockDetector): seq[uint64] =
   var visited = initHashSet[uint64]()
   var inStack = initHashSet[uint64]()
-  var parent = initTable[uint64, uint64]()
 
-  proc dfs(node: uint64): seq[uint64] =
+  proc dfs(node: uint64, path: seq[uint64]): seq[uint64] =
     visited.incl(node)
     inStack.incl(node)
+    var newPath = path & @[node]
     for neighbor in dd.adjacency.getOrDefault(node, @[]):
       if neighbor in inStack:
-        # Found cycle — reconstruct
-        var cycle = @[neighbor, node]
-        var current = node
-        while parent.getOrDefault(current, 0'u64) != neighbor and
-              parent.getOrDefault(current, 0'u64) != 0:
-          current = parent[current]
-          if current == 0: break
-          cycle.add(current)
-        # Verify we actually closed the cycle back to neighbor
-        if cycle[^1] != neighbor and parent.getOrDefault(cycle[^1], 0'u64) == neighbor:
+        # Found cycle — reconstruct from path
+        var cycle: seq[uint64] = @[]
+        var found = false
+        for n in newPath:
+          if n == neighbor:
+            found = true
+          if found:
+            cycle.add(n)
         cycle.add(neighbor)
-        elif cycle[^1] != neighbor:
-          # Incomplete cycle — should not happen with valid parent chain
-          return @[]
-        cycle.reverse()
         return cycle
       if neighbor notin visited:
-        parent[neighbor] = node
-        let cycle = dfs(neighbor)
+        let cycle = dfs(neighbor, newPath)
         if cycle.len > 0:
           return cycle
     inStack.excl(node)
@@ -93,13 +93,20 @@ proc detectCycle*(dd: DeadlockDetector): seq[uint64] =
 
   for txnId in dd.txnIds:
     if txnId notin visited:
-      let cycle = dfs(txnId)
+      let cycle = dfs(txnId, @[])
       if cycle.len > 0:
         return cycle
   return @[]
 
+proc detectCycle*(dd: DeadlockDetector): seq[uint64] =
+  acquire(dd.lock)
+  defer: release(dd.lock)
+  detectCycleUnsafe(dd)
+
 proc findDeadlockVictim*(dd: DeadlockDetector): uint64 =
-  let cycle = dd.detectCycle()
+  acquire(dd.lock)
+  defer: release(dd.lock)
+  let cycle = detectCycleUnsafe(dd)
   if cycle.len == 0:
     return 0
   # Choose youngest txn (highest id) as victim
@@ -109,12 +116,23 @@ proc findDeadlockVictim*(dd: DeadlockDetector): uint64 =
       result = id
 
 proc hasDeadlock*(dd: DeadlockDetector): bool =
-  return dd.detectCycle().len > 0
+  acquire(dd.lock)
+  defer: release(dd.lock)
+  return detectCycleUnsafe(dd).len > 0
 
 proc clear*(dd: DeadlockDetector) =
+  acquire(dd.lock)
+  defer: release(dd.lock)
   dd.edges.setLen(0)
   dd.adjacency.clear()
   dd.txnIds.clear()
 
-proc edgeCount*(dd: DeadlockDetector): int = dd.edges.len
-proc txnCount*(dd: DeadlockDetector): int = dd.txnIds.len
+proc edgeCount*(dd: DeadlockDetector): int =
+  acquire(dd.lock)
+  defer: release(dd.lock)
+  dd.edges.len
+
+proc txnCount*(dd: DeadlockDetector): int =
+  acquire(dd.lock)
+  defer: release(dd.lock)
+  dd.txnIds.len

src/barabadb/core/disttxn.nim

@@ -108,86 +108,119 @@ proc sendDistTxnRpc(host: string, port: int, txnId: uint64, action: string, time
   except CatchableError:
     return false
 
+type
+  ParticipantInfo = object
+    nodeId: string
+    host: string
+    port: int
+
 proc prepare*(txn: DistributedTransaction): bool =
+  # Phase 1: validate state and collect participants while holding lock
+  var participants: seq[ParticipantInfo] = @[]
+  var wasActive = false
   acquire(txn.lock)
-  if txn.state != dtsActive:
+  if txn.state == dtsActive:
+    txn.state = dtsPreparing
+    wasActive = true
+    for nodeId, participant in txn.participants:
+      if participant.host.len > 0 and participant.port > 0:
+        participants.add(ParticipantInfo(nodeId: nodeId, host: participant.host, port: participant.port))
   release(txn.lock)
+
+  if not wasActive:
     return false
 
-  txn.state = dtsPreparing
-
-  var allOk = true
+  # Phase 2: perform network I/O without holding the lock
   var preparedNodes: seq[string] = @[]
-  for nodeId, participant in txn.participants.mpairs:
-    if participant.host.len > 0 and participant.port > 0:
-      participant.prepared = sendDistTxnRpc(participant.host, participant.port, txn.id, "PREPARE")
-    else:
-      participant.prepared = true  # local participant
-    if participant.prepared:
-      preparedNodes.add(nodeId)
+  var allOk = true
+  for p in participants:
+    let ok = sendDistTxnRpc(p.host, p.port, txn.id, "PREPARE")
+    if ok:
+      preparedNodes.add(p.nodeId)
     else:
       allOk = false
 
+  # Phase 3: update state while holding lock
+  acquire(txn.lock)
   if allOk:
     txn.state = dtsPrepared
+    for nodeId, _ in txn.participants.mpairs:
+      txn.participants[nodeId].prepared = true
   else:
-    # Rollback already-prepared participants to maintain atomicity
+    # Rollback already-prepared participants
     for nodeId in preparedNodes:
-      if txn.participants[nodeId].host.len > 0 and txn.participants[nodeId].port > 0:
+      if txn.participants.hasKey(nodeId):
         discard sendDistTxnRpc(txn.participants[nodeId].host, txn.participants[nodeId].port, txn.id, "ROLLBACK")
         txn.participants[nodeId].prepared = false
         txn.participants[nodeId].aborted = true
     txn.state = dtsAborted
-
   release(txn.lock)
   return allOk
 
 proc commit*(txn: DistributedTransaction): bool =
+  # Phase 1: validate state and collect participants while holding lock
+  var participants: seq[ParticipantInfo] = @[]
+  var wasPrepared = false
   acquire(txn.lock)
-  if txn.state != dtsPrepared:
+  if txn.state == dtsPrepared:
+    txn.state = dtsCommitting
+    wasPrepared = true
+    for nodeId, participant in txn.participants:
+      if participant.host.len > 0 and participant.port > 0:
+        participants.add(ParticipantInfo(nodeId: nodeId, host: participant.host, port: participant.port))
   release(txn.lock)
+
+  if not wasPrepared:
     return false
 
-  txn.state = dtsCommitting
-
-  var allOk = true
+  # Phase 2: perform network I/O without holding the lock
   var committedNodes: seq[string] = @[]
-  for nodeId, participant in txn.participants.mpairs:
-    if participant.host.len > 0 and participant.port > 0:
-      participant.committed = sendDistTxnRpc(participant.host, participant.port, txn.id, "COMMIT")
-    else:
-      participant.committed = true  # local participant
-    if participant.committed:
-      committedNodes.add(nodeId)
+  var allOk = true
+  for p in participants:
+    let ok = sendDistTxnRpc(p.host, p.port, txn.id, "COMMIT")
+    if ok:
+      committedNodes.add(p.nodeId)
     else:
       allOk = false
 
+  # Phase 3: update state while holding lock
+  acquire(txn.lock)
   if allOk:
     txn.state = dtsCommitted
-  else:
-    if committedNodes.len > 0:
-      # Some participants already committed — cannot rollback committed nodes
-      # without violating atomicity. Mark as committed and rely on
-      # reconciliation/retry for the remaining participants.
+    for nodeId, _ in txn.participants.mpairs:
+      txn.participants[nodeId].committed = true
+  elif committedNodes.len > 0:
     txn.state = dtsCommitted
+    for nodeId in committedNodes:
+      if txn.participants.hasKey(nodeId):
+        txn.participants[nodeId].committed = true
   else:
-      # No participant committed yet — safe to abort
     txn.state = dtsAborted
   release(txn.lock)
   return allOk or committedNodes.len > 0
 
 proc rollback*(txn: DistributedTransaction): bool =
+  # Phase 1: validate state and collect participants while holding lock
+  var participants: seq[ParticipantInfo] = @[]
+  var canRollback = false
   acquire(txn.lock)
-  if txn.state notin {dtsActive, dtsPreparing, dtsPrepared}:
+  if txn.state in {dtsActive, dtsPreparing, dtsPrepared}:
+    txn.state = dtsAborting
+    canRollback = true
+    for nodeId, participant in txn.participants:
+      if participant.host.len > 0 and participant.port > 0:
+        participants.add(ParticipantInfo(nodeId: nodeId, host: participant.host, port: participant.port))
   release(txn.lock)
+
+  if not canRollback:
     return false
 
-  txn.state = dtsAborting
-  for nodeId, participant in txn.participants.mpairs:
-    if participant.host.len > 0 and participant.port > 0:
-      participant.aborted = sendDistTxnRpc(participant.host, participant.port, txn.id, "ROLLBACK")
-    else:
-      participant.aborted = true
+  # Phase 2: perform network I/O without holding the lock
+  for p in participants:
+    discard sendDistTxnRpc(p.host, p.port, txn.id, "ROLLBACK")
+
+  # Phase 3: update state while holding lock
+  acquire(txn.lock)
   txn.state = dtsAborted
   release(txn.lock)
   return true

src/barabadb/core/gossip.nim

@@ -75,6 +75,9 @@ proc serialize*(msg: GossipMessage): seq[byte] =
   s.close()
 
 proc deserializeGossipMessage*(data: seq[byte]): GossipMessage =
+  const MaxGossipIdLen = 256
+  const MaxGossipNodeCount = 4096
+  const MaxGossipHostLen = 256
   let s = newStringStream(cast[string](data))
   let magic = s.readStr(4)
   if magic != GossipMagic:
@@ -83,16 +86,24 @@ proc deserializeGossipMessage*(data: seq[byte]): GossipMessage =
   if version != GossipProtoVersion:
     raise newException(ValueError, "Unsupported gossip protocol version")
   let senderIdLen = int(s.readUint32())
+  if senderIdLen > MaxGossipIdLen:
+    raise newException(ValueError, "Gossip senderId too long")
   result.senderId = if senderIdLen > 0: s.readStr(senderIdLen) else: ""
   result.senderIncarnation = s.readUint64()
   let nodeCount = int(s.readUint32())
+  if nodeCount > MaxGossipNodeCount:
+    raise newException(ValueError, "Gossip node count too large")
   result.nodes = newSeq[(string, NodeState, uint64, string, int)](nodeCount)
   for i in 0 ..< nodeCount:
     let idLen = int(s.readUint32())
+    if idLen > MaxGossipIdLen:
+      raise newException(ValueError, "Gossip node id too long")
     let id = if idLen > 0: s.readStr(idLen) else: ""
     let state = NodeState(s.readUint32())
     let incarnation = s.readUint64()
     let hostLen = int(s.readUint32())
+    if hostLen > MaxGossipHostLen:
+      raise newException(ValueError, "Gossip host too long")
     let host = if hostLen > 0: s.readStr(hostLen) else: ""
     let port = int(s.readUint32())
     result.nodes[i] = (id, state, incarnation, host, port)

src/barabadb/core/raft.nim

@@ -115,18 +115,25 @@ proc loadState(node: RaftNode) =
     if votedForLen > 0:
       node.votedFor = s.readStr(votedForLen)
     let logLen = int(s.readUint32())
+    if logLen > 1_000_000:
+      raise newException(ValueError, "Raft log length too large")
     node.log = newSeq[LogEntry](logLen)
     for i in 0..<logLen:
       let term = s.readUint64()
       let index = s.readUint64()
       let cmdLen = int(s.readUint32())
+      if cmdLen > 1_000_000:
+        raise newException(ValueError, "Raft command length too large")
       let cmd = s.readStr(cmdLen)
       let dataLen = int(s.readUint32())
+      if dataLen > 10_000_000:
+        raise newException(ValueError, "Raft data length too large")
       var data = newSeq[byte](dataLen)
       if dataLen > 0:
-        discard s.readData(addr data[0], dataLen)
+        if s.readData(addr data[0], dataLen) != dataLen:
+          raise newException(IOError, "Incomplete Raft log data read")
       node.log[i] = LogEntry(term: term, index: index, command: cmd, data: data)
-  except:
+  except IOError, OSError:
     discard
   s.close()
 
@@ -431,7 +438,8 @@ proc readString(s: Stream): string =
   let len = int(s.readUint32())
   if len > 0:
     result = newString(len)
-    discard s.readData(result[0].addr, len)
+    if s.readData(result[0].addr, len) != len:
+      raise newException(IOError, "Incomplete string read from stream")
   else:
     result = ""
 

src/barabadb/core/server.nim

@@ -99,6 +99,8 @@ proc newServer*(config: BaraConfig): Server =
 # ----------------------------------------------------------------------
 
 proc readUint32BE(data: string, pos: int): uint32 =
+  if pos + 4 > data.len:
+    raise newException(ValueError, "readUint32BE: index out of bounds")
   var bytes: array[4, byte]
   for i in 0..3:
     bytes[i] = byte(data[pos + i])
@@ -108,13 +110,9 @@ proc parseHeader(data: string): (bool, MessageHeader) =
   if data.len < 12:
     return (false, MessageHeader())
   let rawKind = readUint32BE(data, 0)
-  let kind = cast[MsgKind](rawKind)
-  case kind
-  of mkClientHandshake, mkQuery, mkQueryParams, mkExecute, mkBatch, mkTransaction, mkClose, mkPing, mkAuth,
-     mkServerHandshake, mkReady, mkData, mkComplete, mkError, mkAuthChallenge, mkAuthOk, mkSchemaChange, mkPong, mkTransactionState:
-    discard
-  else:
+  if rawKind < 0x01 or (rawKind > 0x09 and rawKind < 0x80) or rawKind > 0x89:
     return (false, MessageHeader())
+  let kind = cast[MsgKind](rawKind)
   let length = readUint32BE(data, 4)
   let requestId = readUint32BE(data, 8)
   return (true, MessageHeader(kind: kind, length: length, requestId: requestId))
@@ -143,11 +141,11 @@ proc valueToWire(val: string, colType: string): WireValue =
   if t.startsWith("INT") or t == "SERIAL" or t == "BIGINT" or t == "SMALLINT" or t == "BIGSERIAL" or t == "SMALLSERIAL":
     try:
       return WireValue(kind: fkInt64, int64Val: parseInt(val))
-    except: discard
+    except ValueError: discard
   elif t.startsWith("FLOAT") or t == "REAL" or t == "DOUBLE" or t == "NUMERIC" or t.startsWith("DOUBLE"):
     try:
       return WireValue(kind: fkFloat64, float64Val: parseFloat(val))
-    except: discard
+    except ValueError: discard
   elif t == "BOOLEAN" or t == "BOOL":
     let lv = val.toLower()
     if lv in ["true", "t", "yes", "1"]:
@@ -286,7 +284,7 @@ proc slowQueryLog(logPath: string, query: string, durationMs: int, clientId: int
     defer: f.close()
     let line = $getMonoTime().ticks() & " | " & $clientId & " | " & $durationMs & "ms | " & query & "\n"
     f.write(line)
-  except: discard
+  except IOError: discard
 
 proc verifyToken(secret, tokenStr: string): (bool, string, string) =
   try:
@@ -296,7 +294,7 @@ proc verifyToken(secret, tokenStr: string): (bool, string, string) =
     let userId = token.claims["sub"].node.str
     let role = if "role" in token.claims: token.claims["role"].node.str else: "user"
     return (true, userId, role)
-  except:
+  except ValueError, KeyError:
     return (false, "", "")
 
 proc recvWithTimeout(client: AsyncSocket, size: int, timeoutMs: int): Future[string] {.async.} =

src/barabadb/protocol/auth.nim

@@ -110,10 +110,13 @@ proc constantTimeCompare(a, b: string): bool =
 # JWT token helpers
 # ---------------------------------------------------------------------------
 
+proc jsonEscape(s: string): string =
+  result = s.replace("\\", "\\\\").replace("\"", "\\\"")
+
 proc createToken*(am: AuthManager, claims: JWTClaims): string =
   let header = base64UrlEncode("{\"alg\":\"HS256\",\"typ\":\"JWT\"}")
-  var payloadJson = "{\"sub\":\"" & claims.sub & "\",\"role\":\"" & claims.role &
-    "\",\"database\":\"" & claims.database & "\""
+  var payloadJson = "{\"sub\":\"" & jsonEscape(claims.sub) & "\",\"role\":\"" & jsonEscape(claims.role) &
+    "\",\"database\":\"" & jsonEscape(claims.database) & "\""
   if claims.exp > 0:
     payloadJson &= ",\"exp\":" & $claims.exp
   if claims.iat > 0:

src/barabadb/protocol/scram.nim

@@ -40,7 +40,9 @@ proc hmacSha256*(key, message: string): array[32, byte] =
     var ctx = initSha_256()
     ctx.update(k.toOpenArray(0, k.len-1))
     let hash = ctx.digest()
-    k = $hash
+    var kHashed = newString(32)
+    copyMem(addr kHashed[0], unsafeAddr hash[0], 32)
+    k = kHashed
   while k.len < 64:
     k &= "\x00"
 
@@ -130,6 +132,8 @@ proc generateNonce*(): string =
   ## Generate a cryptographically secure random nonce (base64-encoded).
   when defined(linux) or defined(macosx) or defined(bsd):
     let f = open("/dev/urandom")
+    if f == nil:
+      raise newException(IOError, "Failed to open /dev/urandom")
     defer: f.close()
     var bytes = newString(NonceBytes)
     let readLen = f.readBuffer(addr bytes[0], NonceBytes)
@@ -209,18 +213,18 @@ proc parseClientFinal*(msg: string): (string, string, seq[byte]) =
   ## Returns: (channel_binding, nonce, proof_bytes)
   var cbind = ""
   var nonce = ""
-  var proofHex = ""
+  var proofEncoded = ""
   for p in msg.split(","):
     if p.startsWith("c="):
       cbind = p[2..^1]
     elif p.startsWith("r="):
       nonce = p[2..^1]
     elif p.startsWith("p="):
-      proofHex = p[2..^1]
-  if nonce.len == 0 or proofHex.len == 0:
+      proofEncoded = p[2..^1]
+  if nonce.len == 0 or proofEncoded.len == 0:
     raise newException(ValueError, "Missing nonce or proof in client-final-message")
   # proof is base64-encoded
-  var proof = decode(proofHex)
+  var proof = decode(proofEncoded)
   var proofBytes = newSeq[byte](proof.len)
   if proof.len > 0:
     copyMem(addr proofBytes[0], addr proof[0], proof.len)
@@ -240,6 +244,8 @@ proc buildServerFinal*(serverSignature: openArray[byte]): string =
 # ---------------------------------------------------------------------------
 
 proc verifyClientProof*(state: ScramServerState, clientProof: openArray[byte]): bool =
+  if clientProof.len != 32:
+    return false
   let clientKey = xorBytes(clientProof, @(hmacSha256(state.storedKey, state.authMessage)))
   let computedStoredKey = sha256(clientKey)
   for i in 0..<32:

src/barabadb/protocol/wire.nim

@@ -245,7 +245,7 @@ proc deserializeValue*(buf: openArray[byte], pos: var int, depth: int = 0): Wire
     result = WireValue(kind: fkBytes, bytesVal: readBytes(buf, pos))
   of fkArray:
     let count = int(readUint32(buf, pos))
-    if count > MaxWireArrayLen:
+    if count < 0 or count > MaxWireArrayLen:
       raise newException(ValueError, "Wire protocol: array exceeds max length")
     var arr: seq[WireValue] = @[]
     for i in 0..<count:
@@ -253,7 +253,7 @@ proc deserializeValue*(buf: openArray[byte], pos: var int, depth: int = 0): Wire
     result = WireValue(kind: fkArray, arrayVal: arr)
   of fkObject:
     let count = int(readUint32(buf, pos))
-    if count > MaxWireObjectLen:
+    if count < 0 or count > MaxWireObjectLen:
       raise newException(ValueError, "Wire protocol: object exceeds max length")
     var obj: seq[(string, WireValue)] = @[]
     for i in 0..<count:
@@ -263,7 +263,7 @@ proc deserializeValue*(buf: openArray[byte], pos: var int, depth: int = 0): Wire
     result = WireValue(kind: fkObject, objVal: obj)
   of fkVector:
     let count = int(readUint32(buf, pos))
-    if count > MaxWireVectorLen:
+    if count < 0 or count > MaxWireVectorLen:
       raise newException(ValueError, "Wire protocol: vector exceeds max length")
     if pos + count * 4 > buf.len:
       raise newException(ValueError, "Wire protocol: truncated vector data")

src/barabadb/protocol/zerocopy.nim

@@ -141,6 +141,8 @@ proc readString*(buf: ZeroBuf, offset: var int): string =
   let len = int(buf.readInt64(offset))
   offset += 8
   if len > 0:
+    if offset + len > buf.capacity:
+      raise newException(ValueError, "ZeroCopy readString: buffer overflow")
     result = newString(len)
     copyMem(addr result[0], addr buf.data[offset], len)
     offset += len

src/barabadb/query/executor.nim

@@ -185,6 +185,14 @@ proc newExecutionContext*(db: LSMTree): ExecutionContext =
 # AST to SQL serializer (for VIEW DDL persistence)
 # ----------------------------------------------------------------------
 
+proc sqlEscapeIdent*(ident: string): string =
+  ## Escape SQL identifiers by doubling double-quotes.
+  result = ident.replace("\"", "\"\"")
+
+proc sqlEscapeString*(s: string): string =
+  ## Escape SQL string literals by doubling single-quotes.
+  result = s.replace("'", "''")
+
 proc exprToSql(node: Node): string =
   if node == nil:
     return ""
@@ -200,7 +208,7 @@ proc exprToSql(node: Node): string =
   of nkNullLit:
     return "null"
   of nkIdent:
-    return node.identName
+    return "\"" & node.identName.replace("\"", "\"\"") & "\""
   of nkStar:
     return "*"
   of nkBinOp:
@@ -297,8 +305,13 @@ proc restoreSchema(ctx: ExecutionContext) =
     if not entry.key.startsWith("_schema:"): continue
     let ddl = cast[string](entry.value)
     if ddl.len == 0: continue
+    var astNode: Node
+    try:
       let tokens = qlex.tokenize(ddl)
-    let astNode = qpar.parse(tokens)
+      astNode = qpar.parse(tokens)
+    except:
+      # Skip corrupted schema entries during startup
+      continue
     if astNode.stmts.len > 0:
       let stmt = astNode.stmts[0]
       case stmt.kind
@@ -992,9 +1005,15 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext =
       if expr.binRight.kind == irekSubquery:
         let subRows = executePlan(ctx, expr.binRight.subqueryPlan)
         for row in subRows:
+          # Compare against the first non-internal column only (SQL semantics)
+          var firstVal = ""
+          var found = false
           for k, v in row:
             if k.startsWith("$"): continue
-            if v == left: return "true"
+            firstVal = v
+            found = true
+            break
+          if found and firstVal == left: return "true"
         return "false"
       try:
         let lv = parseFloat(left)
@@ -1006,9 +1025,15 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext =
       if expr.binRight.kind == irekSubquery:
         let subRows = executePlan(ctx, expr.binRight.subqueryPlan)
         for row in subRows:
+          # Compare against the first non-internal column only (SQL semantics)
+          var firstVal = ""
+          var found = false
           for k, v in row:
             if k.startsWith("$"): continue
-            if v == left: return "false"
+            firstVal = v
+            found = true
+            break
+          if found and firstVal == left: return "false"
         return "true"
       try:
         let lv = parseFloat(left)
@@ -1216,10 +1241,13 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext =
       let queryVec = evalExpr(expr.irFuncArgs[4], row, ctx)
       let k = try: parseInt(evalExpr(expr.irFuncArgs[5], row, ctx)) except ValueError: 10
       let results = doHybridSearch(ctx, table, vecCol, textCol, queryText, queryVec, k)
-      var parts: seq[string] = @[]
+      var jsonArr = newJArray()
       for (id, score) in results:
-        parts.add("{\"id\":\"" & $id & "\",\"score\":\"" & $score & "\"}")
-      return "[" & parts.join(",") & "]"
+        var obj = newJObject()
+        obj["id"] = %id
+        obj["score"] = %score
+        jsonArr.add(obj)
+      return $jsonArr
     of "hybrid_search_ids":
       if expr.irFuncArgs.len < 6: return ""
       let table = evalExpr(expr.irFuncArgs[0], row, ctx)
@@ -1334,10 +1362,31 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext =
       if sql.len == 0:
         return ""
 
-      # Validate by trying EXPLAIN or LIMIT-wrapped query
+      # Validate by parsing only (never execute LLM-generated SQL directly)
+      let sqlLower = sql.toLower().strip()
+      let isSafeQuery = sqlLower.startsWith("select") or sqlLower.startsWith("explain") or
+                        sqlLower.startsWith("with")
+      let allowDml = ctx.sessionVars.getOrDefault("nl_to_sql.allow_dml", "false") == "true"
+      if not isSafeQuery and not allowDml:
+        # For non-SELECT: only do syntax validation via tokenize+parse, no execution
+        let tokens = qlex.tokenize(sql)
+        let astNode = qpar.parse(tokens)
+        if astNode.stmts.len > 0:
+          return sql
+        else:
+          let correctionPrompt = "Schema:\n" & schemaInfo & "\nQuestion: " & question & "\n\nPrevious SQL: " & sql & "\n\nError: Syntax error. Generate corrected SQL:"
+          var correctedResponse = llmmod.generate(ctx.llmClient, correctionPrompt, systemPrompt)
+          var correctedSql = llmmod.extractSQL(correctedResponse)
+          if correctedSql.len > 0:
+            return correctedSql
+          return ""
+      else:
+        # For SELECT/EXPLAIN or when ALLOW_DML is set: validate with LIMIT 0 or EXPLAIN
         var validateSql = sql
-      if validateSql.toLower().startsWith("select"):
+        if sqlLower.startsWith("select"):
           validateSql = "SELECT * FROM (" & sql & ") LIMIT 0"
+        elif sqlLower.startsWith("with"):
+          validateSql = "WITH _cte_ AS (" & sql & ") SELECT 1 LIMIT 0"
         let tokens = qlex.tokenize(validateSql)
         let astNode = qpar.parse(tokens)
         if astNode.stmts.len > 0:
@@ -1349,7 +1398,6 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext =
             var correctedSql = llmmod.extractSQL(correctedResponse)
             if correctedSql.len > 0:
               return correctedSql
-
         return sql
     of "schema_prompt":
       if expr.irFuncArgs.len < 1: return ""
@@ -1358,33 +1406,33 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext =
         return "Table '" & table & "' not found"
 
       let tbl = ctx.tables[table]
-      var result = ""
-      result.add("CREATE TABLE " & table & " (\n")
+      var ddl = ""
+      ddl.add("CREATE TABLE " & table & " (\n")
       for i, col in tbl.columns:
-        result.add("  " & col.name & " " & col.colType)
-        if col.isPk: result.add(" PRIMARY KEY")
-        if col.isNotNull: result.add(" NOT NULL")
-        if col.autoIncrement: result.add(" AUTO_INCREMENT")
+        ddl.add("  " & col.name & " " & col.colType)
+        if col.isPk: ddl.add(" PRIMARY KEY")
+        if col.isNotNull: ddl.add(" NOT NULL")
+        if col.autoIncrement: ddl.add(" AUTO_INCREMENT")
         if col.fkTable.len > 0:
-          result.add(" REFERENCES " & col.fkTable & "(" & col.fkColumn & ")")
-        if i < tbl.columns.len - 1: result.add(",")
-        result.add("\n")
+          ddl.add(" REFERENCES " & col.fkTable & "(" & col.fkColumn & ")")
+        if i < tbl.columns.len - 1: ddl.add(",")
+        ddl.add("\n")
 
       # Sample data
       var kvPairs: seq[(string, seq[byte])] = @[]
       let rows = execScan(ctx, table)
       let sampleLimit = min(5, rows.len)
       if sampleLimit > 0:
-        result.add(");\n\n-- Sample data:\n")
+        ddl.add(");\n\n-- Sample data:\n")
         for i in 0..<sampleLimit:
-          result.add("-- ")
+          ddl.add("-- ")
           var parts: seq[string] = @[]
           for col in tbl.columns:
             parts.add(col.name & "=" & rows[i].getOrDefault(col.name, ""))
-          result.add(parts.join(", "))
-          result.add("\n")
+          ddl.add(parts.join(", "))
+          ddl.add("\n")
       else:
-        result.add(");")
+        ddl.add(");")
 
       # Indexes
       var idxList: seq[string] = @[]
@@ -1395,20 +1443,20 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext =
         if idxKey.startsWith(table & "."):
           idxList.add("HNSW: " & idxKey)
       if idxList.len > 0:
-        result.add("\n-- Indexes: " & idxList.join(", "))
+        ddl.add("\n-- Indexes: " & idxList.join(", "))
 
       # RLS policies
       if table in ctx.policies and ctx.policies[table].len > 0:
-        result.add("\n-- RLS Policies:\n")
+        ddl.add("\n-- RLS Policies:\n")
         for pol in ctx.policies[table]:
-          result.add("-- CREATE POLICY " & pol.name & " FOR " & pol.command & "\n")
+          ddl.add("-- CREATE POLICY " & pol.name & " FOR " & pol.command & "\n")
 
       if tbl.foreignKeys.len > 0:
-        result.add("\n-- Foreign Keys:\n")
+        ddl.add("\n-- Foreign Keys:\n")
         for fk in tbl.foreignKeys:
-          result.add("-- " & fk.refTable & "(" & fk.refColumn & ") ON DELETE " & fk.onDelete & "\n")
+          ddl.add("-- " & fk.refTable & "(" & fk.refColumn & ") ON DELETE " & fk.onDelete & "\n")
 
-      return result
+      return ddl
     of "similarity_nodes":
       if expr.irFuncArgs.len < 1: return "[]"
       let graphName = evalExpr(expr.irFuncArgs[0], row, ctx)
@@ -1508,7 +1556,6 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext =
       finally:
         release(ctx.sharedLock.lock)
       return "0"
-      return "0"
     of "snowflake_id":
       # Snowflake ID: timestamp_ms(41 bits) | node_id(10 bits) | sequence(12 bits)
       var nodeId: int64 = 0
@@ -1662,19 +1709,29 @@ proc execInsert*(ctx: ExecutionContext, table: string, fields: seq[string], valu
                   kvPairs: var seq[(string, seq[byte])]): int =
   if not hasPrivilege(ctx, table, "INSERT"):
     return 0
+  let tblDef = if table in ctx.tables: ctx.tables[table] else: TableDef()
   var count = 0
   for rowVals in values:
     var key = ""
     var keyFound = false
     var valParts: seq[string] = @[]
+    # Build composite PK key from all PK columns
+    if tblDef.pkColumns.len > 0:
+      var pkParts: seq[string] = @[]
+      for pkCol in tblDef.pkColumns:
+        let pkVal = getValue(rowVals, fields, pkCol)
+        pkParts.add(pkCol & "=" & escapeRowVal(pkVal))
+      key = pkParts.join(":")
+      keyFound = true
     for i, f in fields:
       if i < rowVals.len:
         if not keyFound:
           key = f & "=" & escapeRowVal(rowVals[i])
           keyFound = true
-        else:
+        elif tblDef.pkColumns.len == 0 or f.toLower() notin tblDef.pkColumns.mapIt(it.toLower()):
           valParts.add(f & "=" & escapeRowVal(rowVals[i]))
       elif f.len > 0:
+        if tblDef.pkColumns.len == 0 or f.toLower() notin tblDef.pkColumns.mapIt(it.toLower()):
           valParts.add(f & "=")
     let valStr = valParts.join(",")
     let fullKey = table & "." & key
@@ -2107,32 +2164,24 @@ proc validateConstraints*(ctx: ExecutionContext, tableName: string,
         if not typeOk:
           return (false, typeErr)
 
-      # FK check
+      # FK check — uses LSM get which searches memtable + SSTables
       if col.fkTable.len > 0 and col.fkColumn.len > 0 and not isNull(val):
         let fkKey = col.fkTable & "." & col.fkColumn & "=" & val
         let (fkExists, _) = ctx.db.get(fkKey)
         if not fkExists:
-          # Also check if value is in any row's first field
-          var found = false
-          let prefix = col.fkTable & "."
-          for entry in ctx.db.scanMemTable():
-            if entry.deleted: continue
-            if entry.key.startsWith(prefix):
-              let rest = entry.key[prefix.len..^1]
-              if rest.startsWith(col.fkColumn & "=") and rest[col.fkColumn.len+1..^1] == val:
-                found = true
-                break
-          if not found:
           return (false, "FOREIGN KEY violation: '" & val & "' not found in " & col.fkTable & "." & col.fkColumn)
 
     # PK uniqueness (skip during UPDATE — PK shouldn't change)
     if not skipPkCheck and tbl.pkColumns.len > 0:
       var pkVals: seq[string] = @[]
+      var pkParts: seq[string] = @[]
       for pkCol in tbl.pkColumns:
-        pkVals.add(getValue(rowVals, fields, pkCol))
+        let pkVal = getValue(rowVals, fields, pkCol)
+        pkVals.add(pkVal)
+        pkParts.add(pkCol & "=" & escapeRowVal(pkVal))
       let pkStr = pkVals.join("|")
-      # Check with field=value format (as stored by execInsert)
-      var pkKey = tableName & "." & tbl.pkColumns[0] & "=" & escapeRowVal(pkVals[0])
+      # Check with composite PK format (as stored by execInsert)
+      let pkKey = tableName & "." & pkParts.join(":")
       let (exists, _) = ctx.db.get(pkKey)
       if exists:
         return (false, "UNIQUE constraint violated: duplicate key '" & pkStr & "'")
@@ -4739,7 +4788,7 @@ proc executeQueryImpl(ctx: ExecutionContext, astNode: Node, params: seq[WireValu
     ctx.views[stmt.cvName] = stmt.cvQuery
     let viewKey = "_schema:views:" & stmt.cvName
     let viewSql = selectToSql(stmt.cvQuery)
-    let viewDdl = "CREATE VIEW " & stmt.cvName & " AS " & viewSql
+    let viewDdl = "CREATE VIEW \"" & sqlEscapeIdent(stmt.cvName) & "\" AS " & viewSql
     ctx.db.put(viewKey, cast[seq[byte]](viewDdl))
     return okResult(msg="CREATE VIEW " & stmt.cvName)
 
@@ -4762,7 +4811,7 @@ proc executeQueryImpl(ctx: ExecutionContext, astNode: Node, params: seq[WireValu
     ctx.tables[stmt.trigTable].triggers = triggers
     # Persist trigger to LSM-Tree
     let trigKey = "_schema:triggers:" & stmt.trigTable & ":" & stmt.trigName
-    let trigDdl = "CREATE TRIGGER " & stmt.trigName & " ON " & stmt.trigTable & " " &
+    let trigDdl = "CREATE TRIGGER \"" & sqlEscapeIdent(stmt.trigName) & "\" ON \"" & sqlEscapeIdent(stmt.trigTable) & "\" " &
                   stmt.trigTiming & " " & stmt.trigEvent & " AS " & stmt.trigAction.strVal
     ctx.db.put(trigKey, cast[seq[byte]](trigDdl))
     return okResult(msg="CREATE TRIGGER " & stmt.trigName)
@@ -5031,7 +5080,7 @@ proc executeQueryImpl(ctx: ExecutionContext, astNode: Node, params: seq[WireValu
     ctx.users[stmt.cuName] = UserDef(name: stmt.cuName, passwordHash: stmt.cuPassword,
                                      isSuperuser: stmt.cuSuperuser, roles: @[])
     let userKey = "_schema:users:" & stmt.cuName
-    let userDdl = "CREATE USER " & stmt.cuName & " WITH PASSWORD '" & stmt.cuPassword & "'" &
+    let userDdl = "CREATE USER \"" & sqlEscapeIdent(stmt.cuName) & "\" WITH PASSWORD '" & sqlEscapeString(stmt.cuPassword) & "'" &
                   (if stmt.cuSuperuser: " SUPERUSER" else: " NOSUPERUSER")
     ctx.db.put(userKey, cast[seq[byte]](userDdl))
     return okResult(msg="CREATE USER " & stmt.cuName)
@@ -5050,7 +5099,7 @@ proc executeQueryImpl(ctx: ExecutionContext, astNode: Node, params: seq[WireValu
                        withCheckExpr: stmt.cpWithCheck))
     ctx.policies[stmt.cpTable] = pols
     let polKey = "_schema:policies:" & stmt.cpTable & ":" & stmt.cpName
-    var polDdl = "CREATE POLICY " & stmt.cpName & " ON " & stmt.cpTable
+    var polDdl = "CREATE POLICY \"" & sqlEscapeIdent(stmt.cpName) & "\" ON \"" & sqlEscapeIdent(stmt.cpTable) & "\""
     if stmt.cpCommand != "ALL":
       polDdl.add(" FOR " & stmt.cpCommand)
     if stmt.cpUsing != nil:

src/barabadb/query/parser.nim

@@ -181,8 +181,8 @@ proc parsePrimary(p: var Parser): Node =
     Node(kind: nkCase, caseExpr: caseExpr, caseWhens: whens, caseElse: elseExpr,
          line: tok.line, col: tok.col)
   else:
-    discard p.advance()
-    Node(kind: nkNullLit, line: tok.line, col: tok.col)
+    raise newException(ValueError,
+      "Unexpected token " & $tok.kind & " at line " & $tok.line & ", col " & $tok.col)
 
 proc parseOverClause(p: var Parser): Node =
   ## Parse OVER ( [PARTITION BY expr, ...] [ORDER BY expr [ASC|DESC], ...] [frame] )
@@ -324,7 +324,10 @@ proc parseComparison(p: var Parser): Node =
     if p.peek().kind == tkNot:
       discard p.advance()
       negated = true
-    discard p.advance()  # consume NULL token (assumed)
+    if p.peek().kind != tkNull:
+      raise newException(ValueError,
+        "Expected NULL after IS " & (if negated: "NOT " else: "") & "at line " & $tok.line)
+    discard p.advance()  # consume NULL token
     return Node(kind: nkIsExpr, isExpr: result, isNegated: negated,
                 line: tok.line, col: tok.col)
   while p.peek().kind in {tkEq, tkNotEq, tkLt, tkLtEq, tkGt, tkGtEq, tkFtsMatch, tkDistanceOp, tkJsonContains, tkJsonContainedBy, tkJsonHasAny, tkJsonHasAll}:

src/barabadb/storage/bloom.nim

@@ -81,7 +81,7 @@ proc deserialize*(bf: var BloomFilter, data: seq[byte]) =
   )
   let numBytes = (bf.size + 7) div 8
   if data.len < 8 + numBytes:
-    return
+    raise newException(ValueError, "Bloom filter data too short: expected " & $(8 + numBytes) & " bytes, got " & $data.len)
   for i in 0..<bf.size:
     if (data[8 + i div 8] and (1'u8 shl (i mod 8))) != 0:
       bf.bits[i] = true

src/barabadb/storage/btree.nim

@@ -62,9 +62,12 @@ proc splitChild[K, V](parent: BTreeNode[K, V], index: int, order: int) =
   let midKey = child.keys[mid]
   parent.keys.insert(midKey, index)
   parent.children.insert(newNode, index + 1)
-  child.keys.setLen(mid)
+  # In B+ tree, leaf nodes must keep the boundary key for range scans
   if child.isLeaf:
-    child.values.setLen(mid)
+    child.keys.setLen(mid + 1)
+    child.values.setLen(mid + 1)
+  else:
+    child.keys.setLen(mid)
 
 proc insertNonFull[K, V](node: BTreeNode[K, V], key: K, value: V, order: int) =
   var i = node.keys.len - 1

src/barabadb/storage/compaction.nim

@@ -98,10 +98,9 @@ proc compact*(cs: CompactionStrategy, level: int): CompactionResult =
       merged.add(entry)
       lastKey = entry.key
 
-  # Filter out tombstones (deleted entries)
+  # Keep tombstones to prevent deleted keys from resurrecting in lower levels
   var final: seq[Entry] = @[]
   for entry in merged:
-    if not entry.deleted:
     final.add(entry)
 
   # Write merged SSTable

src/barabadb/storage/lsm.nim

@@ -59,6 +59,8 @@ proc len*(mt: MemTable): int = mt.entries.len
 
 proc put*(mt: var MemTable, key: string, value: seq[byte], timestamp: uint64, deleted: bool = false): bool =
   let entrySize = key.len + value.len + 16
+  if entrySize > mt.maxSize:
+    return false
   if mt.size + entrySize > mt.maxSize and mt.entries.len > 0:
     return false
   let entry = Entry(key: key, value: value, timestamp: timestamp, deleted: deleted)
@@ -318,7 +320,7 @@ proc newLSMTree*(dir: string, memMaxSize: int = DefaultMemTableSize): LSMTree =
       except:
         discard  # skip corrupt SSTables
 
-  sstables.sort(proc(a, b: SSTable): int = cmp(b.id, a.id))
+  sstables.sort(proc(a, b: SSTable): int = cmp(a.id, b.id))
 
   new(result)
   initLock(result.lock)
@@ -405,6 +407,21 @@ proc delete*(db: LSMTree, key: string) =
     db.memTable = newMemTable(db.memMaxSize)
     discard db.memTable.put(key, @[], ts, deleted = true)
 
+proc putUnsafe*(db: LSMTree, key: string, value: seq[byte], deleted: bool = false) =
+  ## Direct LSM insert without WAL logging — used by recovery.
+  let ts = uint64(getMonoTime().ticks())
+  acquire(db.lock)
+  defer: release(db.lock)
+  if not db.memTable.put(key, value, ts, deleted):
+    if db.immutableMem.len > 0:
+      db.flushUnsafe()
+    db.immutableMem = db.memTable
+    db.memTable = newMemTable(db.memMaxSize)
+    discard db.memTable.put(key, value, ts, deleted)
+
+proc deleteUnsafe*(db: LSMTree, key: string) =
+  putUnsafe(db, key, @[], deleted = true)
+
 proc getUnsafe(db: LSMTree, key: string): (bool, seq[byte]) =
   let (found, entry) = db.memTable.get(key)
   if found:
@@ -478,6 +495,9 @@ proc flush*(db: LSMTree) =
 proc close*(db: LSMTree) =
   acquire(db.lock)
   defer: release(db.lock)
+  # Flush both memtables to avoid data loss
+  while db.immutableMem.len > 0:
+    flushUnsafe(db)
   flushUnsafe(db)
   for sst in db.sstables.mitems:
     sst.close()
@@ -531,7 +551,8 @@ proc scanAll*(db: LSMTree): seq[(string, seq[byte])] =
         result.add((e.key, e.value))
 
   # Scan SSTables from newest to oldest
-  for sst in db.sstables:
+  for i in countdown(db.sstables.high, db.sstables.low):
+    let sst = db.sstables[i]
     for key, offset in sst.index:
       if key notin seen:
         seen[key] = true

src/barabadb/storage/mmap.nim

@@ -54,14 +54,18 @@ proc openMmap*(path: string, mode: MmapMode = mmReadOnly): MmapFile =
 
   let mapped = mmap(nil, fileSize, prot, flags, fd, 0)
   if mapped == MAP_FAILED:
-    discard close(fd)
+    discard posix.close(fd)
     return MmapFile(path: path, regions: @[], totalSize: 0, pageSize: PageSize)
 
+  # Kernel keeps the mapping alive independently of the fd; close it now
+  # to avoid leaking one fd per SSTable.
+  discard posix.close(fd)
+
   let region = MmapRegion(
     data: cast[ptr UncheckedArray[byte]](mapped),
     size: fileSize,
     offset: 0,
-    fd: fd,
+    fd: -1,
     mode: mode,
   )
 
@@ -116,15 +120,20 @@ proc adviseRandom*(mf: MmapFile) =
 
 proc adviseWillNeed*(mf: MmapFile, offset: int, size: int) =
   if mf.regions.len > 0:
+    if offset < 0 or offset + size > mf.regions[0].size:
+      return
     discard madvise(addr mf.regions[0].data[offset], size, MADV_WILLNEED)
 
 proc adviseDontNeed*(mf: MmapFile, offset: int, size: int) =
   if mf.regions.len > 0:
+    if offset < 0 or offset + size > mf.regions[0].size:
+      return
     discard madvise(addr mf.regions[0].data[offset], size, MADV_DONTNEED)
 
 proc close*(mf: MmapFile) =
   for region in mf.regions:
     discard munmap(region.data, region.size)
+    if region.fd != -1:
       discard close(cint(region.fd))
   mf.regions.setLen(0)
 

src/barabadb/storage/recovery.nim

@@ -32,6 +32,7 @@ type
     dataDir*: string
     entries*: seq[RecoveredEntry]
     result*: RecoveryResult
+    lastTxnId*: uint64  # tracks commits seen in WAL
 
 proc newCrashRecovery*(walDir: string, dataDir: string): CrashRecovery =
   CrashRecovery(
@@ -101,6 +102,7 @@ proc scanWAL*(rec: CrashRecovery): seq[RecoveredEntry] =
       discard
 
   stream.close()
+  rec.lastTxnId = txnId
 
 proc analyze*(rec: CrashRecovery): RecoveryResult =
   rec.entries = rec.scanWAL()
@@ -109,10 +111,7 @@ proc analyze*(rec: CrashRecovery): RecoveryResult =
     rec.result = RecoveryResult(state: recDone, applied: false)
     return rec.result
 
-  var lastCommitted: uint64 = 0
-  for entry in rec.entries:
-    if entry.txnId > lastCommitted:
-      lastCommitted = entry.txnId
+  var lastCommitted = rec.lastTxnId
 
   var redoCount = 0
   var undoCount = 0
@@ -149,11 +148,11 @@ proc recover*(rec: CrashRecovery, db: LSMTree = nil): RecoveryResult =
   var undoCount = 0
   for entry in rec.entries:
     if entry.txnId < analysis.lastTxn:
-      # Committed — redo
+      # Committed — redo (bypass WAL to avoid duplicate entries)
       if entry.isDelete:
-        db.delete(entry.key)
+        db.deleteUnsafe(entry.key)
       else:
-        db.put(entry.key, entry.value)
+        db.putUnsafe(entry.key, entry.value)
       inc redoCount
     else:
       # Uncommitted — skip (undo)

src/barabadb/storage/wal.nim

@@ -30,10 +30,11 @@ proc newWriteAheadLog*(dir: string, syncOnWrite: bool = true): WriteAheadLog =
   createDir(dir)
   let path = dir / "wal.log"
   let exists = fileExists(path)
+  let isEmpty = if exists: getFileSize(path) == 0 else: true
   let stream = if exists: newFileStream(path, fmAppend) else: newFileStream(path, fmWrite)
   if stream == nil:
     raise newException(IOError, "Cannot open WAL: " & path)
-  if not exists:
+  if not exists or isEmpty:
     stream.write(WALMagic)
     stream.write(WALVersion)
     stream.flush()
@@ -78,14 +79,17 @@ proc writeCommit*(wal: var WriteAheadLog, timestamp: uint64) =
 
 proc sync*(wal: var WriteAheadLog) =
   wal.stream.flush()
-  let fd = posix.open(cstring(wal.path), O_RDONLY)
+  # Re-open with O_RDWR so fsync operates on a write-capable fd.
+  # Not ideal (two fds for same file) but avoids accessing private
+  # FileStream internals that vary across Nim versions.
+  let fd = posix.open(cstring(wal.path), O_RDWR)
   if fd != -1:
     discard posix.fsync(fd)
     discard posix.close(fd)
 
 proc close*(wal: var WriteAheadLog) =
   wal.stream.flush()
-  let fd = posix.open(cstring(wal.path), O_RDONLY)
+  let fd = posix.open(cstring(wal.path), O_RDWR)
   if fd != -1:
     discard posix.fsync(fd)
     discard posix.close(fd)
@@ -100,10 +104,10 @@ proc readEntries*(walPath: string, untilTimestamp: uint64 = 0): seq[WalEntry] =
   let s = newFileStream(walPath, fmRead)
   if s == nil: return
   # Skip header
-  var magic: uint32
-  var version: uint32
-  discard s.readData(addr magic, 4)
-  discard s.readData(addr version, 4)
+  var magic: uint32 = 0
+  var version: uint32 = 0
+  if s.readData(addr magic, 4) != 4: return
+  if s.readData(addr version, 4) != 4: return
   if magic != WALMagic: return
   while not s.atEnd:
     var kind: uint8

tests/test_all.nim

@@ -3845,6 +3845,59 @@ suite "Foreign Key Enforcement":
     check childSel.rows.len == 0
 
 
+suite "Composite Primary Key":
+  var db: LSMTree
+  var ctx: qexec.ExecutionContext
+  var tmpDir: string
+
+  setup:
+    tmpDir = getTempDir() / "baradb_compk_test_" & $getMonoTime().ticks
+    db = newLSMTree(tmpDir)
+    ctx = qexec.newExecutionContext(db)
+
+  teardown:
+    removeDir(tmpDir)
+
+  test "Composite PK insert stores distinct rows":
+    discard qexec.executeQuery(ctx, parse("CREATE TABLE orders (user_id INT, order_id INT, amount INT, PRIMARY KEY (user_id, order_id))"))
+    discard qexec.executeQuery(ctx, parse("INSERT INTO orders (user_id, order_id, amount) VALUES (1, 100, 50)"))
+    discard qexec.executeQuery(ctx, parse("INSERT INTO orders (user_id, order_id, amount) VALUES (1, 101, 75)"))
+    let sel = qexec.executeQuery(ctx, parse("SELECT * FROM orders"))
+    check sel.success
+    check sel.rows.len == 2
+
+  test "Composite PK prevents duplicate key":
+    discard qexec.executeQuery(ctx, parse("CREATE TABLE items (warehouse INT, sku TEXT, qty INT, PRIMARY KEY (warehouse, sku))"))
+    discard qexec.executeQuery(ctx, parse("INSERT INTO items (warehouse, sku, qty) VALUES (1, 'ABC', 10)"))
+    let dup = qexec.executeQuery(ctx, parse("INSERT INTO items (warehouse, sku, qty) VALUES (1, 'ABC', 20)"))
+    check not dup.success
+    check dup.message.contains("UNIQUE constraint violated")
+
+  test "Composite PK allows same first column with different second":
+    discard qexec.executeQuery(ctx, parse("CREATE TABLE scores (player INT, level INT, score INT, PRIMARY KEY (player, level))"))
+    discard qexec.executeQuery(ctx, parse("INSERT INTO scores (player, level, score) VALUES (1, 1, 100)"))
+    discard qexec.executeQuery(ctx, parse("INSERT INTO scores (player, level, score) VALUES (1, 2, 200)"))
+    let sel = qexec.executeQuery(ctx, parse("SELECT * FROM scores"))
+    check sel.success
+    check sel.rows.len == 2
+
+  test "Composite PK SELECT returns correct rows":
+    discard qexec.executeQuery(ctx, parse("CREATE TABLE pair (a INT, b INT, val TEXT, PRIMARY KEY (a, b))"))
+    discard qexec.executeQuery(ctx, parse("INSERT INTO pair (a, b, val) VALUES (1, 2, 'first')"))
+    discard qexec.executeQuery(ctx, parse("INSERT INTO pair (a, b, val) VALUES (1, 3, 'second')"))
+    let sel = qexec.executeQuery(ctx, parse("SELECT * FROM pair"))
+    check sel.success
+    check sel.rows.len == 2
+    # Verify both rows exist (order may vary)
+    var foundFirst = false
+    var foundSecond = false
+    for row in sel.rows:
+      if row["val"] == "first": foundFirst = true
+      if row["val"] == "second": foundSecond = true
+    check foundFirst
+    check foundSecond
+
+
 suite "Hybrid RAG Search":
   var db: LSMTree
   var ctx: qexec.ExecutionContext