From b5f624f519d377dbf31f98c80281e57d209b31b2 Mon Sep 17 00:00:00 2001 From: Darshan <35736874+darshankabariya@users.noreply.github.com> Date: Tue, 28 Jul 2026 23:34:49 +0530 Subject: [PATCH] fix: report the wire sender's id in channel_message_received (#4073) --- logos_delivery/channels/reliable_channel.nim | 10 +++--- .../scalable_data_sync/scalable_data_sync.nim | 32 ++++++++++++------- .../test_reliable_channel_send_receive.nim | 9 ++++-- 3 files changed, 31 insertions(+), 20 deletions(-) diff --git a/logos_delivery/channels/reliable_channel.nim b/logos_delivery/channels/reliable_channel.nim index 0286e8f27..1a4116380 100644 --- a/logos_delivery/channels/reliable_channel.nim +++ b/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, diff --git a/logos_delivery/channels/scalable_data_sync/scalable_data_sync.nim b/logos_delivery/channels/scalable_data_sync/scalable_data_sync.nim index 81e1f6ac9..985ec2fde 100644 --- a/logos_delivery/channels/scalable_data_sync/scalable_data_sync.nim +++ b/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) diff --git a/tests/channels/test_reliable_channel_send_receive.nim b/tests/channels/test_reliable_channel_send_receive.nim index 8a7b02231..0ffbe7fed 100644 --- a/tests/channels/test_reliable_channel_send_receive.nim +++ b/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")