Remove leaking CancelledFuture from public facing + as a consequence it is tuneled into handling CatchableError everywhere

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)

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()

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