diff --git a/sds.nim b/sds.nim index baf1c7c..9935fff 100644 --- a/sds.nim +++ b/sds.nim @@ -87,9 +87,7 @@ proc wrapOutgoingMessage*( message: seq[byte], messageId: SdsMessageID, channelId: SdsChannelID, -): Future[Result[seq[byte], ReliabilityError]] {. - async: (raises: [CancelledError]), gcsafe -.} = +): Future[Result[seq[byte], ReliabilityError]] {.async: (raises: []), gcsafe.} = ## Wraps an outgoing message with reliability metadata. if message.len == 0: return err(ReliabilityError.reInvalidArgument) @@ -160,8 +158,6 @@ proc wrapOutgoingMessage*( return err(ReliabilityError.reSerializationError) finally: rm.lock.release() - except CancelledError as e: - raise e except CatchableError: error "Failed to wrap message (lock)", channelId = channelId, msg = getCurrentExceptionMsg() @@ -227,7 +223,7 @@ proc unwrapReceivedMessage*( tuple[message: seq[byte], missingDeps: seq[HistoryEntry], channelId: SdsChannelID], ReliabilityError, ] -] {.async: (raises: [CancelledError]).} = +] {.async: (raises: []).} = ## Unwraps a received message and processes its reliability metadata. try: let channelId = extractChannelId(message).valueOr: @@ -338,15 +334,13 @@ proc unwrapReceivedMessage*( await rm.persistence.saveOutgoingRepair(channelId, dep.messageId, outEntry) return ok((msg.content, missingDeps, channelId)) - except CancelledError as e: - raise e except CatchableError: error "Failed to unwrap message", msg = getCurrentExceptionMsg() return err(ReliabilityError.reDeserializationError) proc markDependenciesMet*( rm: ReliabilityManager, messageIds: seq[SdsMessageID], channelId: SdsChannelID -): Future[Result[void, ReliabilityError]] {.async: (raises: [CancelledError]).} = +): Future[Result[void, ReliabilityError]] {.async: (raises: []).} = ## Marks the specified message dependencies as met. try: if channelId notin rm.channels: @@ -374,8 +368,6 @@ proc markDependenciesMet*( await rm.processIncomingBuffer(channelId) return ok() - except CancelledError as e: - raise e except CatchableError: error "Failed to mark dependencies as met", channelId = channelId, msg = getCurrentExceptionMsg() @@ -389,7 +381,7 @@ proc setCallbacks*( onPeriodicSync: PeriodicSyncCallback = nil, onRetrievalHint: RetrievalHintProvider = nil, onRepairReady: RepairReadyCallback = nil, -) {.async: (raises: [CancelledError]).} = +) {.async: (raises: []).} = ## Sets the callback functions for various events in the ReliabilityManager. try: await rm.lock.acquire() @@ -402,14 +394,12 @@ proc setCallbacks*( rm.onRepairReady = onRepairReady finally: rm.lock.release() - except CancelledError as e: - raise e except CatchableError: error "Failed to set callbacks", msg = getCurrentExceptionMsg() proc checkUnacknowledgedMessages( rm: ReliabilityManager, channelId: SdsChannelID -) {.async: (raises: [CancelledError]).} = +) {.async: (raises: []).} = try: await rm.lock.acquire() try: @@ -441,8 +431,6 @@ proc checkUnacknowledgedMessages( channel.outgoingBuffer = newOutgoingBuffer finally: rm.lock.release() - except CancelledError as e: - raise e except CatchableError: error "Failed to check unacknowledged messages", channelId = channelId, msg = getCurrentExceptionMsg() @@ -453,11 +441,8 @@ proc periodicBufferSweep(rm: ReliabilityManager) {.async: (raises: [CancelledErr for channelId, channel in rm.channels: 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)) proc periodicSyncMessage(rm: ReliabilityManager) {.async: (raises: [CancelledError]).} = @@ -470,7 +455,7 @@ proc periodicSyncMessage(rm: ReliabilityManager) {.async: (raises: [CancelledErr error "Error in periodic sync", msg = getCurrentExceptionMsg() await sleepAsync(chronos.seconds(rm.config.syncMessageInterval.inSeconds)) -proc runRepairSweep*(rm: ReliabilityManager) {.async: (raises: [CancelledError]).} = +proc runRepairSweep*(rm: ReliabilityManager) {.async: (raises: []).} = ## 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 @@ -511,15 +496,16 @@ proc runRepairSweep*(rm: ReliabilityManager) {.async: (raises: [CancelledError]) channelId = channelId, msg = getCurrentExceptionMsg() finally: rm.lock.release() - except CancelledError as e: - raise e except CatchableError: error "Error in repair sweep", msg = getCurrentExceptionMsg() proc periodicRepairSweep(rm: ReliabilityManager) {.async: (raises: [CancelledError]).} = ## SDS-R: Periodically checks repair buffers for expired entries. while true: - await rm.runRepairSweep() + try: + await rm.runRepairSweep() + except CatchableError: + error "Error in periodic repair sweep", msg = getCurrentExceptionMsg() await sleepAsync(chronos.milliseconds(rm.config.repairSweepInterval.inMilliseconds)) proc startPeriodicTasks*(rm: ReliabilityManager) = @@ -533,7 +519,7 @@ proc startPeriodicTasks*(rm: ReliabilityManager) = proc resetReliabilityManager*( rm: ReliabilityManager -): Future[Result[void, ReliabilityError]] {.async: (raises: [CancelledError]).} = +): Future[Result[void, ReliabilityError]] {.async: (raises: []).} = ## Resets the ReliabilityManager to its initial state. try: await rm.lock.acquire() @@ -557,8 +543,6 @@ proc resetReliabilityManager*( return err(ReliabilityError.reInternalError) finally: rm.lock.release() - except CancelledError as e: - raise e 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 a8cfaaa..105102f 100644 --- a/sds/sds_utils.nim +++ b/sds/sds_utils.nim @@ -55,7 +55,7 @@ proc cleanup*(rm: ReliabilityManager) {.async: (raises: []).} = proc cleanBloomFilter*( rm: ReliabilityManager, channelId: SdsChannelID -) {.async: (raises: [CancelledError]).} = +) {.async: (raises: []).} = try: await rm.lock.acquire() try: @@ -63,8 +63,6 @@ proc cleanBloomFilter*( rm.channels[channelId].bloomFilter.clean() finally: rm.lock.release() - except CancelledError as e: - raise e except CatchableError: error "Failed to clean bloom filter", error = getCurrentExceptionMsg(), channelId = channelId @@ -217,7 +215,7 @@ proc checkDependencies*( proc getMessageHistory*( rm: ReliabilityManager, channelId: SdsChannelID -): Future[seq[SdsMessageID]] {.async: (raises: [CancelledError]).} = +): Future[seq[SdsMessageID]] {.async: (raises: []).} = try: await rm.lock.acquire() try: @@ -230,8 +228,6 @@ proc getMessageHistory*( return @[] finally: rm.lock.release() - except CancelledError as e: - raise e except CatchableError: error "Failed to get message history", channelId = channelId, error = getCurrentExceptionMsg() @@ -239,7 +235,7 @@ proc getMessageHistory*( proc getOutgoingBuffer*( rm: ReliabilityManager, channelId: SdsChannelID -): Future[seq[UnacknowledgedMessage]] {.async: (raises: [CancelledError]).} = +): Future[seq[UnacknowledgedMessage]] {.async: (raises: []).} = try: await rm.lock.acquire() try: @@ -249,8 +245,6 @@ proc getOutgoingBuffer*( return @[] finally: rm.lock.release() - except CancelledError as e: - raise e except CatchableError: error "Failed to get outgoing buffer", channelId = channelId, error = getCurrentExceptionMsg() @@ -258,9 +252,7 @@ proc getOutgoingBuffer*( proc getIncomingBuffer*( rm: ReliabilityManager, channelId: SdsChannelID -): Future[Table[SdsMessageID, IncomingMessage]] {. - async: (raises: [CancelledError]), gcsafe -.} = +): Future[Table[SdsMessageID, IncomingMessage]] {.async: (raises: []), gcsafe.} = try: await rm.lock.acquire() try: @@ -270,8 +262,6 @@ proc getIncomingBuffer*( return initTable[SdsMessageID, IncomingMessage]() finally: rm.lock.release() - except CancelledError as e: - raise e except CatchableError: error "Failed to get incoming buffer", channelId = channelId, error = getCurrentExceptionMsg() @@ -313,23 +303,19 @@ proc getOrCreateChannel*( proc ensureChannel*( rm: ReliabilityManager, channelId: SdsChannelID -): Future[Result[void, ReliabilityError]] {.async: (raises: [CancelledError]).} = +): Future[Result[void, ReliabilityError]] {.async: (raises: []).} = try: await rm.lock.acquire() try: try: discard await rm.getOrCreateChannel(channelId) return ok() - except CancelledError as e: - raise e except CatchableError: error "Failed to ensure channel", channelId = channelId, msg = getCurrentExceptionMsg() return err(ReliabilityError.reInternalError) finally: rm.lock.release() - except CancelledError as e: - raise e except CatchableError: error "Failed to ensure channel (lock)", channelId = channelId, msg = getCurrentExceptionMsg() @@ -337,7 +323,7 @@ proc ensureChannel*( proc removeChannel*( rm: ReliabilityManager, channelId: SdsChannelID -): Future[Result[void, ReliabilityError]] {.async: (raises: [CancelledError]).} = +): Future[Result[void, ReliabilityError]] {.async: (raises: []).} = try: await rm.lock.acquire() try: @@ -358,8 +344,6 @@ proc removeChannel*( return err(ReliabilityError.reInternalError) finally: rm.lock.release() - except CancelledError as e: - raise e except CatchableError: error "Failed to remove channel (lock)", channelId = channelId, msg = getCurrentExceptionMsg() diff --git a/tests/test_reliability.nim b/tests/test_reliability.nim index eee3bbf..9c14fe5 100644 --- a/tests/test_reliability.nim +++ b/tests/test_reliability.nim @@ -1783,7 +1783,7 @@ proc deliverExcept( senderId: SdsParticipantID, bytes: seq[byte], exclude: seq[SdsParticipantID], -) {.async: (raises: [CancelledError]).} = +) {.async: (raises: []).} = for pid, peer in bus.peers: if pid == senderId or pid in exclude: continue