From d2acdf4f452dc93ba8209eb54e39eb583e4fdf8c Mon Sep 17 00:00:00 2001 From: mfw78 Date: Fri, 17 Jul 2026 09:20:43 +0000 Subject: [PATCH] sdk: add the keeper macro and the typed venue client The keeper author gets the mirror of #[venue]: #[videre_sdk::keeper] on a handler impl emits the per-cdylib module world (requiring the client capability), remaps the videre interfaces onto the SDK's shared shims, completes async handlers on the synchronous guest boundary, and folds ClientError into the wire fault so ? works in handlers. VenueClient replaces the stringly byte-level trait: a Venue marker carries the VenueId and body schema, calls encode through IntentBody, and the byte seam under it (VenueTransport, native AFIT) dispatches statically with zero boxing. HostVenues binds the seam to the module's own videre:venue/client import; the SDK bindgen collapses to one import-only world so linking it never obliges a module to export the adapter face. CowClient becomes the VenueClient alias, and the echo-keeper example drives echo-venue end to end (quote, submit, status, cancel), proven by a platform e2e test. --- .github/workflows/ci.yml | 4 +- Cargo.lock | 9 + Cargo.toml | 1 + crates/cow-venue/src/client.rs | 99 ++++---- crates/cow-venue/src/lib.rs | 2 +- crates/videre-host/tests/platform.rs | 88 +++++++ crates/videre-macros/Cargo.toml | 2 +- crates/videre-macros/src/keeper.rs | 292 +++++++++++++++++++++++ crates/videre-macros/src/lib.rs | 59 ++++- crates/videre-sdk/Cargo.toml | 2 +- crates/videre-sdk/src/bindings.rs | 37 ++- crates/videre-sdk/src/client.rs | 224 ++++++++++++----- crates/videre-sdk/src/faults.rs | 23 ++ crates/videre-sdk/src/keeper.rs | 98 +++++--- crates/videre-sdk/src/lib.rs | 46 ++-- crates/videre-sdk/src/rt.rs | 38 +++ crates/videre-sdk/tests/adapter.rs | 80 ++++--- modules/examples/echo-keeper/Cargo.toml | 18 ++ modules/examples/echo-keeper/module.toml | 40 ++++ modules/examples/echo-keeper/src/lib.rs | 109 +++++++++ 20 files changed, 1058 insertions(+), 213 deletions(-) create mode 100644 crates/videre-macros/src/keeper.rs create mode 100644 crates/videre-sdk/src/rt.rs create mode 100644 modules/examples/echo-keeper/Cargo.toml create mode 100644 modules/examples/echo-keeper/module.toml create mode 100644 modules/examples/echo-keeper/src/lib.rs 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(()) + } +}