diff --git a/baradadb.nimble b/baradadb.nimble index db4f5b9..8f3666c 100644 --- a/baradadb.nimble +++ b/baradadb.nimble @@ -23,7 +23,7 @@ task build_release, "Build release version": exec "nim c -d:ssl -d:release --opt:speed -o:build/baradadb src/baradadb.nim" task test, "Run all tests": - exec "nim c -d:ssl -r tests/test_all.nim" + exec "nim c --path:src -d:ssl -r tests/test_all.nim" task bench, "Run benchmarks": - exec "nim c -d:ssl -d:release -r benchmarks/bench_storage.nim" + exec "nim c --path:src -d:ssl -d:release -r benchmarks/bench_storage.nim" diff --git a/src/barabadb/core/disttxn.nim b/src/barabadb/core/disttxn.nim index 526e70b..179402c 100644 --- a/src/barabadb/core/disttxn.nim +++ b/src/barabadb/core/disttxn.nim @@ -4,6 +4,7 @@ import std/locks import std/monotimes import std/net import std/strutils +import std/nativesockets type DistTxnState* = enum @@ -75,13 +76,29 @@ proc addParticipant*(txn: DistributedTransaction, nodeId: string, ) release(txn.lock) +proc connectWithTimeout(sock: Socket, host: string, port: Port, timeoutMs: int): bool = + ## Non-blocking connect with timeout to avoid hanging on unreachable hosts. + sock.getFd.setBlocking(false) + try: + sock.connect(host, port) + sock.getFd.setBlocking(true) + return true + except OSError: + var fds = @[sock.getFd] + if selectWrite(fds, timeoutMs) <= 0: + return false + sock.getFd.setBlocking(true) + return true + proc sendDistTxnRpc(host: string, port: int, txnId: uint64, action: string, timeoutMs: int = 5000): bool = ## Send 2PC RPC to participant node via TCP text protocol. ## Protocol: "DISTTXN \n" where action = PREPARE|COMMIT|ROLLBACK ## Response: "OK\n" or "ERR \n" try: var sock = newSocket() - sock.connect(host, Port(port)) + if not connectWithTimeout(sock, host, Port(port), timeoutMs): + sock.close() + return false let msg = "DISTTXN " & $txnId & " " & action & "\n" sock.send(msg) var response = "" diff --git a/src/barabadb/core/httpserver.nim b/src/barabadb/core/httpserver.nim index d8c0e8d..2636309 100644 --- a/src/barabadb/core/httpserver.nim +++ b/src/barabadb/core/httpserver.nim @@ -16,7 +16,6 @@ import ../query/executor import ../storage/lsm import ../core/mvcc import ../protocol/wire -import ../protocol/ratelimit import ../core/websocket import jwt as jwtlib @@ -147,30 +146,30 @@ proc queryHandler(server: HttpServer): RequestHandler = of JString: params.add(WireValue(kind: fkString, strVal: p.getStr())) else: params.add(WireValue(kind: fkString, strVal: $p)) - let result = executor.executeQuery(reqCtx, astNode, params) + let res = executor.executeQuery(reqCtx, astNode, params) - if result.success: + if res.success: var jsonRows = newJArray() - for row in result.rows: + for row in res.rows: var jsonRow = newJObject() - for col in result.columns: + for col in res.columns: let key = col var val = "" if key in row: val = row[key] jsonRow[key] = %val jsonRows.add(jsonRow) var jsonCols = newJArray() - for c in result.columns: + for c in res.columns: jsonCols.add(%c) ctx.json(%*{ "rows": jsonRows, - "affectedRows": result.affectedRows, + "affectedRows": res.affectedRows, "columns": jsonCols, - "message": if result.message.len > 0: %result.message else: newJNull() + "message": if res.message.len > 0: %res.message else: newJNull() }) else: server.metrics.queryErrors += 1 - ctx.json(%*{"error": result.message}, 400) + ctx.json(%*{"error": res.message}, 400) proc healthHandler(): RequestHandler = return proc(request: Request) {.gcsafe.} = diff --git a/src/barabadb/core/replication.nim b/src/barabadb/core/replication.nim index 2a0eb7c..bdba388 100644 --- a/src/barabadb/core/replication.nim +++ b/src/barabadb/core/replication.nim @@ -4,6 +4,7 @@ import std/sets import std/locks import std/net import std/strutils +import std/nativesockets type @@ -72,13 +73,29 @@ proc connectReplica*(rm: ReplicationManager, id: string) = rm.replicas[id].connected = true release(rm.lock) +proc connectWithTimeout(sock: Socket, host: string, port: Port, timeoutMs: int): bool = + ## Non-blocking connect with timeout to avoid hanging on unreachable hosts. + sock.getFd.setBlocking(false) + try: + sock.connect(host, port) + sock.getFd.setBlocking(true) + return true + except OSError: + var fds = @[sock.getFd] + if selectWrite(fds, timeoutMs) <= 0: + return false + sock.getFd.setBlocking(true) + return true + proc shipToReplica(replica: Replica, lsn: uint64, data: seq[byte]): bool = ## Send replication data to a replica via TCP. ## Protocol: "REP \n" ## Response: "ACK \n" on success try: var sock = newSocket() - sock.connect(replica.host, Port(replica.port)) + if not connectWithTimeout(sock, replica.host, Port(replica.port), 500): + sock.close() + return false let header = "REP " & $lsn & " " & $data.len & "\n" sock.send(header) if data.len > 0: @@ -108,20 +125,22 @@ proc writeLsn*(rm: ReplicationManager, data: seq[byte]): uint64 = discard shipToReplica(replica, lsn, data) return lsn of rmSync: - rm.pendingAcks[lsn] = initHashSet[string]() - for replica in replicasToShip: - rm.pendingAcks[lsn].incl(replica.id) + if replicasToShip.len > 0: + rm.pendingAcks[lsn] = initHashSet[string]() + for replica in replicasToShip: + rm.pendingAcks[lsn].incl(replica.id) release(rm.lock) for replica in replicasToShip: discard shipToReplica(replica, lsn, data) return lsn of rmSemiSync: - rm.pendingAcks[lsn] = initHashSet[string]() - var count = 0 - for replica in replicasToShip: - if count < rm.syncReplicaCount: - rm.pendingAcks[lsn].incl(replica.id) - inc count + if replicasToShip.len > 0: + rm.pendingAcks[lsn] = initHashSet[string]() + var count = 0 + for replica in replicasToShip: + if count < rm.syncReplicaCount: + rm.pendingAcks[lsn].incl(replica.id) + inc count release(rm.lock) for replica in replicasToShip: discard shipToReplica(replica, lsn, data) diff --git a/src/barabadb/core/server.nim b/src/barabadb/core/server.nim index 0d7cf14..fc9e25c 100644 --- a/src/barabadb/core/server.nim +++ b/src/barabadb/core/server.nim @@ -33,11 +33,6 @@ type tls*: TLSContext activeConnections*: int - ClientConnection = ref object - socket: AsyncSocket - id: int - currentTxn: Transaction - proc newServer*(config: BaraConfig): Server = let dataDir = config.dataDir / "server" let db = newLSMTree(dataDir) @@ -61,12 +56,6 @@ proc readUint32BE(data: string, pos: int): uint32 = bytes[i] = byte(data[pos + i]) bigEndian32(addr result, unsafeAddr bytes) -proc writeUint32BE(val: uint32): array[4, byte] = - bigEndian32(addr result, unsafeAddr val) - -proc writeUint64BE(val: uint64): array[8, byte] = - bigEndian64(addr result, unsafeAddr val) - proc parseHeader(data: string): (bool, MessageHeader) = if data.len < 12: return (false, MessageHeader()) @@ -123,18 +112,18 @@ proc executeQuery(db: LSMTree, ctx: ExecutionContext, query: string, params: seq if astNode.stmts.len == 0: return (true, QueryResult(), "") - let result = executor.executeQuery(ctx, astNode, params) - if result.success: + let res = executor.executeQuery(ctx, astNode, params) + if res.success: # Ship written key-value pairs to replicas - if replication != nil and result.keyValuePairs.len > 0: - for (key, value) in result.keyValuePairs: + if replication != nil and res.keyValuePairs.len > 0: + for (key, value) in res.keyValuePairs: var data = newSeq[byte](key.len + 1 + value.len) for i, c in key: data[i] = byte(c) data[key.len] = byte(0) for i, c in value: data[key.len + 1 + i] = c discard replication.writeLsn(data) - var qr = QueryResult(affectedRows: result.affectedRows, rowCount: result.rows.len) - qr.columns = result.columns + var qr = QueryResult(affectedRows: res.affectedRows, rowCount: res.rows.len) + qr.columns = res.columns var colTypes: seq[string] = @[] var tableName = "" @@ -147,7 +136,7 @@ proc executeQuery(db: LSMTree, ctx: ExecutionContext, query: string, params: seq if tableName.len > 0 and tableName in ctx.tables: let tbl = ctx.tables[tableName] - for col in result.columns: + for col in res.columns: var found = "" for c in tbl.columns: if c.name.toLower() == col.toLower(): @@ -155,20 +144,20 @@ proc executeQuery(db: LSMTree, ctx: ExecutionContext, query: string, params: seq break colTypes.add(found) else: - colTypes = newSeq[string](result.columns.len) + colTypes = newSeq[string](res.columns.len) qr.columnTypes = colTypes.mapIt(typeToFieldKind(it)) qr.rows = @[] - for row in result.rows: + for row in res.rows: var wireRow: seq[WireValue] = @[] - for i, col in result.columns: + for i, col in res.columns: let val = if col in row: row[col] else: "" let cType = if i < colTypes.len: colTypes[i] else: "" wireRow.add(valueToWire(val, cType)) qr.rows.add(wireRow) - return (true, qr, result.message) + return (true, qr, res.message) else: - return (false, QueryResult(), result.message) + return (false, QueryResult(), res.message) except Exception as e: return (false, QueryResult(), e.msg) @@ -265,7 +254,6 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.} info("Client " & $clientId & " connected") var connCtx = cloneForConnection(server.ctx) let idleTimeout = server.config.idleTimeoutMs - let queryTimeout = server.config.queryTimeoutMs let slowThreshold = server.config.slowQueryThresholdMs let slowLog = server.config.slowQueryLogPath var authenticated = not server.config.authEnabled @@ -366,7 +354,7 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.} case header.kind of mkAuth: let tokenStr = parseAuthMessage(cast[seq[byte]](payload)) - let (valid, userId, role) = verifyToken(secret, tokenStr) + let (valid, userId, _) = verifyToken(secret, tokenStr) if valid: authenticated = true let okMsg = makeAuthOkMessage(header.requestId) @@ -420,7 +408,6 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.} await client.send(cast[string](errorMsg)) of mkPing: - var pongPayload: seq[byte] = @[] var pongMsg = WireMessage( header: MessageHeader(kind: mkPong, length: 0, requestId: header.requestId), payload: @[], diff --git a/src/barabadb/core/tracing.nim b/src/barabadb/core/tracing.nim index eb3c3a9..e687e31 100644 --- a/src/barabadb/core/tracing.nim +++ b/src/barabadb/core/tracing.nim @@ -16,8 +16,10 @@ type spanId*: string parentSpanId*: string name*: string - startTime*: int64 # monotonic ticks + startTime*: int64 # monotonic ticks (for duration) + startTimeEpochNs*: int64 # Unix epoch nanoseconds (for OTLP) endTime*: int64 + endTimeEpochNs*: int64 durationMs*: float64 status*: SpanStatus attributes*: Table[string, string] @@ -36,12 +38,14 @@ proc genId(tracer: Tracer): string = proc beginSpan*(tracer: Tracer, name: string, attributes: Table[string, string] = initTable[string, string]()): Span = if not tracer.enabled: return nil + let nowNs = int64(epochTime() * 1_000_000_000) let span = Span( traceId: if tracer.activeSpan != nil: tracer.activeSpan.traceId else: genId(tracer), spanId: genId(tracer), parentSpanId: if tracer.activeSpan != nil: tracer.activeSpan.spanId else: "", name: name, startTime: getMonoTime().ticks(), + startTimeEpochNs: nowNs, attributes: attributes, ) tracer.spans.add(span) @@ -51,6 +55,7 @@ proc beginSpan*(tracer: Tracer, name: string, attributes: Table[string, string] proc endSpan*(tracer: Tracer, span: Span, status: SpanStatus = ssOk) = if span == nil or not tracer.enabled: return span.endTime = getMonoTime().ticks() + span.endTimeEpochNs = int64(epochTime() * 1_000_000_000) span.durationMs = float64(span.endTime - span.startTime) / 1_000_000.0 span.status = status if span.parentSpanId.len > 0: @@ -100,8 +105,8 @@ proc exportOtlp*(tracer: Tracer, endpoint: string = "http://localhost:4318/v1/tr "parentSpanId": span.parentSpanId, "name": span.name, "kind": "SPAN_KIND_INTERNAL", - "startTimeUnixNano": $span.startTime, - "endTimeUnixNano": $span.endTime, + "startTimeUnixNano": $span.startTimeEpochNs, + "endTimeUnixNano": $span.endTimeEpochNs, "status": {"code": if span.status == ssOk: "STATUS_CODE_OK" else: "STATUS_CODE_ERROR"}, "attributes": attrs, }) diff --git a/src/barabadb/query/executor.nim b/src/barabadb/query/executor.nim index d528c67..a8fb156 100644 --- a/src/barabadb/query/executor.nim +++ b/src/barabadb/query/executor.nim @@ -1246,8 +1246,6 @@ proc executePlan*(ctx: ExecutionContext, plan: IRPlan): seq[Row] = let v = evalExpr(expr.aggArgs[0], row) if maxVal == "" or v > maxVal: maxVal = v newRow[alias] = maxVal - else: - newRow[alias] = "" else: let val = evalExpr(expr, if sourceRows.len > 0: sourceRows[0] else: initTable[string, string]()) if alias.len > 0: newRow[alias] = val @@ -2275,11 +2273,11 @@ proc executeQuery*(ctx: ExecutionContext, astNode: Node, params: seq[WireValue] storedRec.checksum[0..<16] & ", Expected: " & expectedChecksum[0..<16]) let startTime = epochTime() - let result = executeMigrationSql(ctx, upBody) + let res = executeMigrationSql(ctx, upBody) let durationMs = int((epochTime() - startTime) * 1000) - if not result.success: - return errResult("Migration '" & stmt.amName & "' failed: " & result.message) + if not res.success: + return errResult("Migration '" & stmt.amName & "' failed: " & res.message) ctx.db.put(migrationAppliedKey(stmt.amName), cast[seq[byte]]("applied")) setMigrationRecord(ctx, MigrationRecord( @@ -2332,10 +2330,10 @@ proc executeQuery*(ctx: ExecutionContext, astNode: Node, params: seq[WireValue] if not found: return errResult("Migration '" & name & "' not found during batch apply") let startTime = epochTime() - let result = executeMigrationSql(ctx, upBody) + let res = executeMigrationSql(ctx, upBody) let durationMs = int((epochTime() - startTime) * 1000) - if not result.success: - return errResult("Migration '" & name & "' failed: " & result.message & + if not res.success: + return errResult("Migration '" & name & "' failed: " & res.message & " (" & $appliedCount & " migrations applied before failure)") ctx.db.put(migrationAppliedKey(name), cast[seq[byte]]("applied")) setMigrationRecord(ctx, MigrationRecord( @@ -2375,9 +2373,9 @@ proc executeQuery*(ctx: ExecutionContext, astNode: Node, params: seq[WireValue] return errResult("Migration '" & name & "' not found during rollback") if downBody.len == 0: return errResult("Migration '" & name & "' has no DOWN script") - let result = executeMigrationSql(ctx, downBody) - if not result.success: - return errResult("Rollback of '" & name & "' failed: " & result.message) + let res = executeMigrationSql(ctx, downBody) + if not res.success: + return errResult("Rollback of '" & name & "' failed: " & res.message) ctx.db.delete(migrationAppliedKey(name)) var rec = getMigrationRecord(ctx, name) rec.rolledBack = true