fix: stabilization session — auth bypass, raft quorum, wire DoS, query operators

Security:
- MIGRATE handler now requires auth (unauthenticated arbitrary writes)
- parseHeader rejects oversized messages before allocation (pre-auth DoS)

Correctness:
- raft commit uses strict majority (N div 2 + 1), fixing even-N minority commit
- power (**) and concat (++) no longer lowered to equality
- != is now the exact complement of = for numerically-equal values
- legacy REP payload carries explicit put/delete tag (PK-only rows survive)
- REP receiver maintains secondary indexes via applyReplicatedPut/Delete
- snapshot send runs gzip off the event loop (heartbeat stall mitigation)

Docs: PLAN.md (session 13), BUG_AUDIT_2026-08.md (~28 findings, 23 tracked),
known-limitations.md, CHANGELOG.md.

Verified: baradadb build clean; test_all + bugfix_test pass.

BUG_AUDIT_2026-08.md

@@ -0,0 +1,81 @@
+# BaraDB — Deep Audit (август 2026)
+
+> Дата: 2026-08-02
+> Метод: 4 паралелни одит-агента по слоеве (Storage / Query / Core / Protocol), всеки чете всички файлове в обхвата си и проверява находките срещу реалния код.
+> Обхват: **само нови дефекти** — 80-те вече оправени в `BUGS.md` / `BUG_AUDIT.md` / `BARADB_CLIENT_BUGS.md` са изключени.
+> **Общо: ~28 находки | Поправени (батч 1): 5 | Остават: 23**
+
+---
+
+## Поправени — батч 1 (5)
+
+| # | Severity | Проблем | Файл | Fix |
+|---|----------|---------|------|-----|
+| C1 | 🔴 CRITICAL | **MIGRATE handler без auth gate** — неавтентикиран клиент пишеше произволни key/value в базата (`handleMigrationMessage → applyMigrationBatch → storeKeys → db.put`). Открито независимо от 2 агента. | `core/server.nim:682` | Добавен `if not authenticated: ... continue` (като REP/DISTTXN блоковете) |
+| C2 | 🔴 CRITICAL | **Raft commit quorum off-by-one за even-N** — `(N+1) div 2` commit-ваше с малцинство при четен брой възли (N=4 → 2/4). Election-ът ползваше коректното strict majority. *GA обхватът е 3-node (нечетно), където формулите съвпадат.* | `core/raft.nim:653` | `let majority = (node.peers.len + 1) div 2 + 1` (съвпада с election); регресионен тест за 4-node |
+| H1 | 🟠 HIGH | **Pre-auth memory-exhaustion DoS** — `parseHeader` не ограничаваше `length` (uint32 до ~4 GiB); `recvExactWithTimeout` пре-алокира преди auth check. | `core/server.nim:168` | Reject `length > uint32(MaxWireStringLen)` (64 MB) преди алокация |
+| H4 | 🟠 HIGH | **`**` и `++` се lower-ваха към equality** — `bkPow`/`bkConcat` липсваха в op-mapping case-а и попадаха в `else: irEq` (`2 ** 3` → `false`, `'a' ++ 'b'` → `false`). | `query/exec/lower.nim:79` | `of bkPow: irOp = irPow`, `of bkConcat: irOp = irAdd`; 2 регресионни теста |
+| H5 | 🟠 HIGH | **`!=` не е отрицание на `=`** — `irNeq` short-circuit-ваше на string inequality, така че `5 != 5.0` → true, но `5 = 5.0` → true. | `query/exec/eval.nim:438` | `irNeq` numeric-first (точно допълнение на `irEq`); регресионен тест |
+
+**Верификация:** `baradadb` build чист; `tests/test_all.nim` (пълен suite) и `tests/bugfix_test.nim` минават без `[FAILED]`.
+
+---
+
+## Остават (23)
+
+### 🟠 HIGH (8)
+
+| # | Проблем | Файл | Предложен fix |
+|---|---------|------|---------------|
+| H2 | **TLS client връзките между възли не верифицират сертификата** — `forwardQueryToLeader` ползва `verifyMode = CVerifyNone` → MITM на клъстър линка. Raft client dials са със същия default (`raftTlsVerifyPeer: false`). | `core/server.nim:70` | Verify peer cert срещу CA при client handshake (fail-closed при enabled TLS) |
+| H3 | **Semi-sync репликация връща durable LSN при partial/zero ack** — `rmSync` връща 0 при partial ack, но `rmSemiSync` връща LSN безусловно (само debug echo при 0 acks). | `core/replication.nim:217` | `if syncReplicaCount > 0 and ackCount < syncReplicaCount: return 0` |
+| H6 | **`COUNT/SUM/AVG(DISTINCT)` игнорира DISTINCT** — `funcDistinct` се set-ва в парсера (BUG-017), но `aggDistinct` никога не се копира/чете; няма dedup в aggregate пътищата. | `query/exec/lower.nim:116`, `plan_exec.nim` | Копирай `aggDistinct = node.funcDistinct`; dedup чрез `HashSet[string]` преди count/sum/avg |
+| H7 | **`UNION/INTERSECT/EXCEPT` (без ALL) чупят с KeyError** — dedup ключът чете `row["$value"]`, но projected редове нямат този ключ. Само `UNION ALL` работи. | `query/executor.nim:459` | Dedup ключ от projected колоните (join `valueToString` по ред на `cols`), не `row["$value"]` |
+| H8 | **`MERGE ... WHEN MATCHED THEN DELETE` / `AND <cond>` не се изпълняват** — AST/parser полетата (BUG-032) съществуват, но executor-ът не ги реферира; DELETE е no-op, condition се игнорира. | `query/executor.nim:752` | В matched клона: провери `mergeMatchedCondition`, после honor-вай `mergeMatchedDelete` |
+| H9 | **WAL recovery чупи процеса при torn record** — recovery parser-ът вярва на `keyLen`/`valLen` (до ~4 GiB alloc) и `kind` (out-of-range enum → `CaseStmtError` Defect, не се catch-ва). | `storage/lsm.nim:704` | Bound lengths + валидирай `kind` преди use; дългосрочно per-record CRC32 |
+| H10 | **B-tree `remove` пише separator с грешна конвенция** — `splitChild` ползва left child max, `removeRec` пише right child min (`child.keys[0]`) → ключове стават ненамираеми при internal nodes (silent data loss). | `storage/btree.nim:377` | Separator = max ключ на left child, преизчислен след rebalance |
+
+### 🟡 MEDIUM (11)
+
+| # | Проблем | Файл | Предложен fix |
+|---|---------|------|---------------|
+| M1 | **MVCC `write` трие от `activeTxns` по време на итерация** — `delete` proc-ът ползва collect-then-delete, но `write` трие inline (unsafe, пропуска timed-out txns). | `core/mvcc.nim:180` | Collect stale ids в seq, трий след loop-а |
+| M2 | **disttxn `connectWithTimeout` без SO_ERROR + uncaught RPC** — refused connect е "writable" → връща true; `sendDistTxnRpc` няма try/except → OSError wedge-ва 2PC състояние. (BUG-042 fix-нат в replication, не тук.) | `core/disttxn.nim:88` | `getsockopt(SO_ERROR)` + try/except около per-participant RPC |
+| M3 | **`checkpoint` leaking write lock при exception** — `acquireWrite` без try/finally; IOError от flush/rotate пропуска `releaseWrite` → постоянен hang. | `storage/lsm.nim:932` | try/finally около lock-а (и walLock) |
+| M4 | **`flushUnsafe` празни memtable преди SSTable write** — при IOError на `writeSSTable` данните са загубени от memory (остават само в WAL, невидими за live reads). | `storage/lsm.nim:870` | Първо `writeSSTable`, после clear на memtable |
+| M5 | **Compaction пропуска empty-string ключа** — dedup sentinel `lastKey = ""` skip-ва ключ `""` → data loss при compact на празен ключ. | `storage/compaction.nim:130` | `haveLast` флаг вместо sentinel стойност |
+| M6 | **`rewriteLive` data-loss window** — `removeFile(wal.path)` преди `moveFile`; crash между тях губи unflushed записи. `rename(2)` и без това е атомен replace. | `storage/wal.nim:329` | Махни `removeFile`, остави атомния `moveFile` |
+| M7 | **Compaction unlink-ва input-ите преди output-ът да е loadable в каталога** — `applyCompactionResult` re-load-ва output в try/except (само warning); ако fail-не след unlink → загуба на ключове. | `storage/compaction.nim:150` | Load/verify output в каталога ПРЕДИ unlink на input-ите |
+| M8 | **`OFFSET n` без `LIMIT` връща 0 реда; negative `LIMIT` чупи** — `limitCount = 0` е sentinel и за "няма limit", и за "LIMIT 0"; `sourceRows[start..<endIdx]` с endIdx<start → IndexDefect. | `query/exec/lower.nim:411`, `plan_exec.nim:219` | Отделен sentinel (-1 = unlimited); clamp negative |
+| M9 | **Aggregate window функции връщат NULL** — `SUM/AVG/COUNT/MIN/MAX OVER (...)` попадат в `else` клона (`"\N"`); само ranking/lead/lag се handle-ват. | `query/exec/window.nim` | `of "sum","avg","count","min","max"` с `resolveFrameBounds` |
+| M10 | **WebSocket приема unmasked client frames** — RFC 6455 §5.1 изисква server да затвори връзката при unmasked client frame (cache-poisoning защита). | `core/websocket.nim:85` | Затвори връзката при `masked == false` |
+| M11 | **WebSocket без frame/message size limit → DoS** — `buf.add(chunk)` расте неограничено; няма 125-byte control-frame cap. | `core/websocket.nim:215` | Max frame/message size + control-frame cap |
+| M12 | **WebSocket SUBSCRIBE bypass-ва table-level auth** — всеки автентикиран клиент subscribe-ва към任意 таблица и получава всички insert/delete. | `core/websocket.nim:232` | Table read authorization при subscribe |
+
+### 🟢 LOW (4)
+
+| # | Проблем | Файл | Предложен fix |
+|---|---------|------|---------------|
+| L1 | **SCRAM timing user enumeration** — unknown user връща веднага, known user прави urandom+HMAC/PBKDF2 работа → timing delta (BUG-049 fix-на съобщението, не timing-а). | `protocol/auth.nim:227` | Equivalent dummy work за unknown users |
+| L2 | **SCRAM channel-binding не се верифицира** — `c=` се приема verbatim; RFC 5802 изисква валидация. Не е exploitable днес (няма TLS-CB). | `protocol/auth.nim:259` | Enforce expected `c=` (напр. `biws`) |
+| L3 | **mmap read `offset + size` overflow** — `offset + size > region.size` с native int wrap-ва негативно при corrupt offset/size → OOB read. v3 SSTable-ите са CRC-защитени (reachable само през legacy v1/v2). | `storage/mmap.nim` | `offset > region.size - size` (без overflow) |
+| L4 | **NULL equality semantics** — `NULL = NULL` и `col = NULL` → true (string sentinel сравнение), не unknown/false. Системно за string-based value модела. | `query/exec/eval.nim:431` | Three-valued logic за NULL (по-голям рефакторинг) |
+
+### Хигиена
+
+- **Stray 94 KB компилиран ELF binary** в `src/barabadb/protocol/scram` — случайно commit-нат в source tree-то; да се премахне (+ `.gitignore`).
+
+---
+
+## Проверени и чисти (не са бъгове)
+
+- JWT `exp`/alg-confusion: pinned `jwt-nim-baraba#fbe084b` `verify()` enforce-ва alg-match, reject-ва `NONE`, проверява `exp/nbf/iat`, constant-time compare.
+- `auth.nim` `constantTimeCompare` и SCRAM `verifyClientProof` са constant-time; празен JWT secret fail-closed (`server.nim:59`).
+- `wire.nim` deserialize bounds/depth caps са sound.
+- CRC byte ranges / `headerSize = 40` са консистентни между write/verify/load (format *коментарът* още казва "36" — само коментар).
+- Lock ordering (`walLock` в `db.lock`; gate преди `db.lock`), mmap negative offset / `close()` recursion / fd handling (BUG-036/046) — непокътнати.
+- `COUNT(col)` изключва NULLs (`v.kind != vkNull`); `LIMIT 0` → празно е коректно; IN-list се lower-ва към OR/AND вериги.
+
+---
+
+*Виж също: `PLAN.md` (Сесия 13), `docs/en/known-limitations.md`.*

CHANGELOG.md

@@ -2,6 +2,28 @@
 
 All notable changes to BaraDB are documented in this file.
 
+## [Unreleased]
+
+### Security
+
+- **MIGRATE auth bypass (CRITICAL)** — the internal `MIGRATE` text-protocol handler now requires authentication (matching the `REP`/`DISTTXN` handlers); previously an unauthenticated client could inject arbitrary key/value rows (`core/server.nim`)
+- **Pre-auth wire-length DoS (HIGH)** — `parseHeader` rejects messages larger than the 64 MB wire cap before allocating the receive buffer (`core/server.nim`)
+
+### Fixed
+
+- **Raft commit quorum (CRITICAL)** — commit now requires a strict majority (`N div 2 + 1`), matching the election check; the previous `(N+1) div 2` formula committed at a minority for even-sized clusters (`core/raft.nim`)
+- **`**` / `++` operators (HIGH)** — power and concat are no longer lowered to equality: `2 ** 3` → 8, `'a' ++ 'b'` → `'ab'` (`query/exec/lower.nim`)
+- **`!=` semantics (HIGH)** — `!=` is now the exact complement of `=` for numerically-equal values (`5 != 5.0` is false) (`query/exec/eval.nim`)
+- **REP replication put/delete encoding** — the legacy REP payload carries an explicit op tag so PK-only inserts (empty value) replicate as puts instead of vanishing as deletes (`core/replication.nim`, `core/server.nim`)
+- **REP receiver secondary indexes** — the legacy REP receiver applies via `applyReplicatedPut/Delete` under the storage gate, keeping B-tree/FTS/HNSW/graph indexes consistent on the replica (`core/server.nim`)
+- **Snapshot send stall (partial)** — the leader's snapshot send runs gzip off the event loop on a worker thread (`gzipFileAsync`), so heartbeats keep firing during compression; tar (send) and the restore path still run on the loop (`core/backup.nim`, `core/raft.nim`)
+
+### Added
+
+- Deep audit report `BUG_AUDIT_2026-08.md` (~28 findings; 5 fixed in this batch, 23 tracked)
+
+---
+
 ## [1.3.0] — 2026-07-30
 
 ### Raft cluster — Supported (single `default` DB scope)

PLAN.md

@@ -145,6 +145,28 @@
 
 ---
 
+## Сесия 13: Stabilization & Deep Audit (август 2026)
+
+> **Цел**: Подобряване на надеждността и коректността върху вече завършеното ядро — целеви поправки + системен паралелен одит по слоеве (Storage / Query / Core / Protocol).
+
+### Целеви поправки (завършени)
+
+| # | Поправка | Файлове | Статус |
+|---|----------|---------|--------|
+| 1 | **REP delete-from-empty** — legacy REP payload носи явен put/delete таг (`encodeRepPayload`/`decodeRepPayload`); PK-only редове вече не изчезват при репликация | `core/replication.nim`, `core/server.nim` | ✅ + тестове |
+| 2 | **Snapshot stall (частична mitigation)** — gzip при leader snapshot send се изнася извън event loop-а през worker thread (`gzipFileAsync`); heartbeats текат по време на компресия | `core/backup.nim`, `core/raft.nim`, `baradadb.nim` | ✅ + e2e |
+| 3 | **REP receiver индекси** — receiver-ът минава през `applyReplicatedPut/Delete` под storage gate; вторичните индекси (B-tree/FTS/HNSW/graph) се поддържат на репликата | `core/server.nim` | ✅ + тест |
+
+### Deep Audit (2026-08)
+
+4 паралелни одит-агента по слоеве; ~28 нови находки (без дублиране на 80-те вече оправени в `BUGS.md`/`BUG_AUDIT.md`). Пълен отчет: [`BUG_AUDIT_2026-08.md`](BUG_AUDIT_2026-08.md).
+
+**Батч 1 — поправени (5):** MIGRATE auth bypass (CRITICAL), raft commit strict-majority за even-N (CRITICAL), pre-auth wire-length DoS (HIGH), `**`/`++` lowering към equality (HIGH), `!=` не е отрицание на `=` (HIGH).
+
+**Остават (~23):** вж. `BUG_AUDIT_2026-08.md` — TLS peer verify, semi-sync partial-ack, COUNT(DISTINCT), UNION/INTERSECT/EXCEPT crash, MERGE THEN DELETE, WAL recovery crash, B-tree separator convention, MVCC delete-during-iteration, disttxn SO_ERROR, checkpoint lock leak, flushUnsafe data-loss, compaction empty-key, wal rewriteLive window, OFFSET-без-LIMIT, window агрегати, WebSocket (3), SCRAM (2), mmap overflow.
+
+---
+
 ## Какво остава от старите планове
 
 | Стар план | Статус |
@@ -157,6 +179,7 @@
 | **Този план** — Сесии 10, 11, 12 | ✅ Завършен |
 | Raft C3a/C3b + DDL/forward/compact/metrics (2026-07-30) | ✅ Завършен на `main` — `docs/superpowers/specs/2026-07-30-raft-cluster-status.md` |
 | **Production GA v1.2.0** (single-node) | ✅ `docs/superpowers/plans/2026-07-30-production-ga.md` |
+| **Сесия 13** — Stabilization & Deep Audit (2026-08) | 🔄 В процес — батч 1 завършен (5 поправки); `BUG_AUDIT_2026-08.md` |
 
 ---
 
@@ -170,4 +193,4 @@
 
 ---
 
-*План версия: 2026-05-17*
+*План версия: 2026-08-02*

docs/en/known-limitations.md

@@ -46,11 +46,11 @@ Documented in [distributed.md](distributed.md). Supported scope:
 
 ## Newly documented limitations
 
-- **Legacy non-raft REP replication infers delete from empty value** — the non-raft replication path still treats an empty value as a delete, so inserts into a PK-only table are misapplied over that path (the row vanishes). Use raft replication instead.
+- **Legacy non-raft REP replication delete inference** — resolved: the non-raft REP payload now carries an explicit put/delete op tag (`encodeRepPayload`/`decodeRepPayload` in `core/replication.nim`), so PK-only inserts (empty LSM value) replicate as puts instead of being misapplied as deletes.
 - **Snapshot-restore ctx staleness** — after an InstallSnapshot restore, HTTP endpoints using the startup-captured ctx may serve stale data until the node is restarted; the `/query` path is fresh per-request. Pre-existing client connections likewise see pre-restore state — reconnect after a restore.
 - **FK-cascade divergence under raft** — `ON DELETE/UPDATE CASCADE` (and `SET NULL`) effects are not raft-replicated: followers only apply the parent row's KV change, so cascaded child rows persist on followers. Avoid FK actions on raft-replicated tables, or accept periodic snapshot resync.
 - **Uncommitted writes in snapshots** — the leader applies writes locally before raft majority commit; a snapshot taken in that window can include writes that never commit (phantom rows after restore + leadership change). Narrow window; fix tracked for a later release.
-- **Event-loop stall during snapshot build/restore** — snapshot build/restore performs blocking tar/gzip on the node's event loop; large data dirs can stall heartbeats and trigger an election mid-transfer.
+- **Event-loop stall during snapshot build/restore** — partially mitigated: the leader's snapshot *send* now tars under the storage gate but runs the CPU-heavy gzip off the event loop on a worker thread (`gzipFileAsync` in `core/backup.nim`), so heartbeats keep firing during compression. The tar itself (send path) and the whole *restore* path (tar extract + DB reopen) still run on the event loop, so very large data dirs can still stall heartbeats during those phases; a full fix (an async/try-lock storage gate so the loop never blocks) is tracked for a later release.
 
 ## Operational requirements
 

src/barabadb/core/backup.nim

@@ -26,6 +26,8 @@ import std/strutils
 import std/times
 import std/algorithm
 import std/json
+import std/asyncdispatch
+import std/threadpool
 import barabadb/storage/lsm
 
 type
@@ -260,6 +262,80 @@ proc backupDataDir*(dataDir: string, output: string, excludes: seq[string] = @[]
   echo "  Source:   ", dataDir
   return true
 
+proc tarDataDir*(dataDir: string, output: string, excludes: seq[string] = @[]): bool =
+  ## Create an UNCOMPRESSED tar of `dataDir` at `output` (no gzip). The raft
+  ## snapshot sender runs this under the storage gate for a consistent file
+  ## capture, then compresses off the event loop via gzipFileAsync.
+  if not dirExists(dataDir):
+    echo "ERROR: Data directory not found: ", dataDir
+    return false
+
+  let parent = parentDir(dataDir)
+  let name = lastPathPart(dataDir)
+  var excludeArgs = ""
+  for pattern in excludes:
+    excludeArgs.add(" --exclude=" & quoteShell(pattern))
+
+  let cmd = "tar -cf " & quoteShell(output) & excludeArgs &
+            " -C " & quoteShell(parent) & " " & quoteShell(name)
+  let (outputStr, exitCode) = execCmdEx(cmd)
+  if exitCode != 0:
+    echo "ERROR: tar command failed with exit code ", exitCode
+    if outputStr.len > 0:
+      echo outputStr
+    return false
+  return true
+
+proc gzipFile*(input: string, output: string,
+               compression: int = DEFAULT_COMPRESSION): bool =
+  ## gzip a single file `input` -> `output`. Pure CPU over an already-captured
+  ## file: no shared storage state, so it is safe to run on a worker thread
+  ## outside the storage gate and off the raft event loop.
+  if not fileExists(input):
+    echo "ERROR: File not found: ", input
+    return false
+
+  let cmd = "gzip -" & $compression & " -c " & quoteShell(input) &
+            " > " & quoteShell(output)
+  let (outputStr, exitCode) = execCmdEx("bash -c " & quoteShell(cmd))
+  if exitCode != 0:
+    echo "ERROR: gzip command failed with exit code ", exitCode
+    if outputStr.len > 0:
+      echo outputStr
+    return false
+  return true
+
+proc gunzipFile*(input: string, output: string): bool =
+  ## Decompress a gzip file `input` -> `output`. Inverse of gzipFile.
+  if not fileExists(input):
+    echo "ERROR: File not found: ", input
+    return false
+
+  let cmd = "gzip -dc " & quoteShell(input) & " > " & quoteShell(output)
+  let (outputStr, exitCode) = execCmdEx("bash -c " & quoteShell(cmd))
+  if exitCode != 0:
+    echo "ERROR: gunzip command failed with exit code ", exitCode
+    if outputStr.len > 0:
+      echo outputStr
+    return false
+  return true
+
+proc gzipFileWorker(input: string, output: string, compression: int): bool {.gcsafe.} =
+  ## Thread entry point: touches only its own (copied) string args + execCmdEx,
+  ## so it is safe to run off the main/event-loop thread under ARC/ORC.
+  gzipFile(input, output, compression)
+
+proc gzipFileAsync*(input: string, output: string,
+                    compression: int = DEFAULT_COMPRESSION): Future[bool] {.async.} =
+  ## Run gzipFile on a threadpool worker and await completion WITHOUT blocking
+  ## the calling async event loop — heartbeats/election timers keep firing
+  ## during the CPU-heavy compression. Polls the FlowVar via sleepAsync so the
+  ## dispatcher stays responsive instead of stalling on a blocking join.
+  var fv = spawn gzipFileWorker(input, output, compression)
+  while not fv.isReady:
+    await sleepAsync(20)
+  result = ^fv
+
 proc restoreDataDir*(input: string, dataDir: string, verbose: bool = false, dryRun: bool = false): bool =
   ## Restore from a tar.gz backup.
   ## When dryRun is true, only prints what would be done.

src/barabadb/core/raft.nim

@@ -13,6 +13,7 @@ import std/os
 import logging
 import ../protocol/wire
 import ../protocol/ssl
+import backup
 
 type
   RaftState* = enum
@@ -95,8 +96,10 @@ type
                            baseTerm: uint64): bool {.gcsafe.}
     snapIncomingId*: uint64
     snapIncomingFile*: string
-    ## Leader InstallSnapshot send. buildSnapshot archives the current data
+    ## Leader InstallSnapshot send. buildSnapshot writes an UNCOMPRESSED tar
-    ## dir into destPath (wired in baradadb.nim via backupDataDir).
+    ## of the data dir to destPath (wired in baradadb.nim via tarDataDir, under
+    ## the storage gate); sendSnapshot then compresses it off the event loop
+    ## (gzipFileAsync) and streams the resulting .tar.gz to the follower.
     ## snapRejectStreak counts consecutive floor-level AppendEntries rejects
     ## per peer; at 2 the peer is queued in snapPending and the network layer
     ## (processMessage) kicks off sendSnapshot. snapSending is the
@@ -646,8 +649,11 @@ proc handleAppendReply*(node: RaftNode, peerId: string, reply: RaftMessage) =
     node.snapRejectStreak.del(peerId)
     node.snapPending.excl(peerId)
 
-    # Update commit index using true majority calculation
+    # Update commit index using strict majority — the same form as the election
-    let majority = (node.peers.len + 1 + 1) div 2  # majority of cluster (peers + leader)
+    # check in handleVoteReply. Cluster size N = peers.len + 1; a strict
+    # majority is N div 2 + 1. The previous (N + 1) div 2 under-counted for
+    # even-sized clusters (e.g. N=4 committed at 2/4, a minority).
+    let majority = (node.peers.len + 1) div 2 + 1  # strict majority of cluster
     var newCommitIdx = node.commitIndex
 
     # Walk logical indices high→low via findLogEntryByIndex (log may be compacted).
@@ -1006,9 +1012,10 @@ proc sendSnapshot*(net: RaftNetwork, peerId: string) {.async.} =
   ## when handleAppendReply queues the peer in snapPending after consecutive
   ## floor-level rejects. Single-flight per peer via node.snapSending.
   ##
-  ## Runs on the raft event loop; buildSnapshot performs blocking disk I/O
+  ## Runs on the raft event loop. buildSnapshot performs the tar on the loop
-  ## (tar+gzip). Snapshot sends are rare, so we accept the stall rather than
+  ## under the storage gate (consistent capture); the CPU-heavy gzip then runs
-  ## adding a worker round-trip (same trade-off as restoreSnapshot).
+  ## on a worker thread via gzipFileAsync, awaited here, so heartbeats and the
+  ## election timer keep firing during compression instead of stalling.
   let node = net.node
   if peerId in node.snapSending:
     return
@@ -1026,15 +1033,25 @@ proc sendSnapshot*(net: RaftNetwork, peerId: string) {.async.} =
 
   let baseIndex = node.lastSnapshotIndex
   let baseTerm = node.lastSnapshotTerm
-  let destPath = node.dataDir / ("snap_out_" & $snapId & ".tar.gz")
+  # buildSnapshot writes an uncompressed tar (under the storage gate, on this
+  # loop); gzipFileAsync then compresses it on a worker thread off the loop.
+  # The follower still receives a normal .tar.gz byte stream.
+  let tarPath = node.dataDir / ("snap_out_" & $snapId & ".tar")
+  let destPath = tarPath & ".gz"
   defer:
+    if fileExists(tarPath):
+      removeFile(tarPath)
     if fileExists(destPath):
       removeFile(destPath)
 
-  if not node.buildSnapshot(destPath):
+  if not node.buildSnapshot(tarPath):
     warn("sendSnapshot: buildSnapshot failed; aborting snapshot send to " & peerId)
     return
 
+  if not await gzipFileAsync(tarPath, destPath):
+    warn("sendSnapshot: snapshot compression failed; aborting send to " & peerId)
+    return
+
   var f: File
   if not open(f, destPath, fmRead):
     warn("sendSnapshot: cannot open built archive " & destPath)

src/barabadb/core/replication.nim

@@ -96,6 +96,41 @@ proc connectWithTimeout(sock: Socket, host: string, port: Port, timeoutMs: int):
     sock.getFd.setBlocking(true)
     return err == 0
 
