From 42c675224c00c476979fa87b66b2b6d703ecfc10 Mon Sep 17 00:00:00 2001 From: dimgigov Date: Wed, 13 May 2026 14:01:13 +0300 Subject: [PATCH] fix: distributed transaction and sharding bugs - disttxn: rollback now correctly updates participant state (was modifying a copy) - disttxn: bare except replaced with CatchableError - sharding: migrateData condition was always false, now correctly migrates data - sharding: unsafe cast[string] replaced with manual byte-to-char conversion - sharding: bare except replaced with CatchableError - server: parseUInt replaced with parseBiggestUint for uint64 portability --- src/barabadb/core/disttxn.nim | 11 +++++------ src/barabadb/core/server.nim | 2 +- src/barabadb/core/sharding.nim | 14 ++++++-------- 3 files changed, 12 insertions(+), 15 deletions(-) diff --git a/src/barabadb/core/disttxn.nim b/src/barabadb/core/disttxn.nim index 7fc2537..5244c5e 100644 --- a/src/barabadb/core/disttxn.nim +++ b/src/barabadb/core/disttxn.nim @@ -105,7 +105,7 @@ proc sendDistTxnRpc(host: string, port: int, txnId: uint64, action: string, time sock.readLine(response) sock.close() return response.strip() == "OK" - except: + except CatchableError: return false proc prepare*(txn: DistributedTransaction): bool = @@ -133,11 +133,10 @@ proc prepare*(txn: DistributedTransaction): bool = else: # Rollback already-prepared participants to maintain atomicity for nodeId in preparedNodes: - var p = txn.participants[nodeId] - if p.host.len > 0 and p.port > 0: - discard sendDistTxnRpc(p.host, p.port, txn.id, "ROLLBACK") - p.prepared = false - p.aborted = true + if txn.participants[nodeId].host.len > 0 and txn.participants[nodeId].port > 0: + discard sendDistTxnRpc(txn.participants[nodeId].host, txn.participants[nodeId].port, txn.id, "ROLLBACK") + txn.participants[nodeId].prepared = false + txn.participants[nodeId].aborted = true txn.state = dtsAborted release(txn.lock) diff --git a/src/barabadb/core/server.nim b/src/barabadb/core/server.nim index 8bf0a9e..7ae8bb5 100644 --- a/src/barabadb/core/server.nim +++ b/src/barabadb/core/server.nim @@ -326,7 +326,7 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.} rest.add(more) let parts = rest.strip().split(" ") if parts.len >= 2: - let txnId = try: parseUInt(parts[0]) except: 0'u64 + let txnId = try: uint64(parseBiggestUint(parts[0])) except: 0'u64 let action = parts[1].toUpper() if server.distTxnManager != nil: let txn = server.distTxnManager.getTxn(txnId) diff --git a/src/barabadb/core/sharding.nim b/src/barabadb/core/sharding.nim index efb1ffa..587c63c 100644 --- a/src/barabadb/core/sharding.nim +++ b/src/barabadb/core/sharding.nim @@ -163,13 +163,15 @@ proc sendMigrationBatch(host: string, port: int, shardId: int, sock.send(header) for (key, value) in entries: # Use \0 as separator between key and value, \n as entry delimiter - let entry = key & "\0" & cast[string](value) & "\n" + var valStr = newString(value.len) + for i, b in value: valStr[i] = char(b) + let entry = key & "\0" & valStr & "\n" sock.send(entry) var response = "" sock.readLine(response) sock.close() return response.strip().startsWith("MIGRATE_OK") - except: + except CatchableError: return false proc migrateData*(router: var ShardRouter, nodes: seq[string], @@ -182,13 +184,9 @@ proc migrateData*(router: var ShardRouter, nodes: seq[string], if router.localNodeId.len == 0: return - # Find shards this node previously owned but no longer does - for i, shard in router.shards: - let wasOwner = router.localNodeId in shard.nodeIds - for shard in router.shards.mitems: - let nowOwner = router.localNodeId in shard.nodeIds - if not nowOwner and router.localNodeId in shard.nodeIds: + let isOwner = router.localNodeId in shard.nodeIds + if not isOwner: # We previously owned this shard but no longer do — ship data to new owner let entries = router.iterateKeys(shard.id) if entries.len > 0: