Skip to content

cow: run the cow keeper on the generic venue client #468

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
mfw78 wants to merge 1 commit into
base: feat/m4-cow-adapter-cdylib
Choose a base branch
from
Open

cow: run the cow keeper on the generic venue client #468

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

27 changes: 17 additions & 10 deletions crates/shepherd-sdk-test/tests/mock_venue.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,15 +9,21 @@ use cowprotocol::{BuyTokenDestination, GPv2OrderData, OrderKind, SellTokenSource
use nexum_sdk::host::{Fault, LocalStoreHost as _, RateLimit};
use nexum_sdk::keeper::{ConditionalSource, Journal, Tick, WatchRef, WatchSet, watch_key};
use shepherd_sdk::cow::{
CowApiError, CowHost, CowIntent, CowIntentBody, OrderRejection, SignedOrder,
gpv2_to_order_data, order_data_to_body, order_uid_hex, run,
CowApiError, CowApiTransport, CowClient, CowHost, CowIntent, CowIntentBody, OrderRejection,
SignedOrder, gpv2_to_order_data, order_data_to_body, order_uid_hex, run,
};
use shepherd_sdk_test::{MockHost, MockVenue};

const SEPOLIA: u64 = 11_155_111;

type VenueHost = MockHost<MockVenue>;

/// The typed client every test drives `run` with: the transitional
/// cow-api bridge over the scripted venue host.
fn venue(host: &VenueHost) -> CowClient<CowApiTransport<'_, VenueHost>> {
CowClient::with_transport(CowApiTransport::new(host, SEPOLIA))
}

/// Closure-backed source so each test scripts its own outcome.
struct FnSource<F>(F);

Expand Down Expand Up @@ -146,7 +152,7 @@ fn keeper_retries_a_transient_rejection_then_submits() {
host.cow_api.enqueue_submit(Ok(client_uid(&order)));

let source = ready_source(&order);
run(&host, &source, &sample_tick()).unwrap();
run(&host, &venue(&host), &source, &sample_tick()).unwrap();
assert_eq!(host.cow_api.call_count(), 1);
assert!(host.store.snapshot().contains_key(&key), "watch survives");
assert!(
Expand All @@ -155,7 +161,7 @@ fn keeper_retries_a_transient_rejection_then_submits() {
.unwrap()
);

run(&host, &source, &sample_tick()).unwrap();
run(&host, &venue(&host), &source, &sample_tick()).unwrap();
assert_eq!(host.cow_api.call_count(), 2);
assert!(
Journal::submitted(&host)
Expand Down Expand Up @@ -185,7 +191,7 @@ fn keeper_backs_off_on_rate_limit_and_submits_after_the_gate() {

let t0 = sample_tick();
let source = ready_source(&order);
run(&host, &source, &t0).unwrap();
run(&host, &venue(&host), &source, &t0).unwrap();
assert_eq!(host.cow_api.call_count(), 1);

// 2500ms rounds up to a 3s epoch gate: a tick inside it never
Expand All @@ -194,14 +200,14 @@ fn keeper_backs_off_on_rate_limit_and_submits_after_the_gate() {
epoch_s: t0.epoch_s + 2,
..t0
};
run(&host, &source, &gated).unwrap();
run(&host, &venue(&host), &source, &gated).unwrap();
assert_eq!(host.cow_api.call_count(), 1, "gated tick must not submit");

let clear = Tick {
epoch_s: t0.epoch_s + 3,
..t0
};
run(&host, &source, &clear).unwrap();
run(&host, &venue(&host), &source, &clear).unwrap();
assert_eq!(host.cow_api.call_count(), 2);
assert!(
Journal::submitted(&host)
Expand All @@ -222,7 +228,7 @@ fn keeper_survives_a_venue_outage_and_submits_on_recovery() {
.inject_fault(CowApiError::Fault(Fault::Unavailable("venue down".into())));

let source = ready_source(&order);
run(&host, &source, &sample_tick()).unwrap();
run(&host, &venue(&host), &source, &sample_tick()).unwrap();
let snapshot = host.store.snapshot();
assert!(snapshot.contains_key(&key));
assert!(!snapshot.contains_key(&watch_key.next_block_key()));
Expand All @@ -231,7 +237,7 @@ fn keeper_survives_a_venue_outage_and_submits_on_recovery() {

host.cow_api.clear_fault();
host.cow_api.enqueue_submit(Ok(client_uid(&order)));
run(&host, &source, &sample_tick()).unwrap();
run(&host, &venue(&host), &source, &sample_tick()).unwrap();
assert_eq!(host.cow_api.call_count(), 2);
assert!(
Journal::submitted(&host)
Expand All @@ -249,7 +255,7 @@ fn keeper_drops_the_watch_on_a_scripted_permanent_rejection() {
host.cow_api
.enqueue_submit(Err(rejection("InvalidSignature")));

run(&host, &ready_source(&order), &sample_tick()).unwrap();
run(&host, &venue(&host), &ready_source(&order), &sample_tick()).unwrap();

assert!(host.store.is_empty(), "watch and gates must go");
assert_eq!(host.cow_api.call_count(), 1);
Expand All @@ -276,6 +282,7 @@ fn keeper_sweep_ignores_sibling_namespace_watches() {
let polls = std::cell::Cell::new(0_u32);
run(
&host,
&venue(&host),
&src(|_, _, _, _| {
polls.set(polls.get() + 1);
ready_outcome(&order)
Expand Down
3 changes: 3 additions & 0 deletions crates/shepherd-sdk/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,9 @@ cow-venue = { path = "../cow-venue", features = ["client", "assembly"] }
# The structured poll seam the keeper run dispatches on.
composable-cow = { path = "../composable-cow" }
nexum-sdk = { path = "../nexum-sdk" }
# The typed client seam the keeper run submits through; also the
# `VenueTransport` contract the legacy cow-api bridge implements.
videre-sdk = { path = "../videre-sdk" }
cowprotocol = { version = "0.2.0", default-features = false }
alloy-primitives.workspace = true
serde_json.workspace = true
Expand Down
28 changes: 18 additions & 10 deletions crates/shepherd-sdk/src/cow/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,9 +1,10 @@
//! CoW Protocol bridging.
//!
//! ABI decoding helpers, the orderbook error surface, and [`run()`] -
//! the poll/submit composition over the keeper stores. The chain-edge
//! order projections live in the `cow-venue` `assembly` slice (the
//! venue adapter owns them) and are re-exported here.
//! the poll/submit composition over the keeper stores, submitting
//! through the typed [`CowClient`] on the `videre:venue/client` seam.
//! The chain-edge order projections live in the `cow-venue` `assembly`
//! slice (the venue adapter owns them) and are re-exported here.
//!
//! The poll seam is the structured
//! [`Verdict`](composable_cow::Verdict), carried by the
Expand All @@ -14,11 +15,14 @@
//! primitive arguments (`&[u8]`, `Option<&str>`, slices) so they can
//! be unit-tested without wit-bindgen scaffolding and re-used
//! unchanged by TWAP, EthFlow, and future strategy modules. The
//! keeper run is generic over the host traits alone.
//! keeper run is generic over the host traits and the venue transport
//! alone; [`CowApiTransport`] carries it over the legacy
//! `shepherd:cow/cow-api` import until the module worlds flip.

pub mod error;
pub mod events;
pub mod run;
pub mod transport;

/// Chain-edge order assembly, re-exported from the `cow-venue`
/// `assembly` slice the venue adapter owns.
Expand All @@ -28,19 +32,23 @@ pub use error::{
classify_submit_error, is_already_submitted,
};
pub use run::run;
pub use transport::CowApiTransport;

/// The venue-neutral intent body types and their borsh `IntentBody`
/// codec, re-exported from the `cow-venue` default slice. The shim keeps
/// this path stable while the module ports move off the legacy surface.
/// codec, re-exported from the `cow-venue` default slice, plus the
/// typed CoW venue client. The shim keeps this path stable while the
/// module ports move off the legacy surface.
pub use cow_venue::{
BuyToken, BuyTokenDestination, CowIntent, CowIntentBody, OrderBody, OrderBuilder, OrderKind,
OrderUid, SellToken, SellTokenSource, SignedOrder, intent_id,
BuyToken, BuyTokenDestination, CowClient, CowIntent, CowIntentBody, CowVenue, OrderBody,
OrderBuilder, OrderKind, OrderUid, SellToken, SellTokenSource, SignedOrder, intent_id,
};

use nexum_sdk::host::Host;

/// `shepherd:cow/cow-api` - orderbook submission path. The CoW-domain
/// sibling of the core host traits in [`nexum_sdk::host`].
/// `shepherd:cow/cow-api` - the legacy orderbook submission path,
/// retiring. The keeper [`run()`] submits through the typed
/// [`CowClient`]; [`CowApiTransport`] bridges it onto this seam until
/// the module worlds flip to `videre:venue/client`.
pub trait CowApiHost {
/// Submit an `OrderCreation` JSON body. The host returns the
/// canonical order UID on success. A rejection surfaces as a typed
Expand Down
133 changes: 61 additions & 72 deletions crates/shepherd-sdk/src/cow/run.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,40 +4,43 @@
//! [`run`] walks the keeper watch set, polls each gate-ready
//! watch through a [`ConditionalSource`], and runs the
//! [`Verdict`]'s effect: lifecycle outcomes update the gate and
//! watch stores, `Post` drives one submission through the
//! [`CowApiHost`](super::CowApiHost) seam with the `submitted:`
//! journal as the idempotency guard - keyed on the venue-and-body
//! [`intent_id`] - and the keeper [`Retrier`]
//! 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.
//!
//! Store faults abort the sweep (the next tick replays it);
//! submission failures never do - they classify into a
//! [`RetryAction`], the ledger applies the effect, and the sweep
//! moves on. Diagnostics go through the guest `tracing` facade -
//! submission failures never do - they fold into a
//! [`RetryAction`] through the videre
//! [`retry_action`] table, 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};
use alloy_primitives::{Address, Bytes, hex};
use composable_cow::Verdict;
use cow_venue::assembly::build_order_creation;
use cowprotocol::GPv2OrderData;
use nexum_sdk::host::Fault;
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 super::{
CowApiError, CowHost, CowIntent, CowIntentBody, SignedOrder, classify_submit_error,
gpv2_to_order_data, intent_id, is_already_submitted, order_data_to_body,
CowClient, CowIntent, CowIntentBody, SignedOrder, gpv2_to_order_data, intent_id,
order_data_to_body,
};

/// 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 `submit_order` call.
pub fn run<H, S>(host: &H, source: &S, tick: &Tick) -> Result<(), Fault>
/// most one venue submit through `venue`.
pub fn run<H, S, T>(host: &H, venue: &CowClient<T>, source: &S, tick: &Tick) -> Result<(), Fault>
where
H: CowHost,
H: LocalStoreHost,
S: ConditionalSource<H, Outcome = Verdict>,
T: VenueTransport,
{
let watches = WatchSet::new(host);
let gates = Gates::new(host);
Expand All @@ -55,7 +58,7 @@ where
Verdict::Post {
order, signature, ..
} => {
submit_ready(host, watch, &order, signature, tick, source.label())?;
submit_ready(host, venue, watch, &order, signature, tick, source.label())?;
}
Verdict::TryNextBlock { .. } => {}
Verdict::WaitBlock { wait_until, .. } => gates.set_next_block(watch, wait_until)?,
Expand All @@ -74,24 +77,27 @@ where
Ok(())
}

/// Submit one freshly-polled `Ready` order, guarding on the
/// `submitted:` journal and dispatching any failure through the retry
/// ledger.
/// Submit one freshly-polled `Ready` order through the typed client,
/// guarding on the `submitted:` journal and 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 a venue submit carries - never from the assembled
/// `OrderCreation` - so the guard survives assembly moving into the
/// venue adapter. The orderbook's UID is the receipt; it rides the
/// log only.
fn submit_ready<H: CowHost>(
/// 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.
fn submit_ready<H, T>(
host: &H,
venue: &CowClient<T>,
watch: WatchRef<'_>,
order: &GPv2OrderData,
signature: Bytes,
tick: &Tick,
label: &str,
) -> Result<(), Fault> {
) -> Result<(), Fault>
where
H: LocalStoreHost,
T: VenueTransport,
{
let Ok(owner) = watch.owner_hex().parse::<Address>() else {
tracing::warn!(
"watch {} carries an unparseable owner; skipping submit",
Expand Down Expand Up @@ -127,73 +133,56 @@ fn submit_ready<H: CowHost>(
tracing::info!("{label} {intent_id} already submitted; skipping re-submit");
return Ok(());
}
let creation = match build_order_creation(&order_data, &signature, owner) {
Ok(creation) => creation,
Err(err) => {
// A constructor rejection (zero `from`, `validTo` beyond
// the client-side max horizon) is deterministic for this
// polled payload: keeping the watch would re-poll and
// re-warn on every block forever. Drop through the ledger
// - the same net effect as the pre-keeper flow, where
// the orderbook rejected the shipped body and the
// classifier dropped the watch.
tracing::warn!("{label} submit dropped watch for {owner:#x}: {err}");
Retrier::new(host).apply(watch, RetryAction::Drop, tick.epoch_s)?;
return Ok(());
}
};
let body = match serde_json::to_vec(&creation) {
Ok(body) => body,
Err(e) => {
tracing::error!("OrderCreation JSON encode failed: {e}");
return Ok(());
}
};

match host.submit_order(tick.chain_id, &body) {
Ok(receipt) => {
let Some(outcome) = rt::complete(venue.submit(&intent)) else {
// Guest transports never suspend; a pending future means a
// foreign transport misbehaved. The watch stays for the next
// tick.
tracing::error!("{label} submit future suspended; retrying next tick");
return Ok(());
};
match outcome {
Ok(SubmitOutcome::Accepted(receipt)) => {
// 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
// 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 {receipt})");
}
Err(CowApiError::Rejected(rejection)) if is_already_submitted(&rejection) => {
// Success wearing an error status: the orderbook already
// holds this order. Journal the intent-id and keep the
// watch so the next tick short-circuits instead of
// re-posting. As above, a journal fault post-submit only
// forfeits the short-circuit; it must not abort the sweep.
if let Err(fault) = journal.record(&intent_id) {
tracing::error!(
"orderbook already holds {intent_id} but journal write failed: {fault}"
);
}
tracing::info!(
"orderbook already holds this order ({}); intent-id journalled",
rejection.error_type,
"submitted {intent_id} (receipt {})",
hex::encode_prefixed(&receipt),
);
}
Err(err) => {
let action = classify_submit_error(&err);
Ok(SubmitOutcome::RequiresSigning(_)) => {
// A sweep cannot sign; nothing is journalled, so the next
// tick surfaces the same ask afresh.
tracing::warn!("{label} submit for {owner:#x} requires signing; not journalled");
}
Err(ClientError::Body(err)) => {
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)?;
match action {
RetryAction::TryNextBlock => tracing::warn!("submit retry-next-block: {err}"),
RetryAction::TryNextBlock => tracing::warn!("submit retry-next-block: {fault}"),
RetryAction::Backoff { seconds } => {
tracing::warn!("submit backoff {seconds}s: {err}");
tracing::warn!("submit backoff {seconds}s: {fault}");
}
RetryAction::Drop => tracing::warn!("submit dropped watch: {err}"),
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.
other => {
let action_label: &'static str = other.into();
tracing::warn!("submit retry action {action_label}: {err}");
tracing::warn!("submit retry action {action_label}: {fault}");
}
}
}
// `ClientError` is non-exhaustive; a future case leaves the
// watch for the next tick.
Err(err) => tracing::error!("submit failed: {err}"),
}
Ok(())
}
Loading
Loading