From 965d67cb01b8b4718457d9ccfd2573d951375042 Mon Sep 17 00:00:00 2001 From: NagyZoltanPeter <113987313+NagyZoltanPeter@users.noreply.github.com> Date: Wed, 20 May 2026 12:08:29 +0200 Subject: [PATCH] feat: make Persistence interface async The 14 Persistence proc fields now return Future[...] with {.async: (raises: []), gcsafe.}, allowing real I/O backends (SQLite, encrypted file, network) to suspend rather than block the Chronos event loop the manager runs on. Propagates through: - ReliabilityManager.lock: system.Lock -> chronos.AsyncLock. Acquired across awaits cleanly; matches the single-threaded Chronos worker the FFI uses. Multi-OS-thread use is now explicitly the caller's responsibility. - sds_utils + sds.nim public API procs (wrapOutgoingMessage, unwrapReceivedMessage, markDependenciesMet, setCallbacks, resetReliabilityManager, cleanup, ensureChannel, removeChannel, the getter snapshots, etc.) are now async. - FFI request handlers in library/sds_thread/... await the new API. - Tests converted via an asyncTest template that wraps each test body in an async proc; setup/teardown use waitFor for their single async call (ensureChannel / cleanup). Lock scope is preserved exactly: the same call sites that held the kernel Lock today hold AsyncLock now -- no new locking added. Co-Authored-By: Claude Opus 4.7 --- .../requests/sds_dependencies_request.nim | 2 +- .../requests/sds_lifecycle_request.nim | 4 +- .../requests/sds_message_request.nim | 8 +- sds.nim | 487 +++++++++------- sds/sds_utils.nim | 201 ++++--- sds/types/persistence.nim | 117 ++-- sds/types/reliability_manager.nim | 11 +- tests/in_memory_persistence.nim | 55 +- tests/test_persistence.nim | 237 ++++---- tests/test_reliability.nim | 528 +++++++++--------- 10 files changed, 941 insertions(+), 709 deletions(-) diff --git a/library/sds_thread/inter_thread_communication/requests/sds_dependencies_request.nim b/library/sds_thread/inter_thread_communication/requests/sds_dependencies_request.nim index 49b3fd1..d3f73ff 100644 --- a/library/sds_thread/inter_thread_communication/requests/sds_dependencies_request.nim +++ b/library/sds_thread/inter_thread_communication/requests/sds_dependencies_request.nim @@ -43,7 +43,7 @@ proc process*( let messageIdsC = self.messageIds.toSeq() let messageIds = messageIdsC.mapIt($it) - markDependenciesMet(rm[], messageIds, $self.channelId).isOkOr: + (await markDependenciesMet(rm[], messageIds, $self.channelId)).isOkOr: error "MARK_DEPENDENCIES_MET failed", error = error return err("error processing MARK_DEPENDENCIES_MET request: " & $error) diff --git a/library/sds_thread/inter_thread_communication/requests/sds_lifecycle_request.nim b/library/sds_thread/inter_thread_communication/requests/sds_lifecycle_request.nim index 1ca6674..a5befc5 100644 --- a/library/sds_thread/inter_thread_communication/requests/sds_lifecycle_request.nim +++ b/library/sds_thread/inter_thread_communication/requests/sds_lifecycle_request.nim @@ -41,7 +41,7 @@ proc createReliabilityManager( error "Failed creating reliability manager", error = error return err("Failed creating reliability manager: " & $error) - rm.setCallbacks( + await rm.setCallbacks( appCallbacks.messageReadyCb, appCallbacks.messageSentCb, appCallbacks.missingDependenciesCb, appCallbacks.periodicSyncCb, appCallbacks.retrievalHintProvider, appCallbacks.repairReadyCb, @@ -61,7 +61,7 @@ proc process*( error "CREATE_RELIABILITY_MANAGER failed", error = error return err("error processing CREATE_RELIABILITY_MANAGER request: " & $error) of RESET_RELIABILITY_MANAGER: - resetReliabilityManager(rm[]).isOkOr: + (await resetReliabilityManager(rm[])).isOkOr: error "RESET_RELIABILITY_MANAGER failed", error = error return err("error processing RESET_RELIABILITY_MANAGER request: " & $error) of START_PERIODIC_TASKS: diff --git a/library/sds_thread/inter_thread_communication/requests/sds_message_request.nim b/library/sds_thread/inter_thread_communication/requests/sds_message_request.nim index 89d283d..97ab7cb 100644 --- a/library/sds_thread/inter_thread_communication/requests/sds_message_request.nim +++ b/library/sds_thread/inter_thread_communication/requests/sds_message_request.nim @@ -53,7 +53,9 @@ proc process*( of WRAP_MESSAGE: let messageBytes = self.message.toSeq() - let wrappedMessage = wrapOutgoingMessage(rm[], messageBytes, $self.messageId, $self.channelId).valueOr: + let wrappedMessage = ( + await wrapOutgoingMessage(rm[], messageBytes, $self.messageId, $self.channelId) + ).valueOr: error "WRAP_MESSAGE failed", error = error return err("error processing WRAP_MESSAGE request: " & $error) @@ -62,7 +64,9 @@ proc process*( of UNWRAP_MESSAGE: let messageBytes = self.message.toSeq() - let (unwrappedMessage, missingDeps, extractedChannelId) = unwrapReceivedMessage(rm[], messageBytes).valueOr: + let (unwrappedMessage, missingDeps, extractedChannelId) = ( + await unwrapReceivedMessage(rm[], messageBytes) + ).valueOr: return err("error processing UNWRAP_MESSAGE request: " & $error) let res = SdsUnwrapResponse(message: unwrappedMessage, missingDeps: missingDeps, channelId: extractedChannelId) diff --git a/sds.nim b/sds.nim index 5e1ba09..e426637 100644 --- a/sds.nim +++ b/sds.nim @@ -1,4 +1,4 @@ -import std/[algorithm, times, locks, tables, sets, options] +import std/[algorithm, times, tables, sets, options] import chronos, results, chronicles import sds/[types, protobuf, sds_utils, rolling_bloom_filter] @@ -33,165 +33,191 @@ proc isAcknowledged*( return false -proc reviewAckStatus(rm: ReliabilityManager, msg: SdsMessage) {.gcsafe.} = - var rbf: Option[RollingBloomFilter] - if msg.bloomFilter.len > 0: - let bfResult = deserializeBloomFilter(msg.bloomFilter) - if bfResult.isOk(): - let bf = bfResult.get() - rbf = some( - RollingBloomFilter.init( - filter = bf, - capacity = bf.capacity, - minCapacity = (bf.capacity.float * (100 - CapacityFlexPercent).float / 100.0).int, - maxCapacity = (bf.capacity.float * (100 + CapacityFlexPercent).float / 100.0).int, +proc reviewAckStatus( + rm: ReliabilityManager, msg: SdsMessage +): Future[void] {.async: (raises: []), gcsafe.} = + try: + var rbf: Option[RollingBloomFilter] + if msg.bloomFilter.len > 0: + let bfResult = deserializeBloomFilter(msg.bloomFilter) + if bfResult.isOk(): + let bf = bfResult.get() + rbf = some( + RollingBloomFilter.init( + filter = bf, + capacity = bf.capacity, + minCapacity = (bf.capacity.float * (100 - CapacityFlexPercent).float / 100.0).int, + maxCapacity = (bf.capacity.float * (100 + CapacityFlexPercent).float / 100.0).int, + ) ) - ) + else: + error "Failed to deserialize bloom filter", error = bfResult.error + rbf = none[RollingBloomFilter]() else: - error "Failed to deserialize bloom filter", error = bfResult.error rbf = none[RollingBloomFilter]() - else: - rbf = none[RollingBloomFilter]() - if msg.channelId notin rm.channels: - return + if msg.channelId notin rm.channels: + return - let channel = rm.channels[msg.channelId] - var toDelete: seq[(int, SdsMessageID)] = @[] - var i = 0 + let channel = rm.channels[msg.channelId] + var toDelete: seq[(int, SdsMessageID)] = @[] + var i = 0 - while i < channel.outgoingBuffer.len: - let outMsg = channel.outgoingBuffer[i] - if outMsg.isAcknowledged(msg.causalHistory, rbf): - if not rm.onMessageSent.isNil(): - rm.onMessageSent(outMsg.message.messageId, outMsg.message.channelId) - toDelete.add((i, outMsg.message.messageId)) - inc i + while i < channel.outgoingBuffer.len: + let outMsg = channel.outgoingBuffer[i] + if outMsg.isAcknowledged(msg.causalHistory, rbf): + if not rm.onMessageSent.isNil(): + {.cast(raises: []).}: + rm.onMessageSent(outMsg.message.messageId, outMsg.message.channelId) + toDelete.add((i, outMsg.message.messageId)) + inc i - for k in countdown(toDelete.high, 0): - let (idx, ackedId) = toDelete[k] - channel.outgoingBuffer.delete(idx) - rm.persistence.removeOutgoing(msg.channelId, ackedId) + for k in countdown(toDelete.high, 0): + let (idx, ackedId) = toDelete[k] + channel.outgoingBuffer.delete(idx) + await rm.persistence.removeOutgoing(msg.channelId, ackedId) + except CatchableError: + error "Failed to review ack status", msg = getCurrentExceptionMsg() proc wrapOutgoingMessage*( rm: ReliabilityManager, message: seq[byte], messageId: SdsMessageID, channelId: SdsChannelID, -): Result[seq[byte], ReliabilityError] = +): Future[Result[seq[byte], ReliabilityError]] {.async: (raises: []), gcsafe.} = ## Wraps an outgoing message with reliability metadata. if message.len == 0: return err(ReliabilityError.reInvalidArgument) if message.len > MaxMessageSize: return err(ReliabilityError.reMessageTooLarge) - withLock rm.lock: + try: + await rm.lock.acquire() try: - let channel = rm.getOrCreateChannel(channelId) - rm.updateLamportTimestamp(getTime().toUnix, channelId) + try: + let channel = await rm.getOrCreateChannel(channelId) + await rm.updateLamportTimestamp(getTime().toUnix, channelId) - let bfResult = serializeBloomFilter(channel.bloomFilter.filter) - if bfResult.isErr: - error "Failed to serialize bloom filter", channelId = channelId + let bfResult = serializeBloomFilter(channel.bloomFilter.filter) + if bfResult.isErr: + error "Failed to serialize bloom filter", channelId = channelId + return err(ReliabilityError.reSerializationError) + + # SDS-R: collect eligible expired repair requests to attach. Per + # spec (sds-r-send-message, RECOMMENDED), prioritise the entries with + # the smallest minTimeRepairReq — they are the most overdue and the + # ones the network most needs us to ask about. + var repairReqs: seq[HistoryEntry] = @[] + let now = getTime() + var expiredKeys: seq[SdsMessageID] = @[] + var eligible: seq[(SdsMessageID, OutgoingRepairEntry)] = @[] + for msgId, repairEntry in channel.outgoingRepairBuffer: + if now >= repairEntry.minTimeRepairReq: + eligible.add((msgId, repairEntry)) + eligible.sort do(a, b: (SdsMessageID, OutgoingRepairEntry)) -> int: + cmp(a[1].minTimeRepairReq, b[1].minTimeRepairReq) + let take = min(eligible.len, rm.config.maxRepairRequests) + for i in 0 ..< take: + repairReqs.add(eligible[i][1].outHistEntry) + expiredKeys.add(eligible[i][0]) + for key in expiredKeys: + channel.outgoingRepairBuffer.del(key) + await rm.persistence.removeOutgoingRepair(channelId, key) + + let msg = SdsMessage.init( + messageId = messageId, + lamportTimestamp = channel.lamportTimestamp, + causalHistory = + await rm.getRecentHistoryEntries(rm.config.maxCausalHistory, channelId), + channelId = channelId, + content = message, + bloomFilter = bfResult.get(), + senderId = rm.participantId, + repairRequest = repairReqs, + ) + + let unackMsg = UnacknowledgedMessage.init( + message = msg, sendTime = getTime(), resendAttempts = 0 + ) + channel.outgoingBuffer.add(unackMsg) + await rm.persistence.saveOutgoing(channelId, unackMsg) + + channel.bloomFilter.add(msg.messageId) + # The full SdsMessage carries senderId and content, so a single + # addToHistory replaces the old triple-write to messageHistory, + # messageCache, and messageSenders. + await rm.addToHistory(msg, channelId) + + return serializeMessage(msg) + except CatchableError: + error "Failed to wrap message", + channelId = channelId, msg = getCurrentExceptionMsg() return err(ReliabilityError.reSerializationError) + finally: + rm.lock.release() + except CatchableError: + error "Failed to wrap message (lock)", + channelId = channelId, msg = getCurrentExceptionMsg() + return err(ReliabilityError.reSerializationError) - # SDS-R: collect eligible expired repair requests to attach. Per - # spec (sds-r-send-message, RECOMMENDED), prioritise the entries with - # the smallest minTimeRepairReq — they are the most overdue and the - # ones the network most needs us to ask about. - var repairReqs: seq[HistoryEntry] = @[] - let now = getTime() - var expiredKeys: seq[SdsMessageID] = @[] - var eligible: seq[(SdsMessageID, OutgoingRepairEntry)] = @[] - for msgId, repairEntry in channel.outgoingRepairBuffer: - if now >= repairEntry.minTimeRepairReq: - eligible.add((msgId, repairEntry)) - eligible.sort do(a, b: (SdsMessageID, OutgoingRepairEntry)) -> int: - cmp(a[1].minTimeRepairReq, b[1].minTimeRepairReq) - let take = min(eligible.len, rm.config.maxRepairRequests) - for i in 0 ..< take: - repairReqs.add(eligible[i][1].outHistEntry) - expiredKeys.add(eligible[i][0]) - for key in expiredKeys: - channel.outgoingRepairBuffer.del(key) - rm.persistence.removeOutgoingRepair(channelId, key) +proc processIncomingBuffer( + rm: ReliabilityManager, channelId: SdsChannelID +): Future[void] {.async: (raises: []), gcsafe.} = + try: + await rm.lock.acquire() + try: + if channelId notin rm.channels: + error "Channel does not exist", channelId = channelId + return - let msg = SdsMessage.init( - messageId = messageId, - lamportTimestamp = channel.lamportTimestamp, - causalHistory = rm.getRecentHistoryEntries(rm.config.maxCausalHistory, channelId), - channelId = channelId, - content = message, - bloomFilter = bfResult.get(), - senderId = rm.participantId, - repairRequest = repairReqs, - ) + let channel = rm.channels[channelId] + if channel.incomingBuffer.len == 0: + return - let unackMsg = - UnacknowledgedMessage.init(message = msg, sendTime = getTime(), resendAttempts = 0) - channel.outgoingBuffer.add(unackMsg) - rm.persistence.saveOutgoing(channelId, unackMsg) + var processed = initHashSet[SdsMessageID]() + var readyToProcess = newSeq[SdsMessageID]() - channel.bloomFilter.add(msg.messageId) - # The full SdsMessage carries senderId and content, so a single - # addToHistory replaces the old triple-write to messageHistory, - # messageCache, and messageSenders. - rm.addToHistory(msg, channelId) + for msgId, entry in channel.incomingBuffer: + if entry.missingDeps.len == 0: + readyToProcess.add(msgId) - return serializeMessage(msg) - except Exception: - error "Failed to wrap message", - channelId = channelId, msg = getCurrentExceptionMsg() - return err(ReliabilityError.reSerializationError) + while readyToProcess.len > 0: + let msgId = readyToProcess.pop() + if msgId in processed: + continue -proc processIncomingBuffer(rm: ReliabilityManager, channelId: SdsChannelID) {.gcsafe.} = - withLock rm.lock: - if channelId notin rm.channels: - error "Channel does not exist", channelId = channelId - return + if msgId in channel.incomingBuffer: + await rm.addToHistory(channel.incomingBuffer[msgId].message, channelId) + if not rm.onMessageReady.isNil(): + {.cast(raises: []).}: + rm.onMessageReady(msgId, channelId) + processed.incl(msgId) - let channel = rm.channels[channelId] - if channel.incomingBuffer.len == 0: - return + for remainingId, entry in channel.incomingBuffer: + if remainingId notin processed: + if msgId in entry.missingDeps: + channel.incomingBuffer[remainingId].missingDeps.excl(msgId) + await rm.persistence.saveIncoming( + channelId, channel.incomingBuffer[remainingId] + ) + if channel.incomingBuffer[remainingId].missingDeps.len == 0: + readyToProcess.add(remainingId) - var processed = initHashSet[SdsMessageID]() - var readyToProcess = newSeq[SdsMessageID]() - - for msgId, entry in channel.incomingBuffer: - if entry.missingDeps.len == 0: - readyToProcess.add(msgId) - - while readyToProcess.len > 0: - let msgId = readyToProcess.pop() - if msgId in processed: - continue - - if msgId in channel.incomingBuffer: - rm.addToHistory(channel.incomingBuffer[msgId].message, channelId) - if not rm.onMessageReady.isNil(): - rm.onMessageReady(msgId, channelId) - processed.incl(msgId) - - for remainingId, entry in channel.incomingBuffer: - if remainingId notin processed: - if msgId in entry.missingDeps: - channel.incomingBuffer[remainingId].missingDeps.excl(msgId) - rm.persistence.saveIncoming( - channelId, channel.incomingBuffer[remainingId] - ) - if channel.incomingBuffer[remainingId].missingDeps.len == 0: - readyToProcess.add(remainingId) - - for msgId in processed: - channel.incomingBuffer.del(msgId) - rm.persistence.removeIncoming(channelId, msgId) + for msgId in processed: + channel.incomingBuffer.del(msgId) + await rm.persistence.removeIncoming(channelId, msgId) + finally: + rm.lock.release() + except CatchableError: + error "Failed to process incoming buffer", + channelId = channelId, msg = getCurrentExceptionMsg() proc unwrapReceivedMessage*( rm: ReliabilityManager, message: seq[byte] -): Result[ +): Future[Result[ tuple[message: seq[byte], missingDeps: seq[HistoryEntry], channelId: SdsChannelID], ReliabilityError, -] = +]] {.async: (raises: []), gcsafe.} = ## Unwraps a received message and processes its reliability metadata. try: let channelId = extractChannelId(message).valueOr: @@ -200,22 +226,22 @@ proc unwrapReceivedMessage*( let msg = deserializeMessage(message).valueOr: return err(ReliabilityError.reDeserializationError) - let channel = rm.getOrCreateChannel(channelId) + let channel = await rm.getOrCreateChannel(channelId) # SDS-R: opportunistic repair-buffer cleanup — applies to duplicates too, # so rebroadcasts cancel redundant responses on peers that already have the message. channel.outgoingRepairBuffer.del(msg.messageId) - rm.persistence.removeOutgoingRepair(channelId, msg.messageId) + await rm.persistence.removeOutgoingRepair(channelId, msg.messageId) channel.incomingRepairBuffer.del(msg.messageId) - rm.persistence.removeIncomingRepair(channelId, msg.messageId) + await rm.persistence.removeIncomingRepair(channelId, msg.messageId) if msg.messageId in channel.messageHistory: return ok((msg.content, @[], channelId)) channel.bloomFilter.add(msg.messageId) - rm.updateLamportTimestamp(msg.lamportTimestamp, channelId) - rm.reviewAckStatus(msg) + await rm.updateLamportTimestamp(msg.lamportTimestamp, channelId) + await rm.reviewAckStatus(msg) # SDS-R: process incoming repair requests from this message. We can only # answer for messages we have actually delivered (i.e. that live in @@ -225,7 +251,7 @@ proc unwrapReceivedMessage*( for repairEntry in msg.repairRequest: # Remove from our own outgoing repair buffer (someone else is also requesting) channel.outgoingRepairBuffer.del(repairEntry.messageId) - rm.persistence.removeOutgoingRepair(channelId, repairEntry.messageId) + await rm.persistence.removeOutgoingRepair(channelId, repairEntry.messageId) if repairEntry.messageId in channel.messageHistory and rm.participantId.len > 0 and repairEntry.senderId.len > 0: if isInResponseGroup( @@ -244,7 +270,9 @@ proc unwrapReceivedMessage*( minTimeRepairResp: now + tResp, ) channel.incomingRepairBuffer[repairEntry.messageId] = inEntry - rm.persistence.saveIncomingRepair(channelId, repairEntry.messageId, inEntry) + await rm.persistence.saveIncomingRepair( + channelId, repairEntry.messageId, inEntry + ) var missingDeps = rm.checkDependencies(msg.causalHistory, channelId) @@ -258,9 +286,9 @@ proc unwrapReceivedMessage*( let entry = IncomingMessage.init(message = msg, missingDeps = initHashSet[SdsMessageID]()) channel.incomingBuffer[msg.messageId] = entry - rm.persistence.saveIncoming(channelId, entry) + await rm.persistence.saveIncoming(channelId, entry) else: - rm.addToHistory(msg, channelId) + await rm.addToHistory(msg, channelId) # Unblock any buffered messages that were waiting on this one. var unblocked: seq[SdsMessageID] = @[] for pendingId, entry in channel.incomingBuffer: @@ -268,19 +296,21 @@ proc unwrapReceivedMessage*( channel.incomingBuffer[pendingId].missingDeps.excl(msg.messageId) unblocked.add(pendingId) for pendingId in unblocked: - rm.persistence.saveIncoming(channelId, channel.incomingBuffer[pendingId]) - rm.processIncomingBuffer(channelId) + await rm.persistence.saveIncoming(channelId, channel.incomingBuffer[pendingId]) + await rm.processIncomingBuffer(channelId) if not rm.onMessageReady.isNil(): - rm.onMessageReady(msg.messageId, channelId) + {.cast(raises: []).}: + rm.onMessageReady(msg.messageId, channelId) else: let entry = IncomingMessage.init( message = msg, missingDeps = missingDeps.getMessageIds().toHashSet(), ) channel.incomingBuffer[msg.messageId] = entry - rm.persistence.saveIncoming(channelId, entry) + await rm.persistence.saveIncoming(channelId, entry) if not rm.onMissingDependencies.isNil(): - rm.onMissingDependencies(msg.messageId, missingDeps, channelId) + {.cast(raises: []).}: + rm.onMissingDependencies(msg.messageId, missingDeps, channelId) # SDS-R: add missing deps to outgoing repair buffer if rm.participantId.len > 0: @@ -295,16 +325,16 @@ proc unwrapReceivedMessage*( minTimeRepairReq: now + tReq, ) channel.outgoingRepairBuffer[dep.messageId] = outEntry - rm.persistence.saveOutgoingRepair(channelId, dep.messageId, outEntry) + await rm.persistence.saveOutgoingRepair(channelId, dep.messageId, outEntry) return ok((msg.content, missingDeps, channelId)) - except Exception: + except CatchableError: error "Failed to unwrap message", msg = getCurrentExceptionMsg() return err(ReliabilityError.reDeserializationError) proc markDependenciesMet*( rm: ReliabilityManager, messageIds: seq[SdsMessageID], channelId: SdsChannelID -): Result[void, ReliabilityError] = +): Future[Result[void, ReliabilityError]] {.async: (raises: []), gcsafe.} = ## Marks the specified message dependencies as met. try: if channelId notin rm.channels: @@ -322,17 +352,17 @@ proc markDependenciesMet*( channel.incomingBuffer[pendingId].missingDeps.excl(msgId) unblocked.add(pendingId) for pendingId in unblocked: - rm.persistence.saveIncoming(channelId, channel.incomingBuffer[pendingId]) + await rm.persistence.saveIncoming(channelId, channel.incomingBuffer[pendingId]) # SDS-R: clear from repair buffers (dependency now met) channel.outgoingRepairBuffer.del(msgId) - rm.persistence.removeOutgoingRepair(channelId, msgId) + await rm.persistence.removeOutgoingRepair(channelId, msgId) channel.incomingRepairBuffer.del(msgId) - rm.persistence.removeIncomingRepair(channelId, msgId) + await rm.persistence.removeIncomingRepair(channelId, msgId) - rm.processIncomingBuffer(channelId) + await rm.processIncomingBuffer(channelId) return ok() - except Exception: + except CatchableError: error "Failed to mark dependencies as met", channelId = channelId, msg = getCurrentExceptionMsg() return err(ReliabilityError.reInternalError) @@ -345,45 +375,59 @@ proc setCallbacks*( onPeriodicSync: PeriodicSyncCallback = nil, onRetrievalHint: RetrievalHintProvider = nil, onRepairReady: RepairReadyCallback = nil, -) = +): Future[void] {.async: (raises: []), gcsafe.} = ## Sets the callback functions for various events in the ReliabilityManager. - withLock rm.lock: - rm.onMessageReady = onMessageReady - rm.onMessageSent = onMessageSent - rm.onMissingDependencies = onMissingDependencies - rm.onPeriodicSync = onPeriodicSync - rm.onRetrievalHint = onRetrievalHint - rm.onRepairReady = onRepairReady + try: + await rm.lock.acquire() + try: + rm.onMessageReady = onMessageReady + rm.onMessageSent = onMessageSent + rm.onMissingDependencies = onMissingDependencies + rm.onPeriodicSync = onPeriodicSync + rm.onRetrievalHint = onRetrievalHint + rm.onRepairReady = onRepairReady + finally: + rm.lock.release() + except CatchableError: + error "Failed to set callbacks", msg = getCurrentExceptionMsg() proc checkUnacknowledgedMessages( rm: ReliabilityManager, channelId: SdsChannelID -) {.gcsafe.} = - withLock rm.lock: - if channelId notin rm.channels: - error "Channel does not exist", channelId = channelId - return +): Future[void] {.async: (raises: []), gcsafe.} = + try: + await rm.lock.acquire() + try: + if channelId notin rm.channels: + error "Channel does not exist", channelId = channelId + return - let channel = rm.channels[channelId] - let now = getTime() - var newOutgoingBuffer: seq[UnacknowledgedMessage] = @[] + let channel = rm.channels[channelId] + let now = getTime() + var newOutgoingBuffer: seq[UnacknowledgedMessage] = @[] - for unackMsg in channel.outgoingBuffer: - let elapsed = now - unackMsg.sendTime - if elapsed > rm.config.resendInterval: - if unackMsg.resendAttempts < rm.config.maxResendAttempts: - var updatedMsg = unackMsg - updatedMsg.resendAttempts += 1 - updatedMsg.sendTime = now - newOutgoingBuffer.add(updatedMsg) - rm.persistence.saveOutgoing(channelId, updatedMsg) + for unackMsg in channel.outgoingBuffer: + let elapsed = now - unackMsg.sendTime + if elapsed > rm.config.resendInterval: + if unackMsg.resendAttempts < rm.config.maxResendAttempts: + var updatedMsg = unackMsg + updatedMsg.resendAttempts += 1 + updatedMsg.sendTime = now + newOutgoingBuffer.add(updatedMsg) + await rm.persistence.saveOutgoing(channelId, updatedMsg) + else: + if not rm.onMessageSent.isNil(): + {.cast(raises: []).}: + rm.onMessageSent(unackMsg.message.messageId, channelId) + await rm.persistence.removeOutgoing(channelId, unackMsg.message.messageId) else: - if not rm.onMessageSent.isNil(): - rm.onMessageSent(unackMsg.message.messageId, channelId) - rm.persistence.removeOutgoing(channelId, unackMsg.message.messageId) - else: - newOutgoingBuffer.add(unackMsg) + newOutgoingBuffer.add(unackMsg) - channel.outgoingBuffer = newOutgoingBuffer + channel.outgoingBuffer = newOutgoingBuffer + finally: + rm.lock.release() + except CatchableError: + error "Failed to check unacknowledged messages", + channelId = channelId, msg = getCurrentExceptionMsg() proc periodicBufferSweep( rm: ReliabilityManager @@ -391,13 +435,11 @@ proc periodicBufferSweep( while true: try: for channelId, channel in rm.channels: - try: - rm.checkUnacknowledgedMessages(channelId) - rm.cleanBloomFilter(channelId) - except Exception: - error "Error in buffer sweep for channel", - channelId = channelId, msg = getCurrentExceptionMsg() - except Exception: + await rm.checkUnacknowledgedMessages(channelId) + await rm.cleanBloomFilter(channelId) + except CancelledError as e: + raise e + except CatchableError: error "Error in periodic buffer sweep", msg = getCurrentExceptionMsg() await sleepAsync(chronos.milliseconds(rm.config.bufferSweepInterval.inMilliseconds)) @@ -408,19 +450,23 @@ proc periodicSyncMessage( while true: try: if not rm.onPeriodicSync.isNil(): - rm.onPeriodicSync() - except Exception: + {.cast(raises: []).}: + rm.onPeriodicSync() + except CatchableError: error "Error in periodic sync", msg = getCurrentExceptionMsg() await sleepAsync(chronos.seconds(rm.config.syncMessageInterval.inSeconds)) -proc runRepairSweep*(rm: ReliabilityManager) {.gcsafe, raises: [].} = +proc runRepairSweep*( + rm: ReliabilityManager +): Future[void] {.async: (raises: []), gcsafe.} = ## SDS-R: Runs a single pass of the repair sweep. ## - Incoming: fires onRepairReady for expired T_resp entries and removes them ## - Outgoing: drops entries past T_max window ## Exposed so it can be driven directly in tests; also invoked by periodicRepairSweep. ## Acquires rm.lock so the repair buffers cannot be observed mid-mutation by - ## a concurrent wrapOutgoingMessage / unwrapReceivedMessage on another thread. - withLock rm.lock: + ## a concurrent wrapOutgoingMessage / unwrapReceivedMessage on another task. + try: + await rm.lock.acquire() try: let now = getTime() for channelId, channel in rm.channels: @@ -434,9 +480,10 @@ proc runRepairSweep*(rm: ReliabilityManager) {.gcsafe, raises: [].} = for msgId in toRebroadcast: let entry = channel.incomingRepairBuffer[msgId] channel.incomingRepairBuffer.del(msgId) - rm.persistence.removeIncomingRepair(channelId, msgId) + await rm.persistence.removeIncomingRepair(channelId, msgId) if not rm.onRepairReady.isNil(): - rm.onRepairReady(entry.cachedMessage, channelId) + {.cast(raises: []).}: + rm.onRepairReady(entry.cachedMessage, channelId) # Drop expired outgoing repair entries past T_max var toRemove: seq[SdsMessageID] = @[] @@ -446,19 +493,21 @@ proc runRepairSweep*(rm: ReliabilityManager) {.gcsafe, raises: [].} = toRemove.add(msgId) for msgId in toRemove: channel.outgoingRepairBuffer.del(msgId) - rm.persistence.removeOutgoingRepair(channelId, msgId) - except Exception: + await rm.persistence.removeOutgoingRepair(channelId, msgId) + except CatchableError: error "Error in repair sweep for channel", channelId = channelId, msg = getCurrentExceptionMsg() - except Exception: - error "Error in repair sweep", msg = getCurrentExceptionMsg() + finally: + rm.lock.release() + except CatchableError: + error "Error in repair sweep", msg = getCurrentExceptionMsg() proc periodicRepairSweep( rm: ReliabilityManager ) {.async: (raises: [CancelledError]), gcsafe.} = ## SDS-R: Periodically checks repair buffers for expired entries. while true: - rm.runRepairSweep() + await rm.runRepairSweep() await sleepAsync(chronos.milliseconds(rm.config.repairSweepInterval.inMilliseconds)) proc startPeriodicTasks*(rm: ReliabilityManager) = @@ -467,22 +516,32 @@ proc startPeriodicTasks*(rm: ReliabilityManager) = asyncSpawn rm.periodicSyncMessage() asyncSpawn rm.periodicRepairSweep() -proc resetReliabilityManager*(rm: ReliabilityManager): Result[void, ReliabilityError] = +proc resetReliabilityManager*( + rm: ReliabilityManager +): Future[Result[void, ReliabilityError]] {.async: (raises: []), gcsafe.} = ## Resets the ReliabilityManager to its initial state. - withLock rm.lock: + try: + await rm.lock.acquire() try: - for channelId, channel in rm.channels: - rm.dropChannelFromPersistence(channelId) - channel.lamportTimestamp = 0 - channel.messageHistory.clear() - channel.outgoingBuffer.setLen(0) - channel.incomingBuffer.clear() - channel.outgoingRepairBuffer.clear() - channel.incomingRepairBuffer.clear() - channel.bloomFilter = - RollingBloomFilter.init(rm.config.bloomFilterCapacity, rm.config.bloomFilterErrorRate) - rm.channels.clear() - return ok() - except Exception: - error "Failed to reset ReliabilityManager", msg = getCurrentExceptionMsg() - return err(ReliabilityError.reInternalError) + try: + for channelId, channel in rm.channels: + await rm.dropChannelFromPersistence(channelId) + channel.lamportTimestamp = 0 + channel.messageHistory.clear() + channel.outgoingBuffer.setLen(0) + channel.incomingBuffer.clear() + channel.outgoingRepairBuffer.clear() + channel.incomingRepairBuffer.clear() + channel.bloomFilter = RollingBloomFilter.init( + rm.config.bloomFilterCapacity, rm.config.bloomFilterErrorRate + ) + rm.channels.clear() + return ok() + except CatchableError: + error "Failed to reset ReliabilityManager", msg = getCurrentExceptionMsg() + return err(ReliabilityError.reInternalError) + finally: + rm.lock.release() + except CatchableError: + error "Failed to reset ReliabilityManager (lock)", msg = getCurrentExceptionMsg() + return err(ReliabilityError.reInternalError) diff --git a/sds/sds_utils.nim b/sds/sds_utils.nim index 394c826..9770f30 100644 --- a/sds/sds_utils.nim +++ b/sds/sds_utils.nim @@ -1,5 +1,5 @@ -import std/[times, locks, tables, sequtils, hashes] -import chronicles, results +import std/[times, tables, sequtils, hashes] +import chronos, chronicles, results import ./rolling_bloom_filter import ./types/[ sds_message_id, history_entry, sds_message, unacknowledged_message, incoming_message, @@ -16,44 +16,53 @@ proc defaultConfig*(): ReliabilityConfig = proc dropChannelFromPersistence*( rm: ReliabilityManager, channelId: SdsChannelID -) {.gcsafe, raises: [].} = +): Future[void] {.async: (raises: []), gcsafe.} = ## Wipes all persisted state for a channel via a single backend call. ## Called by removeChannel / resetReliabilityManager before they clear ## in-memory state. Backend executes the wipe in one transaction. - rm.persistence.dropChannel(channelId) + await rm.persistence.dropChannel(channelId) -proc cleanup*(rm: ReliabilityManager) {.raises: [].} = +proc cleanup*( + rm: ReliabilityManager +): Future[void] {.async: (raises: []), gcsafe.} = ## Releases in-memory state. Does NOT wipe persistence — the manager may be ## reconstructed against the same backend after cleanup, so disk state must ## survive. For deliberate disk wipe, use `removeChannel` or ## `resetReliabilityManager`. - if not rm.isNil(): + if rm.isNil(): + return + try: + await rm.lock.acquire() try: - withLock rm.lock: - for channelId, channel in rm.channels: - channel.outgoingBuffer.setLen(0) - channel.incomingBuffer.clear() - channel.messageHistory.clear() - channel.outgoingRepairBuffer.clear() - channel.incomingRepairBuffer.clear() - rm.channels.clear() - except Exception: - error "Error during cleanup", error = getCurrentExceptionMsg() + for channelId, channel in rm.channels: + channel.outgoingBuffer.setLen(0) + channel.incomingBuffer.clear() + channel.messageHistory.clear() + channel.outgoingRepairBuffer.clear() + channel.incomingRepairBuffer.clear() + rm.channels.clear() + finally: + rm.lock.release() + except CatchableError: + error "Error during cleanup", error = getCurrentExceptionMsg() proc cleanBloomFilter*( rm: ReliabilityManager, channelId: SdsChannelID -) {.gcsafe, raises: [].} = - withLock rm.lock: +): Future[void] {.async: (raises: []), gcsafe.} = + try: + await rm.lock.acquire() try: if channelId in rm.channels: rm.channels[channelId].bloomFilter.clean() - except Exception: - error "Failed to clean bloom filter", - error = getCurrentExceptionMsg(), channelId = channelId + finally: + rm.lock.release() + except CatchableError: + error "Failed to clean bloom filter", + error = getCurrentExceptionMsg(), channelId = channelId proc addToHistory*( rm: ReliabilityManager, msg: SdsMessage, channelId: SdsChannelID -) {.gcsafe, raises: [].} = +): Future[void] {.async: (raises: []), gcsafe.} = ## Inserts a delivered message into the channel's history map and evicts the ## eldest entries when the bound is exceeded. The full SdsMessage is kept so ## senderId is available for downstream causal-history population and the @@ -62,27 +71,27 @@ proc addToHistory*( if channelId in rm.channels: let channel = rm.channels[channelId] channel.messageHistory[msg.messageId] = msg - rm.persistence.appendLogEntry(channelId, msg) + await rm.persistence.appendLogEntry(channelId, msg) while channel.messageHistory.len > rm.config.maxMessageHistory: var firstKey: SdsMessageID for k in channel.messageHistory.keys: firstKey = k break channel.messageHistory.del(firstKey) - rm.persistence.removeLogEntry(channelId, firstKey) - except Exception: + await rm.persistence.removeLogEntry(channelId, firstKey) + except CatchableError: error "Failed to add to history", channelId = channelId, msgId = msg.messageId, error = getCurrentExceptionMsg() proc updateLamportTimestamp*( rm: ReliabilityManager, msgTs: int64, channelId: SdsChannelID -) {.gcsafe, raises: [].} = +): Future[void] {.async: (raises: []), gcsafe.} = try: if channelId in rm.channels: let channel = rm.channels[channelId] channel.lamportTimestamp = max(msgTs, channel.lamportTimestamp) + 1 - rm.persistence.saveLamport(channelId, channel.lamportTimestamp) - except Exception: + await rm.persistence.saveLamport(channelId, channel.lamportTimestamp) + except CatchableError: error "Failed to update lamport timestamp", channelId = channelId, msgTs = msgTs, error = getCurrentExceptionMsg() @@ -100,9 +109,9 @@ proc getMessageIds*(causalHistory: seq[HistoryEntry]): seq[SdsMessageID] = proc computeTReq*( participantId: SdsParticipantID, messageId: SdsMessageID, - tMin: Duration, - tMax: Duration, -): Duration = + tMin: times.Duration, + tMax: times.Duration, +): times.Duration = ## Computes the repair request backoff duration per SDS-R spec: ## T_req = hash(participant_id, message_id) % (T_max - T_min) + T_min let h = abs(hash(participantId.string & messageId)) @@ -116,8 +125,8 @@ proc computeTResp*( participantId: SdsParticipantID, senderId: SdsParticipantID, messageId: SdsMessageID, - tMax: Duration, -): Duration = + tMax: times.Duration, +): times.Duration = ## Computes the repair response backoff duration per SDS-R spec: ## distance = hash(participant_id) XOR hash(sender_id) ## T_resp = distance * hash(message_id) % T_max @@ -149,7 +158,7 @@ proc isInResponseGroup*( proc getRecentHistoryEntries*( rm: ReliabilityManager, n: int, channelId: SdsChannelID -): seq[HistoryEntry] = +): Future[seq[HistoryEntry]] {.async: (raises: []), gcsafe.} = ## Get recent history entries for sending in causal history. ## Populates retrieval hints and senderId (SDS-R) for each entry. try: @@ -164,15 +173,16 @@ proc getRecentHistoryEntries*( for msgId in recentMessageIds: var entry = HistoryEntry(messageId: msgId) if not rm.onRetrievalHint.isNil(): - entry.retrievalHint = rm.onRetrievalHint(msgId) + {.cast(raises: []).}: + entry.retrievalHint = rm.onRetrievalHint(msgId) if entry.retrievalHint.len > 0: - rm.persistence.setRetrievalHint(msgId, entry.retrievalHint) + await rm.persistence.setRetrievalHint(msgId, entry.retrievalHint) entry.senderId = channel.messageHistory[msgId].senderId entries.add(entry) return entries else: return @[] - except Exception: + except CatchableError: error "Failed to get recent history entries", channelId = channelId, n = n, error = getCurrentExceptionMsg() return @[] @@ -197,8 +207,9 @@ proc checkDependencies*( proc getMessageHistory*( rm: ReliabilityManager, channelId: SdsChannelID -): seq[SdsMessageID] = - withLock rm.lock: +): Future[seq[SdsMessageID]] {.async: (raises: []), gcsafe.} = + try: + await rm.lock.acquire() try: if channelId in rm.channels: var ids: seq[SdsMessageID] = @[] @@ -207,42 +218,50 @@ proc getMessageHistory*( return ids else: return @[] - except Exception: - error "Failed to get message history", - channelId = channelId, error = getCurrentExceptionMsg() - return @[] + finally: + rm.lock.release() + except CatchableError: + error "Failed to get message history", + channelId = channelId, error = getCurrentExceptionMsg() + return @[] proc getOutgoingBuffer*( rm: ReliabilityManager, channelId: SdsChannelID -): seq[UnacknowledgedMessage] = - withLock rm.lock: +): Future[seq[UnacknowledgedMessage]] {.async: (raises: []), gcsafe.} = + try: + await rm.lock.acquire() try: if channelId in rm.channels: return rm.channels[channelId].outgoingBuffer else: return @[] - except Exception: - error "Failed to get outgoing buffer", - channelId = channelId, error = getCurrentExceptionMsg() - return @[] + finally: + rm.lock.release() + except CatchableError: + error "Failed to get outgoing buffer", + channelId = channelId, error = getCurrentExceptionMsg() + return @[] proc getIncomingBuffer*( rm: ReliabilityManager, channelId: SdsChannelID -): Table[SdsMessageID, IncomingMessage] = - withLock rm.lock: +): Future[Table[SdsMessageID, IncomingMessage]] {.async: (raises: []), gcsafe.} = + try: + await rm.lock.acquire() try: if channelId in rm.channels: return rm.channels[channelId].incomingBuffer else: return initTable[SdsMessageID, IncomingMessage]() - except Exception: - error "Failed to get incoming buffer", - channelId = channelId, error = getCurrentExceptionMsg() - return initTable[SdsMessageID, IncomingMessage]() + finally: + rm.lock.release() + except CatchableError: + error "Failed to get incoming buffer", + channelId = channelId, error = getCurrentExceptionMsg() + return initTable[SdsMessageID, IncomingMessage]() proc getOrCreateChannel*( rm: ReliabilityManager, channelId: SdsChannelID -): ChannelContext = +): Future[ChannelContext] {.async: (raises: [CatchableError]), gcsafe.} = ## Returns the channel context, creating and bootstrapping it from the ## persistence backend if it does not yet exist in memory. The bloom filter ## is rebuilt deterministically from the loaded message history rather than @@ -252,7 +271,7 @@ proc getOrCreateChannel*( let channel = ChannelContext.new( RollingBloomFilter.init(rm.config.bloomFilterCapacity, rm.config.bloomFilterErrorRate) ) - let snapshot = rm.persistence.loadAllForChannel(channelId) + let snapshot = await rm.persistence.loadAllForChannel(channelId) channel.lamportTimestamp = snapshot.lamportTimestamp for msg in snapshot.messageHistory: channel.messageHistory[msg.messageId] = msg @@ -267,39 +286,55 @@ proc getOrCreateChannel*( channel.incomingRepairBuffer[msgId] = entry rm.channels[channelId] = channel return rm.channels[channelId] - except Exception: + except CatchableError as e: error "Failed to get or create channel", channelId = channelId, error = getCurrentExceptionMsg() - raise + raise e proc ensureChannel*( rm: ReliabilityManager, channelId: SdsChannelID -): Result[void, ReliabilityError] = - withLock rm.lock: +): Future[Result[void, ReliabilityError]] {.async: (raises: []), gcsafe.} = + try: + await rm.lock.acquire() try: - discard rm.getOrCreateChannel(channelId) - return ok() - except Exception: - error "Failed to ensure channel", - channelId = channelId, msg = getCurrentExceptionMsg() - return err(ReliabilityError.reInternalError) + try: + discard await rm.getOrCreateChannel(channelId) + return ok() + except CatchableError: + error "Failed to ensure channel", + channelId = channelId, msg = getCurrentExceptionMsg() + return err(ReliabilityError.reInternalError) + finally: + rm.lock.release() + except CatchableError: + error "Failed to ensure channel (lock)", + channelId = channelId, msg = getCurrentExceptionMsg() + return err(ReliabilityError.reInternalError) proc removeChannel*( rm: ReliabilityManager, channelId: SdsChannelID -): Result[void, ReliabilityError] = - withLock rm.lock: +): Future[Result[void, ReliabilityError]] {.async: (raises: []), gcsafe.} = + try: + await rm.lock.acquire() try: - if channelId in rm.channels: - let channel = rm.channels[channelId] - rm.dropChannelFromPersistence(channelId) - channel.outgoingBuffer.setLen(0) - channel.incomingBuffer.clear() - channel.messageHistory.clear() - channel.outgoingRepairBuffer.clear() - channel.incomingRepairBuffer.clear() - rm.channels.del(channelId) - return ok() - except Exception: - error "Failed to remove channel", - channelId = channelId, msg = getCurrentExceptionMsg() - return err(ReliabilityError.reInternalError) + try: + if channelId in rm.channels: + let channel = rm.channels[channelId] + await rm.dropChannelFromPersistence(channelId) + channel.outgoingBuffer.setLen(0) + channel.incomingBuffer.clear() + channel.messageHistory.clear() + channel.outgoingRepairBuffer.clear() + channel.incomingRepairBuffer.clear() + rm.channels.del(channelId) + return ok() + except CatchableError: + error "Failed to remove channel", + channelId = channelId, msg = getCurrentExceptionMsg() + return err(ReliabilityError.reInternalError) + finally: + rm.lock.release() + except CatchableError: + error "Failed to remove channel (lock)", + channelId = channelId, msg = getCurrentExceptionMsg() + return err(ReliabilityError.reInternalError) diff --git a/sds/types/persistence.nim b/sds/types/persistence.nim index 673c7d7..42b51ba 100644 --- a/sds/types/persistence.nim +++ b/sds/types/persistence.nim @@ -1,3 +1,4 @@ +import chronos import ./sds_message_id import ./sds_message import ./unacknowledged_message @@ -14,6 +15,9 @@ export ## protocol calls into one of these procs immediately after the in-memory ## change, so on-disk state stays in lockstep with in-memory state. ## +## All proc fields are async (return `Future`) so backends can do real I/O +## without blocking the Chronos event loop the manager runs on. +## ## Bloom filter is intentionally not persisted: it is rebuilt from the local ## history log on bootstrap. Async timers are likewise recomputed from the ## absolute timestamps stored in the repair buffer entries. @@ -36,91 +40,136 @@ type ## Pluggable persistence contract. The caller supplies an instance of this ## type at `newReliabilityManager` construction time. Each proc field is ## invoked by nim-sds at the corresponding state-mutation point. + ## All fields are async; nim-sds awaits each call to keep on-disk and + ## in-memory state in lockstep without blocking the event loop. # Per-channel lamport clock - saveLamport*: - proc(channelId: SdsChannelID, lamport: int64) {.gcsafe, raises: [].} + saveLamport*: proc(channelId: SdsChannelID, lamport: int64): Future[void] {. + async: (raises: []), gcsafe + .} # Local log (delivered messages) - appendLogEntry*: - proc(channelId: SdsChannelID, msg: SdsMessage) {.gcsafe, raises: [].} + appendLogEntry*: proc(channelId: SdsChannelID, msg: SdsMessage): Future[void] {. + async: (raises: []), gcsafe + .} removeLogEntry*: - proc(channelId: SdsChannelID, msgId: SdsMessageID) {.gcsafe, raises: [].} - setRetrievalHint*: - proc(msgId: SdsMessageID, hint: seq[byte]) {.gcsafe, raises: [].} + proc(channelId: SdsChannelID, msgId: SdsMessageID): Future[void] {. + async: (raises: []), gcsafe + .} + setRetrievalHint*: proc(msgId: SdsMessageID, hint: seq[byte]): Future[void] {. + async: (raises: []), gcsafe + .} # Outgoing unacknowledged buffer saveOutgoing*: - proc(channelId: SdsChannelID, msg: UnacknowledgedMessage) {.gcsafe, raises: [].} + proc(channelId: SdsChannelID, msg: UnacknowledgedMessage): Future[void] {. + async: (raises: []), gcsafe + .} removeOutgoing*: - proc(channelId: SdsChannelID, msgId: SdsMessageID) {.gcsafe, raises: [].} + proc(channelId: SdsChannelID, msgId: SdsMessageID): Future[void] {. + async: (raises: []), gcsafe + .} # Incoming dependency-waiting buffer saveIncoming*: - proc(channelId: SdsChannelID, msg: IncomingMessage) {.gcsafe, raises: [].} + proc(channelId: SdsChannelID, msg: IncomingMessage): Future[void] {. + async: (raises: []), gcsafe + .} removeIncoming*: - proc(channelId: SdsChannelID, msgId: SdsMessageID) {.gcsafe, raises: [].} + proc(channelId: SdsChannelID, msgId: SdsMessageID): Future[void] {. + async: (raises: []), gcsafe + .} # SDS-R outgoing repair buffer saveOutgoingRepair*: proc( channelId: SdsChannelID, msgId: SdsMessageID, entry: OutgoingRepairEntry - ) {.gcsafe, raises: [].} + ): Future[void] {.async: (raises: []), gcsafe.} removeOutgoingRepair*: - proc(channelId: SdsChannelID, msgId: SdsMessageID) {.gcsafe, raises: [].} + proc(channelId: SdsChannelID, msgId: SdsMessageID): Future[void] {. + async: (raises: []), gcsafe + .} # SDS-R incoming repair buffer saveIncomingRepair*: proc( channelId: SdsChannelID, msgId: SdsMessageID, entry: IncomingRepairEntry - ) {.gcsafe, raises: [].} + ): Future[void] {.async: (raises: []), gcsafe.} removeIncomingRepair*: - proc(channelId: SdsChannelID, msgId: SdsMessageID) {.gcsafe, raises: [].} + proc(channelId: SdsChannelID, msgId: SdsMessageID): Future[void] {. + async: (raises: []), gcsafe + .} # Wipe all persisted state for a channel in one transactional call. # Called by removeChannel / resetReliabilityManager. Backends should # implement this atomically (e.g. one BEGIN/COMMIT) — a per-row loop on # the nim-sds side would mean N fsyncs per drop. - dropChannel*: - proc(channelId: SdsChannelID) {.gcsafe, raises: [].} + dropChannel*: proc(channelId: SdsChannelID): Future[void] {. + async: (raises: []), gcsafe + .} # Bootstrap on `addChannel` / `getOrCreateChannel`. - loadAllForChannel*: - proc(channelId: SdsChannelID): ChannelSnapshot {.gcsafe, raises: [].} + loadAllForChannel*: proc(channelId: SdsChannelID): Future[ChannelSnapshot] {. + async: (raises: []), gcsafe + .} proc noOpPersistence*(): Persistence = ## Default backend that discards every write and returns an empty snapshot. ## Used so existing callers (and tests) that don't care about durability ## keep working without supplying a real backend. Persistence( - saveLamport: proc(channelId: SdsChannelID, lamport: int64) = + saveLamport: proc( + channelId: SdsChannelID, lamport: int64 + ): Future[void] {.async: (raises: []), gcsafe.} = discard, - appendLogEntry: proc(channelId: SdsChannelID, msg: SdsMessage) = + appendLogEntry: proc( + channelId: SdsChannelID, msg: SdsMessage + ): Future[void] {.async: (raises: []), gcsafe.} = discard, - removeLogEntry: proc(channelId: SdsChannelID, msgId: SdsMessageID) = + removeLogEntry: proc( + channelId: SdsChannelID, msgId: SdsMessageID + ): Future[void] {.async: (raises: []), gcsafe.} = discard, - setRetrievalHint: proc(msgId: SdsMessageID, hint: seq[byte]) = + setRetrievalHint: proc( + msgId: SdsMessageID, hint: seq[byte] + ): Future[void] {.async: (raises: []), gcsafe.} = discard, - saveOutgoing: proc(channelId: SdsChannelID, msg: UnacknowledgedMessage) = + saveOutgoing: proc( + channelId: SdsChannelID, msg: UnacknowledgedMessage + ): Future[void] {.async: (raises: []), gcsafe.} = discard, - removeOutgoing: proc(channelId: SdsChannelID, msgId: SdsMessageID) = + removeOutgoing: proc( + channelId: SdsChannelID, msgId: SdsMessageID + ): Future[void] {.async: (raises: []), gcsafe.} = discard, - saveIncoming: proc(channelId: SdsChannelID, msg: IncomingMessage) = + saveIncoming: proc( + channelId: SdsChannelID, msg: IncomingMessage + ): Future[void] {.async: (raises: []), gcsafe.} = discard, - removeIncoming: proc(channelId: SdsChannelID, msgId: SdsMessageID) = + removeIncoming: proc( + channelId: SdsChannelID, msgId: SdsMessageID + ): Future[void] {.async: (raises: []), gcsafe.} = discard, saveOutgoingRepair: proc( channelId: SdsChannelID, msgId: SdsMessageID, entry: OutgoingRepairEntry - ) = + ): Future[void] {.async: (raises: []), gcsafe.} = discard, - removeOutgoingRepair: proc(channelId: SdsChannelID, msgId: SdsMessageID) = + removeOutgoingRepair: proc( + channelId: SdsChannelID, msgId: SdsMessageID + ): Future[void] {.async: (raises: []), gcsafe.} = discard, saveIncomingRepair: proc( channelId: SdsChannelID, msgId: SdsMessageID, entry: IncomingRepairEntry - ) = + ): Future[void] {.async: (raises: []), gcsafe.} = discard, - removeIncomingRepair: proc(channelId: SdsChannelID, msgId: SdsMessageID) = + removeIncomingRepair: proc( + channelId: SdsChannelID, msgId: SdsMessageID + ): Future[void] {.async: (raises: []), gcsafe.} = discard, - dropChannel: proc(channelId: SdsChannelID) = + dropChannel: proc( + channelId: SdsChannelID + ): Future[void] {.async: (raises: []), gcsafe.} = discard, - loadAllForChannel: proc(channelId: SdsChannelID): ChannelSnapshot = - ChannelSnapshot(), + loadAllForChannel: proc( + channelId: SdsChannelID + ): Future[ChannelSnapshot] {.async: (raises: []), gcsafe.} = + return ChannelSnapshot(), ) diff --git a/sds/types/reliability_manager.nim b/sds/types/reliability_manager.nim index 3c42850..d487c48 100644 --- a/sds/types/reliability_manager.nim +++ b/sds/types/reliability_manager.nim @@ -1,4 +1,5 @@ -import std/[tables, locks] +import std/tables +import chronos import ./sds_message_id import ./history_entry import ./callbacks @@ -15,7 +16,11 @@ type ReliabilityManager* = ref object participantId*: SdsParticipantID persistence*: Persistence ## Pluggable durability backend; defaults to a no-op when not supplied. - lock*: Lock + lock*: AsyncLock + ## Single-threaded Chronos cooperative lock. Serializes mutators against + ## one another at await points; the manager assumes all calls come from + ## the same Chronos event loop (the FFI worker thread). Multi-OS-thread + ## use is the caller's responsibility. onMessageReady*: proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} onMessageSent*: proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} onMissingDependencies*: proc( @@ -42,6 +47,6 @@ proc new*( config: config, participantId: participantId, persistence: persistence, + lock: newAsyncLock(), ) - rm.lock.initLock() return rm diff --git a/tests/in_memory_persistence.nim b/tests/in_memory_persistence.nim index f3c7d88..d88d4ea 100644 --- a/tests/in_memory_persistence.nim +++ b/tests/in_memory_persistence.nim @@ -1,4 +1,5 @@ import std/tables +import chronos import sds ## Test-only Persistence backend backed by Nim tables. Lets tests verify the @@ -23,76 +24,98 @@ proc newInMemoryStore*(): InMemoryStore = proc newInMemoryPersistence*(store: InMemoryStore): Persistence = Persistence( - saveLamport: proc(channelId: SdsChannelID, lamport: int64) {.gcsafe, raises: [].} = + saveLamport: proc( + channelId: SdsChannelID, lamport: int64 + ): Future[void] {.async: (raises: []), gcsafe.} = store.lamports[channelId] = lamport, - appendLogEntry: proc(channelId: SdsChannelID, msg: SdsMessage) {.gcsafe, raises: [].} = + appendLogEntry: proc( + channelId: SdsChannelID, msg: SdsMessage + ): Future[void] {.async: (raises: []), gcsafe.} = {.cast(raises: []).}: if channelId notin store.log: store.log[channelId] = initOrderedTable[SdsMessageID, SdsMessage]() store.log[channelId][msg.messageId] = msg, - removeLogEntry: proc(channelId: SdsChannelID, msgId: SdsMessageID) {.gcsafe, raises: [].} = + removeLogEntry: proc( + channelId: SdsChannelID, msgId: SdsMessageID + ): Future[void] {.async: (raises: []), gcsafe.} = {.cast(raises: []).}: if channelId in store.log: store.log[channelId].del(msgId), - setRetrievalHint: proc(msgId: SdsMessageID, hint: seq[byte]) {.gcsafe, raises: [].} = + setRetrievalHint: proc( + msgId: SdsMessageID, hint: seq[byte] + ): Future[void] {.async: (raises: []), gcsafe.} = store.hints[msgId] = hint, - saveOutgoing: proc(channelId: SdsChannelID, msg: UnacknowledgedMessage) {.gcsafe, raises: [].} = + saveOutgoing: proc( + channelId: SdsChannelID, msg: UnacknowledgedMessage + ): Future[void] {.async: (raises: []), gcsafe.} = {.cast(raises: []).}: if channelId notin store.outgoing: store.outgoing[channelId] = initOrderedTable[SdsMessageID, UnacknowledgedMessage]() store.outgoing[channelId][msg.message.messageId] = msg, - removeOutgoing: proc(channelId: SdsChannelID, msgId: SdsMessageID) {.gcsafe, raises: [].} = + removeOutgoing: proc( + channelId: SdsChannelID, msgId: SdsMessageID + ): Future[void] {.async: (raises: []), gcsafe.} = {.cast(raises: []).}: if channelId in store.outgoing: store.outgoing[channelId].del(msgId), - saveIncoming: proc(channelId: SdsChannelID, msg: IncomingMessage) {.gcsafe, raises: [].} = + saveIncoming: proc( + channelId: SdsChannelID, msg: IncomingMessage + ): Future[void] {.async: (raises: []), gcsafe.} = {.cast(raises: []).}: if channelId notin store.incoming: store.incoming[channelId] = initOrderedTable[SdsMessageID, IncomingMessage]() store.incoming[channelId][msg.message.messageId] = msg, - removeIncoming: proc(channelId: SdsChannelID, msgId: SdsMessageID) {.gcsafe, raises: [].} = + removeIncoming: proc( + channelId: SdsChannelID, msgId: SdsMessageID + ): Future[void] {.async: (raises: []), gcsafe.} = {.cast(raises: []).}: if channelId in store.incoming: store.incoming[channelId].del(msgId), saveOutgoingRepair: proc( channelId: SdsChannelID, msgId: SdsMessageID, entry: OutgoingRepairEntry - ) {.gcsafe, raises: [].} = + ): Future[void] {.async: (raises: []), gcsafe.} = {.cast(raises: []).}: if channelId notin store.outgoingRepair: store.outgoingRepair[channelId] = initOrderedTable[SdsMessageID, OutgoingRepairEntry]() store.outgoingRepair[channelId][msgId] = entry, - removeOutgoingRepair: proc(channelId: SdsChannelID, msgId: SdsMessageID) {.gcsafe, raises: [].} = + removeOutgoingRepair: proc( + channelId: SdsChannelID, msgId: SdsMessageID + ): Future[void] {.async: (raises: []), gcsafe.} = {.cast(raises: []).}: if channelId in store.outgoingRepair: store.outgoingRepair[channelId].del(msgId), saveIncomingRepair: proc( channelId: SdsChannelID, msgId: SdsMessageID, entry: IncomingRepairEntry - ) {.gcsafe, raises: [].} = + ): Future[void] {.async: (raises: []), gcsafe.} = {.cast(raises: []).}: if channelId notin store.incomingRepair: store.incomingRepair[channelId] = initOrderedTable[SdsMessageID, IncomingRepairEntry]() store.incomingRepair[channelId][msgId] = entry, - removeIncomingRepair: proc(channelId: SdsChannelID, msgId: SdsMessageID) {.gcsafe, raises: [].} = + removeIncomingRepair: proc( + channelId: SdsChannelID, msgId: SdsMessageID + ): Future[void] {.async: (raises: []), gcsafe.} = {.cast(raises: []).}: if channelId in store.incomingRepair: store.incomingRepair[channelId].del(msgId), - dropChannel: proc(channelId: SdsChannelID) {.gcsafe, raises: [].} = + dropChannel: proc( + channelId: SdsChannelID + ): Future[void] {.async: (raises: []), gcsafe.} = {.cast(raises: []).}: store.lamports.del(channelId) store.log.del(channelId) @@ -103,7 +126,9 @@ proc newInMemoryPersistence*(store: InMemoryStore): Persistence = store.dropChannelCalls[channelId] = store.dropChannelCalls.getOrDefault(channelId) + 1, - loadAllForChannel: proc(channelId: SdsChannelID): ChannelSnapshot {.gcsafe, raises: [].} = + loadAllForChannel: proc( + channelId: SdsChannelID + ): Future[ChannelSnapshot] {.async: (raises: []), gcsafe.} = {.cast(raises: []).}: var snap = ChannelSnapshot() if channelId in store.lamports: @@ -123,5 +148,5 @@ proc newInMemoryPersistence*(store: InMemoryStore): Persistence = if channelId in store.incomingRepair: for msgId, entry in store.incomingRepair[channelId]: snap.incomingRepairBuffer.add((msgId, entry)) - snap, + return snap, ) diff --git a/tests/test_persistence.nim b/tests/test_persistence.nim index 4d3fc3c..981780d 100644 --- a/tests/test_persistence.nim +++ b/tests/test_persistence.nim @@ -1,4 +1,4 @@ -import unittest, results, std/[tables, sets, times] +import unittest, results, chronos, std/[tables, sets, times] import sds import ./in_memory_persistence @@ -6,46 +6,63 @@ converter toParticipantID(s: string): SdsParticipantID = s.SdsParticipantID const testChannel = "testChannel" +template asyncTest(name: string, body: untyped) = + ## Wraps a unittest `test` body in an async proc and runs it to completion. + ## Tests can now `await` rm.* and rm.persistence.* calls directly. + ## unittest's `check` raises Exception (wider than chronos's CatchableError), + ## so catch inside the async body and re-raise after waitFor. + test name: + var asyncTestErr {.inject.}: ref Exception = nil + proc inner() {.async.} = + {.cast(gcsafe).}: + try: + body + except Exception as e: + asyncTestErr = e + waitFor inner() + if asyncTestErr != nil: + raise asyncTestErr + suite "Persistence: write → restart → read-back": - test "outgoing buffer survives restart": + asyncTest "outgoing buffer survives restart": let store = newInMemoryStore() let p1 = newInMemoryPersistence(store) - let rm1 = newReliabilityManager(participantId = "alice", persistence =p1).get() - check rm1.ensureChannel(testChannel).isOk() - let wrapped = rm1.wrapOutgoingMessage(@[1.byte, 2, 3], "msg-1", testChannel) + let rm1 = newReliabilityManager(participantId = "alice", persistence = p1).get() + check (await rm1.ensureChannel(testChannel)).isOk() + let wrapped = await rm1.wrapOutgoingMessage(@[1.byte, 2, 3], "msg-1", testChannel) check wrapped.isOk() check store.outgoing[testChannel].len == 1 check "msg-1" in store.outgoing[testChannel] - rm1.cleanup() + await rm1.cleanup() # Simulate restart: fresh manager, same backend let p2 = newInMemoryPersistence(store) - let rm2 = newReliabilityManager(participantId = "alice", persistence =p2).get() - check rm2.ensureChannel(testChannel).isOk() - let buf = rm2.getOutgoingBuffer(testChannel) + let rm2 = newReliabilityManager(participantId = "alice", persistence = p2).get() + check (await rm2.ensureChannel(testChannel)).isOk() + let buf = await rm2.getOutgoingBuffer(testChannel) check buf.len == 1 check buf[0].message.messageId == "msg-1" - rm2.cleanup() + await rm2.cleanup() - test "lamport clock survives restart": + asyncTest "lamport clock survives restart": let store = newInMemoryStore() let p1 = newInMemoryPersistence(store) - let rm1 = newReliabilityManager(participantId = "alice", persistence =p1).get() - check rm1.ensureChannel(testChannel).isOk() - rm1.updateLamportTimestamp(42, testChannel) + let rm1 = newReliabilityManager(participantId = "alice", persistence = p1).get() + check (await rm1.ensureChannel(testChannel)).isOk() + await rm1.updateLamportTimestamp(42, testChannel) check store.lamports[testChannel] == 43 # max(42, 0) + 1 - rm1.cleanup() + await rm1.cleanup() let p2 = newInMemoryPersistence(store) - let rm2 = newReliabilityManager(participantId = "alice", persistence =p2).get() - check rm2.ensureChannel(testChannel).isOk() + let rm2 = newReliabilityManager(participantId = "alice", persistence = p2).get() + check (await rm2.ensureChannel(testChannel)).isOk() check rm2.channels[testChannel].lamportTimestamp == 43 - test "delivered messages survive restart and rebuild bloom": + asyncTest "delivered messages survive restart and rebuild bloom": let store = newInMemoryStore() let p1 = newInMemoryPersistence(store) - let rm1 = newReliabilityManager(participantId = "alice", persistence =p1).get() - check rm1.ensureChannel(testChannel).isOk() + let rm1 = newReliabilityManager(participantId = "alice", persistence = p1).get() + check (await rm1.ensureChannel(testChannel)).isOk() let msg = SdsMessage.init( messageId = "delivered-1", lamportTimestamp = 1, @@ -55,25 +72,25 @@ suite "Persistence: write → restart → read-back": bloomFilter = @[], senderId = "alice", ) - rm1.addToHistory(msg, testChannel) + await rm1.addToHistory(msg, testChannel) check store.log[testChannel].len == 1 - rm1.cleanup() + await rm1.cleanup() let p2 = newInMemoryPersistence(store) - let rm2 = newReliabilityManager(participantId = "alice", persistence =p2).get() - check rm2.ensureChannel(testChannel).isOk() + let rm2 = newReliabilityManager(participantId = "alice", persistence = p2).get() + check (await rm2.ensureChannel(testChannel)).isOk() let ch = rm2.channels[testChannel] check ch.messageHistory.len == 1 check "delivered-1" in ch.messageHistory # Bloom filter rebuilt from log on bootstrap check ch.bloomFilter.contains("delivered-1") - test "ack removes outgoing entry from persistence": + asyncTest "ack removes outgoing entry from persistence": let store = newInMemoryStore() let p = newInMemoryPersistence(store) - let rm = newReliabilityManager(participantId = "alice", persistence =p).get() - check rm.ensureChannel(testChannel).isOk() - discard rm.wrapOutgoingMessage(@[1.byte], "msg-x", testChannel) + let rm = newReliabilityManager(participantId = "alice", persistence = p).get() + check (await rm.ensureChannel(testChannel)).isOk() + discard await rm.wrapOutgoingMessage(@[1.byte], "msg-x", testChannel) check "msg-x" in store.outgoing[testChannel] # Synthesize an incoming message that ACKs msg-x via causal history @@ -87,22 +104,22 @@ suite "Persistence: write → restart → read-back": senderId = "bob", ) let serialized = serializeMessage(ackMsg).get() - discard rm.unwrapReceivedMessage(serialized) + discard await rm.unwrapReceivedMessage(serialized) check "msg-x" notin store.outgoing[testChannel] - rm.cleanup() + await rm.cleanup() - test "removeChannel issues exactly one dropChannel call and wipes all state": + asyncTest "removeChannel issues exactly one dropChannel call and wipes all state": # Regression for PR #66 review: removal must be a single transactional # drop, not N per-row removes — otherwise SQLite eats N fsyncs per drop. let store = newInMemoryStore() let p = newInMemoryPersistence(store) - let rm = newReliabilityManager(participantId = "alice", persistence =p).get() - check rm.ensureChannel(testChannel).isOk() - discard rm.wrapOutgoingMessage(@[1.byte], "msg-r", testChannel) + let rm = newReliabilityManager(participantId = "alice", persistence = p).get() + check (await rm.ensureChannel(testChannel)).isOk() + discard await rm.wrapOutgoingMessage(@[1.byte], "msg-r", testChannel) check store.outgoing[testChannel].len == 1 check store.lamports[testChannel] > 0 - check rm.removeChannel(testChannel).isOk() + check (await rm.removeChannel(testChannel)).isOk() check store.dropChannelCalls.getOrDefault(testChannel) == 1 check testChannel notin store.outgoing check testChannel notin store.lamports @@ -110,50 +127,52 @@ suite "Persistence: write → restart → read-back": check testChannel notin store.incoming check testChannel notin store.outgoingRepair check testChannel notin store.incomingRepair - rm.cleanup() + await rm.cleanup() - test "noOpPersistence keeps existing manager working": + asyncTest "noOpPersistence keeps existing manager working": let rm = newReliabilityManager(participantId = "alice").get() # default no-op persistence - check rm.ensureChannel(testChannel).isOk() - let wrapped = rm.wrapOutgoingMessage(@[1.byte], "msg-n", testChannel) + check (await rm.ensureChannel(testChannel)).isOk() + let wrapped = await rm.wrapOutgoingMessage(@[1.byte], "msg-n", testChannel) check wrapped.isOk() - check rm.getOutgoingBuffer(testChannel).len == 1 - rm.cleanup() + let buf = await rm.getOutgoingBuffer(testChannel) + check buf.len == 1 + await rm.cleanup() - test "continue operating after restart: lamport stays monotonic": + asyncTest "continue operating after restart: lamport stays monotonic": let store = newInMemoryStore() let p1 = newInMemoryPersistence(store) - let rm1 = newReliabilityManager(participantId = "alice", persistence =p1).get() - check rm1.ensureChannel(testChannel).isOk() - discard rm1.wrapOutgoingMessage(@[1.byte], "m1", testChannel) + let rm1 = newReliabilityManager(participantId = "alice", persistence = p1).get() + check (await rm1.ensureChannel(testChannel)).isOk() + discard await rm1.wrapOutgoingMessage(@[1.byte], "m1", testChannel) let lamportAfterSession1 = store.lamports[testChannel] check lamportAfterSession1 > 0 - rm1.cleanup() + await rm1.cleanup() # Restart and send another message — lamport must not regress. let p2 = newInMemoryPersistence(store) - let rm2 = newReliabilityManager(participantId = "alice", persistence =p2).get() - check rm2.ensureChannel(testChannel).isOk() + let rm2 = newReliabilityManager(participantId = "alice", persistence = p2).get() + check (await rm2.ensureChannel(testChannel)).isOk() check rm2.channels[testChannel].lamportTimestamp == lamportAfterSession1 - discard rm2.wrapOutgoingMessage(@[2.byte], "m2", testChannel) + discard await rm2.wrapOutgoingMessage(@[2.byte], "m2", testChannel) check store.lamports[testChannel] > lamportAfterSession1 - check rm2.getOutgoingBuffer(testChannel).len == 2 - rm2.cleanup() + let buf = await rm2.getOutgoingBuffer(testChannel) + check buf.len == 2 + await rm2.cleanup() - test "multiple restart cycles preserve state": + asyncTest "multiple restart cycles preserve state": let store = newInMemoryStore() for i in 1 .. 3: let p = newInMemoryPersistence(store) - let rm = newReliabilityManager(participantId = "alice", persistence =p).get() - check rm.ensureChannel(testChannel).isOk() - discard rm.wrapOutgoingMessage(@[byte(i)], "m" & $i, testChannel) - rm.cleanup() + let rm = newReliabilityManager(participantId = "alice", persistence = p).get() + check (await rm.ensureChannel(testChannel)).isOk() + discard await rm.wrapOutgoingMessage(@[byte(i)], "m" & $i, testChannel) + await rm.cleanup() # Final session: all three messages must be in the buffer. let pFinal = newInMemoryPersistence(store) - let rmFinal = newReliabilityManager(participantId = "alice", persistence =pFinal).get() - check rmFinal.ensureChannel(testChannel).isOk() - let buf = rmFinal.getOutgoingBuffer(testChannel) + let rmFinal = newReliabilityManager(participantId = "alice", persistence = pFinal).get() + check (await rmFinal.ensureChannel(testChannel)).isOk() + let buf = await rmFinal.getOutgoingBuffer(testChannel) check buf.len == 3 var ids = newSeq[string]() for unack in buf: @@ -161,13 +180,13 @@ suite "Persistence: write → restart → read-back": check "m1" in ids check "m2" in ids check "m3" in ids - rmFinal.cleanup() + await rmFinal.cleanup() - test "incoming dep-waiting buffer survives restart with missingDeps intact": + asyncTest "incoming dep-waiting buffer survives restart with missingDeps intact": let store = newInMemoryStore() let p1 = newInMemoryPersistence(store) - let rm1 = newReliabilityManager(participantId = "alice", persistence =p1).get() - check rm1.ensureChannel(testChannel).isOk() + let rm1 = newReliabilityManager(participantId = "alice", persistence = p1).get() + check (await rm1.ensureChannel(testChannel)).isOk() # Receive a message whose causal-history references an unknown predecessor. let depMsg = SdsMessage.init( @@ -180,47 +199,48 @@ suite "Persistence: write → restart → read-back": senderId = "carol", ) let serialized = serializeMessage(depMsg).get() - discard rm1.unwrapReceivedMessage(serialized) + discard await rm1.unwrapReceivedMessage(serialized) check "msg-with-deps" in store.incoming[testChannel] - rm1.cleanup() + await rm1.cleanup() # Restart — buffered message and its missing-deps set must be back. let p2 = newInMemoryPersistence(store) - let rm2 = newReliabilityManager(participantId = "alice", persistence =p2).get() - check rm2.ensureChannel(testChannel).isOk() - let inbuf = rm2.getIncomingBuffer(testChannel) + let rm2 = newReliabilityManager(participantId = "alice", persistence = p2).get() + check (await rm2.ensureChannel(testChannel)).isOk() + let inbuf = await rm2.getIncomingBuffer(testChannel) check "msg-with-deps" in inbuf check "missing-dep" in inbuf["msg-with-deps"].missingDeps - rm2.cleanup() + await rm2.cleanup() - test "removeChannel + recreate does not inherit stale lamport": + asyncTest "removeChannel + recreate does not inherit stale lamport": # Regression: dropChannel must wipe the lamport row; otherwise a recreate # of the same channelId after restart picks up the old timestamp. let store = newInMemoryStore() let p1 = newInMemoryPersistence(store) - let rm1 = newReliabilityManager(participantId = "alice", persistence =p1).get() - check rm1.ensureChannel(testChannel).isOk() - discard rm1.wrapOutgoingMessage(@[1.byte], "m-old", testChannel) + let rm1 = newReliabilityManager(participantId = "alice", persistence = p1).get() + check (await rm1.ensureChannel(testChannel)).isOk() + discard await rm1.wrapOutgoingMessage(@[1.byte], "m-old", testChannel) check store.lamports[testChannel] > 0 - check rm1.removeChannel(testChannel).isOk() + check (await rm1.removeChannel(testChannel)).isOk() check testChannel notin store.lamports - rm1.cleanup() + await rm1.cleanup() # Recreate the same channelId after a restart — must start fresh. let p2 = newInMemoryPersistence(store) - let rm2 = newReliabilityManager(participantId = "alice", persistence =p2).get() - check rm2.ensureChannel(testChannel).isOk() + let rm2 = newReliabilityManager(participantId = "alice", persistence = p2).get() + check (await rm2.ensureChannel(testChannel)).isOk() check rm2.channels[testChannel].lamportTimestamp == 0 - check rm2.getOutgoingBuffer(testChannel).len == 0 - rm2.cleanup() + let buf = await rm2.getOutgoingBuffer(testChannel) + check buf.len == 0 + await rm2.cleanup() - test "SDS-R outgoing repair buffer survives restart with absolute t_req_at": + asyncTest "SDS-R outgoing repair buffer survives restart with absolute t_req_at": let store = newInMemoryStore() let p1 = newInMemoryPersistence(store) let rm1 = newReliabilityManager( participantId = "alice", persistence = p1 ).get() - check rm1.ensureChannel(testChannel).isOk() + check (await rm1.ensureChannel(testChannel)).isOk() # Receive a message that references an unknown dep — triggers SDS-R repair. let depMsg = SdsMessage.init( @@ -232,32 +252,34 @@ suite "Persistence: write → restart → read-back": bloomFilter = @[], senderId = "bob", ) - discard rm1.unwrapReceivedMessage(serializeMessage(depMsg).get()) + discard await rm1.unwrapReceivedMessage(serializeMessage(depMsg).get()) check "missing-dep" in store.outgoingRepair[testChannel] let originalTReqAt = store.outgoingRepair[testChannel]["missing-dep"].minTimeRepairReq check originalTReqAt.toUnix > 0 - rm1.cleanup() + await rm1.cleanup() # Restart — repair entry must be back with the SAME absolute time, not "now". let p2 = newInMemoryPersistence(store) let rm2 = newReliabilityManager( participantId = "alice", persistence = p2 ).get() - check rm2.ensureChannel(testChannel).isOk() + check (await rm2.ensureChannel(testChannel)).isOk() let buf = rm2.channels[testChannel].outgoingRepairBuffer check "missing-dep" in buf check buf["missing-dep"].minTimeRepairReq == originalTReqAt - rm2.cleanup() + await rm2.cleanup() - test "FIFO eviction state survives restart": + asyncTest "FIFO eviction state survives restart": let store = newInMemoryStore() var smallCfg = defaultConfig() smallCfg.maxMessageHistory = 3 smallCfg.bloomFilterCapacity = 3 let p1 = newInMemoryPersistence(store) - let rm1 = newReliabilityManager(participantId = "alice", config = smallCfg, persistence =p1).get() - check rm1.ensureChannel(testChannel).isOk() + let rm1 = newReliabilityManager( + participantId = "alice", config = smallCfg, persistence = p1 + ).get() + check (await rm1.ensureChannel(testChannel)).isOk() # Add 5 delivered messages — first 2 should be evicted by FIFO. for i in 1 .. 5: let m = SdsMessage.init( @@ -269,16 +291,18 @@ suite "Persistence: write → restart → read-back": bloomFilter = @[], senderId = "alice", ) - rm1.addToHistory(m, testChannel) + await rm1.addToHistory(m, testChannel) check store.log[testChannel].len == 3 check "m1" notin store.log[testChannel] check "m2" notin store.log[testChannel] - rm1.cleanup() + await rm1.cleanup() # Restart — evicted entries must NOT come back; survivors keep order. let p2 = newInMemoryPersistence(store) - let rm2 = newReliabilityManager(participantId = "alice", config = smallCfg, persistence =p2).get() - check rm2.ensureChannel(testChannel).isOk() + let rm2 = newReliabilityManager( + participantId = "alice", config = smallCfg, persistence = p2 + ).get() + check (await rm2.ensureChannel(testChannel)).isOk() let history = rm2.channels[testChannel].messageHistory check history.len == 3 check "m1" notin history @@ -291,16 +315,16 @@ suite "Persistence: write → restart → read-back": channelId = testChannel, content = @[6.byte], bloomFilter = @[], senderId = "alice", ) - rm2.addToHistory(m6, testChannel) + await rm2.addToHistory(m6, testChannel) check "m3" notin store.log[testChannel] check "m6" in store.log[testChannel] - rm2.cleanup() + await rm2.cleanup() - test "dep-clear cascade resumes correctly across a restart": + asyncTest "dep-clear cascade resumes correctly across a restart": let store = newInMemoryStore() let p1 = newInMemoryPersistence(store) - let rm1 = newReliabilityManager(participantId = "alice", persistence =p1).get() - check rm1.ensureChannel(testChannel).isOk() + let rm1 = newReliabilityManager(participantId = "alice", persistence = p1).get() + check (await rm1.ensureChannel(testChannel)).isOk() # Receive c (deps on b), then b (deps on a). Both must buffer. let msgC = SdsMessage.init( @@ -315,17 +339,17 @@ suite "Persistence: write → restart → read-back": channelId = testChannel, content = @[2.byte], bloomFilter = @[], senderId = "bob", ) - discard rm1.unwrapReceivedMessage(serializeMessage(msgC).get()) - discard rm1.unwrapReceivedMessage(serializeMessage(msgB).get()) + discard await rm1.unwrapReceivedMessage(serializeMessage(msgC).get()) + discard await rm1.unwrapReceivedMessage(serializeMessage(msgB).get()) check "c" in store.incoming[testChannel] check "b" in store.incoming[testChannel] - rm1.cleanup() + await rm1.cleanup() # Restart — both still buffered, with intact missingDeps. let p2 = newInMemoryPersistence(store) - let rm2 = newReliabilityManager(participantId = "alice", persistence =p2).get() - check rm2.ensureChannel(testChannel).isOk() - let inbuf = rm2.getIncomingBuffer(testChannel) + let rm2 = newReliabilityManager(participantId = "alice", persistence = p2).get() + check (await rm2.ensureChannel(testChannel)).isOk() + let inbuf = await rm2.getIncomingBuffer(testChannel) check "c" in inbuf check "b" in inbuf @@ -335,11 +359,12 @@ suite "Persistence: write → restart → read-back": channelId = testChannel, content = @[1.byte], bloomFilter = @[], senderId = "alice", ) - discard rm2.unwrapReceivedMessage(serializeMessage(msgA).get()) + discard await rm2.unwrapReceivedMessage(serializeMessage(msgA).get()) let history = rm2.channels[testChannel].messageHistory check "a" in history check "b" in history check "c" in history # Buffer should be drained. - check rm2.getIncomingBuffer(testChannel).len == 0 - rm2.cleanup() + let inbufFinal = await rm2.getIncomingBuffer(testChannel) + check inbufFinal.len == 0 + await rm2.cleanup() diff --git a/tests/test_reliability.nim b/tests/test_reliability.nim index e191de7..a5dcdf1 100644 --- a/tests/test_reliability.nim +++ b/tests/test_reliability.nim @@ -7,6 +7,28 @@ converter toParticipantID(s: string): SdsParticipantID = s.SdsParticipantID const testChannel = "testChannel" +template asyncTest(name: string, body: untyped) = + ## Wraps a unittest `test` body in an async proc so tests can `await` the + ## now-async ReliabilityManager API directly. Setup/teardown blocks still + ## run in the outer (sync) scope — use `waitFor` for any async calls there. + ## unittest's `check` raises `Exception`, which is wider than chronos's + ## default `CatchableError` for async procs — so we catch it inside and + ## re-raise after waitFor, where unittest's normal handling can see it. + ## cast(gcsafe) is needed because suite-level `var rm` looks like a global + ## to the closure capture, but the FFI runtime is single-threaded so the + ## "not gcsafe" warning isn't a real hazard here. + test name: + var asyncTestErr {.inject.}: ref Exception = nil + proc inner() {.async.} = + {.cast(gcsafe).}: + try: + body + except Exception as e: + asyncTestErr = e + waitFor inner() + if asyncTestErr != nil: + raise asyncTestErr + proc seedBloom( rm: ReliabilityManager, channel: SdsChannelID, n: int, prefix = "noise" ) = @@ -25,29 +47,29 @@ suite "Core Operations": let rmResult = newReliabilityManager(participantId = "alice") check rmResult.isOk() rm = rmResult.get() - check rm.ensureChannel(testChannel).isOk() + check (waitFor rm.ensureChannel(testChannel)).isOk() teardown: if not rm.isNil: - rm.cleanup() + waitFor rm.cleanup() - test "can create with default config": + asyncTest "can create with default config": let config = defaultConfig() check: config.bloomFilterCapacity == DefaultBloomFilterCapacity config.bloomFilterErrorRate == DefaultBloomFilterErrorRate config.maxMessageHistory == DefaultMaxMessageHistory - test "basic message wrapping and unwrapping": + asyncTest "basic message wrapping and unwrapping": let msg = @[byte(1), 2, 3] let msgId = "test-msg-1" - let wrappedResult = rm.wrapOutgoingMessage(msg, msgId, testChannel) + let wrappedResult = await rm.wrapOutgoingMessage(msg, msgId, testChannel) check wrappedResult.isOk() let wrapped = wrappedResult.get() check wrapped.len > 0 - let unwrapResult = rm.unwrapReceivedMessage(wrapped) + let unwrapResult = await rm.unwrapReceivedMessage(wrapped) check unwrapResult.isOk() let (unwrapped, missingDeps, channelId) = unwrapResult.get() check: @@ -55,13 +77,13 @@ suite "Core Operations": missingDeps.len == 0 channelId == testChannel - test "basic message wrapping and unwrapping (non-empty bloom)": + asyncTest "basic message wrapping and unwrapping (non-empty bloom)": rm.seedBloom(testChannel, 50) let msg = @[byte(1), 2, 3] let msgId = "test-msg-1" - let wrappedResult = rm.wrapOutgoingMessage(msg, msgId, testChannel) + let wrappedResult = await rm.wrapOutgoingMessage(msg, msgId, testChannel) check wrappedResult.isOk() let wrapped = wrappedResult.get() check wrapped.len > 0 @@ -72,7 +94,7 @@ suite "Core Operations": check decoded.isOk() check decoded.get().bloomFilter.len > 0 - let unwrapResult = rm.unwrapReceivedMessage(wrapped) + let unwrapResult = await rm.unwrapReceivedMessage(wrapped) check unwrapResult.isOk() let (unwrapped, missingDeps, channelId) = unwrapResult.get() check: @@ -80,7 +102,7 @@ suite "Core Operations": missingDeps.len == 0 channelId == testChannel - test "message ordering": + asyncTest "message ordering": # Create messages with different timestamps let msg1 = SdsMessage.init( messageId = "msg1", @@ -107,9 +129,9 @@ suite "Core Operations": serialized2.isOk() # Process out of order - discard rm.unwrapReceivedMessage(serialized2.get()) + discard await rm.unwrapReceivedMessage(serialized2.get()) let timestamp1 = rm.channels[testChannel].lamportTimestamp - discard rm.unwrapReceivedMessage(serialized1.get()) + discard await rm.unwrapReceivedMessage(serialized1.get()) let timestamp2 = rm.channels[testChannel].lamportTimestamp check timestamp2 > timestamp1 @@ -122,18 +144,18 @@ suite "Reliability Mechanisms": let rmResult = newReliabilityManager(participantId = "alice") check rmResult.isOk() rm = rmResult.get() - check rm.ensureChannel(testChannel).isOk() + check (waitFor rm.ensureChannel(testChannel)).isOk() teardown: if not rm.isNil: - rm.cleanup() + waitFor rm.cleanup() - test "dependency detection and resolution": + asyncTest "dependency detection and resolution": var messageReadyCount = 0 var messageSentCount = 0 var missingDepsCount = 0 - rm.setCallbacks( + await rm.setCallbacks( proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} = messageReadyCount += 1, proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} = @@ -173,7 +195,7 @@ suite "Reliability Mechanisms": serialized3.isOk() # First try processing msg3 (which depends on msg2 which depends on msg1) - let unwrapResult3 = rm.unwrapReceivedMessage(serialized3.get()) + let unwrapResult3 = await rm.unwrapReceivedMessage(serialized3.get()) check unwrapResult3.isOk() let (_, missingDeps3, _) = unwrapResult3.get() @@ -184,7 +206,7 @@ suite "Reliability Mechanisms": id2 in missingDeps3.getMessageIds() # Then try processing msg2 (which only depends on msg1) - let unwrapResult2 = rm.unwrapReceivedMessage(serialized2.get()) + let unwrapResult2 = await rm.unwrapReceivedMessage(serialized2.get()) check unwrapResult2.isOk() let (_, missingDeps2, _) = unwrapResult2.get() @@ -195,17 +217,17 @@ suite "Reliability Mechanisms": messageReadyCount == 0 # No messages should be ready yet # Mark first dependency (msg1) as met - let markResult1 = rm.markDependenciesMet(@[id1], testChannel) + let markResult1 = await rm.markDependenciesMet(@[id1], testChannel) check markResult1.isOk() - let incomingBuffer = rm.getIncomingBuffer(testChannel) + let incomingBuffer = await rm.getIncomingBuffer(testChannel) check: incomingBuffer.len == 0 messageReadyCount == 2 # Both msg2 and msg3 should be ready missingDepsCount == 2 # Should still be 2 from the initial missing deps - test "dependency detection and resolution (non-empty bloom)": + asyncTest "dependency detection and resolution (non-empty bloom)": # A populated bloom filter must not short-circuit the dependency check. # Dependency resolution reads messageHistory, not the bloom — but a future # "optimisation" could regress this. Seed the bloom with the dep id so a @@ -213,7 +235,7 @@ suite "Reliability Mechanisms": var missingDepsCount = 0 var messageReadyCount = 0 - rm.setCallbacks( + await rm.setCallbacks( proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} = messageReadyCount += 1, proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} = @@ -241,7 +263,7 @@ suite "Reliability Mechanisms": let serialized2 = serializeMessage(msg2) check serialized2.isOk() - let unwrapResult = rm.unwrapReceivedMessage(serialized2.get()) + let unwrapResult = await rm.unwrapReceivedMessage(serialized2.get()) check unwrapResult.isOk() let (_, missingDeps, _) = unwrapResult.get() @@ -251,12 +273,12 @@ suite "Reliability Mechanisms": id1 in missingDeps.getMessageIds() messageReadyCount == 0 - test "acknowledgment via causal history": + asyncTest "acknowledgment via causal history": var messageReadyCount = 0 var messageSentCount = 0 var missingDepsCount = 0 - rm.setCallbacks( + await rm.setCallbacks( proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} = messageReadyCount += 1, proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} = @@ -268,7 +290,7 @@ suite "Reliability Mechanisms": # Send our message let msg1 = @[byte(1)] let id1 = "msg1" - let wrap1 = rm.wrapOutgoingMessage(msg1, id1, testChannel) + let wrap1 = await rm.wrapOutgoingMessage(msg1, id1, testChannel) check wrap1.isOk() # Create a message that has our message in causal history @@ -286,21 +308,21 @@ suite "Reliability Mechanisms": check serializedMsg2.isOk() # Process the "received" message - should trigger callbacks - let unwrapResult = rm.unwrapReceivedMessage(serializedMsg2.get()) + let unwrapResult = await rm.unwrapReceivedMessage(serializedMsg2.get()) check unwrapResult.isOk() check: messageReadyCount == 1 # For msg2 which we "received" messageSentCount == 1 # For msg1 which was acknowledged via causal history - test "acknowledgment via causal history (non-empty bloom)": + asyncTest "acknowledgment via causal history (non-empty bloom)": # The causal-history ack path must not be perturbed by the local channel # bloom carrying unrelated ids, and the empty bloom on the incoming # message must not spuriously ack any of them. var messageReadyCount = 0 var messageSentCount = 0 - rm.setCallbacks( + await rm.setCallbacks( proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} = messageReadyCount += 1, proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} = @@ -313,7 +335,7 @@ suite "Reliability Mechanisms": let msg1 = @[byte(1)] let id1 = "msg1" - let wrap1 = rm.wrapOutgoingMessage(msg1, id1, testChannel) + let wrap1 = await rm.wrapOutgoingMessage(msg1, id1, testChannel) check wrap1.isOk() let msg2 = SdsMessage.init( @@ -327,17 +349,17 @@ suite "Reliability Mechanisms": let serializedMsg2 = serializeMessage(msg2) check serializedMsg2.isOk() - let unwrapResult = rm.unwrapReceivedMessage(serializedMsg2.get()) + let unwrapResult = await rm.unwrapReceivedMessage(serializedMsg2.get()) check unwrapResult.isOk() check: messageReadyCount == 1 messageSentCount == 1 # exactly id1; no spurious acks for the seeded ids - test "acknowledgment via bloom filter": + asyncTest "acknowledgment via bloom filter": var messageSentCount = 0 - rm.setCallbacks( + await rm.setCallbacks( proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} = discard, proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} = @@ -349,7 +371,7 @@ suite "Reliability Mechanisms": # Send our message let msg1 = @[byte(1)] let id1 = "msg1" - let wrap1 = rm.wrapOutgoingMessage(msg1, id1, testChannel) + let wrap1 = await rm.wrapOutgoingMessage(msg1, id1, testChannel) check wrap1.isOk() # Create a message with bloom filter containing our message @@ -372,18 +394,18 @@ suite "Reliability Mechanisms": let serializedMsg2 = serializeMessage(msg2) check serializedMsg2.isOk() - let unwrapResult = rm.unwrapReceivedMessage(serializedMsg2.get()) + let unwrapResult = await rm.unwrapReceivedMessage(serializedMsg2.get()) check unwrapResult.isOk() check messageSentCount == 1 # Our message should be acknowledged via bloom filter - test "acknowledgment via bloom filter (non-empty bloom)": + asyncTest "acknowledgment via bloom filter (non-empty bloom)": # The peer's bloom contains both our outgoing id and a pile of unrelated # ids. The manager must still ack our message exactly once, and unrelated # ids in the peer's bloom must not produce spurious sent callbacks. var messageSentCount = 0 - rm.setCallbacks( + await rm.setCallbacks( proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} = discard, proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} = @@ -394,7 +416,7 @@ suite "Reliability Mechanisms": let msg1 = @[byte(1)] let id1 = "msg1" - let wrap1 = rm.wrapOutgoingMessage(msg1, id1, testChannel) + let wrap1 = await rm.wrapOutgoingMessage(msg1, id1, testChannel) check wrap1.isOk() var otherPartyBloomFilter = @@ -417,12 +439,12 @@ suite "Reliability Mechanisms": let serializedMsg2 = serializeMessage(msg2) check serializedMsg2.isOk() - let unwrapResult = rm.unwrapReceivedMessage(serializedMsg2.get()) + let unwrapResult = await rm.unwrapReceivedMessage(serializedMsg2.get()) check unwrapResult.isOk() check messageSentCount == 1 - test "outgoing message bloom snapshot reflects channel state": + asyncTest "outgoing message bloom snapshot reflects channel state": # Until now nothing asserts that wrapOutgoingMessage actually attaches # the current bloom snapshot — every other test runs against an empty # filter where the field is empty either way. @@ -439,10 +461,10 @@ suite "Reliability Mechanisms": ) let serIncoming = serializeMessage(incoming) check serIncoming.isOk() - discard rm.unwrapReceivedMessage(serIncoming.get()) + discard await rm.unwrapReceivedMessage(serIncoming.get()) let outId = "outgoing-1" - let wrapped = rm.wrapOutgoingMessage(@[byte(1)], outId, testChannel) + let wrapped = await rm.wrapOutgoingMessage(@[byte(1)], outId, testChannel) check wrapped.isOk() let decoded = deserializeMessage(wrapped.get()) @@ -461,12 +483,12 @@ suite "Reliability Mechanisms": snapshot.contains("delivered-39") snapshot.contains("incoming-1") - test "retrieval hints": + asyncTest "retrieval hints": var messageReadyCount = 0 var messageSentCount = 0 var missingDepsCount = 0 - rm.setCallbacks( + await rm.setCallbacks( proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} = messageReadyCount += 1, proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} = @@ -481,13 +503,13 @@ suite "Reliability Mechanisms": # Send a first message to populate history let msg1 = @[byte(1)] let id1 = "msg1" - let wrap1 = rm.wrapOutgoingMessage(msg1, id1, testChannel) + let wrap1 = await rm.wrapOutgoingMessage(msg1, id1, testChannel) check wrap1.isOk() # Send a second message, which should have the first in its causal history let msg2 = @[byte(2)] let id2 = "msg2" - let wrap2 = rm.wrapOutgoingMessage(msg2, id2, testChannel) + let wrap2 = await rm.wrapOutgoingMessage(msg2, id2, testChannel) check wrap2.isOk() # Check that the wrapped message contains the hint @@ -506,14 +528,14 @@ suite "Reliability Mechanisms": bloomFilter = @[], ) let serialized3 = serializeMessage(msg3).get() - let unwrapResult3 = rm.unwrapReceivedMessage(serialized3) + let unwrapResult3 = await rm.unwrapReceivedMessage(serialized3) check unwrapResult3.isOk() let (_, missingDeps3, _) = unwrapResult3.get() check missingDeps3.len == 1 check missingDeps3[0].messageId == "missing-dep" # The hint is empty because it was not provided by the remote sender check missingDeps3[0].retrievalHint.len == 0 - + # Test with a message that HAS a retrieval hint from remote let msg4 = SdsMessage.init( messageId = "msg4", @@ -524,7 +546,7 @@ suite "Reliability Mechanisms": bloomFilter = @[], ) let serialized4 = serializeMessage(msg4).get() - let unwrapResult4 = rm.unwrapReceivedMessage(serialized4) + let unwrapResult4 = await rm.unwrapReceivedMessage(serialized4) check unwrapResult4.isOk() let (_, missingDeps4, _) = unwrapResult4.get() check missingDeps4.len == 1 @@ -540,16 +562,16 @@ suite "Periodic Tasks & Buffer Management": let rmResult = newReliabilityManager(participantId = "alice") check rmResult.isOk() rm = rmResult.get() - check rm.ensureChannel(testChannel).isOk() + check (waitFor rm.ensureChannel(testChannel)).isOk() teardown: if not rm.isNil: - rm.cleanup() + waitFor rm.cleanup() - test "outgoing buffer management": + asyncTest "outgoing buffer management": var messageSentCount = 0 - rm.setCallbacks( + await rm.setCallbacks( proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} = discard, proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} = @@ -562,10 +584,10 @@ suite "Periodic Tasks & Buffer Management": for i in 0 .. 5: let msg = @[byte(i)] let id = "msg" & $i - let wrap = rm.wrapOutgoingMessage(msg, id, testChannel) + let wrap = await rm.wrapOutgoingMessage(msg, id, testChannel) check wrap.isOk() - let outBuffer = rm.getOutgoingBuffer(testChannel) + let outBuffer = await rm.getOutgoingBuffer(testChannel) check outBuffer.len == 6 # Create message that acknowledges some messages @@ -582,15 +604,15 @@ suite "Periodic Tasks & Buffer Management": check serializedAck.isOk() # Process the acknowledgment - discard rm.unwrapReceivedMessage(serializedAck.get()) + discard await rm.unwrapReceivedMessage(serializedAck.get()) - let finalBuffer = rm.getOutgoingBuffer(testChannel) + let finalBuffer = await rm.getOutgoingBuffer(testChannel) check: finalBuffer.len == 3 # Should have removed acknowledged messages messageSentCount == 3 # Should have triggered sent callback for acknowledged messages - test "periodic buffer sweep and bloom clean": + asyncTest "periodic buffer sweep and bloom clean": var messageSentCount = 0 var config = defaultConfig() @@ -602,9 +624,9 @@ suite "Periodic Tasks & Buffer Management": let rmResultP = newReliabilityManager(participantId = "alice", config = config) check rmResultP.isOk() let rm = rmResultP.get() - check rm.ensureChannel(testChannel).isOk() + check (await rm.ensureChannel(testChannel)).isOk() - rm.setCallbacks( + await rm.setCallbacks( proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} = discard, proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} = @@ -616,10 +638,10 @@ suite "Periodic Tasks & Buffer Management": # First message - should be cleaned from bloom filter later let msg1 = @[byte(1)] let id1 = "msg1" - let wrap1 = rm.wrapOutgoingMessage(msg1, id1, testChannel) + let wrap1 = await rm.wrapOutgoingMessage(msg1, id1, testChannel) check wrap1.isOk() - let initialBuffer = rm.getOutgoingBuffer(testChannel) + let initialBuffer = await rm.getOutgoingBuffer(testChannel) check: initialBuffer[0].resendAttempts == 0 rm.channels[testChannel].bloomFilter.contains(id1) @@ -627,20 +649,20 @@ suite "Periodic Tasks & Buffer Management": rm.startPeriodicTasks() # Wait long enough for bloom filter - waitFor sleepAsync(chronos.milliseconds(500)) + await sleepAsync(chronos.milliseconds(500)) # Add new messages let msg2 = @[byte(2)] let id2 = "msg2" - let wrap2 = rm.wrapOutgoingMessage(msg2, id2, testChannel) + let wrap2 = await rm.wrapOutgoingMessage(msg2, id2, testChannel) check wrap2.isOk() let msg3 = @[byte(3)] let id3 = "msg3" - let wrap3 = rm.wrapOutgoingMessage(msg3, id3, testChannel) + let wrap3 = await rm.wrapOutgoingMessage(msg3, id3, testChannel) check wrap3.isOk() - let finalBuffer = rm.getOutgoingBuffer(testChannel) + let finalBuffer = await rm.getOutgoingBuffer(testChannel) check: finalBuffer.len == 2 # Only msg2 and msg3 should be in buffer, msg1 should be removed after max retries @@ -649,11 +671,11 @@ suite "Periodic Tasks & Buffer Management": not rm.channels[testChannel].bloomFilter.contains(id1) # Bloom filter cleaning check rm.channels[testChannel].bloomFilter.contains(id3) # New message still in filter - rm.cleanup() + await rm.cleanup() - test "periodic sync callback": + asyncTest "periodic sync callback": var syncCallCount = 0 - rm.setCallbacks( + await rm.setCallbacks( proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} = discard, proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} = @@ -665,8 +687,8 @@ suite "Periodic Tasks & Buffer Management": ) rm.startPeriodicTasks() - waitFor sleepAsync(chronos.seconds(1)) - rm.cleanup() + await sleepAsync(chronos.seconds(1)) + await rm.cleanup() check syncCallCount > 0 @@ -678,26 +700,26 @@ suite "Special Cases Handling": let rmResult = newReliabilityManager(participantId = "alice") check rmResult.isOk() rm = rmResult.get() - check rm.ensureChannel(testChannel).isOk() + check (waitFor rm.ensureChannel(testChannel)).isOk() teardown: if not rm.isNil: - rm.cleanup() + waitFor rm.cleanup() - test "message history limits": + asyncTest "message history limits": # Add messages up to max history size for i in 0 .. rm.config.maxMessageHistory + 5: let msg = @[byte(i)] let id = "msg" & $i - let wrap = rm.wrapOutgoingMessage(msg, id, testChannel) + let wrap = await rm.wrapOutgoingMessage(msg, id, testChannel) check wrap.isOk() - let history = rm.getMessageHistory(testChannel) + let history = await rm.getMessageHistory(testChannel) check: history.len <= rm.config.maxMessageHistory history[^1] == "msg" & $(rm.config.maxMessageHistory + 5) - test "invalid bloom filter handling": + asyncTest "invalid bloom filter handling": let msgInvalid = SdsMessage.init( messageId = "invalid-bf", lamportTimestamp = 1, @@ -712,14 +734,14 @@ suite "Special Cases Handling": check serializedInvalid.isOk() # Should handle invalid bloom filter gracefully - let result = rm.unwrapReceivedMessage(serializedInvalid.get()) + let result = await rm.unwrapReceivedMessage(serializedInvalid.get()) check: result.isOk() result.get()[1].len == 0 # No missing dependencies - test "duplicate message handling": + asyncTest "duplicate message handling": var messageReadyCount = 0 - rm.setCallbacks( + await rm.setCallbacks( proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} = messageReadyCount += 1, proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} = @@ -742,45 +764,45 @@ suite "Special Cases Handling": check serialized.isOk() # Process same message twice - let result1 = rm.unwrapReceivedMessage(serialized.get()) + let result1 = await rm.unwrapReceivedMessage(serialized.get()) check result1.isOk() - let result2 = rm.unwrapReceivedMessage(serialized.get()) + let result2 = await rm.unwrapReceivedMessage(serialized.get()) check: result2.isOk() result2.get()[1].len == 0 # No missing deps on second process messageReadyCount == 1 # Message should only be processed once - test "error handling": + asyncTest "error handling": # Empty message let emptyMsg: seq[byte] = @[] - let emptyResult = rm.wrapOutgoingMessage(emptyMsg, "empty", testChannel) + let emptyResult = await rm.wrapOutgoingMessage(emptyMsg, "empty", testChannel) check: not emptyResult.isOk() emptyResult.error == reInvalidArgument # Oversized message let largeMsg = newSeq[byte](MaxMessageSize + 1) - let largeResult = rm.wrapOutgoingMessage(largeMsg, "large", testChannel) + let largeResult = await rm.wrapOutgoingMessage(largeMsg, "large", testChannel) check: not largeResult.isOk() largeResult.error == reMessageTooLarge suite "cleanup": - test "cleanup works correctly": + asyncTest "cleanup works correctly": let rmResult = newReliabilityManager(participantId = "alice") check rmResult.isOk() let rm = rmResult.get() - check rm.ensureChannel(testChannel).isOk() + check (await rm.ensureChannel(testChannel)).isOk() # Add some messages let msg = @[byte(1), 2, 3] let msgId = "test-msg-1" - discard rm.wrapOutgoingMessage(msg, msgId, testChannel) + discard await rm.wrapOutgoingMessage(msg, msgId, testChannel) - rm.cleanup() + await rm.cleanup() - let outBuffer = rm.getOutgoingBuffer(testChannel) - let history = rm.getMessageHistory(testChannel) + let outBuffer = await rm.getOutgoingBuffer(testChannel) + let history = await rm.getMessageHistory(testChannel) check: outBuffer.len == 0 history.len == 0 @@ -795,43 +817,43 @@ suite "Multi-Channel ReliabilityManager Tests": teardown: if not rm.isNil: - rm.cleanup() + waitFor rm.cleanup() - test "can create multi-channel manager without channel ID": + asyncTest "can create multi-channel manager without channel ID": check rm.channels.len == 0 - test "channel management": + asyncTest "channel management": let channel1 = "channel1" let channel2 = "channel2" # Ensure channels - check rm.ensureChannel(channel1).isOk() - check rm.ensureChannel(channel2).isOk() + check (await rm.ensureChannel(channel1)).isOk() + check (await rm.ensureChannel(channel2)).isOk() check rm.channels.len == 2 # Remove channel - check rm.removeChannel(channel1).isOk() + check (await rm.removeChannel(channel1)).isOk() check rm.channels.len == 1 check channel1 notin rm.channels check channel2 in rm.channels - test "stateless message unwrapping with channel extraction": + asyncTest "stateless message unwrapping with channel extraction": let channel1 = "test-channel-1" let channel2 = "test-channel-2" # Create and wrap messages for different channels let msg1 = @[byte(1), 2, 3] let msgId1 = "msg1" - let wrapped1 = rm.wrapOutgoingMessage(msg1, msgId1, channel1) + let wrapped1 = await rm.wrapOutgoingMessage(msg1, msgId1, channel1) check wrapped1.isOk() let msg2 = @[byte(4), 5, 6] let msgId2 = "msg2" - let wrapped2 = rm.wrapOutgoingMessage(msg2, msgId2, channel2) + let wrapped2 = await rm.wrapOutgoingMessage(msg2, msgId2, channel2) check wrapped2.isOk() # Unwrap messages - should extract channel ID and route correctly - let unwrap1 = rm.unwrapReceivedMessage(wrapped1.get()) + let unwrap1 = await rm.unwrapReceivedMessage(wrapped1.get()) check unwrap1.isOk() let (content1, deps1, extractedChannel1) = unwrap1.get() check: @@ -839,7 +861,7 @@ suite "Multi-Channel ReliabilityManager Tests": deps1.len == 0 extractedChannel1 == channel1 - let unwrap2 = rm.unwrapReceivedMessage(wrapped2.get()) + let unwrap2 = await rm.unwrapReceivedMessage(wrapped2.get()) check unwrap2.isOk() let (content2, deps2, extractedChannel2) = unwrap2.get() check: @@ -847,22 +869,22 @@ suite "Multi-Channel ReliabilityManager Tests": deps2.len == 0 extractedChannel2 == channel2 - test "channel isolation": + asyncTest "channel isolation": let channel1 = "isolated-channel-1" let channel2 = "isolated-channel-2" # Add messages to different channels let msg1 = @[byte(1)] let msgId1 = "isolated-msg1" - discard rm.wrapOutgoingMessage(msg1, msgId1, channel1) + discard await rm.wrapOutgoingMessage(msg1, msgId1, channel1) let msg2 = @[byte(2)] let msgId2 = "isolated-msg2" - discard rm.wrapOutgoingMessage(msg2, msgId2, channel2) + discard await rm.wrapOutgoingMessage(msg2, msgId2, channel2) # Check channel-specific data is isolated - let history1 = rm.getMessageHistory(channel1) - let history2 = rm.getMessageHistory(channel2) + let history1 = await rm.getMessageHistory(channel1) + let history2 = await rm.getMessageHistory(channel2) check: history1.len == 1 @@ -872,20 +894,20 @@ suite "Multi-Channel ReliabilityManager Tests": msgId1 notin history2 msgId2 notin history1 - test "channel isolation (non-empty bloom)": + asyncTest "channel isolation (non-empty bloom)": # With both channels carrying populated blooms, ids on one channel must # not appear in the other's filter. An empty-bloom test cannot observe # this — there is nothing to bleed across. let channel1 = "iso-bloom-1" let channel2 = "iso-bloom-2" - check rm.ensureChannel(channel1).isOk() - check rm.ensureChannel(channel2).isOk() + check (await rm.ensureChannel(channel1)).isOk() + check (await rm.ensureChannel(channel2)).isOk() rm.seedBloom(channel1, 25, prefix = "ch1-") rm.seedBloom(channel2, 25, prefix = "ch2-") - let wrap1 = rm.wrapOutgoingMessage(@[byte(1)], "iso-msg-1", channel1) - let wrap2 = rm.wrapOutgoingMessage(@[byte(2)], "iso-msg-2", channel2) + let wrap1 = await rm.wrapOutgoingMessage(@[byte(1)], "iso-msg-1", channel1) + let wrap2 = await rm.wrapOutgoingMessage(@[byte(2)], "iso-msg-2", channel2) check wrap1.isOk() and wrap2.isOk() let bf1 = rm.channels[channel1].bloomFilter @@ -900,12 +922,12 @@ suite "Multi-Channel ReliabilityManager Tests": not bf2.contains("ch1-0") not bf2.contains("iso-msg-1") - test "multi-channel callbacks": + asyncTest "multi-channel callbacks": var readyMessageCount = 0 var sentMessageCount = 0 var missingDepsCount = 0 - rm.setCallbacks( + await rm.setCallbacks( proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} = readyMessageCount += 1, proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} = @@ -920,12 +942,12 @@ suite "Multi-Channel ReliabilityManager Tests": # Send messages from both channels let msg1 = @[byte(1)] let msgId1 = "callback-msg1" - let wrapped1 = rm.wrapOutgoingMessage(msg1, msgId1, channel1) + let wrapped1 = await rm.wrapOutgoingMessage(msg1, msgId1, channel1) check wrapped1.isOk() let msg2 = @[byte(2)] let msgId2 = "callback-msg2" - let wrapped2 = rm.wrapOutgoingMessage(msg2, msgId2, channel2) + let wrapped2 = await rm.wrapOutgoingMessage(msg2, msgId2, channel2) check wrapped2.isOk() # Create acknowledgment messages that include our message IDs in causal history @@ -955,25 +977,25 @@ suite "Multi-Channel ReliabilityManager Tests": serializedAck2.isOk() # Process acknowledgment messages - should trigger callbacks - discard rm.unwrapReceivedMessage(serializedAck1.get()) - discard rm.unwrapReceivedMessage(serializedAck2.get()) + discard await rm.unwrapReceivedMessage(serializedAck1.get()) + discard await rm.unwrapReceivedMessage(serializedAck2.get()) check: readyMessageCount == 2 # Both ack messages should trigger ready callbacks sentMessageCount == 2 # Both original messages should be marked as sent missingDepsCount == 0 # No missing dependencies - test "channel-specific dependency management": + asyncTest "channel-specific dependency management": let channel1 = "dep-channel-1" let channel2 = "dep-channel-2" let depIds = @["dep1", "dep2", "dep3"] # Ensure both channels exist first - check rm.ensureChannel(channel1).isOk() - check rm.ensureChannel(channel2).isOk() + check (await rm.ensureChannel(channel1)).isOk() + check (await rm.ensureChannel(channel2)).isOk() # Mark dependencies as met for specific channel - check rm.markDependenciesMet(depIds, channel1).isOk() + check (await rm.markDependenciesMet(depIds, channel1)).isOk() # Dependencies should only affect the specified channel # Dependencies in channel1 should not affect channel2 @@ -1034,16 +1056,16 @@ suite "SDS-R: Repair Buffer Management": ) check rmResult.isOk() rm = rmResult.get() - check rm.ensureChannel(testChannel).isOk() + check (waitFor rm.ensureChannel(testChannel)).isOk() teardown: if not rm.isNil: - rm.cleanup() + waitFor rm.cleanup() - test "missing deps added to outgoing repair buffer": + asyncTest "missing deps added to outgoing repair buffer": var missingDepsCount = 0 - rm.setCallbacks( + await rm.setCallbacks( proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} = discard, proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} = discard, proc(messageId: SdsMessageID, missingDeps: seq[HistoryEntry], channelId: SdsChannelID) {.gcsafe.} = @@ -1061,7 +1083,7 @@ suite "SDS-R: Repair Buffer Management": ) let serialized = serializeMessage(msg).get() - let result = rm.unwrapReceivedMessage(serialized) + let result = await rm.unwrapReceivedMessage(serialized) check result.isOk() # msg1 should be in the outgoing repair buffer @@ -1070,8 +1092,8 @@ suite "SDS-R: Repair Buffer Management": missingDepsCount == 1 "msg1" in channel.outgoingRepairBuffer - test "receiving message clears it from repair buffers": - rm.setCallbacks( + asyncTest "receiving message clears it from repair buffers": + await rm.setCallbacks( proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} = discard, proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} = discard, proc(messageId: SdsMessageID, missingDeps: seq[HistoryEntry], channelId: SdsChannelID) {.gcsafe.} = discard, @@ -1086,7 +1108,7 @@ suite "SDS-R: Repair Buffer Management": content = @[byte(2)], bloomFilter = @[], ) - discard rm.unwrapReceivedMessage(serializeMessage(msg2).get()) + discard await rm.unwrapReceivedMessage(serializeMessage(msg2).get()) check "msg1" in rm.channels[testChannel].outgoingRepairBuffer # Now receive msg1 — should clear from repair buffer @@ -1098,11 +1120,11 @@ suite "SDS-R: Repair Buffer Management": content = @[byte(1)], bloomFilter = @[], ) - discard rm.unwrapReceivedMessage(serializeMessage(msg1).get()) + discard await rm.unwrapReceivedMessage(serializeMessage(msg1).get()) check "msg1" notin rm.channels[testChannel].outgoingRepairBuffer - test "markDependenciesMet clears repair buffers": - rm.setCallbacks( + asyncTest "markDependenciesMet clears repair buffers": + await rm.setCallbacks( proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} = discard, proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} = discard, proc(messageId: SdsMessageID, missingDeps: seq[HistoryEntry], channelId: SdsChannelID) {.gcsafe.} = discard, @@ -1116,15 +1138,15 @@ suite "SDS-R: Repair Buffer Management": content = @[byte(2)], bloomFilter = @[], ) - discard rm.unwrapReceivedMessage(serializeMessage(msg2).get()) + discard await rm.unwrapReceivedMessage(serializeMessage(msg2).get()) check "msg1" in rm.channels[testChannel].outgoingRepairBuffer # Mark as met via store retrieval - check rm.markDependenciesMet(@["msg1"], testChannel).isOk() + check (await rm.markDependenciesMet(@["msg1"], testChannel)).isOk() check "msg1" notin rm.channels[testChannel].outgoingRepairBuffer - test "expired repair requests attached to outgoing messages": - rm.setCallbacks( + asyncTest "expired repair requests attached to outgoing messages": + await rm.setCallbacks( proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} = discard, proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} = discard, proc(messageId: SdsMessageID, missingDeps: seq[HistoryEntry], channelId: SdsChannelID) {.gcsafe.} = discard, @@ -1138,7 +1160,7 @@ suite "SDS-R: Repair Buffer Management": ) # Send a message — should pick up the expired repair request - let wrapped = rm.wrapOutgoingMessage(@[byte(1)], "new-msg", testChannel) + let wrapped = await rm.wrapOutgoingMessage(@[byte(1)], "new-msg", testChannel) check wrapped.isOk() let unwrapped = deserializeMessage(wrapped.get()).get() @@ -1148,11 +1170,11 @@ suite "SDS-R: Repair Buffer Management": # Should be removed from buffer after attaching "missing-msg" notin channel.outgoingRepairBuffer - test "expired repair requests attach the most-overdue first when capped": + asyncTest "expired repair requests attach the most-overdue first when capped": # Per spec (sds-r-send-message, RECOMMENDED): when more entries are # eligible than maxRepairRequests, attach the ones with the smallest # minTimeRepairReq — i.e. the most overdue. - rm.setCallbacks( + await rm.setCallbacks( proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} = discard, proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} = discard, proc(messageId: SdsMessageID, missingDeps: seq[HistoryEntry], channelId: SdsChannelID) {.gcsafe.} = discard, @@ -1169,7 +1191,7 @@ suite "SDS-R: Repair Buffer Management": minTimeRepairReq: now - initDuration(seconds = 50 - i * 10), ) - let wrapped = rm.wrapOutgoingMessage(@[byte(1)], "outbound", testChannel) + let wrapped = await rm.wrapOutgoingMessage(@[byte(1)], "outbound", testChannel) check wrapped.isOk() let attached = deserializeMessage(wrapped.get()).get().repairRequest @@ -1185,8 +1207,8 @@ suite "SDS-R: Repair Buffer Management": "second" notin channel.outgoingRepairBuffer "third" notin channel.outgoingRepairBuffer - test "incoming repair request adds to incoming repair buffer when eligible": - rm.setCallbacks( + asyncTest "incoming repair request adds to incoming repair buffer when eligible": + await rm.setCallbacks( proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} = discard, proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} = discard, proc(messageId: SdsMessageID, missingDeps: seq[HistoryEntry], channelId: SdsChannelID) {.gcsafe.} = discard, @@ -1218,7 +1240,7 @@ suite "SDS-R: Repair Buffer Management": senderId: "test-participant", # Same as our participantId so we're in response group )], ) - discard rm.unwrapReceivedMessage(serializeMessage(msgWithRepair).get()) + discard await rm.unwrapReceivedMessage(serializeMessage(msgWithRepair).get()) # We should have added it to the incoming repair buffer (we have the message and are in response group) check "cached-msg" in channel.incomingRepairBuffer @@ -1370,14 +1392,13 @@ suite "SDS-R: Edge Cases and Defensive Branches": check other.inMilliseconds >= selfD.inMilliseconds suite "SDS-R: Lifecycle and State": - test "empty participantId disables outgoing repair creation": + asyncTest "empty participantId disables outgoing repair creation": # Explicitly pass empty id to exercise the SDS-R no-op branch. Required-arg # signature means callers can no longer accidentally land here. let rm = newReliabilityManager(participantId = "".SdsParticipantID).get() - defer: rm.cleanup() - check rm.ensureChannel(testChannel).isOk() + check (await rm.ensureChannel(testChannel)).isOk() - rm.setCallbacks( + await rm.setCallbacks( proc(msgId: SdsMessageID, ch: SdsChannelID) {.gcsafe.} = discard, proc(msgId: SdsMessageID, ch: SdsChannelID) {.gcsafe.} = discard, proc(msgId: SdsMessageID, deps: seq[HistoryEntry], ch: SdsChannelID) {.gcsafe.} = discard, @@ -1391,13 +1412,13 @@ suite "SDS-R: Lifecycle and State": content = @[byte(2)], bloomFilter = @[], ) - discard rm.unwrapReceivedMessage(serializeMessage(msg).get()) + discard await rm.unwrapReceivedMessage(serializeMessage(msg).get()) check rm.channels[testChannel].outgoingRepairBuffer.len == 0 + await rm.cleanup() - test "empty senderId in incoming repair request is ignored": + asyncTest "empty senderId in incoming repair request is ignored": let rm = newReliabilityManager(participantId = "bob").get() - defer: rm.cleanup() - check rm.ensureChannel(testChannel).isOk() + check (await rm.ensureChannel(testChannel)).isOk() let channel = rm.channels[testChannel] channel.messageHistory["m-wanted"] = SdsMessage.init( messageId = "m-wanted", @@ -1408,7 +1429,7 @@ suite "SDS-R: Lifecycle and State": bloomFilter = @[], ) - rm.setCallbacks( + await rm.setCallbacks( proc(msgId: SdsMessageID, ch: SdsChannelID) {.gcsafe.} = discard, proc(msgId: SdsMessageID, ch: SdsChannelID) {.gcsafe.} = discard, proc(msgId: SdsMessageID, deps: seq[HistoryEntry], ch: SdsChannelID) {.gcsafe.} = discard, @@ -1423,42 +1444,42 @@ suite "SDS-R: Lifecycle and State": bloomFilter = @[], repairRequest = @[HistoryEntry(messageId: "m-wanted", senderId: "")], ) - discard rm.unwrapReceivedMessage(serializeMessage(msg).get()) + discard await rm.unwrapReceivedMessage(serializeMessage(msg).get()) check "m-wanted" notin channel.incomingRepairBuffer + await rm.cleanup() - test "wrapOutgoingMessage records the message in history with our senderId": + asyncTest "wrapOutgoingMessage records the message in history with our senderId": # Proves Bug 1 is fixed — the original sender can serve her own message. # In the consolidated history model, the SdsMessage itself carries senderId # and can be re-serialized on demand for repair, so a single membership # check + senderId read covers both halves of the original assertion. let rm = newReliabilityManager(participantId = "alice").get() - defer: rm.cleanup() - check rm.ensureChannel(testChannel).isOk() + check (await rm.ensureChannel(testChannel)).isOk() - discard rm.wrapOutgoingMessage(@[byte(1), 2, 3], "m1", testChannel) + discard await rm.wrapOutgoingMessage(@[byte(1), 2, 3], "m1", testChannel) let channel = rm.channels[testChannel] check: "m1" in channel.messageHistory channel.messageHistory["m1"].senderId == "alice" channel.messageHistory["m1"].content == @[byte(1), 2, 3] + await rm.cleanup() - test "getRecentHistoryEntries carries senderId for own messages": + asyncTest "getRecentHistoryEntries carries senderId for own messages": let rm = newReliabilityManager(participantId = "alice").get() - defer: rm.cleanup() - check rm.ensureChannel(testChannel).isOk() + check (await rm.ensureChannel(testChannel)).isOk() - discard rm.wrapOutgoingMessage(@[byte(1)], "m1", testChannel) - discard rm.wrapOutgoingMessage(@[byte(2)], "m2", testChannel) - let entries = rm.getRecentHistoryEntries(10, testChannel) + discard await rm.wrapOutgoingMessage(@[byte(1)], "m1", testChannel) + discard await rm.wrapOutgoingMessage(@[byte(2)], "m2", testChannel) + let entries = await rm.getRecentHistoryEntries(10, testChannel) check: entries.len == 2 entries[0].senderId == "alice" entries[1].senderId == "alice" + await rm.cleanup() - test "resetReliabilityManager clears all SDS-R state": + asyncTest "resetReliabilityManager clears all SDS-R state": let rm = newReliabilityManager(participantId = "alice").get() - defer: rm.cleanup() - check rm.ensureChannel(testChannel).isOk() + check (await rm.ensureChannel(testChannel)).isOk() let channel = rm.channels[testChannel] channel.outgoingRepairBuffer["a"] = OutgoingRepairEntry( @@ -1480,21 +1501,21 @@ suite "SDS-R: Lifecycle and State": senderId = "someone", ) - check rm.resetReliabilityManager().isOk() - check rm.ensureChannel(testChannel).isOk() + check (await rm.resetReliabilityManager()).isOk() + check (await rm.ensureChannel(testChannel)).isOk() let ch2 = rm.channels[testChannel] check: ch2.outgoingRepairBuffer.len == 0 ch2.incomingRepairBuffer.len == 0 ch2.messageHistory.len == 0 + await rm.cleanup() - test "SDS-R state is isolated per channel": + asyncTest "SDS-R state is isolated per channel": let rm = newReliabilityManager(participantId = "alice").get() - defer: rm.cleanup() - check rm.ensureChannel("ch-A").isOk() - check rm.ensureChannel("ch-B").isOk() + check (await rm.ensureChannel("ch-A")).isOk() + check (await rm.ensureChannel("ch-B")).isOk() - rm.setCallbacks( + await rm.setCallbacks( proc(msgId: SdsMessageID, ch: SdsChannelID) {.gcsafe.} = discard, proc(msgId: SdsMessageID, ch: SdsChannelID) {.gcsafe.} = discard, proc(msgId: SdsMessageID, deps: seq[HistoryEntry], ch: SdsChannelID) {.gcsafe.} = discard, @@ -1508,20 +1529,20 @@ suite "SDS-R: Lifecycle and State": content = @[byte(2)], bloomFilter = @[], ) - discard rm.unwrapReceivedMessage(serializeMessage(msg).get()) + discard await rm.unwrapReceivedMessage(serializeMessage(msg).get()) check: rm.channels["ch-A"].outgoingRepairBuffer.len == 1 rm.channels["ch-B"].outgoingRepairBuffer.len == 0 + await rm.cleanup() - test "duplicate message arrival cancels pending incoming repair entry": + asyncTest "duplicate message arrival cancels pending incoming repair entry": # Covers the dedup-before-cleanup fix: a rebroadcast arriving at a peer who # already has the message must clear that peer's incomingRepairBuffer entry. let rm = newReliabilityManager(participantId = "carol").get() - defer: rm.cleanup() - check rm.ensureChannel(testChannel).isOk() + check (await rm.ensureChannel(testChannel)).isOk() let channel = rm.channels[testChannel] - rm.setCallbacks( + await rm.setCallbacks( proc(msgId: SdsMessageID, ch: SdsChannelID) {.gcsafe.} = discard, proc(msgId: SdsMessageID, ch: SdsChannelID) {.gcsafe.} = discard, proc(msgId: SdsMessageID, deps: seq[HistoryEntry], ch: SdsChannelID) {.gcsafe.} = discard, @@ -1552,24 +1573,25 @@ suite "SDS-R: Lifecycle and State": bloomFilter = @[], senderId = "alice", ) - discard rm.unwrapReceivedMessage(serializeMessage(msg).get()) + discard await rm.unwrapReceivedMessage(serializeMessage(msg).get()) check "m1" notin channel.incomingRepairBuffer + await rm.cleanup() suite "SDS-R: Repair Sweep": var rm: ReliabilityManager setup: rm = newReliabilityManager(participantId = "bob").get() - check rm.ensureChannel(testChannel).isOk() + check (waitFor rm.ensureChannel(testChannel)).isOk() teardown: if not rm.isNil: - rm.cleanup() + waitFor rm.cleanup() - test "runRepairSweep fires onRepairReady for expired tResp": + asyncTest "runRepairSweep fires onRepairReady for expired tResp": var fireCount = 0 var firstBytes: seq[byte] = @[] - rm.setCallbacks( + await rm.setCallbacks( proc(msgId: SdsMessageID, ch: SdsChannelID) {.gcsafe.} = discard, proc(msgId: SdsMessageID, ch: SdsChannelID) {.gcsafe.} = discard, proc(msgId: SdsMessageID, deps: seq[HistoryEntry], ch: SdsChannelID) {.gcsafe.} = discard, @@ -1592,7 +1614,7 @@ suite "SDS-R: Repair Sweep": minTimeRepairResp: getTime() + initDuration(minutes = 10), # far future ) - rm.runRepairSweep() + await rm.runRepairSweep() check: fireCount == 1 @@ -1600,8 +1622,8 @@ suite "SDS-R: Repair Sweep": "m-ready" notin channel.incomingRepairBuffer "m-not-ready" in channel.incomingRepairBuffer - test "runRepairSweep drops outgoing entries past T_max window": - rm.setCallbacks( + asyncTest "runRepairSweep drops outgoing entries past T_max window": + await rm.setCallbacks( proc(msgId: SdsMessageID, ch: SdsChannelID) {.gcsafe.} = discard, proc(msgId: SdsMessageID, ch: SdsChannelID) {.gcsafe.} = discard, proc(msgId: SdsMessageID, deps: seq[HistoryEntry], ch: SdsChannelID) {.gcsafe.} = discard, @@ -1618,22 +1640,22 @@ suite "SDS-R: Repair Sweep": minTimeRepairReq: getTime(), ) - rm.runRepairSweep() + await rm.runRepairSweep() check: "m-stale" notin channel.outgoingRepairBuffer "m-fresh" in channel.outgoingRepairBuffer - test "runRepairSweep no-op when buffers are empty": + asyncTest "runRepairSweep no-op when buffers are empty": var fireCount = 0 - rm.setCallbacks( + await rm.setCallbacks( proc(msgId: SdsMessageID, ch: SdsChannelID) {.gcsafe.} = discard, proc(msgId: SdsMessageID, ch: SdsChannelID) {.gcsafe.} = discard, proc(msgId: SdsMessageID, deps: seq[HistoryEntry], ch: SdsChannelID) {.gcsafe.} = discard, onRepairReady = proc(bytes: seq[byte], ch: SdsChannelID) {.gcsafe.} = fireCount += 1, ) - rm.runRepairSweep() + await rm.runRepairSweep() check fireCount == 0 # --- Multi-participant in-process bus for integration tests --------------- @@ -1662,25 +1684,25 @@ proc deliverExcept( senderId: SdsParticipantID, bytes: seq[byte], exclude: seq[SdsParticipantID], -) {.gcsafe.} = +): Future[void] {.async: (raises: []), gcsafe.} = for pid, peer in bus.peers: if pid == senderId or pid in exclude: continue - discard peer.unwrapReceivedMessage(bytes) + discard await peer.unwrapReceivedMessage(bytes) proc addPeer( bus: TestBus, participantId: SdsParticipantID, config: ReliabilityConfig = defaultConfig(), -): ReliabilityManager = +): Future[ReliabilityManager] {.async.} = let rm = newReliabilityManager(participantId, config).get() - doAssert rm.ensureChannel(testChannel).isOk() + doAssert (await rm.ensureChannel(testChannel)).isOk() bus.peers[participantId] = rm bus.delivered[participantId] = @[] let pid = participantId let busRef = bus - rm.setCallbacks( + await rm.setCallbacks( proc(msgId: SdsMessageID, ch: SdsChannelID) {.gcsafe.} = {.cast(gcsafe).}: busRef.delivered[pid].add(msgId), @@ -1689,9 +1711,12 @@ proc addPeer( onRepairReady = proc(bytes: seq[byte], ch: SdsChannelID) {.gcsafe.} = {.cast(gcsafe).}: busRef.recordWire(pid, bytes) - busRef.deliverExcept(pid, bytes, @[]), + # Fire-and-forget delivery from a sync callback context — we cannot + # await here, so spawn the async delivery onto the same event loop. + asyncSpawn(busRef.deliverExcept(pid, bytes, @[])) + , ) - rm + return rm proc broadcast( bus: TestBus, @@ -1699,12 +1724,12 @@ proc broadcast( content: seq[byte], messageId: SdsMessageID, dropAt: seq[SdsParticipantID] = @[], -) = +): Future[void] {.async.} = let rm = bus.peers[senderId] - let wrapped = rm.wrapOutgoingMessage(content, messageId, testChannel) + let wrapped = await rm.wrapOutgoingMessage(content, messageId, testChannel) doAssert wrapped.isOk() bus.recordWire(senderId, wrapped.get()) - bus.deliverExcept(senderId, wrapped.get(), dropAt) + await bus.deliverExcept(senderId, wrapped.get(), dropAt) proc forceOutgoingExpired( rm: ReliabilityManager, messageId: SdsMessageID @@ -1727,20 +1752,20 @@ proc forceIncomingExpired( suite "SDS-R: Multi-Participant Integration": - test "basic single-gap repair (Alice -> Bob misses -> Carol's message triggers repair)": + asyncTest "basic single-gap repair (Alice -> Bob misses -> Carol's message triggers repair)": let bus = newTestBus() - let alice = bus.addPeer("alice") - let bob = bus.addPeer("bob") - let carol = bus.addPeer("carol") + let alice = await bus.addPeer("alice") + let bob = await bus.addPeer("bob") + let carol = await bus.addPeer("carol") # Alice sends M1, but Bob is offline for this one. - bus.broadcast("alice", @[byte(1)], "m1", dropAt = @["bob".SdsParticipantID]) + await bus.broadcast("alice", @[byte(1)], "m1", dropAt = @["bob".SdsParticipantID]) # Carol now has M1; Bob does not. check "m1" in carol.channels[testChannel].messageHistory check "m1" notin bob.channels[testChannel].messageHistory # Carol sends M2 with causal history referencing M1. - bus.broadcast("carol", @[byte(2)], "m2") + await bus.broadcast("carol", @[byte(2)], "m2") # Bob detects M1 missing and populates his outgoingRepairBuffer. check "m1" in bob.channels[testChannel].outgoingRepairBuffer # Bob should have buffered M2. @@ -1751,7 +1776,7 @@ suite "SDS-R: Multi-Participant Integration": bob.forceOutgoingExpired("m1") # Bob sends M3 — it must carry repair_request =[M1, sender =alice]. - bus.broadcast("bob", @[byte(3)], "m3") + await bus.broadcast("bob", @[byte(3)], "m3") # Alice received M3, saw the repair_request, cached-bypass and response-group # checks pass, so she has an incomingRepairBuffer entry for M1 with tResp =0. @@ -1760,28 +1785,31 @@ suite "SDS-R: Multi-Participant Integration": # Force alice's tResp to past just to be safe (it's already 0 for self), # then run her sweep. She rebroadcasts M1. alice.forceIncomingExpired("m1") - alice.runRepairSweep() + await alice.runRepairSweep() + + # Allow any asyncSpawn'd deliveries from the repair callback to run. + await sleepAsync(chronos.milliseconds(10)) # Bob now has M1 and M2 delivered. check: "m1" in bus.delivered["bob"] "m2" in bus.delivered["bob"] - test "response cancellation: only one rebroadcast on the wire": + asyncTest "response cancellation: only one rebroadcast on the wire": let bus = newTestBus() - let alice = bus.addPeer("alice") - let bob = bus.addPeer("bob") - let carol = bus.addPeer("carol") + let alice = await bus.addPeer("alice") + let bob = await bus.addPeer("bob") + let carol = await bus.addPeer("carol") # Alice sends M1, Bob offline. - bus.broadcast("alice", @[byte(1)], "m1", dropAt = @["bob".SdsParticipantID]) + await bus.broadcast("alice", @[byte(1)], "m1", dropAt = @["bob".SdsParticipantID]) # Carol sends M2; Bob sees M1 missing. - bus.broadcast("carol", @[byte(2)], "m2") + await bus.broadcast("carol", @[byte(2)], "m2") check "m1" in bob.channels[testChannel].outgoingRepairBuffer # Bob requests repair. bob.forceOutgoingExpired("m1") - bus.broadcast("bob", @[byte(3)], "m3") + await bus.broadcast("bob", @[byte(3)], "m3") # Both Alice and Carol now have an incomingRepairBuffer entry for M1. check: @@ -1791,14 +1819,16 @@ suite "SDS-R: Multi-Participant Integration": # Alice fires first (T_resp =0 for self). Her rebroadcast should cancel Carol's # pending entry when Carol receives the rebroadcast. alice.forceIncomingExpired("m1") - alice.runRepairSweep() + await alice.runRepairSweep() + await sleepAsync(chronos.milliseconds(10)) # Carol's pending response must have been cleared by the dedup-path cleanup. check "m1" notin carol.channels[testChannel].incomingRepairBuffer # Even if we now force-run Carol's sweep, nothing should fire. let wireCountBefore = bus.wireLog.len - carol.runRepairSweep() + await carol.runRepairSweep() + await sleepAsync(chronos.milliseconds(10)) check bus.wireLog.len == wireCountBefore # Bob received exactly one rebroadcast of M1. @@ -1811,32 +1841,32 @@ suite "SDS-R: Multi-Participant Integration": # Two "m1" entries total on wire: (1) Alice's original broadcast, (2) Alice's rebroadcast. check m1RebroadcastCount == 2 - test "cancellation on incoming repair request: peer drops its own pending request": + asyncTest "cancellation on incoming repair request: peer drops its own pending request": let bus = newTestBus() - let alice = bus.addPeer("alice") - let bob = bus.addPeer("bob") - let carol = bus.addPeer("carol") + let alice = await bus.addPeer("alice") + let bob = await bus.addPeer("bob") + let carol = await bus.addPeer("carol") # Alice sends M1 — drop at both Bob and Carol, so both miss it. - bus.broadcast( + await bus.broadcast( "alice", @[byte(1)], "m1", dropAt = @["bob".SdsParticipantID, "carol".SdsParticipantID], ) # Alice sends M2 referencing M1 — both Bob and Carol see M1 missing. - bus.broadcast("alice", @[byte(2)], "m2") + await bus.broadcast("alice", @[byte(2)], "m2") check: "m1" in bob.channels[testChannel].outgoingRepairBuffer "m1" in carol.channels[testChannel].outgoingRepairBuffer # Bob's T_req fires first. He sends a repair request for M1. bob.forceOutgoingExpired("m1") - bus.broadcast("bob", @[byte(3)], "m3") + await bus.broadcast("bob", @[byte(3)], "m3") # Carol, on receiving Bob's repair request, must have dropped her own # pending outgoingRepairBuffer entry for M1 (cancellation). check "m1" notin carol.channels[testChannel].outgoingRepairBuffer - test "response group filtering: only group members respond": + asyncTest "response group filtering: only group members respond": # With numGroups =10, roughly 1/10 of receivers will be in the group. # Construct a sender+message where a specific non-sender is NOT in the group. var cfg = defaultConfig() @@ -1855,17 +1885,17 @@ suite "SDS-R: Multi-Participant Integration": check chosenMsg.len > 0 let bus = newTestBus() - discard bus.addPeer("alice", cfg) - let bob = bus.addPeer("bob", cfg) - let carol = bus.addPeer("carol", cfg) + discard await bus.addPeer("alice", cfg) + let bob = await bus.addPeer("bob", cfg) + let carol = await bus.addPeer("carol", cfg) # Both Bob and Carol receive the original M1 (so both have it in messageHistory). - bus.broadcast("alice", @[byte(1)], chosenMsg) + await bus.broadcast("alice", @[byte(1)], chosenMsg) # Now Dave arrives: build a fake requester message manually so its repair_request # names Alice as senderId for chosenMsg. # We inject directly by calling unwrapReceivedMessage on bob/carol. - let dave = bus.addPeer("dave", cfg) + discard await bus.addPeer("dave", cfg) # Dave has no messages, but we can hand-craft a repair request he would send. let reqMsg = SdsMessage.init( messageId = "req-from-dave", @@ -1878,28 +1908,28 @@ suite "SDS-R: Multi-Participant Integration": repairRequest = @[HistoryEntry(messageId: chosenMsg, senderId: "alice")], ) let bytes = serializeMessage(reqMsg).get() - discard bob.unwrapReceivedMessage(bytes) - discard carol.unwrapReceivedMessage(bytes) + discard await bob.unwrapReceivedMessage(bytes) + discard await carol.unwrapReceivedMessage(bytes) check: chosenMsg in bob.channels[testChannel].incomingRepairBuffer chosenMsg notin carol.channels[testChannel].incomingRepairBuffer - test "multi-gap batch repair: many missing deps split across requests": + asyncTest "multi-gap batch repair: many missing deps split across requests": let bus = newTestBus() - discard bus.addPeer("alice") - let bob = bus.addPeer("bob") + discard await bus.addPeer("alice") + let bob = await bus.addPeer("bob") # Alice sends 5 messages while Bob is offline. let drops = @["bob".SdsParticipantID] - bus.broadcast("alice", @[byte(1)], "m1", dropAt = drops) - bus.broadcast("alice", @[byte(2)], "m2", dropAt = drops) - bus.broadcast("alice", @[byte(3)], "m3", dropAt = drops) - bus.broadcast("alice", @[byte(4)], "m4", dropAt = drops) - bus.broadcast("alice", @[byte(5)], "m5", dropAt = drops) + await bus.broadcast("alice", @[byte(1)], "m1", dropAt = drops) + await bus.broadcast("alice", @[byte(2)], "m2", dropAt = drops) + await bus.broadcast("alice", @[byte(3)], "m3", dropAt = drops) + await bus.broadcast("alice", @[byte(4)], "m4", dropAt = drops) + await bus.broadcast("alice", @[byte(5)], "m5", dropAt = drops) # Bob comes online and receives M6 which depends on m1..m5. - bus.broadcast("alice", @[byte(6)], "m6") + await bus.broadcast("alice", @[byte(6)], "m6") # Bob should have 5 outgoing repair entries. let channel = bob.channels[testChannel] @@ -1910,22 +1940,22 @@ suite "SDS-R: Multi-Participant Integration": for id in ["m1", "m2", "m3", "m4", "m5"]: bob.forceOutgoingExpired(id) - let wrapped = bob.wrapOutgoingMessage(@[byte(99)], "bob-msg-1", testChannel).get() + let wrapped = (await bob.wrapOutgoingMessage(@[byte(99)], "bob-msg-1", testChannel)).get() let decoded = deserializeMessage(wrapped).get() check decoded.repairRequest.len <= bob.config.maxRepairRequests # The attached entries should be removed from the outgoing buffer. check channel.outgoingRepairBuffer.len == 5 - decoded.repairRequest.len - test "markDependenciesMet externally clears pending repair entry": + asyncTest "markDependenciesMet externally clears pending repair entry": let bus = newTestBus() - discard bus.addPeer("alice") - let bob = bus.addPeer("bob") + discard await bus.addPeer("alice") + let bob = await bus.addPeer("bob") - bus.broadcast("alice", @[byte(1)], "m1", dropAt = @["bob".SdsParticipantID]) - bus.broadcast("alice", @[byte(2)], "m2") + await bus.broadcast("alice", @[byte(1)], "m1", dropAt = @["bob".SdsParticipantID]) + await bus.broadcast("alice", @[byte(2)], "m2") check "m1" in bob.channels[testChannel].outgoingRepairBuffer # Simulate Bob fetching M1 via an out-of-band store query. - check bob.markDependenciesMet(@["m1"], testChannel).isOk() + check (await bob.markDependenciesMet(@["m1"], testChannel)).isOk() check "m1" notin bob.channels[testChannel].outgoingRepairBuffer