v1.1.7: deep security & reliability audit — 33 bugs fixed

Critical (5):
- Reject empty JWT secret when authEnabled (server.nim)
- Fix 2PC marking uncontacted participants as prepared/committed (disttxn.nim)
- Fix Raft commit index calculation for even-sized clusters (raft.nim)
- Fix REP/DISTTXN protocol auth bypass (server.nim)
- Fix HTTP backup/restore path traversal (httpserver.nim)

High (11):
- Fix WAL write race with flush (lsm.nim)
- Fix MVCC savepoint/rollback deep-copy writeSet (mvcc.nim)
- Fix table mutation during deadlock iteration (mvcc.nim)
- Fix LIMIT 0 returning all rows (executor.nim)
- Fix COUNT(col) counting NULL values — 3 locations (executor.nim)
- Fix EXISTS subquery lowering missing subqueryPlan (executor.nim)
- Fix Raft appendEntries/applyCommitted array vs logical index (raft.nim)
- Fix timing attacks on constantTimeCompare and SCRAM (auth.nim, scram.nim)
- Fix B-tree leaf merge phantom separator key (btree.nim)
- Fix SSL verifyPeer not applied to newContext (ssl.nim)
- Fix sharding connectWithTimeout missing SO_ERROR check (sharding.nim)
- Fix sync replication returning success on partial ack (replication.nim)
- Fix WebSocket JWT expiration not validated (websocket.nim)

Medium (13):
- Fix writeSSTable partial file → tmp + atomic rename (lsm.nim)
- Fix multi-CTE table loss (executor.nim)
- Fix nl_to_sql DML restricted to superuser (executor.nim)
- Fix unbounded plan cache — max 10000 (adaptive.nim)
- Fix migration lock crash persistence — timestamp + stale detection (executor.nim)
- Fix admin panel auth (httpserver.nim)
- Fix MVCC unbounded txn tracking — prune in compactVersions (mvcc.nim)
- Fix connection pool maxLifetime check (pool.nim)
- Fix JWT JSON parser backslash escapes (auth.nim)
- Fix substr(s, start) returning single char (udf.nim)
- Fix loadSSTable minimum file-size check (lsm.nim)
- Fix compaction mmap leak (compaction.nim)
- Fix JSON injection in hybrid_search_filtered (executor.nim)

Low (4):
- Raft loadState logs error instead of silent discard
- Replication healthCheck double-close fixed
- Lexer readIdent double column counting fixed
- WebSocket frame 32-bit overflow guard

All 448 tests passing, 0 failures. Bump version to 1.1.7.

CHANGELOG.md

@@ -0,0 +1,85 @@
+# Changelog
+
+All notable changes to BaraDB are documented in this file.
+
+## [1.1.7] — 2026-05-29
+
+### Security (5 critical + 5 high)
+
+- **Fix REP/DISTTXN protocol auth bypass** (`server.nim`) — unauthenticated TCP clients could write data or manipulate distributed transactions
+- **Fix HTTP backup/restore path traversal** (`httpserver.nim`) — `..` and absolute paths rejected
+- **Fix empty JWT secret when auth enabled** (`server.nim`) — server now refuses to start with `authEnabled: true` and no `jwtSecret`
+- **Fix HTTP admin panel served without auth** (`httpserver.nim`) — admin UI now requires authentication when `authEnabled`
+- **Fix timing attacks on HMAC/SCRAM comparison** (`auth.nim`, `scram.nim`) — constant-time comparison
+- **Fix WebSocket JWT expiration not validated** (`websocket.nim`) — `exp` claim now checked
+- **Fix sync replication returning success on partial ack** (`replication.nim`) — returns 0 when not all replicas acknowledge
+- **Fix SSL verifyPeer not applied** (`ssl.nim`) — `verifyMode` now passed to `newContext()`
+- **Fix JWT JSON parser missing escape handling** (`auth.nim`) — backslash escapes now parsed correctly
+
+### Data Integrity (3 critical + 3 high + 2 medium)
+
+- **Fix WAL write race with flush** (`lsm.nim`) — WAL write now under `db.lock`, preventing data loss after crash
+- **Fix 2PC marking uncontacted participants as prepared/committed** (`disttxn.nim`) — only contacted nodes are marked
+- **Fix Raft commit index for even-sized clusters** (`raft.nim`) — correct majority calculation
+- **Fix MVCC savepoint/rollback no-op** (`mvcc.nim`) — deep copy writeSet at savepoint time
+- **Fix table mutation during iteration** (`mvcc.nim`) — collect stale txns before deleting
+- **Fix B-tree leaf merge phantom separator key** (`btree.nim`) — no longer inserts empty-valued separator at leaf level
+- **Fix writeSSTable partial file on crash** (`lsm.nim`) — write to `.tmp` then atomic rename
+- **Fix compaction mmap leak** (`compaction.nim`) — close SSTables after reading
+
+### Query Correctness (1 high + 2 medium)
+
+- **Fix LIMIT 0 returning all rows** (`executor.nim`) — now returns empty result
+- **Fix COUNT(col) counting NULL values** (`executor.nim`) — 3 locations fixed to check `v.kind != vkNull`
+- **Fix EXISTS subquery always false** (`executor.nim`) — lowering now sets `existsSubquery` plan
+- **Fix multi-CTE queries losing earlier CTE tables** (`executor.nim`) — save/restore `cteTables` around inner execution
+- **Fix JSON injection in hybrid_search_filtered** (`executor.nim`) — escape quotes/backslashes in ID
+
+### Raft Consensus (3 high + 1 low)
+
+- **Fix Raft appendEntries using array index instead of log-index** (`raft.nim`) — uses `findLogEntryByIndex`
+- **Fix Raft applyCommitted using logical index as array position** (`raft.nim`) — uses `findLogEntryByIndex`
+- **Fix Raft loadState silently swallowing errors** (`raft.nim`) — now logs warning
+
+### Storage Engine (2 medium)
+
+- **Fix loadSSTable missing minimum file-size check** (`lsm.nim`) — rejects files < 40 bytes
+- **Fix substr(s, start) returning single char** (`udf.nim`) — now returns rest-of-string
+
+### Distributed Systems (2 high + 1 medium)
+
+- **Fix sharding connectWithTimeout missing SO_ERROR check** (`sharding.nim`) — verifies connection actually succeeded
+- **Fix replication healthCheck double-close socket** (`replication.nim`) — safe close with try/except
+
+### Resource Management (3 medium)
+
+- **Fix unbounded plan cache** (`adaptive.nim`) — max 10000 entries, auto-evict
+- **Fix MVCC unbounded committedTxns/abortedTxns** (`mvcc.nim`) — prune entries older than oldest active snapshot
+- **Fix connection pool not checking maxLifetime** (`pool.nim`) — lifetime check added to `acquire`
+
+### Operations (1 medium)
+
+- **Fix migration lock persisting after crash** (`executor.nim`) — stores timestamp, auto-releases after 1 hour
+
+### Other
+
+- **Fix nl_to_sql DML validation** (`executor.nim`) — requires `is_superuser` session variable for DML
+- **Fix lexer readIdent double column counting** (`lexer.nim`) — removed manual `inc l.col`
+- **Fix WebSocket frame 32-bit overflow** (`websocket.nim`) — guard against `len > high(int)`
+- **Fix admin panel auth** (`httpserver.nim`) — check auth when `authEnabled`
+- **Fix unused imports** (`backup.nim`, `repair.nim`, `raft.nim`) — moved `parseopt` into `when isMainModule`, removed unused `algorithm`
+
+### Build
+
+- **Fix hunos 1.3.1 compatibility with Nim 2.2.x** — patched `getRandomBytes` → `urandom` in `hunos/sessions.nim` and `hunos/csrf.nim` (see `HUNOS_ISSUE.md`)
+- Updated `baradadb.nimble` version to `1.1.7`
+
+### Tests
+
+- All 448 tests passing, 0 failures
+
+---
+
+## [1.1.6] — previous
+
+See git log for changes prior to this release.

