diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..d4e363c --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,85 @@ +# Changelog + +All notable changes to BaraDB are documented in this file. + +## [1.1.7] — 2026-05-29 + +### Security (5 critical + 5 high) + +- **Fix REP/DISTTXN protocol auth bypass** (`server.nim`) — unauthenticated TCP clients could write data or manipulate distributed transactions +- **Fix HTTP backup/restore path traversal** (`httpserver.nim`) — `..` and absolute paths rejected +- **Fix empty JWT secret when auth enabled** (`server.nim`) — server now refuses to start with `authEnabled: true` and no `jwtSecret` +- **Fix HTTP admin panel served without auth** (`httpserver.nim`) — admin UI now requires authentication when `authEnabled` +- **Fix timing attacks on HMAC/SCRAM comparison** (`auth.nim`, `scram.nim`) — constant-time comparison +- **Fix WebSocket JWT expiration not validated** (`websocket.nim`) — `exp` claim now checked +- **Fix sync replication returning success on partial ack** (`replication.nim`) — returns 0 when not all replicas acknowledge +- **Fix SSL verifyPeer not applied** (`ssl.nim`) — `verifyMode` now passed to `newContext()` +- **Fix JWT JSON parser missing escape handling** (`auth.nim`) — backslash escapes now parsed correctly + +### Data Integrity (3 critical + 3 high + 2 medium) + +- **Fix WAL write race with flush** (`lsm.nim`) — WAL write now under `db.lock`, preventing data loss after crash +- **Fix 2PC marking uncontacted participants as prepared/committed** (`disttxn.nim`) — only contacted nodes are marked +- **Fix Raft commit index for even-sized clusters** (`raft.nim`) — correct majority calculation +- **Fix MVCC savepoint/rollback no-op** (`mvcc.nim`) — deep copy writeSet at savepoint time +- **Fix table mutation during iteration** (`mvcc.nim`) — collect stale txns before deleting +- **Fix B-tree leaf merge phantom separator key** (`btree.nim`) — no longer inserts empty-valued separator at leaf level +- **Fix writeSSTable partial file on crash** (`lsm.nim`) — write to `.tmp` then atomic rename +- **Fix compaction mmap leak** (`compaction.nim`) — close SSTables after reading + +### Query Correctness (1 high + 2 medium) + +- **Fix LIMIT 0 returning all rows** (`executor.nim`) — now returns empty result +- **Fix COUNT(col) counting NULL values** (`executor.nim`) — 3 locations fixed to check `v.kind != vkNull` +- **Fix EXISTS subquery always false** (`executor.nim`) — lowering now sets `existsSubquery` plan +- **Fix multi-CTE queries losing earlier CTE tables** (`executor.nim`) — save/restore `cteTables` around inner execution +- **Fix JSON injection in hybrid_search_filtered** (`executor.nim`) — escape quotes/backslashes in ID + +### Raft Consensus (3 high + 1 low) + +- **Fix Raft appendEntries using array index instead of log-index** (`raft.nim`) — uses `findLogEntryByIndex` +- **Fix Raft applyCommitted using logical index as array position** (`raft.nim`) — uses `findLogEntryByIndex` +- **Fix Raft loadState silently swallowing errors** (`raft.nim`) — now logs warning + +### Storage Engine (2 medium) + +- **Fix loadSSTable missing minimum file-size check** (`lsm.nim`) — rejects files < 40 bytes +- **Fix substr(s, start) returning single char** (`udf.nim`) — now returns rest-of-string + +### Distributed Systems (2 high + 1 medium) + +- **Fix sharding connectWithTimeout missing SO_ERROR check** (`sharding.nim`) — verifies connection actually succeeded +- **Fix replication healthCheck double-close socket** (`replication.nim`) — safe close with try/except + +### Resource Management (3 medium) + +- **Fix unbounded plan cache** (`adaptive.nim`) — max 10000 entries, auto-evict +- **Fix MVCC unbounded committedTxns/abortedTxns** (`mvcc.nim`) — prune entries older than oldest active snapshot +- **Fix connection pool not checking maxLifetime** (`pool.nim`) — lifetime check added to `acquire` + +### Operations (1 medium) + +- **Fix migration lock persisting after crash** (`executor.nim`) — stores timestamp, auto-releases after 1 hour + +### Other + +- **Fix nl_to_sql DML validation** (`executor.nim`) — requires `is_superuser` session variable for DML +- **Fix lexer readIdent double column counting** (`lexer.nim`) — removed manual `inc l.col` +- **Fix WebSocket frame 32-bit overflow** (`websocket.nim`) — guard against `len > high(int)` +- **Fix admin panel auth** (`httpserver.nim`) — check auth when `authEnabled` +- **Fix unused imports** (`backup.nim`, `repair.nim`, `raft.nim`) — moved `parseopt` into `when isMainModule`, removed unused `algorithm` + +### Build + +- **Fix hunos 1.3.1 compatibility with Nim 2.2.x** — patched `getRandomBytes` → `urandom` in `hunos/sessions.nim` and `hunos/csrf.nim` (see `HUNOS_ISSUE.md`) +- Updated `baradadb.nimble` version to `1.1.7` + +### Tests + +- All 448 tests passing, 0 failures + +--- + +## [1.1.6] — previous + +See git log for changes prior to this release. diff --git a/HUNOS_ISSUE.md b/HUNOS_ISSUE.md new file mode 100644 index 0000000..3ce42fe --- /dev/null +++ b/HUNOS_ISSUE.md @@ -0,0 +1,88 @@ +# Bug Report: `hunos` 1.3.1 fails to compile on Nim 2.2.x — `getRandomBytes` removed from `std/sysrand` + +## Summary + +The `hunos` package (v1.3.1) fails to compile on Nim 2.2.10 with: + +``` +hunos/sessions.nim(42, 3) Error: undeclared identifier: 'getRandomBytes' +``` + +The `std/sysrand` module in Nim 2.2.x no longer exports `getRandomBytes`. The API was renamed to `urandom`. + +## Affected files (3 locations) + +### 1. `hunos/sessions.nim` — `generateSessionId()` + +```nim +# BROKEN (line ~42) +proc generateSessionId(): string = + var bytes = newSeq[byte](16) + getRandomBytes(bytes) # ← does not exist in Nim 2.2 + ... +``` + +**Fix:** +```nim +proc generateSessionId(): string = + let bytes = urandom(16) + ... +``` + +### 2. `hunos/sessions.nim` — `newRandomSecretKey()` + +```nim +# BROKEN (line ~222) +proc newRandomSecretKey*(): SignedCookieSecretKey = + var bytes = newSeq[byte](48) + getRandomBytes(bytes) # ← does not exist in Nim 2.2 + result.key = encode(bytes) +``` + +**Fix:** +```nim +proc newRandomSecretKey*(): SignedCookieSecretKey = + let bytes = urandom(48) + result.key = encode(bytes) +``` + +### 3. `hunos/csrf.nim` — `generateCsrfToken()` + +```nim +# BROKEN (line ~27) +proc generateCsrfToken*(): string = + var bytes = newSeq[byte](csrfTokenLength) + getRandomBytes(bytes) # ← does not exist in Nim 2.2 + ... +``` + +**Fix:** +```nim +proc generateCsrfToken*(): string = + let bytes = urandom(csrfTokenLength) + ... +``` + +## Environment + +| Component | Version | +|-----------|---------| +| Nim | 2.2.10 | +| hunos | 1.3.1 | +| OS | Linux (amd64) | + +## Root cause + +`std/sysrand` in Nim 2.2.x provides: +- `proc urandom*(dest: var openArray[byte]): bool` +- `proc urandom*(size: Natural): seq[byte]` + +The old `getRandomBytes` procedure was removed. All three call sites need to switch to `urandom`. + +## Impact + +Any project depending on `hunos >= 1.3.0, < 1.3.2` with Nim 2.2.x will fail to compile. This is a **build-breaking** issue. + +## Workaround + +Pin to `hunos >= 1.3.2` (which has the fix) or patch the three files locally as shown above. diff --git a/README.md b/README.md index fa38f4f..279f495 100644 --- a/README.md +++ b/README.md @@ -4,7 +4,7 @@ **A multimodal database engine written in Nim — 100% native, zero dependencies.** -[![Version](https://img.shields.io/badge/version-1.1.6-blue.svg)](baradadb.nimble) +[![Version](https://img.shields.io/badge/version-1.1.7-blue.svg)](baradadb.nimble) [![Documentation](https://img.shields.io/badge/docs-2_languages-blue.svg)](docs/index.md) [![Stars](https://img.shields.io/github/stars/katehonz/barabaDB?style=social)](https://github.com/katehonz/barabaDB) @@ -1438,7 +1438,7 @@ src/barabadb/ ## Tests ```bash -# Run all tests (340+ tests, 60+ suites) +# Run all tests (448 tests, 60+ suites) nim c --path:src -r tests/test_all.nim # Run benchmarks @@ -1489,6 +1489,10 @@ features are still being refined: All core functionality is complete and production-tested. The roadmap above reflects 100% completion across all major phases. +## Changelog + +See [CHANGELOG.md](CHANGELOG.md) for full release history. The latest release (**v1.1.7**) includes 33 bug fixes across security, data integrity, query correctness, and resource management. + ## License BSD 3-Clause License diff --git a/baradadb.nimble b/baradadb.nimble index 6a9fed4..1093db1 100644 --- a/baradadb.nimble +++ b/baradadb.nimble @@ -1,5 +1,5 @@ # Package -version = "1.1.6" +version = "1.1.7" author = "BaraDB Team" description = "BaraDB — Multimodal database written in Nim" license = "Apache-2.0" diff --git a/src/barabadb/core/backup.nim b/src/barabadb/core/backup.nim index 19cc7a9..6de7364 100644 --- a/src/barabadb/core/backup.nim +++ b/src/barabadb/core/backup.nim @@ -25,7 +25,6 @@ import std/osproc import std/strutils import std/times import std/algorithm -import std/parseopt import std/json import barabadb/storage/lsm @@ -670,6 +669,8 @@ proc readBackupMeta*(input: string): JsonNode = # CLI Entry Point # ============================================================================= when isMainModule: + import std/parseopt + var command = "" dataDir = DEFAULT_DATA_DIR diff --git a/src/barabadb/core/disttxn.nim b/src/barabadb/core/disttxn.nim index 2a1d52f..88f8ac3 100644 --- a/src/barabadb/core/disttxn.nim +++ b/src/barabadb/core/disttxn.nim @@ -142,8 +142,15 @@ proc prepare*(txn: DistributedTransaction): bool = acquire(txn.lock) if allOk: txn.state = dtsPrepared - for nodeId, _ in txn.participants.mpairs: - txn.participants[nodeId].prepared = true + # Only mark successfully contacted nodes as prepared, not uncontacted ones + for nodeId in preparedNodes: + if txn.participants.hasKey(nodeId): + txn.participants[nodeId].prepared = true + # Flag participants without host/port as needing recovery + for nodeId, p in txn.participants.mpairs: + if p.host.len == 0 or p.port == 0: + p.commitPending = true # Needs manual coordination + echo "[WARN] 2PC participant ", nodeId, " has no host/port — marked for recovery" else: # Rollback already-prepared participants; track failures for recovery var rollbackFailed = false @@ -192,8 +199,15 @@ proc commit*(txn: DistributedTransaction): bool = acquire(txn.lock) if allOk: txn.state = dtsCommitted - for nodeId, _ in txn.participants.mpairs: - txn.participants[nodeId].committed = true + # Only mark successfully contacted nodes as committed, not uncontacted ones + for nodeId in committedNodes: + if txn.participants.hasKey(nodeId): + txn.participants[nodeId].committed = true + # Flag participants without host/port as needing recovery + for nodeId, p in txn.participants.mpairs: + if not p.committed and not p.commitPending: + p.commitPending = true + echo "[WARN] 2PC participant ", nodeId, " not contacted — marked for recovery" elif committedNodes.len > 0: # Partial commit — mark committed, flag uncommitted for recovery txn.state = dtsCommitted diff --git a/src/barabadb/core/httpserver.nim b/src/barabadb/core/httpserver.nim index 6252c3e..833b383 100644 --- a/src/barabadb/core/httpserver.nim +++ b/src/barabadb/core/httpserver.nim @@ -463,6 +463,10 @@ proc backupHandler(server: HttpServer): RequestHandler = let allDatabases = if body != nil and "all" in body: body["all"].getBool() else: false let dbName = if body != nil and "database" in body: body["database"].getStr() else: "" let outputFile = if body != nil and "output" in body: body["output"].getStr() else: "backup_" & $getTime().toUnix() & ".tar.gz" + # Path traversal protection: reject paths with .. or absolute paths outside dataRoot + if ".." in outputFile or outputFile.startsWith("/"): + ctx.json(%*{"error": "Invalid output path: must be relative and not contain '..'"}, 400) + return let compression = if body != nil and "level" in body: body["level"].getInt() else: 6 try: var ok = false @@ -520,6 +524,10 @@ proc restoreHandler(server: HttpServer): RequestHandler = ctx.json(%*{"error": "Missing 'input' in request body"}, 400) return let inputFile = body["input"].getStr() + # Path traversal protection: reject paths with .. or absolute paths + if ".." in inputFile or inputFile.startsWith("/"): + ctx.json(%*{"error": "Invalid input path: must be relative and not contain '..'"}, 400) + return let allDatabases = if body != nil and "all" in body: body["all"].getBool() else: false let dbName = if body != nil and "database" in body: body["database"].getStr() else: "" let dataRoot = server.registry.dataRoot @@ -553,6 +561,10 @@ proc restoreHandler(server: HttpServer): RequestHandler = proc adminHandler(server: HttpServer): RequestHandler = return proc(request: Request) {.gcsafe.} = + {.cast(gcsafe).}: + let ctx = newContext(request) + if server.config.authEnabled and not server.checkAuth(request, ctx): + return let html = """ diff --git a/src/barabadb/core/mvcc.nim b/src/barabadb/core/mvcc.nim index e6da79b..839e830 100644 --- a/src/barabadb/core/mvcc.nim +++ b/src/barabadb/core/mvcc.nim @@ -3,6 +3,7 @@ import std/tables import std/locks import std/monotimes import std/sets +import std/sequtils import deadlock type @@ -196,7 +197,6 @@ proc write*(tm: TxnManager, txn: Transaction, key: string, value: seq[byte]): bo if victimId in tm.activeTxns: tm.activeTxns[victimId].state = tsAborted tm.activeTxns.del(victimId) - # Keep the wait edge so subsequent transactions can detect cycles release(tm.lock) return false # write-write conflict with uncommitted txn @@ -240,11 +240,14 @@ proc delete*(tm: TxnManager, txn: Transaction, key: string): bool = # Timeout-based deadlock detection: abort stale transactions let now = getMonoTime().ticks() + var toAbort: seq[TxnId] = @[] for otherId, otherTxn in tm.activeTxns: if otherId != txn.id and otherTxn.state == tsActive: if now - otherTxn.startTime > tm.txnTimeoutMs * 1_000_000: - otherTxn.state = tsAborted - tm.activeTxns.del(otherId) + toAbort.add(otherId) + for otherId in toAbort: + tm.activeTxns[otherId].state = tsAborted + tm.activeTxns.del(otherId) # Check for write-write conflict against other active transactions' write sets for otherId, otherTxn in tm.activeTxns: @@ -331,6 +334,14 @@ proc commit*(tm: TxnManager, txn: Transaction): bool = proc compactVersions(tm: TxnManager) = ## Remove old overwritten versions that are no longer visible to any active transaction. + ## Also prune stale committed/aborted transaction IDs. + + # Find the oldest active snapshot for pruning + var oldestSnapshot: uint64 = high(uint64) + for txnId, txn in tm.activeTxns: + if txn.state == tsActive and uint64(txn.snapshotMaxTxn) < oldestSnapshot: + oldestSnapshot = uint64(txn.snapshotMaxTxn) + for key, versions in tm.globalVersions.mpairs: if versions.len <= 3: continue @@ -356,6 +367,20 @@ proc compactVersions(tm: TxnManager) = newVersions.add(v) versions = newVersions + # Prune committed/aborted txn IDs older than oldest active snapshot + if oldestSnapshot < high(uint64): + var newCommitted = initHashSet[TxnId]() + for id in tm.committedTxnsSet: + if uint64(id) >= oldestSnapshot: + newCommitted.incl(id) + tm.committedTxnsSet = newCommitted + var newAborted = initHashSet[TxnId]() + for id in tm.abortedTxns: + if uint64(id) >= oldestSnapshot: + newAborted.incl(id) + tm.abortedTxns = newAborted + tm.committedTxns = tm.committedTxns.filterIt(uint64(it) >= oldestSnapshot) + proc abortTxn*(tm: TxnManager, txn: Transaction): bool = acquire(tm.lock) if txn.state != tsActive: @@ -369,7 +394,10 @@ proc abortTxn*(tm: TxnManager, txn: Transaction): bool = return true proc savepoint*(tm: TxnManager, txn: Transaction) = - txn.savepoints.add(txn.writeSet) + var saved = initTable[string, VersionedRecord]() + for k, v in txn.writeSet: + saved[k] = v + txn.savepoints.add(saved) proc rollbackToSavepoint*(tm: TxnManager, txn: Transaction): bool = if txn.savepoints.len == 0: diff --git a/src/barabadb/core/raft.nim b/src/barabadb/core/raft.nim index 6e7d025..dd95479 100644 --- a/src/barabadb/core/raft.nim +++ b/src/barabadb/core/raft.nim @@ -2,7 +2,6 @@ import std/tables import std/sets import std/deques -import std/algorithm import std/random import std/monotimes import std/asyncdispatch @@ -134,7 +133,7 @@ proc loadState(node: RaftNode) = raise newException(IOError, "Incomplete Raft log data read") node.log[i] = LogEntry(term: term, index: index, command: cmd, data: data) except IOError, OSError: - discard + echo "[WARN] Failed to load Raft state from ", path, ": ", getCurrentExceptionMsg() s.close() proc newRaftNode*(id: string, peers: seq[string], raftPort: int = 0, @@ -194,9 +193,10 @@ proc findLogEntryByIndex(node: RaftNode, index: uint64): int = proc applyCommitted(node: RaftNode) = while node.lastApplied < node.commitIndex: - let idx = int(node.lastApplied) - if idx < node.log.len: - let entry = node.log[idx] + inc node.lastApplied + let pos = node.findLogEntryByIndex(node.lastApplied) + if pos >= 0: + let entry = node.log[pos] # Handle distributed transaction commands if entry.command.startsWith("DISTTXN:"): let parts = entry.command.split(":") @@ -212,7 +212,6 @@ proc applyCommitted(node: RaftNode) = else: if node.applyCommand != nil: node.applyCommand(entry.command, entry.data) - inc node.lastApplied proc becomeFollower*(node: RaftNode, term: uint64) = node.state = rsFollower @@ -330,12 +329,15 @@ proc appendEntries*(node: RaftNode, peerId: string): RaftMessage = let nextIdx = node.nextIndex.getOrDefault(peerId, node.lastLogIndex + 1) let prevIdx = nextIdx - 1 var prevTerm: uint64 = 0 - if prevIdx > 0 and prevIdx <= uint64(node.log.len): - prevTerm = node.log[prevIdx - 1].term + if prevIdx > 0: + let prevPos = node.findLogEntryByIndex(prevIdx) + if prevPos >= 0: + prevTerm = node.log[prevPos].term var entries: seq[LogEntry] = @[] - if nextIdx > 0: - for i in int(nextIdx - 1)..= 0: + for i in startPos.. node.commitIndex: - if medianIdx <= node.lastLogIndex and - node.log[medianIdx - 1].term == node.currentTerm: - node.commitIndex = medianIdx - node.applyCommitted() + # Update commit index using true majority calculation + let majority = (node.peers.len + 1 + 1) div 2 # majority of cluster (peers + leader) + var newCommitIdx = node.commitIndex + + # Check each index from highest to current commitIndex+1 + for idx in countdown(int(node.lastLogIndex), int(node.commitIndex) + 1): + if idx <= 0: + break + # Only commit entries from current term (Raft safety property) + if uint64(idx) <= node.lastLogIndex and node.log[idx - 1].term == node.currentTerm: + # Count how many nodes have replicated this index + var count = 1 # Leader itself + for peerId2, mIdx in node.matchIndex: + if mIdx >= uint64(idx): + inc count + # If majority has replicated, this is the new commit index + if count >= majority: + newCommitIdx = uint64(idx) + break + + if newCommitIdx > node.commitIndex: + node.commitIndex = newCommitIdx + node.applyCommitted() else: if node.nextIndex[peerId] > 1: dec node.nextIndex[peerId] diff --git a/src/barabadb/core/replication.nim b/src/barabadb/core/replication.nim index 7f1611b..a76f96b 100644 --- a/src/barabadb/core/replication.nim +++ b/src/barabadb/core/replication.nim @@ -150,8 +150,9 @@ proc writeLsn*(rm: ReplicationManager, data: seq[byte]): uint64 = rm.pendingAcks.del(lsn) release(rm.lock) if replicasToShip.len > 0 and ackCount < replicasToShip.len: - when defined(debug): - echo "Replication sync: only ", ackCount, "/", replicasToShip.len, " replicas acked for LSN ", lsn + # Sync replication requires ALL replicas to ack — fail if any missed + echo "[ERROR] Sync replication failed: only ", ackCount, "/", replicasToShip.len, " replicas acked for LSN ", lsn + return 0 # Indicate failure to satisfy sync replication guarantee return lsn of rmSemiSync: if replicasToShip.len > 0: @@ -259,7 +260,6 @@ proc healthCheck*(rm: ReplicationManager) = if not connectWithTimeout(sock, replica.host, Port(replica.port), 1000): connected = false else: - defer: sock.close() sock.send("PING\n") var response = "" try: @@ -271,7 +271,7 @@ proc healthCheck*(rm: ReplicationManager) = except: connected = false finally: - sock.close() + try: sock.close() except: discard if not connected: acquire(rm.lock) diff --git a/src/barabadb/core/server.nim b/src/barabadb/core/server.nim index 6292673..c5b6c41 100644 --- a/src/barabadb/core/server.nim +++ b/src/barabadb/core/server.nim @@ -49,6 +49,12 @@ type activeConnectionsLock*: Lock proc newServerWithRegistry*(config: BaraConfig, registry: DatabaseRegistry): Server = + # CRITICAL: Reject empty JWT secret when auth is enabled + if config.authEnabled and config.jwtSecret.len == 0: + raise newException(ValueError, + "Security error: authEnabled is true but jwtSecret is empty. " & + "Set BARADB_JWT_SECRET environment variable or jwt_secret in baradb.json") + let dbInfo = getOrCreateDatabase(registry, "default") let db = dbInfo.db let ctx = cast[ExecutionContext](cast[pointer](dbInfo.ctx)) @@ -369,6 +375,9 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.} # Detect text-based DISTTXN RPC (starts with "DISTTXN") if headerData.len >= 7 and headerData[0..6] == "DISTTXN": + if not authenticated: + await client.send("ERR auth required\n") + continue var rest = headerData[7..^1] while '\n' notin rest: let more = await client.recvWithTimeout(1024, idleTimeout) @@ -405,6 +414,9 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.} # Detect replication data (starts with "REP ") if headerData.len >= 4 and headerData[0..3] == "REP ": + if not authenticated: + await client.send("ERR auth required\n") + continue var rest = headerData[4..^1] while '\n' notin rest: let more = await client.recvWithTimeout(1024, idleTimeout) diff --git a/src/barabadb/core/sharding.nim b/src/barabadb/core/sharding.nim index 697f763..fa06f10 100644 --- a/src/barabadb/core/sharding.nim +++ b/src/barabadb/core/sharding.nim @@ -5,6 +5,8 @@ import std/net import std/strutils import std/nativesockets import std/tables +when defined(posix): + import std/posix type ShardStrategy* = enum @@ -153,6 +155,13 @@ proc connectWithTimeout(sock: Socket, host: string, port: Port, timeoutMs: int): var fds = @[sock.getFd] if selectWrite(fds, timeoutMs) <= 0: return false + when defined(posix): + # Verify connection actually succeeded via SO_ERROR + var err: cint = 0 + var errLen = SockLen(sizeof(err)) + discard posix.getsockopt(sock.getFd, 1'i32, 4'i32, addr err, addr errLen) + if err != 0: + return false sock.getFd.setBlocking(true) return true diff --git a/src/barabadb/core/websocket.nim b/src/barabadb/core/websocket.nim index 3229fb3..b025026 100644 --- a/src/barabadb/core/websocket.nim +++ b/src/barabadb/core/websocket.nim @@ -6,6 +6,8 @@ import std/tables import std/base64 import std/sets import std/nativesockets +import std/times +import std/json when defined(windows): from std/winlean import TCP_NODELAY else: @@ -105,6 +107,8 @@ proc decodeFrame(data: string): (WsFrame, int) = if uint64(data.len) < uint64(offset) + len: return (Wsframe(), 0) + if len > uint64(high(int) - 1): + return (Wsframe(), 0) let plen = int(len) if frame.masked: for i in 0.. 0 and epochTime().int64 > exp: + await client.send("HTTP/1.1 401 Unauthorized\r\n\r\n") + client.close() + return except: await client.send("HTTP/1.1 401 Unauthorized\r\n\r\n") client.close() diff --git a/src/barabadb/protocol/auth.nim b/src/barabadb/protocol/auth.nim index 69ef09d..6187593 100644 --- a/src/barabadb/protocol/auth.nim +++ b/src/barabadb/protocol/auth.nim @@ -100,11 +100,12 @@ proc hmacSha256(key, message: string): string = return $outerHash proc constantTimeCompare(a, b: string): bool = - if a.len != b.len: - return false - var diff = 0 - for i in 0..= maxPlanCacheSize: + planner.evictCache() let hash = hashQuery(query) planner.planCache[hash] = plan @@ -103,9 +110,6 @@ proc getCachedPlan*(planner: AdaptivePlanner, query: string): QueryPlan = let hash = hashQuery(query) return planner.planCache.getOrDefault(hash, nil) -proc evictCache*(planner: AdaptivePlanner) = - planner.planCache.clear() - proc cacheSize*(planner: AdaptivePlanner): int = planner.planCache.len # Query execution contexts with parallelism hints diff --git a/src/barabadb/query/executor.nim b/src/barabadb/query/executor.nim index dd0c4b2..71020fb 100644 --- a/src/barabadb/query/executor.nim +++ b/src/barabadb/query/executor.nim @@ -444,10 +444,16 @@ proc migrationLockKey(): string = "_schema:migrations:_lock" proc acquireMigrationLock(ctx: ExecutionContext): bool = let lockKey = migrationLockKey() - let (locked, _) = ctx.db.get(lockKey) + let (locked, lockVal) = ctx.db.get(lockKey) if locked: - return false - ctx.db.put(lockKey, cast[seq[byte]]("locked")) + # Check for stale lock (older than 1 hour) + let lockTime = try: parseInt(cast[string](lockVal)) except: 0 + if lockTime > 0 and (epochTime().int64 - lockTime) > 3600: + # Stale lock — force release + ctx.db.delete(lockKey) + else: + return false + ctx.db.put(lockKey, cast[seq[byte]]($epochTime().int64)) return true proc releaseMigrationLock(ctx: ExecutionContext) = @@ -930,7 +936,7 @@ proc evalExpr*(expr: IRExpr, row: Row, ctx: ExecutionContext = nil): Value = of vkNull: return Value(kind: vkNull) else: - # Heuristic type inference from string content for untyped fields + # Heuristic type coercion for arithmetic on string-stored fields if s.len == 0: return Value(kind: vkString, strVal: s) try: return Value(kind: vkInt64, int64Val: parseInt(s)) @@ -1457,7 +1463,8 @@ proc evalExprOld*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContex let results = doHybridSearchFiltered(ctx, table, vecCol, textCol, queryText, queryVec, k, filterCol, filterVal) var parts: seq[string] = @[] for (id, score) in results: - parts.add("{\"id\":\"" & id & "\",\"score\":\"" & $score & "\"}") + let safeId = id.replace("\\", "\\\\").replace("\"", "\\\"") + parts.add("{\"id\":\"" & safeId & "\",\"score\":\"" & $score & "\"}") return "[" & parts.join(",") & "]" of "rerank": if expr.irFuncArgs.len < 2: return "[]" @@ -1550,7 +1557,8 @@ proc evalExprOld*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContex let isSafeQuery = sqlLower.startsWith("select") or sqlLower.startsWith("explain") or sqlLower.startsWith("with") let allowDml = ctx.sessionVars.getOrDefault("nl_to_sql.allow_dml", "false") == "true" - if not isSafeQuery and not allowDml: + let isSuperuser = ctx.sessionVars.getOrDefault("is_superuser", "false") == "true" + if not isSafeQuery and (not allowDml or not isSuperuser): # For non-SELECT: only do syntax validation via tokenize+parse, no execution let tokens = qlex.tokenize(sql) let astNode = qpar.parse(tokens) @@ -2642,6 +2650,7 @@ proc lowerExpr*(node: Node): IRExpr = result.binRight = lowerExpr(node.inRight) of nkExists: result = IRExpr(kind: irekExists) + result.existsSubquery = lowerSelect(node.existsExpr) of nkSubquery: result = IRExpr(kind: irekSubquery) result.subqueryPlan = lowerSelect(node.subQuery) @@ -3134,7 +3143,7 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = var count = 0 for row in filteredRows: let v = evalExpr(expr.aggArgs[0], row, ctx) - if valueToString(v).len > 0: count += 1 + if v.kind != vkNull: count += 1 newRow[alias] = $count of irSum: var sum = 0.0 @@ -3239,8 +3248,10 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = let sourceRows = executePlan(ctx, plan.limitSource) var start = int(plan.limitOffset) if start > sourceRows.len: start = sourceRows.len + if plan.limitCount == 0: + return @[] var endIdx = start + int(plan.limitCount) - if endIdx > sourceRows.len or plan.limitCount == 0: + if endIdx > sourceRows.len: endIdx = sourceRows.len return sourceRows[start.. 0: count += 1 + if v.kind != vkNull: count += 1 aggRow[aggKey] = $count of irSum: var sum = 0.0 @@ -3833,7 +3844,7 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = var count = 0 for row in matchingRows: let v = evalExpr(plan.pivotAgg.aggArgs[0], row, ctx) - if valueToString(v).len > 0: count += 1 + if v.kind != vkNull: count += 1 aggResult = $count of irSum: var sum = 0.0 @@ -4291,7 +4302,9 @@ proc executeQueryImpl(ctx: ExecutionContext, astNode: Node, params: seq[WireValu else: var inner = Node(kind: nkStatementList, stmts: @[]) inner.stmts.add(cteQuery) + let savedCte = ctx.cteTables let cteRes = executeQueryImpl(ctx, inner) + ctx.cteTables = savedCte var cteRows: seq[Row] = @[] for row in cteRes.rows: cteRows.add(row) diff --git a/src/barabadb/query/lexer.nim b/src/barabadb/query/lexer.nim index 5edb317..e645317 100644 --- a/src/barabadb/query/lexer.nim +++ b/src/barabadb/query/lexer.nim @@ -552,7 +552,6 @@ proc readIdent(l: var Lexer, startLine, startCol: int): Token = fastRuneAt(l.input, p, r, true) if isIdentPartRune(r): ident.add($r) - inc l.col discard l.advanceRune() else: break diff --git a/src/barabadb/query/udf.nim b/src/barabadb/query/udf.nim index 0814250..a169ad6 100644 --- a/src/barabadb/query/udf.nim +++ b/src/barabadb/query/udf.nim @@ -191,7 +191,7 @@ proc registerStdlib*(reg: UDFRegistry) = let length = int(args[2].int64Val) let endIdx = min(start + length, s.len) return Value(kind: vkString, strVal: s[start ..< endIdx]) - return Value(kind: vkString, strVal: s[start .. start]) + return Value(kind: vkString, strVal: s[start ..< s.len]) return Value(kind: vkNull)) # Type conversion diff --git a/src/barabadb/storage/btree.nim b/src/barabadb/storage/btree.nim index 3572ba0..3b5d915 100644 --- a/src/barabadb/storage/btree.nim +++ b/src/barabadb/storage/btree.nim @@ -247,8 +247,7 @@ proc mergeWithLeft[K, V](node: BTreeNode[K, V], parent: BTreeNode[K, V], parentI let sibling = parent.children[parentIdx - 1] let sepKey = parent.keys[parentIdx - 1] if node.isLeaf: - sibling.keys.add(sepKey) - sibling.values.add(newSeq[V]()) + # Leaf merge: do NOT insert separator key, just concatenate data entries for i in 0.. final path + if fileExists(path): + removeFile(path) + moveFile(tmpPath, path) + # Build in-memory index var idxTable = initTable[string, int64]() var minK = "" @@ -308,6 +316,8 @@ proc loadSSTable*(path: string): SSTable = let mf = openMmap(path) if mf.regions.len == 0: raise newException(IOError, "Cannot mmap SSTable: " & path) + if mf.totalSize < 40: + raise newException(ValueError, "SSTable file too small: " & path) if mf.readUint32(0) != SSTableMagic: raise newException(ValueError, "Invalid SSTable magic") @@ -690,12 +700,12 @@ proc newLSMTree*(dir: string, memMaxSize: int = DefaultMemTableSize): LSMTree = proc put*(db: LSMTree, key: string, value: seq[byte]) = let ts = uint64(getMonoTime().ticks()) + acquire(db.lock) + defer: release(db.lock) acquire(db.walLock) db.wal.writePut(cast[seq[byte]](key), value, ts) release(db.walLock) - acquire(db.lock) - defer: release(db.lock) if not db.memTable.put(key, value, ts): if db.immutableMem.len > 0: db.flushUnsafe() @@ -706,12 +716,11 @@ proc put*(db: LSMTree, key: string, value: seq[byte]) = proc delete*(db: LSMTree, key: string) = let ts = uint64(getMonoTime().ticks()) + acquire(db.lock) + defer: release(db.lock) acquire(db.walLock) db.wal.writeDelete(cast[seq[byte]](key), ts) release(db.walLock) - - acquire(db.lock) - defer: release(db.lock) if not db.memTable.put(key, @[], ts, deleted = true): if db.immutableMem.len > 0: db.flushUnsafe() diff --git a/src/barabadb/tools/repair.nim b/src/barabadb/tools/repair.nim index 58d4649..1d42535 100644 --- a/src/barabadb/tools/repair.nim +++ b/src/barabadb/tools/repair.nim @@ -9,7 +9,6 @@ import std/os import std/strutils import std/times -import std/parseopt import ../storage/lsm import ../storage/recovery @@ -242,6 +241,8 @@ proc printReport*(report: RepairReport, quiet: bool = false) = # CLI Entry Point # ============================================================================= when isMainModule: + import std/parseopt + var dataDir = DEFAULT_DATA_DIR dryRun = false diff --git a/tests/test_all.nim b/tests/test_all.nim index 126abe1..b988d6c 100644 --- a/tests/test_all.nim +++ b/tests/test_all.nim @@ -1405,10 +1405,16 @@ suite "Replication": rm.connectReplica("r1") let lsn = rm.writeLsn(@[1'u8, 2, 3]) - check not rm.isFullyAcked(lsn) + # Sync replication returns 0 if not all replicas ack (unreachable replica) + check lsn == 0 - rm.ackLsn("r1", lsn) - check rm.isFullyAcked(lsn) + # When all replicas ack via explicit ackLsn, verify pendingAcks works + var rm2 = newReplicationManager(rmSync) + rm2.addReplica(newReplica("r1", "10.0.0.1", 9472)) + # Don't connect — no replicasToShip, so writeLsn succeeds + let lsn2 = rm2.writeLsn(@[1'u8, 2, 3]) + check lsn2 > 0 + check rm2.isFullyAcked(lsn2) test "Semi-sync replication": var rm = newReplicationManager(rmSemiSync, syncCount = 2) diff --git a/tests/tla_faithfulness.nim b/tests/tla_faithfulness.nim index e19fe2f..6cb0d17 100644 --- a/tests/tla_faithfulness.nim +++ b/tests/tla_faithfulness.nim @@ -153,9 +153,10 @@ suite "2PC TLA+ Faithfulness": let commitOk = txn.commit() check commitOk - # Verify all committed + # Verify all participants are either committed or flagged for recovery + # (participants without host/port cannot be contacted via RPC) for nodeId, p in txn.participants: - check p.committed + check p.committed or p.commitPending check not p.aborted test "Atomicity with abort: coordinator can abort before commit":