feat: harden storage, schema persistence, fair benches, fix wire crash

Core storage: hash MemTable, WAL group commit, L0 compaction rebuild,
reader-writer lock, and a global StorageGate so HTTP workers and TCP
share the LSM safely under multi-thread access.

Schema: durable CREATE/ALTER/DROP under _schema:tables:* with full LSM
restore on open. Executor types/values/schema split into query/exec/.

Wire protocol: switch default MM to ARC — ORC cycle collector segfaulted
after ~20 async INSERTs. Fair multi-tier benchmarks (SQLite/HTTP/wire/PG)
and honesty docs for mixed-tier comparisons.

src/barabadb/storage/compaction.nim

@@ -8,6 +8,7 @@ import ../storage/lsm
 const
   MaxLevel* = 7
   LevelMultiplier* = 10  # each level is 10x the previous
+  ## L0 uses file-count trigger (overlapping ranges); lower levels use size.
 
 type
   SSTableMeta* = object
@@ -29,20 +30,44 @@ type
     levels*: seq[seq[SSTableMeta]]
     dataDir*: string
     maxSizePerLevel*: seq[int]
+    l0FileLimit*: int
 
-proc newCompactionStrategy*(dataDir: string): CompactionStrategy =
+proc newCompactionStrategy*(dataDir: string, l0FileLimit: int = L0CompactionTrigger): CompactionStrategy =
   result = CompactionStrategy(
     levels: newSeq[seq[SSTableMeta]](MaxLevel),
     dataDir: dataDir,
     maxSizePerLevel: newSeq[int](MaxLevel),
+    l0FileLimit: l0FileLimit,
   )
   for i in 0..<MaxLevel:
     result.levels[i] = @[]
     result.maxSizePerLevel[i] = int(float64(1024 * 1024) * pow(float64(LevelMultiplier), float64(i)))  # 1MB, 10MB, 100MB...
 
+proc clear*(cs: CompactionStrategy) =
+  ## Drop all registered tables (used before rebuild-from-LSM).
+  for i in 0..<MaxLevel:
+    cs.levels[i].setLen(0)
+
 proc addTable*(cs: CompactionStrategy, meta: SSTableMeta) =
-  if meta.level < MaxLevel:
-    cs.levels[meta.level].add(meta)
+  let lvl = clamp(meta.level, 0, MaxLevel - 1)
+  cs.levels[lvl].add(meta)
+
+proc rebuildFromLSM*(cs: CompactionStrategy, db: LSMTree) =
+  ## Rebuild level layout from the live LSMTree catalog — single source of truth.
+  ## Avoids drift when flushes add SSTables the strategy never saw.
+  cs.clear()
+  cs.dataDir = db.dir
+  for sst in db.sstables:
+    let size = try: int(getFileSize(sst.path)) except: sst.entryCount * 64
+    cs.addTable(SSTableMeta(
+      path: sst.path,
+      level: sst.level,
+      minKey: sst.minKey,
+      maxKey: sst.maxKey,
+      entryCount: sst.entryCount,
+      sizeBytes: size,
+      createdAt: sst.id,  # stable ordering by id / creation sequence
+    ))
 
 proc totalSize*(cs: CompactionStrategy, level: int): int =
   result = 0
@@ -52,6 +77,9 @@ proc totalSize*(cs: CompactionStrategy, level: int): int =
 proc needsCompaction*(cs: CompactionStrategy, level: int): bool =
   if level >= MaxLevel - 1:
     return false
+  if level == 0:
+    # L0 files can overlap — count-based trigger (RocksDB-style)
+    return cs.levels[0].len >= cs.l0FileLimit
   return cs.totalSize(level) > cs.maxSizePerLevel[level]
 
 proc pickTablesForCompaction*(cs: CompactionStrategy, level: int): seq[SSTableMeta] =

src/barabadb/storage/gate.nim

@@ -0,0 +1,41 @@
+## Global storage gate — exclusive multi-thread entry to LSM / executor.
+##
+## Why: Hunos HTTP runs handlers on a worker-thread pool (`spawn` + internal
+## workers). The TCP server runs on the main async loop. Both share the same
+## `LSMTree` / `ExecutionContext` refs. Nim's default ORC memory manager is not
+## safe for concurrent refcount ops on the same objects from multiple OS threads.
+##
+## Holding this gate for the full duration of a query/compaction/DDL ensures
+## only one thread mutates or reads GC-managed storage state at a time.
+##
+## Ordering: always acquire StorageGate **before** any per-DB `LSMTree.lock`.
+## Call `initStorageGate()` once from main before accepting connections.
+import std/locks
+
+var
+  gGate: Lock
+  gInited*: bool
+
+proc initStorageGate*() =
+  ## Idempotent when called from a single thread at startup.
+  if not gInited:
+    initLock(gGate)
+    gInited = true
+
+proc acquireStorageGate*() {.inline.} =
+  ## Prefer calling initStorageGate() once at process start (main).
+  ## Lazy-init is allowed for unit tests (single-threaded).
+  if not gInited:
+    initStorageGate()
+  acquire(gGate)
+
+proc releaseStorageGate*() {.inline.} =
+  release(gGate)
+
+template withStorageGate*(body: untyped) =
+  ## Exclusive ownership of the storage engine for `body`.
+  acquireStorageGate()
+  try:
+    body
+  finally:
+    releaseStorageGate()

