Stylistics, async default implication addressed, nph style run

Author: NagyZoltanPeter

library/events/json_message_ready_event.nim

@@ -5,8 +5,12 @@ type JsonMessageReadyEvent* = ref object of JsonEvent
   messageId*: SdsMessageID
   channelId*: SdsChannelID
 
-proc new*(T: type JsonMessageReadyEvent, messageId: SdsMessageID, channelId: SdsChannelID): T =
-  return JsonMessageReadyEvent(eventType: "message_ready", messageId: messageId, channelId: channelId)
+proc new*(
+    T: type JsonMessageReadyEvent, messageId: SdsMessageID, channelId: SdsChannelID
+): T =
+  return JsonMessageReadyEvent(
+    eventType: "message_ready", messageId: messageId, channelId: channelId
+  )
 
 method `$`*(jsonMessageReady: JsonMessageReadyEvent): string =
   $(%*jsonMessageReady)

library/events/json_message_sent_event.nim

@@ -5,8 +5,12 @@ type JsonMessageSentEvent* = ref object of JsonEvent
   messageId*: SdsMessageID
   channelId*: SdsChannelID
 
-proc new*(T: type JsonMessageSentEvent, messageId: SdsMessageID, channelId: SdsChannelID): T =
-  return JsonMessageSentEvent(eventType: "message_sent", messageId: messageId, channelId: channelId)
+proc new*(
+    T: type JsonMessageSentEvent, messageId: SdsMessageID, channelId: SdsChannelID
+): T =
+  return JsonMessageSentEvent(
+    eventType: "message_sent", messageId: messageId, channelId: channelId
+  )
 
 method `$`*(jsonMessageSent: JsonMessageSentEvent): string =
   $(%*jsonMessageSent)

library/events/json_missing_dependencies_event.nim

@@ -13,7 +13,10 @@ proc new*(
     channelId: SdsChannelID,
 ): T =
   return JsonMissingDependenciesEvent(
-    eventType: "missing_dependencies", messageId: messageId, missingDeps: missingDeps, channelId: channelId
+    eventType: "missing_dependencies",
+    messageId: messageId,
+    missingDeps: missingDeps,
+    channelId: channelId,
   )
 
 method `$`*(jsonMissingDependencies: JsonMissingDependenciesEvent): string =

library/libsds.nim

@@ -63,7 +63,8 @@ var
 
 proc acquireCtx(callback: SdsCallBack, userData: pointer): ptr SdsContext =
   ctxPoolLock.acquire()
-  defer: ctxPoolLock.release()
+  defer:
+    ctxPoolLock.release()
   if ctxPool.len > 0:
     result = ctxPool.pop()
   else:
@@ -74,7 +75,8 @@ proc acquireCtx(callback: SdsCallBack, userData: pointer): ptr SdsContext =
 
 proc releaseCtx(ctx: ptr SdsContext) =
   ctxPoolLock.acquire()
-  defer: ctxPoolLock.release()
+  defer:
+    ctxPoolLock.release()
   ctx.userData = nil
   ctx.eventCallback = nil
   ctx.eventUserData = nil
@@ -105,7 +107,9 @@ proc onMessageSent(ctx: ptr SdsContext): MessageSentCallback =
       $JsonMessageSentEvent.new(messageId, channelId)
 
 proc onMissingDependencies(ctx: ptr SdsContext): MissingDependenciesCallback =
-  return proc(messageId: SdsMessageID, missingDeps: seq[HistoryEntry], channelId: SdsChannelID) {.gcsafe.} =
+  return proc(
+      messageId: SdsMessageID, missingDeps: seq[HistoryEntry], channelId: SdsChannelID
+  ) {.gcsafe.} =
     callEventCallback(ctx, "onMissingDependencies"):
       $JsonMissingDependenciesEvent.new(messageId, missingDeps, channelId)
 
@@ -135,7 +139,7 @@ proc onRetrievalHint(ctx: ptr SdsContext): RetrievalHintProvider =
       copyMem(addr hintBytes[0], hint, hintLen)
       deallocShared(hint)
       return hintBytes
-    
+
     return @[]
 
 ### End of not-exported components

library/sds_thread/inter_thread_communication/requests/sds_message_request.nim

@@ -69,7 +69,9 @@ proc process*(
     ).valueOr:
       return err("error processing UNWRAP_MESSAGE request: " & $error)
 
-    let res = SdsUnwrapResponse(message: unwrappedMessage, missingDeps: missingDeps, channelId: extractedChannelId)
+    let res = SdsUnwrapResponse(
+      message: unwrappedMessage, missingDeps: missingDeps, channelId: extractedChannelId
+    )
 
     # return the result as a json string
     var node = newJObject()

library/sds_thread/sds_thread.nim

@@ -4,10 +4,7 @@
 
 import std/[options, atomics, os, net, locks]
 import chronicles, chronos, chronos/threadsync, taskpools/channels_spsc_single, results
-import
-  ../ffi_types,
-  ./inter_thread_communication/sds_thread_request,
-  sds/sds_utils
+import ../ffi_types, ./inter_thread_communication/sds_thread_request, sds/sds_utils
 
 type SdsContext* = object
   thread: Thread[(ptr SdsContext)]

sds.nim

@@ -35,7 +35,7 @@ proc isAcknowledged*(
 
 proc reviewAckStatus(
     rm: ReliabilityManager, msg: SdsMessage
-): Future[void] {.async: (raises: [CatchableError]), gcsafe.} =
+) {.async: (raises: [CatchableError]).} =
   try:
     var rbf: Option[RollingBloomFilter]
     if msg.bloomFilter.len > 0:
@@ -46,8 +46,10 @@ proc reviewAckStatus(
           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,
+            minCapacity =
+              (bf.capacity.float * (100 - CapacityFlexPercent).float / 100.0).int,
+            maxCapacity =
+              (bf.capacity.float * (100 + CapacityFlexPercent).float / 100.0).int,
           )
         )
       else:
@@ -167,7 +169,7 @@ proc wrapOutgoingMessage*(
 
 proc processIncomingBuffer(
     rm: ReliabilityManager, channelId: SdsChannelID
-): Future[void] {.async: (raises: [CatchableError]), gcsafe.} =
+) {.async: (raises: [CatchableError]).} =
   try:
     await rm.lock.acquire()
     try:
@@ -220,10 +222,12 @@ proc processIncomingBuffer(
 
 proc unwrapReceivedMessage*(
     rm: ReliabilityManager, message: seq[byte]
-): Future[Result[
-    tuple[message: seq[byte], missingDeps: seq[HistoryEntry], channelId: SdsChannelID],
-    ReliabilityError,
-]] {.async: (raises: [CancelledError]), gcsafe.} =
+): Future[
+    Result[
+      tuple[message: seq[byte], missingDeps: seq[HistoryEntry], channelId: SdsChannelID],
+      ReliabilityError,
+    ]
+] {.async: (raises: [CancelledError]).} =
   ## Unwraps a received message and processes its reliability metadata.
   try:
     let channelId = extractChannelId(message).valueOr:
