From caad4077e9370559d581c5299563090f7c2bb664 Mon Sep 17 00:00:00 2001 From: ajaysehwal Date: Mon, 21 Sep 2026 21:56:18 +0530 Subject: [PATCH 1/2] Reclaim forwarded-payment replay markers instead of leaking them --- src/payment/forwarding_store.rs | 425 ++++++++++++++++++++++++++++---- 1 file changed, 378 insertions(+), 47 deletions(-) diff --git a/src/payment/forwarding_store.rs b/src/payment/forwarding_store.rs index 7bf88dcddf..da6770f1bb 100644 --- a/src/payment/forwarding_store.rs +++ b/src/payment/forwarding_store.rs @@ -74,9 +74,15 @@ pub(crate) struct ForwardRecord<'a> { #[derive(Clone, Debug, PartialEq, Eq)] struct ForwardedPaymentReplayMarker { id: ForwardedPaymentId, + /// When the forward this marker guards against replay was recorded. Defaults to `0` on read + /// for markers written before this field existed, so they're pruned immediately. + forwarded_at_timestamp: u64, } -impl_writeable_tlv_based!(ForwardedPaymentReplayMarker, { (0, id, required) }); +impl_writeable_tlv_based!(ForwardedPaymentReplayMarker, { + (0, id, required), + (2, forwarded_at_timestamp, (default_value, 0)), +}); impl StorableObject for ForwardedPaymentReplayMarker { type Id = ForwardedPaymentId; @@ -312,12 +318,13 @@ impl ForwardingStore { // Keep this marker after the event is handled. LDK can replay an older event after later // events have replaced the directional retry tokens, and it provides no callback after its // handled-event state is durable. - self.replay_markers.insert(ForwardedPaymentReplayMarker { id: forward_id }).await.map_err( - |e| { + self.replay_markers + .insert(ForwardedPaymentReplayMarker { id: forward_id, forwarded_at_timestamp }) + .await + .map_err(|e| { log_error!(self.logger, "Failed to store forwarded payment replay marker: {e}"); e - }, - )?; + })?; Ok(()) } @@ -549,36 +556,19 @@ async fn aggregate_forwarded_payments_and_log( } } +/// Runs for the life of the node in every tracking mode, `Stats` included. An earlier version +/// exited once both the details and marker stores were empty -- which, in `Stats` mode, is true +/// from the very first check on a node that hasn't forwarded anything yet, so the task would +/// return before ever reaching its periodic loop. Nothing re-spawns it, so every marker any later +/// forward wrote would go unreclaimed until the next restart -- silently defeating the point of +/// this module's cleanup. Running unconditionally costs a handful of cheap, mostly-empty store +/// reads once an hour when idle, which is negligible next to that failure mode. pub(crate) async fn run_forwarded_payment_aggregation( mut stop_receiver: tokio::sync::watch::Receiver<()>, forwarding_store: Arc, retention_secs: u64, ) { - if retention_secs == 0 { - match forwarding_store.details.is_empty().await { - Ok(true) => return, - Ok(false) => {}, - Err(e) => log_error!( - forwarding_store.logger, - "Failed to check forwarded payment store: {}", - e - ), - } - } - aggregate_forwarded_payments_and_log(&forwarding_store, retention_secs).await; - if retention_secs == 0 { - match forwarding_store.details.is_empty().await { - Ok(true) => return, - Ok(false) => {}, - Err(e) => log_error!( - forwarding_store.logger, - "Failed to check forwarded payment store: {}", - e - ), - } - } - let period = Duration::from_secs(FORWARDED_PAYMENT_AGGREGATION_BUCKET_SIZE_SECS); let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or(Duration::from_secs(0)).as_secs(); @@ -594,13 +584,6 @@ pub(crate) async fn run_forwarded_payment_aggregation( _ = stop_receiver.changed() => break, _ = interval.tick() => { aggregate_forwarded_payments_and_log(&forwarding_store, retention_secs).await; - if retention_secs == 0 { - match forwarding_store.details.is_empty().await { - Ok(true) => break, - Ok(false) => {}, - Err(e) => log_error!(forwarding_store.logger, "Failed to check forwarded payment store: {}", e), - } - } } } } @@ -616,7 +599,7 @@ async fn aggregate_expired_forwarded_payments( ) -> Result<(u64, u64), Error> { let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or(Duration::from_secs(0)).as_secs(); - aggregate_expired_forwarded_payments_at( + let result = aggregate_expired_forwarded_payments_at( forwarded_payment_store, replay_marker_store, channel_pair_stats_store, @@ -625,7 +608,84 @@ async fn aggregate_expired_forwarded_payments( now, logger, ) + .await?; + + // Reclaim markers independent of `retention_secs`, so `Stats` mode (no details to piggyback + // on) still reclaims them. Logged, not propagated -- the next pass retries. + match prune_expired_replay_markers( + forwarded_payment_store, + replay_marker_store, + FORWARDED_PAYMENT_AGGREGATION_BUCKET_SIZE_SECS, + now, + logger, + ) .await + { + Ok(removed) if removed > 0 => { + log_debug!(logger, "Reclaimed {} expired forwarded payment replay markers", removed); + }, + Err(e) => { + log_error!(logger, "Failed to reclaim forwarded payment replay markers: {}", e); + }, + _ => {}, + } + + Ok(result) +} + +/// Removes replay markers older than one bucket width **and** whose detail, if any, is gone. +/// +/// Both conditions matter: age alone isn't safe. A bucket with one detail still missing its +/// marker gets deferred whole by the aggregation pass above, siblings included. Pruning an +/// old-enough sibling marker by age alone would strand its still-present detail: every later pass +/// would see it missing a marker again and defer the bucket forever. Requiring the detail to +/// already be gone avoids that -- a marker outlives its own detail, however long that takes. +/// +/// The cutoff ignores `retention_secs` on purpose: that knob is about how long `Detailed` mode +/// keeps analytics data, not how long a marker needs to survive to do its job. +async fn prune_expired_replay_markers( + forwarded_payment_store: &ForwardedPaymentStore, + replay_marker_store: &ForwardedPaymentReplayMarkerStore, bucket_size_secs: u64, now: u64, + logger: &Arc, +) -> Result { + if bucket_size_secs == 0 { + return Ok(0); + } + let oldest_retained_bucket_start = + (now.saturating_sub(bucket_size_secs) / bucket_size_secs).saturating_mul(bucket_size_secs); + + // `Stats` mode never writes a detail at all, so every marker would otherwise cost an + // uncached round-trip here just to learn that. Skip it in one read when there's nothing a + // marker could possibly still be guarding. + let no_details_exist = forwarded_payment_store.is_empty().await?; + + let mut expired_ids = Vec::new(); + let mut page_token = None; + loop { + let page = replay_marker_store.list_page(page_token).await?; + for marker in page.objects { + if marker.forwarded_at_timestamp >= oldest_retained_bucket_start { + continue; + } + if !no_details_exist && forwarded_payment_store.contains_key(&marker.id).await? { + // Still guarding a live detail record -- its bucket hasn't closed yet. + continue; + } + expired_ids.push(marker.id); + } + let Some(next_page_token) = page.next_page_token else { break }; + page_token = Some(next_page_token); + } + + let mut removed = 0u64; + for id in expired_ids { + replay_marker_store.remove(&id).await.map_err(|e| { + log_error!(logger, "Failed to remove replay marker {:?}: {}", id, e); + e + })?; + removed += 1; + } + Ok(removed) } async fn aggregate_expired_forwarded_payments_at( @@ -952,8 +1012,12 @@ mod forwarding_stats_tests { replay_marker_store: &TestReplayMarkerStore, payment: ForwardedPaymentDetails, ) { let id = payment.id(); + let forwarded_at_timestamp = payment.forwarded_at_timestamp; forwarded_payment_store.insert(payment).await.unwrap(); - replay_marker_store.insert(ForwardedPaymentReplayMarker { id }).await.unwrap(); + replay_marker_store + .insert(ForwardedPaymentReplayMarker { id, forwarded_at_timestamp }) + .await + .unwrap(); } fn forwarded_payment( @@ -1118,7 +1182,10 @@ mod forwarding_stats_tests { ); replay_marker_store - .insert(ForwardedPaymentReplayMarker { id: payment.id() }) + .insert(ForwardedPaymentReplayMarker { + id: payment.id(), + forwarded_at_timestamp: payment.forwarded_at_timestamp, + }) .await .unwrap(); assert_eq!( @@ -1157,19 +1224,35 @@ mod forwarding_stats_tests { forwarding_store.details.insert(payment.clone()).await.unwrap(); forwarding_store .replay_markers - .insert(ForwardedPaymentReplayMarker { id: payment.id() }) + .insert(ForwardedPaymentReplayMarker { + id: payment.id(), + forwarded_at_timestamp: payment.forwarded_at_timestamp, + }) .await .unwrap(); - let (_stop_sender, stop_receiver) = tokio::sync::watch::channel(()); - - tokio::time::timeout( - Duration::from_secs(1), - run_forwarded_payment_aggregation(stop_receiver, Arc::clone(&forwarding_store), 0), - ) + let (stop_sender, stop_receiver) = tokio::sync::watch::channel(()); + + // The task no longer returns on its own once drained (see + // `run_forwarded_payment_aggregation`'s doc comment) -- it must keep running to reclaim + // whatever forwards happen next. Poll for the immediate first pass to land instead of + // awaiting the future directly. + let handle = tokio::spawn(run_forwarded_payment_aggregation( + stop_receiver, + Arc::clone(&forwarding_store), + 0, + )); + tokio::time::timeout(Duration::from_secs(1), async { + loop { + if forwarding_store.details.is_empty().await.unwrap() { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) .await .unwrap(); - assert!(forwarding_store.details.is_empty().await.unwrap()); + assert!(forwarding_store.replay_markers.is_empty().await.unwrap()); let bucket_id = channel_pair_stats_id(&payment.prev_channel_id, &payment.next_channel_id, 0); assert_eq!( @@ -1182,6 +1265,40 @@ mod forwarding_stats_tests { .payment_count, 1 ); + assert!(!handle.is_finished(), "the loop must keep running once drained"); + + stop_sender.send(()).unwrap(); + tokio::time::timeout(Duration::from_secs(1), handle).await.unwrap().unwrap(); + } + + /// Regression test: on a node that starts with both stores empty -- + /// e.g. before its first forward -- the loop used to return immediately and nothing ever + /// respawned it, so any marker written by a later forward would leak until the next restart. + #[tokio::test] + async fn loop_stays_alive_in_stats_mode_when_stores_start_empty() { + let kv_store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let logger = Arc::new(Logger::new_log_facade()); + let forwarding_store = Arc::new(ForwardingStore::new( + Vec::new(), + ForwardedPaymentTrackingMode::Stats, + kv_store, + logger, + )); + let (stop_sender, stop_receiver) = tokio::sync::watch::channel(()); + + let handle = tokio::spawn(run_forwarded_payment_aggregation( + stop_receiver, + Arc::clone(&forwarding_store), + 0, + )); + tokio::time::sleep(Duration::from_millis(200)).await; + assert!( + !handle.is_finished(), + "loop exited on startup; markers written later will never be reclaimed" + ); + + stop_sender.send(()).unwrap(); + tokio::time::timeout(Duration::from_secs(1), handle).await.unwrap().unwrap(); } #[tokio::test] @@ -1499,6 +1616,220 @@ mod forwarding_stats_tests { ); } + /// Regression test: replay markers used to leak forever (nothing removed them, and `Stats` + /// mode writes no detail to key cleanup off). Covers both tracking modes. + #[tokio::test] + async fn old_replay_markers_are_reclaimed_regardless_of_tracking_mode() { + for mode in [ForwardedPaymentTrackingMode::Stats, ForwardedPaymentTrackingMode::Detailed] { + let kv_store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let logger = Arc::new(Logger::new_log_facade()); + let forwarding_store = + ForwardingStore::new(Vec::new(), mode.clone(), Arc::clone(&kv_store), logger); + + let prev_channel_id = ChannelId([1; 32]); + let next_channel_id = ChannelId([2; 32]); + const FORWARDS: u64 = 50; + for htlc_id in 0..FORWARDS { + let prev_htlcs = [InboundHTLCLocator { + channel_id: prev_channel_id, + htlc_id: Some(htlc_id), + amount_msat: Some(1_000), + user_channel_id: None, + node_id: None, + }]; + let next_htlcs = [OutboundHTLCLocator { + channel_id: next_channel_id, + amount_msat: Some(999), + user_channel_id: None, + node_id: None, + }]; + forwarding_store + .record_forward(ForwardRecord { + prev_htlcs: &prev_htlcs, + next_htlcs: &next_htlcs, + total_fee_earned_msat: Some(1), + skimmed_fee_msat: None, + claim_from_onchain_tx: false, + outbound_amount_forwarded_msat: 999, + }) + .await + .unwrap(); + } + assert!( + !forwarding_store.replay_markers.is_empty().await.unwrap(), + "{mode:?}: forwarding should have written markers" + ); + + // Run aggregation as if a full bucket (plus margin) had already elapsed, so every + // marker just written is eligible for pruning -- mirroring what an operator sees after + // the node has been up for a couple of hours. + let real_now = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or(Duration::from_secs(0)) + .as_secs(); + let far_future = real_now + FORWARDED_PAYMENT_AGGREGATION_BUCKET_SIZE_SECS * 3; + let (_, _) = aggregate_expired_forwarded_payments_at( + &forwarding_store.details, + &forwarding_store.replay_markers, + &forwarding_store.channel_pair_stats, + FORWARDED_PAYMENT_AGGREGATION_BUCKET_SIZE_SECS, + 0, + far_future, + &forwarding_store.logger, + ) + .await + .unwrap(); + let removed = prune_expired_replay_markers( + &forwarding_store.details, + &forwarding_store.replay_markers, + FORWARDED_PAYMENT_AGGREGATION_BUCKET_SIZE_SECS, + far_future, + &forwarding_store.logger, + ) + .await + .unwrap(); + + assert_eq!(removed, FORWARDS, "{mode:?}: every expired marker should be reclaimed"); + assert!( + forwarding_store.replay_markers.is_empty().await.unwrap(), + "{mode:?}: no replay markers should remain" + ); + } + } + + /// Pins the exact age cutoff: a marker from the still-open or immediately-preceding bucket + /// must survive (it may still be protecting a detail record the aggregation pass above hasn't + /// resolved yet), while one from two bucket widths ago must be reclaimed. + #[tokio::test] + async fn prune_expired_replay_markers_respects_the_bucket_cutoff() { + let (forwarded_payment_store, replay_marker_store, _, logger, _) = test_stores_with_kv(); + let bucket_size_secs = FORWARDED_PAYMENT_AGGREGATION_BUCKET_SIZE_SECS; + let now = bucket_size_secs * 10; + + let recent_id = ForwardedPaymentId([1; 32]); + let old_id = ForwardedPaymentId([2; 32]); + replay_marker_store + .insert(ForwardedPaymentReplayMarker { + id: recent_id, + forwarded_at_timestamp: now - bucket_size_secs / 2, + }) + .await + .unwrap(); + replay_marker_store + .insert(ForwardedPaymentReplayMarker { + id: old_id, + forwarded_at_timestamp: now - bucket_size_secs * 2, + }) + .await + .unwrap(); + + let removed = prune_expired_replay_markers( + &forwarded_payment_store, + &replay_marker_store, + bucket_size_secs, + now, + &logger, + ) + .await + .unwrap(); + + assert_eq!(removed, 1); + assert!(replay_marker_store.contains_key(&recent_id).await.unwrap()); + assert!(!replay_marker_store.contains_key(&old_id).await.unwrap()); + } + + /// Regression test: age-only pruning would strand a sibling detail. If a bucket has one detail + /// missing its marker (crash mid-write), the whole bucket is deferred -- pruning an old + /// sibling marker by age alone would make it look broken too, deferring the bucket forever. + #[tokio::test] + async fn pruning_never_orphans_a_sibling_detail_in_a_deferred_bucket() { + let (forwarded_payment_store, replay_marker_store, channel_pair_stats_store, logger) = + test_stores(); + + // Same bucket (bucket width 60, both timestamps floor to 840), one crashed mid-write + // (`straggler`, no marker yet), one fully committed and already old enough to prune. + let straggler = forwarded_payment(1, 850, 110, 100, 10); + let sibling = forwarded_payment(2, 851, 220, 200, 20); + forwarded_payment_store.insert(straggler.clone()).await.unwrap(); + insert_completed_payment(&forwarded_payment_store, &replay_marker_store, sibling.clone()) + .await; + + // A pass far enough ahead that both would be prune-eligible by age alone. + let now = 10_000; + let bucket_size_secs = 60; + let retention_secs = 60; + + let aggregate_result = aggregate_expired_forwarded_payments_at( + &forwarded_payment_store, + &replay_marker_store, + &channel_pair_stats_store, + bucket_size_secs, + retention_secs, + now, + &logger, + ) + .await + .unwrap(); + assert_eq!(aggregate_result, (0, 0), "the whole bucket must be deferred"); + + let removed = prune_expired_replay_markers( + &forwarded_payment_store, + &replay_marker_store, + bucket_size_secs, + now, + &logger, + ) + .await + .unwrap(); + assert_eq!(removed, 0, "the sibling's marker still guards a live detail"); + + // Both details, and the sibling's marker, must have survived untouched. + assert_eq!( + forwarded_payment_store.get(&sibling.id()).await.unwrap(), + Some(sibling.clone()) + ); + assert_eq!( + forwarded_payment_store.get(&straggler.id()).await.unwrap(), + Some(straggler.clone()) + ); + assert!(replay_marker_store.contains_key(&sibling.id()).await.unwrap()); + + // The straggler's write completes: its marker finally appears. + replay_marker_store + .insert(ForwardedPaymentReplayMarker { + id: straggler.id(), + forwarded_at_timestamp: straggler.forwarded_at_timestamp, + }) + .await + .unwrap(); + + let aggregate_result = aggregate_expired_forwarded_payments_at( + &forwarded_payment_store, + &replay_marker_store, + &channel_pair_stats_store, + bucket_size_secs, + retention_secs, + now, + &logger, + ) + .await + .unwrap(); + assert_eq!(aggregate_result, (1, 2), "the now-complete bucket aggregates normally"); + assert!(forwarded_payment_store.get(&sibling.id()).await.unwrap().is_none()); + assert!(forwarded_payment_store.get(&straggler.id()).await.unwrap().is_none()); + + let removed = prune_expired_replay_markers( + &forwarded_payment_store, + &replay_marker_store, + bucket_size_secs, + now, + &logger, + ) + .await + .unwrap(); + assert_eq!(removed, 2, "both markers are reclaimed once their details are gone"); + } + #[tokio::test] async fn zero_retention_cleans_up_after_the_current_bucket_closes() { let (forwarded_payment_store, replay_marker_store, channel_pair_stats_store, logger) = From eb251c46c75b86ec918132ab71e93e388ae8082e Mon Sep 17 00:00:00 2001 From: ajaysehwal Date: Tue, 22 Sep 2026 08:06:58 +0530 Subject: [PATCH 2/2] Delay replay-marker pruning and pair it with detail removal --- src/payment/forwarding_store.rs | 169 +++++++++++++++++++++++++------- 1 file changed, 132 insertions(+), 37 deletions(-) diff --git a/src/payment/forwarding_store.rs b/src/payment/forwarding_store.rs index da6770f1bb..1761823b25 100644 --- a/src/payment/forwarding_store.rs +++ b/src/payment/forwarding_store.rs @@ -74,14 +74,13 @@ pub(crate) struct ForwardRecord<'a> { #[derive(Clone, Debug, PartialEq, Eq)] struct ForwardedPaymentReplayMarker { id: ForwardedPaymentId, - /// When the forward this marker guards against replay was recorded. Defaults to `0` on read - /// for markers written before this field existed, so they're pruned immediately. + /// When the forward this marker guards against replay was recorded. forwarded_at_timestamp: u64, } impl_writeable_tlv_based!(ForwardedPaymentReplayMarker, { (0, id, required), - (2, forwarded_at_timestamp, (default_value, 0)), + (2, forwarded_at_timestamp, required), }); impl StorableObject for ForwardedPaymentReplayMarker { @@ -384,6 +383,24 @@ impl ForwardingStore { ) .await } + + /// Sweeps markers the aggregation pass above didn't already remove alongside their detail -- + /// `Stats` mode, which writes no detail to pair a removal with, and crash orphans. See + /// [`prune_expired_replay_markers`] for why age alone isn't a safe criterion here. + pub(crate) async fn prune_stale_replay_markers(&self) -> Result { + let now = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or(Duration::from_secs(0)) + .as_secs(); + prune_expired_replay_markers( + &self.details, + &self.replay_markers, + FORWARDED_PAYMENT_AGGREGATION_BUCKET_SIZE_SECS, + now, + &self.logger, + ) + .await + } } fn forwarded_payment_id(channel_id: &ChannelId, htlc_id: u64) -> ForwardedPaymentId { @@ -556,17 +573,23 @@ async fn aggregate_forwarded_payments_and_log( } } -/// Runs for the life of the node in every tracking mode, `Stats` included. An earlier version -/// exited once both the details and marker stores were empty -- which, in `Stats` mode, is true -/// from the very first check on a node that hasn't forwarded anything yet, so the task would -/// return before ever reaching its periodic loop. Nothing re-spawns it, so every marker any later -/// forward wrote would go unreclaimed until the next restart -- silently defeating the point of -/// this module's cleanup. Running unconditionally costs a handful of cheap, mostly-empty store -/// reads once an hour when idle, which is negligible next to that failure mode. +/// Runs for the life of the node in every tracking mode, `Stats` included: on an idle node the +/// only cost is a handful of cheap, mostly-empty store reads once an hour, which is negligible +/// next to leaving future markers permanently unreclaimed if it stopped. +/// +/// The age-based marker sweep only starts after one bucket width of uptime. On startup the +/// background processor may still need to replay an event we recorded but hadn't yet durably +/// drained before a prior crash; `record_forward` relies on that event's marker still being there +/// to recognize the replay and skip double-counting it. Sweeping on the immediate startup pass +/// could delete that marker first. Aggregation itself runs immediately regardless -- it only ever +/// removes a marker alongside the detail it was confirmed to guard, which carries no such risk. pub(crate) async fn run_forwarded_payment_aggregation( mut stop_receiver: tokio::sync::watch::Receiver<()>, forwarding_store: Arc, retention_secs: u64, ) { + let started_at = tokio::time::Instant::now(); + let prune_delay = Duration::from_secs(FORWARDED_PAYMENT_AGGREGATION_BUCKET_SIZE_SECS); + aggregate_forwarded_payments_and_log(&forwarding_store, retention_secs).await; let period = Duration::from_secs(FORWARDED_PAYMENT_AGGREGATION_BUCKET_SIZE_SECS); @@ -584,11 +607,34 @@ pub(crate) async fn run_forwarded_payment_aggregation( _ = stop_receiver.changed() => break, _ = interval.tick() => { aggregate_forwarded_payments_and_log(&forwarding_store, retention_secs).await; + if started_at.elapsed() >= prune_delay { + prune_stale_replay_markers_and_log(&forwarding_store).await; + } } } } } +async fn prune_stale_replay_markers_and_log(forwarding_store: &ForwardingStore) { + match forwarding_store.prune_stale_replay_markers().await { + Ok(removed) if removed > 0 => { + log_debug!( + forwarding_store.logger, + "Reclaimed {} expired forwarded payment replay markers", + removed + ); + }, + Err(e) => { + log_error!( + forwarding_store.logger, + "Failed to reclaim forwarded payment replay markers: {}", + e + ); + }, + _ => {}, + } +} + /// Aggregate forwarded payments older than the configured retention period into fixed-width /// channel-pair statistics buckets. async fn aggregate_expired_forwarded_payments( @@ -599,7 +645,7 @@ async fn aggregate_expired_forwarded_payments( ) -> Result<(u64, u64), Error> { let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or(Duration::from_secs(0)).as_secs(); - let result = aggregate_expired_forwarded_payments_at( + aggregate_expired_forwarded_payments_at( forwarded_payment_store, replay_marker_store, channel_pair_stats_store, @@ -608,29 +654,7 @@ async fn aggregate_expired_forwarded_payments( now, logger, ) - .await?; - - // Reclaim markers independent of `retention_secs`, so `Stats` mode (no details to piggyback - // on) still reclaims them. Logged, not propagated -- the next pass retries. - match prune_expired_replay_markers( - forwarded_payment_store, - replay_marker_store, - FORWARDED_PAYMENT_AGGREGATION_BUCKET_SIZE_SECS, - now, - logger, - ) .await - { - Ok(removed) if removed > 0 => { - log_debug!(logger, "Reclaimed {} expired forwarded payment replay markers", removed); - }, - Err(e) => { - log_error!(logger, "Failed to reclaim forwarded payment replay markers: {}", e); - }, - _ => {}, - } - - Ok(result) } /// Removes replay markers older than one bucket width **and** whose detail, if any, is gone. @@ -828,6 +852,15 @@ async fn aggregate_expired_forwarded_payments_at( log_error!(logger, "Failed to remove forwarded payment {:?}: {}", payment_id, e); e })?; + // The marker's job ends with its detail: this bucket was only aggregated because the + // marker confirmed the detail was durably recorded, so nothing needs it anymore. Removing + // it here, rather than leaving it to the separate age-based sweep, means that sweep rarely + // has to fall back to its per-marker existence check -- almost every Detailed-mode marker + // is gone by the time it would otherwise become a candidate. + replay_marker_store.remove(&payment_id).await.map_err(|e| { + log_error!(logger, "Failed to remove replay marker {:?}: {}", payment_id, e); + e + })?; removed_payment_count += 1; } @@ -1301,6 +1334,57 @@ mod forwarding_stats_tests { tokio::time::timeout(Duration::from_secs(1), handle).await.unwrap().unwrap(); } + /// Regression test: the immediate startup pass must not sweep an old marker with no detail + /// behind it -- after a crash, that's indistinguishable from an event the background + /// processor still needs to replay, and `record_forward` depends on the marker being there to + /// recognize the replay. `prune_expired_replay_markers_respects_the_bucket_cutoff` above + /// already covers the sweep itself eventually reclaiming a marker this old; this test is + /// narrowly about the startup pass not being the one to do it. + #[tokio::test] + async fn startup_pass_does_not_sweep_an_old_marker() { + let kv_store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let logger = Arc::new(Logger::new_log_facade()); + let forwarding_store = Arc::new(ForwardingStore::new( + Vec::new(), + ForwardedPaymentTrackingMode::Stats, + kv_store, + Arc::clone(&logger), + )); + + // Old enough to be sweep-eligible by age alone, and with no detail (as in `Stats` mode, + // or a crash-orphaned marker either way) -- exactly what the startup pass must not touch. + let marker_id = ForwardedPaymentId([7; 32]); + let now = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or(Duration::from_secs(0)) + .as_secs(); + forwarding_store + .replay_markers + .insert(ForwardedPaymentReplayMarker { + id: marker_id, + forwarded_at_timestamp: now + .saturating_sub(FORWARDED_PAYMENT_AGGREGATION_BUCKET_SIZE_SECS * 2), + }) + .await + .unwrap(); + + let (stop_sender, stop_receiver) = tokio::sync::watch::channel(()); + let handle = tokio::spawn(run_forwarded_payment_aggregation( + stop_receiver, + Arc::clone(&forwarding_store), + 0, + )); + + tokio::time::sleep(Duration::from_millis(200)).await; + assert!( + forwarding_store.replay_markers.contains_key(&marker_id).await.unwrap(), + "the startup pass must not sweep a marker before the uptime delay elapses" + ); + + stop_sender.send(()).unwrap(); + tokio::time::timeout(Duration::from_secs(1), handle).await.unwrap().unwrap(); + } + #[tokio::test] async fn forwarding_store_records_details_and_channel_stats() { let kv_store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); @@ -1668,7 +1752,10 @@ mod forwarding_stats_tests { .unwrap_or(Duration::from_secs(0)) .as_secs(); let far_future = real_now + FORWARDED_PAYMENT_AGGREGATION_BUCKET_SIZE_SECS * 3; - let (_, _) = aggregate_expired_forwarded_payments_at( + // In `Detailed` mode, aggregation removes each marker alongside its now-resolved + // detail; in `Stats` mode there's no detail to pair with, so the markers are left for + // the sweep below. Either way, every marker should be gone by the end. + let (_, aggregated_removed) = aggregate_expired_forwarded_payments_at( &forwarding_store.details, &forwarding_store.replay_markers, &forwarding_store.channel_pair_stats, @@ -1679,7 +1766,7 @@ mod forwarding_stats_tests { ) .await .unwrap(); - let removed = prune_expired_replay_markers( + let swept = prune_expired_replay_markers( &forwarding_store.details, &forwarding_store.replay_markers, FORWARDED_PAYMENT_AGGREGATION_BUCKET_SIZE_SECS, @@ -1689,7 +1776,11 @@ mod forwarding_stats_tests { .await .unwrap(); - assert_eq!(removed, FORWARDS, "{mode:?}: every expired marker should be reclaimed"); + assert_eq!( + aggregated_removed + swept, + FORWARDS, + "{mode:?}: every expired marker should be reclaimed, one way or the other" + ); assert!( forwarding_store.replay_markers.is_empty().await.unwrap(), "{mode:?}: no replay markers should remain" @@ -1817,6 +1908,10 @@ mod forwarding_stats_tests { assert_eq!(aggregate_result, (1, 2), "the now-complete bucket aggregates normally"); assert!(forwarded_payment_store.get(&sibling.id()).await.unwrap().is_none()); assert!(forwarded_payment_store.get(&straggler.id()).await.unwrap().is_none()); + // Aggregation removes each marker alongside the detail it just confirmed and resolved -- + // both are already gone here, before the separate sweep ever runs. + assert!(!replay_marker_store.contains_key(&sibling.id()).await.unwrap()); + assert!(!replay_marker_store.contains_key(&straggler.id()).await.unwrap()); let removed = prune_expired_replay_markers( &forwarded_payment_store, @@ -1827,7 +1922,7 @@ mod forwarding_stats_tests { ) .await .unwrap(); - assert_eq!(removed, 2, "both markers are reclaimed once their details are gone"); + assert_eq!(removed, 0, "nothing left for the sweep once aggregation already removed both"); } #[tokio::test]