From 83ceddb8ea80992403e572befb88123adc3bc7c1 Mon Sep 17 00:00:00 2001 From: Vaibhav Jindal Date: Tue, 14 Jul 2026 01:49:35 +0530 Subject: [PATCH] refactor(rust-client): collapse NoteObserver into OnNoteReceived Fold the standalone NoteObserver trait into OnNoteReceived, removing one abstraction. StateSync now holds a list of OnNoteReceived observers instead of a single screener plus a separate observer list. on_note_received takes the committed/public note by reference and gains the note's attachments; the trait gains defaulted name() and post-sync apply() hooks. NoteUpdateAction gains an Observe variant that marks a block relevant without storing the note; when multiple observers vote, verdicts fold by precedence (Commit > Insert > Observe > Discard). PswapChainObserver is re-expressed as an OnNoteReceived. No behavioral change to PSWAP lineage tracking. Also extends pswap_multi_round_chain_tracking_test with a private-note case (registering the actual minted round notes), confirming private multi-round lineage tracking works end-to-end. Attachments are passed as a separate parameter for now; moving them onto CommittedNote is deferred to a follow-up (https://github.com/0xMiden/rust-sdk/issues/2118). --- CHANGELOG.md | 1 + crates/rust-client/src/errors.rs | 2 +- crates/rust-client/src/note/note_screener.rs | 13 +- crates/rust-client/src/pswap/observer.rs | 31 +-- crates/rust-client/src/sync/mod.rs | 5 +- crates/rust-client/src/sync/note_observer.rs | 36 ---- crates/rust-client/src/sync/state_sync.rs | 186 ++++++++++++------ .../rust-client/src/transaction/observer.rs | 2 +- .../testing/miden-client-tests/src/tests.rs | 171 +++++++++++++++- 9 files changed, 316 insertions(+), 131 deletions(-) delete mode 100644 crates/rust-client/src/sync/note_observer.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 46b0c9ad0b..3c971c94f0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,7 @@ * [BREAKING][store] The SQLite store now stores account IDs as serialized `BLOB` columns instead of hex `TEXT` ([#2309](https://github.com/0xMiden/rust-sdk/pull/2309)). * [BREAKING][param][store] `Store::insert_block_header` now takes a `nodes` argument and persists the header with its MMR authentication nodes in a single transaction; the standalone `Store::insert_partial_blockchain_nodes` is removed. Header-only inserts (e.g. genesis) pass an empty slice ([#2294](https://github.com/0xMiden/rust-sdk/pull/2294)). * [BREAKING][behavior][store] The `ConsumedExternal` note-metadata layout added in [#2308](https://github.com/0xMiden/rust-sdk/pull/2308) is now the only supported serialized format. The backward-compatible decoding of the older metadata-less layout is removed, so existing stores are not compatible and must be recreated ([#2313](https://github.com/0xMiden/rust-sdk/pull/2313)). +* [BREAKING][rust] The `NoteObserver` trait is removed and folded into `OnNoteReceived`: `on_note_received` now takes `committed_note` and `public_note` by reference plus a new `attachments: Option<&NoteAttachments>` argument, and the trait gains defaulted `name()` and post-sync `apply()` methods. `StateSync` now holds a list of `OnNoteReceived` observers (`StateSync::with_note_observer` now takes `Arc`), and `NoteUpdateAction` gains an `Observe` variant that marks a block relevant without storing the note; when multiple observers vote on a note the verdicts fold by precedence (`Commit` > `Insert` > `Observe` > `Discard`) ([#2279](https://github.com/0xMiden/rust-sdk/issues/2279)). ### Fixes diff --git a/crates/rust-client/src/errors.rs b/crates/rust-client/src/errors.rs index 9866ccac9a..defe02fc06 100644 --- a/crates/rust-client/src/errors.rs +++ b/crates/rust-client/src/errors.rs @@ -209,7 +209,7 @@ pub enum ClientError { /// Logs a non-fatal observer failure without propagating it, so one observer /// can't abort the others or the surrounding sync/transaction step. Shared by -/// the `NoteObserver` and `TransactionObserver` fan-out loops. +/// the `OnNoteReceived` and `TransactionObserver` fan-out loops. pub(crate) fn log_observer_failure( observer: &'static str, op: &str, diff --git a/crates/rust-client/src/note/note_screener.rs b/crates/rust-client/src/note/note_screener.rs index 9e8b2dd104..40d5c749d9 100644 --- a/crates/rust-client/src/note/note_screener.rs +++ b/crates/rust-client/src/note/note_screener.rs @@ -5,7 +5,7 @@ use alloc::vec::Vec; use async_trait::async_trait; use miden_protocol::account::{AccountCode, AccountId}; -use miden_protocol::note::{Note, NoteId}; +use miden_protocol::note::{Note, NoteAttachments, NoteId}; use miden_standards::note::NoteConsumptionStatus; use miden_tx::{ NoteCheckerError, @@ -195,8 +195,9 @@ impl OnNoteReceived for NoteScreener { /// to check its relevance. async fn on_note_received( &self, - committed_note: CommittedNote, - public_note: Option, + committed_note: &CommittedNote, + public_note: Option<&InputNoteRecord>, + _attachments: Option<&NoteAttachments>, ) -> Result { let note_id = *committed_note.note_id(); @@ -224,7 +225,7 @@ impl OnNoteReceived for NoteScreener { if input_note_present || output_note_present { // The note is being tracked by the client so it is relevant - return Ok(NoteUpdateAction::Commit(committed_note)); + return Ok(NoteUpdateAction::Commit(committed_note.clone())); } match public_note { @@ -233,7 +234,7 @@ impl OnNoteReceived for NoteScreener { if let Some(metadata) = public_note.metadata() && self.store.get_unique_note_tags().await?.contains(&metadata.tag()) { - return Ok(NoteUpdateAction::Insert(public_note)); + return Ok(NoteUpdateAction::Insert(public_note.clone())); } // The note is not being tracked by the client and is public so we can screen it @@ -247,7 +248,7 @@ impl OnNoteReceived for NoteScreener { .await?; let is_relevant = !new_note_relevance.is_empty(); if is_relevant { - Ok(NoteUpdateAction::Insert(public_note)) + Ok(NoteUpdateAction::Insert(public_note.clone())) } else { Ok(NoteUpdateAction::Discard) } diff --git a/crates/rust-client/src/pswap/observer.rs b/crates/rust-client/src/pswap/observer.rs index 064af7fca7..cf4c48acd0 100644 --- a/crates/rust-client/src/pswap/observer.rs +++ b/crates/rust-client/src/pswap/observer.rs @@ -1,5 +1,5 @@ -//! Per-note observer that collects every PSWAP-attachment note seen -//! during sync. Lineage-scope filtering happens later, in `discovery`. +//! Per-note [`OnNoteReceived`](crate::sync::OnNoteReceived) observer that collects every +//! PSWAP-attachment note seen during sync. Lineage-scope filtering happens later, in `discovery`. use alloc::boxed::Box; use alloc::sync::Arc; @@ -15,8 +15,8 @@ use crate::ClientError; use crate::pswap::discovery::discover_pswap_rounds; use crate::pswap::lineage::ObservedPswapNote; use crate::rpc::domain::note::CommittedNote; -use crate::store::Store; -use crate::sync::NoteObserver; +use crate::store::{InputNoteRecord, Store}; +use crate::sync::{NoteUpdateAction, OnNoteReceived}; use crate::utils::RwLock; // PSWAP CHAIN OBSERVER @@ -24,15 +24,15 @@ use crate::utils::RwLock; /// Per-sync collector of PSWAP-attachment notes seen this sync. /// -/// - `observe()` runs per-note during sync: reads the PSWAP attachment word straight off the note's -/// resolved attachments (carried inline on the sync window) and records a `ObservedPswapNote`. No -/// RPC round trip, no DB write. +/// - `on_note_received()` runs per-note during sync: reads the PSWAP attachment word straight off +/// the note's resolved attachments (carried inline on the sync window), records an +/// `ObservedPswapNote`, and votes [`NoteUpdateAction::Observe`] so the block is retained. /// - `apply()` runs once post-sync: drains the collector, runs the correlator, applies round /// updates. pub struct PswapChainObserver { store: Arc, - /// `observe()` writes, `apply()` drains; never concurrent. The observer is - /// shared via the outer `Arc` and only ever touched + /// `on_note_received()` writes, `apply()` drains; never concurrent. The observer is + /// shared via the outer `Arc` and only ever touched /// through `&self`, so the `RwLock` alone provides the needed interior /// mutability — no inner `Arc`. chain_note_updates: RwLock>, @@ -48,23 +48,24 @@ impl PswapChainObserver { } #[async_trait(?Send)] -impl NoteObserver for PswapChainObserver { +impl OnNoteReceived for PswapChainObserver { fn name(&self) -> &'static str { "PswapChainObserver" } - async fn observe( + async fn on_note_received( &self, committed_note: &CommittedNote, + _public_note: Option<&InputNoteRecord>, attachments: Option<&NoteAttachments>, - ) -> Result { + ) -> Result { // Notes without a PSWAP attachment are the common case; `extract_pswap_attachment` // fast-rejects them. Foreign-order filtering happens later in `discovery`. let Some(attachments) = attachments else { - return Ok(false); + return Ok(NoteUpdateAction::Discard); }; let Some(attachment) = extract_pswap_attachment(attachments) else { - return Ok(false); + return Ok(NoteUpdateAction::Discard); }; let inclusion_proof = committed_note.inclusion_proof().clone(); @@ -76,7 +77,7 @@ impl NoteObserver for PswapChainObserver { block_num: inclusion_proof.location().block_num(), inclusion_proof, }); - Ok(true) + Ok(NoteUpdateAction::Observe) } /// Drains the collector, runs the correlator, applies round updates. diff --git a/crates/rust-client/src/sync/mod.rs b/crates/rust-client/src/sync/mod.rs index cce42a14b0..92f7c665ed 100644 --- a/crates/rust-client/src/sync/mod.rs +++ b/crates/rust-client/src/sync/mod.rs @@ -77,9 +77,6 @@ mod block_header; mod tag; pub use tag::{NoteTagRecord, NoteTagSource}; -mod note_observer; -pub use note_observer::NoteObserver; - mod state_sync; pub use state_sync::{NoteUpdateAction, OnNoteReceived, StateSync, StateSyncInput}; @@ -120,7 +117,7 @@ where self.ensure_genesis_in_place().await?; self.ensure_rpc_limits_in_place().await?; - // Each `NoteObserver` owns its own per-sync state; `with_note_observer` just attaches. + // Each observer owns its own per-sync state; `with_note_observer` just attaches. let note_screener = self.note_screener(); let state_sync = StateSync::new(self.rpc_api.clone(), Arc::new(note_screener), self.tx_discard_delta) diff --git a/crates/rust-client/src/sync/note_observer.rs b/crates/rust-client/src/sync/note_observer.rs deleted file mode 100644 index 647f02ca9d..0000000000 --- a/crates/rust-client/src/sync/note_observer.rs +++ /dev/null @@ -1,36 +0,0 @@ -//! Side-effect-only observer trait for per-note arrivals during sync. - -use alloc::boxed::Box; - -use async_trait::async_trait; -use miden_protocol::note::NoteAttachments; - -use crate::ClientError; -use crate::rpc::domain::note::CommittedNote; -use crate::sync::StateSyncUpdate; - -/// Per-note + post-sync side-channel into [`crate::sync::StateSync`]. -/// Attach via `StateSync::with_note_observer(...)`. Multiple observers -/// run independently; errors are logged, never abort sync. -#[async_trait(?Send)] -pub trait NoteObserver { - /// Identifier surfaced on `tracing::warn!` events for this observer. - fn name(&self) -> &'static str; - - /// Per-note hook. Runs before the screener verdict. `attachments` is the note's resolved - /// attachment content for this sync window (`None` if absent). - /// - /// Returns `true` to mark the enclosing block as relevant even if the screener discards it, - /// so sync persists its header. - async fn observe( - &self, - committed_note: &CommittedNote, - attachments: Option<&NoteAttachments>, - ) -> Result; - - /// Post-sync hook, invoked once after the sync window closes. - /// Default impl is a no-op for observers that only need `observe()`. - async fn apply(&self, _sync_update: &StateSyncUpdate) -> Result<(), ClientError> { - Ok(()) - } -} diff --git a/crates/rust-client/src/sync/state_sync.rs b/crates/rust-client/src/sync/state_sync.rs index f0bfae3267..2bec70d24f 100644 --- a/crates/rust-client/src/sync/state_sync.rs +++ b/crates/rust-client/src/sync/state_sync.rs @@ -16,7 +16,6 @@ use tracing::info; use super::state_sync_update::TransactionUpdateTracker; use super::{ AccountUpdates, - NoteObserver, PartialBlockchainUpdates, PublicAccountDelta, PublicAccountUpdate, @@ -101,7 +100,11 @@ pub struct StateSyncInput { // SYNC CALLBACKS // ================================================================================================ -/// The action to be taken when a note update is received as part of the sync response. +/// The action a note observer votes for when a note inclusion is received during sync. +/// +/// When several observers are attached, their verdicts are folded by precedence +/// (`Commit` > `Insert` > `Observe` > `Discard`); the first observer at the highest precedence +/// supplies the payload. #[allow(clippy::large_enum_variant)] pub enum NoteUpdateAction { /// The note commit update is relevant and the specified note should be marked as committed in @@ -109,12 +112,37 @@ pub enum NoteUpdateAction { Commit(CommittedNote), /// The public note is relevant and should be inserted into the store. Insert(InputNoteRecord), + /// The note is not stored, but its enclosing block is relevant, so its header is persisted. + /// Used by side-channel observers (e.g. PSWAP tracking) that need the block, not the note. + Observe, /// The note update is not relevant and should be discarded. Discard, } +impl NoteUpdateAction { + /// Fold precedence when multiple observers vote on the same note: higher wins. + /// `Commit` > `Insert` > `Observe` > `Discard`. + fn precedence(&self) -> u8 { + match self { + NoteUpdateAction::Commit(_) => 3, + NoteUpdateAction::Insert(_) => 2, + NoteUpdateAction::Observe => 1, + NoteUpdateAction::Discard => 0, + } + } +} + +/// A per-note sync observer. Multiple observers can be attached to [`StateSync`]; each is invoked +/// for every note inclusion received during sync, and once more post-sync via [`Self::apply`]. #[async_trait(?Send)] pub trait OnNoteReceived { + /// Identifier used to tag this observer in `tracing::warn!` events when its post-sync `apply` + /// fails. Observers with a fallible `apply` should override this; the default suits + /// screening-only observers whose `apply` never fails and so are never named in a log. + fn name(&self) -> &'static str { + "OnNoteReceived" + } + /// Callback that gets executed when a new note is received as part of the sync response. /// /// It receives: @@ -122,14 +150,23 @@ pub trait OnNoteReceived { /// - The committed note received from the network. /// - An optional note record that corresponds to the state of the note in the network (only if /// the note is public). + /// - The note's resolved attachment content for this sync window (`None` if absent). /// - /// It returns an enum indicating the action to be taken for the received note update. Whether - /// the note updated should be committed, new public note inserted, or ignored. + /// It returns the [`NoteUpdateAction`] this observer votes for: commit the note, insert a new + /// public note, mark the block relevant without storing the note + /// ([`NoteUpdateAction::Observe`]), or discard. async fn on_note_received( &self, - committed_note: CommittedNote, - public_note: Option, + committed_note: &CommittedNote, + public_note: Option<&InputNoteRecord>, + attachments: Option<&NoteAttachments>, ) -> Result; + + /// Post-sync hook, invoked once after the sync window closes and before the update is + /// persisted. Default is a no-op, so screening-only observers need not implement it. + async fn apply(&self, _sync_update: &StateSyncUpdate) -> Result<(), ClientError> { + Ok(()) + } } // STATE SYNC // ================================================================================================ @@ -141,12 +178,11 @@ pub trait OnNoteReceived { pub struct StateSync { /// The RPC client used to communicate with the node. rpc_api: Arc, - /// Responsible for checking the relevance of notes and executing the - /// [`OnNoteReceived`] callback when a new note inclusion is received. - note_screener: Arc, - /// Per-note observers (see [`NoteObserver`]), invoked *before* the - /// screener verdict in `note_state_sync`. Empty by default. - note_observers: Vec>, + /// Note observers invoked for every note inclusion in `note_state_sync` and once post-sync in + /// [`Self::run_apply_hooks`]. The first entry is the screener passed to [`Self::new`]; more + /// are appended via [`Self::with_note_observer`]. Per-note verdicts are folded by + /// precedence. + note_observers: Vec>, /// Number of blocks after which pending transactions are considered stale and discarded. /// If `None`, there is no limit and transactions will be kept indefinitely. tx_discard_delta: Option, @@ -173,18 +209,17 @@ impl StateSync { ) -> Self { Self { rpc_api, - note_screener, - note_observers: Vec::new(), + note_observers: alloc::vec![note_screener], tx_discard_delta, sync_nullifiers: true, } } - /// Attaches a [`NoteObserver`] to this sync component. Observers run - /// in attachment order *before* the screener verdict; failures are - /// logged (tagged with [`NoteObserver::name`]) and never abort sync. + /// Appends a note observer. Handlers run in attachment order; their per-note verdicts are + /// folded by precedence, and their post-sync `apply` failures are logged (tagged with + /// [`OnNoteReceived::name`]) and never abort sync. #[must_use] - pub fn with_note_observer(mut self, observer: Arc) -> Self { + pub fn with_note_observer(mut self, observer: Arc) -> Self { self.note_observers.push(observer); self } @@ -206,9 +241,9 @@ impl StateSync { /// Runs each attached observer's `apply()` hook against `state_sync_update`. /// Called by the orchestrator after [`Self::sync_state`] returns but /// before the caller persists the sync update. Per-observer failures are - /// logged (tagged with the observer's [`NoteObserver::name`]) and never - /// abort the rest of the pass — symmetric with the per-note `observe()` - /// dispatcher. + /// logged (tagged with the observer's [`OnNoteReceived::name`]) and never + /// abort the rest of the pass. Screening-only observers inherit the default + /// no-op `apply`, so they never surface here. pub(crate) async fn run_apply_hooks( &self, state_sync_update: &StateSyncUpdate, @@ -216,7 +251,7 @@ impl StateSync { for observer in &self.note_observers { crate::errors::log_observer_failure( observer.name(), - "NoteObserver::apply", + "OnNoteReceived::apply", observer.apply(state_sync_update).await, ); } @@ -234,7 +269,7 @@ impl StateSync { /// 1. Fetch sync data from the node (MMR delta, note inclusions, transactions). /// 2. Update account states (fetch updated public accounts, flag mismatched private ones). /// 3. Advance the partial MMR to the chain tip. - /// 4. Screen note inclusions via the configured [`OnNoteReceived`] callback and track relevant + /// 4. Screen note inclusions via the configured [`OnNoteReceived`] observers and track relevant /// blocks in the MMR. /// 5. Process transaction inclusions (commit local txs, record external consumers, discard /// stale/expired txs, commit output notes). @@ -1012,42 +1047,37 @@ impl StateSync { for (_, committed_note) in note_inclusions { let public_note = (committed_note.note_type() != NoteType::Private) .then(|| public_notes.get(committed_note.note_id())) - .flatten() - .cloned(); - - // Observers run BEFORE the screener: they are a side-effect - // channel independent of the Commit/Insert/Discard decision, - // and a failing screener must not rob them of the note. Clone - // is skipped when no observers are attached (the common case). - if !self.note_observers.is_empty() { - // Resolve attachment content for the note from the sync window: public note - // bodies carry their attachments on the cached `InputNoteRecord`; private-note - // attachments arrive in their own side-table. Both are keyed by note ID. - let note_attachments = if committed_note.note_type() == NoteType::Private { - private_attachments.get(committed_note.note_id()) - } else { - public_note.as_ref().map(InputNoteRecord::attachments) - }; - for obs in &self.note_observers { - match obs.observe(&committed_note, note_attachments).await { - Ok(true) => found_relevant_note = true, - Ok(false) => {}, - Err(err) => { - tracing::warn!( - observer = obs.name(), - error = ?err, - "note observer failed; sync continues", - ); - }, - } + .flatten(); + + // Resolve attachment content for the note from the sync window: public note bodies + // carry their attachments on the cached `InputNoteRecord`; private-note attachments + // arrive in their own side-table. Both are keyed by note ID. + let note_attachments = if committed_note.note_type() == NoteType::Private { + private_attachments.get(committed_note.note_id()) + } else { + public_note.map(InputNoteRecord::attachments) + }; + + // Each observer votes on the note; keep the highest-precedence action + // (Commit > Insert > Observe > Discard). The first observer at that precedence supplies + // the payload. An error aborts the sync — a failed screen must not be silently treated + // as a discard, which could drop a relevant note. + let mut selected_action = NoteUpdateAction::Discard; + for observer in &self.note_observers { + let action = observer + .on_note_received(&committed_note, public_note, note_attachments) + .await?; + if action.precedence() > selected_action.precedence() { + selected_action = action; } } - match self.note_screener.on_note_received(committed_note, public_note).await? { + match selected_action { NoteUpdateAction::Commit(committed_note) => { // Only mark the downloaded block header as relevant if we are talking about // an input note (output notes get marked as committed but we don't need the - // block for anything there) + // block for anything there). Attachments here are private-only, matching the + // committed-note record. let attachments = private_attachments.get(committed_note.note_id()); found_relevant_note |= note_updates.apply_committed_note_state_transitions( &committed_note, @@ -1060,6 +1090,9 @@ impl StateSync { note_updates.apply_new_public_note(public_note, block_header)?; }, + NoteUpdateAction::Observe => { + found_relevant_note = true; + }, NoteUpdateAction::Discard => {}, } } @@ -1323,11 +1356,47 @@ mod tests { impl OnNoteReceived for MockScreener { async fn on_note_received( &self, - _committed_note: CommittedNote, - _public_note: Option, + _committed_note: &CommittedNote, + _public_note: Option<&InputNoteRecord>, + _attachments: Option<&NoteAttachments>, + ) -> Result { + Ok(NoteUpdateAction::Discard) + } + } + + /// Observer whose post-sync `apply` always errors. `on_note_received` is a no-op discard; only + /// `apply` is exercised here (via `run_apply_hooks`). + struct ErroringApply; + + #[async_trait(?Send)] + impl OnNoteReceived for ErroringApply { + async fn on_note_received( + &self, + _committed_note: &CommittedNote, + _public_note: Option<&InputNoteRecord>, + _attachments: Option<&NoteAttachments>, ) -> Result { Ok(NoteUpdateAction::Discard) } + + async fn apply(&self, _sync_update: &StateSyncUpdate) -> Result<(), ClientError> { + Err(ClientError::ChainValidationError("apply failure".into())) + } + } + + /// Regression (#2279): an observer whose post-sync `apply` errors is swallowed by + /// `run_apply_hooks` (logged, not propagated), so one observer's failure cannot abort the sync. + /// Pins the current behavior; changing it to abort should be a deliberate, test-visible change. + #[tokio::test] + async fn apply_hook_error_is_swallowed() { + let mut builder = MockChainBuilder::new(); + let _account = builder.add_existing_mock_account(miden_testing::Auth::IncrNonce).unwrap(); + let rpc_api = MockRpcApi::new(builder.build().unwrap()); + let state_sync = StateSync::new(Arc::new(rpc_api), Arc::new(ErroringApply), None); + + let result = state_sync.run_apply_hooks(&StateSyncUpdate::default()).await; + + assert!(result.is_ok(), "run_apply_hooks must swallow an observer's apply() error"); } fn empty() -> StateSyncInput { @@ -1842,10 +1911,11 @@ mod tests { impl OnNoteReceived for CommitAllScreener { async fn on_note_received( &self, - committed_note: CommittedNote, - _public_note: Option, + committed_note: &CommittedNote, + _public_note: Option<&InputNoteRecord>, + _attachments: Option<&NoteAttachments>, ) -> Result { - Ok(NoteUpdateAction::Commit(committed_note)) + Ok(NoteUpdateAction::Commit(committed_note.clone())) } } diff --git a/crates/rust-client/src/transaction/observer.rs b/crates/rust-client/src/transaction/observer.rs index 4eb3255cd5..bebaee64fa 100644 --- a/crates/rust-client/src/transaction/observer.rs +++ b/crates/rust-client/src/transaction/observer.rs @@ -1,6 +1,6 @@ //! Side-effect-only observer trait for committed transactions. //! -//! Analogous to [`crate::sync::NoteObserver`] but scoped to +//! Analogous to [`crate::sync::OnNoteReceived`] but scoped to //! `Client::apply_transaction`. Lets feature subsystems (e.g. PSWAP //! chain tracking) hook into the post-apply pipeline without //! `apply_transaction` knowing about them by name. diff --git a/crates/testing/miden-client-tests/src/tests.rs b/crates/testing/miden-client-tests/src/tests.rs index 6d3e8cca6d..50745678b4 100644 --- a/crates/testing/miden-client-tests/src/tests.rs +++ b/crates/testing/miden-client-tests/src/tests.rs @@ -616,8 +616,9 @@ async fn sync_persists_auth_nodes_for_skipped_blocks() { impl OnNoteReceived for DiscardAllNotes { async fn on_note_received( &self, - _committed_note: CommittedNote, - _public_note: Option, + _committed_note: &CommittedNote, + _public_note: Option<&InputNoteRecord>, + _attachments: Option<&NoteAttachments>, ) -> Result { Ok(NoteUpdateAction::Discard) } @@ -679,6 +680,128 @@ async fn sync_persists_auth_nodes_for_skipped_blocks() { ); } +/// Regression (#2279): an observer that votes `Observe` marks note-bearing blocks relevant even +/// when the screener discards their notes. Baseline (`sync_persists_auth_nodes_for_skipped_blocks`) +/// stores only the chain tip; adding a second observer retains blocks 1 and 4 as well. +#[tokio::test] +async fn sync_observe_retains_blocks_a_discarding_screener_would_skip() { + use miden_client::async_trait; + use miden_client::rpc::domain::note::CommittedNote; + use miden_client::store::InputNoteRecord; + use miden_client::sync::{NoteUpdateAction, OnNoteReceived, StateSync, StateSyncInput}; + use miden_protocol::crypto::merkle::mmr::{Forest, MmrPeaks, PartialMmr}; + + // Screener discards everything; a second observer votes `Observe` on every note, which outranks + // `Discard` in the precedence fold, so note-bearing blocks are still marked relevant. + struct DiscardAllNotes; + #[async_trait(?Send)] + impl OnNoteReceived for DiscardAllNotes { + async fn on_note_received( + &self, + _committed_note: &CommittedNote, + _public_note: Option<&InputNoteRecord>, + _attachments: Option<&NoteAttachments>, + ) -> Result { + Ok(NoteUpdateAction::Discard) + } + } + + struct ObserveAllNotes; + #[async_trait(?Send)] + impl OnNoteReceived for ObserveAllNotes { + async fn on_note_received( + &self, + _committed_note: &CommittedNote, + _public_note: Option<&InputNoteRecord>, + _attachments: Option<&NoteAttachments>, + ) -> Result { + Ok(NoteUpdateAction::Observe) + } + } + + let (_client, rpc_api, _) = Box::pin(create_test_client()).await; + + let genesis = rpc_api.get_block_header_by_number(Some(0.into()), false).await.unwrap().0; + let mut partial_mmr = PartialMmr::from_peaks(MmrPeaks::new(Forest::empty(), vec![]).unwrap()); + partial_mmr.add(genesis.commitment(), true).unwrap(); + + let state_sync = StateSync::new(Arc::new(rpc_api.clone()), Arc::new(DiscardAllNotes), None) + .with_note_observer(Arc::new(ObserveAllNotes)); + + let state_sync_update = state_sync + .sync_state( + &mut partial_mmr, + StateSyncInput { + accounts: vec![], + note_tags: BTreeSet::from([NoteTag::new(0)]), + input_notes: vec![], + output_notes: vec![], + uncommitted_transactions: vec![], + }, + ) + .await + .unwrap(); + + let stored_blocks: Vec = state_sync_update + .partial_blockchain_updates + .block_headers() + .map(|(header, ..)| header.block_num().as_usize()) + .collect(); + + assert!( + stored_blocks.contains(&1) && stored_blocks.contains(&4), + "Observe must retain note-bearing blocks 1 and 4 that the screener discarded; got {stored_blocks:?}" + ); +} + +/// Regression (#2279): an error returned from an observer's `on_note_received` aborts the whole +/// sync — per-note screening errors are fatal, not silently swallowed. +#[tokio::test] +async fn sync_aborts_when_an_observer_errors() { + use miden_client::async_trait; + use miden_client::rpc::domain::note::CommittedNote; + use miden_client::store::InputNoteRecord; + use miden_client::sync::{NoteUpdateAction, OnNoteReceived, StateSync, StateSyncInput}; + use miden_protocol::crypto::merkle::mmr::{Forest, MmrPeaks, PartialMmr}; + + struct ErroringScreener; + #[async_trait(?Send)] + impl OnNoteReceived for ErroringScreener { + async fn on_note_received( + &self, + _committed_note: &CommittedNote, + _public_note: Option<&InputNoteRecord>, + _attachments: Option<&NoteAttachments>, + ) -> Result { + Err(ClientError::ChainValidationError("observer failure".into())) + } + } + + let (_client, rpc_api, _) = Box::pin(create_test_client()).await; + + let genesis = rpc_api.get_block_header_by_number(Some(0.into()), false).await.unwrap().0; + let mut partial_mmr = PartialMmr::from_peaks(MmrPeaks::new(Forest::empty(), vec![]).unwrap()); + partial_mmr.add(genesis.commitment(), true).unwrap(); + + let state_sync = StateSync::new(Arc::new(rpc_api.clone()), Arc::new(ErroringScreener), None); + + // Block 1 carries notes (tag 0), so `on_note_received` is invoked and its error propagates. + let result = state_sync + .sync_state( + &mut partial_mmr, + StateSyncInput { + accounts: vec![], + note_tags: BTreeSet::from([NoteTag::new(0)]), + input_notes: vec![], + output_notes: vec![], + uncommitted_transactions: vec![], + }, + ) + .await; + + assert!(result.is_err(), "an observer error in on_note_received must abort the sync"); +} + /// Tests that a public account modified across multiple sync steps only triggers a single /// `/GetAccount` RPC call, not one per sync step. #[tokio::test] @@ -695,8 +818,9 @@ async fn sync_state_no_redundant_get_account_calls() { impl OnNoteReceived for DiscardAllNotes { async fn on_note_received( &self, - _committed_note: CommittedNote, - _public_note: Option, + _committed_note: &CommittedNote, + _public_note: Option<&InputNoteRecord>, + _attachments: Option<&NoteAttachments>, ) -> Result { Ok(NoteUpdateAction::Discard) } @@ -3381,10 +3505,12 @@ async fn pswap_full_fill_chain_tracking_test(#[case] note_type: NoteType) { /// 1 it instead lands in Bob's store as a Committed note via the output-note screening path, which /// makes it cleanly consumable in round 2. Alice's lineage must walk depth 0 → 1 → 2 and then /// Reclaim the final tip. +#[rstest] +#[case::public_pswap(NoteType::Public)] +#[case::private_pswap(NoteType::Private)] #[allow(clippy::too_many_lines)] #[tokio::test] -async fn pswap_multi_round_chain_tracking_test() { - let note_type = NoteType::Public; +async fn pswap_multi_round_chain_tracking_test(#[case] note_type: NoteType) { let mock_rpc_api = MockRpcApi::new(Box::pin(create_prebuilt_mock_chain()).await); let (mut alice_client, alice_keystore) = create_pswap_test_client(&mock_rpc_api).await; let (mut bob_client, bob_keystore) = create_pswap_test_client(&mock_rpc_api).await; @@ -3434,10 +3560,15 @@ async fn pswap_multi_round_chain_tracking_test() { mock_rpc_api.prove_block(); alice_client.sync_state().await.unwrap(); - // Bob discovers the public order via the asset-pair tag. - let pswap_tag = PswapNote::create_tag(note_type, &offered_asset, &requested_asset); - bob_client.add_note_tag(pswap_tag).await.unwrap(); - bob_client.sync_state().await.unwrap(); + // Public: Bob discovers the order via the asset-pair tag. Private: Bob is handed `pswap_note` + // off-chain; Alice discovers Bob's paybacks/remainders via their attachments, which the mock is + // taught (after each of Bob's fills) from the actual notes he mints — a real node returns + // these. + if note_type == NoteType::Public { + let pswap_tag = PswapNote::create_tag(note_type, &offered_asset, &requested_asset); + bob_client.add_note_tag(pswap_tag).await.unwrap(); + bob_client.sync_state().await.unwrap(); + } // ── Round 1: Bob fills 25 ETH → 50 BTC payout, leaving 50 BTC / 25 ETH. ── let consume_request = TransactionRequestBuilder::new() @@ -3448,6 +3579,16 @@ async fn pswap_multi_round_chain_tracking_test() { .unwrap(); mock_rpc_api.prove_block(); bob_client.sync_state().await.unwrap(); + if note_type == NoteType::Private { + for record in bob_client.get_output_notes(NoteFilter::All).await.unwrap() { + if let Ok(note) = Note::try_from(record) + && !note.attachments().is_empty() + { + mock_rpc_api + .register_private_note_attachments(note.id(), note.attachments().clone()); + } + } + } alice_client.sync_state().await.unwrap(); let lineage = alice_client.pswap_lineage(order_id).await.unwrap().unwrap(); @@ -3479,6 +3620,16 @@ async fn pswap_multi_round_chain_tracking_test() { .unwrap(); mock_rpc_api.prove_block(); bob_client.sync_state().await.unwrap(); + if note_type == NoteType::Private { + for record in bob_client.get_output_notes(NoteFilter::All).await.unwrap() { + if let Ok(note) = Note::try_from(record) + && !note.attachments().is_empty() + { + mock_rpc_api + .register_private_note_attachments(note.id(), note.attachments().clone()); + } + } + } alice_client.sync_state().await.unwrap(); let lineage = alice_client.pswap_lineage(order_id).await.unwrap().unwrap();