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

@@ -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 txn.participants[nodeId].host.len > 0 and txn.participants[nodeId].port > 0:
-      if p.host.len > 0 and p.port > 0:
+        discard sendDistTxnRpc(txn.participants[nodeId].host, txn.participants[nodeId].port, txn.id, "ROLLBACK")
-        discard sendDistTxnRpc(p.host, p.port, txn.id, "ROLLBACK")
+      txn.participants[nodeId].prepared = false
-      p.prepared = false
+      txn.participants[nodeId].aborted = true
-      p.aborted = true
     txn.state = dtsAborted
 
   release(txn.lock)

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)

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
+    let isOwner = router.localNodeId in shard.nodeIds
-    if not nowOwner and 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: