From ed060553b0c8b8d25aa32bb9781ee5b7055ff41b Mon Sep 17 00:00:00 2001 From: darshankabariya Date: Fri, 17 Jul 2026 16:17:04 +0530 Subject: [PATCH] feat: propagation-based send confirmation mode New messaging option send-confirmation = store (default) | propagation. In propagation mode MessageSent fires from the publish path (relay mesh / lightpush relayPeerCount) right after MessagePropagated, with no store polling; finalized tasks no longer re-enter the task cache, so the confirmation is emitted exactly once. Store mode is bit-identical to before. The option flows through both the structured and legacy flat JSON config shapes. Part of the "Store as a startup-only dependency" experiment (step 9). Co-Authored-By: Claude Fable 5 --- .../api/conf/logos_delivery_conf_json.nim | 6 ++- logos_delivery/api/conf/messaging_conf.nim | 3 ++ .../send_service/send_service.nim | 27 +++++++++-- logos_delivery/messaging/messaging_client.nim | 13 ++++- tests/api/test_api_send.nim | 47 +++++++++++++++++++ 5 files changed, 91 insertions(+), 5 deletions(-) diff --git a/logos_delivery/api/conf/logos_delivery_conf_json.nim b/logos_delivery/api/conf/logos_delivery_conf_json.nim index 6440a63ac..97abb83f3 100644 --- a/logos_delivery/api/conf/logos_delivery_conf_json.nim +++ b/logos_delivery/api/conf/logos_delivery_conf_json.nim @@ -17,6 +17,8 @@ const # a flat blob before the WakuNodeConf walker sees them (it would reject them). KeyReliabilityEnabled = "reliabilityenabled" KeyReliability = "reliability" + KeySendConfirmation = "sendconfirmation" + KeySendConfirmationName = "send-confirmation" proc parseMode(s: string): Result[LogosDeliveryMode, string] = case s.strip().toLowerAscii() @@ -52,7 +54,9 @@ proc parseFlatConf( ## `WakuNodeConf`. Full stack. Delete this proc and its call site to drop support. var messaging = MessagingClientConf() var reliabilityFields: Table[string, (string, JsonNode)] - for key in [KeyReliabilityEnabled, KeyReliability]: + for key in [ + KeyReliabilityEnabled, KeyReliability, KeySendConfirmation, KeySendConfirmationName + ]: if topJsonNode.hasKey(key): reliabilityFields[key] = topJsonNode.getOrDefault(key) topJsonNode.del(key) diff --git a/logos_delivery/api/conf/messaging_conf.nim b/logos_delivery/api/conf/messaging_conf.nim index a8c91acd7..9d06b953e 100644 --- a/logos_delivery/api/conf/messaging_conf.nim +++ b/logos_delivery/api/conf/messaging_conf.nim @@ -38,6 +38,9 @@ type MessagingClientConf* = object ## RLN epoch size, in seconds. reliabilityEnabled* {.name: "reliability".}: Opt[bool] ## Enable store-based send reliability. + sendConfirmation* {.name: "send-confirmation".}: Opt[string] + ## How MessageSent is confirmed: "store" (store witness, default) or + ## "propagation" (publish-path peer count; no store dependency). store*: Opt[bool] ## Enable the store protocol. storenode* {.name: "storenode".}: Opt[string] storeMessageDbUrl* {.name: "store-message-db-url".}: Opt[string] diff --git a/logos_delivery/messaging/delivery_service/send_service/send_service.nim b/logos_delivery/messaging/delivery_service/send_service/send_service.nim index 097bb0752..99ffeb7ea 100644 --- a/logos_delivery/messaging/delivery_service/send_service/send_service.nim +++ b/logos_delivery/messaging/delivery_service/send_service/send_service.nim @@ -38,6 +38,10 @@ const ArchiveTime = chronos.seconds(3) ## Estimation of the time we wait until we start confirming that a message has been properly ## received and archived by a store node +type SendConfirmationMode* {.pure.} = enum + Store ## MessageSent fires when a store node confirms the message (default) + Propagation ## MessageSent fires from the publish path (relay/lightpush peer count) + type SendService* = ref object of RootObj brokerCtx: BrokerContext taskCache: seq[DeliveryTask] @@ -48,6 +52,7 @@ type SendService* = ref object of RootObj sendProcessor: BaseSendProcessor waku: Waku + confirmationMode: SendConfirmationMode checkStoreForMessages: bool lastStoreCheckTime: Moment ## throttles store validation queries to ArchiveTime cadence @@ -77,14 +82,19 @@ proc setupSendProcessorChain( return ok(processors[0]) proc new*( - T: typedesc[SendService], preferP2PReliability: bool, waku: Waku + T: typedesc[SendService], + preferP2PReliability: bool, + waku: Waku, + confirmationMode = SendConfirmationMode.Store, ): Result[T, string] = if not waku.hasRelay() and not waku.hasLightpush(): return err( "Could not create SendService. wakuRelay or wakuLightpushClient should be set" ) - let checkStoreForMessages = preferP2PReliability and waku.isStoreMounted() + let checkStoreForMessages = + confirmationMode == SendConfirmationMode.Store and preferP2PReliability and + waku.isStoreMounted() let sendProcessorChain = setupSendProcessorChain(waku, waku.brokerCtx).valueOr: return err("failed to setup SendProcessorChain: " & $error) @@ -95,6 +105,7 @@ proc new*( serviceLoopHandle: nil, sendProcessor: sendProcessorChain, waku: waku, + confirmationMode: confirmationMode, checkStoreForMessages: checkStoreForMessages, lastStoreCheckTime: Moment.now(), ) @@ -170,6 +181,13 @@ proc reportTaskResult(self: SendService, task: DeliveryTask) = self.brokerCtx, task.requestId, task.msgHash.to0xHex() ) task.propagateEventEmitted = true + if self.confirmationMode == SendConfirmationMode.Propagation: + # The publish path is the confirmation signal: report sent right after + # propagation; the validated task is dropped by evaluateAndCleanUp. + info "Message successfully sent (propagation confirmed)", + requestId = task.requestId, msgHash = task.msgHash.to0xHex() + MessageSentEvent.emit(self.brokerCtx, task.requestId, task.msgHash.to0xHex()) + task.state = DeliveryState.SuccessfullyValidated return of DeliveryState.SuccessfullyValidated: info "Message successfully sent", @@ -276,5 +294,8 @@ proc send*(self: SendService, task: DeliveryTask) {.async.} = await self.sendProcessor.process(task) reportTaskResult(self, task) - if task.state != DeliveryState.FailedToDeliver: + # finalized tasks (failed, or already confirmed in propagation mode) must + # not enter the cache: the service loop would report them a second time + if task.state notin + {DeliveryState.FailedToDeliver, DeliveryState.SuccessfullyValidated}: self.addTask(task) diff --git a/logos_delivery/messaging/messaging_client.nim b/logos_delivery/messaging/messaging_client.nim index d883878fb..e7c9bd37c 100644 --- a/logos_delivery/messaging/messaging_client.nim +++ b/logos_delivery/messaging/messaging_client.nim @@ -1,6 +1,7 @@ ## Messaging layer core: the `MessagingClient` type plus its construction and ## lifecycle. The public operations (subscribe / unsubscribe / send) live in ## `messaging/api.nim`. +import std/strutils import results, chronos, chronicles import logos_delivery/api/conf/messaging_conf, @@ -24,7 +25,17 @@ proc new*( ## The messaging layer chains onto Waku: it drives the underlying Waku kernel ## for transport while exposing its own send/recv API. let reliability = conf.reliabilityEnabled.get(DefaultP2pReliability) - let sendService = ?SendService.new(reliability, waku) + + let confirmationMode = + case conf.sendConfirmation.get("store").strip().toLowerAscii() + of "store": + SendConfirmationMode.Store + of "propagation": + SendConfirmationMode.Propagation + else: + return err("invalid send-confirmation mode: " & conf.sendConfirmation.get("")) + + let sendService = ?SendService.new(reliability, waku, confirmationMode) let recvService = RecvService.new(waku) return ok( T( diff --git a/tests/api/test_api_send.nim b/tests/api/test_api_send.nim index 6470b0168..5d852080c 100644 --- a/tests/api/test_api_send.nim +++ b/tests/api/test_api_send.nim @@ -319,6 +319,53 @@ suite "Waku API - Send": (await node.stop()).isOkOr: raiseAssert "Failed to stop node: " & error + asyncTest "Propagation mode: send confirmed without a store node": + ## send-confirmation=propagation: MessageSent fires from the publish path; + ## no store node is connected (same shape as "Send only propagates", which + ## in store mode never fires MessageSent). + var node: LogosDelivery + lockNewGlobalBrokerContext: + node = ( + await LogosDelivery.new( + LogosDeliveryConf( + kernelConf: KernelConf(createApiNodeConf()), + messagingConf: + Opt.some(MessagingClientConf(sendConfirmation: Opt.some("propagation"))), + channelsConf: Opt.some(ReliableChannelManagerConf()), + ) + ) + ).valueOr: + raiseAssert error + (await node.start()).isOkOr: + raiseAssert "Failed to start Waku node: " & error + + await node.waku.node.connectToNodes(@[relayNode1PeerInfo]) + + let eventManager = newSendEventListenerManager(node.waku.brokerCtx) + defer: + await eventManager.teardown() + + let envelope = MessageEnvelope.init( + ContentTopic("/waku/2/default-content/proto"), "test payload" + ) + + let requestId = (await node.messagingClient.send(envelope)).valueOr: + raiseAssert error + + const eventTimeout = 10.seconds + discard await eventManager.waitForEvents(eventTimeout) + + eventManager.validate( + {SendEventOutcome.Sent, SendEventOutcome.Propagated}, requestId + ) + # exactly one confirmation: the finalized task must not be re-reported + # by later service-loop ticks + check eventManager.sentCount == 1 + check eventManager.propagatedCount == 1 + + (await node.stop()).isOkOr: + raiseAssert "Failed to stop node: " & error + asyncTest "Send only propagates fallback to lightpush": var node: LogosDelivery lockNewGlobalBrokerContext: