fix: harden exception handling, break ARC cycles, sync license/client
CI / test (push) Has been cancelled
CI / verify (push) Has been cancelled
Clients CI / build-server (push) Has been cancelled
Clients CI / test-python (push) Has been cancelled
Clients CI / test-javascript (push) Has been cancelled
Clients CI / test-nim (push) Has been cancelled
Clients CI / test-rust (push) Has been cancelled
CI / test (push) Has been cancelled
CI / verify (push) Has been cancelled
Clients CI / build-server (push) Has been cancelled
Clients CI / test-python (push) Has been cancelled
Clients CI / test-javascript (push) Has been cancelled
Clients CI / test-nim (push) Has been cancelled
Clients CI / test-rust (push) Has been cancelled
Replace bare except: with CatchableError across storage, query, Raft, backup, and protocol code so Defects are not swallowed. Break uncollectable ARC cycles in server shard/gossip callbacks via Server-owned refs and cursor locals. Align package license with LICENSE (BSD-3-Clause), sync README version, and point test_all at the canonical clients/nim baradb client (parseConnectionString + aliases).
This commit is contained in:
@@ -146,7 +146,7 @@ proc formatTimestamp*(ts: int64): string =
|
||||
try:
|
||||
let dt = fromUnix(ts)
|
||||
result = format(dt, "yyyy-MM-dd HH:mm:ss")
|
||||
except:
|
||||
except CatchableError:
|
||||
result = $ts
|
||||
|
||||
proc parseBackupFilename*(filename: string): int64 =
|
||||
@@ -163,7 +163,7 @@ proc parseBackupFilename*(filename: string): int64 =
|
||||
result = 0
|
||||
else:
|
||||
result = 0
|
||||
except:
|
||||
except CatchableError:
|
||||
result = 0
|
||||
|
||||
proc getArchiveSize*(input: string): int64 =
|
||||
@@ -175,7 +175,7 @@ proc getArchiveSize*(input: string): int64 =
|
||||
if exitCode == 0:
|
||||
try:
|
||||
result = parseBiggestInt(strip(outStr))
|
||||
except:
|
||||
except CatchableError:
|
||||
result = getFileSize(input) # fallback
|
||||
else:
|
||||
result = getFileSize(input)
|
||||
@@ -189,7 +189,7 @@ proc getFreeSpace*(path: string): int64 =
|
||||
if exitCode == 0:
|
||||
try:
|
||||
result = parseBiggestInt(strip(outStr))
|
||||
except:
|
||||
except CatchableError:
|
||||
result = -1
|
||||
else:
|
||||
result = -1
|
||||
@@ -701,14 +701,14 @@ when isMainModule:
|
||||
of "input", "i": target = val
|
||||
of "keep", "k":
|
||||
try: keepCount = parseInt(val)
|
||||
except: quit("ERROR: --keep must be a number", 1)
|
||||
except CatchableError: quit("ERROR: --keep must be a number", 1)
|
||||
of "exclude", "e": excludes.add(val)
|
||||
of "level", "l":
|
||||
try:
|
||||
compression = parseInt(val)
|
||||
if compression < 0 or compression > 9:
|
||||
quit("ERROR: --level must be between 0 and 9", 1)
|
||||
except: quit("ERROR: --level must be a number", 1)
|
||||
except CatchableError: quit("ERROR: --level must be a number", 1)
|
||||
of "dry-run": dryRun = true
|
||||
of "force", "f": force = true
|
||||
of "online": online = true
|
||||
|
||||
@@ -277,7 +277,7 @@ proc sendGossipUdp(gp: GossipProtocol, target: GossipNode, msg: GossipMessage) =
|
||||
let data = serialize(msg)
|
||||
sock.sendTo(target.host, Port(target.port), cast[string](data))
|
||||
sock.close()
|
||||
except:
|
||||
except CatchableError:
|
||||
discard
|
||||
|
||||
proc broadcastGossip(gp: GossipProtocol) =
|
||||
@@ -298,7 +298,7 @@ proc handleIncomingGossip(gp: GossipProtocol, data: string, senderAddr: string)
|
||||
let parts = host.split(":")
|
||||
host = parts[0]
|
||||
if parts[1].len > 0:
|
||||
port = try: parseInt(parts[1]) except: gp.gossipPort
|
||||
port = try: parseInt(parts[1]) except CatchableError: gp.gossipPort
|
||||
let newNode = GossipNode(
|
||||
id: msg.senderId, host: host, port: port,
|
||||
state: nsAlive, incarnation: msg.senderIncarnation,
|
||||
@@ -306,7 +306,7 @@ proc handleIncomingGossip(gp: GossipProtocol, data: string, senderAddr: string)
|
||||
)
|
||||
gp.addMember(newNode)
|
||||
gp.applyGossipMessage(msg)
|
||||
except:
|
||||
except CatchableError:
|
||||
discard
|
||||
|
||||
proc startHealthCheck*(gp: GossipProtocol, intervalMs: int = 1000) {.async.} =
|
||||
@@ -342,14 +342,14 @@ proc startGossipListener*(gp: GossipProtocol) {.async.} =
|
||||
# Recreate socket after too many errors
|
||||
try:
|
||||
gp.sock.close()
|
||||
except:
|
||||
except CatchableError:
|
||||
discard
|
||||
try:
|
||||
gp.sock = newAsyncSocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)
|
||||
gp.sock.setSockOpt(OptReuseAddr, true)
|
||||
gp.sock.bindAddr(Port(gp.gossipPort))
|
||||
consecutiveErrors = 0
|
||||
except:
|
||||
except CatchableError:
|
||||
break
|
||||
# Exponential backoff with cap
|
||||
let delayMs = min(baseRetryDelayMs * (1 shl min(consecutiveErrors, 6)), 5000)
|
||||
|
||||
@@ -103,7 +103,7 @@ proc verifyToken*(server: HttpServer, tokenStr: string): (bool, string, string)
|
||||
let userId = token.claims["sub"].node.str
|
||||
let role = if "role" in token.claims: token.claims["role"].node.str else: "user"
|
||||
return (true, userId, role)
|
||||
except:
|
||||
except CatchableError:
|
||||
return (false, "", "")
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
|
||||
@@ -202,7 +202,7 @@ proc applyCommitted(node: RaftNode) =
|
||||
let parts = entry.command.split(":")
|
||||
if parts.len >= 3:
|
||||
let action = parts[1]
|
||||
let txnId = try: parseUInt(parts[2]) except: 0'u64
|
||||
let txnId = try: parseUInt(parts[2]) except CatchableError: 0'u64
|
||||
if action == "PREPARE" and node.onDistTxnPrepare != nil:
|
||||
discard node.onDistTxnPrepare(txnId, @[])
|
||||
elif action == "COMMIT" and node.onDistTxnCommit != nil:
|
||||
@@ -564,7 +564,7 @@ proc connectToPeer(net: RaftNetwork, peerId: string) {.async.} =
|
||||
let sock = newAsyncSocket()
|
||||
await sock.connect(host, Port(port))
|
||||
net.peerSockets[peerId] = sock
|
||||
except:
|
||||
except CatchableError:
|
||||
discard
|
||||
|
||||
proc send*(net: RaftNetwork, peerId: string, msg: RaftMessage) {.async.} =
|
||||
@@ -577,7 +577,7 @@ proc send*(net: RaftNetwork, peerId: string, msg: RaftMessage) {.async.} =
|
||||
bigEndian32(addr header[0], unsafeAddr payloadLen)
|
||||
try:
|
||||
await net.peerSockets[peerId].send(cast[string](header) & cast[string](data))
|
||||
except:
|
||||
except CatchableError:
|
||||
net.peerSockets.del(peerId)
|
||||
|
||||
proc broadcast*(net: RaftNetwork, msgs: seq[RaftMessage]) {.async.} =
|
||||
@@ -615,9 +615,9 @@ proc receiveLoop(net: RaftNetwork, client: AsyncSocket) {.async.} =
|
||||
let msg = deserializeRaftMessage(payload)
|
||||
try:
|
||||
await net.processMessage(msg)
|
||||
except:
|
||||
except CatchableError:
|
||||
discard
|
||||
except:
|
||||
except CatchableError:
|
||||
discard
|
||||
finally:
|
||||
client.close()
|
||||
@@ -641,7 +641,7 @@ proc run*(net: RaftNetwork) {.async.} =
|
||||
try:
|
||||
let client = await net.socket.accept()
|
||||
asyncCheck net.receiveLoop(client)
|
||||
except:
|
||||
except CatchableError:
|
||||
break
|
||||
|
||||
proc stop*(net: RaftNetwork) =
|
||||
|
||||
@@ -266,12 +266,12 @@ proc healthCheck*(rm: ReplicationManager) =
|
||||
sock.readLine(response)
|
||||
if response.strip() != "PONG":
|
||||
connected = false
|
||||
except:
|
||||
except CatchableError:
|
||||
connected = false
|
||||
except:
|
||||
except CatchableError:
|
||||
connected = false
|
||||
finally:
|
||||
try: sock.close() except: discard
|
||||
try: sock.close() except CatchableError: discard
|
||||
|
||||
if not connected:
|
||||
acquire(rm.lock)
|
||||
|
||||
@@ -65,55 +65,53 @@ proc newServerWithRegistry*(config: BaraConfig, registry: DatabaseRegistry): Ser
|
||||
let tlsConfig = newTLSConfig(config.certFile, config.keyFile)
|
||||
tls = newTLSContext(tlsConfig)
|
||||
|
||||
# Initialize sharding
|
||||
let shardRouter = newShardRouter()
|
||||
# Initialize sharding / gossip. Server fields own the refs; locals used inside
|
||||
# callback closures are {.cursor.} so ARC does not form uncollectable cycles
|
||||
# (local + closure env + object callback fields).
|
||||
let localId = if config.raftNodeId.len > 0: config.raftNodeId else: "node-" & $config.port
|
||||
let cm = newClusterMembership(shardRouter, localId)
|
||||
|
||||
# Wire shard migration callbacks to LSM (use default database)
|
||||
shardRouter.iterateKeys = proc(shardId: int): seq[(string, seq[byte])] {.gcsafe.} =
|
||||
var entries: seq[(string, seq[byte])] = @[]
|
||||
for (key, value) in db.scanAll():
|
||||
if shardRouter.getShard(key) == shardId:
|
||||
entries.add((key, value))
|
||||
return entries
|
||||
|
||||
shardRouter.storeKeys = proc(shardId: int, entries: seq[(string, seq[byte])]) {.gcsafe.} =
|
||||
for (key, value) in entries:
|
||||
db.put(key, value)
|
||||
|
||||
shardRouter.deleteKeys = proc(keys: seq[string]) {.gcsafe.} =
|
||||
for key in keys:
|
||||
db.delete(key)
|
||||
|
||||
# Initialize gossip
|
||||
let gossipPort = config.raftPort + 100
|
||||
let gp = newGossipProtocol(localId, config.address, config.port, gossipPort = gossipPort)
|
||||
|
||||
# Wire gossip → cluster membership
|
||||
gp.onJoin = proc(node: GossipNode) {.gcsafe.} =
|
||||
cm.onNodeJoin(node.id, node.host, node.port)
|
||||
|
||||
gp.onLeave = proc(nodeId: string) {.gcsafe.} =
|
||||
cm.onNodeLeave(nodeId)
|
||||
|
||||
gp.onSuspect = proc(nodeId: string) {.gcsafe.} =
|
||||
cm.onNodeSuspect(nodeId)
|
||||
|
||||
# Initialize rate limiter
|
||||
let rl = newRateLimiter(rlaTokenBucket, config.rateLimitGlobal, config.rateLimitPerClient)
|
||||
|
||||
result = Server(config: config, running: false, db: db, ctx: ctx,
|
||||
registry: registry,
|
||||
txnManager: ctx.txnManager, distTxnManager: newDistTxnManager(),
|
||||
replicationManager: newReplicationManager(),
|
||||
shardRouter: shardRouter,
|
||||
clusterMembership: cm,
|
||||
gossipProtocol: gp,
|
||||
shardRouter: newShardRouter(),
|
||||
clusterMembership: nil,
|
||||
gossipProtocol: newGossipProtocol(localId, config.address, config.port, gossipPort = gossipPort),
|
||||
tls: tls,
|
||||
rateLimiter: rl)
|
||||
result.clusterMembership = newClusterMembership(result.shardRouter, localId)
|
||||
initLock(result.activeConnectionsLock)
|
||||
|
||||
# Wire shard migration callbacks to LSM (default database)
|
||||
block:
|
||||
let shardRouter {.cursor.} = result.shardRouter
|
||||
let dbRef {.cursor.} = db
|
||||
shardRouter.iterateKeys = proc(shardId: int): seq[(string, seq[byte])] {.gcsafe.} =
|
||||
var entries: seq[(string, seq[byte])] = @[]
|
||||
for (key, value) in dbRef.scanAll():
|
||||
if shardRouter.getShard(key) == shardId:
|
||||
entries.add((key, value))
|
||||
return entries
|
||||
shardRouter.storeKeys = proc(shardId: int, entries: seq[(string, seq[byte])]) {.gcsafe.} =
|
||||
for (key, value) in entries:
|
||||
dbRef.put(key, value)
|
||||
shardRouter.deleteKeys = proc(keys: seq[string]) {.gcsafe.} =
|
||||
for key in keys:
|
||||
dbRef.delete(key)
|
||||
|
||||
# Wire gossip → cluster membership
|
||||
block:
|
||||
let gp {.cursor.} = result.gossipProtocol
|
||||
let cm {.cursor.} = result.clusterMembership
|
||||
gp.onJoin = proc(node: GossipNode) {.gcsafe.} =
|
||||
cm.onNodeJoin(node.id, node.host, node.port)
|
||||
gp.onLeave = proc(nodeId: string) {.gcsafe.} =
|
||||
cm.onNodeLeave(nodeId)
|
||||
gp.onSuspect = proc(nodeId: string) {.gcsafe.} =
|
||||
cm.onNodeSuspect(nodeId)
|
||||
|
||||
proc newServerWithDb*(config: BaraConfig, db: LSMTree): Server =
|
||||
let registry = newDatabaseRegistry(config)
|
||||
let ctx = newExecutionContext(db, registry)
|
||||
@@ -389,7 +387,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: uint64(parseBiggestUint(parts[0])) except: 0'u64
|
||||
let txnId = try: uint64(parseBiggestUint(parts[0])) except CatchableError: 0'u64
|
||||
let action = parts[1].toUpper()
|
||||
if server.distTxnManager != nil:
|
||||
let txn = server.distTxnManager.getTxn(txnId)
|
||||
@@ -428,8 +426,8 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.}
|
||||
rest.add(more)
|
||||
let parts = rest.strip().split(" ")
|
||||
if parts.len >= 2:
|
||||
let lsn = try: parseUInt(parts[0]) except: 0'u64
|
||||
let dataLen = try: parseInt(parts[1]) except: 0
|
||||
let lsn = try: parseUInt(parts[0]) except CatchableError: 0'u64
|
||||
let dataLen = try: parseInt(parts[1]) except CatchableError: 0
|
||||
if dataLen > 0:
|
||||
var data = ""
|
||||
while data.len < dataLen:
|
||||
@@ -460,7 +458,7 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.}
|
||||
let headerLine = "MIGRATE " & rest.strip()
|
||||
let parts = rest.strip().split(" ")
|
||||
if parts.len >= 2:
|
||||
let entryCount = try: parseInt(parts[1]) except: 0
|
||||
let entryCount = try: parseInt(parts[1]) except CatchableError: 0
|
||||
var data = ""
|
||||
if entryCount > 0:
|
||||
# Read all entries (each entry is key\0value\n)
|
||||
@@ -551,7 +549,7 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.}
|
||||
# Shard-aware routing: check if this node should handle the write
|
||||
var shardCheck = true
|
||||
if server.clusterMembership.nodes.len > 0:
|
||||
let stmts = try: parse(tokenize(queryStr)) except: nil
|
||||
let stmts = try: parse(tokenize(queryStr)) except CatchableError: nil
|
||||
if stmts != nil:
|
||||
for stmt in stmts.stmts:
|
||||
if stmt.kind in {nkInsert, nkUpdate, nkDelete}:
|
||||
|
||||
@@ -337,8 +337,8 @@ proc handleMigrationMessage*(headerLine: string, data: string,
|
||||
if parts.len < 3:
|
||||
return "ERR invalid migrate header\n"
|
||||
|
||||
let shardId = try: parseInt(parts[1]) except: -1
|
||||
let entryCount = try: parseInt(parts[2]) except: 0
|
||||
let shardId = try: parseInt(parts[1]) except CatchableError: -1
|
||||
let entryCount = try: parseInt(parts[2]) except CatchableError: 0
|
||||
|
||||
if shardId < 0 or entryCount < 0:
|
||||
return "ERR invalid shard id or entry count\n"
|
||||
|
||||
@@ -128,5 +128,5 @@ proc exportOtlp*(tracer: Tracer, endpoint: string = "http://localhost:4318/v1/tr
|
||||
client.close()
|
||||
tracer.spans = @[]
|
||||
return true
|
||||
except:
|
||||
except CatchableError:
|
||||
return false
|
||||
|
||||
@@ -189,7 +189,7 @@ proc notifyClient(client: WsClient, msg: string) {.async.} =
|
||||
try:
|
||||
let frame = encodeFrame(0x1, msg)
|
||||
await client.socket.send(frame)
|
||||
except:
|
||||
except CatchableError:
|
||||
discard
|
||||
|
||||
proc broadcastToTable*(server: WsServer, table: string, msg: string) {.async.} =
|
||||
@@ -247,7 +247,7 @@ proc handleWsClient(server: WsServer, client: AsyncSocket, id: int) {.async.} =
|
||||
|
||||
buf = buf[consumed..^1]
|
||||
|
||||
except:
|
||||
except CatchableError:
|
||||
discard
|
||||
finally:
|
||||
echo "WebSocket client ", id, " disconnected"
|
||||
@@ -305,7 +305,7 @@ proc handleConnection(server: WsServer, client: AsyncSocket) {.async.} =
|
||||
await client.send("HTTP/1.1 401 Unauthorized\r\n\r\n")
|
||||
client.close()
|
||||
return
|
||||
except:
|
||||
except CatchableError:
|
||||
await client.send("HTTP/1.1 401 Unauthorized\r\n\r\n")
|
||||
client.close()
|
||||
return
|
||||
|
||||
Reference in New Issue
Block a user