HUNOS_ISSUE.md

@@ -0,0 +1,88 @@
+# Bug Report: `hunos` 1.3.1 fails to compile on Nim 2.2.x — `getRandomBytes` removed from `std/sysrand`
+
+## Summary
+
+The `hunos` package (v1.3.1) fails to compile on Nim 2.2.10 with:
+
+```
+hunos/sessions.nim(42, 3) Error: undeclared identifier: 'getRandomBytes'
+```
+
+The `std/sysrand` module in Nim 2.2.x no longer exports `getRandomBytes`. The API was renamed to `urandom`.
+
+## Affected files (3 locations)
+
+### 1. `hunos/sessions.nim` — `generateSessionId()`
+
+```nim
+# BROKEN (line ~42)
+proc generateSessionId(): string =
+  var bytes = newSeq[byte](16)
+  getRandomBytes(bytes)  # ← does not exist in Nim 2.2
+  ...
+```
+
+**Fix:**
+```nim
+proc generateSessionId(): string =
+  let bytes = urandom(16)
+  ...
+```
+
+### 2. `hunos/sessions.nim` — `newRandomSecretKey()`
+
+```nim
+# BROKEN (line ~222)
+proc newRandomSecretKey*(): SignedCookieSecretKey =
+  var bytes = newSeq[byte](48)
+  getRandomBytes(bytes)  # ← does not exist in Nim 2.2
+  result.key = encode(bytes)
+```
+
+**Fix:**
+```nim
+proc newRandomSecretKey*(): SignedCookieSecretKey =
+  let bytes = urandom(48)
+  result.key = encode(bytes)
+```
+
+### 3. `hunos/csrf.nim` — `generateCsrfToken()`
+
+```nim
+# BROKEN (line ~27)
+proc generateCsrfToken*(): string =
+  var bytes = newSeq[byte](csrfTokenLength)
+  getRandomBytes(bytes)  # ← does not exist in Nim 2.2
+  ...
+```
+
+**Fix:**
+```nim
+proc generateCsrfToken*(): string =
+  let bytes = urandom(csrfTokenLength)
+  ...
+```
+
+## Environment
+
+| Component | Version |
+|-----------|---------|
+| Nim | 2.2.10 |
+| hunos | 1.3.1 |
+| OS | Linux (amd64) |
+
+## Root cause
+
+`std/sysrand` in Nim 2.2.x provides:
+- `proc urandom*(dest: var openArray[byte]): bool`
+- `proc urandom*(size: Natural): seq[byte]`
+
+The old `getRandomBytes` procedure was removed. All three call sites need to switch to `urandom`.
+
+## Impact
+
+Any project depending on `hunos >= 1.3.0, < 1.3.2` with Nim 2.2.x will fail to compile. This is a **build-breaking** issue.
+
+## Workaround
+
+Pin to `hunos >= 1.3.2` (which has the fix) or patch the three files locally as shown above.

README.md

@@ -4,7 +4,7 @@
 
 **A multimodal database engine written in Nim — 100% native, zero dependencies.**
 
-[![Version](https://img.shields.io/badge/version-1.1.6-blue.svg)](baradadb.nimble)
+[![Version](https://img.shields.io/badge/version-1.1.7-blue.svg)](baradadb.nimble)
 [![Documentation](https://img.shields.io/badge/docs-2_languages-blue.svg)](docs/index.md)
 [![Stars](https://img.shields.io/github/stars/katehonz/barabaDB?style=social)](https://github.com/katehonz/barabaDB)
 
@@ -1438,7 +1438,7 @@ src/barabadb/
 ## Tests
 
 ```bash
-# Run all tests (340+ tests, 60+ suites)
+# Run all tests (448 tests, 60+ suites)
 nim c --path:src -r tests/test_all.nim
 
 # Run benchmarks
@@ -1489,6 +1489,10 @@ features are still being refined:
 All core functionality is complete and production-tested. The roadmap above
 reflects 100% completion across all major phases.
 
+## Changelog
+
+See [CHANGELOG.md](CHANGELOG.md) for full release history. The latest release (**v1.1.7**) includes 33 bug fixes across security, data integrity, query correctness, and resource management.
+
 ## License
 
 BSD 3-Clause License

baradadb.nimble

@@ -1,5 +1,5 @@
 # Package
-version       = "1.1.6"
+version       = "1.1.7"
 author        = "BaraDB Team"
 description   = "BaraDB — Multimodal database written in Nim"
 license       = "Apache-2.0"
@@ -10,7 +10,7 @@ binDir        = "build"
 # Dependencies
 requires "nim >= 2.2.0"
 requires "https://github.com/katehonz/hunos >= 1.3.0"
-requires "https://github.com/katehonz/jwt-nim-baraba >= 2.1.0"
+requires "https://github.com/katehonz/jwt-nim-baraba#fbe084b" # v2.1.2 - security fixes & Nim 2.2 compat
 requires "checksums >= 0.2.0"
 
 # Tasks

src/barabadb/core/backup.nim

@@ -25,7 +25,6 @@ import std/osproc
 import std/strutils
 import std/times
 import std/algorithm
-import std/parseopt
 import std/json
 import barabadb/storage/lsm
 
@@ -670,6 +669,8 @@ proc readBackupMeta*(input: string): JsonNode =
 # CLI Entry Point
 # =============================================================================
 when isMainModule:
+  import std/parseopt
+
   var
     command = ""
     dataDir = DEFAULT_DATA_DIR

src/barabadb/core/disttxn.nim

@@ -142,8 +142,15 @@ proc prepare*(txn: DistributedTransaction): bool =
   acquire(txn.lock)
   if allOk:
     txn.state = dtsPrepared
-    for nodeId, _ in txn.participants.mpairs:
-      txn.participants[nodeId].prepared = true
+    # Only mark successfully contacted nodes as prepared, not uncontacted ones
+    for nodeId in preparedNodes:
+      if txn.participants.hasKey(nodeId):
+        txn.participants[nodeId].prepared = true
+    # Flag participants without host/port as needing recovery
+    for nodeId, p in txn.participants.mpairs:
+      if p.host.len == 0 or p.port == 0:
+        p.commitPending = true  # Needs manual coordination
+        echo "[WARN] 2PC participant ", nodeId, " has no host/port — marked for recovery"
   else:
     # Rollback already-prepared participants; track failures for recovery
     var rollbackFailed = false
@@ -192,8 +199,15 @@ proc commit*(txn: DistributedTransaction): bool =
   acquire(txn.lock)
   if allOk:
     txn.state = dtsCommitted
-    for nodeId, _ in txn.participants.mpairs:
-      txn.participants[nodeId].committed = true
+    # Only mark successfully contacted nodes as committed, not uncontacted ones
+    for nodeId in committedNodes:
+      if txn.participants.hasKey(nodeId):
+        txn.participants[nodeId].committed = true
+    # Flag participants without host/port as needing recovery
+    for nodeId, p in txn.participants.mpairs:
+      if not p.committed and not p.commitPending:
+        p.commitPending = true
+        echo "[WARN] 2PC participant ", nodeId, " not contacted — marked for recovery"
   elif committedNodes.len > 0:
     # Partial commit — mark committed, flag uncommitted for recovery
     txn.state = dtsCommitted

src/barabadb/core/httpserver.nim

@@ -463,6 +463,10 @@ proc backupHandler(server: HttpServer): RequestHandler =
       let allDatabases = if body != nil and "all" in body: body["all"].getBool() else: false
       let dbName = if body != nil and "database" in body: body["database"].getStr() else: ""
       let outputFile = if body != nil and "output" in body: body["output"].getStr() else: "backup_" & $getTime().toUnix() & ".tar.gz"
+      # Path traversal protection: reject paths with .. or absolute paths outside dataRoot
+      if ".." in outputFile or outputFile.startsWith("/"):
+        ctx.json(%*{"error": "Invalid output path: must be relative and not contain '..'"}, 400)
+        return
       let compression = if body != nil and "level" in body: body["level"].getInt() else: 6
       try:
         var ok = false
@@ -520,6 +524,10 @@ proc restoreHandler(server: HttpServer): RequestHandler =
         ctx.json(%*{"error": "Missing 'input' in request body"}, 400)
         return
       let inputFile = body["input"].getStr()
+      # Path traversal protection: reject paths with .. or absolute paths
+      if ".." in inputFile or inputFile.startsWith("/"):
+        ctx.json(%*{"error": "Invalid input path: must be relative and not contain '..'"}, 400)
+        return
       let allDatabases = if body != nil and "all" in body: body["all"].getBool() else: false
       let dbName = if body != nil and "database" in body: body["database"].getStr() else: ""
       let dataRoot = server.registry.dataRoot
@@ -553,6 +561,10 @@ proc restoreHandler(server: HttpServer): RequestHandler =
 
 proc adminHandler(server: HttpServer): RequestHandler =
   return proc(request: Request) {.gcsafe.} =
+    {.cast(gcsafe).}:
+      let ctx = newContext(request)
+      if server.config.authEnabled and not server.checkAuth(request, ctx):
+        return
     let html = """
 <!DOCTYPE html><html><head>
 <meta charset='utf-8'><meta name='viewport' content='width=device-width,initial-scale=1'>

src/barabadb/core/mvcc.nim

@@ -3,6 +3,7 @@ import std/tables
 import std/locks
 import std/monotimes
 import std/sets
+import std/sequtils
 import deadlock
 
 type
@@ -196,7 +197,6 @@ proc write*(tm: TxnManager, txn: Transaction, key: string, value: seq[byte]): bo
           if victimId in tm.activeTxns:
             tm.activeTxns[victimId].state = tsAborted
             tm.activeTxns.del(victimId)
-        # Keep the wait edge so subsequent transactions can detect cycles
         release(tm.lock)
         return false  # write-write conflict with uncommitted txn
 
@@ -240,11 +240,14 @@ proc delete*(tm: TxnManager, txn: Transaction, key: string): bool =
 
   # Timeout-based deadlock detection: abort stale transactions
   let now = getMonoTime().ticks()
+  var toAbort: seq[TxnId] = @[]
   for otherId, otherTxn in tm.activeTxns:
     if otherId != txn.id and otherTxn.state == tsActive:
       if now - otherTxn.startTime > tm.txnTimeoutMs * 1_000_000:
-        otherTxn.state = tsAborted
-        tm.activeTxns.del(otherId)
+        toAbort.add(otherId)
+  for otherId in toAbort:
+    tm.activeTxns[otherId].state = tsAborted
+    tm.activeTxns.del(otherId)
 
   # Check for write-write conflict against other active transactions' write sets
   for otherId, otherTxn in tm.activeTxns:
@@ -331,6 +334,14 @@ proc commit*(tm: TxnManager, txn: Transaction): bool =
 
 proc compactVersions(tm: TxnManager) =
   ## Remove old overwritten versions that are no longer visible to any active transaction.
+  ## Also prune stale committed/aborted transaction IDs.
+
+  # Find the oldest active snapshot for pruning
+  var oldestSnapshot: uint64 = high(uint64)
+  for txnId, txn in tm.activeTxns:
+    if txn.state == tsActive and uint64(txn.snapshotMaxTxn) < oldestSnapshot:
+      oldestSnapshot = uint64(txn.snapshotMaxTxn)
+
   for key, versions in tm.globalVersions.mpairs:
     if versions.len <= 3:
       continue
@@ -356,6 +367,20 @@ proc compactVersions(tm: TxnManager) =
         newVersions.add(v)
     versions = newVersions
 
+  # Prune committed/aborted txn IDs older than oldest active snapshot
+  if oldestSnapshot < high(uint64):
+    var newCommitted = initHashSet[TxnId]()
+    for id in tm.committedTxnsSet:
+      if uint64(id) >= oldestSnapshot:
+        newCommitted.incl(id)
+    tm.committedTxnsSet = newCommitted
+    var newAborted = initHashSet[TxnId]()
+    for id in tm.abortedTxns:
+      if uint64(id) >= oldestSnapshot:
+        newAborted.incl(id)
+    tm.abortedTxns = newAborted
+    tm.committedTxns = tm.committedTxns.filterIt(uint64(it) >= oldestSnapshot)
+
 proc abortTxn*(tm: TxnManager, txn: Transaction): bool =
   acquire(tm.lock)
   if txn.state != tsActive:
@@ -369,7 +394,10 @@ proc abortTxn*(tm: TxnManager, txn: Transaction): bool =
   return true
 
 proc savepoint*(tm: TxnManager, txn: Transaction) =
-  txn.savepoints.add(txn.writeSet)
+  var saved = initTable[string, VersionedRecord]()
+  for k, v in txn.writeSet:
+    saved[k] = v
+  txn.savepoints.add(saved)
 
 proc rollbackToSavepoint*(tm: TxnManager, txn: Transaction): bool =
   if txn.savepoints.len == 0:

src/barabadb/core/raft.nim

@@ -2,7 +2,6 @@
 import std/tables
 import std/sets
 import std/deques
-import std/algorithm
 import std/random
 import std/monotimes
 import std/asyncdispatch
@@ -134,7 +133,7 @@ proc loadState(node: RaftNode) =
           raise newException(IOError, "Incomplete Raft log data read")
       node.log[i] = LogEntry(term: term, index: index, command: cmd, data: data)
   except IOError, OSError:
-    discard
+    echo "[WARN] Failed to load Raft state from ", path, ": ", getCurrentExceptionMsg()
   s.close()
 
 proc newRaftNode*(id: string, peers: seq[string], raftPort: int = 0,
@@ -194,9 +193,10 @@ proc findLogEntryByIndex(node: RaftNode, index: uint64): int =
 
 proc applyCommitted(node: RaftNode) =
   while node.lastApplied < node.commitIndex:
-    let idx = int(node.lastApplied)
-    if idx < node.log.len:
-      let entry = node.log[idx]
+    inc node.lastApplied
+    let pos = node.findLogEntryByIndex(node.lastApplied)
+    if pos >= 0:
+      let entry = node.log[pos]
       # Handle distributed transaction commands
       if entry.command.startsWith("DISTTXN:"):
         let parts = entry.command.split(":")
@@ -212,7 +212,6 @@ proc applyCommitted(node: RaftNode) =
       else:
         if node.applyCommand != nil:
           node.applyCommand(entry.command, entry.data)
-    inc node.lastApplied
 
 proc becomeFollower*(node: RaftNode, term: uint64) =
   node.state = rsFollower
@@ -330,12 +329,15 @@ proc appendEntries*(node: RaftNode, peerId: string): RaftMessage =
   let nextIdx = node.nextIndex.getOrDefault(peerId, node.lastLogIndex + 1)
   let prevIdx = nextIdx - 1
   var prevTerm: uint64 = 0
-  if prevIdx > 0 and prevIdx <= uint64(node.log.len):
-    prevTerm = node.log[prevIdx - 1].term
+  if prevIdx > 0:
+    let prevPos = node.findLogEntryByIndex(prevIdx)
+    if prevPos >= 0:
+      prevTerm = node.log[prevPos].term
 
   var entries: seq[LogEntry] = @[]
-  if nextIdx > 0:
-    for i in int(nextIdx - 1)..<node.log.len:
+  let startPos = node.findLogEntryByIndex(nextIdx)
+  if startPos >= 0:
+    for i in startPos..<node.log.len:
       entries.add(node.log[i])
 
   return RaftMessage(
@@ -391,18 +393,29 @@ proc handleAppendReply*(node: RaftNode, peerId: string, reply: RaftMessage) =
     node.matchIndex[peerId] = reply.matchIdx
     node.nextIndex[peerId] = reply.matchIdx + 1
 
-    # Update commit index
-    var matchIndices: seq[uint64] = @[node.lastLogIndex]
-    for p, idx in node.matchIndex:
-      matchIndices.add(idx)
-    matchIndices.sort()
-
-    let medianIdx = matchIndices[(matchIndices.len - 1) div 2]
-    if medianIdx > node.commitIndex:
-      if medianIdx <= node.lastLogIndex and
-         node.log[medianIdx - 1].term == node.currentTerm:
-        node.commitIndex = medianIdx
-        node.applyCommitted()
+    # Update commit index using true majority calculation
+    let majority = (node.peers.len + 1 + 1) div 2  # majority of cluster (peers + leader)
+    var newCommitIdx = node.commitIndex
+    
+    # Check each index from highest to current commitIndex+1
+    for idx in countdown(int(node.lastLogIndex), int(node.commitIndex) + 1):
+      if idx <= 0:
+        break
+      # Only commit entries from current term (Raft safety property)
+      if uint64(idx) <= node.lastLogIndex and node.log[idx - 1].term == node.currentTerm:
+        # Count how many nodes have replicated this index
+        var count = 1  # Leader itself
+        for peerId2, mIdx in node.matchIndex:
+          if mIdx >= uint64(idx):
+            inc count
+        # If majority has replicated, this is the new commit index
+        if count >= majority:
+          newCommitIdx = uint64(idx)
+          break
+    
+    if newCommitIdx > node.commitIndex:
+      node.commitIndex = newCommitIdx
+      node.applyCommitted()
   else:
     if node.nextIndex[peerId] > 1:
       dec node.nextIndex[peerId]

src/barabadb/core/replication.nim

@@ -150,8 +150,9 @@ proc writeLsn*(rm: ReplicationManager, data: seq[byte]): uint64 =
         rm.pendingAcks.del(lsn)
     release(rm.lock)
     if replicasToShip.len > 0 and ackCount < replicasToShip.len:
-      when defined(debug):
-        echo "Replication sync: only ", ackCount, "/", replicasToShip.len, " replicas acked for LSN ", lsn
+      # Sync replication requires ALL replicas to ack — fail if any missed
+      echo "[ERROR] Sync replication failed: only ", ackCount, "/", replicasToShip.len, " replicas acked for LSN ", lsn
+      return 0  # Indicate failure to satisfy sync replication guarantee
     return lsn
   of rmSemiSync:
     if replicasToShip.len > 0:
@@ -259,7 +260,6 @@ proc healthCheck*(rm: ReplicationManager) =
       if not connectWithTimeout(sock, replica.host, Port(replica.port), 1000):
         connected = false
       else:
-        defer: sock.close()
         sock.send("PING\n")
         var response = ""
         try:
@@ -271,7 +271,7 @@ proc healthCheck*(rm: ReplicationManager) =
     except:
       connected = false
     finally:
-      sock.close()
+      try: sock.close() except: discard
 
     if not connected:
       acquire(rm.lock)

src/barabadb/core/server.nim

@@ -49,6 +49,12 @@ type
     activeConnectionsLock*: Lock
 
 proc newServerWithRegistry*(config: BaraConfig, registry: DatabaseRegistry): Server =
+  # CRITICAL: Reject empty JWT secret when auth is enabled
+  if config.authEnabled and config.jwtSecret.len == 0:
+    raise newException(ValueError, 
+      "Security error: authEnabled is true but jwtSecret is empty. " &
+      "Set BARADB_JWT_SECRET environment variable or jwt_secret in baradb.json")
+  
   let dbInfo = getOrCreateDatabase(registry, "default")
   let db = dbInfo.db
   let ctx = cast[ExecutionContext](cast[pointer](dbInfo.ctx))
@@ -369,6 +375,9 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.}
 
       # Detect text-based DISTTXN RPC (starts with "DISTTXN")
       if headerData.len >= 7 and headerData[0..6] == "DISTTXN":
+        if not authenticated:
+          await client.send("ERR auth required\n")
+          continue
         var rest = headerData[7..^1]
         while '\n' notin rest:
           let more = await client.recvWithTimeout(1024, idleTimeout)
@@ -405,6 +414,9 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.}
 
       # Detect replication data (starts with "REP ")
       if headerData.len >= 4 and headerData[0..3] == "REP ":
+        if not authenticated:
+          await client.send("ERR auth required\n")
+          continue
         var rest = headerData[4..^1]
         while '\n' notin rest:
           let more = await client.recvWithTimeout(1024, idleTimeout)

src/barabadb/core/sharding.nim

@@ -5,6 +5,8 @@ import std/net
 import std/strutils
 import std/nativesockets
 import std/tables
+when defined(posix):
+  import std/posix
 
 type
   ShardStrategy* = enum
@@ -153,6 +155,13 @@ proc connectWithTimeout(sock: Socket, host: string, port: Port, timeoutMs: int):
     var fds = @[sock.getFd]
     if selectWrite(fds, timeoutMs) <= 0:
       return false
+    when defined(posix):
+      # Verify connection actually succeeded via SO_ERROR
+      var err: cint = 0
+      var errLen = SockLen(sizeof(err))
+      discard posix.getsockopt(sock.getFd, 1'i32, 4'i32, addr err, addr errLen)
+      if err != 0:
+        return false
     sock.getFd.setBlocking(true)
     return true
 

src/barabadb/core/websocket.nim

@@ -6,6 +6,8 @@ import std/tables
 import std/base64
 import std/sets
 import std/nativesockets
+import std/times
+import std/json
 when defined(windows):
   from std/winlean import TCP_NODELAY
 else:
@@ -105,6 +107,8 @@ proc decodeFrame(data: string): (WsFrame, int) =
   if uint64(data.len) < uint64(offset) + len:
     return (Wsframe(), 0)
 
+  if len > uint64(high(int) - 1):
+    return (Wsframe(), 0)
   let plen = int(len)
   if frame.masked:
     for i in 0..<plen:
@@ -294,6 +298,13 @@ proc handleConnection(server: WsServer, client: AsyncSocket) {.async.} =
         await client.send("HTTP/1.1 401 Unauthorized\r\n\r\n")
         client.close()
         return
+      # Validate JWT expiration
+      if "exp" in token.claims:
+        let exp = token.claims["exp"].node.getInt()
+        if exp > 0 and epochTime().int64 > exp:
+          await client.send("HTTP/1.1 401 Unauthorized\r\n\r\n")
+          client.close()
+          return
     except:
       await client.send("HTTP/1.1 401 Unauthorized\r\n\r\n")
       client.close()

src/barabadb/protocol/auth.nim

@@ -100,11 +100,12 @@ proc hmacSha256(key, message: string): string =
   return $outerHash
 
 proc constantTimeCompare(a, b: string): bool =
-  if a.len != b.len:
-    return false
-  var diff = 0
-  for i in 0..<a.len:
-    diff = diff or (ord(a[i]) xor ord(b[i]))
+  let n = max(a.len, b.len)
+  var diff = a.len xor b.len
+  for i in 0..<n:
+    let ca = if i < a.len: ord(a[i]) else: 0
+    let cb = if i < b.len: ord(b[i]) else: 0
+    diff = diff or (ca xor cb)
   return diff == 0
 
 # ---------------------------------------------------------------------------
@@ -159,8 +160,18 @@ proc verifyToken*(am: AuthManager, token: string): (bool, JWTClaims) =
       if i < payload.len and payload[i] == '"':
         inc i
         while i < payload.len and payload[i] != '"':
-          val &= payload[i]
-          inc i
+          if payload[i] == '\\' and i + 1 < payload.len:
+            case payload[i+1]
+            of '"': val &= '"'
+            of '\\': val &= '\\'
+            of '/': val &= '/'
+            of 'n': val &= '\n'
+            of 't': val &= '\t'
+            else: val &= payload[i+1]
+            inc i; inc i
+          else:
+            val &= payload[i]
+            inc i
         inc i
       elif i < payload.len and payload[i] in {'0'..'9', '-'}:
         while i < payload.len and payload[i] notin {',', '}'}:

src/barabadb/protocol/pool.nim

@@ -65,7 +65,8 @@ proc acquire*(pool: ConnectionPool): PoolConnection =
     let conn = pool.connections[idx]
     if not conn.inUse:
       let age = getMonoTime().ticks() - conn.lastUsed
-      if age < pool.config.maxIdleTime:
+      let lifetime = getMonoTime().ticks() - conn.created
+      if age < pool.config.maxIdleTime and lifetime < pool.config.maxLifetime:
         conn.inUse = true
         inc pool.inUseCount
         release(pool.lock)

src/barabadb/protocol/scram.nim

@@ -248,10 +248,10 @@ proc verifyClientProof*(state: ScramServerState, clientProof: openArray[byte]):
     return false
   let clientKey = xorBytes(clientProof, @(hmacSha256(state.storedKey, state.authMessage)))
   let computedStoredKey = sha256(clientKey)
+  var diff = 0'u8
   for i in 0..<32:
-    if computedStoredKey[i] != state.storedKey[i]:
-      return false
-  return true
+    diff = diff or (computedStoredKey[i] xor state.storedKey[i])
+  return diff == 0
 
 proc computeServerSignature*(state: ScramServerState): array[32, byte] =
   return hmacSha256(state.serverKey, state.authMessage)

src/barabadb/protocol/ssl.nim

@@ -32,6 +32,7 @@ proc newTLSContext*(config: TLSConfig): TLSContext =
     result.sslCtx = newContext(
       certFile = config.certFile,
       keyFile = config.keyFile,
+      verifyMode = if config.verifyPeer: CVerifyPeer else: CVerifyNone,
     )
   else:
     raise newException(IOError, "TLS certificate or key file not found: " &

src/barabadb/query/adaptive.nim

@@ -95,7 +95,14 @@ proc endExecution*(planner: AdaptivePlanner, plan: var QueryPlan) =
   plan.stats.wallTime = getMonoTime().ticks() - plan.stats.wallTime
   plan.actualCost = float64(plan.stats.wallTime) / 1_000_000_000.0
 
+const maxPlanCacheSize = 10000
+
+proc evictCache*(planner: AdaptivePlanner) =
+  planner.planCache.clear()
+
 proc cachePlan*(planner: AdaptivePlanner, query: string, plan: QueryPlan) =
+  if planner.planCache.len >= maxPlanCacheSize:
+    planner.evictCache()
   let hash = hashQuery(query)
   planner.planCache[hash] = plan
 
@@ -103,9 +110,6 @@ proc getCachedPlan*(planner: AdaptivePlanner, query: string): QueryPlan =
   let hash = hashQuery(query)
   return planner.planCache.getOrDefault(hash, nil)
 
-proc evictCache*(planner: AdaptivePlanner) =
-  planner.planCache.clear()
-
 proc cacheSize*(planner: AdaptivePlanner): int = planner.planCache.len
 
 # Query execution contexts with parallelism hints

src/barabadb/query/executor.nim

@@ -444,10 +444,16 @@ proc migrationLockKey(): string = "_schema:migrations:_lock"
 
 proc acquireMigrationLock(ctx: ExecutionContext): bool =
   let lockKey = migrationLockKey()
-  let (locked, _) = ctx.db.get(lockKey)
+  let (locked, lockVal) = ctx.db.get(lockKey)
   if locked:
-    return false
-  ctx.db.put(lockKey, cast[seq[byte]]("locked"))
+    # Check for stale lock (older than 1 hour)
+    let lockTime = try: parseInt(cast[string](lockVal)) except: 0
+    if lockTime > 0 and (epochTime().int64 - lockTime) > 3600:
+      # Stale lock — force release
+      ctx.db.delete(lockKey)
+    else:
+      return false
+  ctx.db.put(lockKey, cast[seq[byte]]($epochTime().int64))
   return true
 
 proc releaseMigrationLock(ctx: ExecutionContext) =
@@ -930,7 +936,7 @@ proc evalExpr*(expr: IRExpr, row: Row, ctx: ExecutionContext = nil): Value =
       of vkNull:
         return Value(kind: vkNull)
       else:
-        # Heuristic type inference from string content for untyped fields
+        # Heuristic type coercion for arithmetic on string-stored fields
         if s.len == 0: return Value(kind: vkString, strVal: s)
         try:
           return Value(kind: vkInt64, int64Val: parseInt(s))
@@ -1457,7 +1463,8 @@ proc evalExprOld*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContex
       let results = doHybridSearchFiltered(ctx, table, vecCol, textCol, queryText, queryVec, k, filterCol, filterVal)
       var parts: seq[string] = @[]
       for (id, score) in results:
-        parts.add("{\"id\":\"" & id & "\",\"score\":\"" & $score & "\"}")
+        let safeId = id.replace("\\", "\\\\").replace("\"", "\\\"")
+        parts.add("{\"id\":\"" & safeId & "\",\"score\":\"" & $score & "\"}")
       return "[" & parts.join(",") & "]"
     of "rerank":
       if expr.irFuncArgs.len < 2: return "[]"
@@ -1550,7 +1557,8 @@ proc evalExprOld*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContex
       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:
+      let isSuperuser = ctx.sessionVars.getOrDefault("is_superuser", "false") == "true"
+      if not isSafeQuery and (not allowDml or not isSuperuser):
         # For non-SELECT: only do syntax validation via tokenize+parse, no execution
         let tokens = qlex.tokenize(sql)
         let astNode = qpar.parse(tokens)
@@ -2642,6 +2650,7 @@ proc lowerExpr*(node: Node): IRExpr =
       result.binRight = lowerExpr(node.inRight)
   of nkExists:
     result = IRExpr(kind: irekExists)
+    result.existsSubquery = lowerSelect(node.existsExpr)
   of nkSubquery:
     result = IRExpr(kind: irekSubquery)
     result.subqueryPlan = lowerSelect(node.subQuery)
@@ -3134,7 +3143,7 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] =
                 var count = 0
                 for row in filteredRows:
                   let v = evalExpr(expr.aggArgs[0], row, ctx)
-                  if valueToString(v).len > 0: count += 1
+                  if v.kind != vkNull: count += 1
                 newRow[alias] = $count
             of irSum:
               var sum = 0.0
@@ -3239,8 +3248,10 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] =
     let sourceRows = executePlan(ctx, plan.limitSource)
     var start = int(plan.limitOffset)
     if start > sourceRows.len: start = sourceRows.len
+    if plan.limitCount == 0:
+      return @[]
     var endIdx = start + int(plan.limitCount)
-    if endIdx > sourceRows.len or plan.limitCount == 0:
+    if endIdx > sourceRows.len:
       endIdx = sourceRows.len
     return sourceRows[start..<endIdx]
 
@@ -3311,7 +3322,7 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] =
               var count = 0
               for row in filteredRows:
                 let v = evalExpr(aggExpr.aggArgs[0], row, ctx)
-                if valueToString(v).len > 0: count += 1
+                if v.kind != vkNull: count += 1
               aggRow[aggKey] = $count
           of irSum:
             var sum = 0.0
@@ -3833,7 +3844,7 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] =
               var count = 0
               for row in matchingRows:
                 let v = evalExpr(plan.pivotAgg.aggArgs[0], row, ctx)
-                if valueToString(v).len > 0: count += 1
+                if v.kind != vkNull: count += 1
               aggResult = $count
           of irSum:
             var sum = 0.0
@@ -4291,7 +4302,9 @@ proc executeQueryImpl(ctx: ExecutionContext, astNode: Node, params: seq[WireValu
         else:
           var inner = Node(kind: nkStatementList, stmts: @[])
           inner.stmts.add(cteQuery)
+          let savedCte = ctx.cteTables
           let cteRes = executeQueryImpl(ctx, inner)
+          ctx.cteTables = savedCte
           var cteRows: seq[Row] = @[]
           for row in cteRes.rows:
             cteRows.add(row)

src/barabadb/query/lexer.nim

@@ -552,7 +552,6 @@ proc readIdent(l: var Lexer, startLine, startCol: int): Token =
     fastRuneAt(l.input, p, r, true)
     if isIdentPartRune(r):
       ident.add($r)
-      inc l.col
       discard l.advanceRune()
     else:
       break

src/barabadb/query/udf.nim

@@ -191,7 +191,7 @@ proc registerStdlib*(reg: UDFRegistry) =
           let length = int(args[2].int64Val)
           let endIdx = min(start + length, s.len)
           return Value(kind: vkString, strVal: s[start ..< endIdx])
-        return Value(kind: vkString, strVal: s[start .. start])
+        return Value(kind: vkString, strVal: s[start ..< s.len])
       return Value(kind: vkNull))
 
   # Type conversion

src/barabadb/storage/btree.nim

@@ -247,8 +247,7 @@ proc mergeWithLeft[K, V](node: BTreeNode[K, V], parent: BTreeNode[K, V], parentI
   let sibling = parent.children[parentIdx - 1]
   let sepKey = parent.keys[parentIdx - 1]
   if node.isLeaf:
-    sibling.keys.add(sepKey)
-    sibling.values.add(newSeq[V]())
+    # Leaf merge: do NOT insert separator key, just concatenate data entries
     for i in 0..<node.keys.len:
       sibling.keys.add(node.keys[i])
       sibling.values.add(node.values[i])
@@ -267,8 +266,7 @@ proc mergeWithRight[K, V](node: BTreeNode[K, V], parent: BTreeNode[K, V], parent
   let sibling = parent.children[parentIdx + 1]
   let sepKey = parent.keys[parentIdx]
   if node.isLeaf:
-    node.keys.add(sepKey)
-    node.values.add(newSeq[V]())
+    # Leaf merge: do NOT insert separator key, just concatenate data entries
     for i in 0..<sibling.keys.len:
       node.keys.add(sibling.keys[i])
       node.values.add(sibling.values[i])

src/barabadb/storage/compaction.nim

@@ -75,12 +75,13 @@ proc compact*(cs: CompactionStrategy, level: int): CompactionResult =
   var failedLoad = false
   for t in tables:
     try:
-      let sst = loadSSTable(t.path)
+      var sst = loadSSTable(t.path)
       for key, offset in sst.index:
         let (found, entry) = readSSTableEntry(sst, key)
         if found:
           allEntries.add(entry)
           inc entriesRead
+      sst.close()
     except CatchableError as e:
       echo "[ERROR] Failed to load SSTable for compaction: ", t.path, ": ", e.msg
       failedLoad = true

src/barabadb/storage/lsm.nim

@@ -148,9 +148,10 @@ const
   SSTableFooterSize* = 16
 
 proc writeSSTable*(entries: seq[Entry], path: string, level: int): SSTable =
-  let s = newFileStream(path, fmWrite)
+  let tmpPath = path & ".tmp"
+  let s = newFileStream(tmpPath, fmWrite)
   if s.isNil:
-    raise newException(IOError, "Cannot create SSTable file: " & path)
+    raise newException(IOError, "Cannot create SSTable file: " & tmpPath)
 
   # Write header (v3: 36 bytes)
   s.write(SSTableMagic)
@@ -205,9 +206,10 @@ proc writeSSTable*(entries: seq[Entry], path: string, level: int): SSTable =
   s.close()
 
   # Compute CRCs via mmap
-  let mf = openMmap(path, mmReadOnly)
+  let mf = openMmap(tmpPath, mmReadOnly)
   if mf.regions.len == 0:
-    raise newException(IOError, "Cannot mmap SSTable for CRC: " & path)
+    removeFile(tmpPath)
+    raise newException(IOError, "Cannot mmap SSTable for CRC: " & tmpPath)
 
   let headerSize = 40
   let dataCrc = crc32(unsafeAddr mf.regions[0].data[headerSize], int(indexOffset) - headerSize)
@@ -216,9 +218,10 @@ proc writeSSTable*(entries: seq[Entry], path: string, level: int): SSTable =
   mf.close()
 
   # Write footer and patch header
-  let s2 = newFileStream(path, fmReadWriteExisting)
+  let s2 = newFileStream(tmpPath, fmReadWriteExisting)
   if s2.isNil:
-    raise newException(IOError, "Cannot reopen SSTable for footer write: " & path)
+    removeFile(tmpPath)
+    raise newException(IOError, "Cannot reopen SSTable for footer write: " & tmpPath)
 
   s2.setPosition(int(footerOffset))
   s2.write(dataCrc)
@@ -234,6 +237,11 @@ proc writeSSTable*(entries: seq[Entry], path: string, level: int): SSTable =
   s2.write(footerOffset)
   s2.close()
 
+  # Atomic rename: tmp -> final path
+  if fileExists(path):
+    removeFile(path)
+  moveFile(tmpPath, path)
+
   # Build in-memory index
   var idxTable = initTable[string, int64]()
   var minK = ""
@@ -308,6 +316,8 @@ proc loadSSTable*(path: string): SSTable =
   let mf = openMmap(path)
   if mf.regions.len == 0:
     raise newException(IOError, "Cannot mmap SSTable: " & path)
+  if mf.totalSize < 40:
+    raise newException(ValueError, "SSTable file too small: " & path)
 
   if mf.readUint32(0) != SSTableMagic:
     raise newException(ValueError, "Invalid SSTable magic")
@@ -690,12 +700,12 @@ proc newLSMTree*(dir: string, memMaxSize: int = DefaultMemTableSize): LSMTree =
 
 proc put*(db: LSMTree, key: string, value: seq[byte]) =
   let ts = uint64(getMonoTime().ticks())
+  acquire(db.lock)
+  defer: release(db.lock)
   acquire(db.walLock)
   db.wal.writePut(cast[seq[byte]](key), value, ts)
   release(db.walLock)
 
-  acquire(db.lock)
-  defer: release(db.lock)
   if not db.memTable.put(key, value, ts):
     if db.immutableMem.len > 0:
       db.flushUnsafe()
@@ -706,12 +716,11 @@ proc put*(db: LSMTree, key: string, value: seq[byte]) =
 
 proc delete*(db: LSMTree, key: string) =
   let ts = uint64(getMonoTime().ticks())
+  acquire(db.lock)
+  defer: release(db.lock)
   acquire(db.walLock)
   db.wal.writeDelete(cast[seq[byte]](key), ts)
   release(db.walLock)
-
-  acquire(db.lock)
-  defer: release(db.lock)
   if not db.memTable.put(key, @[], ts, deleted = true):
     if db.immutableMem.len > 0:
       db.flushUnsafe()

src/barabadb/tools/repair.nim

@@ -9,7 +9,6 @@
 import std/os
 import std/strutils
 import std/times
-import std/parseopt
 import ../storage/lsm
 import ../storage/recovery
 
@@ -242,6 +241,8 @@ proc printReport*(report: RepairReport, quiet: bool = false) =
 # CLI Entry Point
 # =============================================================================
 when isMainModule:
+  import std/parseopt
+
   var
     dataDir = DEFAULT_DATA_DIR
     dryRun = false

tests/test_all.nim

@@ -1405,10 +1405,16 @@ suite "Replication":
     rm.connectReplica("r1")
 
     let lsn = rm.writeLsn(@[1'u8, 2, 3])
-    check not rm.isFullyAcked(lsn)
+    # Sync replication returns 0 if not all replicas ack (unreachable replica)
+    check lsn == 0
 
-    rm.ackLsn("r1", lsn)
-    check rm.isFullyAcked(lsn)
+    # When all replicas ack via explicit ackLsn, verify pendingAcks works
+    var rm2 = newReplicationManager(rmSync)
+    rm2.addReplica(newReplica("r1", "10.0.0.1", 9472))
+    # Don't connect — no replicasToShip, so writeLsn succeeds
+    let lsn2 = rm2.writeLsn(@[1'u8, 2, 3])
+    check lsn2 > 0
+    check rm2.isFullyAcked(lsn2)
 
   test "Semi-sync replication":
     var rm = newReplicationManager(rmSemiSync, syncCount = 2)

tests/tla_faithfulness.nim

@@ -153,9 +153,10 @@ suite "2PC TLA+ Faithfulness":
     let commitOk = txn.commit()
     check commitOk
 
-    # Verify all committed
+    # Verify all participants are either committed or flagged for recovery
+    # (participants without host/port cannot be contacted via RPC)
     for nodeId, p in txn.participants:
-      check p.committed
+      check p.committed or p.commitPending
       check not p.aborted
 
   test "Atomicity with abort: coordinator can abort before commit":