diff --git a/crates/composable-cow/src/run.rs b/crates/composable-cow/src/run.rs index a3ee56f2..ac90f64a 100644 --- a/crates/composable-cow/src/run.rs +++ b/crates/composable-cow/src/run.rs @@ -1,14 +1,17 @@ //! Keeper run: the poll-loop composition conditional- //! commitment modules share. //! -//! [`run`] walks the keeper watch set, polls each gate-ready -//! watch through a [`Poller`], and runs the +//! [`run`] first drives the shared +//! [`reconcile`](videre_sdk::reconcile) pass over the `submitted:` +//! reserve/commit journal, then walks the keeper watch set, polls each +//! gate-ready watch through a [`Poller`], and runs the //! [`Verdict`]'s effect: lifecycle outcomes update the gate and -//! watch stores, `Post` drives one submission through the typed -//! [`CowClient`] onto the `videre:venue/client` seam with the -//! `submitted:` journal as the idempotency guard - keyed on the -//! venue-and-body [`intent_id`] - and the keeper [`Retrier`] -//! as the failure dispatch. +//! watch stores, `Post` reserves the encoded body on the venue-and-body +//! submission key and drives one submission through the typed +//! [`CowClient`] onto the `videre:venue/client` seam, committing on +//! acceptance, with the keeper [`Retrier`] as the failure dispatch. A +//! reservation whose submit outcome is lost is resubmitted by the next +//! tick's reconcile pass, never dropped. //! //! Store faults abort the run (the next tick replays it); //! submission failures never do - they fold into a @@ -23,27 +26,55 @@ 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, classify_denied, intent_id}; +use cow_venue::{CowClient, CowIntent, CowIntentBody, CowVenue, SignedOrder, classify_denied}; use cowprotocol::GPv2OrderData; use nexum_sdk::host::{Fault, LocalStoreHost}; -use nexum_sdk::keeper::{Gates, Journal, Poller, Retrier, RetryAction, Tick, WatchRef, WatchSet}; +use nexum_sdk::keeper::{ + Gates, Journal, Mark, Poller, Retrier, RetryAction, Tick, WatchRef, WatchSet, +}; use std::task::Poll; use videre_sdk::client::poll_once; -use videre_sdk::keeper::retry_action; -use videre_sdk::{ClientError, SubmitOutcome, VenueFault, VenueTransport}; +use videre_sdk::keeper::{retry_action, submission_key}; +use videre_sdk::{ + ClientError, IntentBody as _, SubmitOutcome, Venue as _, VenueFault, VenueTransport, +}; use crate::Verdict; /// Poll every gate-ready watch once at `tick` and run each outcome's -/// effect. One source poll per ready watch; a `Post` outcome makes at -/// most one venue submit through `venue`. +/// effect. The top-of-sweep [`reconcile`](videre_sdk::reconcile) pass +/// resolves stranded reservations first, then one source poll per ready +/// watch; a `Post` outcome makes at most one venue submit through +/// `venue`. pub fn run(host: &H, venue: &CowClient, source: &S, tick: &Tick) -> Result<(), Fault> where H: LocalStoreHost, S: Poller, T: VenueTransport, { + // Resolve any stranded reservation before polling fresh watches, so a + // submit whose outcome was lost is resubmitted, never dropped (#572). + // The helper is async and the guest boundary synchronous, so drive it + // with `poll_once`. + let journal = Journal::submitted(host); + match poll_once(videre_sdk::reconcile( + &CowVenue::ID, + venue.transport(), + &journal, + tick, + videre_sdk::DEFAULT_RECONCILE_BUDGET, + )) { + Poll::Ready(res) => { + res?; + } + Poll::Pending => { + // A misbehaving guest transport suspended; leave the RESERVED + // markers for the next tick rather than dropping them. + tracing::error!("cow reconcile suspended; skipping this tick"); + } + } + let watches = WatchSet::new(host); let gates = Gates::new(host); for key in watches.list()? { @@ -79,14 +110,16 @@ where Ok(()) } -/// Submit one freshly-polled `Ready` order through the typed client, -/// guarding on the `submitted:` journal and dispatching any venue +/// Submit one freshly-polled `Ready` order through the typed client on +/// the `submitted:` reserve/commit journal, dispatching any venue /// refusal through the retry ledger. /// -/// The journal keys on the deterministic venue-and-body [`intent_id`], -/// derived before any network work from the same body bytes the venue -/// submit carries, so the guard is independent of where assembly -/// happens. The venue's receipt rides the log only. +/// The journal keys on the deterministic venue-and-body submission key. +/// A `COMMITTED` marker is an idempotent skip; a `RESERVED` marker is +/// owned by this tick's reconcile pass and never re-submitted here. A +/// fresh order reserves its encoded body before the submit and commits +/// on acceptance; release runs only on a known synchronous non-accept, +/// never on a pending or accepted path. fn submit_ready( host: &H, venue: &CowClient, @@ -123,56 +156,76 @@ where owner: owner.into_array(), signature: signature.to_vec(), })); - let intent_id = match intent_id(&intent) { - Ok(id) => id, + // Reserve the exact wire bytes the venue submit and the reconcile + // resubmit both carry, so the id and the reservation agree. + let encoded = match intent.to_bytes() { + Ok(bytes) => bytes, Err(err) => { tracing::error!("intent body encode failed: {err}"); return Ok(()); } }; + let intent_id = submission_key(&CowVenue::ID, &encoded); let journal = Journal::submitted(host); - if journal.contains(&intent_id)? { - tracing::info!("{label} {intent_id} already submitted; skipping re-submit"); - return Ok(()); + match journal.mark(&intent_id)? { + Some(Mark::Committed) => { + tracing::info!("{label} {intent_id} already committed; skipping re-submit"); + return Ok(()); + } + Some(Mark::Reserved) => { + // Owned by this tick's reconcile pass; never a second submit. + tracing::info!("{label} {intent_id} reserved; reconcile owns it"); + return Ok(()); + } + None => {} } + // Reserve the real body before any network work: a crash or lost + // outcome now strands a RESERVED marker the next tick's reconcile + // resolves, never a silent drop (#572). + journal.reserve(&intent_id, &encoded)?; let Poll::Ready(outcome) = poll_once(venue.submit(&intent)) else { // Guest transports never suspend; a pending future means a - // foreign transport misbehaved. Route through the retrier for - // symmetry with the venue-refusal arm; a next-block retry keeps - // the watch for the next tick. + // foreign transport misbehaved. Leave the marker RESERVED for the + // next tick's reconcile and retry the watch next block; never + // release on a pending path. tracing::error!("{label} submit future suspended; retrying next block"); return Retrier::new(host).apply(watch, RetryAction::TryNextBlock, tick); }; match outcome { Ok(SubmitOutcome::Accepted(receipt)) => { + // The submit landed; commit the reservation best-effort. A + // commit fault leaves the marker RESERVED for reconcile, never + // released or aborted. + if let Err(fault) = journal.commit(&intent_id) { + tracing::error!("submitted {intent_id} but commit write failed: {fault}"); + } // An acceptance ends any refusal episode: clear the - // first-refusal marker so a later independent refusal - // earns a fresh one-block grace. + // 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 run or unwind the accepted order. - // Log and carry on - the already-submitted arm keeps the - // next tick's re-post idempotent. - if let Err(fault) = journal.record(&intent_id) { - tracing::error!("submitted {intent_id} but journal write failed: {fault}"); - } tracing::info!( "submitted {intent_id} (receipt {})", hex::encode_prefixed(&receipt), ); } Ok(SubmitOutcome::RequiresSigning(_)) => { - // A run cannot sign; nothing is journalled, so the next - // tick surfaces the same ask afresh. + // A known non-accept: a run cannot sign, so release the reserve + // and re-pose the ask next tick. + journal.release(&intent_id)?; tracing::warn!("{label} submit for {owner:#x} requires signing; not journalled"); } Err(ClientError::Body(err)) => { + // A known non-accept before any order is placed: release. + journal.release(&intent_id)?; tracing::error!("intent body encode failed: {err}"); } Err(ClientError::Venue(fault)) => { + // A known venue refusal: release the reserve, then fold it + // through the ledger. + journal.release(&intent_id)?; let action = match &fault { VenueFault::Denied(detail) => classify_denied(detail), other => retry_action(other), @@ -193,8 +246,9 @@ where } } } - // `ClientError` is non-exhaustive; a future case leaves the - // watch for the next tick. + // `ClientError` is non-exhaustive; an unknown outcome is neither a + // known accept nor a known refusal, so leave the marker RESERVED + // for reconcile rather than releasing. Err(err) => tracing::error!("submit failed: {err}"), } Ok(()) diff --git a/crates/composable-cow/tests/run.rs b/crates/composable-cow/tests/run.rs index a466e14b..e3bf7ba1 100644 --- a/crates/composable-cow/tests/run.rs +++ b/crates/composable-cow/tests/run.rs @@ -2,16 +2,16 @@ //! scripted venue transport on the `videre:venue/client` seam. use std::cell::{Cell, RefCell}; -use std::collections::VecDeque; +use std::collections::{HashSet, VecDeque}; use alloy_primitives::{Address, B256, U256, address, hex, keccak256}; use composable_cow::{Verdict, run}; use cow_venue::assembly::{gpv2_to_order_data, order_data_to_body}; use cow_venue::{CowClient, CowIntent, CowIntentBody, CowVenue, SignedOrder}; use cowprotocol::{BuyTokenDestination, GPv2OrderData, OrderKind, SellTokenSource}; -use nexum_sdk::host::LocalStoreHost as _; -use nexum_sdk::keeper::{Gates, Journal, Poller, Tick, WatchRef, WatchSet}; -use nexum_sdk_test::{MockHost, capture_tracing}; +use nexum_sdk::host::{Fault, LocalStoreHost}; +use nexum_sdk::keeper::{Gates, Journal, Mark, Poller, Tick, WatchRef, WatchSet}; +use nexum_sdk_test::{MockHost, MockLocalStore, capture_tracing}; use videre_sdk::client::sealed::SealedTransport; use videre_sdk::keeper::submission_key; use videre_sdk::{ @@ -733,3 +733,289 @@ fn ready_submits_the_encoded_intent_body_through_the_venue_seam() { "the journal keys on the generic submission key", ); } + +// ---- #573 reserve/commit + top-of-sweep reconcile ---- + +/// Venue that models the CoW re-POST floor: a body it already holds +/// re-accepts (the `DuplicatedOrder` -> AlreadyHeld -> Accepted fold), +/// so a reconcile resubmit is always safe. A fresh body gets the +/// programmed outcome; an accepted body joins the held set. Every POST +/// is recorded. +struct HoldingVenue { + outcome: RefCell>, + posts: RefCell>>, + held: RefCell>>, +} + +impl HoldingVenue { + fn new(outcome: Result) -> Self { + Self { + outcome: RefCell::new(outcome), + posts: RefCell::new(Vec::new()), + held: RefCell::new(HashSet::new()), + } + } + + fn accepting() -> Self { + Self::new(Ok(SubmitOutcome::Accepted(vec![0xAB]))) + } + + fn posts(&self) -> Vec> { + self.posts.borrow().clone() + } + + fn post_count(&self) -> usize { + self.posts.borrow().len() + } + + fn held_count(&self) -> usize { + self.held.borrow().len() + } + + /// Pre-seed a held body: a POST the venue received before the caller + /// lost its outcome. + fn preload(&self, body: &[u8]) { + self.held.borrow_mut().insert(body.to_vec()); + } +} + +impl SealedTransport for &HoldingVenue {} + +impl VenueTransport for &HoldingVenue { + async fn quote(&self, _venue: &VenueId, _body: Vec) -> Result { + unreachable!("quote not exercised") + } + + async fn submit(&self, _venue: &VenueId, body: Vec) -> Result { + self.posts.borrow_mut().push(body.clone()); + if self.held.borrow().contains(&body) { + return Ok(SubmitOutcome::Accepted(vec![0xAB])); + } + let outcome = self.outcome.borrow().clone(); + if let Ok(SubmitOutcome::Accepted(_)) = &outcome { + self.held.borrow_mut().insert(body); + } + outcome + } + + async fn status(&self, _venue: &VenueId, _receipt: &[u8]) -> Result { + unreachable!("status not exercised") + } + + async fn cancel(&self, _venue: &VenueId, _receipt: &[u8]) -> Result<(), VenueFault> { + unreachable!("cancel not exercised") + } +} + +fn holding_client(venue: &HoldingVenue) -> CowClient<&HoldingVenue> { + CowClient::with_transport(venue) +} + +/// Wraps a store, faulting the first `COMMITTED` write to `submitted:` +/// once, then delegating. Models an accepted submit whose commit write +/// faults: the `RESERVED` marker persists and no release runs. +struct FlakyCommit { + inner: MockLocalStore, + arm: Cell, +} + +impl FlakyCommit { + fn new() -> Self { + Self { + inner: MockLocalStore::default(), + arm: Cell::new(true), + } + } +} + +impl LocalStoreHost for FlakyCommit { + fn get(&self, key: &str) -> Result>, Fault> { + self.inner.get(key) + } + + fn set(&self, key: &str, value: &[u8]) -> Result<(), Fault> { + // 0x02 is the journal COMMITTED tag. + if self.arm.get() && key.starts_with("submitted:") && value.first() == Some(&0x02) { + self.arm.set(false); + return Err(Fault::Unavailable("commit write faulted".into())); + } + self.inner.set(key, value) + } + + fn delete(&self, key: &str) -> Result<(), Fault> { + self.inner.delete(key) + } + + fn list_keys(&self, prefix: &str) -> Result, Fault> { + self.inner.list_keys(prefix) + } + + fn contains(&self, key: &str) -> Result { + self.inner.contains(key) + } + + fn len(&self, key: &str) -> Result, Fault> { + LocalStoreHost::len(&self.inner, key) + } + + fn count(&self, prefix: &str) -> Result { + self.inner.count(prefix) + } +} + +/// A source that never submits: exercises the reconcile pass alone. +struct Idle; + +impl Poller for Idle { + type Outcome = Verdict; + + fn poll(&self, _host: &H, _watch: WatchRef<'_>, _params: &[u8], _tick: &Tick) -> Verdict { + Verdict::TryNextBlock { reason: [0; 4] } + } +} + +/// A source that posts one fixed order on every poll. +struct PostOnce(GPv2OrderData); + +impl Poller for PostOnce { + type Outcome = Verdict; + + fn poll(&self, _host: &H, _watch: WatchRef<'_>, _params: &[u8], _tick: &Tick) -> Verdict { + ready_outcome(&self.0) + } +} + +/// Seed a stranded `RESERVED` marker on the encoded body, as a prior +/// tick's reserve whose submit outcome never landed. +fn seed_reserved(host: &impl LocalStoreHost, order: &GPv2OrderData) { + Journal::submitted(host) + .reserve(&intent_id(order), &intent_bytes(order)) + .unwrap(); +} + +fn cow_mark(host: &impl LocalStoreHost, order: &GPv2OrderData) -> Option { + Journal::submitted(host).mark(&intent_id(order)).unwrap() +} + +/// W1: reserved, but the venue never saw the POST. The next tick's +/// reconcile resubmits to exactly one held order. +#[test] +fn w1_reserved_but_venue_never_saw_the_post_reconciles() { + let host = MockLocalStore::default(); + let order = submittable_order(); + seed_reserved(&host, &order); + let venue = HoldingVenue::accepting(); + + run(&host, &holding_client(&venue), &Idle, &sample_tick()).unwrap(); + + assert_eq!( + venue.post_count(), + 1, + "reconcile resubmits the stranded body" + ); + assert_eq!(venue.held_count(), 1, "exactly one held order"); + assert_eq!( + venue.posts()[0], + intent_bytes(&order), + "the reserved body round-trips", + ); + assert_eq!(cow_mark(&host, &order), Some(Mark::Committed)); +} + +/// W2: accepted, then the commit faults, leaving the marker RESERVED. +/// The next tick's reconcile resubmits, the venue dedups +/// (AlreadyHeld -> Accepted), the commit lands: two POSTs, one held. +#[test] +fn w2_accepted_then_commit_faults_reconciles_without_double_holding() { + let host = FlakyCommit::new(); + WatchSet::new(&host) + .put(&sample_owner(), &sample_hash(), b"params") + .unwrap(); + let order = submittable_order(); + let venue = HoldingVenue::accepting(); + + // Tick A: reserve, venue accepts (POST #1), the commit write faults; + // the RESERVED marker persists, no release runs. + run( + &host, + &holding_client(&venue), + &PostOnce(order.clone()), + &sample_tick(), + ) + .unwrap(); + assert_eq!(venue.post_count(), 1); + assert_eq!( + cow_mark(&host, &order), + Some(Mark::Reserved), + "a commit fault leaves the marker RESERVED", + ); + + // Tick B: reconcile re-POSTs (POST #2), the venue dedups, the commit + // lands. The fresh loop then sees COMMITTED and never re-posts. + run( + &host, + &holding_client(&venue), + &PostOnce(order.clone()), + &sample_tick(), + ) + .unwrap(); + assert_eq!( + venue.post_count(), + 2, + "reconcile re-POSTs the reserved body" + ); + assert_eq!(venue.held_count(), 1, "one held order despite two POSTs"); + assert_eq!(cow_mark(&host, &order), Some(Mark::Committed)); +} + +/// W3: the run was abandoned after the venue received the POST. The next +/// tick's reconcile re-POSTs, the AlreadyHeld backstop accepts, one held. +#[test] +fn w3_abandoned_after_the_post_reconciles_to_one_held() { + let host = MockLocalStore::default(); + let order = submittable_order(); + seed_reserved(&host, &order); + let venue = HoldingVenue::accepting(); + venue.preload(&intent_bytes(&order)); + + run(&host, &holding_client(&venue), &Idle, &sample_tick()).unwrap(); + + assert_eq!(venue.post_count(), 1); + assert_eq!(venue.held_count(), 1, "the already-held order stays single"); + assert_eq!(cow_mark(&host, &order), Some(Mark::Committed)); + // The venue-never-saw-it sub-case is W1 above. +} + +/// Anti-#572: a RESERVED marker MUST drive a reconcile POST routed +/// THROUGH `venue.submit`, where the AlreadyHeld -> Accepted backstop +/// catches the duplicate. A pre-venue skip (the #572 shape) would defeat +/// the backstop and drop the order. +#[test] +fn anti_572_reserved_marker_drives_a_reconcile_post_through_the_venue() { + let host = MockLocalStore::default(); + let order = submittable_order(); + seed_reserved(&host, &order); + let venue = HoldingVenue::accepting(); + // The venue already holds it: the reconcile POST hits the + // AlreadyHeld -> Accepted path, not a fresh accept. + venue.preload(&intent_bytes(&order)); + + run(&host, &holding_client(&venue), &Idle, &sample_tick()).unwrap(); + + assert_eq!( + venue.post_count(), + 1, + "the reserved marker POSTs, never a silent skip", + ); + assert_eq!( + venue.posts()[0], + intent_bytes(&order), + "the exact reserved bytes re-POST", + ); + assert_eq!(venue.held_count(), 1, "no duplicate order"); + assert_eq!( + cow_mark(&host, &order), + Some(Mark::Committed), + "the backstop-accepted resubmit commits", + ); +} diff --git a/crates/videre-sdk/src/client.rs b/crates/videre-sdk/src/client.rs index 08a535b0..f3384d44 100644 --- a/crates/videre-sdk/src/client.rs +++ b/crates/videre-sdk/src/client.rs @@ -257,6 +257,13 @@ impl VenueClient { V::ID } + /// The bound transport, so a keeper reconcile pass resubmits reserved + /// bodies through the same seam this client submits on. + #[must_use] + pub fn transport(&self) -> &T { + &self.transport + } + /// Encode the typed body and price it at the bound venue. The /// returned [`Quoted`] carries the encoded bytes, so `submit` sends /// exactly the body the venue priced.