diff --git a/logos_delivery/messaging/delivery_service/send_service/delivery_task.nim b/logos_delivery/messaging/delivery_service/send_service/delivery_task.nim index f0c9d9942..0871fbe82 100644 --- a/logos_delivery/messaging/delivery_service/send_service/delivery_task.nim +++ b/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 diff --git a/logos_delivery/messaging/delivery_service/send_service/lightpush_processor.nim b/logos_delivery/messaging/delivery_service/send_service/lightpush_processor.nim index f4de74143..2c9ea777c 100644 --- a/logos_delivery/messaging/delivery_service/send_service/lightpush_processor.nim +++ b/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 diff --git a/logos_delivery/messaging/delivery_service/send_service/relay_processor.nim b/logos_delivery/messaging/delivery_service/send_service/relay_processor.nim index 801750f5a..ee3799e77 100644 --- a/logos_delivery/messaging/delivery_service/send_service/relay_processor.nim +++ b/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 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 bbfb4ea7d..90787f415 100644 --- a/logos_delivery/messaging/delivery_service/send_service/send_service.nim +++ b/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. diff --git a/tests/messaging/test_all.nim b/tests/messaging/test_all.nim index 3608348e8..cf54567f7 100644 --- a/tests/messaging/test_all.nim +++ b/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 diff --git a/tests/messaging/test_delivery_task_reaping.nim b/tests/messaging/test_delivery_task_reaping.nim new file mode 100644 index 000000000..959328d80 --- /dev/null +++ b/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)