twap: grant one next-block retry before dropping on InvalidEip1271Signature

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.

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<SubmitOutcome, VenueFault> {
+    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.

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!(

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"

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()),