src/barabadb/storage/lsm.nim

@@ -13,6 +13,11 @@ import bloom
 import wal
 import mmap
 import crc32
+import rwlock
+
+# Re-export WAL durability knobs for callers of newLSMTree
+export wal
+export rwlock
 
 const
   SSTableMagic* = 0x53535442'u32  # "SSTB"
@@ -21,6 +26,8 @@ const
   DefaultBloomFpRate* = 0.01
   ManifestVersion* = 1
   ManifestFileName* = "MANIFEST"
+  ## Trigger L0 compaction when this many L0 SSTables exist.
+  L0CompactionTrigger* = 4
 
 type
   Entry* = object
@@ -29,9 +36,10 @@ type
     timestamp*: uint64
     deleted*: bool
 
+  ## Hash-table MemTable: O(1) put/get. Sorted only when flushing to SSTable.
   MemTable* = object
-    entries: seq[Entry]
-    size: int
+    map: Table[string, Entry]
+    size: int      ## approximate byte size of live entries
     maxSize: int
 
   SSTable* = object
@@ -56,55 +64,64 @@ type
     currentSeq: uint64
     nextSSTableId*: int
     manifestSequence*: int64
-    lock*: Lock
+    ## Reader-writer lock: concurrent gets; exclusive put/flush/compact.
+    ## `acquire(db.lock)` is exclusive (write) for backward compatibility.
+    lock*: RwLock
     walLock*: Lock
+    ## Set by flush when L0 file count hits L0CompactionTrigger (hint for compactors).
+    needsCompaction*: bool
+    ## When true, flushUnsafe skips WAL rewrite (recovery still holds the WAL file open).
+    recovering: bool
 
 proc newMemTable(maxSize: int = DefaultMemTableSize): MemTable =
-  MemTable(entries: @[], size: 0, maxSize: maxSize)
+  MemTable(map: initTable[string, Entry](), size: 0, maxSize: maxSize)
 
-proc len*(mt: MemTable): int = mt.entries.len
+proc len*(mt: MemTable): int = mt.map.len
+
+proc byteSize*(mt: MemTable): int = mt.size
 
 proc put*(mt: var MemTable, key: string, value: seq[byte], timestamp: uint64, deleted: bool = false): bool =
+  ## O(1) average-case insert/update. Returns false if the new key would exceed maxSize.
   let entrySize = key.len + value.len + 16
   if entrySize > mt.maxSize:
     return false
   let entry = Entry(key: key, value: value, timestamp: timestamp, deleted: deleted)
-  let pos = mt.entries.lowerBound(entry, proc(a, b: Entry): int = cmp(a.key, b.key))
-  if pos < mt.entries.len and mt.entries[pos].key == key:
-    let oldSize = mt.entries[pos].key.len + mt.entries[pos].value.len + 16
-    mt.entries[pos] = entry
+  if key in mt.map:
+    let old = mt.map[key]
+    # Only accept equal-or-newer timestamps (WAL recovery may replay older values)
+    if timestamp < old.timestamp:
+      return true
+    let oldSize = old.key.len + old.value.len + 16
+    mt.map[key] = entry
     mt.size += entrySize - oldSize
   else:
-    if mt.size + entrySize > mt.maxSize and mt.entries.len > 0:
+    if mt.size + entrySize > mt.maxSize and mt.map.len > 0:
       return false
-    mt.entries.insert(entry, pos)
+    mt.map[key] = entry
     mt.size += entrySize
   return true
 
 proc get*(mt: MemTable, key: string): (bool, Entry) =
-  if mt.entries.len == 0:
-    return (false, Entry())
-  var lo = 0
-  var hi = mt.entries.len - 1
-  while lo <= hi:
-    let mid = (lo + hi) div 2
-    let c = cmp(mt.entries[mid].key, key)
-    if c == 0:
-      return (true, mt.entries[mid])
-    elif c < 0:
-      lo = mid + 1
-    else:
-      hi = mid - 1
+  if key in mt.map:
+    return (true, mt.map[key])
   return (false, Entry())
 
+proc sortedEntries*(mt: MemTable): seq[Entry] =
+  ## Materialize entries sorted by key — used for SSTable flush and ordered scans.
+  result = newSeqOfCap[Entry](mt.map.len)
+  for _, entry in mt.map:
+    result.add(entry)
+  result.sort(proc(a, b: Entry): int = cmp(a.key, b.key))
+
 proc scan*(mt: MemTable, startKey, endKey: string): seq[Entry] =
   result = @[]
-  for entry in mt.entries:
-    if entry.key >= startKey and entry.key <= endKey:
+  for key, entry in mt.map:
+    if key >= startKey and key <= endKey:
       result.add(entry)
