From 19fa7606043ed014d1abc663266ed2773ba18699 Mon Sep 17 00:00:00 2001 From: dimgigov Date: Sun, 17 May 2026 11:57:55 +0300 Subject: [PATCH] =?UTF-8?q?Week=204:=20Production=20Hardening=20=E2=80=94?= =?UTF-8?q?=20Thread=20Safety,=20Fuzz=20Tests,=20Property-Based=20Tests,?= =?UTF-8?q?=20NimForum=20Smoke=20Test?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Thread Safety: - Add lock to BTreeIndex (insert/get/scan/remove/contains) - Add lock to InvertedIndex (addDocument/removeDocument/search/bm25/tfidf/fuzzy/regex) - Add ctxLock to ExecutionContext for DDL/schema metadata, autoIncCounters, sequences - Deep-copy sessionVars in cloneForConnection() to isolate per-connection state - Fix activeConnections check-then-act race with Lock Wire Protocol Hardening: - Fix RangeDefect in deserializeValue for invalid FieldKind bytes - Fix IndexDefect in fkFloat32/fkFloat64 by reusing readUint32/readUint64 - Fix int32/int64 cast from uint32/uint64 to use cast[] instead of constructor - Fix parseHeader to validate MsgKind instead of holey enum cast Property-Based Tests (tests/prop_test.nim): - 8 invariants for evalExprValue: literal types, commutativity, identity, double negation, NULL propagation Wire Protocol Fuzz (tests/fuzz_test.nim): - 7 fuzz tests: random bytes deserialization, truncated buffers, MsgKind casts, parseHeader logic, message roundtrips NimForum Adapter Smoke Test (tests/nimforum_smoke_test.nim): - Full TCP server lifecycle with adapter CRUD and parameterized queries Test Results: 423 tests pass, 0 failures --- src/barabadb/core/server.nim | 41 +++- src/barabadb/fts/engine.nim | 335 ++++++++++++++++++-------------- src/barabadb/protocol/wire.nim | 25 +-- src/barabadb/query/executor.nim | 100 +++++++--- src/barabadb/storage/btree.nim | 134 ++++++++----- tests/fuzz_test.nim | 87 +++++++++ tests/nimforum_smoke_test.nim | 69 +++++++ tests/prop_test.nim | 123 ++++++++++++ tests/test_all.nim | 3 + 9 files changed, 676 insertions(+), 241 deletions(-) create mode 100644 tests/fuzz_test.nim create mode 100644 tests/nimforum_smoke_test.nim create mode 100644 tests/prop_test.nim diff --git a/src/barabadb/core/server.nim b/src/barabadb/core/server.nim index 231569a..a461f21 100644 --- a/src/barabadb/core/server.nim +++ b/src/barabadb/core/server.nim @@ -7,6 +7,7 @@ import std/tables import std/os import std/endians import std/monotimes +import std/locks import config import logging import ../protocol/wire @@ -37,6 +38,7 @@ type gossipProtocol*: GossipProtocol tls*: TLSContext activeConnections*: int + activeConnectionsLock*: Lock proc newServer*(config: BaraConfig): Server = let dataDir = config.dataDir / "server" @@ -83,13 +85,14 @@ proc newServer*(config: BaraConfig): Server = gp.onSuspect = proc(nodeId: string) {.gcsafe.} = cm.onNodeSuspect(nodeId) - Server(config: config, running: false, db: db, ctx: ctx, + result = Server(config: config, running: false, db: db, ctx: ctx, txnManager: ctx.txnManager, distTxnManager: newDistTxnManager(), replicationManager: newReplicationManager(), shardRouter: shardRouter, clusterMembership: cm, gossipProtocol: gp, tls: tls) + initLock(result.activeConnectionsLock) # ---------------------------------------------------------------------- # Wire Protocol Helpers @@ -105,9 +108,13 @@ proc parseHeader(data: string): (bool, MessageHeader) = if data.len < 12: return (false, MessageHeader()) let rawKind = readUint32BE(data, 0) - {.push warning[HoleEnumConv]: off.} - let kind = MsgKind(rawKind) - {.pop.} + let kind = cast[MsgKind](rawKind) + case kind + of mkClientHandshake, mkQuery, mkQueryParams, mkExecute, mkBatch, mkTransaction, mkClose, mkPing, mkAuth, + mkServerHandshake, mkReady, mkData, mkComplete, mkError, mkAuthChallenge, mkAuthOk, mkSchemaChange, mkPong, mkTransactionState: + discard + else: + return (false, MessageHeader()) let length = readUint32BE(data, 4) let requestId = readUint32BE(data, 8) return (true, MessageHeader(kind: kind, length: length, requestId: requestId)) @@ -523,8 +530,12 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.} except Exception as e: errorMsg("Client " & $clientId & " error: " & e.msg) finally: - if server.activeConnections > 0: - dec server.activeConnections + acquire(server.activeConnectionsLock) + try: + if server.activeConnections > 0: + dec server.activeConnections + finally: + release(server.activeConnectionsLock) info("Client " & $clientId & " disconnected") client.close() @@ -547,7 +558,16 @@ proc run*(server: Server) {.async.} = info("BaraDB listening on " & server.config.address & ":" & $server.config.port) while server.running: let client = await sock.accept() - if server.config.maxConnections > 0 and server.activeConnections >= server.config.maxConnections: + acquire(server.activeConnectionsLock) + var shouldAccept = true + try: + if server.config.maxConnections > 0 and server.activeConnections >= server.config.maxConnections: + shouldAccept = false + else: + inc server.activeConnections + finally: + release(server.activeConnectionsLock) + if not shouldAccept: client.close() continue if server.tls != nil: @@ -555,10 +575,15 @@ proc run*(server: Server) {.async.} = server.tls.wrapServer(client) except Exception as e: errorMsg("TLS handshake failed: " & e.msg) + acquire(server.activeConnectionsLock) + try: + if server.activeConnections > 0: + dec server.activeConnections + finally: + release(server.activeConnectionsLock) client.close() continue inc clientId - inc server.activeConnections asyncCheck server.handleClient(client, clientId) proc stop*(server: Server) = diff --git a/src/barabadb/fts/engine.nim b/src/barabadb/fts/engine.nim index 003624f..c7acbad 100644 --- a/src/barabadb/fts/engine.nim +++ b/src/barabadb/fts/engine.nim @@ -4,6 +4,7 @@ import std/strutils import std/unicode import std/math import std/algorithm +import std/locks type TermFreq* = Table[string, int] @@ -20,6 +21,7 @@ type docCount*: int avgDocLen*: float64 totalTerms*: int + lock*: Lock SearchResult* = object docId*: uint64 @@ -111,59 +113,68 @@ proc tokenize*(text: string, config: TokenizerConfig = defaultTokenizerConfig()) result.add(token) proc newInvertedIndex*(): InvertedIndex = - InvertedIndex( + result = InvertedIndex( postings: initTable[string, seq[PostingEntry]](), docLengths: initTable[uint64, int](), docCount: 0, avgDocLen: 0.0, totalTerms: 0, ) + initLock(result.lock) proc addDocument*(idx: InvertedIndex, docId: uint64, text: string, config: TokenizerConfig = defaultTokenizerConfig()) = - let tokens = tokenize(text, config) - var termFreqs = initTable[string, int]() - var positions = initTable[string, seq[int]]() + acquire(idx.lock) + try: + let tokens = tokenize(text, config) + var termFreqs = initTable[string, int]() + var positions = initTable[string, seq[int]]() - for i, token in tokens: - if token notin termFreqs: - termFreqs[token] = 0 - positions[token] = @[] - inc termFreqs[token] - positions[token].add(i) + for i, token in tokens: + if token notin termFreqs: + termFreqs[token] = 0 + positions[token] = @[] + inc termFreqs[token] + positions[token].add(i) - for term, freq in termFreqs: - if term notin idx.postings: - idx.postings[term] = @[] - idx.postings[term].add(PostingEntry( - docId: docId, - termFreq: freq, - positions: positions[term], - )) + for term, freq in termFreqs: + if term notin idx.postings: + idx.postings[term] = @[] + idx.postings[term].add(PostingEntry( + docId: docId, + termFreq: freq, + positions: positions[term], + )) - idx.docLengths[docId] = tokens.len - inc idx.docCount - idx.totalTerms += tokens.len - idx.avgDocLen = float64(idx.totalTerms) / float64(idx.docCount) + idx.docLengths[docId] = tokens.len + inc idx.docCount + idx.totalTerms += tokens.len + idx.avgDocLen = float64(idx.totalTerms) / float64(idx.docCount) + finally: + release(idx.lock) proc removeDocument*(idx: InvertedIndex, docId: uint64) = - if docId notin idx.docLengths: - return - let docLen = idx.docLengths[docId] - idx.docLengths.del(docId) - dec idx.docCount - idx.totalTerms -= docLen - if idx.docCount > 0: - idx.avgDocLen = float64(idx.totalTerms) / float64(idx.docCount) + acquire(idx.lock) + try: + if docId notin idx.docLengths: + return + let docLen = idx.docLengths[docId] + idx.docLengths.del(docId) + dec idx.docCount + idx.totalTerms -= docLen + if idx.docCount > 0: + idx.avgDocLen = float64(idx.totalTerms) / float64(idx.docCount) - for term, postings in idx.postings.mpairs: - var newPostings: seq[PostingEntry] = @[] - for entry in postings: - if entry.docId != docId: - newPostings.add(entry) - postings = newPostings + for term, postings in idx.postings.mpairs: + var newPostings: seq[PostingEntry] = @[] + for entry in postings: + if entry.docId != docId: + newPostings.add(entry) + postings = newPostings + finally: + release(idx.lock) -proc bm25Score*(idx: InvertedIndex, term: string, docId: uint64, +proc bm25ScoreUnsafe(idx: InvertedIndex, term: string, docId: uint64, k1: float64 = 1.2, b: float64 = 0.75): float64 = if term notin idx.postings: return 0.0 @@ -190,49 +201,72 @@ proc bm25Score*(idx: InvertedIndex, term: string, docId: uint64, (float64(tf) + k1 * (1.0 - b + b * docLen / idx.avgDocLen)) return idf * tfNorm +proc bm25Score*(idx: InvertedIndex, term: string, docId: uint64, + k1: float64 = 1.2, b: float64 = 0.75): float64 = + acquire(idx.lock) + try: + result = bm25ScoreUnsafe(idx, term, docId, k1, b) + finally: + release(idx.lock) + proc search*(idx: InvertedIndex, query: string, limit: int = 10, config: TokenizerConfig = defaultTokenizerConfig()): seq[SearchResult] = - let queryTokens = tokenize(query, config) - if queryTokens.len == 0: - return @[] + acquire(idx.lock) + try: + let queryTokens = tokenize(query, config) + if queryTokens.len == 0: + return @[] - var docScores = initTable[uint64, float64]() - var docHighlights = initTable[uint64, seq[(int, int)]]() + var docScores = initTable[uint64, float64]() + var docHighlights = initTable[uint64, seq[(int, int)]]() - for token in queryTokens: - if token notin idx.postings: - continue - for entry in idx.postings[token]: - let score = bm25Score(idx, token, entry.docId) - if entry.docId notin docScores: - docScores[entry.docId] = 0.0 - docHighlights[entry.docId] = @[] - docScores[entry.docId] += score - for pos in entry.positions: - let start = pos - let stop = pos + token.len - docHighlights[entry.docId].add((start, stop)) + for token in queryTokens: + if token notin idx.postings: + continue + for entry in idx.postings[token]: + let score = bm25ScoreUnsafe(idx, token, entry.docId) + if entry.docId notin docScores: + docScores[entry.docId] = 0.0 + docHighlights[entry.docId] = @[] + docScores[entry.docId] += score + for pos in entry.positions: + let start = pos + let stop = pos + token.len + docHighlights[entry.docId].add((start, stop)) - var results: seq[SearchResult] = @[] - for docId, score in docScores: - results.add(SearchResult( - docId: docId, - score: score, - highlights: docHighlights.getOrDefault(docId, @[]), - )) + var results: seq[SearchResult] = @[] + for docId, score in docScores: + results.add(SearchResult( + docId: docId, + score: score, + highlights: docHighlights.getOrDefault(docId, @[]), + )) - results.sort(proc(a, b: SearchResult): int = cmp(b.score, a.score)) + results.sort(proc(a, b: SearchResult): int = cmp(b.score, a.score)) - if results.len > limit: - results = results[0.. limit: + results = results[0.. limit: - results = results[0.. limit: + results = results[0.. limit: - results = results[0.. limit: + results = results[0.. 0: - var match = false - # Check if pattern starts with/ends with or contains - if pattern.startsWith("*") and pattern.endsWith("*"): - let inner = pattern[1..^2] - if term.find(inner) >= 0: - match = true - elif pattern.startsWith("*"): - let suffix = pattern[1..^1] - if term.endsWith(suffix): - match = true - elif pattern.endsWith("*"): - let prefix = pattern[0..^2] - if term.startsWith(prefix): - match = true - else: - if term == pattern: - match = true + for term in idx.postings.keys: + # Simple pattern matching: check if pattern is substring + if pattern.len > 0: + var match = false + # Check if pattern starts with/ends with or contains + if pattern.startsWith("*") and pattern.endsWith("*"): + let inner = pattern[1..^2] + if term.find(inner) >= 0: + match = true + elif pattern.startsWith("*"): + let suffix = pattern[1..^1] + if term.endsWith(suffix): + match = true + elif pattern.endsWith("*"): + let prefix = pattern[0..^2] + if term.startsWith(prefix): + match = true + else: + if term == pattern: + match = true - if match: - for entry in idx.postings[term]: - if entry.docId notin docScores: - docScores[entry.docId] = 0.0 - docScores[entry.docId] += float64(entry.termFreq) + if match: + for entry in idx.postings[term]: + if entry.docId notin docScores: + docScores[entry.docId] = 0.0 + docScores[entry.docId] += float64(entry.termFreq) - var results: seq[SearchResult] = @[] - for docId, score in docScores: - results.add(SearchResult(docId: docId, score: score, highlights: @[])) + var results: seq[SearchResult] = @[] + for docId, score in docScores: + results.add(SearchResult(docId: docId, score: score, highlights: @[])) - results.sort(proc(a, b: SearchResult): int = cmp(b.score, a.score)) - if results.len > limit: - results = results[0.. limit: + results = results[0..= buf.len: raise newException(ValueError, "Wire protocol: unexpected end of buffer") - let kind = FieldKind(buf[pos]) + let kindByte = buf[pos] + if kindByte > byte(high(FieldKind)): + raise newException(ValueError, "Wire protocol: invalid field kind") + let kind = FieldKind(kindByte) inc pos case kind of fkNull: result = WireValue(kind: fkNull) @@ -223,26 +226,18 @@ proc deserializeValue*(buf: openArray[byte], pos: var int, depth: int = 0): Wire pos += 2 result = WireValue(kind: fkInt16, int16Val: val) of fkInt32: - result = WireValue(kind: fkInt32, int32Val: int32(readUint32(buf, pos))) + result = WireValue(kind: fkInt32, int32Val: cast[int32](readUint32(buf, pos))) of fkInt64: - result = WireValue(kind: fkInt64, int64Val: int64(readUint64(buf, pos))) + result = WireValue(kind: fkInt64, int64Val: cast[int64](readUint64(buf, pos))) of fkFloat32: var fl: float32 - var bytes: array[4, byte] - for i in 0..3: bytes[i] = buf[pos + i] - var i32: int32 - bigEndian32(addr i32, unsafeAddr bytes) - fl = cast[float32](i32) - pos += 4 + let raw = readUint32(buf, pos) + fl = cast[float32](raw) result = WireValue(kind: fkFloat32, float32Val: fl) of fkFloat64: var fl: float64 - var bytes: array[8, byte] - for i in 0..7: bytes[i] = buf[pos + i] - var i64: int64 - bigEndian64(addr i64, unsafeAddr bytes) - fl = cast[float64](i64) - pos += 8 + let raw = readUint64(buf, pos) + fl = cast[float64](raw) result = WireValue(kind: fkFloat64, float64Val: fl) of fkString: result = WireValue(kind: fkString, strVal: readString(buf, pos)) diff --git a/src/barabadb/query/executor.nim b/src/barabadb/query/executor.nim index 3e984ca..d2ba55c 100644 --- a/src/barabadb/query/executor.nim +++ b/src/barabadb/query/executor.nim @@ -11,6 +11,7 @@ import std/times import std/json import std/random import std/monotimes +import std/locks import lexer as qlex import parser as qpar import ast @@ -74,6 +75,7 @@ type sessionVars*: Table[string, string] # session-scoped key/value variables autoIncCounters*: Table[string, int64] # table.col -> next auto-increment value sequences*: Table[string, int64] # sequence name -> current value + ctxLock*: Lock # protects tables, views, btrees, ftsIndexes, users, policies, autoIncCounters, sequences MigrationRecord* = object name*: string @@ -160,6 +162,7 @@ proc newExecutionContext*(db: LSMTree): ExecutionContext = autoIncCounters: initTable[string, int64](), sequences: initTable[string, int64](), onChange: nil) + initLock(result.ctxLock) restoreSchema(result) # ---------------------------------------------------------------------- @@ -332,7 +335,10 @@ proc restoreSchema(ctx: ExecutionContext) = else: discard proc cloneForConnection*(ctx: ExecutionContext): ExecutionContext = - ExecutionContext(db: ctx.db, tables: ctx.tables, + var svCopy = initTable[string, string]() + for k, v in ctx.sessionVars: + svCopy[k] = v + result = ExecutionContext(db: ctx.db, tables: ctx.tables, btrees: ctx.btrees, views: ctx.views, cteTables: initTable[string, seq[Row]](), ftsIndexes: ctx.ftsIndexes, @@ -340,10 +346,11 @@ proc cloneForConnection*(ctx: ExecutionContext): ExecutionContext = users: ctx.users, policies: ctx.policies, txnManager: ctx.txnManager, currentUser: ctx.currentUser, currentRole: ctx.currentRole, - sessionVars: ctx.sessionVars, + sessionVars: svCopy, autoIncCounters: ctx.autoIncCounters, sequences: ctx.sequences, pendingTxn: nil, onChange: ctx.onChange) + initLock(result.ctxLock) # ---------------------------------------------------------------------- # Migration Helpers @@ -1059,18 +1066,27 @@ proc evalExpr*(expr: IRExpr, row: Table[string, string], ctx: ExecutionContext = if ctx == nil: return "0" let seqName = evalExpr(expr.irFuncArgs[0], row, ctx) var val: int64 = 0 - if seqName in ctx.sequences: - val = ctx.sequences[seqName] - val += 1 - ctx.sequences[seqName] = val + acquire(ctx.ctxLock) + try: + if seqName in ctx.sequences: + val = ctx.sequences[seqName] + val += 1 + ctx.sequences[seqName] = val + finally: + release(ctx.ctxLock) return $val of "currval": if expr.irFuncArgs.len < 1: return "0" if ctx == nil: return "0" let seqName = evalExpr(expr.irFuncArgs[0], row, ctx) - if seqName in ctx.sequences: - return $ctx.sequences[seqName] + acquire(ctx.ctxLock) + try: + if seqName in ctx.sequences: + return $ctx.sequences[seqName] + finally: + release(ctx.ctxLock) + return "0" return "0" of "snowflake_id": # Snowflake ID: timestamp_ms(41 bits) | node_id(10 bits) | sequence(12 bits) @@ -1457,6 +1473,7 @@ proc validateType*(colType: string, value: string): (bool, string) = return (false, "Vector dimension mismatch: expected " & $expectedDim & " but got " & $vec.len) return (true, "") +proc executeQueryImpl(ctx: ExecutionContext, astNode: Node, params: seq[WireValue] = @[]): ExecResult proc executeQuery*(ctx: ExecutionContext, astNode: Node, params: seq[WireValue] = @[]): ExecResult proc executeMigrationSql(ctx: ExecutionContext, sql: string): ExecResult @@ -1468,7 +1485,7 @@ proc fireTriggers*(ctx: ExecutionContext, tableName: string, timing: string, eve let tokens = qlex.tokenize(trig.action.strVal) let astNode = qpar.parse(tokens) if astNode.stmts.len > 0: - discard executeQuery(ctx, astNode) + discard executeQueryImpl(ctx, astNode) proc validateConstraints*(ctx: ExecutionContext, tableName: string, fields: seq[string], values: seq[seq[string]], skipPkCheck: bool = false): (bool, string) = @@ -3075,7 +3092,21 @@ proc getSelectColumns(stmt: Node): seq[string] = else: result.add("col" & $i) -proc executeQuery*(ctx: ExecutionContext, astNode: Node, params: seq[WireValue] = @[]): ExecResult = +proc isDDL(stmt: Node): bool = + case stmt.kind + of nkCreateTable, nkDropTable, nkAlterTable, + nkCreateView, nkDropView, + nkCreateIndex, nkDropIndex, + nkCreateTrigger, nkDropTrigger, + nkCreateUser, nkDropUser, + nkCreatePolicy, nkDropPolicy, + nkGrant, nkRevoke, + nkEnableRLS, nkDisableRLS: + result = true + else: + result = false + +proc executeQueryImpl(ctx: ExecutionContext, astNode: Node, params: seq[WireValue] = @[]): ExecResult = if astNode == nil or astNode.stmts.len == 0: return okResult() @@ -3109,7 +3140,7 @@ proc executeQuery*(ctx: ExecutionContext, astNode: Node, params: seq[WireValue] # Step 1: Execute the non-recursive anchor (left side of UNION) var innerLeft = Node(kind: nkStatementList, stmts: @[]) innerLeft.stmts.add(cteQuery.setOpLeft) - let anchorRes = executeQuery(ctx, innerLeft) + let anchorRes = executeQueryImpl(ctx, innerLeft) for row in anchorRes.rows: allRows.add(row) @@ -3125,7 +3156,7 @@ proc executeQuery*(ctx: ExecutionContext, astNode: Node, params: seq[WireValue] var innerRight = Node(kind: nkStatementList, stmts: @[]) innerRight.stmts.add(cteQuery.setOpRight) - let rightRes = executeQuery(ctx, innerRight) + let rightRes = executeQueryImpl(ctx, innerRight) ctx.cteTables = savedCte @@ -3159,7 +3190,7 @@ proc executeQuery*(ctx: ExecutionContext, astNode: Node, params: seq[WireValue] # Recursive CTE without UNION — treat as non-recursive fallback var inner = Node(kind: nkStatementList, stmts: @[]) inner.stmts.add(cteQuery) - let cteRes = executeQuery(ctx, inner) + let cteRes = executeQueryImpl(ctx, inner) var cteRows: seq[Row] = @[] for row in cteRes.rows: cteRows.add(row) @@ -3167,7 +3198,7 @@ proc executeQuery*(ctx: ExecutionContext, astNode: Node, params: seq[WireValue] else: var inner = Node(kind: nkStatementList, stmts: @[]) inner.stmts.add(cteQuery) - let cteRes = executeQuery(ctx, inner) + let cteRes = executeQueryImpl(ctx, inner) var cteRows: seq[Row] = @[] for row in cteRes.rows: cteRows.add(row) @@ -3180,7 +3211,7 @@ proc executeQuery*(ctx: ExecutionContext, astNode: Node, params: seq[WireValue] # Execute the view's underlying query var inner = Node(kind: nkStatementList, stmts: @[]) inner.stmts.add(viewQuery) - let innerResult = executeQuery(ctx, inner) + let innerResult = executeQueryImpl(ctx, inner) # Now filter and project with outer query constraints var filteredRows = innerResult.rows var cols = innerResult.columns @@ -3405,11 +3436,11 @@ proc executeQuery*(ctx: ExecutionContext, astNode: Node, params: seq[WireValue] # Execute left and right queries var innerLeft = Node(kind: nkStatementList, stmts: @[]) innerLeft.stmts.add(stmt.setOpLeft) - let leftRes = executeQuery(ctx, innerLeft) + let leftRes = executeQueryImpl(ctx, innerLeft) var innerRight = Node(kind: nkStatementList, stmts: @[]) innerRight.stmts.add(stmt.setOpRight) - let rightRes = executeQuery(ctx, innerRight) + let rightRes = executeQueryImpl(ctx, innerRight) # Derive columns from left side var cols = leftRes.columns @@ -3493,9 +3524,13 @@ proc executeQuery*(ctx: ExecutionContext, astNode: Node, params: seq[WireValue] if col.autoIncrement and col.name notin mutableFields: let counterKey = stmt.insTarget & "." & col.name var nextVal: int64 = 1 - if counterKey in ctx.autoIncCounters: - nextVal = ctx.autoIncCounters[counterKey] - ctx.autoIncCounters[counterKey] = nextVal + int64(mutableValues.len) + acquire(ctx.ctxLock) + try: + if counterKey in ctx.autoIncCounters: + nextVal = ctx.autoIncCounters[counterKey] + ctx.autoIncCounters[counterKey] = nextVal + int64(mutableValues.len) + finally: + release(ctx.ctxLock) # Insert at position 0 so it becomes the primary storage key mutableFields.insert(col.name, 0) for i in 0..= ctx.autoIncCounters[counterKey]: - ctx.autoIncCounters[counterKey] = intVal + 1 + acquire(ctx.ctxLock) + try: + if counterKey notin ctx.autoIncCounters or intVal >= ctx.autoIncCounters[counterKey]: + ctx.autoIncCounters[counterKey] = intVal + 1 + finally: + release(ctx.ctxLock) except: discard applyDefaultValues(tbl, mutableFields, mutableValues) @@ -3657,7 +3696,7 @@ proc executeQuery*(ctx: ExecutionContext, astNode: Node, params: seq[WireValue] var sourceRows: seq[Row] = @[] if stmt.mergeSource != nil: if stmt.mergeSource.kind == nkSelect: - let srcRes = executeQuery(ctx, Node(kind: nkStatementList, stmts: @[stmt.mergeSource])) + let srcRes = executeQueryImpl(ctx, Node(kind: nkStatementList, stmts: @[stmt.mergeSource])) sourceRows = srcRes.rows elif stmt.mergeSource.kind == nkIdent: sourceRows = execScan(ctx, stmt.mergeSource.identName) @@ -4236,9 +4275,22 @@ proc executeQuery*(ctx: ExecutionContext, astNode: Node, params: seq[WireValue] return errResult("Unsupported statement type: " & $stmt.kind) +proc executeQuery*(ctx: ExecutionContext, astNode: Node, params: seq[WireValue] = @[]): ExecResult = + if astNode == nil or astNode.stmts.len == 0: + return okResult() + let stmt = astNode.stmts[0] + if isDDL(stmt): + acquire(ctx.ctxLock) + try: + result = executeQueryImpl(ctx, astNode, params) + finally: + release(ctx.ctxLock) + else: + result = executeQueryImpl(ctx, astNode, params) + proc executeMigrationSql(ctx: ExecutionContext, sql: string): ExecResult = let tokens = qlex.tokenize(sql) let astNode = qpar.parse(tokens) if astNode.stmts.len > 0: - return executeQuery(ctx, astNode) + return executeQueryImpl(ctx, astNode) return okResult(msg="Empty migration body") diff --git a/src/barabadb/storage/btree.nim b/src/barabadb/storage/btree.nim index 5eef249..3e91ddf 100644 --- a/src/barabadb/storage/btree.nim +++ b/src/barabadb/storage/btree.nim @@ -1,5 +1,6 @@ ## B-Tree Index — ordered key-value index import std/tables +import std/locks const DefaultBTreeOrder* = 32 @@ -16,6 +17,7 @@ type root: BTreeNode[K, V] order: int size: int + lock*: Lock proc newBTreeNode[K, V](isLeaf: bool = true): BTreeNode[K, V] = BTreeNode[K, V]( @@ -24,7 +26,8 @@ proc newBTreeNode[K, V](isLeaf: bool = true): BTreeNode[K, V] = ) proc newBTreeIndex*[K, V](order: int = DefaultBTreeOrder): BTreeIndex[K, V] = - BTreeIndex[K, V](root: newBTreeNode[K, V](), order: order, size: 0) + result = BTreeIndex[K, V](root: newBTreeNode[K, V](), order: order, size: 0) + initLock(result.lock) proc search[K, V](node: BTreeNode[K, V], key: K): seq[V] = var i = 0 @@ -84,66 +87,91 @@ proc insertNonFull[K, V](node: BTreeNode[K, V], key: K, value: V, order: int) = insertNonFull(node.children[i], key, value, order) proc insert*[K, V](btree: var BTreeIndex[K, V], key: K, value: V) = - if btree.root.keys.len == btree.order - 1: - var newRoot = newBTreeNode[K, V](isLeaf = false) - newRoot.children.add(btree.root) - splitChild(newRoot, 0, btree.order) - btree.root = newRoot - insertNonFull(btree.root, key, value, btree.order) - else: - insertNonFull(btree.root, key, value, btree.order) - inc btree.size + acquire(btree.lock) + try: + if btree.root.keys.len == btree.order - 1: + var newRoot = newBTreeNode[K, V](isLeaf = false) + newRoot.children.add(btree.root) + splitChild(newRoot, 0, btree.order) + btree.root = newRoot + insertNonFull(btree.root, key, value, btree.order) + else: + insertNonFull(btree.root, key, value, btree.order) + inc btree.size + finally: + release(btree.lock) proc get*[K, V](btree: BTreeIndex[K, V], key: K): seq[V] = - search(btree.root, key) + acquire(btree.lock) + try: + result = search(btree.root, key) + finally: + release(btree.lock) proc contains*[K, V](btree: BTreeIndex[K, V], key: K): bool = - return btree.get(key).len > 0 + acquire(btree.lock) + try: + result = search(btree.root, key).len > 0 + finally: + release(btree.lock) proc scan*[K, V](btree: BTreeIndex[K, V], startKey, endKey: K): seq[(K, seq[V])] = - result = @[] - var node = btree.root - while not node.isLeaf: - var i = 0 - while i < node.keys.len and startKey > node.keys[i]: - inc i - node = node.children[i] + acquire(btree.lock) + try: + result = @[] + var node = btree.root + while not node.isLeaf: + var i = 0 + while i < node.keys.len and startKey > node.keys[i]: + inc i + node = node.children[i] - while node != nil: - for i in 0..= startKey: - if node.keys[i] <= endKey: - result.add((node.keys[i], node.values[i])) - else: - return - node = node.next + while node != nil: + for i in 0..= startKey: + if node.keys[i] <= endKey: + result.add((node.keys[i], node.values[i])) + else: + return + node = node.next + finally: + release(btree.lock) -proc len*[K, V](btree: BTreeIndex[K, V]): int = btree.size +proc len*[K, V](btree: BTreeIndex[K, V]): int = + acquire(btree.lock) + try: + result = btree.size + finally: + release(btree.lock) proc remove*[K, V](btree: var BTreeIndex[K, V], key: K, value: V) = - proc removeRec(node: BTreeNode[K, V]): bool = - var i = 0 - while i < node.keys.len and key > node.keys[i]: - inc i - if node.isLeaf: - if i < node.keys.len and key == node.keys[i]: - var vals = node.values[i] - var idx = -1 - for j in 0..= 0: - vals.del(idx) - if vals.len == 0: - node.keys.del(i) - node.values.del(i) - else: - node.values[i] = vals - return true - return false - else: - return removeRec(node.children[i]) + acquire(btree.lock) + try: + proc removeRec(node: BTreeNode[K, V]): bool = + var i = 0 + while i < node.keys.len and key > node.keys[i]: + inc i + if node.isLeaf: + if i < node.keys.len and key == node.keys[i]: + var vals = node.values[i] + var idx = -1 + for j in 0..= 0: + vals.del(idx) + if vals.len == 0: + node.keys.del(i) + node.values.del(i) + else: + node.values[i] = vals + return true + return false + else: + return removeRec(node.children[i]) - if removeRec(btree.root): - dec btree.size + if removeRec(btree.root): + dec btree.size + finally: + release(btree.lock) diff --git a/tests/fuzz_test.nim b/tests/fuzz_test.nim new file mode 100644 index 0000000..0d4e374 --- /dev/null +++ b/tests/fuzz_test.nim @@ -0,0 +1,87 @@ +## Wire Protocol Fuzz Tests +import std/unittest +import std/random +import std/strutils + +import barabadb/protocol/wire + +suite "Wire Protocol Fuzz": + test "deserializeValue survives random bytes": + var rng = initRand(12345) + for i in 0..<500: + var buf = newSeq[byte](rng.rand(1..256)) + for b in buf.mitems: + b = byte(rng.rand(0..255)) + var pos = 0 + try: + discard deserializeValue(buf, pos) + except: + discard # Exceptions are expected for garbage input + + test "deserializeValue survives truncated buffers": + var rng = initRand(12346) + for i in 0..<200: + # Start with a valid-ish prefix then truncate + var buf: seq[byte] = @[byte(rng.rand(0..12))] # random FieldKind + let extra = rng.rand(0..64) + for j in 0..= 12 + + test "serializeMessage roundtrip with random payloads": + var rng = initRand(12350) + for i in 0..<100: + let reqId = uint32(rng.rand(int64(high(uint32)))) + var msg = WireMessage( + header: MessageHeader(kind: mkQuery, length: uint32(rng.rand(0..1000)), requestId: reqId), + payload: newSeq[byte](rng.rand(0..200)) + ) + for j in 0..= 12 + + test "makeErrorMessage with large IDs": + var rng = initRand(12351) + for i in 0..<100: + let reqId = uint32(rng.rand(int64(high(uint32)))) + let err = makeErrorMessage(reqId, uint32(rng.rand(0..255)), "fuzz") + check err.len > 0 diff --git a/tests/nimforum_smoke_test.nim b/tests/nimforum_smoke_test.nim new file mode 100644 index 0000000..c886ca4 --- /dev/null +++ b/tests/nimforum_smoke_test.nim @@ -0,0 +1,69 @@ +## NimForum Adapter Smoke Test +import std/unittest +import std/osproc +import std/os +import std/strutils +import std/strtabs +import std/times + +import ../adaptors/nim/baradb_sqlite as sqlite + +suite "NimForum Adapter Smoke Test": + var serverProcess: Process + var port: int + var dataDir: string + + setup: + port = 35000 + (getTime().toUnix.int mod 10000) + dataDir = getTempDir() / "baradb_nimforum_" & $port + createDir(dataDir) + var env = newStringTable() + for key, val in envPairs(): + env[key] = val + env["BARADB_PORT"] = $port + env["BARADB_DATA_DIR"] = dataDir + env["BARADB_LOG_LEVEL"] = "error" + serverProcess = startProcess("./build/baradadb", env=env, options={poStdErrToStdOut, poDaemon}) + sleep(800) + + teardown: + if serverProcess != nil: + serverProcess.terminate() + discard serverProcess.waitForExit() + removeDir(dataDir) + + test "Adapter basic CRUD over TCP": + var db = open("127.0.0.1:" & $port, "", "", "default") + + db.exec(sql"CREATE TABLE nf_test (id INT PRIMARY KEY, name STRING)") + db.exec(sql"INSERT INTO nf_test (id, name) VALUES (1, 'hello')") + db.exec(sql"INSERT INTO nf_test (id, name) VALUES (2, 'world')") + + let rows = db.getAllRows(sql"SELECT * FROM nf_test") + check rows.len == 2 + + let row = db.getRow(sql"SELECT * FROM nf_test WHERE id = 1") + check row.len == 2 + check row[0] == "1" + check row[1] == "hello" + + let val = db.getValue(sql"SELECT name FROM nf_test WHERE id = 2") + check val == "world" + + let cnt = db.getValue(sql"SELECT count(*) FROM nf_test") + check cnt == "2" + + db.close() + + test "Adapter parameterized queries": + var db = open("127.0.0.1:" & $port, "", "", "default") + + db.exec(sql"CREATE TABLE nf_params (id INT PRIMARY KEY, val STRING)") + db.exec(sql"INSERT INTO nf_params (id, val) VALUES (?, ?)", 10, "ten") + db.exec(sql"INSERT INTO nf_params (id, val) VALUES (?, ?)", 20, "twenty") + + let row = db.getRow(sql"SELECT val FROM nf_params WHERE id = ?", 10) + check row.len >= 1 + check row[row.len - 1] == "ten" + + db.close() diff --git a/tests/prop_test.nim b/tests/prop_test.nim new file mode 100644 index 0000000..bec7458 --- /dev/null +++ b/tests/prop_test.nim @@ -0,0 +1,123 @@ +## Property-Based Tests — evalExprValue invariants +import std/unittest +import std/tables +import std/random +import std/os +import std/monotimes +import std/math + +import barabadb/core/types +import barabadb/storage/lsm +import barabadb/query/ir as qir +import barabadb/query/executor as qexec + +suite "Property-Based — evalExprValue Invariants": + setup: + var testDir = getTempDir() / "baradb_prop_test_" & $getCurrentProcessId() & "_" & $getMonoTime().ticks + createDir(testDir) + var db = newLSMTree(testDir) + var ctx = qexec.newExecutionContext(db) + + teardown: + removeDir(testDir) + + proc randIntLit(rng: var Rand, minVal: int = -1000, maxVal: int = 1000): IRExpr = + result = IRExpr(kind: irekLiteral, valueKind: vkInt64) + result.literal = IRLiteral(kind: vkInt64, int64Val: int64(rng.rand(minVal..maxVal))) + + proc randFloatLit(rng: var Rand, minVal: float = -1000.0, maxVal: float = 1000.0): IRExpr = + result = IRExpr(kind: irekLiteral, valueKind: vkFloat64) + result.literal = IRLiteral(kind: vkFloat64, float64Val: minVal + rng.rand(maxVal - minVal)) + + proc randBinaryExpr(rng: var Rand, left, right: IRExpr, op: IROperator): IRExpr = + result = IRExpr(kind: irekBinary) + result.binOp = op + result.binLeft = left + result.binRight = right + if left.valueKind == vkInt64 and right.valueKind == vkInt64 and op != irDiv: + result.valueKind = vkInt64 + else: + result.valueKind = vkFloat64 + + proc randUnaryExpr(rng: var Rand, expr: IRExpr, op: IROperator): IRExpr = + result = IRExpr(kind: irekUnary) + result.unOp = op + result.unExpr = expr + result.valueKind = expr.valueKind + + test "Literal eval returns correct ValueKind (INT)": + var rng = initRand(42) + for i in 0..<100: + let lit = randIntLit(rng) + let v = evalExprValue(lit, initTable[string, string](), nil) + check v.kind == vkInt64 + + test "Literal eval returns correct ValueKind (FLOAT)": + var rng = initRand(43) + for i in 0..<100: + let lit = randFloatLit(rng) + let v = evalExprValue(lit, initTable[string, string](), nil) + check v.kind == vkFloat64 + + test "INT addition is commutative": + var rng = initRand(44) + for i in 0..<100: + let a = randIntLit(rng) + let b = randIntLit(rng) + let sum1 = evalExprValue(randBinaryExpr(rng, a, b, irAdd), initTable[string, string](), nil) + let sum2 = evalExprValue(randBinaryExpr(rng, b, a, irAdd), initTable[string, string](), nil) + if sum1.kind == vkInt64 and sum2.kind == vkInt64: + check sum1.int64Val == sum2.int64Val + + test "FLOAT addition is commutative": + var rng = initRand(45) + for i in 0..<100: + let a = randFloatLit(rng) + let b = randFloatLit(rng) + let sum1 = evalExprValue(randBinaryExpr(rng, a, b, irAdd), initTable[string, string](), nil) + let sum2 = evalExprValue(randBinaryExpr(rng, b, a, irAdd), initTable[string, string](), nil) + if sum1.kind == vkFloat64 and sum2.kind == vkFloat64: + check abs(sum1.float64Val - sum2.float64Val) < 1e-9 + + test "INT multiplication by 1 is identity": + var rng = initRand(46) + for i in 0..<100: + let a = randIntLit(rng) + let one = IRExpr(kind: irekLiteral, valueKind: vkInt64) + one.literal = IRLiteral(kind: vkInt64, int64Val: 1) + let prod = evalExprValue(randBinaryExpr(rng, a, one, irMul), initTable[string, string](), nil) + if prod.kind == vkInt64: + check prod.int64Val == a.literal.int64Val + + test "Double negation of INT is identity": + var rng = initRand(47) + for i in 0..<100: + let a = randIntLit(rng) + let neg = randUnaryExpr(rng, a, irNeg) + let negNeg = randUnaryExpr(rng, neg, irNeg) + let v = evalExprValue(negNeg, initTable[string, string](), nil) + if v.kind == vkInt64: + check v.int64Val == a.literal.int64Val + + test "Double negation of FLOAT is identity": + var rng = initRand(48) + for i in 0..<100: + let a = randFloatLit(rng) + let neg = randUnaryExpr(rng, a, irNeg) + let negNeg = randUnaryExpr(rng, neg, irNeg) + let v = evalExprValue(negNeg, initTable[string, string](), nil) + if v.kind == vkFloat64: + check abs(v.float64Val - a.literal.float64Val) < 1e-9 + + test "NULL literal propagates through arithmetic": + let nullLit = IRExpr(kind: irekLiteral, valueKind: vkNull) + nullLit.literal = IRLiteral(kind: vkNull) + let intLit = IRExpr(kind: irekLiteral, valueKind: vkInt64) + intLit.literal = IRLiteral(kind: vkInt64, int64Val: 5) + for op in [irAdd, irSub, irMul, irDiv]: + let expr = IRExpr(kind: irekBinary) + expr.binOp = op + expr.binLeft = nullLit + expr.binRight = intLit + let v = evalExprValue(expr, initTable[string, string](), nil) + check v.kind == vkNull diff --git a/tests/test_all.nim b/tests/test_all.nim index 8bfca7f..1449bf8 100644 --- a/tests/test_all.nim +++ b/tests/test_all.nim @@ -6,6 +6,7 @@ import std/os import std/asyncdispatch import std/monotimes import std/base64 +import std/random import barabadb/core/types import barabadb/core/mvcc @@ -3728,3 +3729,5 @@ suite "Join Performance — Hash Join & Index Nested Loop": check r.rows[0]["name"] == "Alice" check r.rows[3]["name"] == "Carol" check r.rows[3]["total"] == "\\N" + +