fix: report the wire sender's id in channel_message_received (#4073)

logos_delivery/channels/reliable_channel.nim

@@ -246,9 +246,9 @@ proc send*(
 
   return ok(channelReqId)
 
-proc reportReceived(self: ReliableChannel, content: seq[byte]) =
+proc reportReceived(self: ReliableChannel, deliverable: SdsDeliverable) =
   ## Tail of the ingress pipeline (reassemble -> emit).
-  let reassembled = self.segmentation.handleIncomingSegment(content)
+  let reassembled = self.segmentation.handleIncomingSegment(deliverable.content)
   if reassembled.isSome():
     ## Emit on the captured `brokerCtx` (the manager's), so the
     ## application listener that the manager has set up on that same
@@ -257,7 +257,7 @@ proc reportReceived(self: ReliableChannel, content: seq[byte]) =
       self.brokerCtx,
       ChannelMessageReceivedEvent(
         channelId: self.channelId,
-        senderId: self.senderId,
+        senderId: deliverable.senderId,
         payload: reassembled.get().payload,
       ),
     )
@@ -323,8 +323,8 @@ proc onMessageReceived(
       ),
     )
     return
-  for content in deliverable:
-    self.reportReceived(content)
+  for item in deliverable:
+    self.reportReceived(item)
 
 proc new*(
     T: type ReliableChannel,

logos_delivery/channels/scalable_data_sync/scalable_data_sync.nim

@@ -44,12 +44,17 @@ type
   RebroadcastHandler* = proc(wire: seq[byte]) {.gcsafe, raises: [].}
     ## Invoked with a full SDS envelope to rebroadcast (SDS-R repair).
 
+  SdsDeliverable* = object
+    ## Deliverable segment tagged with the sender id from the SDS envelope.
+    content*: seq[byte]
+    senderId*: SdsParticipantID
+
   SdsHandler* = ref object
     reliabilityManager: ReliabilityManager
     channelId: SdsChannelID
-    pendingContent: OrderedTable[SdsMessageID, seq[byte]]
+    pendingContent: OrderedTable[SdsMessageID, SdsDeliverable]
       ## Segments parked until their causal dependencies arrive.
-    released: seq[seq[byte]]
+    released: seq[SdsDeliverable]
       ## Parked segments released by the unwrap currently in flight;
       ## filled via `onMessageReady`, drained by `handleIncoming`.
     ingressLock: AsyncLock
@@ -128,7 +133,7 @@ proc new*(
   let handler = T(
     reliabilityManager: rm,
     channelId: channelId,
-    pendingContent: initOrderedTable[SdsMessageID, seq[byte]](),
+    pendingContent: initOrderedTable[SdsMessageID, SdsDeliverable](),
     released: @[],
     ingressLock: newAsyncLock(),
     participantId: participantId,
@@ -162,8 +167,8 @@ proc wrapOutgoing*(
 
 proc handleIncoming*(
     self: SdsHandler, wire: seq[byte]
-): Future[Result[seq[seq[byte]], string]] {.async: (raises: []).} =
-  ## Returns the payloads deliverable now, in causal order. Empty when SDS
+): Future[Result[seq[SdsDeliverable], string]] {.async: (raises: []).} =
+  ## Returns the deliverables ready now, in causal order. Empty when SDS
   ## consumed the message; `err` when the bytes are not an SDS envelope.
   let msg = deserializeMessage(wire).valueOr:
     return err("SDS deserialization failed: " & $error)
@@ -173,7 +178,7 @@ proc handleIncoming*(
   if msg.channelId != self.channelId:
     debug "dropping SDS message for foreign channel",
       channelId = self.channelId, wireChannelId = msg.channelId
-    return ok(newSeq[seq[byte]]())
+    return ok(newSeq[SdsDeliverable]())
 
   ## Only the lock acquisition can raise (CancelledError); the unwrap work
   ## below is `raises: []`, so the try stays scoped to exactly the acquire.
@@ -184,7 +189,7 @@ proc handleIncoming*(
 
   ## Funnel every unwrap outcome into `res` so the lock is released once on
   ## the tail path, where `releaseIngressLock` can surface its own error.
-  var res: Result[seq[seq[byte]], string]
+  var res: Result[seq[SdsDeliverable], string]
   block ingress:
     ## Load persisted state before the duplicate check, so a replay right
     ## after a restart is not re-delivered. Idempotent, cheap once loaded.
@@ -203,7 +208,7 @@ proc handleIncoming*(
       break ingress
 
     if isDuplicate:
-      res = ok(newSeq[seq[byte]]())
+      res = ok(newSeq[SdsDeliverable]())
       break ingress
 
     if unwrapped.missingDeps.len > 0:
@@ -215,14 +220,17 @@ proc handleIncoming*(
         self.pendingContent.del(oldest)
         warn "SDS pending-content stash full, dropping oldest entry",
           channelId = self.channelId, dropped = oldest
-      self.pendingContent[msg.messageId] = unwrapped.message
-      res = ok(newSeq[seq[byte]]())
+      self.pendingContent[msg.messageId] =
+        SdsDeliverable(content: unwrapped.message, senderId: msg.senderId)
+      res = ok(newSeq[SdsDeliverable]())
       break ingress
 
-    var deliverable = newSeq[seq[byte]]()
+    var deliverable = newSeq[SdsDeliverable]()
     if unwrapped.message.len > 0:
       ## Empty content is sync traffic: causal metadata only.
-      deliverable.add(unwrapped.message)
+      deliverable.add(
+        SdsDeliverable(content: unwrapped.message, senderId: msg.senderId)
+      )
     deliverable.add(self.released)
     self.released.setLen(0)
     res = ok(deliverable)

tests/channels/test_reliable_channel_send_receive.nim

@@ -71,13 +71,13 @@ suite "Reliable Channel - ingress":
       .createReliableChannel(channelId, contentTopic, SdsParticipantID("local"))
       .expect("createReliableChannel")
 
-    let received = newFuture[seq[byte]]("channel-message-received")
+    let received = newFuture[ChannelMessageReceivedEvent]("channel-message-received")
     discard ChannelMessageReceivedEvent
       .listen(
         brokerCtx,
         proc(evt: ChannelMessageReceivedEvent) {.async: (raises: []).} =
           if not received.finished() and evt.channelId == channelId:
-            received.complete(evt.payload)
+            received.complete(evt)
         ,
       )
       .expect("listen ChannelMessageReceivedEvent")
@@ -108,7 +108,10 @@ suite "Reliable Channel - ingress":
     let arrived = await received.withTimeout(TestTimeout)
     check arrived
     if arrived:
-      check received.read() == appPayload
+      let evt = received.read()
+      check evt.payload == appPayload
+      ## wire sender's id, not the local participant id
+      check evt.senderId == SdsParticipantID("remote")
 
     (await waku.stop()).expect("stop")