From ccc54e8f18d050689de18659e80a577c3ffb657c Mon Sep 17 00:00:00 2001 From: dimgigov Date: Sun, 2 Aug 2026 22:49:30 +0300 Subject: [PATCH] =?UTF-8?q?fix:=20stabilization=20session=20=E2=80=94=20au?= =?UTF-8?q?th=20bypass,=20raft=20quorum,=20wire=20DoS,=20query=20operators?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 81 +++++++++++++++++++++++++++++++ CHANGELOG.md | 22 +++++++++ PLAN.md | 25 +++++++++- docs/en/known-limitations.md | 4 +- src/barabadb/core/backup.nim | 76 +++++++++++++++++++++++++++++ src/barabadb/core/raft.nim | 35 +++++++++---- src/barabadb/core/replication.nim | 35 +++++++++++++ src/barabadb/core/server.nim | 50 +++++++++++++------ src/barabadb/query/exec/eval.nim | 7 +-- src/barabadb/query/exec/lower.nim | 2 + src/baradadb.nim | 15 +++--- tests/bugfix_test.nim | 73 ++++++++++++++++++++++++++++ tests/test_all.nim | 73 +++++++++++++++++++++++++++- 13 files changed, 458 insertions(+), 40 deletions(-) create mode 100644 BUG_AUDIT_2026-08.md diff --git a/BUG_AUDIT_2026-08.md b/BUG_AUDIT_2026-08.md new file mode 100644 index 0000000..eadbf3e --- /dev/null +++ b/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 ` не се изпълняват** — 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.. 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`.* diff --git a/CHANGELOG.md b/CHANGELOG.md index a601d75..dafab0c 100644 --- a/CHANGELOG.md +++ b/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) diff --git a/PLAN.md b/PLAN.md index 68b3028..bf5d4d4 100644 --- a/PLAN.md +++ b/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* diff --git a/docs/en/known-limitations.md b/docs/en/known-limitations.md index 47597c4..8e43c74 100644 --- a/docs/en/known-limitations.md +++ b/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 diff --git a/src/barabadb/core/backup.nim b/src/barabadb/core/backup.nim index 6d13d23..e8b54b5 100644 --- a/src/barabadb/core/backup.nim +++ b/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. diff --git a/src/barabadb/core/raft.nim b/src/barabadb/core/raft.nim index bf00b3c..5543872 100644 --- a/src/barabadb/core/raft.nim +++ b/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 - ## dir into destPath (wired in baradadb.nim via backupDataDir). + ## Leader InstallSnapshot send. buildSnapshot writes an UNCOMPRESSED tar + ## 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 - let majority = (node.peers.len + 1 + 1) div 2 # majority of cluster (peers + leader) + # Update commit index using strict majority — the same form as the election + # 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 - ## (tar+gzip). Snapshot sends are rare, so we accept the stall rather than - ## adding a worker round-trip (same trade-off as restoreSnapshot). + ## Runs on the raft event loop. buildSnapshot performs the tar on the loop + ## under the storage gate (consistent capture); the CPU-heavy gzip then runs + ## 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) diff --git a/src/barabadb/core/replication.nim b/src/barabadb/core/replication.nim index 1d30702..cf813ca 100644 --- a/src/barabadb/core/replication.nim +++ b/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 \n" diff --git a/src/barabadb/core/server.nim b/src/barabadb/core/server.nim index d268eaa..e1cb551 100644 --- a/src/barabadb/core/server.nim +++ b/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 - # on the receiver. Deletes ship an empty value as before. - let value = if pair.deleted: @[] else: pair.value - var data = newSeq[byte](pair.key.len + 1 + value.len) - for i, c in pair.key: data[i] = byte(c) - data[pair.key.len] = byte(0) - for i, c in value: data[pair.key.len + 1 + i] = c - discard replication.writeLsn(data) + # Legacy REP wire format: explicit 'P'/'D' op tag (see + # encodeRepPayload). The tag — not an empty value — distinguishes + # a put from a delete, so PK-only rows (empty value) replicate as + # puts instead of vanishing as deletes. + discard replication.writeLsn( + encodeRepPayload(pair.deleted, pair.key, pair.value)) 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') - if nullPos >= 0: - let key = data[0.. 0: - server.db.put(key, stringToBytes(value)) - else: - server.db.delete(key) + # Op tag — not value length — decides put vs delete, so a PK-only + # put (empty value) is applied as a put and the row survives. + let decoded = decodeRepPayload(cast[seq[byte]](data)) + case decoded.op + of ropPut, ropDelete: + # Apply through applyReplicatedPut/Delete (not raw db.put/delete) + # so secondary B-tree/FTS/HNSW/graph indexes stay consistent on + # 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) diff --git a/src/barabadb/query/exec/eval.nim b/src/barabadb/query/exec/eval.nim index 4f15125..82134bc 100644 --- a/src/barabadb/query/exec/eval.nim +++ b/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" - # Try numeric comparison + # Numeric-first so `!=` is the exact complement of `=` (irEq): string + # 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" diff --git a/src/barabadb/query/exec/lower.nim b/src/barabadb/query/exec/lower.nim index 2b84a98..735a2a0 100644 --- a/src/barabadb/query/exec/lower.nim +++ b/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) diff --git a/src/baradadb.nim b/src/baradadb.nim index e15d191..a7fbe15 100644 --- a/src/baradadb.nim +++ b/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 - # into the path raft picks (dataDir/raft/snap_out_.tar.gz). Like - # restoreSnapshot this runs on the raft event loop and blocks on disk I/O - # (tar+gzip); snapshot sends are rare, so we accept the stall. + # Leader InstallSnapshot send: tar the default DB's data directory into the + # path raft picks (dataDir/raft/snap_out_.tar). The tar runs here on + # the raft event loop under the storage gate; sendSnapshot then gzips it on + # 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 diff --git a/tests/bugfix_test.nim b/tests/bugfix_test.nim index 0b12970..a18d27f 100644 --- a/tests/bugfix_test.nim +++ b/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) diff --git a/tests/test_all.nim b/tests/test_all.nim index 738acc2..0f829dc 100644 --- a/tests/test_all.nim +++ b/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)