fix: charge admission once per task and exempt budget-parked tasks from the reaper

Two defects the send-service seam had once admission actually enforces,
both fixed by a single DeliveryTask.firstAdmittedTime field:

- The retry loop gated admission on firstPropagatedTime.isNone(), so a
  task that failed to propagate (e.g. no peers) was re-admitted on every
  1s tick, re-charging the epoch budget and — with the shipped default of
  one message per epoch — starving all other traffic. Admission is now
  gated on firstAdmittedTime.isNone(): a task charges one slot / draws one
  nonce for its lifetime, and retries reuse it.

- reportTaskResult reaped any never-propagated task older than
  MaxTimeInCache (60s) measured from message creation, so an over-budget
  task parked for a 600s epoch was hard-failed with a misleading "Unable
  to send within retry time window" long before the epoch could roll
  (issue #4049). The reaper now runs only for admitted tasks and measures
  from admission, so a task waiting for epoch budget is exempt and gets a
  full delivery window once it is finally admitted.

The RLN-rejection branch in both processors resets firstAdmittedTime
along with clearing the proof, so the regenerated proof's fresh nonce is
re-admitted rather than sent uncharged.

The reap decision is extracted to DeliveryTask.isDeliveryTimedOut and
unit-tested (tests/messaging/test_delivery_task_reaping.nim); a
scheduler-level integration test of park-and-release is a follow-up.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

logos_delivery/messaging/delivery_service/send_service/delivery_task.nim

@@ -26,6 +26,14 @@ type DeliveryTask* = ref object
   firstPropagatedTime*: Opt[Moment]
     ## Set once on the first successful propagation; never reset on re-publish.
     ## Anchors the store-validation time cap (see propagationAge).
+  firstAdmittedTime*: Opt[Moment]
+    ## Set when the task first passes rate-limit admission (consumes a budget
+    ## slot / draws an RLN nonce); `none` while still parked waiting for epoch
+    ## budget. Guards re-admission on retry and anchors the delivery-timeout
+    ## reaper from the first real send attempt rather than message creation, so
+    ## a task parked for budget is not aged out before it can be sent. Reset to
+    ## `none` when an RLN rejection clears the proof, since regenerating draws a
+    ## fresh nonce that must be re-admitted.
   propagateEventEmitted*: bool
   errorDesc*: string
 
@@ -84,5 +92,22 @@ proc propagationAge*(self: DeliveryTask): timer.Duration =
   else:
     ZeroDuration
 
+proc admissionAge*(self: DeliveryTask): timer.Duration =
+  ## Time elapsed since the task first passed admission; ZeroDuration while it
+  ## has never been admitted (i.e. still parked waiting for epoch budget).
+  if self.firstAdmittedTime.isSome():
+    timer.Moment.now() - self.firstAdmittedTime.get()
+  else:
+    ZeroDuration
+
+proc isDeliveryTimedOut*(self: DeliveryTask, maxTime: timer.Duration): bool =
+  ## True when the task was admitted (drew a slot) and has been trying to
+  ## deliver longer than `maxTime` without ever propagating. A task never
+  ## admitted — still parked waiting for epoch budget — is exempt: it is waiting
+  ## for the epoch to roll, not failing to deliver, and the clock runs from
+  ## admission, not message creation, so budget wait does not count against it.
+  self.firstAdmittedTime.isSome() and self.firstPropagatedTime.isNone() and
+    self.admissionAge() > maxTime
+
 proc isEphemeral*(self: DeliveryTask): bool =
   return self.msg.ephemeral

logos_delivery/messaging/delivery_service/send_service/lightpush_processor.nim

@@ -37,10 +37,11 @@ method sendImpl*(
     if error.isRlnRejection():
       ## The proof was refused, so it must not be sent again: drop it and let
       ## the refreshed merkle path produce a new one on the next round.
-      ## Re-admission gates the regeneration, so a task cannot spin through the
-      ## epoch budget by retrying.
+      ## Resetting admission re-charges the fresh nonce that regeneration draws,
+      ## so a task cannot spin through the epoch budget by retrying.
       self.waku.onRlnProofRejected()
       task.msg.proof = @[]
+      task.firstAdmittedTime = Opt.none(Moment)
       task.state = DeliveryState.NextRoundRetry
       return
 

logos_delivery/messaging/delivery_service/send_service/relay_processor.nim

@@ -69,9 +69,11 @@ method sendImpl*(self: RelaySendProcessor, task: DeliveryTask) {.async.} =
       ## The relay validator refused the proof. Dropping it and retrying is not
       ## the same as failing: the message is valid, its proof went stale against
       ## a moved merkle root. Clearing it makes the next round regenerate one
-      ## against the refreshed path.
+      ## against the refreshed path; resetting admission re-charges the fresh
+      ## nonce that regeneration draws.
       self.waku.onRlnProofRejected()
       task.msg.proof = @[]
+      task.firstAdmittedTime = Opt.none(Moment)
       task.state = DeliveryState.NextRoundRetry
       return
 

logos_delivery/messaging/delivery_service/send_service/send_service.nim

@@ -199,15 +199,19 @@ proc reportTaskResult(self: SendService, task: DeliveryTask) =
     # rest of the states are intermediate and does not translate to event
     discard
 
-  # Only tasks that never propagated are reported as hard send failures here.
+  # Hard-fail a task that was admitted (drew a slot) but has been trying to
+  # deliver longer than MaxTimeInCache without ever propagating. A task still
+  # parked for epoch budget (never admitted) is exempt — it is waiting for the
+  # epoch to roll, not failing to deliver, and the timeout is measured from
+  # admission, not message creation, so budget wait does not count against it.
   # Propagated-but-not-store-validated tasks are handled (warn + drop, no event)
   # in evaluateAndCleanUp.
-  if task.firstPropagatedTime.isNone() and task.messageAge() > MaxTimeInCache:
+  if task.isDeliveryTimedOut(MaxTimeInCache):
     error "Failed to send message",
       requestId = task.requestId,
       msgHash = task.msgHash.to0xHex(),
       error = "Message too old",
-      age = task.messageAge()
+      age = task.admissionAge()
     task.state = DeliveryState.FailedToDeliver
     MessageErrorEvent.emit(
       self.brokerCtx,
@@ -255,12 +259,16 @@ proc trySendMessages(self: SendService) {.async.} =
 
   for task in tasksToSend:
     # Todo, check if it has any perf gain to run them concurrent...
-    if task.firstPropagatedTime.isNone():
-      ## Never propagated: this transmission consumes a fresh epoch slot, so
-      ## it must be admitted. Tasks that stay over budget remain in
-      ## `NextRoundRetry` and are retried as the epoch rolls over.
+    if task.firstAdmittedTime.isNone():
+      ## Not yet admitted: this transmission will consume a fresh epoch slot, so
+      ## it must be admitted (and draw a nonce). Guarding on admission rather
+      ## than propagation means a task that failed to propagate is not
+      ## re-charged every tick — it keeps its slot and its proof. Tasks that
+      ## stay over budget remain in `NextRoundRetry` and are retried as the
+      ## epoch rolls over.
       (await self.rateLimitManager.admit(task.msg.payload)).isOkOr:
         continue
+      task.firstAdmittedTime = Opt.some(Moment.now())
 
       ## Strictly after admission, so a rejected message never draws a nonce.
       ## A no-op when RLN is not mounted, or when a prior round already
@@ -305,6 +313,7 @@ proc send*(self: SendService, task: DeliveryTask) {.async.} =
     task.state = DeliveryState.NextRoundRetry
     self.addTask(task)
     return
+  task.firstAdmittedTime = Opt.some(Moment.now())
 
   ## Strictly after admission, so a rejected message never draws a nonce.
   ## A no-op when RLN is not mounted.

tests/messaging/test_all.nim

@@ -1,3 +1,3 @@
 {.used.}
 
-import ./test_rate_limit_manager, ./test_rln_proof_attach
+import ./test_rate_limit_manager, ./test_rln_proof_attach, ./test_delivery_task_reaping

tests/messaging/test_delivery_task_reaping.nim

@@ -0,0 +1,39 @@
+{.used.}
+
+import results, chronos, testutils/unittests
+
+import logos_delivery/messaging/delivery_service/send_service/delivery_task
+
+const MaxTime = chronos.minutes(1)
+
+proc taskWith(admitted, propagated: Opt[Moment]): DeliveryTask =
+  ## Builds a DeliveryTask directly (bypassing `new`, which needs a broker) with
+  ## only the fields the reaping predicate reads.
+  DeliveryTask(firstAdmittedTime: admitted, firstPropagatedTime: propagated)
+
+suite "DeliveryTask - delivery-timeout reaping":
+  test "a task parked for budget (never admitted) is exempt, however old":
+    ## The #4049 fix: budget-parked tasks must survive to the epoch roll instead
+    ## of being aged out with a misleading failure.
+    let task = taskWith(Opt.none(Moment), Opt.none(Moment))
+    check not task.isDeliveryTimedOut(MaxTime)
+
+  test "an admitted, never-propagated task past the window times out":
+    let task = taskWith(Opt.some(Moment.now() - chronos.minutes(2)), Opt.none(Moment))
+    check task.isDeliveryTimedOut(MaxTime)
+
+  test "an admitted task still within the window does not time out":
+    let task = taskWith(Opt.some(Moment.now()), Opt.none(Moment))
+    check not task.isDeliveryTimedOut(MaxTime)
+
+  test "a propagated task is never timed out here (store validation owns it)":
+    let task =
+      taskWith(Opt.some(Moment.now() - chronos.minutes(2)), Opt.some(Moment.now()))
+    check not task.isDeliveryTimedOut(MaxTime)
+
+  test "the timeout clock runs from admission, not message creation":
+    ## A task that waited a long time for budget then just got admitted has a
+    ## fresh clock — it is not reaped immediately on admission.
+    let task = taskWith(Opt.some(Moment.now()), Opt.none(Moment))
+    check task.admissionAge() < MaxTime
+    check not task.isDeliveryTimedOut(MaxTime)