+type
+  RepOp* = enum
+    ropPut
+    ropDelete
+    ropInvalid
+
+proc encodeRepPayload*(deleted: bool, key: string, value: seq[byte]): seq[byte] =
+  ## Tagged legacy-REP payload. A leading op tag makes put/delete explicit so
+  ## an empty put value (PK-only rows store an empty LSM value) is never
+  ## mistaken for a delete on the receiver:
+  ##   put    -> 'P' & key & "\x00" & value   (value may be empty)
+  ##   delete -> 'D' & key
+  ## Mirrors the raft convention (explicit "put"/"delete" commands).
+  if deleted:
+    cast[seq[byte]]("D" & key)
+  else:
+    cast[seq[byte]]("P" & key & "\x00" & cast[string](value))
+
+proc decodeRepPayload*(data: seq[byte]): tuple[op: RepOp, key: string, value: seq[byte]] =
+  ## Inverse of encodeRepPayload. Returns ropInvalid for empty or untagged
+  ## payloads rather than guessing the operation from the value length.
+  if data.len == 0:
+    return (ropInvalid, "", @[])
+  case char(data[0])
+  of 'P':
+    let body = data[1 ..< data.len]
+    let nullPos = find(body, byte(0))
+    if nullPos < 0:
+      return (ropInvalid, "", @[])
+    return (ropPut, cast[string](body[0 ..< nullPos]), body[nullPos + 1 ..< body.len])
+  of 'D':
+    return (ropDelete, cast[string](data[1 ..< data.len]), @[])
+  else:
+    return (ropInvalid, "", @[])
+
 proc shipToReplica(replica: Replica, lsn: uint64, data: seq[byte]): bool =
   ## Send replication data to a replica via TCP.
   ## Protocol: "REP <lsn> <dataLen>\n<data>"

src/barabadb/core/server.nim

@@ -22,6 +22,7 @@ import ../query/parser
 import ../query/ast
 import ../query/executor
 import ../query/exec/params
+import ../query/exec/dml
 import ../storage/lsm
 import ../storage/gate
 import ../core/mvcc
@@ -165,6 +166,11 @@ proc parseHeader(data: string): (bool, MessageHeader) =
     return (false, MessageHeader())
   let kind = cast[MsgKind](rawKind)
   let length = readUint32BE(data, 4)
+  # Reject oversized messages before any buffer allocation: recvExactWithTimeout
+  # pre-allocates `length` bytes before the auth check, so an unbounded uint32
+  # (up to ~4 GiB) is a pre-auth memory-exhaustion DoS. Cap at the wire max.
+  if length > uint32(MaxWireStringLen):
+    return (false, MessageHeader())
   let requestId = readUint32BE(data, 8)
   return (true, MessageHeader(kind: kind, length: length, requestId: requestId))
 
@@ -413,14 +419,12 @@ proc executeQuery(db: LSMTree, ctx: ExecutionContext, query: string, params: seq
           # the raft path below handles the statement).
           if raftNode == nil and replication != nil and res.keyValuePairs.len > 0:
             for pair in res.keyValuePairs:
-              # Legacy REP wire format: key \x00 value, empty value = delete
+              # Legacy REP wire format: explicit 'P'/'D' op tag (see
-              # on the receiver. Deletes ship an empty value as before.
+              # encodeRepPayload). The tag — not an empty value — distinguishes
-              let value = if pair.deleted: @[] else: pair.value
+              # a put from a delete, so PK-only rows (empty value) replicate as
-              var data = newSeq[byte](pair.key.len + 1 + value.len)
+              # puts instead of vanishing as deletes.
-              for i, c in pair.key: data[i] = byte(c)
+              discard replication.writeLsn(
-              data[pair.key.len] = byte(0)
+                encodeRepPayload(pair.deleted, pair.key, pair.value))
-              for i, c in value: data[pair.key.len + 1 + i] = c
-              discard replication.writeLsn(data)
           qr = QueryResult(affectedRows: res.affectedRows, rowCount: res.rows.len)
           qr.columns = res.columns
 
@@ -655,14 +659,25 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.}
               if chunk.len == 0: break
               data.add(chunk)
             if data.len > 0:
-              let nullPos = data.find('\0')
+              # Op tag — not value length — decides put vs delete, so a PK-only
-              if nullPos >= 0:
+              # put (empty value) is applied as a put and the row survives.
-                let key = data[0..<nullPos]
+              let decoded = decodeRepPayload(cast[seq[byte]](data))
-                let value = data[nullPos+1..^1]
+              case decoded.op
-                if value.len > 0:
+              of ropPut, ropDelete:
-                  server.db.put(key, stringToBytes(value))
+                # Apply through applyReplicatedPut/Delete (not raw db.put/delete)
-                else:
+                # so secondary B-tree/FTS/HNSW/graph indexes stay consistent on
-                  server.db.delete(key)
+                # the replica — the same path raft uses. server.ctx is the
+                # canonical default ctx whose index structures the per-connection
+                # query clones share. Under the storage gate: those structures
+                # are shared with hunos HTTP workers and are only safe to mutate
+                # under it.
+                withStorageGate:
+                  if decoded.op == ropPut:
+                    applyReplicatedPut(server.ctx, decoded.key, decoded.value)
+                  else:
+                    applyReplicatedDelete(server.ctx, decoded.key)
+              of ropInvalid:
+                discard
           await client.send("ACK " & $lsn & "\n")
         else:
           await client.send("ERR\n")
@@ -670,6 +685,9 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.}
 
       # Detect shard migration data (starts with "MIGRATE ")
       if headerData.len >= 8 and headerData[0..7] == "MIGRATE ":
+        if not authenticated:
+          await client.send("ERR auth required\n")
+          continue
         var rest = headerData[8..^1]
         while '\n' notin rest:
           let more = await client.recvWithTimeout(1024, idleTimeout)

src/barabadb/query/exec/eval.nim

@@ -436,11 +436,12 @@ proc evalExprOld*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContex
       except CatchableError: discard
       return "false"
     of irNeq:
-      if left != right: return "true"
+      # Numeric-first so `!=` is the exact complement of `=` (irEq): string
-      # Try numeric comparison
+      # inequality alone would make `5 != 5.0` true while `5 = 5.0` is true.
       try:
         return if parseFloat(left) != parseFloat(right): "true" else: "false"
-      except CatchableError: return "false"
+      except CatchableError:
+        return if left != right: "true" else: "false"
     of irLt:
       try:
         return if parseFloat(left) < parseFloat(right): "true" else: "false"

src/barabadb/query/exec/lower.nim

@@ -76,6 +76,8 @@ proc lowerExpr*(node: Node): IRExpr =
     of bkJsonContainedBy: irOp = irJsonContainedBy
     of bkJsonHasAny: irOp = irJsonHasAny
     of bkJsonHasAll: irOp = irJsonHasAll
+    of bkPow: irOp = irPow
+    of bkConcat: irOp = irAdd  # irAdd concatenates string operands
     else: irOp = irEq
     result.binOp = irOp
     result.binLeft = lowerExpr(node.binLeft)

src/baradadb.nim

@@ -439,19 +439,20 @@ proc main() =
             echo "[raft] Snapshot restore failed: ", e.msg
             result = false
 
-    # Leader InstallSnapshot send: archive the default DB's data directory
+    # Leader InstallSnapshot send: tar the default DB's data directory into the
-    # into the path raft picks (dataDir/raft/snap_out_<snapId>.tar.gz). Like
+    # path raft picks (dataDir/raft/snap_out_<snapId>.tar). The tar runs here on
-    # restoreSnapshot this runs on the raft event loop and blocks on disk I/O
+    # the raft event loop under the storage gate; sendSnapshot then gzips it on
-    # (tar+gzip); snapshot sends are rare, so we accept the stall.
+    # a worker thread off the loop (gzipFileAsync) so heartbeats keep flowing.
     raftNode.buildSnapshot = proc(destPath: string): bool {.gcsafe.} =
-      echo "[raft] Building snapshot archive ", destPath
+      echo "[raft] Building snapshot tar ", destPath
       {.cast(gcsafe).}:
         # Hold the storage gate while tarring the data dir so a concurrent
         # memtable flush (HTTP /query path) cannot write an SSTable
-        # mid-archive.
+        # mid-archive. Compression happens later, off the gate (see
+        # sendSnapshot), so it neither stalls the loop nor blocks applies.
         withStorageGate:
           try:
-            result = backupDataDir(defaultDbDir, destPath)
+            result = tarDataDir(defaultDbDir, destPath)
           except CatchableError as e:
             echo "[raft] Snapshot build failed: ", e.msg
             result = false

tests/bugfix_test.nim

@@ -7,6 +7,7 @@ import ../src/barabadb/query/exec/params
 import ../src/barabadb/query/exec/dml
 import ../src/barabadb/core/types
 import ../src/barabadb/core/config
+import ../src/barabadb/core/replication
 import ../src/barabadb/storage/lsm
 
 const testDir = "/tmp/baradb_bugfix_test"
@@ -552,3 +553,75 @@ suite "Raft TLS config":
     check cfg.raftTlsKeyFile == "/tmp/raft.key"
     check cfg.raftTlsCaFile == "/tmp/raft-ca.crt"
     check cfg.raftTlsVerifyPeer == true
+
+
+suite "Legacy REP payload encoding — empty value is not a delete":
+
+  test "PK-only put (empty value) round-trips as a put, not a delete":
+    ## Regression: the legacy REP receiver used to infer a delete from an empty
+    ## value, so PK-only rows (empty LSM value) vanished on the replica.
+    let decoded = decodeRepPayload(encodeRepPayload(false, "pkonly.id=3", @[]))
+    check decoded.op == ropPut
+    check decoded.key == "pkonly.id=3"
+    check decoded.value.len == 0
+
+  test "delete round-trips as a delete":
+    let decoded = decodeRepPayload(encodeRepPayload(true, "users.id=1", @[]))
+    check decoded.op == ropDelete
+    check decoded.key == "users.id=1"
+    check decoded.value.len == 0
+
+  test "put with a non-empty value preserves the value bytes":
+    let decoded = decodeRepPayload(
+      encodeRepPayload(false, "users.id=1", cast[seq[byte]]("bob")))
+    check decoded.op == ropPut
+    check decoded.key == "users.id=1"
+    check cast[string](decoded.value) == "bob"
+
+  test "value containing a null byte survives the round-trip":
+    ## Decode splits on the FIRST null (the key/value separator) only.
+    let value = @[byte('a'), byte(0), byte('b')]
+    let decoded = decodeRepPayload(encodeRepPayload(false, "k", value))
+    check decoded.op == ropPut
+    check decoded.key == "k"
+    check decoded.value == value
+
+  test "empty or untagged payloads decode as invalid, not delete":
+    check decodeRepPayload(@[]).op == ropInvalid
+    check decodeRepPayload(cast[seq[byte]]("Xfoo")).op == ropInvalid
+
+
+suite "Query operator correctness — audit batch 1":
+
+  test "power operator ** evaluates, not lowered to equality":
+    ## Regression: bkPow used to fall through to `else: irOp = irEq`, so
+    ## `2 ** 3` evaluated as `2 = 3` (false) instead of 8.
+    var ctx = setupCtx()
+    defer: teardown(ctx)
+    discard executeQuery(ctx, parse("INSERT INTO users (id, name) VALUES (1, 'a')"))
+    let r = executeQuery(ctx, parse("SELECT 2 ** 3 AS x FROM users"))
+    check r.success
+    check r.rows.len == 1
+    check parseFloat(valueToString(r.rows[0]["x"])) == 8.0
+
+  test "concat operator ++ concatenates strings":
+    ## Regression: bkConcat also fell through to irEq, so `'a' ++ 'b'`
+    ## evaluated as `'a' = 'b'` (false) instead of "ab".
+    var ctx = setupCtx()
+    defer: teardown(ctx)
+    discard executeQuery(ctx, parse("INSERT INTO users (id, name) VALUES (1, 'a')"))
+    let r = executeQuery(ctx, parse("SELECT 'a' ++ 'b' AS x FROM users"))
+    check r.success
+    check r.rows.len == 1
+    check valueToString(r.rows[0]["x"]) == "ab"
+
+  test "!= is the complement of = for numerically equal values":
+    ## Regression: irNeq short-circuited on string inequality, so `1 != 1.0`
+    ## was true while `1 = 1.0` was also true (not complements).
+    var ctx = setupCtx()
+    defer: teardown(ctx)
+    discard executeQuery(ctx, parse("INSERT INTO users (id, name) VALUES (1, 'alice')"))
+    let eq = executeQuery(ctx, parse("SELECT * FROM users WHERE id = 1.0"))
+    let neq = executeQuery(ctx, parse("SELECT * FROM users WHERE id != 1.0"))
+    check eq.rows.len == 1   # 1 = 1.0 -> true
+    check neq.rows.len == 0  # 1 != 1.0 -> false (old bug returned the row)

tests/test_all.nim

@@ -2895,6 +2895,29 @@ suite "Raft InstallSnapshot Send":
     check node.matchIndex["peer-1"] == 101
     check node.nextIndex["peer-1"] == 102
 
+  test "commit requires strict majority for even-sized clusters":
+    ## Regression: the commit quorum used (N+1) div 2, which for a 4-node
+    ## cluster commits at 2/4 (a minority). Strict majority is N div 2 + 1.
+    var node = newRaftNode("leader", @["p1", "p2", "p3"])
+    node.currentTerm = 5
+    node.state = rsLeader
+    let e = node.appendLog("put", cast[seq[byte]]("k\x00v"))
+    check e.index == 1
+    check e.term == 5
+    node.nextIndex["p1"] = 2
+    node.nextIndex["p2"] = 2
+    node.nextIndex["p3"] = 2
+    # Leader + 1 peer (count=2) is NOT a majority of 4.
+    node.handleAppendReply("p1", RaftMessage(
+      kind: rmkAppendEntriesReply, term: 5, senderId: "p1",
+      success: true, matchIdx: 1))
+    check node.commitIndex == 0
+    # Leader + 2 peers (count=3) IS a strict majority of 4 -> commits.
+    node.handleAppendReply("p2", RaftMessage(
+      kind: rmkAppendEntriesReply, term: 5, senderId: "p2",
+      success: true, matchIdx: 1))
+    check node.commitIndex == 1
+
   test "InstallSnapshotReply success advances match/next index and clears streak":
     var node = newRaftNode("leader", @["peer-1"])
     node.currentTerm = 5
@@ -3021,7 +3044,14 @@ suite "Raft InstallSnapshot Send":
                                       bt: uint64): bool {.gcsafe.} =
         gotBaseIndex = bi
         gotBaseTerm = bt
-        result = readFile(p) == payload
+        # sendSnapshot now gzips the tar off the event loop, so the assembled
+        # archive is gzip-compressed; decompress before comparing the bytes.
+        let raw = p & ".raw"
+        defer:
+          if fileExists(raw): removeFile(raw)
+        if not gunzipFile(p, raw):
+          return false
+        result = readFile(raw) == payload
 
       let netL = newRaftNetwork(leader)
       let netF = newRaftNetwork(follower)
@@ -3049,8 +3079,10 @@ suite "Raft InstallSnapshot Send":
       check follower.lastSnapshotTerm == 4
       check gotBaseIndex == 100
       check gotBaseTerm == 4
-      # Temp archive cleaned up after the transfer
+      # Temp archives (uncompressed tar + compressed .tar.gz) cleaned up after
+      # the transfer
       check not fileExists(tmp / "raft-l" / "snap_out_100.tar.gz")
+      check not fileExists(tmp / "raft-l" / "snap_out_100.tar")
     scenario()
 
   test "sendSnapshot single-flight guard skips a concurrent send":
@@ -3441,6 +3473,43 @@ suite "Raft SQL Write Path":
     let (found, _) = db.get("t.id=1")
     check not found
 
+  test "REP receiver chain (encode -> decode -> apply) maintains indexes":
+    ## Mirrors server.nim's legacy REP handler: decodeRepPayload decides the op,
+    ## then applyReplicatedPut/Delete keep secondary indexes consistent. Guards
+    ## the wiring the receiver relies on — an indexed put must populate the
+    ## B-tree and a PK-only put (empty value) must apply as a put, not vanish.
+    var testDir = getTempDir() / "baradb_rep_recv_idx_" & $getCurrentProcessId() & "_" & $getMonoTime().ticks
+    createDir(testDir)
+    defer: removeDir(testDir)
+    var db = newLSMTree(testDir)
+    var ctx = qexec.newExecutionContext(db)
+    discard qexec.executeQuery(ctx, parse(
+      "CREATE TABLE t (id INT PRIMARY KEY, name STRING)"))
+    discard qexec.executeQuery(ctx, parse(
+      "CREATE INDEX idx_name ON t (name)"))
+    # Leader ships an indexed put; the receiver decodes and applies it.
+    let put = decodeRepPayload(
+      encodeRepPayload(false, "t.id=1", cast[seq[byte]]("name=alice")))
+    check put.op == ropPut
+    if put.op == ropPut:
+      applyReplicatedPut(ctx, put.key, put.value)
+    check ctx.btrees["t.name"].get("alice").len >= 1
+    # A PK-only put (empty value) must apply as a put, not a delete.
+    let pk = decodeRepPayload(encodeRepPayload(false, "t.id=2", @[]))
+    check pk.op == ropPut
+    if pk.op == ropPut:
+      applyReplicatedPut(ctx, pk.key, pk.value)
+    let (foundPk, _) = db.get("t.id=2")
+    check foundPk
+    # Leader ships a delete; the receiver drops the row and the index entry.
+    let del = decodeRepPayload(encodeRepPayload(true, "t.id=1", @[]))
+    check del.op == ropDelete
+    if del.op == ropDelete:
+      applyReplicatedDelete(ctx, del.key)
+    check ctx.btrees["t.name"].get("alice").len == 0
+    let (foundDel, _) = db.get("t.id=1")
+    check not foundDel
+
   test "applyReplicatedPut updates in-memory graphs":
     var testDir = getTempDir() / "baradb_raft_apply_g_" & $getCurrentProcessId() & "_" & $getMonoTime().ticks
     createDir(testDir)