baraba/Baradb
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: non-blocking replication/disttxn timeouts, OTLP epoch time, test paths, compiler warnings

Browse Source
This commit is contained in:
dimgigov
2026-05-07 15:41:33 +03:00
parent d3b38251c3
commit c8633b9589
7 changed files with 87 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
baradadb.nimble
+2 -2
View File
@@ -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"
src/barabadb/core/disttxn.nim
+18 -1
View File
@@ -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 <txnId> <action>\n" where action = PREPARE|COMMIT|ROLLBACK
## Response: "OK\n" or "ERR <msg>\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 = ""
src/barabadb/core/httpserver.nim
+8 -9
View File
@@ -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.} =
src/barabadb/core/replication.nim
+29 -10
View File
@@ -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 <lsn> <dataLen>\n<data>"
## Response: "ACK <lsn>\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)
src/barabadb/core/server.nim
+13 -26
View File
@@ -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: @[],
src/barabadb/core/tracing.nim
+8 -3
View File
@@ -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,
})
src/barabadb/query/executor.nim
+9 -11
View File
@@ -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