diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index c5d2fa81..c2ca1d19 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -78,7 +78,7 @@ jobs: - uses: taiki-e/install-action@43aecc8d72668fbcfe75c31400bc4f890f1c5853 # v2.83.2 with: tool: nextest - # Build all 16 guest module wasms ONCE (release/wasm32-wasip2): the single + # Build all 17 guest module wasms ONCE (release/wasm32-wasip2): the single # source of truth for guest buildability and the artifacts the integration # tests load. Replaces the deleted 9-way build-module matrix, which recompiled # the shared wasm dependency graph ~9x cold. Per-module size report folded in; @@ -88,7 +88,7 @@ jobs: cargo build --release --target wasm32-wasip2 --locked \ -p example -p twap-monitor -p ethflow-watcher -p price-alert \ -p balance-tracker -p stop-loss -p http-probe -p echo-venue \ - -p echo-client -p clock-reader -p flaky-bomb -p flaky-venue \ + -p echo-client -p echo-keeper -p clock-reader -p flaky-bomb -p flaky-venue \ -p fuel-bomb -p memory-bomb -p panic-bomb -p slow-host { echo "### module .wasm sizes" diff --git a/Cargo.lock b/Cargo.lock index 5919a1ae..171b17d9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2159,6 +2159,15 @@ dependencies = [ "wit-bindgen 0.58.0", ] +[[package]] +name = "echo-keeper" +version = "0.1.0" +dependencies = [ + "nexum-sdk", + "videre-sdk", + "wit-bindgen 0.59.0", +] + [[package]] name = "echo-venue" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index a21dfb89..94377aa5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -24,6 +24,7 @@ members = [ "modules/example", "modules/examples/balance-tracker", "modules/examples/echo-client", + "modules/examples/echo-keeper", "modules/examples/echo-venue", "modules/examples/http-probe", "modules/examples/price-alert", diff --git a/crates/cow-venue/src/client.rs b/crates/cow-venue/src/client.rs index b8d4b302..274906c7 100644 --- a/crates/cow-venue/src/client.rs +++ b/crates/cow-venue/src/client.rs @@ -1,65 +1,40 @@ -//! The typed CoW intent client. +//! The CoW venue as a keeper types it. //! -//! [`CowClient`] binds the keeper-facing [`IntentClient`] to the CoW -//! venue id and speaks the venue's own [`CowIntentBody`] over it, so -//! keeper code submits a typed CoW body without naming the venue on -//! every call or handling wire bytes. The classification API +//! [`CowVenue`] names the venue once - the id its adapter registers +//! under and the [`CowIntentBody`] schema it decodes - so keeper code +//! drives it through [`VenueClient`] with typed bodies, never wire +//! bytes. The classification API //! ([`classify`](crate::classification::classify)) travels in the same //! slice so the client that submits an order and the table that //! classifies its rejection version together. -use videre_sdk::client::{ClientError, IntentClient, VenueClient, VenueId}; -use videre_sdk::{IntentStatus, SubmitOutcome}; +use videre_sdk::client::{HostVenues, Venue, VenueClient, VenueId}; use crate::body::CowIntentBody; -/// The venue id the CoW adapter registers under and the registry resolves. -/// Every [`CowClient`] call routes here. -pub const VENUE: &str = "cow"; +/// The CoW venue marker: every [`CowClient`] call routes to +/// [`Venue::ID`] and encodes a [`CowIntentBody`]. +#[derive(Clone, Copy, Debug)] +pub struct CowVenue; -/// A typed intent client pre-bound to the CoW venue. A thin newtype over -/// [`IntentClient`] that fixes the venue id and the body type so callers -/// cannot mis-route or submit a foreign body. -#[derive(Clone, Debug)] -pub struct CowClient