@@ -258,17 +262,18 @@ proc unwrapReceivedMessage*(
       # Remove from our own outgoing repair buffer (someone else is also requesting)
       channel.outgoingRepairBuffer.del(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 repairEntry.messageId in channel.messageHistory and rm.participantId.len > 0 and
+          repairEntry.senderId.len > 0:
         if isInResponseGroup(
-          rm.participantId, repairEntry.senderId,
-          repairEntry.messageId, rm.config.numResponseGroups
+          rm.participantId, repairEntry.senderId, repairEntry.messageId,
+          rm.config.numResponseGroups,
         ):
-          let serialized = serializeMessage(channel.messageHistory[repairEntry.messageId])
+          let serialized =
+            serializeMessage(channel.messageHistory[repairEntry.messageId])
           if serialized.isOk():
             let tResp = computeTResp(
-              rm.participantId, repairEntry.senderId,
-              repairEntry.messageId, rm.config.repairTMax
+              rm.participantId, repairEntry.senderId, repairEntry.messageId,
+              rm.config.repairTMax,
             )
             let inEntry = IncomingRepairEntry(
               inHistEntry: repairEntry,
@@ -302,15 +307,16 @@ proc unwrapReceivedMessage*(
             channel.incomingBuffer[pendingId].missingDeps.excl(msg.messageId)
             unblocked.add(pendingId)
         for pendingId in unblocked:
-          await rm.persistence.saveIncoming(channelId, channel.incomingBuffer[pendingId])
+          await rm.persistence.saveIncoming(
+            channelId, channel.incomingBuffer[pendingId]
+          )
         await rm.processIncomingBuffer(channelId)
         if not rm.onMessageReady.isNil():
           {.cast(raises: []).}:
             rm.onMessageReady(msg.messageId, channelId)
     else:
       let entry = IncomingMessage.init(
-        message = msg,
-        missingDeps = missingDeps.getMessageIds().toHashSet(),
+        message = msg, missingDeps = missingDeps.getMessageIds().toHashSet()
       )
       channel.incomingBuffer[msg.messageId] = entry
       await rm.persistence.saveIncoming(channelId, entry)
@@ -323,13 +329,11 @@ proc unwrapReceivedMessage*(
         for dep in missingDeps:
           if dep.messageId notin channel.outgoingRepairBuffer:
             let tReq = computeTReq(
-              rm.participantId, dep.messageId,
-              rm.config.repairTMin, rm.config.repairTMax
-            )
-            let outEntry = OutgoingRepairEntry(
-              outHistEntry: dep,
-              minTimeRepairReq: now + tReq,
+              rm.participantId, dep.messageId, rm.config.repairTMin,
+              rm.config.repairTMax,
             )
+            let outEntry =
+              OutgoingRepairEntry(outHistEntry: dep, minTimeRepairReq: now + tReq)
             channel.outgoingRepairBuffer[dep.messageId] = outEntry
             await rm.persistence.saveOutgoingRepair(channelId, dep.messageId, outEntry)
 
@@ -342,7 +346,7 @@ proc unwrapReceivedMessage*(
 
 proc markDependenciesMet*(
     rm: ReliabilityManager, messageIds: seq[SdsMessageID], channelId: SdsChannelID
-): Future[Result[void, ReliabilityError]] {.async: (raises: [CancelledError]), gcsafe.} =
+): Future[Result[void, ReliabilityError]] {.async: (raises: [CancelledError]).} =
   ## Marks the specified message dependencies as met.
   try:
     if channelId notin rm.channels:
@@ -385,7 +389,7 @@ proc setCallbacks*(
     onPeriodicSync: PeriodicSyncCallback = nil,
     onRetrievalHint: RetrievalHintProvider = nil,
     onRepairReady: RepairReadyCallback = nil,
-): Future[void] {.async: (raises: [CancelledError]), gcsafe.} =
+) {.async: (raises: [CancelledError]).} =
   ## Sets the callback functions for various events in the ReliabilityManager.
   try:
     await rm.lock.acquire()
@@ -405,7 +409,7 @@ proc setCallbacks*(
 
 proc checkUnacknowledgedMessages(
     rm: ReliabilityManager, channelId: SdsChannelID
-): Future[void] {.async: (raises: [CancelledError]), gcsafe.} =
+) {.async: (raises: [CancelledError]).} =
   try:
     await rm.lock.acquire()
     try:
@@ -443,9 +447,7 @@ proc checkUnacknowledgedMessages(
     error "Failed to check unacknowledged messages",
       channelId = channelId, msg = getCurrentExceptionMsg()
 
-proc periodicBufferSweep(
-    rm: ReliabilityManager
-) {.async: (raises: [CancelledError]), gcsafe.} =
+proc periodicBufferSweep(rm: ReliabilityManager) {.async: (raises: [CancelledError]).} =
   while true:
     try:
       for channelId, channel in rm.channels:
@@ -458,9 +460,7 @@ proc periodicBufferSweep(
 
     await sleepAsync(chronos.milliseconds(rm.config.bufferSweepInterval.inMilliseconds))
 
-proc periodicSyncMessage(
-    rm: ReliabilityManager
-) {.async: (raises: [CancelledError]), gcsafe.} =
+proc periodicSyncMessage(rm: ReliabilityManager) {.async: (raises: [CancelledError]).} =
   while true:
     try:
       if not rm.onPeriodicSync.isNil():
@@ -470,9 +470,7 @@ proc periodicSyncMessage(
       error "Error in periodic sync", msg = getCurrentExceptionMsg()
     await sleepAsync(chronos.seconds(rm.config.syncMessageInterval.inSeconds))
 
-proc runRepairSweep*(
-    rm: ReliabilityManager
-): Future[void] {.async: (raises: [CancelledError]), gcsafe.} =
+proc runRepairSweep*(rm: ReliabilityManager) {.async: (raises: [CancelledError]).} =
   ## 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
@@ -518,9 +516,7 @@ proc runRepairSweep*(
   except CatchableError:
     error "Error in repair sweep", msg = getCurrentExceptionMsg()
 
-proc periodicRepairSweep(
-    rm: ReliabilityManager
-) {.async: (raises: [CancelledError]), gcsafe.} =
+proc periodicRepairSweep(rm: ReliabilityManager) {.async: (raises: [CancelledError]).} =
   ## SDS-R: Periodically checks repair buffers for expired entries.
   while true:
     await rm.runRepairSweep()
@@ -537,7 +533,7 @@ proc startPeriodicTasks*(rm: ReliabilityManager) =
 
 proc resetReliabilityManager*(
     rm: ReliabilityManager
-): Future[Result[void, ReliabilityError]] {.async: (raises: [CancelledError]), gcsafe.} =
+): Future[Result[void, ReliabilityError]] {.async: (raises: [CancelledError]).} =
   ## Resets the ReliabilityManager to its initial state.
   try:
     await rm.lock.acquire()

sds/bloom.nim

@@ -80,7 +80,8 @@ proc initializeBloomFilter*(
 
 proc `$`*(bf: BloomFilter): string =
   ## Prints the configuration of the Bloom filter.
-  return "Bloom filter with $1 capacity, $2 error rate, $3 hash functions, and requiring $4 bits of memory." %
+  return
+    "Bloom filter with $1 capacity, $2 error rate, $3 hash functions, and requiring $4 bits of memory." %
     [
       $bf.capacity,
       formatFloat(bf.errorRate, format = ffScientific, precision = 1),

sds/message.nim

@@ -7,10 +7,5 @@ import ./types/repair_entry
 import ./types/reliability_config
 
 export
-  sds_message_id,
-  history_entry,
-  sds_message,
-  unacknowledged_message,
-  incoming_message,
-  repair_entry,
-  reliability_config
+  sds_message_id, history_entry, sds_message, unacknowledged_message, incoming_message,
+  repair_entry, reliability_config