From c5b52f3b5896ebd7ce0086827d81a30ee80fe659 Mon Sep 17 00:00:00 2001 From: mfw78 Date: Sat, 18 Jul 2026 00:40:16 +0000 Subject: [PATCH] twap: grant one next-block retry before dropping on InvalidEip1271Signature A first-time user's Safe wiring and ComposableCoW create() land in one block, so the orderbook rejects the first submission against its own head and the keeper dropped a valid watch. The classification table gains a drop-on-repeat action, the retry ledger records the refused block and gates one next-block retry, and a denied refusal re-enters the table by its errorType prefix so the grace survives the coarse venue-error collapse. A repeat on a later block still drops. --- Cargo.lock | 1 + crates/composable-cow/src/sweep.rs | 23 ++- crates/composable-cow/tests/sweep.rs | 150 ++++++++++++++++++++ crates/cow-venue/build.rs | 1 + crates/cow-venue/data/classification.toml | 23 ++- crates/cow-venue/src/adapter.rs | 12 ++ crates/cow-venue/src/classification.rs | 47 +++++- crates/cow-venue/src/classification_data.rs | 5 +- crates/cow-venue/src/lib.rs | 2 +- crates/nexum-sdk/src/keeper.rs | 95 +++++++++---- crates/nexum-sdk/tests/keeper.rs | 122 +++++++++++++++- crates/videre-sdk/Cargo.toml | 2 + crates/videre-sdk/src/keeper.rs | 63 +++++++- 13 files changed, 492 insertions(+), 54 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 533912ed..1b2b7ece 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5983,6 +5983,7 @@ dependencies = [ "nexum-sdk-test", "strum", "thiserror 2.0.18", + "tracing", "videre-macros", "videre-status-body", "wit-bindgen 0.59.0", diff --git a/crates/composable-cow/src/sweep.rs b/crates/composable-cow/src/sweep.rs index d3fa5dd7..768d51c4 100644 --- a/crates/composable-cow/src/sweep.rs +++ b/crates/composable-cow/src/sweep.rs @@ -13,21 +13,24 @@ //! Store faults abort the sweep (the next tick replays it); //! submission failures never do - they fold into a //! [`RetryAction`] through the videre -//! [`retry_action`] table, the ledger applies the effect, and the +//! [`retry_action`] table, a `denied` refusal re-entering the CoW +//! classification through its errorType prefix +//! ([`classify_denied`]) so a one-shot row survives the coarse +//! collapse, the ledger applies the effect, and the //! sweep moves on. Diagnostics go through the guest `tracing` facade - //! the same channel strategy code logs on - so module tests observe //! the composed behaviour with one capture. use alloy_primitives::{Address, Bytes, hex}; use cow_venue::assembly::{gpv2_to_order_data, order_data_to_body}; -use cow_venue::{CowClient, CowIntent, CowIntentBody, SignedOrder, intent_id}; +use cow_venue::{CowClient, CowIntent, CowIntentBody, SignedOrder, classify_denied, intent_id}; use cowprotocol::GPv2OrderData; use nexum_sdk::host::{Fault, LocalStoreHost}; use nexum_sdk::keeper::{ ConditionalSource, Gates, Journal, Retrier, RetryAction, Tick, WatchRef, WatchSet, }; use videre_sdk::keeper::retry_action; -use videre_sdk::{ClientError, SubmitOutcome, VenueTransport, rt}; +use videre_sdk::{ClientError, SubmitOutcome, VenueFault, VenueTransport, rt}; use crate::Verdict; @@ -141,6 +144,12 @@ where }; match outcome { Ok(SubmitOutcome::Accepted(receipt)) => { + // An acceptance ends any refusal episode: clear the + // first-refusal marker so a later independent refusal + // earns a fresh one-block grace. + if let Err(fault) = Retrier::new(host).clear_refusal(watch) { + tracing::error!("submitted {intent_id} but refusal-marker clear failed: {fault}"); + } // The submit already succeeded; a journal-store fault here // must not abort the sweep or unwind the accepted order. // Log and carry on - the already-submitted arm keeps the @@ -162,13 +171,17 @@ where tracing::error!("intent body encode failed: {err}"); } Err(ClientError::Venue(fault)) => { - let action = retry_action(&fault); - Retrier::new(host).apply(watch, action, tick.epoch_s)?; + let action = match &fault { + VenueFault::Denied(detail) => classify_denied(detail), + other => retry_action(other), + }; + Retrier::new(host).apply(watch, action, tick)?; match action { RetryAction::TryNextBlock => tracing::warn!("submit retry-next-block: {fault}"), RetryAction::Backoff { seconds } => { tracing::warn!("submit backoff {seconds}s: {fault}"); } + RetryAction::DropOnRepeat => tracing::warn!("submit drop-on-repeat: {fault}"), RetryAction::Drop => tracing::warn!("submit dropped watch: {fault}"), // `RetryAction` is non-exhaustive; the ledger already // ran the effect, so the log needs only the name. diff --git a/crates/composable-cow/tests/sweep.rs b/crates/composable-cow/tests/sweep.rs index 7d400f4c..567a1384 100644 --- a/crates/composable-cow/tests/sweep.rs +++ b/crates/composable-cow/tests/sweep.rs @@ -512,6 +512,156 @@ fn rate_limited_submit_backs_off_through_the_epoch_gate() { assert!(!snapshot.keys().any(|k| k.starts_with("submitted:"))); } +/// The adapter's projection of the same-block wiring+create race: a +/// first-time user's Safe wiring and `create()` land in one block, so +/// the orderbook rejects the first submission against its own head. +fn eip1271_rejection() -> Result { + Err(VenueFault::Denied( + "InvalidEip1271Signature: signature for computed order hash 0x7ee5 is not valid".into(), + )) +} + +/// Same-block wiring+create race: the first rejection gates the watch +/// to the next block instead of dropping it, re-polls within the block +/// stay gated, and the retried submission one block later lands. +#[test] +fn first_eip1271_rejection_retries_on_the_next_block() { + let host = MockHost::new(); + let key = seed_watch(&host); + let watch = WatchRef::parse(&key).unwrap(); + let order = submittable_order(); + let venue = MockVenue::default(); + venue.enqueue_submit(eip1271_rejection()); + venue.enqueue_submit(accepted()); + + let source = { + let order = order.clone(); + src(move |_, _, _, _| ready_outcome(&order)) + }; + let tick = sample_tick(); + let (result, logs) = capture_tracing(|| run(&host, &client(&venue), &source, &tick)); + result.unwrap(); + + let snapshot = host.store.snapshot(); + assert!( + snapshot.contains_key(&key), + "first rejection keeps the watch" + ); + assert_eq!( + snapshot.get(&watch.next_block_key()).unwrap(), + &(tick.block + 1).to_le_bytes().to_vec(), + "the watch gates to the next block", + ); + assert!(logs.any(|e| e.message.contains("drop-on-repeat"))); + + // Sub-block re-polls stay gated: the race is not hammered. + run(&host, &client(&venue), &source, &tick).unwrap(); + assert_eq!(venue.submit_count(), 1); + + // One block later the wiring is visible and the retry lands. + let next = Tick { + block: tick.block + 1, + ..tick + }; + run(&host, &client(&venue), &source, &next).unwrap(); + assert_eq!(venue.submit_count(), 2); + assert!( + Journal::submitted(&host) + .contains(&intent_id(&order)) + .unwrap(), + ); +} + +/// A rejection that repeats on a later block is a genuinely broken +/// signature: the watch and every derived key go. +#[test] +fn repeated_eip1271_rejection_on_a_later_block_drops_the_watch() { + let host = MockHost::new(); + seed_watch(&host); + let order = submittable_order(); + let venue = MockVenue::default(); + venue.enqueue_submit(eip1271_rejection()); + venue.enqueue_submit(eip1271_rejection()); + + let source = src(move |_, _, _, _| ready_outcome(&order)); + let tick = sample_tick(); + run(&host, &client(&venue), &source, &tick).unwrap(); + + let next = Tick { + block: tick.block + 1, + ..tick + }; + run(&host, &client(&venue), &source, &next).unwrap(); + + assert_eq!(venue.submit_count(), 2); + assert!( + host.store.is_empty(), + "a repeated rejection must drop the watch, its gates, and the marker", + ); +} + +/// An accepted submission ends the refusal episode: a later tranche's +/// own first rejection earns a fresh one-block grace instead of an +/// immediate drop on the stale marker from an earlier tranche. +#[test] +fn acceptance_resets_the_one_block_grace_for_later_tranches() { + let host = MockHost::new(); + let key = seed_watch(&host); + let watch = WatchRef::parse(&key).unwrap(); + let tranche_one = submittable_order(); + let mut tranche_two = submittable_order(); + tranche_two.buyAmount = U256::from(1_001_u64); + + let venue = MockVenue::default(); + venue.enqueue_submit(eip1271_rejection()); + venue.enqueue_submit(accepted()); + venue.enqueue_submit(eip1271_rejection()); + + let tick = sample_tick(); + let boundary = tick.block + 5; + let source = src(move |_, _, _, t: &Tick| { + if t.block < boundary { + ready_outcome(&tranche_one) + } else { + ready_outcome(&tranche_two) + } + }); + + // Tranche one: refused at the tick block, accepted one block later. + run(&host, &client(&venue), &source, &tick).unwrap(); + let next = Tick { + block: tick.block + 1, + ..tick + }; + run(&host, &client(&venue), &source, &next).unwrap(); + assert_eq!(venue.submit_count(), 2); + assert!( + !host.store.snapshot().contains_key(&watch.refused_key()), + "acceptance must clear the first-refusal marker", + ); + + // Tranche two: its own first rejection at a later block keeps the + // watch and gates it to the next block. + let later = Tick { + block: boundary, + ..tick + }; + run(&host, &client(&venue), &source, &later).unwrap(); + let snapshot = host.store.snapshot(); + assert!( + snapshot.contains_key(&key), + "a fresh refusal after an acceptance must keep the watch", + ); + assert_eq!( + snapshot.get(&watch.refused_key()).unwrap(), + &later.block.to_le_bytes().to_vec(), + ); + assert_eq!( + snapshot.get(&watch.next_block_key()).unwrap(), + &(later.block + 1).to_le_bytes().to_vec(), + ); +} + /// Restart regression: a keeper that posted, journalled, and then /// restarted over the same persistent local store must not post the /// same order again - one venue submit across both lives. diff --git a/crates/cow-venue/build.rs b/crates/cow-venue/build.rs index 10122c97..f619f788 100644 --- a/crates/cow-venue/build.rs +++ b/crates/cow-venue/build.rs @@ -28,6 +28,7 @@ fn main() { let action = match e.action { Action::TryNextBlock => "GenAction::TryNextBlock", Action::Backoff => "GenAction::Backoff", + Action::DropOnRepeat => "GenAction::DropOnRepeat", Action::Drop => "GenAction::Drop", }; out.push_str(&format!( diff --git a/crates/cow-venue/data/classification.toml b/crates/cow-venue/data/classification.toml index f2cbb40b..2d236248 100644 --- a/crates/cow-venue/data/classification.toml +++ b/crates/cow-venue/data/classification.toml @@ -18,6 +18,12 @@ # watch for `backoff-seconds` before the next # attempt. `backoff-seconds` is required and # must be at least 1. +# action = "drop-on-repeat" Permanent unless first-seen: the first +# rejection retries on the next block (a +# same-block state race can make a valid +# order verify late); a repeat on a later +# block removes the watch. +# # action = "drop" Permanent: no retry can succeed. Remove the # watch and its gates. # @@ -41,7 +47,7 @@ # (a permanent-looking contract rejection is dropped rather than # retried, and the limit-order backoff is shorter) are exactly: # -# InvalidEip1271Signature drop upstream: retry next block +# InvalidEip1271Signature drop-on-repeat upstream: retry next block # InsufficientBalance drop upstream: backoff 10 min # InsufficientAllowance drop upstream: backoff 10 min # InvalidAppData drop upstream: backoff 60 s @@ -85,6 +91,17 @@ error-type = "DuplicateOrder" action = "try-next-block" already-submitted = true +# --- One-block grace: retry once, then drop -------------------------- + +# A first-time user's Safe wiring and conditional-order registration +# can land in the same block as the indexed event; an orderbook node +# verifying against its own head then rejects a signature that is valid +# one block later. The first rejection retries next block; a repeat on +# a later block is a genuinely broken signature and drops the watch. +[[entry]] +error-type = "InvalidEip1271Signature" +action = "drop-on-repeat" + # --- Permanent: drop the watch --------------------------------------- # # Listed for documentation; each is also the default for any unlisted @@ -95,10 +112,6 @@ already-submitted = true error-type = "InvalidSignature" action = "drop" -[[entry]] -error-type = "InvalidEip1271Signature" -action = "drop" - [[entry]] error-type = "WrongOwner" action = "drop" diff --git a/crates/cow-venue/src/adapter.rs b/crates/cow-venue/src/adapter.rs index 333475e7..693154ee 100644 --- a/crates/cow-venue/src/adapter.rs +++ b/crates/cow-venue/src/adapter.rs @@ -420,6 +420,9 @@ fn classified(api: &ApiError) -> VenueError { RetryAction::Backoff { seconds } => VenueError::RateLimited(RateLimit { retry_after_ms: Some(seconds.saturating_mul(1000)), }), + // The one-shot grace is client-side; the wire stays `denied`, + // and the errorType prefix in the detail carries it across. + RetryAction::DropOnRepeat => VenueError::Denied(detail), RetryAction::Drop => VenueError::Denied(detail), _ => VenueError::Denied(detail), } @@ -758,6 +761,15 @@ mod tests { Err(VenueError::Denied(detail)) if detail.contains("InvalidSignature") )); + // A drop-on-repeat row stays `denied` on the wire; the + // errorType prefix carries the one-shot grace to the client. + reject(&fetch, "InvalidEip1271Signature"); + assert!(matches!( + submit_with(&fetch, &config, &signed_bytes()), + Err(VenueError::Denied(detail)) + if detail.starts_with("InvalidEip1271Signature:") + )); + reject(&fetch, "TooManyLimitOrders"); assert!(matches!( submit_with(&fetch, &config, &signed_bytes()), diff --git a/crates/cow-venue/src/classification.rs b/crates/cow-venue/src/classification.rs index aabda91f..a3c6861b 100644 --- a/crates/cow-venue/src/classification.rs +++ b/crates/cow-venue/src/classification.rs @@ -32,6 +32,7 @@ pub const CLASSIFICATION_TOML: &str = include_str!("../data/classification.toml" enum GenAction { TryNextBlock, Backoff, + DropOnRepeat, Drop, } @@ -53,6 +54,7 @@ impl GeneratedRow { GenAction::Backoff => RetryAction::Backoff { seconds: self.backoff_seconds.max(1), }, + GenAction::DropOnRepeat => RetryAction::DropOnRepeat, GenAction::Drop => RetryAction::Drop, } } @@ -116,6 +118,19 @@ pub fn is_already_submitted(error_type: &str) -> bool { table().is_already_submitted(error_type) } +/// Retry action for a coarse `denied` refusal. The adapter spells a +/// classified rejection as `{errorType}: {description}`; the prefix +/// re-enters the table so a [`RetryAction::DropOnRepeat`] row survives +/// the collapse to the wire `venue-error`. Every other denial is +/// permanent. +pub fn classify_denied(detail: &str) -> RetryAction { + let error_type = detail.split_once(':').map_or(detail, |(prefix, _)| prefix); + match classify(error_type) { + RetryAction::DropOnRepeat => RetryAction::DropOnRepeat, + _ => RetryAction::Drop, + } +} + #[cfg(test)] mod tests { use super::*; @@ -141,6 +156,7 @@ mod tests { Action::Backoff => RetryAction::Backoff { seconds: entry.backoff_seconds.max(1), }, + Action::DropOnRepeat => RetryAction::DropOnRepeat, Action::Drop => RetryAction::Drop, }; assert_eq!( @@ -168,6 +184,10 @@ mod tests { classify("TooManyLimitOrders"), RetryAction::Backoff { seconds: 30 }, ); + assert_eq!( + classify("InvalidEip1271Signature"), + RetryAction::DropOnRepeat, + ); assert_eq!(classify("InvalidSignature"), RetryAction::Drop); assert!(is_already_submitted("DuplicatedOrder")); assert!(is_already_submitted("DuplicateOrder")); @@ -181,7 +201,7 @@ mod tests { assert!(!is_already_submitted("NewlyMintedErrorType")); } - /// All three retry arms are reachable from the table alone. + /// Every retry arm is reachable from the table alone. #[test] fn table_reaches_every_arm() { assert_eq!(classify("InsufficientFee"), RetryAction::TryNextBlock); @@ -189,9 +209,34 @@ mod tests { classify("TooManyLimitOrders"), RetryAction::Backoff { .. } )); + assert_eq!( + classify("InvalidEip1271Signature"), + RetryAction::DropOnRepeat, + ); assert_eq!(classify("InvalidSignature"), RetryAction::Drop); } + /// A denied detail re-enters the table by its `errorType` prefix: + /// only a drop-on-repeat row escapes the permanent default, and a + /// transient row cannot be smuggled through a denial. + #[test] + fn denied_detail_refines_by_error_type_prefix() { + assert_eq!( + classify_denied("InvalidEip1271Signature: signature is not valid"), + RetryAction::DropOnRepeat, + ); + assert_eq!( + classify_denied("InvalidSignature: bad sig"), + RetryAction::Drop, + ); + assert_eq!( + classify_denied("InsufficientFee: too low"), + RetryAction::Drop, + ); + assert_eq!(classify_denied("policy refusal"), RetryAction::Drop); + assert_eq!(classify_denied(""), RetryAction::Drop); + } + #[test] fn duplicate_type_is_rejected() { let toml = r#" diff --git a/crates/cow-venue/src/classification_data.rs b/crates/cow-venue/src/classification_data.rs index 49162e0b..bb96cc3f 100644 --- a/crates/cow-venue/src/classification_data.rs +++ b/crates/cow-venue/src/classification_data.rs @@ -10,7 +10,7 @@ use serde::Deserialize; -/// One of the three retry actions an `errorType` maps to on the wire. +/// One of the retry actions an `errorType` maps to on the wire. #[derive(Clone, Copy, Debug, Eq, PartialEq, Deserialize)] #[serde(rename_all = "kebab-case")] pub enum Action { @@ -18,6 +18,9 @@ pub enum Action { TryNextBlock, /// Throttle: gate the watch for `backoff_seconds` before retrying. Backoff, + /// Permanent unless first-seen: retry on the next block once, then + /// drop on a repeat at a later block. + DropOnRepeat, /// Permanent: remove the watch and its gates. Drop, } diff --git a/crates/cow-venue/src/lib.rs b/crates/cow-venue/src/lib.rs index 4d31bbe2..990c4363 100644 --- a/crates/cow-venue/src/lib.rs +++ b/crates/cow-venue/src/lib.rs @@ -81,6 +81,6 @@ pub use order::{ pub use adapter::CowAdapter; #[cfg(feature = "client")] -pub use classification::{ClassificationTable, classify, is_already_submitted}; +pub use classification::{ClassificationTable, classify, classify_denied, is_already_submitted}; #[cfg(feature = "client")] pub use client::{CowClient, CowVenue, intent_id}; diff --git a/crates/nexum-sdk/src/keeper.rs b/crates/nexum-sdk/src/keeper.rs index 637e7985..726bdae4 100644 --- a/crates/nexum-sdk/src/keeper.rs +++ b/crates/nexum-sdk/src/keeper.rs @@ -26,10 +26,10 @@ //! - [`Retrier`] - runs a [`RetryAction`]'s effect through the //! stores after a failed keeper run attempt. //! -//! [`WatchRef`] ties the first two together: gate keys are derived -//! from the exact hex substrings of the stored watch key, and -//! [`WatchSet::remove`] drops a watch together with all of its gate -//! keys so no failure path can orphan a gate. +//! [`WatchRef`] ties the first two together: gate and marker keys are +//! derived from the exact hex substrings of the stored watch key, and +//! [`WatchSet::remove`] drops a watch together with all of its derived +//! keys so no failure path can orphan one. //! //! ``` //! use nexum_sdk::keeper::{Gates, Journal, WatchRef, WatchSet}; @@ -99,6 +99,10 @@ pub const SUBMITTED_PREFIX: &str = "submitted:"; /// the observe-and-verify path (e.g. ethflow) recorded an existing /// upstream order as seen. pub const OBSERVED_PREFIX: &str = "observed:"; +/// Prefix of the first-refusal marker paired with a watch: the block a +/// [`RetryAction::DropOnRepeat`] refusal was first seen at, u64 +/// little-endian. +pub const REFUSED_PREFIX: &str = "refused:"; /// Canonical watch key for an owner / commitment-hash pair (lowercase /// `0x`-prefixed hex on both halves). Free-standing because the key @@ -159,6 +163,11 @@ impl<'k> WatchRef<'k> { pub fn next_epoch_key(&self) -> String { format!("{NEXT_EPOCH_PREFIX}{}:{}", self.owner_hex, self.hash_hex) } + + /// The `refused:` first-refusal marker key paired with this watch. + pub fn refused_key(&self) -> String { + format!("{REFUSED_PREFIX}{}:{}", self.owner_hex, self.hash_hex) + } } /// Watch-set registry: one row per conditional commitment, keyed @@ -199,11 +208,13 @@ impl<'h, H: LocalStoreHost> WatchSet<'h, H> { self.host.list_keys(WATCH_PREFIX) } - /// Drop the watch together with both of its gate keys. Gates go - /// first: a fault part-way leaves the watch row behind so a retry - /// re-drops it, and a gate key can never outlive its watch. + /// Drop the watch together with its gate and refusal keys. The + /// derived keys go first: a fault part-way leaves the watch row + /// behind so a retry re-drops it, and a derived key can never + /// outlive its watch. pub fn remove(&self, watch: WatchRef<'_>) -> Result<(), Fault> { Gates::new(self.host).clear(watch)?; + self.host.delete(&watch.refused_key())?; self.host.delete(&watch.key()) } } @@ -238,12 +249,12 @@ impl<'h, H: LocalStoreHost> Gates<'h, H> { /// inclusive at its threshold. #[must_use = "the readiness verdict gates the poll; `?` alone drops the inner bool"] pub fn is_ready(&self, watch: WatchRef<'_>, block: u64, epoch_s: u64) -> Result { - if let Some(next) = self.read_u64(&watch.next_block_key())? + if let Some(next) = read_u64(self.host, &watch.next_block_key())? && block < next { return Ok(false); } - if let Some(next) = self.read_u64(&watch.next_epoch_key())? + if let Some(next) = read_u64(self.host, &watch.next_epoch_key())? && epoch_s < next { return Ok(false); @@ -256,21 +267,22 @@ impl<'h, H: LocalStoreHost> Gates<'h, H> { self.host.delete(&watch.next_block_key())?; self.host.delete(&watch.next_epoch_key()) } +} - fn read_u64(&self, key: &str) -> Result, Fault> { - // Absent key: silently no gate. Present but wrong length: the - // value is corrupt, so warn before falling open to no gate - - // fail-open is deliberate (a corrupt value can only make the - // watch poll sooner), but it must not pass unobserved. - let Some(b) = self.host.get(key)? else { - return Ok(None); - }; - match <[u8; 8]>::try_from(b.as_slice()) { - Ok(bytes) => Ok(Some(u64::from_le_bytes(bytes))), - Err(_) => { - tracing::warn!(%key, len = b.len(), "gate value corrupt; treating as absent"); - Ok(None) - } +/// Little-endian u64 row read shared by the gate and refusal-marker +/// keys. Absent key: silently `None`. Present but wrong length: the +/// value is corrupt, so warn before falling open to `None` - fail-open +/// is deliberate (a corrupt value can only make the watch act sooner, +/// never wedge it), but it must not pass unobserved. +fn read_u64(host: &H, key: &str) -> Result, Fault> { + let Some(b) = host.get(key)? else { + return Ok(None); + }; + match <[u8; 8]>::try_from(b.as_slice()) { + Ok(bytes) => Ok(Some(u64::from_le_bytes(bytes))), + Err(_) => { + tracing::warn!(%key, len = b.len(), "stored value corrupt; treating as absent"); + Ok(None) } } } @@ -373,14 +385,20 @@ pub enum RetryAction { /// Seconds to wait before retrying. seconds: u64, }, + /// Grant one next-block retry: the first application records the + /// block and gates the watch to the next one; a repeat at a later + /// block removes the watch. + DropOnRepeat, /// Remove the watch and its gates; no retry can succeed. Drop, } /// Retry ledger: runs a [`RetryAction`]'s effect through the keeper /// stores. `Backoff` saturates at `u64::MAX` on the epoch clock; -/// `Drop` delegates to [`WatchSet::remove`], so gates go first and no -/// failure path can orphan one. +/// `DropOnRepeat` keeps a `refused:` block marker so only a repeat on +/// a later block removes the watch; `Drop` delegates to +/// [`WatchSet::remove`], so derived keys go first and no failure path +/// can orphan one. pub struct Retrier<'h, H> { host: &'h H, } @@ -391,20 +409,39 @@ impl<'h, H: LocalStoreHost> Retrier<'h, H> { Self { host } } - /// Apply `action` to the watch, with `now_epoch_s` as the backoff - /// origin. + /// Apply `action` to the watch at `tick`: the backoff origin and + /// the repeat-detection block both read from it. pub fn apply( &self, watch: WatchRef<'_>, action: RetryAction, - now_epoch_s: u64, + tick: &Tick, ) -> Result<(), Fault> { match action { RetryAction::TryNextBlock => Ok(()), RetryAction::Backoff { seconds } => { - Gates::new(self.host).set_next_epoch(watch, now_epoch_s.saturating_add(seconds)) + Gates::new(self.host).set_next_epoch(watch, tick.epoch_s.saturating_add(seconds)) + } + RetryAction::DropOnRepeat => { + let key = watch.refused_key(); + match read_u64(self.host, &key)? { + Some(block) if tick.block > block => WatchSet::new(self.host).remove(watch), + Some(_) => Ok(()), + None => { + self.host.set(&key, &tick.block.to_le_bytes())?; + Gates::new(self.host).set_next_block(watch, tick.block.saturating_add(1)) + } + } } RetryAction::Drop => WatchSet::new(self.host).remove(watch), } } + + /// Clear the watch's first-refusal marker: an accepted submit ends + /// the refusal episode, so a later [`RetryAction::DropOnRepeat`] + /// refusal earns a fresh one-block grace. No-op when no marker is + /// set. + pub fn clear_refusal(&self, watch: WatchRef<'_>) -> Result<(), Fault> { + self.host.delete(&watch.refused_key()) + } } diff --git a/crates/nexum-sdk/tests/keeper.rs b/crates/nexum-sdk/tests/keeper.rs index e271fa41..f73c9adf 100644 --- a/crates/nexum-sdk/tests/keeper.rs +++ b/crates/nexum-sdk/tests/keeper.rs @@ -8,8 +8,8 @@ use alloy_primitives::{Address, B256, address, b256}; use nexum_sdk::host::{Fault, LocalStoreHost as _}; use nexum_sdk::keeper::{ - ConditionalSource, Gates, Journal, NEXT_BLOCK_PREFIX, NEXT_EPOCH_PREFIX, Retrier, RetryAction, - Tick, WATCH_PREFIX, WatchRef, WatchSet, watch_key, + ConditionalSource, Gates, Journal, NEXT_BLOCK_PREFIX, NEXT_EPOCH_PREFIX, REFUSED_PREFIX, + Retrier, RetryAction, Tick, WATCH_PREFIX, WatchRef, WatchSet, watch_key, }; use nexum_sdk_test::MockHost; @@ -362,6 +362,14 @@ fn seeded_watch(host: &MockHost) -> String { .unwrap() } +fn tick_at(block: u64, epoch_s: u64) -> Tick { + Tick { + chain_id: 1, + block, + epoch_s, + } +} + #[test] fn ledger_try_next_block_leaves_the_store_untouched() { let host = MockHost::new(); @@ -372,7 +380,7 @@ fn ledger_try_next_block_leaves_the_store_untouched() { .apply( WatchRef::parse(&key).unwrap(), RetryAction::TryNextBlock, - 1_000, + &tick_at(100, 1_000), ) .unwrap(); @@ -387,7 +395,11 @@ fn ledger_backoff_gates_the_watch_on_the_epoch_clock() { let ledger = Retrier::new(&host); ledger - .apply(watch, RetryAction::Backoff { seconds: 30 }, 1_000) + .apply( + watch, + RetryAction::Backoff { seconds: 30 }, + &tick_at(100, 1_000), + ) .unwrap(); let gates = Gates::new(&host); @@ -410,7 +422,11 @@ fn ledger_backoff_saturates_on_the_epoch_clock() { let watch = WatchRef::parse(&key).unwrap(); Retrier::new(&host) - .apply(watch, RetryAction::Backoff { seconds: u64::MAX }, 1_000) + .apply( + watch, + RetryAction::Backoff { seconds: u64::MAX }, + &tick_at(100, 1_000), + ) .unwrap(); assert_eq!( @@ -427,17 +443,109 @@ fn ledger_drop_removes_the_watch_and_its_gates() { Gates::new(&host).set_next_block(watch, 500).unwrap(); Retrier::new(&host) - .apply(watch, RetryAction::Drop, 1_000) + .apply(watch, RetryAction::Drop, &tick_at(100, 1_000)) .unwrap(); assert!(host.store.is_empty(), "watch and gates must go"); } +#[test] +fn ledger_drop_on_repeat_grants_one_next_block_retry() { + let host = MockHost::new(); + let key = seeded_watch(&host); + let watch = WatchRef::parse(&key).unwrap(); + let ledger = Retrier::new(&host); + + // First refusal: the block is recorded and the watch gates to the + // next block instead of dropping. + ledger + .apply(watch, RetryAction::DropOnRepeat, &tick_at(100, 1_000)) + .unwrap(); + let snapshot = host.store.snapshot(); + assert!(snapshot.contains_key(&key), "first refusal keeps the watch"); + assert_eq!( + snapshot.get(&watch.refused_key()).unwrap(), + &100_u64.to_le_bytes().to_vec(), + ); + assert_eq!( + snapshot.get(&watch.next_block_key()).unwrap(), + &101_u64.to_le_bytes().to_vec(), + ); + + // A repeat at the same block leaves the store untouched. + let before = host.store.snapshot(); + ledger + .apply(watch, RetryAction::DropOnRepeat, &tick_at(100, 1_000)) + .unwrap(); + assert_eq!(host.store.snapshot(), before); + + // A repeat on a later block removes the watch and every derived key. + ledger + .apply(watch, RetryAction::DropOnRepeat, &tick_at(101, 1_012)) + .unwrap(); + assert!(host.store.is_empty(), "watch, gates, and marker must go"); +} + +#[test] +fn ledger_clear_refusal_resets_the_one_block_grace() { + let host = MockHost::new(); + let key = seeded_watch(&host); + let watch = WatchRef::parse(&key).unwrap(); + let ledger = Retrier::new(&host); + + ledger + .apply(watch, RetryAction::DropOnRepeat, &tick_at(100, 1_000)) + .unwrap(); + ledger.clear_refusal(watch).unwrap(); + assert!(!host.store.snapshot().contains_key(&watch.refused_key())); + + // A refusal at a later block after the clear is a fresh first + // refusal: the watch survives and the marker records the new block. + ledger + .apply(watch, RetryAction::DropOnRepeat, &tick_at(105, 1_060)) + .unwrap(); + let snapshot = host.store.snapshot(); + assert!(snapshot.contains_key(&key), "the watch must survive"); + assert_eq!( + snapshot.get(&watch.refused_key()).unwrap(), + &105_u64.to_le_bytes().to_vec(), + ); + assert_eq!( + snapshot.get(&watch.next_block_key()).unwrap(), + &106_u64.to_le_bytes().to_vec(), + ); +} + +#[test] +fn ledger_drop_removes_the_refused_marker() { + let host = MockHost::new(); + let key = seeded_watch(&host); + let watch = WatchRef::parse(&key).unwrap(); + + let ledger = Retrier::new(&host); + ledger + .apply(watch, RetryAction::DropOnRepeat, &tick_at(100, 1_000)) + .unwrap(); + ledger + .apply(watch, RetryAction::Drop, &tick_at(100, 1_000)) + .unwrap(); + + assert!( + !host + .store + .snapshot() + .keys() + .any(|k| k.starts_with(REFUSED_PREFIX)), + "drop must not orphan the refusal marker", + ); +} + #[test] fn retry_action_labels_are_stable_snake_case() { - let cases: [(RetryAction, &str); 3] = [ + let cases: [(RetryAction, &str); 4] = [ (RetryAction::TryNextBlock, "try_next_block"), (RetryAction::Backoff { seconds: 1 }, "backoff"), + (RetryAction::DropOnRepeat, "drop_on_repeat"), (RetryAction::Drop, "drop"), ]; for (action, label) in cases { diff --git a/crates/videre-sdk/Cargo.toml b/crates/videre-sdk/Cargo.toml index 5ccad1f2..0c87c63b 100644 --- a/crates/videre-sdk/Cargo.toml +++ b/crates/videre-sdk/Cargo.toml @@ -36,6 +36,8 @@ videre-status-body = { path = "../videre-status-body" } http.workspace = true strum.workspace = true thiserror.workspace = true +# Best-effort fault logs on the sweep's non-critical store cleanups. +tracing.workspace = true wit-bindgen.workspace = true [dev-dependencies] diff --git a/crates/videre-sdk/src/keeper.rs b/crates/videre-sdk/src/keeper.rs index 8ca91a3e..732a271a 100644 --- a/crates/videre-sdk/src/keeper.rs +++ b/crates/videre-sdk/src/keeper.rs @@ -65,10 +65,14 @@ impl Keeper { /// run every other outcome and every venue refusal through the /// [`Retrier`]. The [`submission_key`] is checked against the /// `submitted:` [`Journal`] before every submit and recorded on - /// acceptance, so an accepted body never reaches the venue twice; - /// a `requires-signing` answer journals nothing and is surfaced - /// afresh each sweep. Store faults abort the sweep; venue refusals - /// never do - they fold into per-watch retry actions. + /// acceptance - the watch's first-refusal marker is then cleared + /// best-effort - so a journalled acceptance is never resubmitted. + /// The record is not atomic with the submit: an acceptance whose + /// journal write faults can still resubmit. A `requires-signing` + /// answer journals nothing and is surfaced afresh each sweep. + /// Store faults abort the sweep, bar the post-acceptance marker + /// clear; venue refusals never do - they fold into per-watch + /// retry actions. pub async fn sweep(&self, host: &H, tick: &Tick) -> Result where H: LocalStoreHost, @@ -105,6 +109,15 @@ impl Keeper { match self.venues.submit(&self.venue, body).await { Ok(SubmitOutcome::Accepted(_)) => { journal.record(&key)?; + // The acceptance is journalled; the marker + // clear is cleanup and must not abort the + // sweep. + if let Err(fault) = retrier.clear_refusal(watch) { + tracing::error!( + %fault, + "refusal-marker clear failed after journalled acceptance", + ); + } report.submitted += 1; continue; } @@ -123,7 +136,7 @@ impl Keeper { RetryAction::Drop => report.dropped += 1, _ => report.retried += 1, } - retrier.apply(watch, action, tick.epoch_s)?; + retrier.apply(watch, action, tick)?; } Ok(report) } @@ -191,6 +204,7 @@ pub fn retry_action(fault: &VenueFault) -> RetryAction { mod tests { use std::cell::RefCell; + use nexum_sdk::host::{Fault, LocalStoreHost as _}; use nexum_sdk::keeper::{Gates, Journal, Tick, WatchRef, WatchSet}; use nexum_sdk::prelude::{Address, B256, hex, keccak256}; use nexum_sdk_test::MockLocalStore; @@ -310,6 +324,45 @@ mod tests { assert_eq!(venue.submitted.borrow().len(), 1); } + #[test] + fn accepted_body_clears_the_refusal_marker() { + let host = MockLocalStore::default(); + let key = put_watch(&host); + let watch = WatchRef::parse(&key).expect("well-formed key"); + host.set(&watch.refused_key(), &50_u64.to_le_bytes()) + .expect("marker writes"); + let venue = StubVenue::new(Ok(SubmitOutcome::Accepted(vec![1]))); + + let report = run(keeper(Sweep::Submit(b"body".to_vec()), &venue).sweep(&host, &TICK)) + .expect("sweep runs"); + assert_eq!(report.submitted, 1); + assert!( + host.get(&watch.refused_key()) + .expect("marker reads") + .is_none(), + "acceptance must clear the first-refusal marker", + ); + } + + #[test] + fn refusal_marker_clear_fault_does_not_abort_a_journalled_acceptance() { + let host = MockLocalStore::default(); + put_watch(&host); + host.fail_on("refused:", Fault::Unavailable("store down".into())); + let venue = StubVenue::new(Ok(SubmitOutcome::Accepted(vec![1]))); + + let report = run(keeper(Sweep::Submit(b"body".to_vec()), &venue).sweep(&host, &TICK)) + .expect("marker-clear fault must not abort the sweep"); + assert_eq!(report.submitted, 1); + let key = format!("stub:{}", hex::encode_prefixed(keccak256(b"body"))); + assert!( + Journal::submitted(&host) + .contains(&key) + .expect("journal reads"), + "acceptance must be journalled before the marker clear", + ); + } + #[test] fn a_changed_body_submits_afresh() { let host = MockLocalStore::default();