From 8fb5dde858272ac4d9b34d3c771b6e4649f49858 Mon Sep 17 00:00:00 2001 From: dimgigov Date: Thu, 14 May 2026 02:22:07 +0300 Subject: [PATCH 1/3] fix(protocol): serialize float32/float64/fkVector in big-endian The JavaScript client reads floats via readFloatBE/readDoubleBE, which expect IEEE-754 values in big-endian byte order. The Nim server was writing them with copyMem in native byte order, so on little-endian machines (x86_64) the JS side deserialized FLOAT64 values as garbage (e.g. 1.0 became 3.03865e-319). Fix serialization by casting to int32/int64 and using bigEndian32/ bigEndian64, mirroring the existing big-endian handling for ints. Apply the same fix to deserialization and to fkVector elements. --- src/barabadb/protocol/wire.nim | 21 +++++++++++++++------ 1 file changed, 15 insertions(+), 6 deletions(-) diff --git a/src/barabadb/protocol/wire.nim b/src/barabadb/protocol/wire.nim index 88963c7..259a132 100644 --- a/src/barabadb/protocol/wire.nim +++ b/src/barabadb/protocol/wire.nim @@ -169,12 +169,14 @@ proc serializeValue*(buf: var seq[byte], val: WireValue) = of fkFloat32: var fl = val.float32Val var bytes: array[4, byte] - copyMem(addr bytes, unsafeAddr fl, 4) + var i32 = cast[int32](fl) + bigEndian32(addr bytes, unsafeAddr i32) buf.add(bytes) of fkFloat64: var fl = val.float64Val var bytes: array[8, byte] - copyMem(addr bytes, unsafeAddr fl, 8) + var i64 = cast[int64](fl) + bigEndian64(addr bytes, unsafeAddr i64) buf.add(bytes) of fkString: buf.writeString(val.strVal) of fkBytes: buf.writeBytes(val.bytesVal) @@ -192,7 +194,8 @@ proc serializeValue*(buf: var seq[byte], val: WireValue) = for f in val.vecVal: var fl = f var bytes: array[4, byte] - copyMem(addr bytes, unsafeAddr fl, 4) + var i32 = cast[int32](fl) + bigEndian32(addr bytes, unsafeAddr i32) buf.add(bytes) of fkJson: buf.writeString(val.jsonVal) @@ -227,14 +230,18 @@ proc deserializeValue*(buf: openArray[byte], pos: var int, depth: int = 0): Wire var fl: float32 var bytes: array[4, byte] for i in 0..3: bytes[i] = buf[pos + i] - copyMem(addr fl, unsafeAddr bytes, 4) + var i32: int32 + bigEndian32(addr i32, unsafeAddr bytes) + fl = cast[float32](i32) pos += 4 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] - copyMem(addr fl, unsafeAddr bytes, 8) + var i64: int64 + bigEndian64(addr i64, unsafeAddr bytes) + fl = cast[float64](i64) pos += 8 result = WireValue(kind: fkFloat64, float64Val: fl) of fkString: @@ -270,7 +277,9 @@ proc deserializeValue*(buf: openArray[byte], pos: var int, depth: int = 0): Wire var fl: float32 var bytes: array[4, byte] for j in 0..3: bytes[j] = buf[pos + j] - copyMem(addr fl, unsafeAddr bytes, 4) + var i32: int32 + bigEndian32(addr i32, unsafeAddr bytes) + fl = cast[float32](i32) pos += 4 vec.add(fl) result = WireValue(kind: fkVector, vecVal: vec) From 398769ff97832a02a5ec4144120339cb8419b610 Mon Sep 17 00:00:00 2001 From: dimgigov Date: Thu, 14 May 2026 02:22:07 +0300 Subject: [PATCH 2/3] feat(client): add TCP request queue for safe concurrency The previous implementation allowed multiple concurrent async calls (query, execute, ping) to interleave writes on the same socket, which corrupted the binary protocol framing when NodeBB fired parallel database operations. Add an internal _requestQueue and _requestLock so that all TCP requests are serialized: each async operation enqueues a task, and tasks are drained one at a time via setImmediate(). --- clients/javascript/baradb.js | 126 ++++++++++++++++++++++------------- 1 file changed, 78 insertions(+), 48 deletions(-) diff --git a/clients/javascript/baradb.js b/clients/javascript/baradb.js index c963a53..74cddef 100644 --- a/clients/javascript/baradb.js +++ b/clients/javascript/baradb.js @@ -244,6 +244,8 @@ class Client { this.requestId = 0; this._buffer = Buffer.alloc(0); this._pendingResolve = null; + this._requestQueue = []; + this._requestLock = false; } async connect() { @@ -478,70 +480,98 @@ class Client { throw new Error(`Unexpected auth response: 0x${header.kind.toString(16)}`); } + async _processQueue() { + if (this._requestLock || this._requestQueue.length === 0) return; + this._requestLock = true; + const { task, resolve, reject } = this._requestQueue.shift(); + try { + const result = await task(); + resolve(result); + } catch (err) { + reject(err); + } finally { + this._requestLock = false; + setImmediate(() => this._processQueue()); + } + } + + _enqueue(task) { + return new Promise((resolve, reject) => { + this._requestQueue.push({ task, resolve, reject }); + this._processQueue(); + }); + } + async ping() { if (!this.connected) throw new Error('Not connected'); - const msg = this._build(MsgKind.PING, Buffer.alloc(0)); - this.socket.write(msg); + return this._enqueue(async () => { + const msg = this._build(MsgKind.PING, Buffer.alloc(0)); + this.socket.write(msg); - const header = await this._readHeader(); - if (header.kind === MsgKind.PONG) return true; - if (header.kind === MsgKind.ERROR) throw await this._readError(header.length); - return false; + const header = await this._readHeader(); + if (header.kind === MsgKind.PONG) return true; + if (header.kind === MsgKind.ERROR) throw await this._readError(header.length); + return false; + }); } async query(sql) { if (!this.connected) throw new Error('Not connected'); - const queryBuf = Buffer.from(sql, 'utf-8'); - const payload = Buffer.alloc(4 + queryBuf.length + 1); - payload.writeUInt32BE(queryBuf.length, 0); - queryBuf.copy(payload, 4); - payload[4 + queryBuf.length] = ResultFormat.BINARY; + return this._enqueue(async () => { + const queryBuf = Buffer.from(sql, 'utf-8'); + const payload = Buffer.alloc(4 + queryBuf.length + 1); + payload.writeUInt32BE(queryBuf.length, 0); + queryBuf.copy(payload, 4); + payload[4 + queryBuf.length] = ResultFormat.BINARY; - const msg = this._build(MsgKind.QUERY, payload); - this.socket.write(msg); + const msg = this._build(MsgKind.QUERY, payload); + this.socket.write(msg); - const header = await this._readHeader(); - if (header.kind === MsgKind.ERROR) throw await this._readError(header.length); - if (header.kind === MsgKind.DATA) return await this._readDataResponse(header.length); - if (header.kind === MsgKind.COMPLETE) { - const data = await this._recvExact(header.length); - const result = new QueryResult(); - result.affectedRows = data.readUInt32BE(0); - return result; - } - return new QueryResult(); + const header = await this._readHeader(); + if (header.kind === MsgKind.ERROR) throw await this._readError(header.length); + if (header.kind === MsgKind.DATA) return await this._readDataResponse(header.length); + if (header.kind === MsgKind.COMPLETE) { + const data = await this._recvExact(header.length); + const result = new QueryResult(); + result.affectedRows = data.readUInt32BE(0); + return result; + } + return new QueryResult(); + }); } async queryParams(sql, params = []) { if (!this.connected) throw new Error('Not connected'); - const queryBuf = Buffer.from(sql, 'utf-8'); - const paramParts = []; - for (const p of params) { - paramParts.push(p.serialize()); - } - const paramsBuf = Buffer.concat(paramParts); + return this._enqueue(async () => { + const queryBuf = Buffer.from(sql, 'utf-8'); + const paramParts = []; + for (const p of params) { + paramParts.push(p.serialize()); + } + const paramsBuf = Buffer.concat(paramParts); - const payload = Buffer.alloc(4 + queryBuf.length + 1 + 4 + paramsBuf.length); - let pos = 0; - payload.writeUInt32BE(queryBuf.length, pos); pos += 4; - queryBuf.copy(payload, pos); pos += queryBuf.length; - payload[pos] = ResultFormat.BINARY; pos++; - payload.writeUInt32BE(params.length, pos); pos += 4; - paramsBuf.copy(payload, pos); + const payload = Buffer.alloc(4 + queryBuf.length + 1 + 4 + paramsBuf.length); + let pos = 0; + payload.writeUInt32BE(queryBuf.length, pos); pos += 4; + queryBuf.copy(payload, pos); pos += queryBuf.length; + payload[pos] = ResultFormat.BINARY; pos++; + payload.writeUInt32BE(params.length, pos); pos += 4; + paramsBuf.copy(payload, pos); - const msg = this._build(MsgKind.QUERY_PARAMS, payload); - this.socket.write(msg); + const msg = this._build(MsgKind.QUERY_PARAMS, payload); + this.socket.write(msg); - const header = await this._readHeader(); - if (header.kind === MsgKind.ERROR) throw await this._readError(header.length); - if (header.kind === MsgKind.DATA) return await this._readDataResponse(header.length); - if (header.kind === MsgKind.COMPLETE) { - const data = await this._recvExact(header.length); - const result = new QueryResult(); - result.affectedRows = data.readUInt32BE(0); - return result; - } - return new QueryResult(); + const header = await this._readHeader(); + if (header.kind === MsgKind.ERROR) throw await this._readError(header.length); + if (header.kind === MsgKind.DATA) return await this._readDataResponse(header.length); + if (header.kind === MsgKind.COMPLETE) { + const data = await this._recvExact(header.length); + const result = new QueryResult(); + result.affectedRows = data.readUInt32BE(0); + return result; + } + return new QueryResult(); + }); } async execute(sql) { From 71dcffeccec4f156c570fb1aac86c0753cb1dff5 Mon Sep 17 00:00:00 2001 From: dimgigov Date: Thu, 14 May 2026 02:22:07 +0300 Subject: [PATCH 3/3] fix(gossip): use async UDP socket to avoid blocking the event loop The gossip listener used synchronous newSocket + recvFrom, which blocks the async event loop and prevents other async operations from running while waiting for UDP packets. Switch to newAsyncSocket + async recvFrom so the loop can continue processing other tasks between gossip rounds. --- src/barabadb/core/gossip.nim | 15 ++++++--------- 1 file changed, 6 insertions(+), 9 deletions(-) diff --git a/src/barabadb/core/gossip.nim b/src/barabadb/core/gossip.nim index da6cbba..f2c37bf 100644 --- a/src/barabadb/core/gossip.nim +++ b/src/barabadb/core/gossip.nim @@ -4,6 +4,7 @@ import std/random import std/monotimes import std/asyncdispatch import std/net +import std/asyncnet import std/strutils import std/streams import std/os @@ -36,7 +37,7 @@ type fanout*: int # number of nodes to gossip to per round gossipPort*: int running*: bool - sock*: Socket + sock*: AsyncSocket onJoin*: proc(node: GossipNode) {.gcsafe.} onLeave*: proc(nodeId: string) {.gcsafe.} onSuspect*: proc(nodeId: string) {.gcsafe.} @@ -305,25 +306,21 @@ proc startGossipRound*(gp: GossipProtocol, intervalMs: int = 2000) {.async.} = gp.broadcastGossip() proc startGossipListener*(gp: GossipProtocol) {.async.} = - gp.sock = newSocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP) + gp.sock = newAsyncSocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP) gp.sock.setSockOpt(OptReuseAddr, true) gp.sock.bindAddr(Port(gp.gossipPort)) gp.running = true while gp.running: try: - var data = newString(65535) - var senderAddr = "" - var senderPort: Port - let bytesRead = gp.sock.recvFrom(data, 65535, senderAddr, senderPort) - if bytesRead > 0: - data.setLen(bytesRead) + let (data, senderAddr, senderPort) = await gp.sock.recvFrom(65535) + if data.len > 0: gp.handleIncomingGossip(data, senderAddr) except: # Small sleep on error to avoid spin gp.sock.close() # Recreate socket for next iteration try: - gp.sock = newSocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP) + gp.sock = newAsyncSocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP) gp.sock.setSockOpt(OptReuseAddr, true) gp.sock.bindAddr(Port(gp.gossipPort)) except: