diff --git a/CHANGELOG.md b/CHANGELOG.md index d7af056f68..bf8d0a6dcb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ ### Enhancements +* Added `miden-client-service` crate that wraps `Client` to provide operation coordination (sync/transaction mutual exclusion), background sync, and async-friendly client access ([#1734](https://github.com/0xMiden/miden-client/pull/1734)). * Fixed the faucet token symbol display when showing account details ([#1985](https://github.com/0xMiden/miden-client/pull/1985)). ## 0.14.0 (2026-04-07) @@ -17,6 +18,7 @@ * Made `GrpcNoteTransportClient` connection lazy, deferring it to the first RPC call instead of connecting eagerly at client initialization ([#1970](https://github.com/0xMiden/miden-client/pull/1970)). * Updated the `GrpcClient` to fetch the RPC limits from the node ([#1724](https://github.com/0xMiden/miden-client/pull/1724)) ([#1737](https://github.com/0xMiden/miden-client/pull/1737), [#1809](https://github.com/0xMiden/miden-client/pull/1809)). +* Added the `@miden-sdk/react` hooks library with a provider, hooks, and an example app for the web client ([#1711](https://github.com/0xMiden/miden-client/pull/1711)). * Added typed error parsing for node RPC endpoints, enabling programmatic error handling instead of string parsing ([#1734](https://github.com/0xMiden/miden-client/pull/1734)). * Added `--rpc-status` flag to `miden-client info` command to display RPC node status information including node version, genesis commitment, store status, and block producer status; also added `get_status_unversioned` to `NodeRpcClient` trait ([#1742](https://github.com/0xMiden/miden-client/pull/1742)). * Prevent a potential unwrap panic in `insert_storage_map_nodes_for_map` ([#1750](https://github.com/0xMiden/miden-client/pull/1750)). diff --git a/Cargo.lock b/Cargo.lock index d77b9d7b74..19ef9ada33 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3003,6 +3003,7 @@ dependencies = [ "async-trait", "clap", "miden-client", + "miden-client-service", "miden-client-sqlite-store", "num_cpus", "rand 0.9.2", @@ -3015,6 +3016,17 @@ dependencies = [ "uuid", ] +[[package]] +name = "miden-client-service" +version = "0.15.0" +dependencies = [ + "miden-client", + "miden-protocol", + "rand 0.9.2", + "tokio", + "tracing", +] + [[package]] name = "miden-client-sqlite-store" version = "0.15.0" diff --git a/Cargo.toml b/Cargo.toml index b133c6427d..9f275a4a5f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -5,6 +5,7 @@ members = [ "bin/integration-tests", "bin/miden-bench", "bin/miden-cli", + "crates/client-service", "crates/idxdb-store", "crates/rust-client", "crates/sqlite-store", @@ -35,6 +36,7 @@ opt-level = 1 # Workspace crates idxdb-store = { default-features = false, package = "miden-idxdb-store", path = "crates/idxdb-store", version = "0.15.0" } miden-client = { default-features = false, path = "crates/rust-client", version = "0.15.0" } +miden-client-service = { path = "crates/client-service", version = "0.15.0" } miden-client-sqlite-store = { default-features = false, path = "crates/sqlite-store", version = "0.15.0" } # Miden protocol dependencies diff --git a/bin/integration-tests/Cargo.toml b/bin/integration-tests/Cargo.toml index 0f1aa1b105..b214817e75 100644 --- a/bin/integration-tests/Cargo.toml +++ b/bin/integration-tests/Cargo.toml @@ -20,6 +20,7 @@ path = "tests/integration.rs" [dependencies] # Workspace dependencies miden-client = { features = ["std", "testing", "tonic"], workspace = true } +miden-client-service = { workspace = true } miden-client-sqlite-store = { workspace = true } # External dependencies diff --git a/bin/integration-tests/src/main.rs b/bin/integration-tests/src/main.rs index df5dccf9bb..3bff109316 100644 --- a/bin/integration-tests/src/main.rs +++ b/bin/integration-tests/src/main.rs @@ -286,6 +286,7 @@ impl std::fmt::Debug for TestCase { #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] enum TestCategory { Client, + ClientService, CustomTransaction, Fpi, NetworkFpi, @@ -300,6 +301,7 @@ impl AsRef for TestCategory { fn as_ref(&self) -> &str { match self { TestCategory::Client => "client", + TestCategory::ClientService => "client_service", TestCategory::CustomTransaction => "custom_transaction", TestCategory::Fpi => "fpi", TestCategory::NetworkFpi => "network_fpi", diff --git a/bin/integration-tests/src/tests/client_service.rs b/bin/integration-tests/src/tests/client_service.rs new file mode 100644 index 0000000000..2f05676462 --- /dev/null +++ b/bin/integration-tests/src/tests/client_service.rs @@ -0,0 +1,264 @@ +//! Integration tests for miden-client-service. + +use std::sync::Arc; +use std::time::Duration; + +use anyhow::{Context, Result, bail}; +use miden_client::account::AccountStorageMode; +use miden_client::asset::{Asset, FungibleAsset}; +use miden_client::auth::RPO_FALCON_SCHEME_ID; +use miden_client::note::NoteType; +use miden_client::testing::common::*; +use miden_client::transaction::{PaymentNoteDescription, TransactionRequestBuilder}; +use miden_client_service::{ClientEvent, ClientService, EventFilter, ServiceConfig}; + +use crate::tests::config::ClientConfig; + +/// Comprehensive test that verifies ClientService sync and transaction submission. +/// Tests: +/// - Creating a ClientService +/// - Performing sync operations +/// - Submitting a transaction through the coordinated service +/// - Verifying the service continues to work after transaction +pub async fn test_client_service_sync_and_transaction(client_config: ClientConfig) -> Result<()> { + let (builder, keystore) = client_config.into_client_builder().await?; + let mut client = builder.build().await.context("failed to build client")?; + + // Wait for node and initial sync + wait_for_node(&mut client).await; + + // Set up a faucet account using the raw client (before wrapping in service) + let (faucet, _faucet_key) = insert_new_fungible_faucet( + &mut client, + AccountStorageMode::Private, + &keystore, + RPO_FALCON_SCHEME_ID, + ) + .await + .context("failed to create faucet")?; + + // Now wrap the client in ClientService + let config = ServiceConfig::new().without_background_sync(); + let service = Arc::new(ClientService::new(client, config)); + + // Perform initial sync through the service + let summary = service.sync_state().await.context("failed to sync")?; + assert!(summary.block_num.as_u32() > 0, "Should sync to a non-zero block"); + + // Build and submit a mint transaction + let target_account_id = miden_client::account::AccountId::try_from(ACCOUNT_ID_REGULAR) + .context("failed to create target account id")?; + let asset = + FungibleAsset::new(faucet.id(), MINT_AMOUNT).context("failed to create fungible asset")?; + + let tx_request = { + let mut client = service.client().await; + TransactionRequestBuilder::new() + .build_mint_fungible_asset(asset, target_account_id, NoteType::Public, client.rng()) + .context("failed to build mint transaction request")? + }; + + let tx_id = service + .submit_transaction(faucet.id(), tx_request) + .await + .context("failed to submit transaction")?; + + println!("Transaction submitted: {tx_id}"); + + // Sync again to verify service still works after transaction + let summary2 = service.sync_state().await.context("failed to sync after transaction")?; + assert!(summary2.block_num >= summary.block_num, "Should sync to same or later block"); + + println!("Final sync to block {}", summary2.block_num); + + Ok(()) +} + +/// Tests that background sync works correctly with periodic automatic syncing. +pub async fn test_client_service_background_sync(client_config: ClientConfig) -> Result<()> { + let (builder, _keystore) = client_config.into_client_builder().await?; + let mut client = builder.build().await.context("failed to build client")?; + + // Wait for node to be ready + wait_for_node(&mut client).await; + println!("Node is ready, creating service..."); + + // Create service with fast background sync for testing + let config = ServiceConfig::new().with_sync_interval(Some(Duration::from_millis(500))); + let service = Arc::new(ClientService::new(client, config)); + + // Do one manual sync to verify the service works + let initial_sync = service.sync_state().await.context("initial sync failed")?; + println!("Initial manual sync succeeded, block_num: {}", initial_sync.block_num); + + // Start background sync + let mut sync_handle = service.start_background_sync(); + assert!(sync_handle.is_active(), "Background sync handle should be active"); + + // Wait for a few sync cycles (500ms interval, wait 2s = ~3-4 cycles) + tokio::time::sleep(Duration::from_secs(2)).await; + + // Stop background sync + sync_handle.stop(); + assert!(!sync_handle.is_active(), "Background sync handle should be inactive after stop"); + + // Verify service still works after stopping background sync + let final_sync = service.sync_state().await.context("final sync failed")?; + assert!( + final_sync.block_num >= initial_sync.block_num, + "Should sync to same or later block" + ); + + println!("Background sync test completed, final block: {}", final_sync.block_num); + + Ok(()) +} + +/// End-to-end reactive flow across two [`ClientService`]s: +/// - A sends an asset to B via `submit_transaction`. +/// - B *reacts* to the `NoteReceived` event with a registered handler that builds a consume tx +/// and pushes it onto a [`TransactionQueueHandle`]. B never calls `submit_transaction` directly +/// — the handler + queue do it. +/// - Test observes the outcome purely through events (`NoteConsumed` on B, balance updated). +/// +/// Exercises: background sync on both sides, `on()` event handlers, `expect()` lossless +/// awaiters, the transaction queue (event → enqueue → submit), and the `persistence-mirror` +/// invariant (events fire only after the store commits). +pub async fn test_client_service_reactive_transfer(client_config: ClientConfig) -> Result<()> { + const TRANSFER: u64 = MINT_AMOUNT / 2; + + // --- Set up client A with faucet + wallet_a and fund wallet_a. --- + let (mut client_a, keystore_a) = client_config.clone().into_client().await?; + let (faucet, _faucet_key) = insert_new_fungible_faucet( + &mut client_a, + AccountStorageMode::Public, + &keystore_a, + RPO_FALCON_SCHEME_ID, + ) + .await + .context("create faucet on A")?; + let (wallet_a, _key_a) = insert_new_wallet( + &mut client_a, + AccountStorageMode::Public, + &keystore_a, + RPO_FALCON_SCHEME_ID, + ) + .await + .context("create wallet on A")?; + + let mint_tx = + mint_and_consume(&mut client_a, wallet_a.id(), faucet.id(), NoteType::Public).await; + wait_for_tx(&mut client_a, mint_tx).await?; + + // --- Set up client B with wallet_b. --- + let (mut client_b, keystore_b) = client_config.into_client().await?; + let (wallet_b, _key_b) = insert_new_wallet( + &mut client_b, + AccountStorageMode::Public, + &keystore_b, + RPO_FALCON_SCHEME_ID, + ) + .await + .context("create wallet on B")?; + let wallet_b_id = wallet_b.id(); + + // --- Wrap both in services with fast background sync. --- + let svc_config = ServiceConfig::new().with_sync_interval(Some(Duration::from_millis(500))); + let svc_a = Arc::new(ClientService::new(client_a, svc_config.clone())); + let svc_b = Arc::new(ClientService::new(client_b, svc_config)); + let _sync_a = svc_a.start_background_sync(); + let _sync_b = svc_b.start_background_sync(); + + // --- B starts a transaction queue; an event handler pushes consume-requests onto it. --- + // + // The NoteReceived event carries the full InputNoteRecord, so the handler doesn't need + // to touch the store at all — just convert, build, enqueue. Fire-and-forget via the + // transaction queue worker. Success surfaces via the NoteConsumed event awaited below. + let tx_queue = svc_b.start_transaction_queue(); + let queue_for_handler = tx_queue.clone(); + + svc_b.on(EventFilter::AnyNoteReceived, move |event, _svc| { + let queue = queue_for_handler.clone(); + async move { + let ClientEvent::NoteReceived { note } = event else { + return; + }; + let Ok(concrete_note) = (*note).clone().try_into() else { + return; + }; + let Ok(req) = TransactionRequestBuilder::new().build_consume_notes(vec![concrete_note]) + else { + return; + }; + drop(queue.enqueue(wallet_b_id, req)); + } + }); + + // --- Kick off the transfer on A. --- + let asset = FungibleAsset::new(faucet.id(), TRANSFER).context("build asset")?; + let (transfer_req, expected_note) = { + let mut client = svc_a.client().await; + let req = TransactionRequestBuilder::new() + .build_pay_to_id( + PaymentNoteDescription::new( + vec![Asset::Fungible(asset)], + wallet_a.id(), + wallet_b_id, + ), + NoteType::Public, + client.rng(), + ) + .context("build pay_to_id")?; + let note = req + .expected_output_own_notes() + .into_iter() + .next() + .context("pay_to_id produces an own output note")?; + (req, note) + }; + + // Pre-register the observable checkpoints we're going to assert against, BEFORE + // submitting. These are lossless — events emitted during the upcoming syncs can't be + // dropped by a slow broadcast subscriber. + let received_awaiter = svc_b.expect(EventFilter::NoteReceived(expected_note.id())); + let consumed_awaiter = svc_b.expect(EventFilter::AnyNoteConsumed); + + let transfer_tx_id = svc_a + .submit_transaction(wallet_a.id(), transfer_req) + .await + .context("A submit transfer")?; + println!("A submitted transfer tx: {transfer_tx_id}"); + + // B must first see the note (event drives the handler, which enqueues). + tokio::time::timeout(Duration::from_secs(60), received_awaiter) + .await + .context("timed out waiting for NoteReceived on B")? + .context("NoteReceived awaiter cancelled")?; + println!("B: NoteReceived event fired — handler should now be enqueueing consume"); + + // Then, the queue-submitted consume tx must land on-chain and trigger NoteConsumed. + tokio::time::timeout(Duration::from_secs(60), consumed_awaiter) + .await + .context("timed out waiting for NoteConsumed on B")? + .context("NoteConsumed awaiter cancelled")?; + println!("B: NoteConsumed event fired — consume tx committed"); + + // --- Verify the balance actually arrived. --- + let balance = { + let client = svc_b.client().await; + client + .account_reader(wallet_b_id) + .get_balance(faucet.id()) + .await + .context("read wallet_b balance")? + }; + if balance != TRANSFER { + bail!("wallet_b balance {balance} != expected {TRANSFER}"); + } + println!("wallet_b balance: {balance}"); + + // Queue shutdown: dropping tx_queue closes the channel, the worker exits. + drop(tx_queue); + + Ok(()) +} diff --git a/bin/integration-tests/src/tests/mod.rs b/bin/integration-tests/src/tests/mod.rs index e9f3767342..c33173f17b 100644 --- a/bin/integration-tests/src/tests/mod.rs +++ b/bin/integration-tests/src/tests/mod.rs @@ -1,4 +1,5 @@ pub mod client; +pub mod client_service; pub mod config; pub mod custom_transaction; pub mod fpi; diff --git a/bin/miden-cli/src/commands/exec.rs b/bin/miden-cli/src/commands/exec.rs index fdd57b9e4f..87f43626d7 100644 --- a/bin/miden-cli/src/commands/exec.rs +++ b/bin/miden-cli/src/commands/exec.rs @@ -50,7 +50,7 @@ pub struct ExecCmd { } impl ExecCmd { - pub async fn execute( + pub async fn execute( &self, mut client: Client, ) -> Result<(), CliError> { diff --git a/bin/miden-cli/src/commands/export.rs b/bin/miden-cli/src/commands/export.rs index e9322c2518..a7349d9514 100644 --- a/bin/miden-cli/src/commands/export.rs +++ b/bin/miden-cli/src/commands/export.rs @@ -55,7 +55,7 @@ impl From<&ExportType> for NoteExportType { } impl ExportCmd { - pub async fn execute( + pub async fn execute( &self, mut client: Client, keystore: FilesystemKeyStore, @@ -116,7 +116,7 @@ async fn export_account( // EXPORT NOTE // ================================================================================================ -async fn export_note( +async fn export_note( client: &mut Client, note_id: &str, filename: Option, diff --git a/bin/miden-cli/src/commands/import.rs b/bin/miden-cli/src/commands/import.rs index dc95d14574..565d6486d8 100644 --- a/bin/miden-cli/src/commands/import.rs +++ b/bin/miden-cli/src/commands/import.rs @@ -25,7 +25,7 @@ pub struct ImportCmd { } impl ImportCmd { - pub async fn execute( + pub async fn execute( &self, mut client: Client, keystore: FilesystemKeyStore, diff --git a/bin/miden-cli/src/commands/new_account.rs b/bin/miden-cli/src/commands/new_account.rs index 53170eb40b..e7246af5d3 100644 --- a/bin/miden-cli/src/commands/new_account.rs +++ b/bin/miden-cli/src/commands/new_account.rs @@ -101,7 +101,7 @@ pub struct NewWalletCmd { } impl NewWalletCmd { - pub async fn execute( + pub async fn execute( &self, mut client: Client, keystore: CliKeyStore, @@ -196,7 +196,7 @@ pub struct NewAccountCmd { } impl NewAccountCmd { - pub async fn execute( + pub async fn execute( &self, mut client: Client, keystore: CliKeyStore, @@ -377,7 +377,7 @@ fn should_add_implicit_auth_controlled( /// and build the account. /// /// If no auth component is detected in the packages, a Falcon-based auth component will be added. -async fn create_client_account( +async fn create_client_account( client: &mut Client, keystore: &CliKeyStore, account_type: AccountType, @@ -465,7 +465,7 @@ async fn create_client_account( } /// Submits a deploy transaction to the node for the specified account. -async fn deploy_account( +async fn deploy_account( client: &mut Client, account: &Account, ) -> Result<(), CliError> { diff --git a/bin/miden-cli/src/commands/new_transactions.rs b/bin/miden-cli/src/commands/new_transactions.rs index 347f666886..4ee54fee53 100644 --- a/bin/miden-cli/src/commands/new_transactions.rs +++ b/bin/miden-cli/src/commands/new_transactions.rs @@ -72,7 +72,7 @@ pub struct MintCmd { } impl MintCmd { - pub async fn execute( + pub async fn execute( &self, mut client: Client, ) -> Result<(), CliError> { @@ -143,7 +143,7 @@ pub struct SendCmd { } impl SendCmd { - pub async fn execute( + pub async fn execute( &self, mut client: Client, ) -> Result<(), CliError> { @@ -221,7 +221,7 @@ pub struct SwapCmd { } impl SwapCmd { - pub async fn execute( + pub async fn execute( &self, mut client: Client, ) -> Result<(), CliError> { @@ -299,7 +299,7 @@ pub struct ConsumeNotesCmd { } impl ConsumeNotesCmd { - pub async fn execute( + pub async fn execute( &self, mut client: Client, ) -> Result<(), CliError> { @@ -368,7 +368,7 @@ impl ConsumeNotesCmd { // EXECUTE TRANSACTION // ================================================================================================ -async fn execute_transaction( +async fn execute_transaction( client: &mut Client, account_id: AccountId, transaction_request: TransactionRequest, diff --git a/bin/miden-cli/src/commands/notes.rs b/bin/miden-cli/src/commands/notes.rs index 80929c72c8..b6d3e5e9f6 100644 --- a/bin/miden-cli/src/commands/notes.rs +++ b/bin/miden-cli/src/commands/notes.rs @@ -72,7 +72,7 @@ pub struct NotesCmd { } impl NotesCmd { - pub async fn execute( + pub async fn execute( &self, mut client: Client, ) -> Result<(), CliError> { @@ -121,7 +121,7 @@ struct CliNoteSummary { // LIST NOTES // ================================================================================================ -async fn list_notes( +async fn list_notes( client: Client, filter: ClientNoteFilter, ) -> Result<(), CliError> { @@ -147,7 +147,7 @@ async fn list_notes( // SHOW NOTE // ================================================================================================ #[allow(clippy::too_many_lines)] -async fn show_note( +async fn show_note( client: Client, note_id: String, with_code: bool, @@ -316,7 +316,7 @@ async fn show_note( // LIST CONSUMABLE INPUT NOTES // ================================================================================================ -async fn list_consumable_notes( +async fn list_consumable_notes( client: Client, account_id: Option<&String>, ) -> Result<(), CliError> { @@ -333,7 +333,7 @@ async fn list_consumable_notes( // ================================================================================================ /// Send a (stored) note -async fn send( +async fn send( client: &mut Client, note_id: &str, address: &str, @@ -359,7 +359,7 @@ async fn send( /// Fetched notes are stored in the store. async fn fetch(client: &mut Client) -> Result<(), CliError> where - AUTH: Keystore + Sync + 'static, + AUTH: Keystore + Send + Sync + 'static, { client.fetch_private_notes().await?; diff --git a/bin/miden-cli/src/commands/sync.rs b/bin/miden-cli/src/commands/sync.rs index 599b0ed70f..ac60616088 100644 --- a/bin/miden-cli/src/commands/sync.rs +++ b/bin/miden-cli/src/commands/sync.rs @@ -9,7 +9,7 @@ use crate::errors::CliError; pub struct SyncCmd {} impl SyncCmd { - pub async fn execute( + pub async fn execute( &self, mut client: Client, ) -> Result<(), CliError> { diff --git a/bin/miden-cli/src/commands/transactions.rs b/bin/miden-cli/src/commands/transactions.rs index 4a196a0220..6d75169798 100644 --- a/bin/miden-cli/src/commands/transactions.rs +++ b/bin/miden-cli/src/commands/transactions.rs @@ -15,7 +15,7 @@ pub struct TransactionCmd { } impl TransactionCmd { - pub async fn execute( + pub async fn execute( &self, client: Client, ) -> Result<(), CliError> { @@ -26,7 +26,7 @@ impl TransactionCmd { // LIST TRANSACTIONS // ================================================================================================ -async fn list_transactions( +async fn list_transactions( client: Client, ) -> Result<(), CliError> { let transactions = client.get_transactions(TransactionFilter::All).await?; diff --git a/bin/miden-cli/src/info.rs b/bin/miden-cli/src/info.rs index 2d8c1e2f89..e436168e63 100644 --- a/bin/miden-cli/src/info.rs +++ b/bin/miden-cli/src/info.rs @@ -11,7 +11,7 @@ use super::config::CliConfig; use crate::commands::account::DEFAULT_ACCOUNT_ID_KEY; use crate::errors::CliError; -pub async fn print_client_info( +pub async fn print_client_info( client: &Client, show_rpc_status: bool, ) -> Result<(), CliError> { @@ -42,7 +42,7 @@ pub async fn print_client_info( // HELPERS // ================================================================================================ -async fn print_client_stats( +async fn print_client_stats( client: &Client, ) -> Result<(), CliError> { println!("Block number: {}", client.get_sync_height().await?); diff --git a/bin/miden-cli/src/lib.rs b/bin/miden-cli/src/lib.rs index 9c6fef97b2..ebf58d6fd0 100644 --- a/bin/miden-cli/src/lib.rs +++ b/bin/miden-cli/src/lib.rs @@ -498,7 +498,7 @@ pub fn create_dynamic_table(headers: &[&str]) -> Table { /// unable to find any note where `note_id_prefix` is a prefix of its ID. /// - Returns [`IdPrefixFetchError::MultipleMatches`](miden_client::IdPrefixFetchError::MultipleMatches) /// if there were more than one note found where `note_id_prefix` is a prefix of its ID. -pub(crate) async fn get_output_note_with_id_prefix( +pub(crate) async fn get_output_note_with_id_prefix( client: &miden_client::Client, note_id_prefix: &str, ) -> Result { diff --git a/crates/client-service/Cargo.toml b/crates/client-service/Cargo.toml new file mode 100644 index 0000000000..99d1f373d2 --- /dev/null +++ b/crates/client-service/Cargo.toml @@ -0,0 +1,37 @@ +[package] +authors.workspace = true +description = "Service wrapper for miden-client with operation coordination and background sync" +edition.workspace = true +keywords = ["client", "miden", "service"] +license.workspace = true +name = "miden-client-service" +readme = "README.md" +repository.workspace = true +rust-version.workspace = true +version.workspace = true + +[features] +default = ["std"] +std = ["miden-client/std", "tokio"] +testing = ["miden-client/testing"] + +[dependencies] +# Workspace crates +miden-client = { workspace = true } + +# Miden protocol dependencies +miden-protocol = { workspace = true } + +# External dependencies +tracing = { workspace = true } + +# Optional native dependencies +tokio = { features = ["rt", "sync", "time"], optional = true, workspace = true } + +[dev-dependencies] +miden-client = { features = ["testing"], workspace = true } +rand = { workspace = true } +tokio = { features = ["macros", "rt"], workspace = true } + +[lints] +workspace = true diff --git a/crates/client-service/src/awaiters.rs b/crates/client-service/src/awaiters.rs new file mode 100644 index 0000000000..8eb1e228b6 --- /dev/null +++ b/crates/client-service/src/awaiters.rs @@ -0,0 +1,162 @@ +//! Pre-registered event awaiters. +//! +//! Unlike the broadcast channel used by [`subscribe`](crate::ClientService::subscribe), +//! awaiters are fulfilled synchronously during event emission and cannot be +//! lost to lag. This makes them suitable for RPC-style "wait for a specific +//! event" patterns where missing the event would be a correctness bug. + +use std::future::Future; +use std::pin::Pin; +use std::sync::Mutex; +use std::task::{Context, Poll}; + +use tokio::sync::oneshot; + +use crate::events::ClientEvent; +use crate::handlers::EventFilter; + +/// Registry of pending awaiters. Consulted in-line by `emit_events` before +/// the broadcast send so awaited events cannot be dropped by a slow broadcast +/// consumer. +pub(crate) struct AwaiterRegistry { + inner: Mutex>, +} + +struct Awaiter { + filter: EventFilter, + tx: oneshot::Sender, +} + +impl AwaiterRegistry { + pub(crate) fn new() -> Self { + Self { inner: Mutex::new(Vec::new()) } + } + + /// Registers an awaiter for the next event matching `filter`. Returns a + /// future that resolves to the matching event, or [`AwaiterCancelled`] if + /// the service is dropped before a match arrives. + pub(crate) fn register(&self, filter: EventFilter) -> AwaiterFuture { + let (tx, rx) = oneshot::channel(); + self.inner + .lock() + .expect("awaiter registry poisoned") + .push(Awaiter { filter, tx }); + AwaiterFuture { rx } + } + + /// Fulfills any awaiters matching the given events. Consumed awaiters are + /// removed; awaiters whose receivers have been dropped are pruned. + pub(crate) fn fulfill(&self, events: &[ClientEvent]) { + if events.is_empty() { + return; + } + let mut inner = self.inner.lock().expect("awaiter registry poisoned"); + let old = std::mem::take(&mut *inner); + for awaiter in old { + if awaiter.tx.is_closed() { + continue; // receiver dropped — prune + } + match events.iter().find(|e| awaiter.filter.matches(e)) { + Some(event) => { + // Fulfilled — consume the awaiter. + let _ = awaiter.tx.send(event.clone()); + }, + None => { + // No match this round — keep waiting. + inner.push(awaiter); + }, + } + } + } +} + +/// Future returned by [`ClientService::expect`](crate::ClientService::expect) +/// and used internally by [`ClientService::once`](crate::ClientService::once). +/// +/// Resolves with the first matching [`ClientEvent`]. Dropping the future +/// cancels the awaiter; the registry prunes stale entries on the next +/// emission. +pub struct AwaiterFuture { + rx: oneshot::Receiver, +} + +impl Future for AwaiterFuture { + type Output = Result; + + fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + Pin::new(&mut self.rx).poll(cx).map_err(|_| AwaiterCancelled) + } +} + +/// Error returned when an awaiter is cancelled because the service was +/// dropped before a matching event arrived. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct AwaiterCancelled; + +impl core::fmt::Display for AwaiterCancelled { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { + write!(f, "awaiter cancelled: service was dropped before event arrived") + } +} + +impl std::error::Error for AwaiterCancelled {} + +#[cfg(test)] +mod tests { + use super::*; + use crate::test_utils::{test_note_arc, test_note_id}; + + #[tokio::test] + async fn fulfills_matching_event() { + let registry = AwaiterRegistry::new(); + let note_id = test_note_id(); + let awaiter = registry.register(EventFilter::NoteCommitted(note_id)); + + registry.fulfill(&[ClientEvent::NoteCommitted { note_id }]); + + let event = awaiter.await.unwrap(); + assert_eq!(event.note_id(), Some(note_id)); + } + + #[tokio::test] + async fn ignores_non_matching_events() { + let registry = AwaiterRegistry::new(); + let note_id = test_note_id(); + let awaiter = registry.register(EventFilter::NoteCommitted(note_id)); + + registry.fulfill(&[ClientEvent::NoteReceived { note: test_note_arc() }]); + + let result = tokio::time::timeout(std::time::Duration::from_millis(50), awaiter).await; + assert!(result.is_err(), "awaiter should not have resolved"); + } + + #[tokio::test] + async fn dropped_awaiter_is_pruned() { + let registry = AwaiterRegistry::new(); + let note_id = test_note_id(); + { + let _awaiter = registry.register(EventFilter::NoteCommitted(note_id)); + // Awaiter dropped at end of block. + } + // Pruning happens opportunistically during fulfill. + registry.fulfill(&[ClientEvent::NoteCommitted { note_id }]); + assert!(registry.inner.lock().unwrap().is_empty()); + } + + #[tokio::test] + async fn only_first_match_wins() { + let registry = AwaiterRegistry::new(); + let note_id = test_note_id(); + let awaiter = registry.register(EventFilter::AnyNoteCommitted); + + registry.fulfill(&[ + ClientEvent::NoteCommitted { note_id }, + ClientEvent::NoteCommitted { note_id }, + ]); + + let event = awaiter.await.unwrap(); + assert!(matches!(event, ClientEvent::NoteCommitted { .. })); + // Second call should find no awaiter. + assert!(registry.inner.lock().unwrap().is_empty()); + } +} diff --git a/crates/client-service/src/config.rs b/crates/client-service/src/config.rs new file mode 100644 index 0000000000..919d1873ac --- /dev/null +++ b/crates/client-service/src/config.rs @@ -0,0 +1,81 @@ +//! Configuration options for the client service. + +use core::num::NonZeroU32; +use std::time::Duration; + +/// Default interval between automatic background sync operations. +pub const DEFAULT_SYNC_INTERVAL: Duration = Duration::from_secs(5); + +/// Configuration options for [`ClientService`](crate::ClientService). +#[derive(Debug, Clone)] +pub struct ServiceConfig { + /// Interval between automatic background sync operations. + /// + /// Set to `None` to disable automatic background sync. + /// Default: [`DEFAULT_SYNC_INTERVAL`]. + pub sync_interval: Option, + + /// Whether `start_background_sync` should trigger an immediate sync before + /// waiting for the first interval tick. + /// + /// Default: `true`. + pub sync_on_start: bool, + + /// If set, each sync advances at most this many blocks past the current + /// sync height. When `None`, every sync runs to the current chain tip. + /// + /// Useful for catch-up scenarios: a client that has been offline for + /// thousands of blocks would otherwise emit a massive burst of events + /// (one per note/transaction/account across the whole range) and hold the + /// client mutex for the full duration of a single huge sync. Chunking + /// breaks that into bounded steps. + /// + /// Default: `None`. + pub sync_chunk_blocks: Option, +} + +impl Default for ServiceConfig { + fn default() -> Self { + Self { + sync_interval: Some(DEFAULT_SYNC_INTERVAL), + sync_on_start: true, + sync_chunk_blocks: None, + } + } +} + +impl ServiceConfig { + /// Creates a new configuration with default values. + pub fn new() -> Self { + Self::default() + } + + /// Sets the sync interval. + #[must_use] + pub fn with_sync_interval(mut self, interval: Option) -> Self { + self.sync_interval = interval; + self + } + + /// Disables automatic background sync. + #[must_use] + pub fn without_background_sync(mut self) -> Self { + self.sync_interval = None; + self + } + + /// Controls whether background sync triggers an immediate sync on start. + #[must_use] + pub fn with_sync_on_start(mut self, sync_on_start: bool) -> Self { + self.sync_on_start = sync_on_start; + self + } + + /// Configures chunked sync. When set, each sync advances at most `n` blocks + /// past the current sync height; `None` disables chunking. + #[must_use] + pub fn with_sync_chunk_blocks(mut self, n: Option) -> Self { + self.sync_chunk_blocks = n; + self + } +} diff --git a/crates/client-service/src/events.rs b/crates/client-service/src/events.rs new file mode 100644 index 0000000000..965b689fc9 --- /dev/null +++ b/crates/client-service/src/events.rs @@ -0,0 +1,200 @@ +//! Event types emitted by the client service. +//! +//! After each sync, the [`ClientService`](crate::ClientService) decomposes the +//! [`SyncSummary`] into individual [`ClientEvent`] variants and publishes them +//! on a broadcast channel. Consumers can react to these events in three ways: +//! +//! ## 1. Raw Event Stream +//! +//! ```rust,ignore +//! let mut events = service.subscribe(); +//! tokio::spawn(async move { +//! while let Ok(event) = events.recv().await { +//! match event { +//! ClientEvent::NoteReceived { note } => println!("received: {}", note.id()), +//! ClientEvent::TransactionCommitted { transaction_id } => println!("confirmed: {transaction_id}"), +//! _ => {} +//! } +//! } +//! }); +//! ``` +//! +//! ## 2. Persistent Handlers +//! +//! ```rust,ignore +//! service.on(EventFilter::AnyNoteReceived, |event, _service| async move { +//! // Handler receives the full InputNoteRecord — no store lookup needed. +//! if let ClientEvent::NoteReceived { note } = event { +//! println!("new note {} carrying {:?}", note.id(), note.assets()); +//! } +//! }); +//! ``` +//! +//! ## 3. One-Shot Awaiters +//! +//! ```rust,ignore +//! let tx_id = service.submit_transaction(account_id, request).await?; +//! service.once(EventFilter::TransactionCommitted(tx_id), Some(Duration::from_secs(60))).await?; +//! ``` + +use std::collections::BTreeMap; +use std::sync::Arc; + +use miden_client::store::InputNoteRecord; +use miden_client::sync::SyncSummary; +use miden_client::transaction::TransactionId; +use miden_protocol::account::AccountId; +use miden_protocol::note::NoteId; + +/// Events emitted by the [`ClientService`](crate::ClientService). +/// +/// Granular events (notes, transactions, accounts) are emitted before +/// [`SyncCompleted`](Self::SyncCompleted) so subscribers see individual +/// changes first. +#[derive(Debug, Clone)] +pub enum ClientEvent { + /// A new public note was received during sync. Carries the full record so + /// handlers can build consume requests, inspect the script, etc. without a + /// store roundtrip. + /// + /// Wrapped in `Arc` so broadcasting to N subscribers is an O(N) refcount + /// bump rather than N deep clones of the (potentially multi-KB) record. + NoteReceived { note: Arc }, + /// A tracked note was committed on-chain. + NoteCommitted { note_id: NoteId }, + /// A note was consumed. + NoteConsumed { note_id: NoteId }, + /// A transaction was committed on-chain. + TransactionCommitted { transaction_id: TransactionId }, + /// An on-chain account was updated. + AccountUpdated { account_id: AccountId }, + /// A private account was locked. + AccountLocked { account_id: AccountId }, + /// Sync completed. Always emitted last in a sync cycle. + SyncCompleted { summary: SyncSummary }, +} + +impl ClientEvent { + /// Returns the [`NoteId`] if this is a note-related event. + pub fn note_id(&self) -> Option { + match self { + Self::NoteReceived { note } => Some(note.id()), + Self::NoteCommitted { note_id } | Self::NoteConsumed { note_id } => Some(*note_id), + _ => None, + } + } + + /// Returns the [`TransactionId`] if this is a transaction-related event. + pub fn transaction_id(&self) -> Option { + match self { + Self::TransactionCommitted { transaction_id } => Some(*transaction_id), + _ => None, + } + } + + /// Returns the [`AccountId`] if this is an account-related event. + pub fn account_id(&self) -> Option { + match self { + Self::AccountUpdated { account_id } | Self::AccountLocked { account_id } => { + Some(*account_id) + }, + _ => None, + } + } + + /// Returns the [`SyncSummary`] if this is a [`SyncCompleted`](Self::SyncCompleted) event. + pub fn summary(&self) -> Option<&SyncSummary> { + match self { + Self::SyncCompleted { summary } => Some(summary), + _ => None, + } + } +} + +/// Decomposes a [`SyncSummary`] into individual [`ClientEvent`] variants. +/// +/// `new_note_records` must contain the full [`InputNoteRecord`] for every entry in +/// `summary.new_public_notes`. Callers build this map by bulk-reading the store after +/// `apply_state_sync`. Entries missing from the map are silently skipped (the note will +/// still be in the store, just not in the event stream). +/// +/// Granular events are emitted first, with [`ClientEvent::SyncCompleted`] last. +pub(crate) fn events_from_sync( + summary: &SyncSummary, + new_note_records: &BTreeMap>, +) -> Vec { + let capacity = summary.new_public_notes.len() + + summary.committed_notes.len() + + summary.consumed_notes.len() + + summary.committed_transactions.len() + + summary.updated_accounts.len() + + summary.locked_accounts.len() + + 1; + let mut events = Vec::with_capacity(capacity); + + for note_id in &summary.new_public_notes { + if let Some(note) = new_note_records.get(note_id) { + events.push(ClientEvent::NoteReceived { note: Arc::clone(note) }); + } + } + for ¬e_id in &summary.committed_notes { + events.push(ClientEvent::NoteCommitted { note_id }); + } + for ¬e_id in &summary.consumed_notes { + events.push(ClientEvent::NoteConsumed { note_id }); + } + for &transaction_id in &summary.committed_transactions { + events.push(ClientEvent::TransactionCommitted { transaction_id }); + } + for &account_id in &summary.updated_accounts { + events.push(ClientEvent::AccountUpdated { account_id }); + } + for &account_id in &summary.locked_accounts { + events.push(ClientEvent::AccountLocked { account_id }); + } + + events.push(ClientEvent::SyncCompleted { summary: summary.clone() }); + + events +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::test_utils::{empty_summary, test_note_arc, test_note_id}; + + #[test] + fn empty_summary_produces_only_sync_completed() { + let events = events_from_sync(&empty_summary(), &BTreeMap::new()); + assert_eq!(events.len(), 1); + assert!(matches!(events[0], ClientEvent::SyncCompleted { .. })); + } + + #[test] + fn sync_completed_is_last() { + let note = test_note_arc(); + let note_id = test_note_id(); + let mut summary = empty_summary(); + summary.new_public_notes = vec![note_id]; + summary.committed_notes = vec![note_id]; + + let mut records = BTreeMap::new(); + records.insert(note_id, Arc::clone(¬e)); + + let events = events_from_sync(&summary, &records); + assert_eq!(events.len(), 3); + assert!(matches!(events[0], ClientEvent::NoteReceived { .. })); + assert!(matches!(events[1], ClientEvent::NoteCommitted { .. })); + assert!(matches!(events[2], ClientEvent::SyncCompleted { .. })); + } + + #[test] + fn new_public_notes_without_records_are_skipped() { + let mut summary = empty_summary(); + summary.new_public_notes = vec![test_note_id()]; + let events = events_from_sync(&summary, &BTreeMap::new()); + // No record in the map → no NoteReceived emitted; SyncCompleted still fires. + assert_eq!(events.len(), 1); + assert!(matches!(events[0], ClientEvent::SyncCompleted { .. })); + } +} diff --git a/crates/client-service/src/handlers.rs b/crates/client-service/src/handlers.rs new file mode 100644 index 0000000000..f74097135f --- /dev/null +++ b/crates/client-service/src/handlers.rs @@ -0,0 +1,225 @@ +//! Event handler registration and dispatch. +//! +//! Each registered handler gets its own spawned worker task reading from a +//! bounded mpsc channel. `dispatch` pushes matching events onto the handler's +//! queue without blocking emission; if the queue is full the event is dropped +//! and a warning is logged. This caps the fanout at `num_handlers` tasks and +//! preserves per-handler event ordering. + +use std::future::Future; +use std::pin::Pin; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, RwLock}; + +use miden_client::transaction::TransactionId; +use miden_protocol::account::AccountId; +use miden_protocol::note::NoteId; +use tokio::sync::mpsc; +use tracing::warn; + +use crate::events::ClientEvent; + +/// Bounded capacity of each handler's event queue. +const HANDLER_QUEUE_CAPACITY: usize = 256; + +type BoxFuture = Pin + Send>>; + +pub(crate) struct StoredHandler(Arc BoxFuture + Send + Sync>); + +impl StoredHandler { + pub(crate) fn new(f: F) -> Self + where + F: Fn(ClientEvent) -> Fut + Send + Sync + 'static, + Fut: Future + Send + 'static, + { + Self(Arc::new(move |event| Box::pin(f(event)))) + } + + fn call(&self, event: ClientEvent) -> BoxFuture { + (self.0)(event) + } +} + +/// Opaque identifier for a registered handler. Used with +/// [`ClientService::remove_handler`](crate::ClientService::remove_handler). +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub struct HandlerId(u64); + +/// Describes which events a handler or awaiter is interested in. +/// +/// Used with [`ClientService::on`](crate::ClientService::on), +/// [`ClientService::once`](crate::ClientService::once), and +/// [`ClientService::expect`](crate::ClientService::expect). +#[derive(Debug, Clone)] +pub enum EventFilter { + /// Match a specific note being received. + NoteReceived(NoteId), + /// Match any note received. + AnyNoteReceived, + /// Match a specific note being committed. + NoteCommitted(NoteId), + /// Match any note committed. + AnyNoteCommitted, + /// Match any note consumed. + AnyNoteConsumed, + /// Match a specific transaction committed. + TransactionCommitted(TransactionId), + /// Match any transaction committed. + AnyTransactionCommitted, + /// Match a specific account updated. + AccountUpdated(AccountId), + /// Match any account updated. + AnyAccountUpdated, + /// Match any sync completed. + AnySyncCompleted, +} + +impl EventFilter { + pub(crate) fn matches(&self, event: &ClientEvent) -> bool { + match (self, event) { + (Self::NoteReceived(id), ClientEvent::NoteReceived { note }) => *id == note.id(), + (Self::NoteCommitted(id), ClientEvent::NoteCommitted { note_id }) => id == note_id, + (Self::AccountUpdated(id), ClientEvent::AccountUpdated { account_id }) => { + id == account_id + }, + ( + Self::TransactionCommitted(id), + ClientEvent::TransactionCommitted { transaction_id }, + ) => id == transaction_id, + (Self::AnyNoteReceived, ClientEvent::NoteReceived { .. }) + | (Self::AnyNoteCommitted, ClientEvent::NoteCommitted { .. }) + | (Self::AnyNoteConsumed, ClientEvent::NoteConsumed { .. }) + | (Self::AnyTransactionCommitted, ClientEvent::TransactionCommitted { .. }) + | (Self::AnyAccountUpdated, ClientEvent::AccountUpdated { .. }) + | (Self::AnySyncCompleted, ClientEvent::SyncCompleted { .. }) => true, + _ => false, + } + } +} + +struct HandlerEntry { + id: HandlerId, + filter: EventFilter, + /// Bounded sender feeding the handler's worker task. Dropping this (via + /// `unregister` or registry drop) closes the channel and terminates the + /// worker. + tx: mpsc::Sender, +} + +pub(crate) struct HandlerRegistry { + handlers: RwLock>, + next_id: AtomicU64, +} + +impl HandlerRegistry { + pub fn new() -> Self { + Self { + handlers: RwLock::new(Vec::new()), + next_id: AtomicU64::new(1), + } + } + + /// Registers a handler and spawns its dedicated worker task. + pub fn register(&self, filter: EventFilter, handler: StoredHandler) -> HandlerId { + let id = HandlerId(self.next_id.fetch_add(1, Ordering::Relaxed)); + let (tx, mut rx) = mpsc::channel::(HANDLER_QUEUE_CAPACITY); + + tokio::spawn(async move { + // The handler is invoked sequentially per event: preserves ordering + // within a handler. The loop exits when all senders drop (which + // happens on `unregister` or when the registry itself drops). + while let Some(event) = rx.recv().await { + handler.call(event).await; + } + }); + + self.handlers.write().expect("handler registry poisoned").push(HandlerEntry { + id, + filter, + tx, + }); + id + } + + /// Unregisters a handler by id. The worker task exits once the channel + /// drains. + pub fn unregister(&self, id: HandlerId) -> bool { + let mut handlers = self.handlers.write().expect("handler registry poisoned"); + let len_before = handlers.len(); + handlers.retain(|entry| entry.id != id); + handlers.len() < len_before + } + + /// Dispatches events to all matching handlers. + /// + /// Non-blocking: uses `try_send` on each handler's bounded queue. If a + /// handler's queue is full, the event is dropped and a warning is logged — + /// emission is never stalled by a slow handler. + pub fn dispatch(&self, events: &[ClientEvent]) { + if events.is_empty() { + return; + } + let handlers = self.handlers.read().expect("handler registry poisoned"); + + for event in events { + for entry in handlers.iter() { + if !entry.filter.matches(event) { + continue; + } + if let Err(mpsc::error::TrySendError::Full(_)) = entry.tx.try_send(event.clone()) { + // Full queue: handler is slow. Drop the event rather than + // stall emission. `Closed` (worker died) is silently + // ignored; the entry will be cleaned up on unregister. + warn!( + handler_id = ?entry.id, + capacity = HANDLER_QUEUE_CAPACITY, + "handler queue full, dropping event", + ); + } + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::test_utils::{empty_summary, test_note_arc, test_note_id}; + + #[test] + fn filter_matches_specific_note() { + let note_id = test_note_id(); + let note = test_note_arc(); + + assert!( + EventFilter::NoteCommitted(note_id).matches(&ClientEvent::NoteCommitted { note_id }) + ); + assert!(!EventFilter::NoteCommitted(note_id).matches(&ClientEvent::NoteReceived { note })); + assert!( + !EventFilter::NoteCommitted(note_id).matches(&ClientEvent::NoteConsumed { note_id }) + ); + } + + #[test] + fn filter_matches_any_variant() { + let note_id = test_note_id(); + let note = test_note_arc(); + + assert!(EventFilter::AnyNoteReceived.matches(&ClientEvent::NoteReceived { note })); + assert!(!EventFilter::AnyNoteReceived.matches(&ClientEvent::NoteCommitted { note_id })); + assert!( + EventFilter::AnySyncCompleted + .matches(&ClientEvent::SyncCompleted { summary: empty_summary() }) + ); + } + + #[tokio::test] + async fn register_and_unregister() { + let registry = HandlerRegistry::new(); + let handler = StoredHandler::new(|_| async {}); + let id = registry.register(EventFilter::NoteCommitted(test_note_id()), handler); + + assert!(registry.unregister(id)); + assert!(!registry.unregister(id)); + } +} diff --git a/crates/client-service/src/lib.rs b/crates/client-service/src/lib.rs new file mode 100644 index 0000000000..2f2631522a --- /dev/null +++ b/crates/client-service/src/lib.rs @@ -0,0 +1,582 @@ +//! A service wrapper for the Miden client that provides operation coordination, +//! background synchronization, and an event system for reacting to on-chain +//! state changes. +//! +//! # Overview +//! +//! The [`ClientService`] wraps a [`Client`](miden_client::Client) to provide: +//! +//! - **Operation Coordination** - Mutual exclusion between sync and transaction operations +//! - **Background Sync** - Periodic automatic synchronization with the network +//! - **Event System** - Subscribe to state changes, register handlers, or await specific events +//! - **Async-Friendly Access** - Direct client access via [`MutexGuard`](tokio::sync::MutexGuard) +//! +//! # Example +//! +//! ```rust,ignore +//! use std::sync::Arc; +//! use std::time::Duration; +//! use miden_client_service::{ClientService, ServiceConfig, EventFilter}; +//! +//! let service = Arc::new(ClientService::new(client, ServiceConfig::default())); +//! let _sync = service.start_background_sync(); +//! +//! // Submit a transaction and wait for it to be committed +//! let tx_id = service.submit_transaction(account_id, tx_request).await?; +//! service.once(EventFilter::TransactionCommitted(tx_id), Some(Duration::from_secs(60))).await?; +//! +//! // Register a persistent handler (has access to the service) +//! service.on(EventFilter::AnyNoteReceived, |event, service| async move { +//! let client = service.client().await; +//! // query client state... +//! }); +//! +//! // Or subscribe to the raw event stream +//! let mut events = service.subscribe(); +//! ``` + +#![cfg_attr(not(feature = "std"), no_std)] + +extern crate alloc; + +use std::collections::BTreeMap; +use std::future::Future; +use std::sync::Arc; +use std::time::Duration; + +use miden_client::auth::TransactionAuthenticator; +use miden_client::store::{InputNoteRecord, NoteFilter}; +use miden_client::sync::{SyncSummary, SyncTarget}; +use miden_client::transaction::{TransactionId, TransactionRequest}; +use miden_client::{Client, ClientError}; +use miden_protocol::account::AccountId; +use miden_protocol::block::BlockNumber; +use miden_protocol::note::NoteId; +use tokio::sync::{Mutex, MutexGuard, broadcast}; +use tracing::{debug, info, warn}; + +mod awaiters; +pub(crate) mod events; +mod handlers; +mod tx_queue; + +mod config; +pub use awaiters::{AwaiterCancelled, AwaiterFuture}; +pub use config::ServiceConfig; +pub use events::ClientEvent; +pub use handlers::{EventFilter, HandlerId}; +/// Re-exported so consumers implementing a custom note filter for +/// [`ClientService::with_note_screener`] don't have to reach into `miden_client`. +/// +/// **This is a filter, not a listener.** Implementing [`OnNoteReceived`] changes what the +/// client persists during sync. To observe notes without changing persistence, use +/// [`ClientService::subscribe`] or [`ClientService::on`]. +pub use miden_client::sync::{NoteUpdateAction, OnNoteReceived}; +pub use tx_queue::{EnqueueError, EnqueuedTx, TransactionQueueHandle}; + +/// Capacity of the broadcast channel backing [`ClientService::subscribe`]. +/// +/// Subscribers that fall more than this many events behind will receive +/// `RecvError::Lagged` and permanently drop those events. A catch-up sync can +/// emit hundreds or thousands of events in a single burst, so this is sized +/// generously by default. +const EVENT_BROADCAST_CAPACITY: usize = 4096; + +#[cfg(test)] +pub(crate) mod test_utils; + +/// A service wrapper for the Miden client that provides coordination, background +/// sync, and an event system. +/// +/// The service is `Send + Sync` and designed to be used behind an `Arc`. +pub struct ClientService +where + AUTH: TransactionAuthenticator + Send + Sync + 'static, +{ + client: Mutex>, + config: ServiceConfig, + event_tx: broadcast::Sender, + handlers: handlers::HandlerRegistry, + awaiters: awaiters::AwaiterRegistry, + /// Optional custom note filter. When `None`, sync uses the default + /// [`NoteScreener`](miden_client::note::NoteScreener) constructed from the client's store and + /// RPC API. + note_screener: Option>, +} + +impl ClientService +where + AUTH: TransactionAuthenticator + Send + Sync + 'static, +{ + /// Creates a new service wrapping the given client. + pub fn new(client: Client, config: ServiceConfig) -> Self { + Self::build(client, config, None) + } + + /// Creates a new service with a caller-supplied note filter. + /// + /// The screener controls **what gets persisted to the store** during sync — Commit, Insert, + /// or Discard per incoming note. Events emitted by the service mirror persistence: a more + /// aggressive `Discard` policy produces fewer events, a more aggressive `Insert` policy + /// produces more. + /// + /// `OnNoteReceived` is a **filter**, not a listener. It is invoked *during* sync to decide + /// what the client keeps. If all you need is to observe notes (for analytics, UI, etc.), + /// use [`subscribe`](Self::subscribe) or [`on`](Self::on) instead — observers run after + /// persistence and do not affect store state. + /// + /// When `None` is passed (or [`new`](Self::new) is used), the default + /// [`NoteScreener`](miden_client::note::NoteScreener) is used: keep tracked notes, keep + /// public notes whose tag is tracked, keep public notes consumable by a tracked account, + /// discard everything else. + pub fn with_note_screener( + client: Client, + config: ServiceConfig, + screener: Arc, + ) -> Self { + Self::build(client, config, Some(screener)) + } + + fn build( + client: Client, + config: ServiceConfig, + note_screener: Option>, + ) -> Self { + let (event_tx, _) = broadcast::channel(EVENT_BROADCAST_CAPACITY); + Self { + client: Mutex::new(client), + config, + event_tx, + handlers: handlers::HandlerRegistry::new(), + awaiters: awaiters::AwaiterRegistry::new(), + note_screener, + } + } + + /// Creates a new service with default configuration. + pub fn with_default_config(client: Client) -> Self { + Self::new(client, ServiceConfig::default()) + } + + /// Returns a reference to the service configuration. + pub fn config(&self) -> &ServiceConfig { + &self.config + } + + /// Returns a guard providing access to the underlying client. + pub async fn client(&self) -> MutexGuard<'_, Client> { + self.client.lock().await + } + + /// Synchronizes the client state with the network. + /// + /// Acquires exclusive access, fetches the latest state, applies it to the + /// local store, and emits [`ClientEvent`]s for each change observed. + /// + /// When [`ServiceConfig::sync_chunk_blocks`] is set, advances at most that + /// many blocks per call; otherwise syncs to the current chain tip. When a + /// custom note filter was supplied via [`with_note_screener`](Self::with_note_screener), + /// it is used instead of the default screener. + pub async fn sync_state(&self) -> Result { + debug!("Starting coordinated sync"); + let mut client = self.client.lock().await; + + let screener: Arc = match &self.note_screener { + Some(s) => Arc::clone(s), + None => Arc::new(client.note_screener()), + }; + let upper_bound = match self.config.sync_chunk_blocks { + Some(chunk) => { + let current = client.get_sync_height().await?; + SyncTarget::BlockNumber(BlockNumber::from( + current.as_u32().saturating_add(chunk.get()), + )) + }, + None => SyncTarget::CommittedChainTip, + }; + + let state_sync_update = client.get_sync_update_with_screener(screener, upper_bound).await?; + let summary: SyncSummary = (&state_sync_update).into(); + client.apply_state_sync(state_sync_update).await?; + + info!(block_num = ?summary.block_num, "Sync completed"); + + // Enrich the event stream: fetch full records for notes we just inserted so + // subscribers get the body inline without doing their own store roundtrip. + // One batched query per sync, skipped entirely if no new public notes arrived. + let new_note_records = fetch_new_note_records(&client, &summary.new_public_notes).await?; + drop(client); + + self.emit_events(events::events_from_sync(&summary, &new_note_records)); + + Ok(summary) + } + + /// Submits a new transaction. + pub async fn submit_transaction( + &self, + account_id: AccountId, + transaction_request: TransactionRequest, + ) -> Result { + debug!(?account_id, "Starting coordinated transaction"); + let mut client = self.client.lock().await; + let tx_id = client.submit_new_transaction(account_id, transaction_request).await?; + info!(?tx_id, "Transaction submitted"); + Ok(tx_id) + } + + /// Starts a background sync loop that periodically syncs with the network. + /// + /// If [`ServiceConfig::sync_on_start`] is `true` (the default), an + /// immediate sync is run before the first interval tick. + /// + /// Returns a handle that can be used to stop the background sync. + /// If background sync is disabled in the config, returns an inactive handle. + pub fn start_background_sync(self: &Arc) -> BackgroundSyncHandle { + let Some(interval) = self.config.sync_interval else { + let (tx, _) = broadcast::channel(1); + return BackgroundSyncHandle::new(tx); + }; + + let (shutdown_tx, mut shutdown_rx) = broadcast::channel(1); + let service = Arc::clone(self); + let sync_on_start = self.config.sync_on_start; + + tokio::spawn(async move { + info!(?interval, sync_on_start, "Starting background sync"); + + if sync_on_start { + match service.sync_state().await { + Ok(summary) => { + debug!(block_num = ?summary.block_num, "Initial sync completed"); + }, + Err(e) => { + warn!(error = %e, "Initial sync failed"); + }, + } + } + + loop { + tokio::select! { + _ = shutdown_rx.recv() => { + info!("Background sync shutting down"); + break; + } + _ = tokio::time::sleep(interval) => { + match service.sync_state().await { + Ok(summary) => { + debug!(block_num = ?summary.block_num, "Background sync completed"); + } + Err(e) => { + warn!(error = %e, "Background sync failed"); + } + } + } + } + } + }); + + BackgroundSyncHandle::new(shutdown_tx) + } + + // ------------------------------------------------------------------- + // Event system + // ------------------------------------------------------------------- + + /// Returns a receiver for the raw event stream. + pub fn subscribe(&self) -> broadcast::Receiver { + self.event_tx.subscribe() + } + + /// Registers a persistent event handler. + /// + /// The handler fires whenever an event matching `filter` is observed. + /// It receives the [`ClientEvent`] and an `Arc` so it can + /// query client state. Each invocation is spawned as a separate task. + /// + /// Returns a [`HandlerId`] that can be used with + /// [`remove_handler`](Self::remove_handler). + pub fn on(self: &Arc, filter: EventFilter, handler: F) -> HandlerId + where + F: Fn(ClientEvent, Arc>) -> Fut + Send + Sync + 'static, + Fut: Future + Send + 'static, + { + let service = Arc::clone(self); + let stored = handlers::StoredHandler::new(move |event| { + let service = Arc::clone(&service); + handler(event, service) + }); + self.handlers.register(filter, stored) + } + + /// Removes a previously registered handler. + pub fn remove_handler(&self, id: HandlerId) -> bool { + self.handlers.unregister(id) + } + + /// Pre-registers an awaiter for the next event matching `filter`. + /// + /// Returns a future that resolves with the first matching [`ClientEvent`]. + /// The awaiter is registered synchronously before this method returns, so + /// events emitted after this call cannot be missed — including events + /// triggered by actions taken while the future is being awaited. + /// + /// Prefer this over [`once`](Self::once) when awaiting an event caused by + /// an action you're about to perform: + /// + /// ```rust,ignore + /// let awaiter = service.expect(EventFilter::TransactionCommitted(tx_id)); + /// service.submit_transaction(account_id, request).await?; + /// let event = awaiter.await?; // cannot miss, even if sync bursts events + /// ``` + /// + /// Unlike [`subscribe`](Self::subscribe)-based awaiting, matching is + /// guaranteed: awaiters are fulfilled inline during event emission and are + /// not subject to broadcast channel lag. + pub fn expect(&self, filter: EventFilter) -> AwaiterFuture { + self.awaiters.register(filter) + } + + /// Waits for the first event matching `filter`, with an optional timeout. + /// + /// Convenience wrapper over [`expect`](Self::expect) plus + /// [`tokio::time::timeout`]. Returns [`EventTimeoutError`] if the timeout + /// fires before a matching event arrives, or if the service is dropped. + pub async fn once( + &self, + filter: EventFilter, + timeout: Option, + ) -> Result { + let awaiter = self.expect(filter); + match timeout { + Some(duration) => match tokio::time::timeout(duration, awaiter).await { + Ok(Ok(event)) => Ok(event), + Ok(Err(_)) | Err(_) => Err(EventTimeoutError), + }, + None => awaiter.await.map_err(|_| EventTimeoutError), + } + } + + fn emit_events(&self, events: Vec) { + // Fulfill awaiters first: they are lossless and callers depend on them. + self.awaiters.fulfill(&events); + // Hand off to per-handler workers (bounded queues, backpressure on the + // handler side — never blocks emission). + self.handlers.dispatch(&events); + // Best-effort broadcast for raw subscribers. + for event in events { + let _ = self.event_tx.send(event); + } + } +} + +/// Bulk-fetches input-note records for the given IDs and wraps each in an `Arc` for cheap +/// broadcasting. A single store query; returns an empty map if `note_ids` is empty. +async fn fetch_new_note_records( + client: &Client, + note_ids: &[NoteId], +) -> Result>, ClientError> +where + AUTH: TransactionAuthenticator + Send + Sync + 'static, +{ + if note_ids.is_empty() { + return Ok(BTreeMap::new()); + } + Ok(client + .get_input_notes(NoteFilter::List(note_ids.to_vec())) + .await? + .into_iter() + .map(|record| (record.id(), Arc::new(record))) + .collect()) +} + +/// Error returned when a [`once`](ClientService::once) call times out. +#[derive(Debug, Clone)] +pub struct EventTimeoutError; + +impl core::fmt::Display for EventTimeoutError { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { + write!(f, "timed out waiting for event") + } +} + +impl std::error::Error for EventTimeoutError {} + +/// A handle to control background sync operations. +pub struct BackgroundSyncHandle { + shutdown_tx: Option>, +} + +impl BackgroundSyncHandle { + fn new(shutdown_tx: broadcast::Sender<()>) -> Self { + Self { shutdown_tx: Some(shutdown_tx) } + } + + /// Signals the background sync to stop. + pub fn stop(&mut self) { + if let Some(tx) = self.shutdown_tx.take() { + let _ = tx.send(()); + } + } + + /// Returns true if the handle can still control the background sync. + pub fn is_active(&self) -> bool { + self.shutdown_tx.as_ref().is_some_and(|tx| tx.receiver_count() > 0) + } +} + +impl Drop for BackgroundSyncHandle { + fn drop(&mut self) { + self.stop(); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::test_utils::{empty_summary, test_note_arc, test_note_id}; + + fn assert_send_sync() {} + + #[test] + fn service_is_send_sync() { + assert_send_sync::>(); + } + + #[tokio::test] + async fn subscribe_receives_events() { + let (event_tx, _) = broadcast::channel(256); + let mut rx = event_tx.subscribe(); + + let note_id = test_note_id(); + let _ = event_tx.send(ClientEvent::NoteCommitted { note_id }); + + let event = rx.recv().await.unwrap(); + assert!(matches!(event, ClientEvent::NoteCommitted { note_id: id } if id == note_id)); + } + + #[tokio::test] + async fn once_resolves_on_match() { + let (event_tx, _) = broadcast::channel(256); + let mut rx = event_tx.subscribe(); + let note_id = test_note_id(); + + let tx = event_tx.clone(); + tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(10)).await; + let _ = tx.send(ClientEvent::NoteCommitted { note_id }); + }); + + let filter = EventFilter::NoteCommitted(note_id); + let event = tokio::time::timeout(Duration::from_secs(1), async { + loop { + match rx.recv().await { + Ok(event) if filter.matches(&event) => return event, + Ok(_) | Err(broadcast::error::RecvError::Lagged(_)) => {}, + Err(broadcast::error::RecvError::Closed) => unreachable!(), + } + } + }) + .await + .unwrap(); + + assert_eq!(event.note_id().unwrap(), note_id); + } + + #[tokio::test] + async fn once_times_out() { + let (_event_tx, mut rx) = broadcast::channel::(256); + + let result = tokio::time::timeout(Duration::from_millis(50), async { + loop { + match rx.recv().await { + Ok(_) | Err(broadcast::error::RecvError::Lagged(_)) => {}, + Err(broadcast::error::RecvError::Closed) => { + std::future::pending::<()>().await; + unreachable!() + }, + } + } + }) + .await; + + assert!(result.is_err()); + } + + #[tokio::test] + async fn handler_fires_for_matching_event() { + let registry = handlers::HandlerRegistry::new(); + let note_id = test_note_id(); + + let (notify_tx, notify_rx) = tokio::sync::oneshot::channel(); + let notify_tx = Arc::new(std::sync::Mutex::new(Some(notify_tx))); + + let handler = handlers::StoredHandler::new(move |_event| { + let tx = notify_tx.clone(); + async move { + if let Some(tx) = tx.lock().unwrap().take() { + let _ = tx.send(()); + } + } + }); + + registry.register(EventFilter::NoteCommitted(note_id), handler); + registry.dispatch(&[ClientEvent::NoteCommitted { note_id }]); + + tokio::time::timeout(Duration::from_secs(1), notify_rx) + .await + .expect("timed out") + .expect("channel dropped"); + } + + #[tokio::test] + async fn handler_skips_non_matching_event() { + let registry = handlers::HandlerRegistry::new(); + let note_id = test_note_id(); + + let (notify_tx, notify_rx) = tokio::sync::oneshot::channel(); + let notify_tx = Arc::new(std::sync::Mutex::new(Some(notify_tx))); + + let handler = handlers::StoredHandler::new(move |_event| { + let tx = notify_tx.clone(); + async move { + if let Some(tx) = tx.lock().unwrap().take() { + let _ = tx.send(()); + } + } + }); + + registry.register(EventFilter::NoteCommitted(note_id), handler); + registry.dispatch(&[ClientEvent::NoteReceived { note: test_note_arc() }]); + + let result = tokio::time::timeout(Duration::from_millis(50), notify_rx).await; + assert!(result.is_err()); + } + + #[tokio::test] + async fn events_from_sync_integration() { + let note_id = test_note_id(); + let note = test_note_arc(); + let mut summary = empty_summary(); + summary.new_public_notes = vec![note_id]; + + let mut records = BTreeMap::new(); + records.insert(note_id, Arc::clone(¬e)); + + let events = events::events_from_sync(&summary, &records); + assert_eq!(events.len(), 2); + assert_eq!(events[0].note_id().unwrap(), note_id); + assert!(events[1].summary().is_some()); + } + + #[test] + fn client_event_accessors() { + let note_id = test_note_id(); + let event = ClientEvent::NoteReceived { note: test_note_arc() }; + + assert_eq!(event.note_id(), Some(note_id)); + assert_eq!(event.transaction_id(), None); + assert_eq!(event.account_id(), None); + assert!(event.summary().is_none()); + } +} diff --git a/crates/client-service/src/test_utils.rs b/crates/client-service/src/test_utils.rs new file mode 100644 index 0000000000..7c05e0aad3 --- /dev/null +++ b/crates/client-service/src/test_utils.rs @@ -0,0 +1,43 @@ +use std::sync::Arc; + +use miden_client::store::InputNoteRecord; +use miden_client::sync::SyncSummary; +use miden_client::testing::NoteBuilder; +use miden_client::testing::account_id::ACCOUNT_ID_SENDER; +use miden_protocol::account::AccountId; +use miden_protocol::block::BlockNumber; +use miden_protocol::note::NoteId; +use rand::SeedableRng; +use rand::rngs::SmallRng; + +/// Returns a deterministic [`InputNoteRecord`] built from a fixed sender and RNG seed. +/// +/// All calls to this function produce a record with the same [`NoteId`], which is what +/// [`test_note_id`] returns. +pub(crate) fn test_note_record() -> InputNoteRecord { + let sender = AccountId::try_from(ACCOUNT_ID_SENDER).expect("valid sender id"); + // Fixed seed → deterministic serial_num → stable NoteId across calls. + let rng = SmallRng::seed_from_u64(0x00c0_ffee); + let note = NoteBuilder::new(sender, rng).build().expect("build test note"); + InputNoteRecord::from(note) +} + +pub(crate) fn test_note_arc() -> Arc { + Arc::new(test_note_record()) +} + +pub(crate) fn test_note_id() -> NoteId { + test_note_record().id() +} + +pub(crate) fn empty_summary() -> SyncSummary { + SyncSummary { + block_num: BlockNumber::from(0u32), + new_public_notes: vec![], + committed_notes: vec![], + consumed_notes: vec![], + updated_accounts: vec![], + locked_accounts: vec![], + committed_transactions: vec![], + } +} diff --git a/crates/client-service/src/tx_queue.rs b/crates/client-service/src/tx_queue.rs new file mode 100644 index 0000000000..547cb29c45 --- /dev/null +++ b/crates/client-service/src/tx_queue.rs @@ -0,0 +1,215 @@ +//! Sequential transaction submission queue. +//! +//! [`ClientService`](crate::ClientService) serializes mutations behind a single client mutex. +//! For most callers this is fine — you just `await` `submit_transaction` and let the mutex +//! queue you in FIFO order. When a caller wants **fire-and-forget** semantics — "take this +//! request, give me back a handle, process it whenever the client is free" — that's what the +//! transaction queue is for. +//! +//! A worker task owns an unbounded mpsc of pending submissions. Each enqueue returns an +//! [`EnqueuedTx`] future that resolves once the worker has submitted the transaction (or +//! failed). Dropping the [`TransactionQueueHandle`] closes the channel and the worker exits +//! after draining; in-flight submissions complete, not-yet-started submissions resolve with +//! [`EnqueueError::QueueShutDown`]. +//! +//! ## Example +//! +//! ```rust,ignore +//! let queue = service.start_transaction_queue(); +//! +//! // React to events by enqueueing a transaction — handler returns immediately. +//! let svc = Arc::clone(&service); +//! let q = queue.clone(); +//! svc.on(EventFilter::AnyNoteCommitted, move |_event, service| { +//! let q = q.clone(); +//! async move { +//! let client = service.client().await; +//! let req = build_some_request(&*client).await.unwrap(); +//! let account_id = client.account_id_to_act_on(); +//! drop(client); +//! +//! let handle = q.enqueue(account_id, req); +//! // Fire-and-forget: drop the handle, or tokio::spawn to log the result. +//! tokio::spawn(async move { +//! match handle.await { +//! Ok(tx_id) => tracing::info!(?tx_id, "tx submitted"), +//! Err(e) => tracing::warn!(error = %e, "tx failed"), +//! } +//! }); +//! } +//! }); +//! ``` + +use std::future::Future; +use std::pin::Pin; +use std::sync::Arc; +use std::task::{Context, Poll}; + +use miden_client::ClientError; +use miden_client::auth::TransactionAuthenticator; +use miden_client::transaction::{TransactionId, TransactionRequest}; +use miden_protocol::account::AccountId; +use tokio::sync::{mpsc, oneshot}; +use tracing::debug; + +use crate::ClientService; + +/// Internal envelope carrying a submission request to the worker. +struct QueuedTx { + account_id: AccountId, + request: TransactionRequest, + result: oneshot::Sender>, +} + +/// Handle to a running transaction queue. +/// +/// Cheap to clone — multiple producers can share one queue. The worker shuts down once the +/// last handle is dropped (and the mpsc channel closes). +#[derive(Clone)] +pub struct TransactionQueueHandle { + sender: mpsc::UnboundedSender, +} + +impl TransactionQueueHandle { + /// Enqueues a transaction for sequential submission. Returns a future resolving to the + /// submission result. + /// + /// Dropping the returned [`EnqueuedTx`] before it resolves is safe — the submission still + /// runs, the result is just discarded. (Fire-and-forget mode.) + pub fn enqueue(&self, account_id: AccountId, request: TransactionRequest) -> EnqueuedTx { + let (result_tx, result_rx) = oneshot::channel(); + match self.sender.send(QueuedTx { account_id, request, result: result_tx }) { + Ok(()) => EnqueuedTx { rx: Some(result_rx) }, + // Queue already shut down; the future will resolve synchronously to an error. + Err(_) => EnqueuedTx { rx: None }, + } + } + + /// Returns `true` if the worker is still running (the channel has not been closed). + pub fn is_active(&self) -> bool { + !self.sender.is_closed() + } +} + +/// Future returned by [`TransactionQueueHandle::enqueue`]. +/// +/// Resolves to the submitted [`TransactionId`] on success, or an [`EnqueueError`] on failure. +pub struct EnqueuedTx { + rx: Option>>, +} + +impl Future for EnqueuedTx { + type Output = Result; + + fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + let Some(rx) = self.rx.as_mut() else { + return Poll::Ready(Err(EnqueueError::QueueShutDown)); + }; + match Pin::new(rx).poll(cx) { + Poll::Ready(Ok(Ok(tx_id))) => Poll::Ready(Ok(tx_id)), + Poll::Ready(Ok(Err(e))) => Poll::Ready(Err(EnqueueError::Submission(e))), + // Sender dropped without sending — worker exited before processing. + Poll::Ready(Err(_)) => Poll::Ready(Err(EnqueueError::QueueShutDown)), + Poll::Pending => Poll::Pending, + } + } +} + +/// Error returned by [`EnqueuedTx`]. +#[derive(Debug)] +pub enum EnqueueError { + /// The queue was shut down before the transaction was submitted. Either the + /// [`TransactionQueueHandle`] was dropped, or the service was dropped. + QueueShutDown, + /// Submission reached the network layer and failed. + Submission(ClientError), +} + +impl core::fmt::Display for EnqueueError { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { + match self { + Self::QueueShutDown => { + write!(f, "transaction queue was shut down before submission") + }, + Self::Submission(_) => write!(f, "transaction submission failed"), + } + } +} + +impl std::error::Error for EnqueueError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + Self::QueueShutDown => None, + Self::Submission(e) => Some(e), + } + } +} + +impl From for EnqueueError { + fn from(value: ClientError) -> Self { + Self::Submission(value) + } +} + +impl ClientService +where + AUTH: TransactionAuthenticator + Send + Sync + 'static, +{ + /// Starts a worker task that processes enqueued transactions sequentially. + /// + /// Transactions submitted via the returned handle are processed in FIFO order: each waits + /// for the previous to complete. The worker calls + /// [`submit_transaction`](Self::submit_transaction) internally, so queued transactions + /// share the same mutex as [`sync_state`](Self::sync_state) and direct client access — + /// they don't bypass coordination, they just free the caller from `await`ing synchronously. + /// + /// Returns a [`TransactionQueueHandle`] that can be cloned and shared. The worker shuts + /// down when all handles are dropped. + pub fn start_transaction_queue(self: &Arc) -> TransactionQueueHandle { + let (sender, mut receiver) = mpsc::unbounded_channel::(); + let service = Arc::clone(self); + + tokio::spawn(async move { + debug!("Transaction queue worker starting"); + while let Some(queued) = receiver.recv().await { + let result = service.submit_transaction(queued.account_id, queued.request).await; + // Result receiver may have been dropped (fire-and-forget caller) — ignore. + let _ = queued.result.send(result); + } + debug!("Transaction queue worker exiting"); + }); + + TransactionQueueHandle { sender } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn enqueue_on_dropped_handle_resolves_to_shutdown_error() { + // Simulate a handle that has no live worker by closing the channel. + let (sender, receiver) = mpsc::unbounded_channel::(); + drop(receiver); + let handle = TransactionQueueHandle { sender }; + + assert!(!handle.is_active()); + + // We can't easily construct a TransactionRequest in a unit test (it needs a real + // client + note), but we can verify the error path by constructing an EnqueuedTx + // whose rx is `None` directly. + let (dropped_tx, dropped_rx) = oneshot::channel::>(); + drop(dropped_tx); + let future = EnqueuedTx { rx: Some(dropped_rx) }; + let err = future.await.expect_err("should fail when sender drops"); + assert!(matches!(err, EnqueueError::QueueShutDown)); + } + + #[tokio::test] + async fn enqueue_on_never_connected_handle_is_immediate_error() { + let future = EnqueuedTx { rx: None }; + let err = future.await.expect_err("None rx should immediately error"); + assert!(matches!(err, EnqueueError::QueueShutDown)); + } +} diff --git a/crates/idxdb-store/src/yarn.lock b/crates/idxdb-store/src/yarn.lock index 5f9a304299..2f1a36f3ab 100644 --- a/crates/idxdb-store/src/yarn.lock +++ b/crates/idxdb-store/src/yarn.lock @@ -252,130 +252,130 @@ "@nodelib/fs.scandir" "2.1.5" fastq "^1.6.0" -"@rollup/rollup-android-arm-eabi@4.59.0": - version "4.59.0" - resolved "https://registry.yarnpkg.com/@rollup/rollup-android-arm-eabi/-/rollup-android-arm-eabi-4.59.0.tgz#a6742c74c7d9d6d604ef8a48f99326b4ecda3d82" - integrity sha512-upnNBkA6ZH2VKGcBj9Fyl9IGNPULcjXRlg0LLeaioQWueH30p6IXtJEbKAgvyv+mJaMxSm1l6xwDXYjpEMiLMg== - -"@rollup/rollup-android-arm64@4.59.0": - version "4.59.0" - resolved "https://registry.yarnpkg.com/@rollup/rollup-android-arm64/-/rollup-android-arm64-4.59.0.tgz#97247be098de4df0c11971089fd2edf80a5da8cf" - integrity sha512-hZ+Zxj3SySm4A/DylsDKZAeVg0mvi++0PYVceVyX7hemkw7OreKdCvW2oQ3T1FMZvCaQXqOTHb8qmBShoqk69Q== - -"@rollup/rollup-darwin-arm64@4.59.0": - version "4.59.0" - resolved "https://registry.yarnpkg.com/@rollup/rollup-darwin-arm64/-/rollup-darwin-arm64-4.59.0.tgz#674852cf14cf11b8056e0b1a2f4e872b523576cf" - integrity sha512-W2Psnbh1J8ZJw0xKAd8zdNgF9HRLkdWwwdWqubSVk0pUuQkoHnv7rx4GiF9rT4t5DIZGAsConRE3AxCdJ4m8rg== - -"@rollup/rollup-darwin-x64@4.59.0": - version "4.59.0" - resolved "https://registry.yarnpkg.com/@rollup/rollup-darwin-x64/-/rollup-darwin-x64-4.59.0.tgz#36dfd7ed0aaf4d9d89d9ef983af72632455b0246" - integrity sha512-ZW2KkwlS4lwTv7ZVsYDiARfFCnSGhzYPdiOU4IM2fDbL+QGlyAbjgSFuqNRbSthybLbIJ915UtZBtmuLrQAT/w== - -"@rollup/rollup-freebsd-arm64@4.59.0": - version "4.59.0" - resolved "https://registry.yarnpkg.com/@rollup/rollup-freebsd-arm64/-/rollup-freebsd-arm64-4.59.0.tgz#2f87c2074b4220260fdb52a9996246edfc633c22" - integrity sha512-EsKaJ5ytAu9jI3lonzn3BgG8iRBjV4LxZexygcQbpiU0wU0ATxhNVEpXKfUa0pS05gTcSDMKpn3Sx+QB9RlTTA== - -"@rollup/rollup-freebsd-x64@4.59.0": - version "4.59.0" - resolved "https://registry.yarnpkg.com/@rollup/rollup-freebsd-x64/-/rollup-freebsd-x64-4.59.0.tgz#9b5a26522a38a95dc06616d1939d4d9a76937803" - integrity sha512-d3DuZi2KzTMjImrxoHIAODUZYoUUMsuUiY4SRRcJy6NJoZ6iIqWnJu9IScV9jXysyGMVuW+KNzZvBLOcpdl3Vg== - -"@rollup/rollup-linux-arm-gnueabihf@4.59.0": - version "4.59.0" - resolved "https://registry.yarnpkg.com/@rollup/rollup-linux-arm-gnueabihf/-/rollup-linux-arm-gnueabihf-4.59.0.tgz#86aa4859385a8734235b5e40a48e52d770758c3a" - integrity sha512-t4ONHboXi/3E0rT6OZl1pKbl2Vgxf9vJfWgmUoCEVQVxhW6Cw/c8I6hbbu7DAvgp82RKiH7TpLwxnJeKv2pbsw== - -"@rollup/rollup-linux-arm-musleabihf@4.59.0": - version "4.59.0" - resolved "https://registry.yarnpkg.com/@rollup/rollup-linux-arm-musleabihf/-/rollup-linux-arm-musleabihf-4.59.0.tgz#cbe70e56e6ece8dac83eb773b624fc9e5a460976" - integrity sha512-CikFT7aYPA2ufMD086cVORBYGHffBo4K8MQ4uPS/ZnY54GKj36i196u8U+aDVT2LX4eSMbyHtyOh7D7Zvk2VvA== - -"@rollup/rollup-linux-arm64-gnu@4.59.0": - version "4.59.0" - resolved "https://registry.yarnpkg.com/@rollup/rollup-linux-arm64-gnu/-/rollup-linux-arm64-gnu-4.59.0.tgz#d14992a2e653bc3263d284bc6579b7a2890e1c45" - integrity sha512-jYgUGk5aLd1nUb1CtQ8E+t5JhLc9x5WdBKew9ZgAXg7DBk0ZHErLHdXM24rfX+bKrFe+Xp5YuJo54I5HFjGDAA== - -"@rollup/rollup-linux-arm64-musl@4.59.0": - version "4.59.0" - resolved "https://registry.yarnpkg.com/@rollup/rollup-linux-arm64-musl/-/rollup-linux-arm64-musl-4.59.0.tgz#2fdd1ddc434ea90aeaa0851d2044789b4d07f6da" - integrity sha512-peZRVEdnFWZ5Bh2KeumKG9ty7aCXzzEsHShOZEFiCQlDEepP1dpUl/SrUNXNg13UmZl+gzVDPsiCwnV1uI0RUA== - -"@rollup/rollup-linux-loong64-gnu@4.59.0": - version "4.59.0" - resolved "https://registry.yarnpkg.com/@rollup/rollup-linux-loong64-gnu/-/rollup-linux-loong64-gnu-4.59.0.tgz#8a181e6f89f969f21666a743cd411416c80099e7" - integrity sha512-gbUSW/97f7+r4gHy3Jlup8zDG190AuodsWnNiXErp9mT90iCy9NKKU0Xwx5k8VlRAIV2uU9CsMnEFg/xXaOfXg== - -"@rollup/rollup-linux-loong64-musl@4.59.0": - version "4.59.0" - resolved "https://registry.yarnpkg.com/@rollup/rollup-linux-loong64-musl/-/rollup-linux-loong64-musl-4.59.0.tgz#904125af2babc395f8061daa27b5af1f4e3f2f78" - integrity sha512-yTRONe79E+o0FWFijasoTjtzG9EBedFXJMl888NBEDCDV9I2wGbFFfJQQe63OijbFCUZqxpHz1GzpbtSFikJ4Q== - -"@rollup/rollup-linux-ppc64-gnu@4.59.0": - version "4.59.0" - resolved "https://registry.yarnpkg.com/@rollup/rollup-linux-ppc64-gnu/-/rollup-linux-ppc64-gnu-4.59.0.tgz#a57970ac6864c9a3447411a658224bdcf948be22" - integrity sha512-sw1o3tfyk12k3OEpRddF68a1unZ5VCN7zoTNtSn2KndUE+ea3m3ROOKRCZxEpmT9nsGnogpFP9x6mnLTCaoLkA== - -"@rollup/rollup-linux-ppc64-musl@4.59.0": - version "4.59.0" - resolved "https://registry.yarnpkg.com/@rollup/rollup-linux-ppc64-musl/-/rollup-linux-ppc64-musl-4.59.0.tgz#bb84de5b26870567a4267666e08891e80bb56a63" - integrity sha512-+2kLtQ4xT3AiIxkzFVFXfsmlZiG5FXYW7ZyIIvGA7Bdeuh9Z0aN4hVyXS/G1E9bTP/vqszNIN/pUKCk/BTHsKA== - -"@rollup/rollup-linux-riscv64-gnu@4.59.0": - version "4.59.0" - resolved "https://registry.yarnpkg.com/@rollup/rollup-linux-riscv64-gnu/-/rollup-linux-riscv64-gnu-4.59.0.tgz#72d00d2c7fb375ce3564e759db33f17a35bffab9" - integrity sha512-NDYMpsXYJJaj+I7UdwIuHHNxXZ/b/N2hR15NyH3m2qAtb/hHPA4g4SuuvrdxetTdndfj9b1WOmy73kcPRoERUg== - -"@rollup/rollup-linux-riscv64-musl@4.59.0": - version "4.59.0" - resolved "https://registry.yarnpkg.com/@rollup/rollup-linux-riscv64-musl/-/rollup-linux-riscv64-musl-4.59.0.tgz#4c166ef58e718f9245bd31873384ba15a5c1a883" - integrity sha512-nLckB8WOqHIf1bhymk+oHxvM9D3tyPndZH8i8+35p/1YiVoVswPid2yLzgX7ZJP0KQvnkhM4H6QZ5m0LzbyIAg== - -"@rollup/rollup-linux-s390x-gnu@4.59.0": - version "4.59.0" - resolved "https://registry.yarnpkg.com/@rollup/rollup-linux-s390x-gnu/-/rollup-linux-s390x-gnu-4.59.0.tgz#bb5025cde9a61db478c2ca7215808ad3bce73a09" - integrity sha512-oF87Ie3uAIvORFBpwnCvUzdeYUqi2wY6jRFWJAy1qus/udHFYIkplYRW+wo+GRUP4sKzYdmE1Y3+rY5Gc4ZO+w== - -"@rollup/rollup-linux-x64-gnu@4.59.0": - version "4.59.0" - resolved "https://registry.yarnpkg.com/@rollup/rollup-linux-x64-gnu/-/rollup-linux-x64-gnu-4.59.0.tgz#9b66b1f9cd95c6624c788f021c756269ffed1552" - integrity sha512-3AHmtQq/ppNuUspKAlvA8HtLybkDflkMuLK4DPo77DfthRb71V84/c4MlWJXixZz4uruIH4uaa07IqoAkG64fg== - -"@rollup/rollup-linux-x64-musl@4.59.0": - version "4.59.0" - resolved "https://registry.yarnpkg.com/@rollup/rollup-linux-x64-musl/-/rollup-linux-x64-musl-4.59.0.tgz#b007ca255dc7166017d57d7d2451963f0bd23fd9" - integrity sha512-2UdiwS/9cTAx7qIUZB/fWtToJwvt0Vbo0zmnYt7ED35KPg13Q0ym1g442THLC7VyI6JfYTP4PiSOWyoMdV2/xg== - -"@rollup/rollup-openbsd-x64@4.59.0": - version "4.59.0" - resolved "https://registry.yarnpkg.com/@rollup/rollup-openbsd-x64/-/rollup-openbsd-x64-4.59.0.tgz#e8b357b2d1aa2c8d76a98f5f0d889eabe93f4ef9" - integrity sha512-M3bLRAVk6GOwFlPTIxVBSYKUaqfLrn8l0psKinkCFxl4lQvOSz8ZrKDz2gxcBwHFpci0B6rttydI4IpS4IS/jQ== - -"@rollup/rollup-openharmony-arm64@4.59.0": - version "4.59.0" - resolved "https://registry.yarnpkg.com/@rollup/rollup-openharmony-arm64/-/rollup-openharmony-arm64-4.59.0.tgz#96c2e3f4aacd3d921981329831ff8dde492204dc" - integrity sha512-tt9KBJqaqp5i5HUZzoafHZX8b5Q2Fe7UjYERADll83O4fGqJ49O1FsL6LpdzVFQcpwvnyd0i+K/VSwu/o/nWlA== - -"@rollup/rollup-win32-arm64-msvc@4.59.0": - version "4.59.0" - resolved "https://registry.yarnpkg.com/@rollup/rollup-win32-arm64-msvc/-/rollup-win32-arm64-msvc-4.59.0.tgz#2d865149d706d938df8b4b8f117e69a77646d581" - integrity sha512-V5B6mG7OrGTwnxaNUzZTDTjDS7F75PO1ae6MJYdiMu60sq0CqN5CVeVsbhPxalupvTX8gXVSU9gq+Rx1/hvu6A== - -"@rollup/rollup-win32-ia32-msvc@4.59.0": - version "4.59.0" - resolved "https://registry.yarnpkg.com/@rollup/rollup-win32-ia32-msvc/-/rollup-win32-ia32-msvc-4.59.0.tgz#abe1593be0fa92325e9971c8da429c5e05b92c36" - integrity sha512-UKFMHPuM9R0iBegwzKF4y0C4J9u8C6MEJgFuXTBerMk7EJ92GFVFYBfOZaSGLu6COf7FxpQNqhNS4c4icUPqxA== - -"@rollup/rollup-win32-x64-gnu@4.59.0": - version "4.59.0" - resolved "https://registry.yarnpkg.com/@rollup/rollup-win32-x64-gnu/-/rollup-win32-x64-gnu-4.59.0.tgz#c4af3e9518c9a5cd4b1c163dc81d0ad4d82e7eab" - integrity sha512-laBkYlSS1n2L8fSo1thDNGrCTQMmxjYY5G0WFWjFFYZkKPjsMBsgJfGf4TLxXrF6RyhI60L8TMOjBMvXiTcxeA== - -"@rollup/rollup-win32-x64-msvc@4.59.0": - version "4.59.0" - resolved "https://registry.yarnpkg.com/@rollup/rollup-win32-x64-msvc/-/rollup-win32-x64-msvc-4.59.0.tgz#4584a8a87b29188a4c1fe987a9fcf701e256d86c" - integrity sha512-2HRCml6OztYXyJXAvdDXPKcawukWY2GpR5/nxKp4iBgiO3wcoEGkAaqctIbZcNB6KlUQBIqt8VYkNSj2397EfA== +"@rollup/rollup-android-arm-eabi@4.60.1": + version "4.60.1" + resolved "https://registry.yarnpkg.com/@rollup/rollup-android-arm-eabi/-/rollup-android-arm-eabi-4.60.1.tgz#043f145716234529052ef9e1ce1d847ffbe9e674" + integrity sha512-d6FinEBLdIiK+1uACUttJKfgZREXrF0Qc2SmLII7W2AD8FfiZ9Wjd+rD/iRuf5s5dWrr1GgwXCvPqOuDquOowA== + +"@rollup/rollup-android-arm64@4.60.1": + version "4.60.1" + resolved "https://registry.yarnpkg.com/@rollup/rollup-android-arm64/-/rollup-android-arm64-4.60.1.tgz#023e1bd146e7519087dfd9e8b29e4cf9f8ecd35c" + integrity sha512-YjG/EwIDvvYI1YvYbHvDz/BYHtkY4ygUIXHnTdLhG+hKIQFBiosfWiACWortsKPKU/+dUwQQCKQM3qrDe8c9BA== + +"@rollup/rollup-darwin-arm64@4.60.1": + version "4.60.1" + resolved "https://registry.yarnpkg.com/@rollup/rollup-darwin-arm64/-/rollup-darwin-arm64-4.60.1.tgz#55ccb5487c02419954c57a7a80602885d616e1ee" + integrity sha512-mjCpF7GmkRtSJwon+Rq1N8+pI+8l7w5g9Z3vWj4T7abguC4Czwi3Yu/pFaLvA3TTeMVjnu3ctigusqWUfjZzvw== + +"@rollup/rollup-darwin-x64@4.60.1": + version "4.60.1" + resolved "https://registry.yarnpkg.com/@rollup/rollup-darwin-x64/-/rollup-darwin-x64-4.60.1.tgz#254b65404b14488c83225e88b8819376ad71a784" + integrity sha512-haZ7hJ1JT4e9hqkoT9R/19XW2QKqjfJVv+i5AGg57S+nLk9lQnJ1F/eZloRO3o9Scy9CM3wQ9l+dkXtcBgN5Ew== + +"@rollup/rollup-freebsd-arm64@4.60.1": + version "4.60.1" + resolved "https://registry.yarnpkg.com/@rollup/rollup-freebsd-arm64/-/rollup-freebsd-arm64-4.60.1.tgz#6377ff38c052c76fcaffb7b2728d3172fe676fe6" + integrity sha512-czw90wpQq3ZsAVBlinZjAYTKduOjTywlG7fEeWKUA7oCmpA8xdTkxZZlwNJKWqILlq0wehoZcJYfBvOyhPTQ6w== + +"@rollup/rollup-freebsd-x64@4.60.1": + version "4.60.1" + resolved "https://registry.yarnpkg.com/@rollup/rollup-freebsd-x64/-/rollup-freebsd-x64-4.60.1.tgz#ba3902309d088eaf7139b916f09b7140b28b406d" + integrity sha512-KVB2rqsxTHuBtfOeySEyzEOB7ltlB/ux38iu2rBQzkjbwRVlkhAGIEDiiYnO2kFOkJp+Z7pUXKyrRRFuFUKt+g== + +"@rollup/rollup-linux-arm-gnueabihf@4.60.1": + version "4.60.1" + resolved "https://registry.yarnpkg.com/@rollup/rollup-linux-arm-gnueabihf/-/rollup-linux-arm-gnueabihf-4.60.1.tgz#e011b9a14638267e53b446286e838dbdaf53f167" + integrity sha512-L+34Qqil+v5uC0zEubW7uByo78WOCIrBvci69E7sFASRl0X7b/MB6Cqd1lky/CtcSVTydWa2WZwFuWexjS5o6g== + +"@rollup/rollup-linux-arm-musleabihf@4.60.1": + version "4.60.1" + resolved "https://registry.yarnpkg.com/@rollup/rollup-linux-arm-musleabihf/-/rollup-linux-arm-musleabihf-4.60.1.tgz#0bce9ce9a009490abd28fd922dd97ed521311afe" + integrity sha512-n83O8rt4v34hgFzlkb1ycniJh7IR5RCIqt6mz1VRJD6pmhRi0CXdmfnLu9dIUS6buzh60IvACM842Ffb3xd6Gg== + +"@rollup/rollup-linux-arm64-gnu@4.60.1": + version "4.60.1" + resolved "https://registry.yarnpkg.com/@rollup/rollup-linux-arm64-gnu/-/rollup-linux-arm64-gnu-4.60.1.tgz#6f6cfbbf324fbb4ceff213abdf7f322fd45d25ff" + integrity sha512-Nql7sTeAzhTAja3QXeAI48+/+GjBJ+QmAH13snn0AJSNL50JsDqotyudHyMbO2RbJkskbMbFJfIJKWA6R1LCJQ== + +"@rollup/rollup-linux-arm64-musl@4.60.1": + version "4.60.1" + resolved "https://registry.yarnpkg.com/@rollup/rollup-linux-arm64-musl/-/rollup-linux-arm64-musl-4.60.1.tgz#f7cb3eecaea9c151ef77342af05f38ae924bf795" + integrity sha512-+pUymDhd0ys9GcKZPPWlFiZ67sTWV5UU6zOJat02M1+PiuSGDziyRuI/pPue3hoUwm2uGfxdL+trT6Z9rxnlMA== + +"@rollup/rollup-linux-loong64-gnu@4.60.1": + version "4.60.1" + resolved "https://registry.yarnpkg.com/@rollup/rollup-linux-loong64-gnu/-/rollup-linux-loong64-gnu-4.60.1.tgz#499bfac6bb669fd88bb664357bf6be996a28b92f" + integrity sha512-VSvgvQeIcsEvY4bKDHEDWcpW4Yw7BtlKG1GUT4FzBUlEKQK0rWHYBqQt6Fm2taXS+1bXvJT6kICu5ZwqKCnvlQ== + +"@rollup/rollup-linux-loong64-musl@4.60.1": + version "4.60.1" + resolved "https://registry.yarnpkg.com/@rollup/rollup-linux-loong64-musl/-/rollup-linux-loong64-musl-4.60.1.tgz#127dfac08764764396bbe04453c545d38a3ab518" + integrity sha512-4LqhUomJqwe641gsPp6xLfhqWMbQV04KtPp7/dIp0nzPxAkNY1AbwL5W0MQpcalLYk07vaW9Kp1PBhdpZYYcEw== + +"@rollup/rollup-linux-ppc64-gnu@4.60.1": + version "4.60.1" + resolved "https://registry.yarnpkg.com/@rollup/rollup-linux-ppc64-gnu/-/rollup-linux-ppc64-gnu-4.60.1.tgz#6a72f4d95852aac18326c5bf708393e8f3a41b70" + integrity sha512-tLQQ9aPvkBxOc/EUT6j3pyeMD6Hb8QF2BTBnCQWP/uu1lhc9AIrIjKnLYMEroIz/JvtGYgI9dF3AxHZNaEH0rw== + +"@rollup/rollup-linux-ppc64-musl@4.60.1": + version "4.60.1" + resolved "https://registry.yarnpkg.com/@rollup/rollup-linux-ppc64-musl/-/rollup-linux-ppc64-musl-4.60.1.tgz#ba8674666b00d6f9066cb9a5771a8430c34d2de6" + integrity sha512-RMxFhJwc9fSXP6PqmAz4cbv3kAyvD1etJFjTx4ONqFP9DkTkXsAMU4v3Vyc5BgzC+anz7nS/9tp4obsKfqkDHg== + +"@rollup/rollup-linux-riscv64-gnu@4.60.1": + version "4.60.1" + resolved "https://registry.yarnpkg.com/@rollup/rollup-linux-riscv64-gnu/-/rollup-linux-riscv64-gnu-4.60.1.tgz#17cc38b2a71e302547cad29bcf78d0db2618c922" + integrity sha512-QKgFl+Yc1eEk6MmOBfRHYF6lTxiiiV3/z/BRrbSiW2I7AFTXoBFvdMEyglohPj//2mZS4hDOqeB0H1ACh3sBbg== + +"@rollup/rollup-linux-riscv64-musl@4.60.1": + version "4.60.1" + resolved "https://registry.yarnpkg.com/@rollup/rollup-linux-riscv64-musl/-/rollup-linux-riscv64-musl-4.60.1.tgz#e36a41e2d8bd247331bd5cfc13b8c951d33454a2" + integrity sha512-RAjXjP/8c6ZtzatZcA1RaQr6O1TRhzC+adn8YZDnChliZHviqIjmvFwHcxi4JKPSDAt6Uhf/7vqcBzQJy0PDJg== + +"@rollup/rollup-linux-s390x-gnu@4.60.1": + version "4.60.1" + resolved "https://registry.yarnpkg.com/@rollup/rollup-linux-s390x-gnu/-/rollup-linux-s390x-gnu-4.60.1.tgz#1687265f1f4bdea0726c761a58c2db9933609d68" + integrity sha512-wcuocpaOlaL1COBYiA89O6yfjlp3RwKDeTIA0hM7OpmhR1Bjo9j31G1uQVpDlTvwxGn2nQs65fBFL5UFd76FcQ== + +"@rollup/rollup-linux-x64-gnu@4.60.1": + version "4.60.1" + resolved "https://registry.yarnpkg.com/@rollup/rollup-linux-x64-gnu/-/rollup-linux-x64-gnu-4.60.1.tgz#56a6a0d9076f2a05a976031493b24a20ddcc0e77" + integrity sha512-77PpsFQUCOiZR9+LQEFg9GClyfkNXj1MP6wRnzYs0EeWbPcHs02AXu4xuUbM1zhwn3wqaizle3AEYg5aeoohhg== + +"@rollup/rollup-linux-x64-musl@4.60.1": + version "4.60.1" + resolved "https://registry.yarnpkg.com/@rollup/rollup-linux-x64-musl/-/rollup-linux-x64-musl-4.60.1.tgz#bc240ebb5b9fd8d41ca8a80cb458452e8c187e0f" + integrity sha512-5cIATbk5vynAjqqmyBjlciMJl1+R/CwX9oLk/EyiFXDWd95KpHdrOJT//rnUl4cUcskrd0jCCw3wpZnhIHdD9w== + +"@rollup/rollup-openbsd-x64@4.60.1": + version "4.60.1" + resolved "https://registry.yarnpkg.com/@rollup/rollup-openbsd-x64/-/rollup-openbsd-x64-4.60.1.tgz#6f80d48a006c4b2ffa7724e95a3e33f6975872af" + integrity sha512-cl0w09WsCi17mcmWqqglez9Gk8isgeWvoUZ3WiJFYSR3zjBQc2J5/ihSjpl+VLjPqjQ/1hJRcqBfLjssREQILw== + +"@rollup/rollup-openharmony-arm64@4.60.1": + version "4.60.1" + resolved "https://registry.yarnpkg.com/@rollup/rollup-openharmony-arm64/-/rollup-openharmony-arm64-4.60.1.tgz#8f6db6f70d0a48abd833b263cd6dd3e7199c4c0e" + integrity sha512-4Cv23ZrONRbNtbZa37mLSueXUCtN7MXccChtKpUnQNgF010rjrjfHx3QxkS2PI7LqGT5xXyYs1a7LbzAwT0iCA== + +"@rollup/rollup-win32-arm64-msvc@4.60.1": + version "4.60.1" + resolved "https://registry.yarnpkg.com/@rollup/rollup-win32-arm64-msvc/-/rollup-win32-arm64-msvc-4.60.1.tgz#b68989bfa815d0b3d4e302ecd90bda744438b177" + integrity sha512-i1okWYkA4FJICtr7KpYzFpRTHgy5jdDbZiWfvny21iIKky5YExiDXP+zbXzm3dUcFpkEeYNHgQ5fuG236JPq0g== + +"@rollup/rollup-win32-ia32-msvc@4.60.1": + version "4.60.1" + resolved "https://registry.yarnpkg.com/@rollup/rollup-win32-ia32-msvc/-/rollup-win32-ia32-msvc-4.60.1.tgz#c098e45338c50f22f1b288476354f025b746285b" + integrity sha512-u09m3CuwLzShA0EYKMNiFgcjjzwqtUMLmuCJLeZWjjOYA3IT2Di09KaxGBTP9xVztWyIWjVdsB2E9goMjZvTQg== + +"@rollup/rollup-win32-x64-gnu@4.60.1": + version "4.60.1" + resolved "https://registry.yarnpkg.com/@rollup/rollup-win32-x64-gnu/-/rollup-win32-x64-gnu-4.60.1.tgz#2c9e15be155b79d05999953b1737b2903842e903" + integrity sha512-k+600V9Zl1CM7eZxJgMyTUzmrmhB/0XZnF4pRypKAlAgxmedUA+1v9R+XOFv56W4SlHEzfeMtzujLJD22Uz5zg== + +"@rollup/rollup-win32-x64-msvc@4.60.1": + version "4.60.1" + resolved "https://registry.yarnpkg.com/@rollup/rollup-win32-x64-msvc/-/rollup-win32-x64-msvc-4.60.1.tgz#23b860113e9f87eea015d1fa3a4240a52b42fcd4" + integrity sha512-lWMnixq/QzxyhTV6NjQJ4SFo1J6PvOX8vUx5Wb4bBPsEb+8xZ89Bz6kOXpfXj9ak9AHTQVQzlgzBEc1SyM27xQ== "@types/chai@^5.2.2": version "5.2.3" @@ -1211,37 +1211,37 @@ reusify@^1.0.4: integrity sha512-g6QUff04oZpHs0eG5p83rFLhHeV00ug/Yf9nZM6fLeUrPguBTkTQOdpAWWspMh55TZfVQDPaN3NQJfbVRAxdIw== rollup@>=4.59.0, rollup@^4.43.0: - version "4.59.0" - resolved "https://registry.yarnpkg.com/rollup/-/rollup-4.59.0.tgz#cf74edac17c1486f562d728a4d923a694abdf06f" - integrity sha512-2oMpl67a3zCH9H79LeMcbDhXW/UmWG/y2zuqnF2jQq5uq9TbM9TVyXvA4+t+ne2IIkBdrLpAaRQAvo7YI/Yyeg== + version "4.60.1" + resolved "https://registry.yarnpkg.com/rollup/-/rollup-4.60.1.tgz#b4aa2bcb3a5e1437b5fad40d43fe42d4bde7a42d" + integrity sha512-VmtB2rFU/GroZ4oL8+ZqXgSA38O6GR8KSIvWmEFv63pQ0G6KaBH9s07PO8XTXP4vI+3UJUEypOfjkGfmSBBR0w== dependencies: "@types/estree" "1.0.8" optionalDependencies: - "@rollup/rollup-android-arm-eabi" "4.59.0" - "@rollup/rollup-android-arm64" "4.59.0" - "@rollup/rollup-darwin-arm64" "4.59.0" - "@rollup/rollup-darwin-x64" "4.59.0" - "@rollup/rollup-freebsd-arm64" "4.59.0" - "@rollup/rollup-freebsd-x64" "4.59.0" - "@rollup/rollup-linux-arm-gnueabihf" "4.59.0" - "@rollup/rollup-linux-arm-musleabihf" "4.59.0" - "@rollup/rollup-linux-arm64-gnu" "4.59.0" - "@rollup/rollup-linux-arm64-musl" "4.59.0" - "@rollup/rollup-linux-loong64-gnu" "4.59.0" - "@rollup/rollup-linux-loong64-musl" "4.59.0" - "@rollup/rollup-linux-ppc64-gnu" "4.59.0" - "@rollup/rollup-linux-ppc64-musl" "4.59.0" - "@rollup/rollup-linux-riscv64-gnu" "4.59.0" - "@rollup/rollup-linux-riscv64-musl" "4.59.0" - "@rollup/rollup-linux-s390x-gnu" "4.59.0" - "@rollup/rollup-linux-x64-gnu" "4.59.0" - "@rollup/rollup-linux-x64-musl" "4.59.0" - "@rollup/rollup-openbsd-x64" "4.59.0" - "@rollup/rollup-openharmony-arm64" "4.59.0" - "@rollup/rollup-win32-arm64-msvc" "4.59.0" - "@rollup/rollup-win32-ia32-msvc" "4.59.0" - "@rollup/rollup-win32-x64-gnu" "4.59.0" - "@rollup/rollup-win32-x64-msvc" "4.59.0" + "@rollup/rollup-android-arm-eabi" "4.60.1" + "@rollup/rollup-android-arm64" "4.60.1" + "@rollup/rollup-darwin-arm64" "4.60.1" + "@rollup/rollup-darwin-x64" "4.60.1" + "@rollup/rollup-freebsd-arm64" "4.60.1" + "@rollup/rollup-freebsd-x64" "4.60.1" + "@rollup/rollup-linux-arm-gnueabihf" "4.60.1" + "@rollup/rollup-linux-arm-musleabihf" "4.60.1" + "@rollup/rollup-linux-arm64-gnu" "4.60.1" + "@rollup/rollup-linux-arm64-musl" "4.60.1" + "@rollup/rollup-linux-loong64-gnu" "4.60.1" + "@rollup/rollup-linux-loong64-musl" "4.60.1" + "@rollup/rollup-linux-ppc64-gnu" "4.60.1" + "@rollup/rollup-linux-ppc64-musl" "4.60.1" + "@rollup/rollup-linux-riscv64-gnu" "4.60.1" + "@rollup/rollup-linux-riscv64-musl" "4.60.1" + "@rollup/rollup-linux-s390x-gnu" "4.60.1" + "@rollup/rollup-linux-x64-gnu" "4.60.1" + "@rollup/rollup-linux-x64-musl" "4.60.1" + "@rollup/rollup-openbsd-x64" "4.60.1" + "@rollup/rollup-openharmony-arm64" "4.60.1" + "@rollup/rollup-win32-arm64-msvc" "4.60.1" + "@rollup/rollup-win32-ia32-msvc" "4.60.1" + "@rollup/rollup-win32-x64-gnu" "4.60.1" + "@rollup/rollup-win32-x64-msvc" "4.60.1" fsevents "~2.3.2" run-parallel@^1.1.9: @@ -1257,9 +1257,9 @@ semver@^7.6.0: integrity sha512-RF0Fw+rO5AMf9MAyaRXI4AV0Ulj5lMHqVxxdSgiVbixSCXoEmmX/jk0CuJw4+3SqroYO9VoUh+HcuJivvtJemA== semver@^7.6.3: - version "7.7.4" - resolved "https://registry.yarnpkg.com/semver/-/semver-7.7.4.tgz#28464e36060e991fa7a11d0279d2d3f3b57a7e8a" - integrity sha512-vFKC2IEtQnVhpT78h1Yp8wzwrf8CM+MzKMHGJZfBtzhZNycRFnXsHk6E5TxIkkMsgNS7mdX3AGB7x2QM2di4lA== + version "7.7.3" + resolved "https://registry.yarnpkg.com/semver/-/semver-7.7.3.tgz#4b5f4143d007633a8dc671cd0a6ef9147b8bb946" + integrity sha512-SdsKMrI9TdgjdweUSR9MweHA4EJ8YxHn8DFaDisvhVlUOe4BF1tLD7GAj0lIqWVl+dPb/rExr0Btby5loQm20Q== shebang-command@^2.0.0: version "2.0.0" diff --git a/crates/rust-client/src/note/import.rs b/crates/rust-client/src/note/import.rs index f8aa51e4bb..31caede98f 100644 --- a/crates/rust-client/src/note/import.rs +++ b/crates/rust-client/src/note/import.rs @@ -34,7 +34,7 @@ use crate::{Client, ClientError}; /// Note importing methods. impl Client where - AUTH: TransactionAuthenticator + Sync + 'static, + AUTH: TransactionAuthenticator + Send + Sync + 'static, { // INPUT NOTE CREATION // -------------------------------------------------------------------------------------------- diff --git a/crates/rust-client/src/note/note_screener.rs b/crates/rust-client/src/note/note_screener.rs index 140d7c3fea..b1682b635b 100644 --- a/crates/rust-client/src/note/note_screener.rs +++ b/crates/rust-client/src/note/note_screener.rs @@ -187,7 +187,8 @@ impl NoteScreener { // DEFAULT CALLBACK IMPLEMENTATIONS // ================================================================================================ -#[async_trait(?Send)] +#[cfg_attr(not(target_arch = "wasm32"), async_trait)] +#[cfg_attr(target_arch = "wasm32", async_trait(?Send))] impl OnNoteReceived for NoteScreener { /// Default implementation of the [`OnNoteReceived`] callback. It queries the store for the /// committed note to check if it's relevant. If the note wasn't being tracked but it came in diff --git a/crates/rust-client/src/note_transport/mod.rs b/crates/rust-client/src/note_transport/mod.rs index e1616a8bdf..e10a9521de 100644 --- a/crates/rust-client/src/note_transport/mod.rs +++ b/crates/rust-client/src/note_transport/mod.rs @@ -69,7 +69,7 @@ impl Client { impl Client where - AUTH: TransactionAuthenticator + Sync + 'static, + AUTH: TransactionAuthenticator + Send + Sync + 'static, { /// Fetch notes for tracked note tags. /// diff --git a/crates/rust-client/src/sync/mod.rs b/crates/rust-client/src/sync/mod.rs index 287c95de1e..69fe5aa901 100644 --- a/crates/rust-client/src/sync/mod.rs +++ b/crates/rust-client/src/sync/mod.rs @@ -31,7 +31,7 @@ //! # use miden_client::{Client, ClientError}; //! # use miden_protocol::{block::BlockHeader, Felt, Word}; //! # use miden_protocol::crypto::rand::FeltRng; -//! # async fn run_sync(client: &mut Client) -> Result<(), ClientError> { +//! # async fn run_sync(client: &mut Client) -> Result<(), ClientError> { //! // Attempt to synchronize the client's state with the Miden network. //! // The requested data is based on the client's state: it gets updates for accounts, relevant //! // notes, etc. For more information on the data that gets requested, see the doc comments for @@ -58,6 +58,7 @@ use alloc::sync::Arc; use alloc::vec::Vec; use core::cmp::max; +use core::num::NonZeroU32; use miden_protocol::account::AccountId; use miden_protocol::block::BlockNumber; @@ -77,6 +78,10 @@ pub use tag::{NoteTagRecord, NoteTagSource}; mod state_sync; pub use state_sync::{NoteUpdateAction, OnNoteReceived, StateSync, StateSyncInput}; +// Re-exported for ergonomics so callers of `sync_state_to` / `get_sync_update_to` +// don't have to import from `rpc::domain`. +pub use crate::rpc::domain::sync::SyncTarget; + mod state_sync_update; pub use state_sync_update::{ AccountUpdates, @@ -88,7 +93,7 @@ pub use state_sync_update::{ /// Client synchronization methods. impl Client where - AUTH: TransactionAuthenticator + Sync + 'static, + AUTH: TransactionAuthenticator + Send + Sync + 'static, { // SYNC STATE // -------------------------------------------------------------------------------------------- @@ -117,6 +122,10 @@ where /// state. /// 7. The MMR is updated with the new peaks and authentication nodes. /// 8. All updates are applied to the store to be persisted. + /// + /// For chunked sync (advancing the client a bounded number of blocks at a time), use + /// [`get_sync_update_by_blocks()`](Self::get_sync_update_by_blocks) followed by + /// [`apply_state_sync()`](Self::apply_state_sync). pub async fn sync_state(&mut self) -> Result { self.ensure_genesis_in_place().await?; self.ensure_rpc_limits_in_place().await?; @@ -137,7 +146,9 @@ where let input = self.build_sync_input().await?; // Get the sync update from the network - let state_sync_update = state_sync.sync_state(&mut current_partial_mmr, input).await?; + let state_sync_update = state_sync + .sync_state(&mut current_partial_mmr, input, SyncTarget::CommittedChainTip) + .await?; let sync_summary: SyncSummary = (&state_sync_update).into(); debug!(sync_summary = ?sync_summary, "Sync summary computed"); @@ -185,6 +196,72 @@ where }) } + /// Gets the state sync update from the network without applying it. + /// + /// This method performs the same network requests as [`sync_state()`](Self::sync_state) but + /// returns the raw [`StateSyncUpdate`] instead of applying it to the store. This is useful + /// when you want to inspect or modify the update before applying it, or when implementing + /// custom sync logic. + /// + /// Use [`apply_state_sync()`](Self::apply_state_sync) to apply the returned update. + /// + /// # Example + /// + /// ```rust,ignore + /// let update = client.get_sync_update().await?; + /// println!("Sync would update {} accounts", update.accounts.updated_accounts.len()); + /// client.apply_state_sync(update).await?; + /// ``` + pub async fn get_sync_update(&self) -> Result { + self.get_sync_update_to(SyncTarget::CommittedChainTip).await + } + + /// Like [`get_sync_update()`](Self::get_sync_update) but only advances at most + /// `block_amount` blocks. + pub async fn get_sync_update_by_blocks( + &self, + block_amount: NonZeroU32, + ) -> Result { + let current = self.get_sync_height().await?; + let target = BlockNumber::from(current.as_u32().saturating_add(block_amount.get())); + self.get_sync_update_to(SyncTarget::BlockNumber(target)).await + } + + /// Like [`get_sync_update()`](Self::get_sync_update) but syncs up to the given + /// [`SyncTarget`] instead of the chain tip. + pub async fn get_sync_update_to( + &self, + upper_bound: SyncTarget, + ) -> Result { + self.get_sync_update_with_screener(Arc::new(self.note_screener()), upper_bound) + .await + } + + /// Like [`get_sync_update_to()`](Self::get_sync_update_to) but uses a caller-supplied + /// [`OnNoteReceived`] implementation to decide relevance per note. + /// + /// `OnNoteReceived` is a **filter**, not a listener. It controls what gets persisted to + /// the store during sync (Commit / Insert / Discard per incoming note). The returned + /// [`StateSyncUpdate`] — and, transitively, any events derived from it — reflects only the + /// notes the screener kept. + /// + /// If you want to *observe* notes without changing persistence semantics, use the default + /// screener and subscribe to events downstream. Implement this trait only when you need + /// different persistence behavior (e.g., tracking extra public notes, ignoring specific + /// tags, or auto-accepting private notes delivered out-of-band). + pub async fn get_sync_update_with_screener( + &self, + screener: Arc, + upper_bound: SyncTarget, + ) -> Result { + let state_sync = StateSync::new(self.rpc_api.clone(), screener, self.tx_discard_delta); + + let mut current_partial_mmr = self.store.get_current_partial_mmr().await?; + let input = self.build_sync_input().await?; + + state_sync.sync_state(&mut current_partial_mmr, input, upper_bound).await + } + /// Applies the state sync update to the store and prunes the irrelevant block headers. /// /// See [`crate::Store::apply_state_sync()`] for what the update implies. @@ -214,7 +291,7 @@ where // ================================================================================================ /// Contains stats about the sync operation. -#[derive(Debug, PartialEq)] +#[derive(Debug, PartialEq, Clone)] pub struct SyncSummary { /// Block number up to which the client has been synced. pub block_num: BlockNumber, diff --git a/crates/rust-client/src/sync/state_sync.rs b/crates/rust-client/src/sync/state_sync.rs index 8164f17952..2a6a2b9c7f 100644 --- a/crates/rust-client/src/sync/state_sync.rs +++ b/crates/rust-client/src/sync/state_sync.rs @@ -94,8 +94,9 @@ pub enum NoteUpdateAction { Discard, } -#[async_trait(?Send)] -pub trait OnNoteReceived { +#[cfg_attr(not(target_arch = "wasm32"), async_trait)] +#[cfg_attr(target_arch = "wasm32", async_trait(?Send))] +pub trait OnNoteReceived: Send + Sync { /// Callback that gets executed when a new note is received as part of the sync response. /// /// It receives: @@ -198,6 +199,7 @@ impl StateSync { &self, current_partial_mmr: &mut PartialMmr, input: StateSyncInput, + upper_bound: SyncTarget, ) -> Result { let StateSyncInput { accounts, @@ -220,7 +222,7 @@ impl StateSync { let note_tags = Arc::new(note_tags); let account_ids: Vec = accounts.iter().map(AccountHeader::id).collect(); let Some(mut sync_data) = self - .fetch_sync_data(state_sync_update.block_num, &account_ids, ¬e_tags) + .fetch_sync_data(state_sync_update.block_num, &account_ids, ¬e_tags, upper_bound) .await? else { // No progress — already at the tip. @@ -286,17 +288,17 @@ impl StateSync { current_block_num: BlockNumber, account_ids: &[AccountId], note_tags: &Arc>, + upper_bound: SyncTarget, ) -> Result, ClientError> { - // Step 1: Fetch the MMR delta and chain tip header. - let chain_mmr_info = self - .rpc_api - .sync_chain_mmr(current_block_num, SyncTarget::CommittedChainTip) - .await?; + // Step 1: Fetch the MMR delta and the header at the requested upper bound. + // The node clamps `SyncTarget::BlockNumber(n)` to the chain tip if `n` is + // past it, so the response's `block_to` is always a valid block num. + let chain_mmr_info = self.rpc_api.sync_chain_mmr(current_block_num, upper_bound).await?; let chain_tip = chain_mmr_info.block_to; - // No progress — already at the tip. + // No progress — already at the requested upper bound. if chain_tip == current_block_num { - info!(block_num = %current_block_num, "Already at chain tip, nothing to sync."); + info!(block_num = %current_block_num, "Already at sync upper bound, nothing to sync."); return Ok(None); } @@ -778,7 +780,7 @@ mod tests { /// Mock note screener that discards all notes, for minimal test setup. struct MockScreener; - #[async_trait(?Send)] + #[async_trait] impl OnNoteReceived for MockScreener { async fn on_note_received( &self, @@ -947,7 +949,8 @@ mod tests { /// This ensures committed notes get their inclusion proofs set during sync. struct CommitAllScreener; - #[async_trait(?Send)] + #[cfg_attr(not(target_arch = "wasm32"), async_trait)] + #[cfg_attr(target_arch = "wasm32", async_trait(?Send))] impl OnNoteReceived for CommitAllScreener { async fn on_note_received( &self, @@ -1052,7 +1055,10 @@ mod tests { uncommitted_transactions: vec![], }; - let update = state_sync.sync_state(&mut partial_mmr, sync_input).await.unwrap(); + let update = state_sync + .sync_state(&mut partial_mmr, sync_input, SyncTarget::CommittedChainTip) + .await + .unwrap(); let updated_notes: Vec<_> = update.note_updates.updated_input_notes().collect(); @@ -1095,7 +1101,10 @@ mod tests { assert_eq!(partial_mmr.forest().num_leaves(), 1); // First sync - let update = state_sync.sync_state(&mut partial_mmr, empty()).await.unwrap(); + let update = state_sync + .sync_state(&mut partial_mmr, empty(), SyncTarget::CommittedChainTip) + .await + .unwrap(); assert_eq!(update.block_num, chain_tip_1); let forest_1 = partial_mmr.forest(); @@ -1106,7 +1115,10 @@ mod tests { mock_rpc.advance_blocks(2); let chain_tip_2 = mock_rpc.get_chain_tip_block_num(); - let update = state_sync.sync_state(&mut partial_mmr, empty()).await.unwrap(); + let update = state_sync + .sync_state(&mut partial_mmr, empty(), SyncTarget::CommittedChainTip) + .await + .unwrap(); assert_eq!(update.block_num, chain_tip_2); let forest_2 = partial_mmr.forest(); @@ -1114,7 +1126,10 @@ mod tests { assert_eq!(forest_2.num_leaves(), chain_tip_2.as_u32() as usize + 1); // Third sync (no new blocks) - let update = state_sync.sync_state(&mut partial_mmr, empty()).await.unwrap(); + let update = state_sync + .sync_state(&mut partial_mmr, empty(), SyncTarget::CommittedChainTip) + .await + .unwrap(); assert_eq!(update.block_num, chain_tip_2); assert_eq!(partial_mmr.forest(), forest_2); @@ -1235,7 +1250,12 @@ mod tests { let mut partial_mmr = PartialMmr::from_peaks(genesis_peaks); let sync_data = state_sync - .fetch_sync_data(BlockNumber::GENESIS, &[], &Arc::new(note_tags.clone())) + .fetch_sync_data( + BlockNumber::GENESIS, + &[], + &Arc::new(note_tags.clone()), + SyncTarget::CommittedChainTip, + ) .await .unwrap() .expect("should have progressed past genesis"); diff --git a/crates/rust-client/src/transaction/mod.rs b/crates/rust-client/src/transaction/mod.rs index 88b55dfa29..cfed50e7b4 100644 --- a/crates/rust-client/src/transaction/mod.rs +++ b/crates/rust-client/src/transaction/mod.rs @@ -35,7 +35,7 @@ //! /// containing 100 tokens of `faucet_id`'s fungible asset. //! async fn create_and_submit_transaction< //! R: rand::Rng, -//! AUTH: TransactionAuthenticator + Sync + 'static, +//! AUTH: TransactionAuthenticator + Send + Sync + 'static, //! >( //! client: &mut Client, //! sender_id: AccountId, @@ -157,7 +157,7 @@ pub use result::TransactionResult; /// Transaction management methods impl Client where - AUTH: TransactionAuthenticator + Sync + 'static, + AUTH: TransactionAuthenticator + Send + Sync + 'static, { // TRANSACTION DATA RETRIEVAL // -------------------------------------------------------------------------------------------- @@ -206,7 +206,7 @@ where &mut self, account_id: AccountId, transaction_request: TransactionRequest, - tx_prover: Arc, + tx_prover: Arc, ) -> Result { // Register any missing NTX scripts before the main transaction. // The registration path contains its own full execute -> prove -> submit pipeline. @@ -365,7 +365,7 @@ where pub async fn prove_transaction_with( &mut self, tx_result: &TransactionResult, - tx_prover: Arc, + tx_prover: Arc, ) -> Result { info!("Proving transaction..."); @@ -649,7 +649,7 @@ where &mut self, account_id: AccountId, scripts: &[NoteScript], - tx_prover: Arc, + tx_prover: Arc, ) -> Result<(), ClientError> { let mut missing_scripts = Vec::new(); diff --git a/crates/testing/miden-client-tests/src/tests.rs b/crates/testing/miden-client-tests/src/tests.rs index 88b3ac2bf1..790a9e79f3 100644 --- a/crates/testing/miden-client-tests/src/tests.rs +++ b/crates/testing/miden-client-tests/src/tests.rs @@ -489,7 +489,8 @@ async fn sync_persists_auth_nodes_for_skipped_blocks() { // for every sync step. This means only the chain tip will have `include_block = true`. struct DiscardAllNotes; - #[async_trait(?Send)] + #[cfg_attr(not(target_arch = "wasm32"), async_trait)] + #[cfg_attr(target_arch = "wasm32", async_trait(?Send))] impl OnNoteReceived for DiscardAllNotes { async fn on_note_received( &self, @@ -528,6 +529,7 @@ async fn sync_persists_auth_nodes_for_skipped_blocks() { output_notes: vec![], uncommitted_transactions: vec![], }, + miden_client::rpc::domain::sync::SyncTarget::CommittedChainTip, ) .await .unwrap(); @@ -564,7 +566,8 @@ async fn sync_state_no_redundant_get_account_calls() { struct DiscardAllNotes; - #[async_trait(?Send)] + #[cfg_attr(not(target_arch = "wasm32"), async_trait)] + #[cfg_attr(target_arch = "wasm32", async_trait(?Send))] impl OnNoteReceived for DiscardAllNotes { async fn on_note_received( &self, @@ -613,7 +616,14 @@ async fn sync_state_no_redundant_get_account_calls() { output_notes: vec![], uncommitted_transactions: vec![], }; - let state_sync_update = state_sync.sync_state(&mut partial_mmr, input).await.unwrap(); + let state_sync_update = state_sync + .sync_state( + &mut partial_mmr, + input, + miden_client::rpc::domain::sync::SyncTarget::CommittedChainTip, + ) + .await + .unwrap(); // Only 1 updated public account entry, not N duplicates assert_eq!( diff --git a/eslint.config.js b/eslint.config.js index a7d2630141..553ee48fc2 100644 --- a/eslint.config.js +++ b/eslint.config.js @@ -3,6 +3,7 @@ module.exports = [ // Ignore patterns ignores: [ "crates/web-client/dist/**/*", + "crates/react-sdk/dist/**/*", "target/**/*", "**/target/**/*", "miden-node/**/*",