+  result.sort(proc(a, b: Entry): int = cmp(a.key, b.key))
 
 proc clear*(mt: var MemTable) =
-  mt.entries.setLen(0)
+  mt.map.clear()
   mt.size = 0
 
 # ----------------------------------------------------------------------
@@ -600,8 +617,15 @@ proc checkStorageConsistency*(db: LSMTree): seq[string] =
 # ----------------------------------------------------------------------
 
 proc flushUnsafe(db: LSMTree) {.gcsafe.}
+proc countL0*(db: LSMTree): int
 
-proc newLSMTree*(dir: string, memMaxSize: int = DefaultMemTableSize): LSMTree =
+proc newLSMTree*(
+    dir: string,
+    memMaxSize: int = DefaultMemTableSize,
+    walSyncMode: WalSyncMode = wsmGroup,
+    walGroupEvery: int = DefaultWalGroupEvery,
+    walGroupIntervalMs: int = 0,
+): LSMTree =
   createDir(dir)
   createDir(dir / "sstables")
 
@@ -641,21 +665,29 @@ proc newLSMTree*(dir: string, memMaxSize: int = DefaultMemTableSize): LSMTree =
       echo "[INFO] Loaded ", sstables.len, " SSTable(s) from directory scan"
 
   new(result)
-  initLock(result.lock)
+  initRwLock(result.lock)
   initLock(result.walLock)
   result.dir = dir
   result.memTable = newMemTable(memMaxSize)
   result.immutableMem = newMemTable(0)
   result.sstables = sstables
-  result.wal = newWriteAheadLog(dir / "wal")
+  result.wal = newWriteAheadLog(
+    dir / "wal",
+    syncMode = walSyncMode,
+    groupEvery = walGroupEvery,
+    groupIntervalMs = walGroupIntervalMs,
+  )
   result.memMaxSize = memMaxSize
   result.currentSeq = 0
   result.nextSSTableId = nextId
   result.manifestSequence = manifestSeq
+  result.recovering = false
+  result.needsCompaction = result.countL0() >= L0CompactionTrigger
 
   # WAL crash recovery — replay unflushed entries into memTable
   let walPath = dir / "wal" / "wal.log"
   if fileExists(walPath):
+    result.recovering = true
     var stream: FileStream = nil
     try:
       stream = newFileStream(walPath, fmRead)
@@ -697,11 +729,30 @@ proc newLSMTree*(dir: string, memMaxSize: int = DefaultMemTableSize): LSMTree =
     finally:
       if stream != nil:
         stream.close()
+      result.recovering = false
+      # After recovery, shrink WAL to live unflushed state only
+      acquire(result.walLock)
+      try:
+        var liveKeys: seq[string] = @[]
+        var liveVals: seq[seq[byte]] = @[]
+        var liveTs: seq[uint64] = @[]
+        var liveDel: seq[bool] = @[]
+        for e in result.immutableMem.sortedEntries():
+          liveKeys.add(e.key); liveVals.add(e.value); liveTs.add(e.timestamp); liveDel.add(e.deleted)
+        for e in result.memTable.sortedEntries():
+          liveKeys.add(e.key); liveVals.add(e.value); liveTs.add(e.timestamp); liveDel.add(e.deleted)
+        if liveKeys.len == 0:
+          result.wal.truncate()
+        else:
+          result.wal.rewriteLive(liveKeys, liveVals, liveTs, liveDel)
+      finally:
+        release(result.walLock)
 
 proc put*(db: LSMTree, key: string, value: seq[byte]) =
   let ts = uint64(getMonoTime().ticks())
-  acquire(db.lock)
-  defer: release(db.lock)
+  acquireWrite(db.lock)
+  defer: releaseWrite(db.lock)
+  # WAL then memtable under the same exclusive lock → crash recovery sees a total order
   acquire(db.walLock)
   db.wal.writePut(cast[seq[byte]](key), value, ts)
   release(db.walLock)
@@ -716,8 +767,8 @@ 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)
+  acquireWrite(db.lock)
+  defer: releaseWrite(db.lock)
   acquire(db.walLock)
   db.wal.writeDelete(cast[seq[byte]](key), ts)
   release(db.walLock)
@@ -732,8 +783,8 @@ proc delete*(db: LSMTree, key: string) =
 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)
+  acquireWrite(db.lock)
+  defer: releaseWrite(db.lock)
   if not db.memTable.put(key, value, ts, deleted):
     if db.immutableMem.len > 0:
       db.flushUnsafe()
