feat: upgrade testbench and claude.md file for better context

Author: darshankabariya

sds.nim

@@ -133,6 +133,7 @@ proc wrapOutgoingMessage*(
         channelId: channelId,
         content: message,
         bloomFilter: bfResult.get(),
+        senderId: rm.participantId,
         repairRequest: repairReqs,
       )
 
@@ -144,7 +145,16 @@ proc wrapOutgoingMessage*(
       channel.bloomFilter.add(msg.messageId)
       rm.addToHistory(msg.messageId, channelId)
 
-      return serializeMessage(msg)
+      # SDS-R: record sender for future causal-history entries
+      if rm.participantId.len > 0:
+        channel.messageSenders[msg.messageId] = rm.participantId
+
+      let serialized = serializeMessage(msg)
+      if serialized.isOk():
+        # SDS-R: cache serialized bytes so we can serve our own message on repair
+        if channel.messageCache.len < rm.config.maxMessageHistory:
+          channel.messageCache[msg.messageId] = serialized.get()
+      return serialized
     except Exception:
       error "Failed to wrap message",
         channelId = channelId, msg = getCurrentExceptionMsg()
@@ -213,6 +223,11 @@ proc unwrapReceivedMessage*(
 
     let channel = 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)
+    channel.incomingRepairBuffer.del(msg.messageId)
+
     if msg.messageId in channel.messageHistory:
       return ok((msg.content, @[], channelId))
 
@@ -222,14 +237,14 @@ proc unwrapReceivedMessage*(
     # Review ACK status for outgoing messages
     rm.reviewAckStatus(msg)
 
-    # SDS-R: remove this message from repair buffers (dependency now met)
-    channel.outgoingRepairBuffer.del(msg.messageId)
-    channel.incomingRepairBuffer.del(msg.messageId)
-
     # SDS-R: cache the raw message for potential repair responses
     if channel.messageCache.len < rm.config.maxMessageHistory:
       channel.messageCache[msg.messageId] = message
 
+    # SDS-R: record sender so our future causal-history entries carry it
+    if msg.senderId.len > 0:
+      channel.messageSenders[msg.messageId] = msg.senderId
+
     # SDS-R: process incoming repair requests from this message
     let now = getTime()
     for repairEntry in msg.repairRequest:
@@ -267,6 +282,10 @@ proc unwrapReceivedMessage*(
       else:
         # All dependencies met, add to history
         rm.addToHistory(msg.messageId, channelId)
+        # Unblock any buffered messages that were waiting on this one.
+        for pendingId, entry in channel.incomingBuffer:
+          if msg.messageId in entry.missingDeps:
+            channel.incomingBuffer[pendingId].missingDeps.excl(msg.messageId)
         rm.processIncomingBuffer(channelId)
         if not rm.onMessageReady.isNil():
           rm.onMessageReady(msg.messageId, channelId)
@@ -415,43 +434,47 @@ proc periodicSyncMessage(
       error "Error in periodic sync", msg = getCurrentExceptionMsg()
     await sleepAsync(chronos.seconds(rm.config.syncMessageInterval.inSeconds))
 
+proc runRepairSweep*(rm: ReliabilityManager) {.gcsafe, 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
+  ## Exposed so it can be driven directly in tests; also invoked by periodicRepairSweep.
+  try:
+    let now = getTime()
+    for channelId, channel in rm.channels:
+      try:
+        # Check incoming repair buffer for expired T_resp (time to rebroadcast)
+        var toRebroadcast: seq[SdsMessageID] = @[]
+        for msgId, entry in channel.incomingRepairBuffer:
+          if now >= entry.tResp:
+            toRebroadcast.add(msgId)
+
+        for msgId in toRebroadcast:
+          let entry = channel.incomingRepairBuffer[msgId]
+          channel.incomingRepairBuffer.del(msgId)
+          if not rm.onRepairReady.isNil():
+            rm.onRepairReady(entry.cachedMessage, channelId)
+
+        # Drop expired outgoing repair entries past T_max
+        var toRemove: seq[SdsMessageID] = @[]
+        let tMaxDuration = rm.config.repairTMax
+        for msgId, entry in channel.outgoingRepairBuffer:
+          if now - entry.tReq > tMaxDuration:
+            toRemove.add(msgId)
+        for msgId in toRemove:
+          channel.outgoingRepairBuffer.del(msgId)
+      except Exception:
+        error "Error in repair sweep for channel",
+          channelId = channelId, msg = getCurrentExceptionMsg()
+  except Exception:
+    error "Error in repair sweep", msg = getCurrentExceptionMsg()
+
 proc periodicRepairSweep(
     rm: ReliabilityManager
 ) {.async: (raises: [CancelledError]), gcsafe.} =
   ## SDS-R: Periodically checks repair buffers for expired entries.
-  ## - Incoming: fires onRepairReady for expired T_resp entries
-  ## - Outgoing: drops entries past T_max
   while true:
-    try:
-      let now = getTime()
-      for channelId, channel in rm.channels:
-        try:
-          # Check incoming repair buffer for expired T_resp (time to rebroadcast)
-          var toRebroadcast: seq[SdsMessageID] = @[]
-          for msgId, entry in channel.incomingRepairBuffer:
-            if now >= entry.tResp:
-              toRebroadcast.add(msgId)
-
-          for msgId in toRebroadcast:
-            let entry = channel.incomingRepairBuffer[msgId]
-            channel.incomingRepairBuffer.del(msgId)
-            if not rm.onRepairReady.isNil():
-              rm.onRepairReady(entry.cachedMessage, channelId)
-
-          # Drop expired outgoing repair entries past T_max
-          var toRemove: seq[SdsMessageID] = @[]
-          let tMaxDuration = rm.config.repairTMax
-          for msgId, entry in channel.outgoingRepairBuffer:
-            if now - entry.tReq > tMaxDuration:
-              toRemove.add(msgId)
-          for msgId in toRemove:
-            channel.outgoingRepairBuffer.del(msgId)
-        except Exception:
-          error "Error in repair sweep for channel",
-            channelId = channelId, msg = getCurrentExceptionMsg()
-    except Exception:
-      error "Error in periodic repair sweep", msg = getCurrentExceptionMsg()
-
+    rm.runRepairSweep()
     await sleepAsync(chronos.milliseconds(rm.config.repairSweepInterval.inMilliseconds))
 
 proc startPeriodicTasks*(rm: ReliabilityManager) =
@@ -476,6 +499,7 @@ proc resetReliabilityManager*(rm: ReliabilityManager): Result[void, ReliabilityE
         channel.outgoingRepairBuffer.clear()
         channel.incomingRepairBuffer.clear()
         channel.messageCache.clear()
+        channel.messageSenders.clear()
         channel.bloomFilter = newRollingBloomFilter(
           rm.config.bloomFilterCapacity, rm.config.bloomFilterErrorRate
         )

sds/message.nim

@@ -17,6 +17,7 @@ type
     channelId*: SdsChannelID
     content*: seq[byte]
     bloomFilter*: seq[byte]
+    senderId*: SdsParticipantID  ## SDS-R: original sender's participant ID
     repairRequest*: seq[HistoryEntry]  ## Capped list of missing entries requesting repair (SDS-R)
 
   UnacknowledgedMessage* = object

sds/protobuf.nim

@@ -34,6 +34,9 @@ proc encode*(msg: SdsMessage): ProtoBuffer =
   pb.write(5, msg.content)
   pb.write(6, msg.bloomFilter)
 
+  if msg.senderId.len > 0:
+    pb.write(7, msg.senderId)
+
   for entry in msg.repairRequest:
     let entryPb = encodeHistoryEntry(entry)
     pb.write(13, entryPb.buffer)
@@ -78,6 +81,9 @@ proc decode*(T: type SdsMessage, buffer: seq[byte]): ProtobufResult[T] =
   if not ?pb.getField(6, msg.bloomFilter):
     msg.bloomFilter = @[] # Empty if not present
 
+  # SDS-R: decode senderId (field 7, optional)
+  discard pb.getField(7, msg.senderId)
+
   # SDS-R: decode repair request (field 13, optional)
   var repairBuffers: seq[seq[byte]]
   if pb.getRepeatedField(13, repairBuffers).isOk():

sds/sds_utils.nim

@@ -54,6 +54,7 @@ type
     outgoingRepairBuffer*: Table[SdsMessageID, OutgoingRepairEntry]
     incomingRepairBuffer*: Table[SdsMessageID, IncomingRepairEntry]
     messageCache*: Table[SdsMessageID, seq[byte]]  ## Cached serialized messages for repair responses
+    messageSenders*: Table[SdsMessageID, SdsParticipantID]  ## SDS-R: msgId -> original sender
 
   ReliabilityManager* = ref object
     channels*: Table[SdsChannelID, ChannelContext]
@@ -109,6 +110,7 @@ proc cleanup*(rm: ReliabilityManager) {.raises: [].} =
           channel.outgoingRepairBuffer.clear()
           channel.incomingRepairBuffer.clear()
           channel.messageCache.clear()
+          channel.messageSenders.clear()
         rm.channels.clear()
     except Exception:
       error "Error during cleanup", error = getCurrentExceptionMsg()
@@ -217,19 +219,20 @@ proc getRecentHistoryEntries*(
     rm: ReliabilityManager, n: int, channelId: SdsChannelID
 ): seq[HistoryEntry] =
   ## Get recent history entries for sending in causal history.
-  ## Populates retrieval hints for our own messages using the provider callback.
+  ## Populates retrieval hints and senderId (SDS-R) for each entry.
   try:
     if channelId in rm.channels:
       let channel = rm.channels[channelId]
       let recentMessageIds = channel.messageHistory[max(0, channel.messageHistory.len - n) .. ^1]
-      if rm.onRetrievalHint.isNil():
-        return toCausalHistory(recentMessageIds)
-      else:
-        var entries: seq[HistoryEntry] = @[]
-        for msgId in recentMessageIds:
-          let hint = rm.onRetrievalHint(msgId)
-          entries.add(newHistoryEntry(msgId, hint))
-        return entries
+      var entries: seq[HistoryEntry] = @[]
+      for msgId in recentMessageIds:
+        var entry = HistoryEntry(messageId: msgId)
+        if not rm.onRetrievalHint.isNil():
+          entry.retrievalHint = rm.onRetrievalHint(msgId)
+        if msgId in channel.messageSenders:
+          entry.senderId = channel.messageSenders[msgId]
+        entries.add(entry)
+      return entries
     else:
       return @[]
   except Exception:
@@ -315,6 +318,7 @@ proc getOrCreateChannel*(
         outgoingRepairBuffer: initTable[SdsMessageID, OutgoingRepairEntry](),
         incomingRepairBuffer: initTable[SdsMessageID, IncomingRepairEntry](),
         messageCache: initTable[SdsMessageID, seq[byte]](),
+        messageSenders: initTable[SdsMessageID, SdsParticipantID](),
       )
     result = rm.channels[channelId]
   except Exception:
@@ -347,6 +351,7 @@ proc removeChannel*(
         channel.outgoingRepairBuffer.clear()
         channel.incomingRepairBuffer.clear()
         channel.messageCache.clear()
+        channel.messageSenders.clear()
         rm.channels.del(channelId)
       return ok()
     except Exception: