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 <noreply@anthropic.com>

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)

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]

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)

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(

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: