From 2e7ecb3b81d5807c69d20b274eb95dacac572718 Mon Sep 17 00:00:00 2001 From: mfw78 Date: Fri, 24 Jul 2026 02:44:16 +0000 Subject: [PATCH] videre-sdk: name the one-poll driver for the seam it asserts --- crates/composable-cow/src/sweep.rs | 16 +++++++---- crates/cow-venue/src/client.rs | 8 ++++-- crates/videre-macros/src/keeper.rs | 8 +++--- crates/videre-macros/src/lib.rs | 2 +- crates/videre-sdk/src/client.rs | 34 ++++++++++++++++++++++ crates/videre-sdk/src/keeper.rs | 5 +++- crates/videre-sdk/src/lib.rs | 8 +++--- crates/videre-sdk/src/rt.rs | 38 ------------------------- crates/videre-sdk/tests/adapter.rs | 5 +++- docs/05-sdk-design.md | 5 ++-- modules/ethflow-watcher/src/strategy.rs | 8 ++++-- 11 files changed, 74 insertions(+), 63 deletions(-) delete mode 100644 crates/videre-sdk/src/rt.rs diff --git a/crates/composable-cow/src/sweep.rs b/crates/composable-cow/src/sweep.rs index 768d51c4..01e0a702 100644 --- a/crates/composable-cow/src/sweep.rs +++ b/crates/composable-cow/src/sweep.rs @@ -29,8 +29,11 @@ use nexum_sdk::host::{Fault, LocalStoreHost}; use nexum_sdk::keeper::{ ConditionalSource, Gates, Journal, 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, rt}; +use videre_sdk::{ClientError, SubmitOutcome, VenueFault, VenueTransport}; use crate::Verdict; @@ -135,12 +138,13 @@ where return Ok(()); } - let Some(outcome) = rt::complete(venue.submit(&intent)) else { + let Poll::Ready(outcome) = poll_once(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(()); + // 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. + tracing::error!("{label} submit future suspended; retrying next block"); + return Retrier::new(host).apply(watch, RetryAction::TryNextBlock, tick); }; match outcome { Ok(SubmitOutcome::Accepted(receipt)) => { diff --git a/crates/cow-venue/src/client.rs b/crates/cow-venue/src/client.rs index 9dd86963..9a5ecfaf 100644 --- a/crates/cow-venue/src/client.rs +++ b/crates/cow-venue/src/client.rs @@ -147,9 +147,11 @@ mod tests { let client = CowClient::with_transport(spy.clone()); assert_eq!(client.venue(), CowVenue::ID); - videre_sdk::rt::complete(client.submit(&body)) - .expect("guest futures complete in one poll") - .expect("submit succeeds"); + let std::task::Poll::Ready(result) = videre_sdk::client::poll_once(client.submit(&body)) + else { + panic!("guest futures complete in one poll"); + }; + result.expect("submit succeeds"); let calls = spy.submitted.borrow(); assert_eq!(calls.len(), 1); diff --git a/crates/videre-macros/src/keeper.rs b/crates/videre-macros/src/keeper.rs index 9af9c569..465f701d 100644 --- a/crates/videre-macros/src/keeper.rs +++ b/crates/videre-macros/src/keeper.rs @@ -4,7 +4,7 @@ //! with the keeper deltas: the `client` capability is required, the //! videre interfaces remap onto the SDK bindings (one shim set, one //! type identity for the typed client), async handlers complete via -//! `videre_sdk::rt::complete`, and `ClientError` folds into the wire +//! `videre_sdk::client::poll_once`, and `ClientError` folds into the wire //! fault so `?` works in handlers. use proc_macro2::TokenStream; @@ -115,9 +115,9 @@ pub(crate) fn expand(input: &ItemImpl) -> syn::Result { // typed internal fault, never a hang. let drive = |call: TokenStream| { quote! { - match ::videre_sdk::rt::complete(#call) { - ::core::option::Option::Some(result) => result, - ::core::option::Option::None => ::core::result::Result::Err( + match ::videre_sdk::client::poll_once(#call) { + ::core::task::Poll::Ready(result) => result, + ::core::task::Poll::Pending => ::core::result::Result::Err( nexum::host::types::Fault::Internal( ::std::string::String::from(#SUSPENDED), ), diff --git a/crates/videre-macros/src/lib.rs b/crates/videre-macros/src/lib.rs index a9cb5677..bbcaccb3 100644 --- a/crates/videre-macros/src/lib.rs +++ b/crates/videre-macros/src/lib.rs @@ -180,7 +180,7 @@ pub fn venue(attr: TokenStream, item: TokenStream) -> TokenStream { /// whose associated functions are the event handlers (`init`, /// `on_block`, `on_chain_logs`, `on_tick`, `on_message`, /// `on_intent_status`); handlers may be `async` and are completed on -/// the synchronous guest boundary (`videre_sdk::rt::complete`), so a +/// the synchronous guest boundary (`videre_sdk::client::poll_once`), so a /// handler can await the typed `VenueClient` directly. /// /// The macro reads the crate's `module.toml`, requires the `client` diff --git a/crates/videre-sdk/src/client.rs b/crates/videre-sdk/src/client.rs index 63ff375d..4a4e9443 100644 --- a/crates/videre-sdk/src/client.rs +++ b/crates/videre-sdk/src/client.rs @@ -15,6 +15,8 @@ use std::borrow::Cow; use std::fmt; use std::future::Future; use std::marker::PhantomData; +use std::pin::pin; +use std::task::{Context, Poll, Waker}; use strum::IntoStaticStr; @@ -133,6 +135,17 @@ pub trait VenueTransport: sealed::SealedTransport { ) -> impl Future>; } +/// Poll a future once and return its state. `videre:venue/client@0.1.0` +/// declares plain funcs, so a [`VenueTransport`] over the host import +/// resolves on the first poll. [`Poll::Pending`] means a foreign +/// [`VenueTransport`] impl suspended, which the keeper macro folds to +/// `Fault::Internal`. +pub fn poll_once(future: F) -> Poll { + let mut future = pin!(future); + let mut cx = Context::from_waker(Waker::noop()); + future.as_mut().poll(&mut cx) +} + /// The module's `videre:venue/client` import behind the /// [`VenueTransport`] seam: the transport every guest-side /// [`VenueClient`] defaults to. @@ -308,3 +321,24 @@ pub enum ClientError { #[error(transparent)] Venue(#[from] VenueFault), } + +#[cfg(test)] +mod tests { + use std::task::Poll; + + use super::poll_once; + + #[test] + fn ready_chain_completes_in_one_poll() { + async fn two() -> u8 { + let one = async { 1u8 }.await; + one + async { 1u8 }.await + } + assert_eq!(poll_once(two()), Poll::Ready(2)); + } + + #[test] + fn suspending_future_reports_pending() { + assert_eq!(poll_once(std::future::pending::<()>()), Poll::Pending); + } +} diff --git a/crates/videre-sdk/src/keeper.rs b/crates/videre-sdk/src/keeper.rs index 732a271a..6e64fdcd 100644 --- a/crates/videre-sdk/src/keeper.rs +++ b/crates/videre-sdk/src/keeper.rs @@ -215,7 +215,10 @@ mod tests { /// Drive a sweep on the test's synchronous boundary. fn run(future: F) -> F::Output { - crate::rt::complete(future).expect("sweep futures complete in one poll") + match crate::client::poll_once(future) { + std::task::Poll::Ready(output) => output, + std::task::Poll::Pending => panic!("sweep futures complete in one poll"), + } } /// Answers every poll with one programmed outcome. diff --git a/crates/videre-sdk/src/lib.rs b/crates/videre-sdk/src/lib.rs index 007eb2e2..3594f2b5 100644 --- a/crates/videre-sdk/src/lib.rs +++ b/crates/videre-sdk/src/lib.rs @@ -25,8 +25,9 @@ //! own `videre:venue/client` import). Lives here (not in the //! strategy SDK) so the codec and the client that speaks it version //! together. `#[videre_sdk::keeper]` on a handler impl wires the -//! import and drives async handlers; [`rt`] completes their futures -//! on the synchronous guest boundary. +//! import and drives async handlers; +//! [`poll_once`](client::poll_once) completes their futures on the +//! synchronous guest boundary. //! //! - [`keeper`](mod@keeper) - the generic sweep assembler: //! [`Keeper::sweep`] runs the world-neutral `nexum_sdk::keeper` @@ -80,7 +81,6 @@ pub mod client; pub mod event; pub mod faults; pub mod keeper; -pub mod rt; pub mod transport; pub use adapter::VenueAdapter; @@ -96,7 +96,7 @@ pub use videre_macros::IntentBody; /// (asserting the `client` capability), remaps the videre interfaces /// onto the SDK bindings so the module drives a [`VenueClient`] with /// shared type identity, dispatches events to the handlers (async ones -/// completed through [`rt::complete`]), and folds [`ClientError`] into +/// completed through [`client::poll_once`]), and folds [`ClientError`] into /// the wire fault so `?` works in handlers. See /// [`videre_macros::keeper`]. pub use videre_macros::keeper; diff --git a/crates/videre-sdk/src/rt.rs b/crates/videre-sdk/src/rt.rs deleted file mode 100644 index 6a62f4f3..00000000 --- a/crates/videre-sdk/src/rt.rs +++ /dev/null @@ -1,38 +0,0 @@ -//! Futures on the synchronous guest boundary. - -use std::future::Future; -use std::pin::pin; -use std::task::{Context, Poll, Waker}; - -/// Complete a future in one poll. Guest host imports are synchronous, -/// so every await in a keeper future resolves immediately and one poll -/// runs it to completion. `None` reports a future that suspended, -/// which nothing built over the host imports does; the keeper macro's -/// emitted glue folds it to a typed fault. -pub fn complete(future: F) -> Option { - let mut future = pin!(future); - let mut cx = Context::from_waker(Waker::noop()); - match future.as_mut().poll(&mut cx) { - Poll::Ready(output) => Some(output), - Poll::Pending => None, - } -} - -#[cfg(test)] -mod tests { - use super::complete; - - #[test] - fn ready_chain_completes_in_one_poll() { - async fn two() -> u8 { - let one = async { 1u8 }.await; - one + async { 1u8 }.await - } - assert_eq!(complete(two()), Some(2)); - } - - #[test] - fn suspending_future_reports_none() { - assert_eq!(complete(std::future::pending::<()>()), None); - } -} diff --git a/crates/videre-sdk/tests/adapter.rs b/crates/videre-sdk/tests/adapter.rs index 0c490331..d43cd252 100644 --- a/crates/videre-sdk/tests/adapter.rs +++ b/crates/videre-sdk/tests/adapter.rs @@ -15,7 +15,10 @@ use videre_sdk::{ /// Drive a client future on the test's synchronous boundary. fn run(future: F) -> F::Output { - videre_sdk::rt::complete(future).expect("client futures complete in one poll") + match videre_sdk::client::poll_once(future) { + std::task::Poll::Ready(output) => output, + std::task::Poll::Pending => panic!("client futures complete in one poll"), + } } /// First published body version: a fixed-price quote. diff --git a/docs/05-sdk-design.md b/docs/05-sdk-design.md index f77078e4..1512ee47 100755 --- a/docs/05-sdk-design.md +++ b/docs/05-sdk-design.md @@ -284,8 +284,9 @@ to an `impl` block whose associated functions are the event handlers `on_intent_status`). It requires the `client` capability (the `videre:venue/client` import is what makes a keeper a keeper), wires that import onto the SDK's shared shims, and lets handlers be `async` -so they can await the typed client directly; `videre_sdk::rt` -completes the futures on the synchronous guest boundary. A +so they can await the typed client directly; +`videre_sdk::client::poll_once` completes the futures on the +synchronous guest boundary. A `From` impl onto the wire fault is emitted, so `?` applies to client calls inside handlers. diff --git a/modules/ethflow-watcher/src/strategy.rs b/modules/ethflow-watcher/src/strategy.rs index 69456554..de3d3238 100644 --- a/modules/ethflow-watcher/src/strategy.rs +++ b/modules/ethflow-watcher/src/strategy.rs @@ -211,7 +211,7 @@ mod tests { use nexum_sdk::host::LocalStoreHost as _; use nexum_sdk_test::{MockHost, capture_tracing}; use videre_sdk::client::VenueId; - use videre_sdk::rt::complete; + use videre_sdk::client::poll_once; use videre_sdk::status_body::IntentStatus as Lifecycle; use videre_sdk::{IntentStatus, Quotation, SubmitOutcome, VenueFault}; @@ -309,8 +309,10 @@ mod tests { logs: &[Log], ) -> Result<(), Fault> { let client = CowClient::with_transport(spy.clone()); - complete(on_chain_logs(host, &client, chain_id, logs)) - .expect("guest futures complete in one poll") + match poll_once(on_chain_logs(host, &client, chain_id, logs)) { + std::task::Poll::Ready(output) => output, + std::task::Poll::Pending => panic!("guest futures complete in one poll"), + } } fn open_status() -> Vec {