{ - inner: IntentClient

, +impl Venue for CowVenue { + const ID: VenueId = VenueId::from_static("cow"); + type Body = CowIntentBody; } -impl CowClient

{ - /// Bind a client handle to the CoW venue. - pub fn new(venues: P) -> Self { - Self { - inner: IntentClient::new(venues, VENUE), - } - } - - /// The venue id every call routes to (always [`VENUE`]). - pub fn venue(&self) -> &VenueId { - self.inner.venue() - } - - /// Encode a typed CoW body and submit it to the venue. - pub fn submit(&self, body: &CowIntentBody) -> Result { - self.inner.submit(body) - } - - /// Report where a previously submitted intent is in its life. - pub fn status(&self, receipt: &[u8]) -> Result { - self.inner.status(receipt) - } - - /// Ask the venue to withdraw an intent. - pub fn cancel(&self, receipt: &[u8]) -> Result<(), ClientError> { - self.inner.cancel(receipt) - } -} +/// A typed client pre-bound to the CoW venue: callers cannot mis-route +/// or submit a foreign body. +pub type CowClient = VenueClient; #[cfg(test)] mod tests { - use super::*; use std::cell::RefCell; use std::rc::Rc; - use videre_sdk::VenueFault; + + use videre_sdk::client::VenueTransport; + use videre_sdk::{IntentStatus, Quotation, SubmitOutcome, VenueFault}; + + use super::*; /// One recorded submit: the venue it routed to and the wire bytes. type SubmitLog = Rc)>>>; @@ -72,27 +47,31 @@ mod tests { submitted: SubmitLog, } - impl VenueClient for SpyClient { - fn quote( - &self, - _venue: &VenueId, - _body: Vec, - ) -> Result { + impl VenueTransport for SpyClient { + async fn quote(&self, _venue: &VenueId, _body: Vec) -> Result { unreachable!("quote not exercised") } - fn submit(&self, venue: &VenueId, body: Vec) -> Result { + async fn submit( + &self, + venue: &VenueId, + body: Vec, + ) -> Result { self.submitted .borrow_mut() .push((venue.to_string(), body.clone())); Ok(SubmitOutcome::Accepted(body)) } - fn status(&self, _venue: &VenueId, _receipt: &[u8]) -> Result { + async fn status( + &self, + _venue: &VenueId, + _receipt: &[u8], + ) -> Result { unreachable!("status not exercised") } - fn cancel(&self, _venue: &VenueId, _receipt: &[u8]) -> Result<(), VenueFault> { + async fn cancel(&self, _venue: &VenueId, _receipt: &[u8]) -> Result<(), VenueFault> { unreachable!("cancel not exercised") } } @@ -124,13 +103,15 @@ mod tests { let body = sample_body(); let expected = body.to_bytes().expect("body encodes"); - let client = CowClient::new(spy.clone()); - assert_eq!(client.venue().as_str(), VENUE); - client.submit(&body).expect("submit succeeds"); + 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 calls = spy.submitted.borrow(); assert_eq!(calls.len(), 1); - assert_eq!(calls[0].0, VENUE); + assert_eq!(calls[0].0, CowVenue::ID.as_str()); assert_eq!(calls[0].1, expected); } } diff --git a/crates/cow-venue/src/lib.rs b/crates/cow-venue/src/lib.rs index 147d4920..3b4054fb 100644 --- a/crates/cow-venue/src/lib.rs +++ b/crates/cow-venue/src/lib.rs @@ -64,4 +64,4 @@ pub use order::{BuyTokenDestination, OrderBody, OrderKind, SellTokenSource}; #[cfg(feature = "client")] pub use classification::{ClassificationTable, classify, is_already_submitted}; #[cfg(feature = "client")] -pub use client::{CowClient, VENUE}; +pub use client::{CowClient, CowVenue}; diff --git a/crates/videre-host/tests/platform.rs b/crates/videre-host/tests/platform.rs index 67ea2120..9c44c425 100644 --- a/crates/videre-host/tests/platform.rs +++ b/crates/videre-host/tests/platform.rs @@ -580,6 +580,94 @@ async fn e2e_echo_module_registry_adapter_round_trip() { ); } +/// The blessed keeper path over the same two real components: the +/// echo-keeper module (built by `#[videre_sdk::keeper]`) drives the +/// echo-venue adapter through the typed `VenueClient` - +/// quote, submit, status, cancel, all with a typed body - and receives +/// the fulfilled `intent-status` the registry polls back. Proves the +/// macro-emitted worker and the typed client end to end, with no +/// hand-written byte marshalling on the keeper side. +#[tokio::test] +async fn e2e_keeper_module_drives_the_venue_through_the_typed_client() { + let (Some(adapter_wasm), Some(module_wasm)) = ( + module_wasm_or_skip("echo-venue"), + module_wasm_or_skip("echo-keeper"), + ) else { + return; + }; + + let chain = MockChainProvider::new(); + chain.on_method(ChainMethod::EthBlockNumber, "\"0x1\""); + let components = nexum_runtime::test_utils::mock_components_from(chain, MockStateStore::new()); + let logs = components.logs.clone(); + + let engine = make_wasmtime_engine(); + let config = EngineConfig { + adapters: vec![AdapterEntry { + path: adapter_wasm, + manifest: Some(workspace_path("modules/examples/echo-venue/module.toml")), + http_allow: Vec::new(), + messaging_topics: Vec::new(), + }], + modules: vec![ModuleEntry { + path: module_wasm, + manifest: Some(workspace_path("modules/examples/echo-keeper/module.toml")), + }], + ..Default::default() + }; + let videre = Arc::new(platform(&config)); + let extensions = videre_assembly(&videre); + let linker = make_linker(&engine, &extensions); + + let mut supervisor = + Supervisor::boot(&engine, &linker, &config, &components, &extensions, None) + .await + .expect("boot"); + assert_eq!( + supervisor.adapter_alive_count(), + 1, + "echo-venue is routable" + ); + assert_eq!(supervisor.alive_count(), 1, "echo-keeper is alive"); + + // One block drives the keeper's async on_block: quote, submit, + // status, cancel, all through the typed client. + assert_eq!(supervisor.dispatch_block(block(1)).await, 1); + + // The accepted receipt is under status watch; echo settles + // instantly, so the first poll fans the terminal status back. + let registry = registry_of(&supervisor); + let mut delivered = 0; + for _ in 0..2 { + for update in registry.poll_status_transitions().await { + assert_eq!(update.venue, "echo-venue"); + delivered += supervisor + .dispatch_extension_event(status_event(update)) + .await; + } + } + assert_eq!(delivered, 1, "one terminal status delivered to the keeper"); + assert_eq!(supervisor.alive_count(), 1, "keeper must remain alive"); + + // Every typed verb observably ran. + let runs = logs.list_runs("echo-keeper"); + assert_eq!(runs.len(), 1, "one run recorded for echo-keeper"); + let page = logs.read(&runs[0].run, 0); + let messages: Vec<&str> = page.records.iter().map(|r| r.message.as_str()).collect(); + for needle in [ + "quoted at echo-venue", + "submitted to echo-venue", + "status at echo-venue", + "cancelled at echo-venue", + "intent status from venue echo-venue", + ] { + assert!( + messages.iter().any(|m| m.contains(needle)), + "missing `{needle}`; records were: {messages:?}", + ); + } +} + /// The body-version handshake refuses a mismatched pair: an adapter /// decoding only v1 against a keeper encoding v2 fails the boot at the /// keeper's install, before instantiation, naming both sides' versions. diff --git a/crates/videre-macros/Cargo.toml b/crates/videre-macros/Cargo.toml index 31f0b9a3..7bb32f49 100644 --- a/crates/videre-macros/Cargo.toml +++ b/crates/videre-macros/Cargo.toml @@ -4,7 +4,7 @@ version = "0.1.0" edition.workspace = true license.workspace = true repository.workspace = true -description = "Proc-macro glue for videre venue adapters: #[venue] turns an impl VenueAdapter into the per-cdylib wit-bindgen and adapter export; derive(IntentBody) emits the versioned body codec." +description = "Proc-macro glue for the videre personas: #[venue] turns an impl VenueAdapter into the per-cdylib wit-bindgen and adapter export; #[keeper] emits the worker world wired to the typed venue client; derive(IntentBody) emits the versioned body codec." [lib] proc-macro = true diff --git a/crates/videre-macros/src/keeper.rs b/crates/videre-macros/src/keeper.rs new file mode 100644 index 00000000..568796d8 --- /dev/null +++ b/crates/videre-macros/src/keeper.rs @@ -0,0 +1,292 @@ +//! Expansion for `#[keeper]`: the keeper-worker mirror of `#[module]`. +//! +//! Same world synthesis and event dispatch as the plain module macro, +//! 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 +//! fault so `?` works in handlers. + +use proc_macro2::TokenStream; +use quote::quote; +use syn::{ImplItem, ItemImpl}; + +/// The handler names recognised on a `#[keeper]` impl; the `#[module]` +/// set, since a keeper is a plain worker. +const HANDLERS: [&str; 6] = [ + "init", + "on_block", + "on_chain_logs", + "on_tick", + "on_message", + "on_intent_status", +]; + +/// The manifest capability granting the client import. +const CLIENT_CAPABILITY: &str = "client"; + +/// The import the `client` capability must map to. +const CLIENT_IMPORT: &str = "videre:venue/client@0.1.0"; + +/// WIT packages the client import needs on the resolve path, in +/// dependency order. +const CLIENT_PACKAGES: [&str; 3] = ["videre-value-flow", "videre-types", "videre-venue"]; + +/// The fault detail for a handler future that suspended. +const SUSPENDED: &str = "keeper handler suspended: guest futures complete in one poll"; + +/// Expand the handler impl into the keeper module glue, or a compile +/// error naming the rule the input broke. +pub(crate) fn expand(input: &ItemImpl) -> syn::Result { + let self_ty = &input.self_ty; + if !crate::is_plain_type(self_ty) { + return Err(syn::Error::new_spanned( + self_ty, + "#[videre_sdk::keeper] must be applied to an inherent impl of a named type", + )); + } + if let Some((_, trait_path, _)) = &input.trait_ { + return Err(syn::Error::new_spanned( + trait_path, + "#[videre_sdk::keeper] must be applied to an inherent impl, not a trait impl", + )); + } + if !input.generics.params.is_empty() { + return Err(syn::Error::new_spanned( + &input.generics, + "#[videre_sdk::keeper] must be applied to a non-generic impl", + )); + } + + // Reserve the `on_` prefix for the recognised handler set, exactly + // as `#[module]` does: a typo'd handler must not silently no-op. + for item in &input.items { + if let ImplItem::Fn(f) = item { + let name = f.sig.ident.to_string(); + if name.starts_with("on_") && !HANDLERS.contains(&name.as_str()) { + return Err(syn::Error::new_spanned( + &f.sig.ident, + format!( + "`{name}` is not a recognised #[videre_sdk::keeper] handler; expected one \ + of {HANDLERS:?} (rename helpers so they do not start with `on_`)" + ), + )); + } + } + } + + // Present handlers with their asyncness: async ones are completed + // on the synchronous guest boundary by the emitted dispatch. + let present: Vec<(&str, bool)> = input + .items + .iter() + .filter_map(|item| match item { + ImplItem::Fn(f) => { + let name = f.sig.ident.to_string(); + HANDLERS + .into_iter() + .find(|h| *h == name) + .map(|h| (h, f.sig.asyncness.is_some())) + } + _ => None, + }) + .collect(); + if present.is_empty() { + return Err(syn::Error::new_spanned( + self_ty, + "#[videre_sdk::keeper] found no recognised handlers on this impl; define at least one \ + of `init`, `on_block`, `on_chain_logs`, `on_tick`, `on_message`, `on_intent_status`", + )); + } + let handler = |name: &str| present.iter().find(|(h, _)| *h == name).copied(); + + let (anchors, module_world) = derive_keeper_world() + .map_err(|msg| syn::Error::new(proc_macro2::Span::call_site(), msg))?; + let wit_paths = crate::resolve_wit_packages(&module_world.packages) + .map_err(|msg| syn::Error::new(proc_macro2::Span::call_site(), msg))?; + let inline_world = &module_world.wit; + let adapter_caps: Vec = module_world + .adapters + .iter() + .map(|cap| syn::Ident::new(cap, proc_macro2::Span::call_site())) + .collect(); + + // Complete an async handler's future in one poll; a suspension is a + // 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( + nexum::host::types::Fault::Internal( + ::std::string::String::from(#SUSPENDED), + ), + ), + } + } + }; + + let init_impl = match handler("init") { + Some((_, is_async)) => { + let call = quote! { <#self_ty>::init(config) }; + let body = if is_async { drive(call) } else { call }; + quote! { + fn init( + config: ::std::vec::Vec<(::std::string::String, ::std::string::String)>, + ) -> ::core::result::Result<(), Fault> { + #body + } + } + } + None => quote! { + fn init( + _config: ::std::vec::Vec<(::std::string::String, ::std::string::String)>, + ) -> ::core::result::Result<(), Fault> { + ::core::result::Result::Ok(()) + } + }, + }; + + let arm = |name: &str, variant: &str| -> TokenStream { + let variant = syn::Ident::new(variant, proc_macro2::Span::call_site()); + match handler(name) { + Some((_, is_async)) => { + let call = syn::Ident::new(name, proc_macro2::Span::call_site()); + let call = quote! { <#self_ty>::#call(payload) }; + let body = if is_async { drive(call) } else { call }; + quote! { nexum::host::types::Event::#variant(payload) => #body, } + } + None => quote! { + nexum::host::types::Event::#variant(_) => ::core::result::Result::Ok(()), + }, + } + }; + let block_arm = arm("on_block", "Block"); + let logs_arm = arm("on_chain_logs", "ChainLogs"); + let tick_arm = arm("on_tick", "Tick"); + let message_arm = arm("on_message", "Message"); + let intent_status_arm = arm("on_intent_status", "IntentStatus"); + + Ok(quote! { + // Anchor a rebuild on the manifest and the extension registry: + // the emitted world is derived from them. + #(const _: &[u8] = ::core::include_bytes!(#anchors);)* + + wit_bindgen::generate!({ + inline: #inline_world, + path: [#(#wit_paths),*], + world: "nexum:module-world/module", + generate_all, + with: { + "videre:types/types@0.1.0": ::videre_sdk::bindings::videre::types::types, + "videre:value-flow/types@0.1.0": + ::videre_sdk::bindings::videre::value_flow::types, + "videre:venue/client@0.1.0": + ::videre_sdk::bindings::videre::venue::client, + }, + }); + + ::nexum_sdk::bind_host_via_wit_bindgen!(caps: [#(#adapter_caps),*]); + + #input + + // Folds a typed client failure into the wire fault, so `?` + // applies to client calls inside handlers. + impl ::core::convert::From<::videre_sdk::ClientError> for nexum::host::types::Fault { + fn from(err: ::videre_sdk::ClientError) -> Self { + ::core::convert::Into::into(::nexum_sdk::host::Fault::from(err)) + } + } + + #[doc(hidden)] + struct __VidereKeeperExport; + + impl Guest for __VidereKeeperExport { + #init_impl + + fn on_event(event: nexum::host::types::Event) -> ::core::result::Result<(), Fault> { + match event { + #block_arm + #logs_arm + #tick_arm + #message_arm + #intent_status_arm + } + } + } + + export!(__VidereKeeperExport); + }) +} + +/// The canonical `client` extension row, injected when the composition +/// root's registry does not carry one. +fn client_row() -> nexum_world::ExtensionRow { + nexum_world::ExtensionRow { + name: CLIENT_CAPABILITY.to_owned(), + import: CLIENT_IMPORT.to_owned(), + packages: CLIENT_PACKAGES.map(str::to_owned).into(), + } +} + +/// Read the consuming crate's `module.toml`, require the worker shape +/// (no `[module] kind`) and the `client` capability, and synthesize the +/// per-module world with the client extension row guaranteed. Returns +/// the rebuild anchor paths alongside the world. +fn derive_keeper_world() -> Result<(Vec, nexum_world::ModuleWorld), String> { + let manifest_path = crate::manifest_dir()?.join("module.toml"); + let text = std::fs::read_to_string(&manifest_path).map_err(|e| { + format!( + "could not read {} ({e}); #[videre_sdk::keeper] derives the component's WIT world \ + from the manifest's [capabilities] section, so the manifest must sit next to \ + Cargo.toml", + manifest_path.display() + ) + })?; + if let Some(kind) = nexum_world::manifest_kind(&text) + .map_err(|e| format!("{}: {e}", manifest_path.display()))? + { + return Err(format!( + "{}: a #[videre_sdk::keeper] module is a plain worker; drop `[module] kind = \ + \"{kind}\"`", + manifest_path.display() + )); + } + let declared = nexum_world::manifest_capabilities(&text) + .map_err(|e| format!("{}: {e}", manifest_path.display()))?; + if !declared.iter().any(|cap| cap == CLIENT_CAPABILITY) { + return Err(format!( + "{}: a keeper drives venues through `{CLIENT_IMPORT}`; declare the \ + `{CLIENT_CAPABILITY}` capability under [capabilities]", + manifest_path.display() + )); + } + let manifest_path = manifest_path.to_string_lossy().into_owned(); + + let mut anchors = vec![manifest_path.clone()]; + let mut extensions = match nexum_world::find_extensions_manifest(&crate::manifest_dir()?) { + None => Vec::new(), + Some(registry) => { + let text = std::fs::read_to_string(®istry) + .map_err(|e| format!("could not read {}: {e}", registry.display()))?; + let rows = nexum_world::manifest_extensions(&text) + .map_err(|e| format!("{}: {e}", registry.display()))?; + anchors.push(registry.to_string_lossy().into_owned()); + rows + } + }; + match extensions.iter().find(|row| row.name == CLIENT_CAPABILITY) { + None => extensions.push(client_row()), + Some(row) if row.import == CLIENT_IMPORT => {} + Some(row) => { + return Err(format!( + "the registered `{CLIENT_CAPABILITY}` extension imports `{}`; \ + #[videre_sdk::keeper] requires `{CLIENT_IMPORT}`", + row.import + )); + } + } + let module_world = nexum_world::synthesize(&declared, &extensions) + .map_err(|e| format!("{manifest_path}: {e}"))?; + Ok((anchors, module_world)) +} diff --git a/crates/videre-macros/src/lib.rs b/crates/videre-macros/src/lib.rs index 77f83e0c..665fb88a 100644 --- a/crates/videre-macros/src/lib.rs +++ b/crates/videre-macros/src/lib.rs @@ -1,20 +1,27 @@ -//! Proc-macro glue for videre venue adapters. +//! Proc-macro glue for the two videre personas. //! //! [`venue`] is the single blessed venue authoring path: applied to an //! `impl VenueAdapter` block it emits the per-cdylib wit-bindgen for a //! manifest-derived world exporting `videre:venue/adapter`, asserts the //! manifest kind, and expands to the SDK's internal export codegen. //! +//! [`keeper`] is its worker mirror: applied to a handler impl it emits +//! the per-cdylib wit-bindgen for a manifest-derived module world, +//! wires the `videre:venue/client` import onto the SDK's shared shims, +//! and dispatches events to the handlers, completing async ones on the +//! synchronous guest boundary. +//! //! [`derive@IntentBody`] implements the venue SDK's versioned body codec //! over a per-venue version enum. //! -//! The module-side macro (`#[module]`) lives in `nexum-module-macros`. +//! The plain module macro (`#[module]`) lives in `nexum-module-macros`. //! //! Consumers reach these through the SDK re-exports -//! (`videre_sdk::venue`, `videre_sdk::IntentBody`) rather than -//! depending on this crate directly. +//! (`videre_sdk::venue`, `videre_sdk::keeper`, `videre_sdk::IntentBody`) +//! rather than depending on this crate directly. mod intent_body; +mod keeper; mod world; use proc_macro::TokenStream; @@ -166,15 +173,53 @@ pub fn venue(attr: TokenStream, item: TokenStream) -> TokenStream { .into() } +/// Generate the per-cdylib glue for a keeper: a worker that drives +/// venues through the typed client. +/// +/// The keeper-author mirror of `#[module]`. Apply to an `impl` block +/// 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 +/// handler can await the typed `VenueClient` directly. +/// +/// The macro reads the crate's `module.toml`, requires the `client` +/// capability (the `videre:venue/client` import is what makes a keeper +/// a keeper), synthesizes the per-module world exactly as `#[module]` +/// does, and remaps the videre interfaces onto the SDK bindings: the +/// module's client import resolves to the SDK's shared shims, so the +/// `VenueClient` a handler drives and the wire speak one set of types. +/// A `From` impl onto the wire fault is emitted, so `?` +/// applies to client calls inside handlers. +/// +/// The same crate-root resolution invariants as `#[module]` apply, and +/// the consuming crate must declare `wit-bindgen`, `videre-sdk`, and +/// `nexum-sdk` as direct dependencies. +#[proc_macro_attribute] +pub fn keeper(attr: TokenStream, item: TokenStream) -> TokenStream { + if !attr.is_empty() { + return syn::Error::new( + proc_macro2::Span::call_site(), + "#[videre_sdk::keeper] takes no arguments", + ) + .to_compile_error() + .into(); + } + let input = syn::parse_macro_input!(item as ItemImpl); + keeper::expand(&input) + .unwrap_or_else(syn::Error::into_compile_error) + .into() +} + /// Whether a type is a plain named path (`Foo`), the only shape a module /// export type may take. -fn is_plain_type(ty: &Type) -> bool { +pub(crate) fn is_plain_type(ty: &Type) -> bool { matches!(ty, Type::Path(tp) if tp.qself.is_none()) } /// The consuming crate's manifest directory, the root every crate-local /// lookup starts from. -fn manifest_dir() -> Result { +pub(crate) fn manifest_dir() -> Result { std::env::var("CARGO_MANIFEST_DIR") .map(std::path::PathBuf::from) .map_err(|_| "CARGO_MANIFEST_DIR is not set".to_string()) @@ -215,7 +260,7 @@ fn derive_venue_world() -> Result<(String, nexum_world::ModuleWorld), String> { /// Resolve each needed WIT package directory crate-locally (vendored /// `wit/deps/`, then own `wit/`), falling back through /// ancestors for the transitional monorepo layout. -fn resolve_wit_packages(packages: &[String]) -> Result, String> { +pub(crate) fn resolve_wit_packages(packages: &[String]) -> Result, String> { Ok( nexum_world::resolve_wit_packages(&manifest_dir()?, packages)? .into_iter() diff --git a/crates/videre-sdk/Cargo.toml b/crates/videre-sdk/Cargo.toml index 47b77c88..54701fd4 100644 --- a/crates/videre-sdk/Cargo.toml +++ b/crates/videre-sdk/Cargo.toml @@ -4,7 +4,7 @@ version = "0.1.0" edition.workspace = true license.workspace = true repository.workspace = true -description = "Guest-side videre SDK: the VenueAdapter trait over the venue-adapter world bindgen, the borsh-versioned IntentBody codec, the typed intent client core, the generic keeper sweep assembler, and typed wrappers over the scoped transport imports." +description = "Guest-side videre SDK: the VenueAdapter trait mirroring the venue-adapter world, the borsh-versioned IntentBody codec, the typed venue client over the native-AFIT transport seam, the generic keeper sweep assembler, and typed wrappers over the scoped transport imports." [lib] # Plain library - adapters link this and emit their own cdylib for the diff --git a/crates/videre-sdk/src/bindings.rs b/crates/videre-sdk/src/bindings.rs index eb426521..9f57dde8 100644 --- a/crates/videre-sdk/src/bindings.rs +++ b/crates/videre-sdk/src/bindings.rs @@ -1,22 +1,43 @@ -//! Guest bindings for the `videre:venue/venue-adapter` world. +//! Guest bindings for the videre SDK, generated once from an +//! import-only inline world carrying every interface both personas +//! speak: the videre type vocabulary, the host types and scoped +//! transport, and the keeper-facing `videre:venue/client` shims. //! //! Unlike event modules, which run `wit_bindgen::generate!` per cdylib, -//! the venue SDK generates the adapter world's bindings once, here: the +//! the SDK generates these bindings once: the //! [`VenueAdapter`](crate::VenueAdapter) trait, the typed transport -//! wrappers, and the intent client core are all expressed over these -//! types. The `#[videre_sdk::venue]` attribute's per-cdylib bindgen -//! remaps `videre:types/types`, `videre:value-flow/types`, and -//! `nexum:host/types` onto these modules with `with`, so a macro-built -//! adapter shares type identity with the SDK and the conformance kit. +//! wrappers, and the client core are all expressed over them. The +//! per-cdylib bindgens (`#[videre_sdk::venue]`, `#[videre_sdk::keeper]`) +//! remap the shared interfaces onto these modules with `with`, so a +//! macro-built component shares type identity (and, for the keeper's +//! client, one shim set) with the SDK and the conformance kit. +//! +//! The world is import-only on purpose: an embedded world section is +//! unioned into every linking module at componentization, where an +//! unused import prunes but an export is a hard obligation. Keeping the +//! adapter export out of the SDK world is what lets a keeper module +//! link this crate without being asked to be a venue; the export face +//! is emitted per-cdylib by `#[videre_sdk::venue]` alone. wit_bindgen::generate!({ + inline: "package videre:sdk-shims; + +world sdk-imports { + import videre:types/types@0.1.0; + import videre:value-flow/types@0.1.0; + import nexum:host/types@0.1.0; + import nexum:host/chain@0.1.0; + import nexum:host/messaging@0.1.0; + import videre:venue/client@0.1.0; +} +", path: [ "../../wit/videre-value-flow", "../../wit/videre-types", "../../wit/nexum-host", "../../wit/videre-venue", ], - world: "videre:venue/venue-adapter", + world: "videre:sdk-shims/sdk-imports", generate_all, additional_derives: [PartialEq], }); diff --git a/crates/videre-sdk/src/client.rs b/crates/videre-sdk/src/client.rs index 8652f849..baa40118 100644 --- a/crates/videre-sdk/src/client.rs +++ b/crates/videre-sdk/src/client.rs @@ -1,27 +1,38 @@ -//! The typed intent client core: [`IntentClient`] over the byte-level -//! [`VenueClient`] seam. +//! The typed venue client: [`VenueClient`] binds one [`Venue`] over the +//! byte-level [`VenueTransport`] seam. //! -//! The client boundary carries opaque bodies; this module is where a -//! typed body meets it. [`IntentClient`] binds one [`VenueId`] and -//! encodes through [`IntentBody`] before submission, so keeper code -//! never handles wire bytes. The seam is byte-level on purpose: the -//! strategy-module SDK implements [`VenueClient`] over its own -//! `videre:venue/client` import shims, tests implement it in memory -//! (an in-process adapter works directly), and the typed layer above is -//! shared by both. +//! The wire carries opaque bodies and a stringly venue selector; typing +//! is recovered here. A venue is named once, as a [`Venue`] marker +//! carrying its [`VenueId`] and body schema, and every call encodes +//! through [`IntentBody`] before the seam, so keeper code never handles +//! wire bytes. [`HostVenues`] is the seam bound to the module's own +//! `videre:venue/client` import; tests and in-process adapters +//! implement [`VenueTransport`] directly. The transport methods are +//! native AFIT, so dispatch is static and nothing on the call path +//! boxes. +use std::borrow::Cow; use std::fmt; +use std::future::Future; +use std::marker::PhantomData; use strum::IntoStaticStr; +use crate::bindings::videre::venue::client as shims; use crate::{BodyError, IntentBody, IntentStatus, Quotation, SubmitOutcome, VenueFault}; /// Venue identifier: the id an adapter registers under and every client /// call routes to. Opaque beyond equality. #[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)] -pub struct VenueId(String); +pub struct VenueId(Cow<'static, str>); impl VenueId { + /// Wrap a static id without allocating: the [`Venue::ID`] spelling. + #[must_use] + pub const fn from_static(id: &'static str) -> Self { + Self(Cow::Borrowed(id)) + } + /// The id at its wire spelling. #[must_use] pub fn as_str(&self) -> &str { @@ -31,13 +42,13 @@ impl VenueId { impl From for VenueId { fn from(id: String) -> Self { - Self(id) + Self(Cow::Owned(id)) } } impl From<&str> for VenueId { fn from(id: &str) -> Self { - Self(id.to_owned()) + Self(Cow::Owned(id.to_owned())) } } @@ -53,52 +64,126 @@ impl fmt::Display for VenueId { } } -/// Byte-level access to the keeper-facing `videre:venue/client` -/// interface, the venue named per call. -pub trait VenueClient { +/// One venue as a keeper types it: the id its adapter registers under +/// and the body schema it decodes. Implement on a unit marker +/// (`struct CowVenue;`) and drive it through [`VenueClient`]. +pub trait Venue { + /// The id the venue's adapter registers under. + const ID: VenueId; + + /// The versioned body schema the venue decodes. + type Body: IntentBody; +} + +/// The byte-level seam under the typed client: `videre:venue/client` +/// with the venue named per call. Native AFIT, so a [`VenueClient`] +/// over any transport dispatches statically. [`HostVenues`] binds it to +/// the module's own import; tests implement it in memory. +pub trait VenueTransport { /// Price an opaque intent body at the named venue. - fn quote(&self, venue: &VenueId, body: Vec) -> Result; + fn quote( + &self, + venue: &VenueId, + body: Vec, + ) -> impl Future>; /// Submit an opaque intent body to the named venue. - fn submit(&self, venue: &VenueId, body: Vec) -> Result; + fn submit( + &self, + venue: &VenueId, + body: Vec, + ) -> impl Future>; /// Report where a previously submitted intent is in its life. - fn status(&self, venue: &VenueId, receipt: &[u8]) -> Result; + fn status( + &self, + venue: &VenueId, + receipt: &[u8], + ) -> impl Future>; /// Ask the venue to withdraw an intent. Success means the venue /// accepted the cancellation, not that an in-flight settlement can /// no longer win the race. - fn cancel(&self, venue: &VenueId, receipt: &[u8]) -> Result<(), VenueFault>; + fn cancel( + &self, + venue: &VenueId, + receipt: &[u8], + ) -> impl Future>; +} + +/// The module's `videre:venue/client` import behind the +/// [`VenueTransport`] seam: the transport every guest-side +/// [`VenueClient`] defaults to. +#[derive(Clone, Copy, Debug, Default)] +pub struct HostVenues; + +impl VenueTransport for HostVenues { + async fn quote(&self, venue: &VenueId, body: Vec) -> Result { + shims::quote(venue.as_str(), &body).map_err(VenueFault::from) + } + + async fn submit(&self, venue: &VenueId, body: Vec) -> Result { + shims::submit(venue.as_str(), &body).map_err(VenueFault::from) + } + + async fn status(&self, venue: &VenueId, receipt: &[u8]) -> Result { + shims::status(venue.as_str(), receipt).map_err(VenueFault::from) + } + + async fn cancel(&self, venue: &VenueId, receipt: &[u8]) -> Result<(), VenueFault> { + shims::cancel(venue.as_str(), receipt).map_err(VenueFault::from) + } +} + +/// A typed client bound to one [`Venue`]: encodes the venue's +/// [`IntentBody`] to wire bytes and forwards through the +/// [`VenueTransport`] seam under [`Venue::ID`]. Zero-sized over the +/// default [`HostVenues`] transport. +pub struct VenueClient { + transport: T, + venue: PhantomData, +} + +impl VenueClient { + /// Bind the venue over the module's own `videre:venue/client` + /// import. + #[must_use] + pub const fn new() -> Self { + Self { + transport: HostVenues, + venue: PhantomData, + } + } } -/// A typed intent client bound to one venue: encodes an [`IntentBody`] -/// to wire bytes and forwards through the [`VenueClient`] seam. -#[derive(Clone, Debug)] -pub struct IntentClient

{ - venues: P, - venue: VenueId, +impl Default for VenueClient { + fn default() -> Self { + Self::new() + } } -impl IntentClient

{ - /// Bind a client handle to the venue id the registry resolves. - pub fn new(venues: P, venue: impl Into) -> Self { +impl VenueClient { + /// Bind the venue over a caller-supplied transport (tests, + /// in-process adapters). + pub const fn with_transport(transport: T) -> Self { Self { - venues, - venue: venue.into(), + transport, + venue: PhantomData, } } - /// The venue every call on this client routes to. - pub fn venue(&self) -> &VenueId { - &self.venue + /// The venue id every call on this client routes to. + #[must_use] + pub fn venue(&self) -> VenueId { + V::ID } - /// Encode a 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. - pub fn quote(&self, body: &B) -> Result, ClientError> { + /// 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. + pub async fn quote(&self, body: &V::Body) -> Result, ClientError> { let bytes = body.to_bytes()?; - let quotation = self.venues.quote(&self.venue, bytes.clone())?; + let quotation = self.transport.quote(&V::ID, bytes.clone()).await?; Ok(Quoted { client: self, bytes, @@ -106,47 +191,74 @@ impl IntentClient

{ }) } - /// Encode a typed body and submit it to the bound venue. - pub fn submit(&self, body: &B) -> Result { + /// Encode the typed body and submit it to the bound venue. + pub async fn submit(&self, body: &V::Body) -> Result { let bytes = body.to_bytes()?; - Ok(self.venues.submit(&self.venue, bytes)?) + Ok(self.transport.submit(&V::ID, bytes).await?) } /// Report where a previously submitted intent is in its life. - pub fn status(&self, receipt: &[u8]) -> Result { - Ok(self.venues.status(&self.venue, receipt)?) + pub async fn status(&self, receipt: &[u8]) -> Result { + Ok(self.transport.status(&V::ID, receipt).await?) } /// Ask the bound venue to withdraw an intent. - pub fn cancel(&self, receipt: &[u8]) -> Result<(), ClientError> { - Ok(self.venues.cancel(&self.venue, receipt)?) + pub async fn cancel(&self, receipt: &[u8]) -> Result<(), ClientError> { + Ok(self.transport.cancel(&V::ID, receipt).await?) + } +} + +impl Clone for VenueClient { + fn clone(&self) -> Self { + Self { + transport: self.transport.clone(), + venue: PhantomData, + } + } +} + +impl fmt::Debug for VenueClient { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("VenueClient") + .field("venue", &V::ID) + .finish_non_exhaustive() } } /// A priced intent: the quotation plus the exact bytes it prices, bound -/// to the client that fetched it. Consuming it with [`submit`](Self::submit) -/// is the only way from a quote to a submission, so a keeper cannot -/// submit a body other than the one quoted. -#[derive(Debug)] -pub struct Quoted<'a, P> { - client: &'a IntentClient

, +/// to the client that fetched it. Consuming it with +/// [`submit`](Self::submit) is the only way from a quote to a +/// submission, so a keeper cannot submit a body other than the one +/// quoted. +pub struct Quoted<'a, V: Venue, T: VenueTransport> { + client: &'a VenueClient, bytes: Vec, quotation: Quotation, } -impl Quoted<'_, P> { +impl Quoted<'_, V, T> { /// The venue's indicative quotation for the body. + #[must_use] pub fn quotation(&self) -> &Quotation { &self.quotation } /// Submit the quoted body to the venue that priced it. - pub fn submit(self) -> Result { - Ok(self.client.venues.submit(&self.client.venue, self.bytes)?) + pub async fn submit(self) -> Result { + Ok(self.client.transport.submit(&V::ID, self.bytes).await?) + } +} + +impl fmt::Debug for Quoted<'_, V, T> { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("Quoted") + .field("venue", &V::ID) + .field("quotation", &self.quotation) + .finish_non_exhaustive() } } -/// Why a typed intent call failed: before the wire (the body failed to +/// Why a typed client call failed: before the wire (the body failed to /// encode) or beyond it (the registry or venue refused). /// /// `IntoStaticStr` yields a snake_case label per case for log and diff --git a/crates/videre-sdk/src/faults.rs b/crates/videre-sdk/src/faults.rs index 9e751747..00d6840c 100644 --- a/crates/videre-sdk/src/faults.rs +++ b/crates/videre-sdk/src/faults.rs @@ -13,6 +13,7 @@ use nexum_sdk::host; use strum::IntoStaticStr; use crate::bindings::nexum::host::types::RateLimit as WireRateLimit; +use crate::client::ClientError; use crate::{Fault, RateLimit, VenueError}; /// Owned mirror of the wire `venue-error` with `Display`: what typed @@ -130,6 +131,28 @@ impl From for VenueError { } } +/// Fold a typed client failure into the SDK-neutral fault a keeper +/// handler returns: an encode failure and a misnamed venue are the +/// caller's `invalid-input`; venue refusals map structurally. +impl From for host::Fault { + fn from(err: ClientError) -> Self { + match err { + ClientError::Body(body) => host::Fault::InvalidInput(body.to_string()), + ClientError::Venue(fault) => match fault { + VenueFault::UnknownVenue => host::Fault::InvalidInput(fault.to_string()), + VenueFault::InvalidBody(s) => host::Fault::InvalidInput(s), + VenueFault::Unsupported => host::Fault::Unsupported(fault.to_string()), + VenueFault::Denied(s) => host::Fault::Denied(s), + VenueFault::RateLimited { retry_after_ms } => { + host::Fault::RateLimited(host::RateLimit { retry_after_ms }) + } + VenueFault::Unavailable(s) => host::Fault::Unavailable(s), + VenueFault::Timeout => host::Fault::Timeout, + }, + } + } +} + /// Fold a wasi:http fetch failure into the venue error an intent /// function returns: an allowlist refusal stays `denied`, a timeout is /// `timeout`, and transport failures (including a request the adapter diff --git a/crates/videre-sdk/src/keeper.rs b/crates/videre-sdk/src/keeper.rs index 23d2ae5d..3240faed 100644 --- a/crates/videre-sdk/src/keeper.rs +++ b/crates/videre-sdk/src/keeper.rs @@ -1,7 +1,7 @@ //! The generic keeper sweep: one pass assembling the world-neutral //! stores - [`WatchSet`] to [`Gates`] to [`ConditionalSource::poll`] to //! [`Retrier`] to [`Journal`] - and routing submissions through the -//! [`VenueClient`] seam. +//! [`VenueTransport`] seam. //! //! [`Sweep`] is the shared poll outcome: the concrete //! [`ConditionalSource::Outcome`] a keeper's sources produce so @@ -15,7 +15,7 @@ use nexum_sdk::keeper::{ }; use nexum_sdk::prelude::{hex, keccak256}; -use crate::client::{VenueClient, VenueId}; +use crate::client::{VenueId, VenueTransport}; use crate::{SubmitOutcome, UnsignedTx, VenueFault}; /// What one poll asks the sweep to do with its watch. @@ -59,9 +59,9 @@ impl Keeper { } } -impl Keeper { +impl Keeper { /// Sweep the watch set once at `tick`: poll every ready watch, - /// submit [`Sweep::Submit`] bodies through the venue client, and + /// submit [`Sweep::Submit`] bodies through the venue seam, and /// run every other outcome and every venue refusal through the /// [`Retrier`]. A venue-and-body key is checked against the /// `submitted:` [`Journal`] before every submit and recorded on @@ -69,7 +69,7 @@ impl Keeper { /// a `requires-signing` answer journals nothing and is surfaced /// afresh each sweep. Store faults abort the sweep; venue refusals /// never do - they fold into per-watch retry actions. - pub fn sweep(&self, host: &H, tick: &Tick) -> Result + pub async fn sweep(&self, host: &H, tick: &Tick) -> Result where H: LocalStoreHost, S: ConditionalSource, @@ -102,7 +102,7 @@ impl Keeper { report.duplicates += 1; continue; } - match self.venues.submit(&self.venue, body) { + match self.venues.submit(&self.venue, body).await { Ok(SubmitOutcome::Accepted(_)) => { journal.record(&key)?; report.submitted += 1; @@ -192,9 +192,14 @@ mod tests { use nexum_sdk_test::MockLocalStore; use super::{Keeper, Sweep, SweepReport}; - use crate::client::{VenueClient, VenueId}; + use crate::client::{VenueId, VenueTransport}; use crate::{IntentStatus, Quotation, SubmitOutcome, UnsignedTx, VenueFault}; + /// 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") + } + /// Answers every poll with one programmed outcome. struct StubSource(Sweep); @@ -232,21 +237,29 @@ mod tests { } } - impl VenueClient for &StubVenue { - fn quote(&self, _venue: &VenueId, _body: Vec) -> Result { + impl VenueTransport for &StubVenue { + async fn quote(&self, _venue: &VenueId, _body: Vec) -> Result { unreachable!("quote not exercised") } - fn submit(&self, _venue: &VenueId, body: Vec) -> Result { + async fn submit( + &self, + _venue: &VenueId, + body: Vec, + ) -> Result { self.submitted.borrow_mut().push(body); self.outcome.clone() } - fn status(&self, _venue: &VenueId, _receipt: &[u8]) -> Result { + async fn status( + &self, + _venue: &VenueId, + _receipt: &[u8], + ) -> Result { unreachable!("status not exercised") } - fn cancel(&self, _venue: &VenueId, _receipt: &[u8]) -> Result<(), VenueFault> { + async fn cancel(&self, _venue: &VenueId, _receipt: &[u8]) -> Result<(), VenueFault> { unreachable!("cancel not exercised") } } @@ -274,7 +287,7 @@ mod tests { let venue = StubVenue::new(Ok(SubmitOutcome::Accepted(vec![0xA5, 0x5A]))); let keeper = keeper(Sweep::Submit(b"body".to_vec()), &venue); - let report = keeper.sweep(&host, &TICK).expect("sweep runs"); + let report = run(keeper.sweep(&host, &TICK)).expect("sweep runs"); assert_eq!(report.polled, 1); assert_eq!(report.submitted, 1); assert_eq!(venue.submitted.borrow().as_slice(), [b"body".to_vec()]); @@ -285,7 +298,7 @@ mod tests { assert_eq!(WatchSet::new(&host).list().expect("list reads").len(), 1); // A later sweep re-polls the watch but never re-posts the body. - let report = keeper.sweep(&host, &TICK).expect("sweep runs"); + let report = run(keeper.sweep(&host, &TICK)).expect("sweep runs"); assert_eq!(report.submitted, 0); assert_eq!(report.duplicates, 1); assert_eq!(venue.submitted.borrow().len(), 1); @@ -303,8 +316,18 @@ mod tests { ])); let keeper = Keeper::new(source, &venue, "stub"); - assert_eq!(keeper.sweep(&host, &TICK).expect("sweep runs").submitted, 1); - assert_eq!(keeper.sweep(&host, &TICK).expect("sweep runs").submitted, 1); + assert_eq!( + run(keeper.sweep(&host, &TICK)) + .expect("sweep runs") + .submitted, + 1 + ); + assert_eq!( + run(keeper.sweep(&host, &TICK)) + .expect("sweep runs") + .submitted, + 1 + ); assert_eq!( venue.submitted.borrow().as_slice(), [b"one".to_vec(), b"two".to_vec()] @@ -324,13 +347,13 @@ mod tests { let venue = StubVenue::new(Ok(SubmitOutcome::RequiresSigning(tx.clone()))); let keeper = keeper(Sweep::Submit(b"body".to_vec()), &venue); - let report = keeper.sweep(&host, &TICK).expect("sweep runs"); + let report = run(keeper.sweep(&host, &TICK)).expect("sweep runs"); assert_eq!(report.unsigned, vec![tx.clone()]); assert_eq!(report.submitted, 0); // Nothing accepted, nothing journalled: the next sweep // surfaces the same transaction again. - let report = keeper.sweep(&host, &TICK).expect("sweep runs"); + let report = run(keeper.sweep(&host, &TICK)).expect("sweep runs"); assert_eq!(report.unsigned, vec![tx]); } @@ -344,8 +367,7 @@ mod tests { .expect("gate writes"); let venue = StubVenue::new(Ok(SubmitOutcome::Accepted(vec![1]))); - let report = keeper(Sweep::Submit(b"body".to_vec()), &venue) - .sweep(&host, &TICK) + let report = run(keeper(Sweep::Submit(b"body".to_vec()), &venue).sweep(&host, &TICK)) .expect("sweep runs"); assert_eq!(report.gated, 1); assert_eq!(report.polled, 0); @@ -358,9 +380,7 @@ mod tests { put_watch(&host); let venue = StubVenue::new(Ok(SubmitOutcome::Accepted(vec![1]))); - let report = keeper(Sweep::Drop, &venue) - .sweep(&host, &TICK) - .expect("sweep runs"); + let report = run(keeper(Sweep::Drop, &venue).sweep(&host, &TICK)).expect("sweep runs"); assert_eq!(report.dropped, 1); assert!(WatchSet::new(&host).list().expect("list reads").is_empty()); } @@ -372,11 +392,11 @@ mod tests { let venue = StubVenue::new(Ok(SubmitOutcome::Accepted(vec![1]))); let keeper = keeper(Sweep::Backoff { seconds: 30 }, &venue); - let report = keeper.sweep(&host, &TICK).expect("sweep runs"); + let report = run(keeper.sweep(&host, &TICK)).expect("sweep runs"); assert_eq!(report.retried, 1); // Still inside the backoff window: gated, not polled. - let report = keeper.sweep(&host, &TICK).expect("sweep runs"); + let report = run(keeper.sweep(&host, &TICK)).expect("sweep runs"); assert_eq!(report.gated, 1); // At the threshold the gate opens again. @@ -384,7 +404,7 @@ mod tests { epoch_s: TICK.epoch_s + 30, ..TICK }; - let report = keeper.sweep(&host, &later).expect("sweep runs"); + let report = run(keeper.sweep(&host, &later)).expect("sweep runs"); assert_eq!(report.polled, 1); } @@ -397,7 +417,7 @@ mod tests { })); let keeper = keeper(Sweep::Submit(b"body".to_vec()), &venue); - let report = keeper.sweep(&host, &TICK).expect("sweep runs"); + let report = run(keeper.sweep(&host, &TICK)).expect("sweep runs"); assert_eq!(report.retried, 1); // 2500 ms rounds up to a 3 s epoch gate. @@ -405,12 +425,18 @@ mod tests { epoch_s: TICK.epoch_s + 2, ..TICK }; - assert_eq!(keeper.sweep(&host, &at_2s).expect("sweep runs").gated, 1); + assert_eq!( + run(keeper.sweep(&host, &at_2s)).expect("sweep runs").gated, + 1 + ); let at_3s = Tick { epoch_s: TICK.epoch_s + 3, ..TICK }; - assert_eq!(keeper.sweep(&host, &at_3s).expect("sweep runs").polled, 1); + assert_eq!( + run(keeper.sweep(&host, &at_3s)).expect("sweep runs").polled, + 1 + ); } #[test] @@ -419,8 +445,7 @@ mod tests { put_watch(&host); let venue = StubVenue::new(Err(VenueFault::Denied("blocked".into()))); - let report = keeper(Sweep::Submit(b"body".to_vec()), &venue) - .sweep(&host, &TICK) + let report = run(keeper(Sweep::Submit(b"body".to_vec()), &venue).sweep(&host, &TICK)) .expect("sweep runs"); assert_eq!(report.dropped, 1); assert!(WatchSet::new(&host).list().expect("list reads").is_empty()); @@ -433,9 +458,12 @@ mod tests { let venue = StubVenue::new(Err(VenueFault::Unavailable("down".into()))); let keeper = keeper(Sweep::Submit(b"body".to_vec()), &venue); - let report = keeper.sweep(&host, &TICK).expect("sweep runs"); + let report = run(keeper.sweep(&host, &TICK)).expect("sweep runs"); assert_eq!(report.retried, 1); - assert_eq!(keeper.sweep(&host, &TICK).expect("sweep runs").polled, 1); + assert_eq!( + run(keeper.sweep(&host, &TICK)).expect("sweep runs").polled, + 1 + ); } #[test] @@ -443,9 +471,7 @@ mod tests { let host = MockLocalStore::default(); let venue = StubVenue::new(Ok(SubmitOutcome::Accepted(vec![1]))); - let report = keeper(Sweep::WaitBlock, &venue) - .sweep(&host, &TICK) - .expect("sweep runs"); + let report = run(keeper(Sweep::WaitBlock, &venue).sweep(&host, &TICK)).expect("sweep runs"); assert_eq!(report, SweepReport::default()); } } diff --git a/crates/videre-sdk/src/lib.rs b/crates/videre-sdk/src/lib.rs index 4c34ada2..149f0a3e 100644 --- a/crates/videre-sdk/src/lib.rs +++ b/crates/videre-sdk/src/lib.rs @@ -18,17 +18,22 @@ //! one-byte version tag plus the borsh payload; an unknown tag fails //! typedly rather than as a stringly decode error. //! -//! - [`client`] - the typed intent client core: [`VenueId`] and -//! [`IntentClient`], which binds a venue and encodes through -//! [`IntentBody`] before the byte-level [`VenueClient`] seam. Lives -//! here (not in the strategy SDK) so the codec and the client that -//! speaks it version together. +//! - [`client`] - the typed venue client: a [`Venue`] marker (its +//! [`VenueId`] plus body schema) drives [`VenueClient`], which +//! encodes through [`IntentBody`] before the byte-level, native-AFIT +//! [`VenueTransport`] seam ([`HostVenues`] binds it to the module's +//! 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. //! -//! - [`keeper`] - the generic sweep assembler: [`Keeper::sweep`] runs -//! the world-neutral `nexum_sdk::keeper` stores over a +//! - [`keeper`](mod@keeper) - the generic sweep assembler: +//! [`Keeper::sweep`] runs the world-neutral `nexum_sdk::keeper` +//! stores over a //! [`ConditionalSource`](nexum_sdk::keeper::ConditionalSource) //! producing the shared [`Sweep`] outcome, submitting through the -//! [`VenueClient`] seam. +//! [`VenueTransport`] seam. //! //! - [`transport`] - typed wrappers over the world's scoped imports: //! [`HostChain`](transport::HostChain) behind the SDK [`ChainHost`] @@ -42,15 +47,16 @@ //! //! ## Why the bindgen lives in this crate //! -//! The adapter world's types generate once, in [`bindings`]: the trait, -//! wrappers, and client core are all typed over them. `#[venue]`'s -//! per-cdylib bindgen remaps the type interfaces onto [`bindings`], so -//! an adapter speaks these types while its world imports stay derived -//! from its own manifest. +//! The shared interfaces generate once, in [`bindings`], from an +//! import-only world: the trait, wrappers, and client core are all +//! typed over them. The per-cdylib bindgens (`#[venue]`, `#[keeper]`) +//! remap the shared interfaces onto [`bindings`], so a macro-built +//! component speaks these types while its world stays derived from its +//! own manifest. //! //! [`ChainHost`]: nexum_sdk::host::ChainHost -//! [`IntentClient`]: client::IntentClient //! [`VenueClient`]: client::VenueClient +//! [`VenueTransport`]: client::VenueTransport #![cfg_attr(not(test), warn(unused_crate_dependencies))] #![warn(missing_docs)] @@ -63,16 +69,26 @@ pub mod body; pub mod client; pub mod faults; pub mod keeper; +pub mod rt; pub mod transport; pub use adapter::VenueAdapter; pub use body::{BodyError, IntentBody}; -pub use client::{ClientError, IntentClient, Quoted, VenueClient, VenueId}; +pub use client::{ClientError, HostVenues, Quoted, Venue, VenueClient, VenueId, VenueTransport}; pub use faults::VenueFault; pub use keeper::{Keeper, Sweep, SweepReport}; /// Derive [`IntentBody`] on the outer per-venue version enum. See /// [`videre_macros::IntentBody`]. pub use videre_macros::IntentBody; +/// The blessed keeper authoring path. Apply to a worker's handler impl: +/// emits the per-cdylib bindgen for a world derived from `module.toml` +/// (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 +/// the wire fault so `?` works in handlers. See +/// [`videre_macros::keeper`]. +pub use videre_macros::keeper; /// The single blessed venue authoring path. Apply to the adapter's /// `impl VenueAdapter for MyVenue` block: emits the per-cdylib bindgen /// for a world derived from `module.toml` (asserting its diff --git a/crates/videre-sdk/src/rt.rs b/crates/videre-sdk/src/rt.rs new file mode 100644 index 00000000..6a62f4f3 --- /dev/null +++ b/crates/videre-sdk/src/rt.rs @@ -0,0 +1,38 @@ +//! 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 e71fe381..9c0814f9 100644 --- a/crates/videre-sdk/tests/adapter.rs +++ b/crates/videre-sdk/tests/adapter.rs @@ -1,18 +1,23 @@ //! Acceptance surface for the venue SDK: a hand-written adapter //! compiles against [`VenueAdapter`] and round-trips a versioned body //! through `#[derive(IntentBody)]` - including the typed -//! unknown-version failure and the typed client core driving the -//! adapter through the [`VenueClient`] seam. The world-export glue is -//! `#[videre_sdk::venue]`'s alone; echo-venue is its worked target. +//! unknown-version failure and the typed [`VenueClient`] driving the +//! adapter through the [`VenueTransport`] seam. The world-export glue +//! is `#[videre_sdk::venue]`'s alone; echo-venue is its worked target. use borsh::{BorshDeserialize, BorshSerialize}; use videre_sdk::value_flow::{Asset, AssetAmount}; use videre_sdk::{ - AuthScheme, BodyError, ClientError, Config, Fault, IntentBody, IntentClient, IntentHeader, - IntentStatus, Quotation, Settlement, SubmitOutcome, VenueAdapter, VenueClient, VenueError, - VenueFault, VenueId, + AuthScheme, BodyError, ClientError, Config, Fault, IntentBody, IntentHeader, IntentStatus, + Quotation, Settlement, SubmitOutcome, Venue, VenueAdapter, VenueClient, VenueError, VenueFault, + VenueId, VenueTransport, }; +/// 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") +} + /// First published body version: a fixed-price quote. #[derive(BorshSerialize, BorshDeserialize, Clone, Debug, PartialEq, Eq)] struct QuoteV1 { @@ -115,33 +120,50 @@ impl VenueAdapter for DemoAdapter { } } -/// In-process client: routes the demo venue id straight into the adapter, -/// standing in for the host registry the keeper-side seam will bind. +/// The demo venue as a keeper types it. +struct DemoVenue; + +impl Venue for DemoVenue { + const ID: VenueId = VenueId::from_static("demo"); + type Body = QuoteBody; +} + +/// A venue id no adapter answers for, over the same body schema. +struct NowhereVenue; + +impl Venue for NowhereVenue { + const ID: VenueId = VenueId::from_static("nowhere"); + type Body = QuoteBody; +} + +/// In-process transport: routes the demo venue id straight into the +/// adapter, standing in for the host registry the keeper-side seam +/// binds. struct InProcessClient; -impl VenueClient for InProcessClient { - fn quote(&self, venue: &VenueId, body: Vec) -> Result { +impl VenueTransport for InProcessClient { + async fn quote(&self, venue: &VenueId, body: Vec) -> Result { if venue.as_str() != "demo" { return Err(VenueFault::UnknownVenue); } DemoAdapter::quote(body).map_err(Into::into) } - fn submit(&self, venue: &VenueId, body: Vec) -> Result { + async fn submit(&self, venue: &VenueId, body: Vec) -> Result { if venue.as_str() != "demo" { return Err(VenueFault::UnknownVenue); } DemoAdapter::submit(body).map_err(Into::into) } - fn status(&self, venue: &VenueId, receipt: &[u8]) -> Result { + async fn status(&self, venue: &VenueId, receipt: &[u8]) -> Result { if venue.as_str() != "demo" { return Err(VenueFault::UnknownVenue); } DemoAdapter::status(receipt.to_vec()).map_err(Into::into) } - fn cancel(&self, venue: &VenueId, receipt: &[u8]) -> Result<(), VenueFault> { + async fn cancel(&self, venue: &VenueId, receipt: &[u8]) -> Result<(), VenueFault> { if venue.as_str() != "demo" { return Err(VenueFault::UnknownVenue); } @@ -237,50 +259,54 @@ fn adapter_reports_an_unknown_version_as_invalid_body() { } #[test] -fn typed_client_round_trips_through_the_client_seam() { - let client = IntentClient::new(InProcessClient, "demo"); +fn typed_client_round_trips_through_the_transport_seam() { + let client = VenueClient::::with_transport(InProcessClient); + assert_eq!(client.venue(), DemoVenue::ID); - let outcome = client.submit(&v2_body()).unwrap(); + let outcome = run(client.submit(&v2_body())).unwrap(); let SubmitOutcome::Accepted(receipt) = outcome else { panic!("demo venue always accepts"); }; assert_eq!(receipt, RECEIPT.to_vec()); - assert_eq!(client.status(&receipt).unwrap(), IntentStatus::Open); - client.cancel(&receipt).unwrap(); + assert_eq!(run(client.status(&receipt)).unwrap(), IntentStatus::Open); + run(client.cancel(&receipt)).unwrap(); assert!(matches!( - client.status(&[0, 1]).unwrap_err(), + run(client.status(&[0, 1])).unwrap_err(), ClientError::Venue(VenueFault::Denied(_)) )); } #[test] fn quote_typestate_prices_then_submits_the_quoted_body() { - fn drive(client: &IntentClient) -> Result { + async fn drive( + client: &VenueClient, + ) -> Result { // The typestate chain under test: a quotation is the only path - // from a priced body to its submission. - client.quote(&v2_body())?.submit() + // from a priced body to its submission. Static dispatch end to + // end: the transport is native AFIT, nothing boxes. + client.quote(&v2_body()).await?.submit().await } - let client = IntentClient::new(InProcessClient, "demo"); + let client = VenueClient::::with_transport(InProcessClient); - let quoted = client.quote(&v2_body()).unwrap(); + let quoted = run(client.quote(&v2_body())).unwrap(); assert_eq!( quoted.quotation().gives.amount, 1_000_000u64.to_be_bytes().to_vec() ); assert_eq!(quoted.quotation().valid_until_ms, 1_700_000_000_000); - let outcome = drive(&client).unwrap(); + let outcome = run(drive(&client)).unwrap(); assert!(matches!(outcome, SubmitOutcome::Accepted(r) if r == RECEIPT.to_vec())); } #[test] fn unbound_venue_is_unknown_at_the_client() { - let client = IntentClient::new(InProcessClient, "nowhere"); + let client = VenueClient::::with_transport(InProcessClient); assert!(matches!( - client.submit(&v2_body()).unwrap_err(), + run(client.submit(&v2_body())).unwrap_err(), ClientError::Venue(VenueFault::UnknownVenue) )); } diff --git a/modules/examples/echo-keeper/Cargo.toml b/modules/examples/echo-keeper/Cargo.toml new file mode 100644 index 00000000..ff843e85 --- /dev/null +++ b/modules/examples/echo-keeper/Cargo.toml @@ -0,0 +1,18 @@ +[package] +name = "echo-keeper" +version = "0.1.0" +edition.workspace = true +license.workspace = true +repository.workspace = true +description = "Shepherd example keeper paired with the echo-venue adapter: drives it through the typed VenueClient emitted by #[videre_sdk::keeper] - quote, submit, status, cancel - and logs the intent-status transitions the registry fans back." + +[lints] +workspace = true + +[lib] +crate-type = ["cdylib"] + +[dependencies] +nexum-sdk = { path = "../../../crates/nexum-sdk" } +videre-sdk = { path = "../../../crates/videre-sdk" } +wit-bindgen = { version = "0.59", default-features = false, features = ["macros", "realloc"] } diff --git a/modules/examples/echo-keeper/module.toml b/modules/examples/echo-keeper/module.toml new file mode 100644 index 00000000..0620286c --- /dev/null +++ b/modules/examples/echo-keeper/module.toml @@ -0,0 +1,40 @@ +# echo-keeper module manifest - the blessed keeper half of the echo +# pair. It drives the echo-venue adapter through the typed client, so it +# declares the `client` capability alongside `logging`; the per-module +# world the macro derives imports exactly videre:venue/client and +# nexum:host/logging. + +[module] +name = "echo-keeper" +version = "0.1.0" +# Placeholder content hash; parsed but not verified in 0.2. +component = "sha256:0000000000000000000000000000000000000000000000000000000000000000" + +[capabilities] +# `client` grants the videre:venue/client import (required by +# #[videre_sdk::keeper]); `logging` the log sink. +required = ["client", "logging"] +optional = [] + +[capabilities.http] +allow = [] + +# Drive the venue on every chain-1 block. +[[subscription]] +kind = "block" +chain_id = 1 + +# Observe the status transitions the registry polls from the echo-venue +# adapter. +[[subscription]] +kind = "intent-status" +venue = "echo-venue" + +[config] +name = "echo-keeper" + +# The one body-schema version this keeper encodes; install refuses the +# keeper unless every installed adapter's [venue] body_versions +# contains it. +[venue] +body_version = 1 diff --git a/modules/examples/echo-keeper/src/lib.rs b/modules/examples/echo-keeper/src/lib.rs new file mode 100644 index 00000000..05e41619 --- /dev/null +++ b/modules/examples/echo-keeper/src/lib.rs @@ -0,0 +1,109 @@ +//! # echo-keeper (reference videre keeper module) +//! +//! The blessed keeper half of the echo pair: on every chain-1 block it +//! drives the echo-venue adapter through the typed +//! `VenueClient` - quote, submit, status, cancel, all with a +//! typed body - and logs each `intent-status` transition the registry +//! fans back. Where echo-client hand-writes byte marshalling over the +//! raw `videre:venue/client` import, this module is +//! `#[videre_sdk::keeper]`: the macro wires the world and the client +//! import, and the author never sees wire bytes. +//! +//! It declares two capabilities (`client`, `logging`), so the built +//! component imports `videre:venue/client` and `nexum:host/logging` and +//! nothing else: the per-module world matches the manifest by +//! construction. + +// wit_bindgen::generate! expands to host-import shims whose arity matches +// the WIT signatures, which can exceed clippy's too-many-arguments threshold. +#![cfg_attr(not(test), warn(unused_crate_dependencies))] +#![allow(clippy::too_many_arguments)] + +use nexum::host::{logging, types}; +use videre_sdk::{SubmitOutcome, Venue, VenueClient, VenueId}; + +/// The echo venue as this keeper types it: the id the paired adapter +/// answers for and the body schema below. +struct EchoVenue; + +impl Venue for EchoVenue { + const ID: VenueId = VenueId::from_static("echo-venue"); + type Body = EchoBody; +} + +/// The keeper's published body schema. The echo venue accepts any +/// bytes, so v1 is just the block number: enough to exercise the typed +/// codec end to end. +#[derive(videre_sdk::IntentBody)] +enum EchoBody { + V1(u64), +} + +struct EchoKeeper; + +#[videre_sdk::keeper] +impl EchoKeeper { + async fn on_block(block: types::Block) -> Result<(), Fault> { + let venue = VenueClient::::new(); + let body = EchoBody::V1(block.number); + + // Quote-then-submit through the typestate: the venue prices + // exactly the bytes it is later handed. ClientError folds into + // the wire fault, so `?` applies throughout. + let quoted = venue.quote(&body).await?; + logging::log( + logging::Level::Info, + &format!( + "quoted at {}: gives {} amount bytes", + EchoVenue::ID, + quoted.quotation().gives.amount.len(), + ), + ); + let receipt = match quoted.submit().await? { + SubmitOutcome::Accepted(receipt) => receipt, + SubmitOutcome::RequiresSigning(_) => { + logging::log( + logging::Level::Warn, + &format!("{} unexpectedly asked for a signature", EchoVenue::ID), + ); + return Ok(()); + } + }; + logging::log( + logging::Level::Info, + &format!( + "submitted to {}: receipt {} bytes", + EchoVenue::ID, + receipt.len(), + ), + ); + + let status = venue.status(&receipt).await?; + logging::log( + logging::Level::Info, + &format!("status at {}: {status:?}", EchoVenue::ID), + ); + + venue.cancel(&receipt).await?; + logging::log( + logging::Level::Info, + &format!("cancelled at {}", EchoVenue::ID), + ); + Ok(()) + } + + fn on_intent_status(update: types::IntentStatusUpdate) -> Result<(), Fault> { + let body = nexum_sdk::status_body::StatusBody::decode(&update.status) + .map_err(|err| Fault::InvalidInput(err.to_string()))?; + logging::log( + logging::Level::Info, + &format!( + "intent status from venue {}: {:?} ({} receipt bytes)", + update.venue, + body.status, + update.receipt.len(), + ), + ); + Ok(()) + } +}