From 8e67ebb1a4e188313dcf5399537197d7e02e2888 Mon Sep 17 00:00:00 2001 From: mfw78 Date: Sat, 25 Jul 2026 12:32:50 +0000 Subject: [PATCH] cow: adopt the Journal guard in the composable-cow keeper submit Rewire submit_ready onto Journal::guard: the six inline journal.mark/reserve/commit/release calls that carry the #572 reserve-before-effect ordering fold into one guard call, with the venue outcome mapped to a Disposition (Accepted commits, requires-signing and known refusals release, an unknown outcome retains for reconcile). Behaviour is identical: all 23 composable-cow run tests including w1/w2/w3 and anti_572 pass unchanged. The remaining code is the CoW-specific outcome handling (refusal clear, retry ledger, tracing) that no primitive absorbs; the value is that the durable-effect ordering is no longer inlined, so a second venue's keeper reuses it rather than re-deriving it. Part of #609. Stacks on the Journal::guard combinator (#629). --- crates/composable-cow/src/run.rs | 96 ++++++++++++-------------------- 1 file changed, 37 insertions(+), 59 deletions(-) diff --git a/crates/composable-cow/src/run.rs b/crates/composable-cow/src/run.rs index 279a02da..591759e5 100644 --- a/crates/composable-cow/src/run.rs +++ b/crates/composable-cow/src/run.rs @@ -20,7 +20,8 @@ use cow_venue::{CowClient, CowIntent, CowIntentBody, CowVenue, SignedOrder, clas use cowprotocol::GPv2OrderData; use nexum_sdk::host::{Fault, LocalStoreHost}; use nexum_sdk::keeper::{ - Gates, Journal, Mark, Poller, Retrier, RetryAction, Tick, WatchRef, WatchSet, + Disposition, Gates, Guarded, Journal, Mark, Poller, Retrier, RetryAction, Tick, WatchRef, + WatchSet, }; use std::task::Poll; @@ -98,14 +99,8 @@ where Ok(()) } -/// Submit one polled `Post` order through the typed client on the -/// `submitted:` reserve/commit journal, dispatching a refusal through -/// the retry ledger. -/// -/// Keyed on the venue-and-body submission key: `COMMITTED` is an -/// idempotent skip, `RESERVED` is owned by reconcile. A fresh order -/// reserves before the submit and commits on acceptance; release runs -/// only on a known synchronous non-accept. +/// Submit one polled `Post` order through the guard, folding a refusal +/// into the retry ledger. fn submit_ready( host: &H, venue: &CowClient, @@ -128,9 +123,7 @@ where }; let Some(order_data) = gpv2_to_order_data(order) else { - // An unknown enum marker means the SDK cannot express this - // payload yet; skip rather than drop so an SDK upgrade can - // still pick the watch up. + // Unknown enum marker: skip, not drop, so an SDK upgrade picks it up. tracing::warn!( "{label} submit skipped for {owner:#x}: GPv2OrderData carried an unknown enum marker" ); @@ -142,8 +135,7 @@ where owner, signature: signature.to_vec(), })); - // Reserve the exact wire bytes the venue submit and the reconcile - // resubmit both carry, so the id and the reservation agree. + // Key on the exact bytes the submit and reconcile both carry. let encoded = match intent.to_bytes() { Ok(bytes) => bytes, Err(err) => { @@ -153,42 +145,40 @@ where }; let intent_id = submission_key(&CowVenue::ID, &encoded); let journal = Journal::submitted(host); - 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(()); + // The guard owns the reserve/commit/reconcile ordering (#572). + let guarded = poll_once(journal.guard(&intent_id, &encoded, || async { + let outcome = venue.submit(&intent).await; + let disposition = match &outcome { + Ok(SubmitOutcome::Accepted(_)) => Disposition::Commit, + Ok(SubmitOutcome::RequiresSigning(_)) + | Err(ClientError::Body(_)) + | Err(ClientError::Venue(_)) => Disposition::Release, + // Unknown outcome: hold for reconcile. + Err(_) => Disposition::Retain, + }; + (disposition, outcome) + })); + let outcome = match guarded { + Poll::Ready(res) => match res? { + Guarded::Skipped(Mark::Committed) => { + tracing::info!("{label} {intent_id} already committed; skipping re-submit"); + return Ok(()); + } + Guarded::Skipped(Mark::Reserved) => { + tracing::info!("{label} {intent_id} reserved; reconcile owns it"); + return Ok(()); + } + Guarded::Ran(outcome) => outcome, + }, + Poll::Pending => { + // Guest transports never suspend; retry next block, reconcile owns the marker. + tracing::error!("{label} submit future suspended; retrying next block"); + return Retrier::new(host).apply(watch, RetryAction::TryNextBlock, tick); } - 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. 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. + // Acceptance ends the refusal episode: clear the first-refusal marker. if let Err(fault) = Retrier::new(host).clear_refusal(watch) { tracing::error!("submitted {intent_id} but refusal-marker clear failed: {fault}"); } @@ -198,20 +188,12 @@ where ); } Ok(SubmitOutcome::RequiresSigning(_)) => { - // 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), @@ -224,17 +206,13 @@ where } 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. + // Non-exhaustive fallthrough: log the action name. other => { let action_label: &'static str = other.into(); tracing::warn!("submit retry action {action_label}: {fault}"); } } } - // `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(())