@@ -745,18 +796,26 @@ proc putUnsafe*(db: LSMTree, key: string, value: seq[byte], deleted: bool = fals
 proc deleteUnsafe*(db: LSMTree, key: string) =
   putUnsafe(db, key, @[], deleted = true)
 
+proc copyBytes(s: seq[byte]): seq[byte] =
+  ## Deep copy so callers on other threads never share ORC-managed seq buffers.
+  result = newSeq[byte](s.len)
+  if s.len > 0:
+    copyMem(addr result[0], unsafeAddr s[0], s.len)
+
 proc getUnsafe(db: LSMTree, key: string): (bool, seq[byte]) =
+  ## Caller must hold at least a read lock.
+  ## Returned values are deep-copied for multi-thread ORC safety (HTTP + TCP share LSM).
   let (found, entry) = db.memTable.get(key)
   if found:
     if entry.deleted:
       return (false, @[])
-    return (true, entry.value)
+    return (true, copyBytes(entry.value))
 
   let (found2, entry2) = db.immutableMem.get(key)
   if found2:
     if entry2.deleted:
       return (false, @[])
-    return (true, entry2.value)
+    return (true, copyBytes(entry2.value))
 
   # Search SSTables from newest to oldest
   for i in countdown(db.sstables.high, db.sstables.low):
@@ -769,21 +828,40 @@ proc getUnsafe(db: LSMTree, key: string): (bool, seq[byte]) =
     if found3:
       if entry3.deleted:
         return (false, @[])
-      return (true, entry3.value)
+      return (true, copyBytes(entry3.value))
 
   return (false, @[])
 
 proc get*(db: LSMTree, key: string): (bool, seq[byte]) =
-  acquire(db.lock)
-  defer: release(db.lock)
+  ## Thread-safe lookup.
+  ## Default: exclusive lock — required for Nim ORC when TCP + HTTP threads share the DB.
+  ## Compile with `-d:baraConcurrentReads` for shared read locks (needs multi-thread-safe MM
+  ## such as a future atomicArc build; unsafe with default ORC across OS threads).
+  when defined(baraConcurrentReads):
+    acquireRead(db.lock)
+    defer: releaseRead(db.lock)
+  else:
+    acquireWrite(db.lock)
+    defer: releaseWrite(db.lock)
   return getUnsafe(db, key)
 
 proc contains*(db: LSMTree, key: string): bool =
-  acquire(db.lock)
-  defer: release(db.lock)
+  when defined(baraConcurrentReads):
+    acquireRead(db.lock)
+    defer: releaseRead(db.lock)
+  else:
+    acquireWrite(db.lock)
+    defer: releaseWrite(db.lock)
   let (found, _) = getUnsafe(db, key)
   return found
 
+proc countL0*(db: LSMTree): int =
+  ## Number of level-0 SSTables (newest, uncompacted).
+  result = 0
+  for sst in db.sstables:
+    if sst.level == 0:
+      inc result
+
 proc flushUnsafe(db: LSMTree) =
   if db.immutableMem.len == 0 and db.memTable.len == 0:
     return
@@ -802,7 +880,8 @@ proc flushUnsafe(db: LSMTree) =
   let path = db.dir / "sstables" / ($db.nextSSTableId & ".sst")
   inc db.nextSSTableId
 
-  var sst = writeSSTable(toFlush.entries, path, level = 0)
+  # Sort once at flush time (O(n log n)) — put/get stay O(1)
+  var sst = writeSSTable(toFlush.sortedEntries(), path, level = 0)
   sst.id = db.nextSSTableId - 1
   db.sstables.add(sst)
   # SSTables are kept in insertion order (newest last) so getUnsafe can search newest-first
@@ -814,22 +893,43 @@ proc flushUnsafe(db: LSMTree) =
   except CatchableError as e:
     echo "[WARN] Failed to write MANIFEST: ", e.msg
 
-  acquire(db.walLock)
-  db.wal.writeCommit(uint64(getMonoTime().ticks()))
-  db.wal.maybeRotate()
-  db.wal.sync()
-  release(db.walLock)
+  # Rewrite WAL to contain only still-unflushed memtable entries.
+  # Skip during recovery — the WAL file is still open for reading.
+  if not db.recovering:
+    acquire(db.walLock)
+    var liveKeys: seq[string] = @[]
+    var liveVals: seq[seq[byte]] = @[]
+    var liveTs: seq[uint64] = @[]
+    var liveDel: seq[bool] = @[]
+    for e in db.immutableMem.sortedEntries():
+      liveKeys.add(e.key)
+      liveVals.add(e.value)
+      liveTs.add(e.timestamp)
+      liveDel.add(e.deleted)
+    for e in db.memTable.sortedEntries():
+      liveKeys.add(e.key)
+      liveVals.add(e.value)
+      liveTs.add(e.timestamp)
+      liveDel.add(e.deleted)
+    if liveKeys.len == 0:
+      db.wal.truncate()
+    else:
+      db.wal.rewriteLive(liveKeys, liveVals, liveTs, liveDel)
+    release(db.walLock)
+
+  if db.countL0() >= L0CompactionTrigger:
+    db.needsCompaction = true
 
 proc flush*(db: LSMTree) =
-  acquire(db.lock)
-  defer: release(db.lock)
+  acquireWrite(db.lock)
+  defer: releaseWrite(db.lock)
   flushUnsafe(db)
 
 proc checkpoint*(db: LSMTree) =
   ## Create a consistent checkpoint: freeze memtable, flush to SSTable,
   ## rotate WAL, and write MANIFEST. This provides a clean boundary
   ## for online backup without stopping the server.
-  acquire(db.lock)
+  acquireWrite(db.lock)
 
   # Flush any pending immutable memtable first
   if db.immutableMem.len > 0:
@@ -850,10 +950,10 @@ proc checkpoint*(db: LSMTree) =
   db.wal.sync()
   release(db.walLock)
 
-  release(db.lock)
+  releaseWrite(db.lock)
 
 proc close*(db: LSMTree) =
-  acquire(db.lock)
+  acquireWrite(db.lock)
   try:
     # Flush both memtables to avoid data loss
     while db.immutableMem.len > 0:
@@ -863,61 +963,109 @@ proc close*(db: LSMTree) =
       sst.close()
     db.wal.close()
   finally:
-    release(db.lock)
+    releaseWrite(db.lock)
+
+template withDataLock(db: LSMTree, body: untyped) =
+  ## Shared or exclusive depending on baraConcurrentReads (see get*).
+  when defined(baraConcurrentReads):
+    acquireRead(db.lock)
+    try:
+      body
+    finally:
+      releaseRead(db.lock)
+  else:
+    acquireWrite(db.lock)
+    try:
+      body
+    finally:
+      releaseWrite(db.lock)
 
 proc memTableSize*(db: LSMTree): int =
-  acquire(db.lock)
-  defer: release(db.lock)
-  return db.memTable.len
+  withDataLock(db):
+    return db.memTable.len
 
 proc sstableCount*(db: LSMTree): int =
-  acquire(db.lock)
-  defer: release(db.lock)
-  return db.sstables.len
+  withDataLock(db):
+    return db.sstables.len
 
 proc dir*(db: LSMTree): string =
-  acquire(db.lock)
-  defer: release(db.lock)
-  return db.dir
+  withDataLock(db):
+    return db.dir
 
 proc scanMemTable*(db: LSMTree): seq[Entry] =
-  acquire(db.lock)
-  defer: release(db.lock)
-  ## Return all entries from memory (memTable + immutableMem)
-  result = @[]
-  for e in db.memTable.entries:
-    result.add(e)
-  for e in db.immutableMem.entries:
-    result.add(e)
+  ## Return all entries from memory (memTable + immutableMem), sorted by key.
+  ## Immutable wins over active memtable only when timestamps are newer (same key rare).
+  withDataLock(db):
+    var merged = initTable[string, Entry]()
+    for e in db.immutableMem.sortedEntries():
+      merged[e.key] = e
+    for e in db.memTable.sortedEntries():
+      if e.key notin merged or e.timestamp >= merged[e.key].timestamp:
+        merged[e.key] = e
+    result = newSeqOfCap[Entry](merged.len)
+    for _, e in merged:
+      result.add(e)
+    result.sort(proc(a, b: Entry): int = cmp(a.key, b.key))
+
+proc scanRange*(db: LSMTree, startKey, endKey: string): seq[(string, seq[byte])] =
+  ## Inclusive key range scan over memtables + SSTables (newest wins).
+  withDataLock(db):
+    var best = initTable[string, Entry]()
+
+    for e in db.memTable.scan(startKey, endKey):
+      best[e.key] = e
+    for e in db.immutableMem.scan(startKey, endKey):
+      if e.key notin best or e.timestamp > best[e.key].timestamp:
+        best[e.key] = e
+
+    for i in countdown(db.sstables.high, db.sstables.low):
+      let sst = db.sstables[i]
+      if sst.maxKey < startKey or sst.minKey > endKey:
+        continue
+      for key, offset in sst.index:
+        if key < startKey or key > endKey:
+          continue
+        if key in best:
+          continue
+        let (found, entry) = readSSTableEntry(sst, key)
+        if found:
+          best[key] = entry
+
+    var keys = newSeqOfCap[string](best.len)
+    for k in best.keys:
+      keys.add(k)
+    keys.sort(cmp)
+    for k in keys:
+      let e = best[k]
+      if not e.deleted:
+        result.add((e.key, e.value))
 
 proc scanAll*(db: LSMTree): seq[(string, seq[byte])] =
   ## Scan all active (non-deleted) entries from memory and SSTables.
   ## Used for shard data migration.
-  acquire(db.lock)
-  defer: release(db.lock)
+  withDataLock(db):
+    var seen = initTable[string, bool]()
 
-  var seen = initTable[string, bool]()
+    # Scan memtable first (most recent)
+    for e in db.memTable.sortedEntries():
+      if e.key notin seen:
+        seen[e.key] = true
+        if not e.deleted:
+          result.add((e.key, e.value))
 
-  # Scan memtable first (most recent)
-  for e in db.memTable.entries:
-    if e.key notin seen:
-      seen[e.key] = true
-      if not e.deleted:
-        result.add((e.key, e.value))
+    # Scan immutable memtable
+    for e in db.immutableMem.sortedEntries():
+      if e.key notin seen:
+        seen[e.key] = true
+        if not e.deleted:
+          result.add((e.key, e.value))
 
-  # Scan immutable memtable
-  for e in db.immutableMem.entries:
-    if e.key notin seen:
-      seen[e.key] = true
-      if not e.deleted:
-        result.add((e.key, e.value))
-
-  # Scan SSTables from newest to oldest
-  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
-        let (found, entry) = readSSTableEntry(sst, key)
-        if found and not entry.deleted:
-          result.add((entry.key, entry.value))
+    # Scan SSTables from newest to oldest
+    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
+          let (found, entry) = readSSTableEntry(sst, key)
+          if found and not entry.deleted:
+            result.add((entry.key, entry.value))

src/barabadb/storage/rwlock.nim

@@ -0,0 +1,80 @@
+## Simple reader-writer lock for LSM concurrent reads.
+## Multiple readers OR one writer. Writers are exclusive.
+## `acquire` / `release` are write-side (backward compatible with Lock-style usage).
+import std/locks
+
+type
+  RwLock* = object
+    mu: Lock
+    readers: int          ## active readers
+    writer: bool          ## writer holds exclusive access
+    waitingWriters: int   ## prefer writers to avoid reader starvation of compact/flush
+    canRead: Cond
+    canWrite: Cond
+
+proc initRwLock*(rw: var RwLock) =
+  initLock(rw.mu)
+  initCond(rw.canRead)
+  initCond(rw.canWrite)
+  rw.readers = 0
+  rw.writer = false
+  rw.waitingWriters = 0
+
+proc deinitRwLock*(rw: var RwLock) =
+  deinitCond(rw.canRead)
+  deinitCond(rw.canWrite)
+  deinitLock(rw.mu)
+
+proc acquireRead*(rw: var RwLock) =
+  ## Shared read lock. Blocks while a writer is active or waiting (writer preference).
+  acquire(rw.mu)
+  while rw.writer or rw.waitingWriters > 0:
+    wait(rw.canRead, rw.mu)
+  inc rw.readers
+  release(rw.mu)
+
+proc releaseRead*(rw: var RwLock) =
+  acquire(rw.mu)
+  dec rw.readers
+  if rw.readers == 0:
+    # Wake one waiting writer
+    signal(rw.canWrite)
+  release(rw.mu)
+
+proc acquireWrite*(rw: var RwLock) =
+  ## Exclusive write lock.
+  acquire(rw.mu)
+  inc rw.waitingWriters
+  while rw.writer or rw.readers > 0:
+    wait(rw.canWrite, rw.mu)
+  dec rw.waitingWriters
+  rw.writer = true
+  release(rw.mu)
+
+proc releaseWrite*(rw: var RwLock) =
+  acquire(rw.mu)
+  rw.writer = false
+  # Prefer draining writers, else open the gate for readers
+  if rw.waitingWriters > 0:
+    signal(rw.canWrite)
+  else:
+    broadcast(rw.canRead)
+  release(rw.mu)
+
+# Lock-compatible names: default exclusive (used by compaction, put, flush)
+proc acquire*(rw: var RwLock) {.inline.} = acquireWrite(rw)
+proc release*(rw: var RwLock) {.inline.} = releaseWrite(rw)
+
+template withReadLock*(rw: var RwLock, body: untyped) =
+  acquireRead(rw)
+  try:
+    body
+  finally:
+    releaseRead(rw)
+
+template withWriteLock*(rw: var RwLock, body: untyped) =
+  acquireWrite(rw)
+  try:
+    body
+  finally:
+    releaseWrite(rw)

src/barabadb/storage/wal.nim

@@ -4,12 +4,16 @@ import std/os
 import std/streams
 import std/strutils
 import std/posix
+import std/monotimes
+import std/times
 
 const
   WALMagic* = 0x42415241'u32  # "BARA"
   WALVersion* = 1'u32
   DefaultMaxWalSegmentSize* = 64 * 1024 * 1024  # 64MB
   WalArchiveDir* = "wal_archive"
+  ## Default group-commit batch size (entries between fsyncs).
+  DefaultWalGroupEvery* = 64
 
 type
   WalEntryKind* = enum
@@ -18,6 +22,15 @@ type
     wekCheckpoint = 3
     wekCommit = 4
 
+  ## Durability policy for WAL writes.
+  ## - wsmNone:  flush userspace buffer only; fsync on truncate/rewrite/close/explicit sync
+  ## - wsmGroup: group commit — fsync every N entries and/or every intervalMs (default)
+  ## - wsmEvery: fsync after every entry (strict, slow)
+  WalSyncMode* = enum
+    wsmNone = "none"
+    wsmGroup = "group"
+    wsmEvery = "every"
+
   WalEntry* = object
     kind*: WalEntryKind
     timestamp*: uint64
@@ -34,14 +47,28 @@ type
     path: string
     stream: FileStream
     entryCount: uint64
-    syncOnWrite: bool
+    syncMode*: WalSyncMode
+    groupEvery*: int          ## entries between fsyncs when mode=group
+    groupIntervalMs*: int     ## time-based fsync when mode=group (0 = off)
+    unsyncedEntries: int      ## entries written since last fsync
+    lastSync: MonoTime
     maxSegmentSize: int64
     currentSequence: int64
+    ## Counters for observability / benchmarks
+    fsyncCount*: uint64
+    bytesSinceSync: int
 
 proc readEntries*(walPath: string, untilTimestamp: uint64 = 0): seq[WalEntry]
 proc listWalArchive*(dir: string): seq[WalSegment]
 proc maybeRotate*(wal: var WriteAheadLog)
 
+proc parseWalSyncMode*(s: string): WalSyncMode =
+  case s.toLowerAscii()
+  of "none", "async", "off", "false", "0": wsmNone
+  of "every", "sync", "full", "true", "1": wsmEvery
+  of "group", "batch", "": wsmGroup
+  else: wsmGroup
+
 proc parseWalSequence*(filename: string): int64 =
   ## Extract sequence from "wal.000042.log"
   try:
@@ -74,6 +101,12 @@ proc nextWalSequence*(dir: string): int64 =
     return 1
   return segments[^1].sequence + 1
 
+proc fsyncPath(path: string) =
+  let fd = posix.open(cstring(path), O_RDWR)
+  if fd != -1:
+    discard posix.fsync(fd)
+    discard posix.close(fd)
+
 proc rotate*(wal: var WriteAheadLog) =
   ## Close current WAL and archive it, then start a new one.
   if wal.stream != nil:
@@ -96,7 +129,12 @@ proc rotate*(wal: var WriteAheadLog) =
   wal.stream.write(WALMagic)
   wal.stream.write(WALVersion)
   wal.stream.flush()
+  fsyncPath(wal.path)
   wal.entryCount = 0
+  wal.unsyncedEntries = 0
+  wal.bytesSinceSync = 0
+  wal.lastSync = getMonoTime()
+  inc wal.fsyncCount
 
 proc maybeRotate*(wal: var WriteAheadLog) =
   ## Rotate if current WAL exceeds max segment size.
@@ -106,7 +144,16 @@ proc maybeRotate*(wal: var WriteAheadLog) =
   if currentSize >= wal.maxSegmentSize:
     wal.rotate()
 
-proc newWriteAheadLog*(dir: string, syncOnWrite: bool = true): WriteAheadLog =
+proc newWriteAheadLog*(
+    dir: string,
+    syncMode: WalSyncMode = wsmGroup,
+    groupEvery: int = DefaultWalGroupEvery,
+    groupIntervalMs: int = 0,
+    syncOnWrite: bool = false,
+): WriteAheadLog =
+  ## Create a WAL.
+  ## - syncMode controls durability (see WalSyncMode).
+  ## - syncOnWrite=true is legacy and forces wsmEvery.
   createDir(dir)
   let path = dir / "wal.log"
   let exists = fileExists(path)
@@ -125,18 +172,62 @@ proc newWriteAheadLog*(dir: string, syncOnWrite: bool = true): WriteAheadLog =
     for e in readEntries(path):
       inc count
 
+  let mode = if syncOnWrite: wsmEvery else: syncMode
+  let ge = if groupEvery <= 0: DefaultWalGroupEvery else: groupEvery
   let seqNum = nextWalSequence(dir)
   WriteAheadLog(
     dir: dir,
     path: path,
     stream: stream,
     entryCount: count,
-    syncOnWrite: syncOnWrite,
+    syncMode: mode,
+    groupEvery: ge,
+    groupIntervalMs: groupIntervalMs,
+    unsyncedEntries: 0,
+    lastSync: getMonoTime(),
     maxSegmentSize: DefaultMaxWalSegmentSize,
     currentSequence: seqNum,
+    fsyncCount: 0,
+    bytesSinceSync: 0,
   )
 
+proc setSyncMode*(wal: var WriteAheadLog, mode: WalSyncMode) =
+  wal.syncMode = mode
+
+proc setGroupEvery*(wal: var WriteAheadLog, n: int) =
+  wal.groupEvery = if n <= 0: DefaultWalGroupEvery else: n
+
+proc setGroupIntervalMs*(wal: var WriteAheadLog, ms: int) =
+  wal.groupIntervalMs = max(0, ms)
+
+proc markSynced(wal: var WriteAheadLog) =
+  wal.unsyncedEntries = 0
+  wal.bytesSinceSync = 0
+  wal.lastSync = getMonoTime()
+  inc wal.fsyncCount
+
+proc maybeGroupSync(wal: var WriteAheadLog, entryBytes: int) =
+  ## Apply durability policy after a buffered write.
+  case wal.syncMode
+  of wsmNone:
+    discard
+  of wsmEvery:
+    fsyncPath(wal.path)
+    wal.markSynced()
+  of wsmGroup:
+    inc wal.unsyncedEntries
+    wal.bytesSinceSync += entryBytes
+    var due = wal.unsyncedEntries >= wal.groupEvery
+    if not due and wal.groupIntervalMs > 0:
+      let elapsedMs = (getMonoTime() - wal.lastSync).inMilliseconds
+      if elapsedMs >= wal.groupIntervalMs:
+        due = true
+    if due:
+      fsyncPath(wal.path)
+      wal.markSynced()
+
 proc writeEntry*(wal: var WriteAheadLog, entry: WalEntry) =
+  let entryBytes = 1 + 8 + 4 + entry.key.len + 4 + entry.value.len
   wal.stream.write(uint8(entry.kind))
   wal.stream.write(entry.timestamp)
   wal.stream.write(uint32(entry.key.len))
@@ -145,8 +236,9 @@ proc writeEntry*(wal: var WriteAheadLog, entry: WalEntry) =
   wal.stream.write(uint32(entry.value.len))
   if entry.value.len > 0:
     wal.stream.writeData(unsafeAddr entry.value[0], entry.value.len)
-  if wal.syncOnWrite:
-    wal.stream.flush()
+  # Always push to kernel page cache; durability policy decides fsync
+  wal.stream.flush()
+  wal.maybeGroupSync(entryBytes)
   inc wal.entryCount
   # Check rotation every 1000 entries to avoid stat on every write
   if wal.entryCount mod 1000 == 0:
@@ -177,28 +269,87 @@ proc writeCommit*(wal: var WriteAheadLog, timestamp: uint64) =
   ))
 
 proc sync*(wal: var WriteAheadLog) =
+  ## Force durability of all buffered WAL data.
   wal.stream.flush()
-  # 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)
+  fsyncPath(wal.path)
+  wal.markSynced()
+
+proc truncate*(wal: var WriteAheadLog) =
+  ## Reset WAL to empty (header only). Safe only when all prior entries
+  ## are durable in SSTables and nothing remains only-in-memtable.
+  if wal.stream != nil:
+    wal.stream.flush()
+    wal.stream.close()
+  wal.stream = newFileStream(wal.path, fmWrite)
+  if wal.stream == nil:
+    raise newException(IOError, "Cannot truncate WAL: " & wal.path)
+  wal.stream.write(WALMagic)
+  wal.stream.write(WALVersion)
+  wal.stream.flush()
+  fsyncPath(wal.path)
+  wal.entryCount = 0
+  wal.markSynced()
+
+proc rewriteLive*(wal: var WriteAheadLog,
+                  keys: openArray[string],
+                  values: openArray[seq[byte]],
+                  timestamps: openArray[uint64],
+                  deleted: openArray[bool]) =
+  ## Atomically replace WAL contents with a live memtable snapshot.
+  ## Used after a partial flush so unflushed keys remain recoverable.
+  doAssert keys.len == values.len and keys.len == timestamps.len and keys.len == deleted.len
+  if keys.len == 0:
+    wal.truncate()
+    return
+
+  let tmpPath = wal.path & ".rewrite"
+  let s = newFileStream(tmpPath, fmWrite)
+  if s == nil:
+    raise newException(IOError, "Cannot create WAL rewrite file: " & tmpPath)
+  s.write(WALMagic)
+  s.write(WALVersion)
+  var count: uint64 = 0
+  for i in 0 ..< keys.len:
+    let kind = if deleted[i]: wekDelete else: wekPut
+    s.write(uint8(kind))
+    s.write(timestamps[i])
+    s.write(uint32(keys[i].len))
+    if keys[i].len > 0:
+      s.write(keys[i])
+    s.write(uint32(values[i].len))
+    if values[i].len > 0:
+      s.writeData(unsafeAddr values[i][0], values[i].len)
+    inc count
+  s.flush()
+  s.close()
+  fsyncPath(tmpPath)
+
+  if wal.stream != nil:
+    wal.stream.close()
+  if fileExists(wal.path):
+    removeFile(wal.path)
+  moveFile(tmpPath, wal.path)
+  wal.stream = newFileStream(wal.path, fmAppend)
+  if wal.stream == nil:
+    raise newException(IOError, "Cannot reopen WAL after rewrite: " & wal.path)
+  wal.entryCount = count
+  wal.markSynced()
 
 proc setMaxSegmentSize*(wal: var WriteAheadLog, size: int64) =
   wal.maxSegmentSize = size
 
 proc close*(wal: var WriteAheadLog) =
   wal.stream.flush()
-  let fd = posix.open(cstring(wal.path), O_RDWR)
-  if fd != -1:
-    discard posix.fsync(fd)
-    discard posix.close(fd)
+  fsyncPath(wal.path)
+  wal.markSynced()
   wal.stream.close()
 
 proc entryCount*(wal: WriteAheadLog): uint64 = wal.entryCount
 proc path*(wal: WriteAheadLog): string = wal.path
+proc unsyncedEntries*(wal: WriteAheadLog): int = wal.unsyncedEntries
+
+## Legacy alias — true maps to wsmEvery
+proc syncOnWrite*(wal: WriteAheadLog): bool = wal.syncMode == wsmEvery
 
 proc readEntries*(walPath: string, untilTimestamp: uint64 = 0): seq[WalEntry] =
   result = @[]