diff --git a/CHANGELOG.md b/CHANGELOG.md index 873f63208b..c4272a4d18 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,16 +4,22 @@ ### Breaking Changes +* [BREAKING][behavior][rust] A Note Transport Layer failure no longer fails `Client::sync_state`. The error is logged and the chain sync still applies; the transport cursor is left where it was, so the next sync requests the same page again ([#2453](https://github.com/0xMiden/rust-sdk/pull/2453)). * [BREAKING][type][rust] Added the `TransactionRequestError::SwapNoteWithZeroAsset` variant, so exhaustive matches on `TransactionRequestError` must handle it ([#2459](https://github.com/0xMiden/rust-sdk/pull/2459)). * [BREAKING][removal][test] Loose helper functions in `miden_client::testing::common` are now methods on `TestClient`. `TestClient::keystore()` exposes the client's keystore, so `ClientConfig::into_client` and `into_unsynced_client` return just the `TestClient` instead of a client/keystore pair ([#2481](https://github.com/0xMiden/rust-sdk/pull/2481)). ### Fixes +* [FIX][rust] A private note fetched from the Note Transport Layer whose nullifier is already on chain is now imported as consumed instead of committed, so `get_consumable_notes` no longer reports notes the node will reject ([#2453](https://github.com/0xMiden/rust-sdk/pull/2453)). * [FIX][rust] Added validation of cached transaction encryption keys during deserialization. Unsupported encryption schemes and empty or oversized key IDs are rejected before reading the key ID bytes ([#2411](https://github.com/0xMiden/rust-sdk/pull/2411)). * [FIX][cli] `miden-client import` now rejects invocations without a file path instead of silently succeeding ([#2450](https://github.com/0xMiden/rust-sdk/pull/2450)). * [FIX][rust] `TransactionRequestBuilder::build_swap` and `build_pswap_create` now reject a zero-amount asset on either side of the exchange. A zero requested asset produced a payback P2ID note carrying nothing, and a zero offered asset produced a note whose consumer pays and receives nothing ([#2459](https://github.com/0xMiden/rust-sdk/pull/2459)). * [FIX][test] The integration tests run again on a chain that charges no fee. A `--funders` path (`MIDEN_FUNDER_ACCOUNTS_DIR`) that is unset, empty, missing, or holds no `.mac` file now leaves the run without funders instead of failing, which is all a fee-free genesis needs, since it declares no wallets for the path to hold. A `.mac` file that is present but unusable stays a hard error ([#2481](https://github.com/0xMiden/rust-sdk/pull/2481)). +### Enhancements + +* [rust] `Client::sync_state` fetches a Note Transport Layer page and the node's chain update concurrently, instead of running a full note transport sync before the chain sync. The transport notes are imported first and their records join the chain sync's note updates, so a note delivered and committed within the same sync is reported by that sync ([#2453](https://github.com/0xMiden/rust-sdk/pull/2453)). + ## 0.16.0 (2026-09-07) ### Breaking Changes diff --git a/crates/rust-client/src/errors.rs b/crates/rust-client/src/errors.rs index 3cb3a3c297..b6910ba060 100644 --- a/crates/rust-client/src/errors.rs +++ b/crates/rust-client/src/errors.rs @@ -244,6 +244,8 @@ pub enum ClientError { /// `From for ClientError` returning `Observer(Box::new(err))`. #[error(transparent)] Observer(Box), + #[error("expected note blocks to be screened before state sync update is built")] + UnscreenedNoteBlocks, } // OBSERVER FAN-OUT diff --git a/crates/rust-client/src/note/import.rs b/crates/rust-client/src/note/import.rs index b1241ed1df..f473c7f16c 100644 --- a/crates/rust-client/src/note/import.rs +++ b/crates/rust-client/src/note/import.rs @@ -25,7 +25,7 @@ use miden_protocol::note::{ use miden_standards::note::NoteFile; use miden_tx::auth::TransactionAuthenticator; -use crate::rpc::domain::note::{FetchedNote, SyncedNote}; +use crate::rpc::domain::note::{FetchedNote, ResolvedSyncNotesBlock}; use crate::rpc::{NoteContentFetch, RpcError}; use crate::store::input_note_states::ExpectedNoteState; use crate::store::{InputNoteRecord, InputNoteState, NoteFilter}; @@ -128,7 +128,7 @@ where previous_note, details, sync_hint.after_block_num(), - Some(sync_hint.tag()), + sync_hint.tag(), )); }, NoteFile::Committed { note, proof } => { @@ -342,47 +342,56 @@ where /// the new information changed them. async fn import_note_records_by_details( &mut self, - requested_notes: Vec<(Option, NoteDetails, BlockNumber, Option)>, + requested_notes: Vec, ) -> Result, ClientError> { let mut lowest_request_block: BlockNumber = u32::MAX.into(); let mut note_requests = vec![]; for (_, details, after_block_num, tag) in &requested_notes { - if let Some(tag) = tag { - note_requests.push((details.commitment(), *tag)); - lowest_request_block = lowest_request_block.min(*after_block_num); - } + note_requests.push((details.commitment(), *tag)); + lowest_request_block = lowest_request_block.min(*after_block_num); } - let mut committed_notes_data = - self.sync_expected_notes(lowest_request_block, note_requests).await?; + let blocks = self.sync_expected_notes(lowest_request_block, ¬e_requests).await?; - let mut note_records = vec![]; + // The blocks arrive with the notes, so a committed note needs no further block lookup. They + // are stored first, so a record is never persisted as committed before the header that + // proves its inclusion is tracked and stored. let mut partial_mmr = self.get_current_partial_mmr().await?; + self.insert_note_blocks(&blocks, &mut partial_mmr).await?; + self.cache_partial_mmr(partial_mmr).await?; + let mut note_records = vec![]; for (previous_note, details, after_block_num, tag) in requested_notes { let mut note_record = previous_note.unwrap_or_else(|| { InputNoteRecord::new( details, NoteAttachments::empty(), self.store.get_current_timestamp(), - ExpectedNoteState { metadata: None, after_block_num, tag }.into(), + ExpectedNoteState { + metadata: None, + after_block_num, + tag: Some(tag), + } + .into(), ) }); // Notes the node has not reported as committed keep their expected record untouched. - let Some(SyncedNote { - committed: committed_note, attachments, .. - }) = committed_notes_data.remove(¬e_record.details_commitment()) - else { + let commitment = note_record.details_commitment(); + let Some((sync_note, block_header)) = blocks.iter().find_map(|block| { + let sync_note = block.notes.values().find(|sync_note| { + NoteId::new(commitment, sync_note.committed.metadata()) + == *sync_note.committed.note_id() + })?; + Some((sync_note, &block.block_header)) + }) else { note_records.push(note_record); continue; }; + let committed_note = &sync_note.committed; // A note that carries no attachments has nothing to apply to the record. - let attachments = (!attachments.is_empty()).then_some(attachments); - - let block_header = self - .get_and_store_authenticated_block(committed_note.block_num(), &mut partial_mmr) - .await?; + let attachments = + (!sync_note.attachments.is_empty()).then(|| sync_note.attachments.clone()); let metadata = *committed_note.metadata(); let mut note_changed = note_record @@ -393,7 +402,7 @@ where } // `block_header_received` transitions the record's state, so it must always run. - note_changed |= note_record.block_header_received(&block_header)?; + note_changed |= note_record.block_header_received(block_header)?; // Once committed, the note no longer needs its expected-note tag. if note_changed { @@ -409,33 +418,77 @@ where note_records.push(note_record); } } - self.cache_partial_mmr(partial_mmr).await?; + + self.mark_externally_consumed(&mut note_records).await?; Ok(note_records) } - /// Checks whether the expected notes (identified by their details commitments and tags) have - /// been committed on chain between `request_block_num` and the current block, returning the - /// matching synced notes keyed by details commitment. + /// Marks a record whose nullifier is already on chain as consumed, when the nullifier commit + /// height is at or below the client's sync height. + /// + /// Only a note the node reported as committed carries the metadata a nullifier is derived from, + /// so the rest are skipped. + async fn mark_externally_consumed( + &self, + note_records: &mut [InputNoteRecord], + ) -> Result<(), ClientError> { + let mut nullifiers = BTreeSet::new(); + let mut lowest_commitment_block: BlockNumber = u32::MAX.into(); + for note_record in note_records.iter() { + let (Some(nullifier), Some(inclusion_proof)) = + (note_record.nullifier(), note_record.inclusion_proof()) + else { + continue; + }; + nullifiers.insert(nullifier); + lowest_commitment_block = + lowest_commitment_block.min(inclusion_proof.location().block_num()); + } + + if nullifiers.is_empty() { + return Ok(()); + } + + let spent_heights = self + .rpc_api + .get_nullifier_commit_heights(nullifiers, lowest_commitment_block) + .await?; + + let sync_height = self.get_sync_height().await?; + for note_record in note_records.iter_mut() { + let Some(nullifier) = note_record.nullifier() else { + continue; + }; + if let Some(Some(spent_at)) = spent_heights.get(&nullifier) + && *spent_at <= sync_height + { + note_record.consumed_externally(nullifier, *spent_at, None)?; + } + } + + Ok(()) + } + + /// Fetches every block between `request_block_num` and the client's sync height that holds a + /// note under one of `sync_tags`. /// - /// Expected notes have no metadata and thus no `NoteId`, so each committed note is matched by - /// reconstructing the id from the committed metadata: `NoteId::new(details_commitment, - /// metadata)`. + /// Each block carries its header and the MMR path proving its inclusion at the sync height, + /// which is the forest the client's partial MMR is at, so a note found here needs no further + /// block lookup. Deciding which of the returned notes answer a request is the caller's. async fn sync_expected_notes( - &mut self, + &self, request_block_num: BlockNumber, // Expected notes' details commitments with their tags. - expected_notes: Vec<(NoteDetailsCommitment, NoteTag)>, - ) -> Result, ClientError> { + expected_notes: &[(NoteDetailsCommitment, NoteTag)], + ) -> Result, ClientError> { let sync_tags: BTreeSet = expected_notes.iter().map(|(_, tag)| *tag).collect(); - - let mut matched_notes = BTreeMap::new(); let current_block_num = self.get_sync_height().await?; // Notes expected only after a block we have not reached can't be committed within our // synced view yet: skip the lookup and let them stay expected until a future sync. if request_block_num > current_block_num { - return Ok(matched_notes); + return Ok(Vec::new()); } let blocks = self @@ -449,12 +502,14 @@ where .await .map_err(ClientError::RpcError)?; + let mut matched_blocks = vec![]; for block in blocks { + let mut block_matches = false; if block.block_header.block_num() > current_block_num { break; } - for sync_note in block.notes.into_values() { + for sync_note in block.notes.values() { let committed = &sync_note.committed; // The note carries its own commit height in its inclusion proof, which is a @@ -465,26 +520,35 @@ where continue; } - let Some((commitment, _)) = expected_notes.iter().find(|(commitment, _)| { + let Some((..)) = expected_notes.iter().find(|(commitment, _)| { NoteId::new(*commitment, committed.metadata()) == *committed.note_id() }) else { continue; }; - matched_notes.insert(*commitment, sync_note); + block_matches = true; + } + + if block_matches { + matched_blocks.push(block); } } - Ok(matched_notes) + Ok(matched_blocks) } } +/// A note to import by details: the stored record it updates when there is one, its details, the +/// block from which to look for its commitment, and the tag to track it under. +pub(crate) type NoteImportByDetailsRequest = + (Option, NoteDetails, BlockNumber, NoteTag); + // HELPERS // ================================================================================================ /// Returns an error if the already-stored note is currently being processed by a local transaction, /// since an in-flight note can't be overwritten by an import. -fn ensure_not_processing(previous_note: Option<&InputNoteRecord>) -> Result<(), ClientError> { +pub fn ensure_not_processing(previous_note: Option<&InputNoteRecord>) -> Result<(), ClientError> { if let Some(note) = previous_note && note.is_processing() { diff --git a/crates/rust-client/src/note/note_update_tracker.rs b/crates/rust-client/src/note/note_update_tracker.rs index a2bd9f66fc..ec9e8559a4 100644 --- a/crates/rust-client/src/note/note_update_tracker.rs +++ b/crates/rust-client/src/note/note_update_tracker.rs @@ -377,6 +377,24 @@ impl NoteUpdateTracker { }) } + /// Tracks additional already-persisted input notes. + /// + /// Used to extend a sync's nullifier check to notes that are about to be written by another + /// path (e.g. the note transport sync) and are therefore absent from the store snapshot this + /// tracker was built from. Notes already tracked for the same details commitment are skipped, + /// so a record built by this sync is never replaced by a stale one. + pub(crate) fn track_existing_input_notes( + &mut self, + notes: impl IntoIterator, + ) { + for note in notes { + if self.input_notes.contains_key(¬e.details_commitment()) { + continue; + } + self.insert_input_note(note, NoteUpdateType::None); + } + } + /// Appends nullifiers to the per-account ordered nullifier list. /// /// Nullifiers from the same account must be in execution order; ordering across different diff --git a/crates/rust-client/src/note_transport/mod.rs b/crates/rust-client/src/note_transport/mod.rs index cfa3e3bd38..c1f6c7fd1c 100644 --- a/crates/rust-client/src/note_transport/mod.rs +++ b/crates/rust-client/src/note_transport/mod.rs @@ -14,7 +14,6 @@ use miden_protocol::address::Address; use miden_protocol::block::BlockNumber; use miden_protocol::note::{Note, NoteDetails, NoteDetailsCommitment, NoteHeader, NoteId, NoteTag}; use miden_protocol::utils::serde::Serializable; -use miden_standards::note::{NoteFile, NoteSyncHint}; use miden_tx::auth::TransactionAuthenticator; use miden_tx::utils::serde::{ ByteReader, @@ -25,6 +24,7 @@ use miden_tx::utils::serde::{ }; pub use self::errors::NoteTransportError; +use crate::note::{NoteFile, NoteSyncHint}; use crate::store::{InputNoteRecord, NoteFilter, SettingScope}; use crate::sync::NoteTagSource; use crate::{Client, ClientError}; @@ -354,17 +354,24 @@ where /// fetches only notes past the stored cursor. Historical notes for a newly tracked tag are /// recovered automatically by [`Client::sync_note_transport`], which backfills each new tag. pub async fn fetch_private_notes(&mut self) -> Result<(), ClientError> { + self.ensure_genesis_in_place().await?; + let note_tags: Vec = self.store.get_unique_note_tags().await?.into_iter().collect(); let cursor = self.store.get_note_transport_cursor().await?; - let (_, new_cursor) = self.fetch_transport_notes(cursor, ¬e_tags).await?; + let mut id_by_commitment = BTreeMap::new(); + let (note_files, new_cursor) = + self.fetch_transport_notes(cursor, ¬e_tags, &mut id_by_commitment).await?; + + self.import_notes(¬e_files).await?; self.store.update_note_transport_cursor(new_cursor).await?; Ok(()) } - /// Backfill historical private notes for tags added after the global cursor advanced. + /// Plans the backfill of historical private notes for tags added after the global cursor + /// advanced. /// /// The global transport cursor is shared across all tracked tags and only moves forward, so a /// tag that starts being tracked late never sees its notes that already sit below the cursor. @@ -376,48 +383,52 @@ where /// the steady-state stream is harmless. /// /// At most [`Self::MAX_BACKFILL_TAGS_PER_SYNC`] tags are backfilled per call; any remainder - /// stays uncovered and is picked up on the next sync. Returns the ids of notes imported here. - pub(crate) async fn backfill_new_tags(&mut self) -> Result, ClientError> { + /// stays uncovered and is picked up on the next sync. + /// + /// Returns the pruned covered set, whether pruning changed it, and the tags to backfill. Reads + /// only: persisting the covered set is left to the apply phase, which writes it after the + /// imported notes so a crash re-backfills instead of skipping a tag whose notes were never + /// written. + async fn plan_backfill(&self) -> Result<(BTreeSet, bool, Vec), ClientError> { let candidates = self.backfill_candidate_tags().await?; let loaded = self.load_covered_tags().await?; // Drop tags no longer tracked. Keeping a removed tag marked covered would make a later // re-add skip its backlog, silently missing notes that arrived while it was untracked. - let mut covered: BTreeSet = loaded.intersection(&candidates).copied().collect(); - if covered.len() != loaded.len() { - self.save_covered_tags(&covered).await?; - } + let covered: BTreeSet = loaded.intersection(&candidates).copied().collect(); + let pruned = covered.len() != loaded.len(); - let new_tags: Vec = candidates.difference(&covered).copied().collect(); - - let mut imported_ids = Vec::new(); - for tag in new_tags.into_iter().take(Self::MAX_BACKFILL_TAGS_PER_SYNC) { - imported_ids.extend(self.backfill_tag(tag).await?); - covered.insert(tag); - // Persist after each tag so a crash mid-backfill keeps completed tags covered. A redo - // is harmless because imports dedupe; the dangerous direction (marking covered before - // the import lands) never happens. - self.save_covered_tags(&covered).await?; - } + let new_tags: Vec = candidates + .difference(&covered) + .copied() + .take(Self::MAX_BACKFILL_TAGS_PER_SYNC) + .collect(); - Ok(imported_ids) + Ok((covered, pruned, new_tags)) } /// Drain a single tag's full history from the transport, paging until the cursor stops /// advancing. Uses a local cursor and never touches the global one, so it cannot regress - /// steady-state progress. Returns the ids of the notes it imported. - async fn backfill_tag(&mut self, tag: NoteTag) -> Result, ClientError> { - let mut imported_ids = Vec::new(); + /// steady-state progress. Returns the note files from every fetched page, in page order and + /// none of them written. + async fn backfill_tag( + &self, + tag: NoteTag, + id_by_commitment: &mut BTreeMap, + ) -> Result, ClientError> { + let mut note_files = Vec::new(); let mut cursor = NoteTransportCursor::init(); for _ in 0..Self::MAX_BACKFILL_ITERATIONS { - let (ids, new_cursor) = self.fetch_transport_notes(cursor, &[tag]).await?; - imported_ids.extend(ids); + let (page_files, new_cursor) = + self.fetch_transport_notes(cursor, &[tag], id_by_commitment).await?; + note_files.extend(page_files); // Terminate on any lack of forward progress. A well-behaved server returns `new_cursor // == cursor` when there are no new notes for this tag (since `rcursor = max(cursor, // max_seq_returned)`); using `<=` also handles implementations that return an `init()` // cursor on empty batches (see the in-tree mock transport). + if new_cursor <= cursor { - return Ok(imported_ids); + return Ok(note_files); } cursor = new_cursor; } @@ -427,23 +438,26 @@ where ))) } - /// Fetch one batch of notes from the note transport network for the provided tags. + /// Fetches and returns one batch of notes from the note transport layer for the provided tags + /// without applying any update to the store. /// - /// The server paginates; this method issues one RPC and returns the imported details - /// commitments together with the new cursor. The returned cursor equals the input cursor when - /// the batch was empty (i.e. no new notes). Callers that want to drain a tag's full backlog - /// should loop until `new_cursor == cursor` (see [`Client::backfill_new_tags`]). Callers that - /// do steady-state polling (see [`Client::sync_state`] / [`Client::fetch_private_notes`]) - /// should call this once per tick with the stored cursor. + /// The server paginates; this method issues one transport call and returns the note files + /// together with the new cursor. The returned cursor equals the input cursor when the batch was + /// empty (i.e. no new notes). Callers that want to drain a tag's full backlog should loop until + /// `new_cursor == cursor` (see [`Client::backfill_tag`]). Callers that do steady-state polling + /// (see [`Client::sync_state`] / [`Client::fetch_private_notes`]) should call this once per + /// tick with the stored cursor. /// - /// Downloaded notes are imported into the local store. Persistence of the returned cursor is - /// left to the caller so that drain loops can guard against regression of an already-advanced - /// stored cursor. - pub(crate) async fn fetch_transport_notes( - &mut self, + /// Each downloaded note's id is recorded in `id_by_commitment` so the caller can resolve the + /// written records back to note ids once the final record set is known. Persistence of the + /// returned cursor is left to the caller so that drain loops can guard against regression of an + /// already-advanced stored cursor. + async fn fetch_transport_notes( + &self, cursor: NoteTransportCursor, tags: &[NoteTag], - ) -> Result<(Vec, NoteTransportCursor), ClientError> { + id_by_commitment: &mut BTreeMap, + ) -> Result<(Vec, NoteTransportCursor), ClientError> { // Fallback lookback window, in blocks, used only for notes the transport delivered without // a sender-provided block hint. Scanning back from sync height handles the race where a // note is committed on-chain just before the NTL delivers its data. Without it, @@ -455,7 +469,6 @@ where // TODO: perhaps we should not need to map received IDs with details commitments, and // instead we may allow `InputNoteRecord` to optionally keep NoteIds. Then within // `import_note` we could match everything by ID and remove this map check - let mut id_by_commitment: BTreeMap = BTreeMap::new(); let (note_infos, rcursor) = self.get_note_transport_api()?.fetch_notes(tags, cursor).await?; for note_info in ¬e_infos { @@ -491,24 +504,108 @@ where let fallback_after_block_num = BlockNumber::from(sync_height.as_u32().saturating_sub(NOTE_LOOKBACK_BLOCKS)); - let mut note_requests = Vec::with_capacity(notes.len()); + let mut note_files = Vec::with_capacity(notes.len()); for (note, block_hint) in notes { let tag = note.metadata().tag(); // Prefer the sender-provided hint, falling back to the lookback window when absent. let after_block_num = block_hint.unwrap_or(fallback_after_block_num); - let note_file = NoteFile::ExpectedNote { + note_files.push(NoteFile::ExpectedNote { details: note.into(), sync_hint: NoteSyncHint::new(after_block_num, tag), - }; - note_requests.push(note_file); + }); } - let imported_commitments = self.import_notes(¬e_requests).await?; - let imported_ids = imported_commitments - .into_iter() - .filter_map(|commitment| id_by_commitment.get(&commitment).copied()) + + Ok((note_files, rcursor)) + } + + /// Fetches the notes the Note Transport Layer holds for the tracked tags. + /// + /// Runs the per-tag backfill and fetches a page of notes. This performs no node call and writes + /// nothing but the relay outbox, so it can run concurrently with the chain fetch. The caller + /// imports the returned files and then persists the cursor and the covered-tag set. + /// + /// Returns empty data when note transport is not configured. + pub(crate) async fn fetch_note_transport_updates( + &self, + ) -> Result { + let mut note_transport_update = NoteTransportLayerUpdate::default(); + if !self.is_note_transport_enabled() { + return Ok(note_transport_update); + } + + // Drain any private notes whose previous relay attempt failed. A flush error is logged, not + // propagated: a failing relay must not block the sync, and the entries stay durable for the + // next attempt. This is the one write this phase performs; it touches only the outbox + // setting, which is independent of everything the apply phase writes. + if let Err(err) = self.flush_relay_outbox().await { + tracing::warn!(?err, "relay outbox flush failed during sync; entries retained"); + } + + // Recover historical private notes for any tag added after the global cursor advanced. This + // drains each newly tracked tag from the start, fetching only that tag's own history. + let (mut covered, pruned, new_tags) = self.plan_backfill().await?; + let backfilled = !new_tags.is_empty(); + for tag in new_tags { + note_transport_update + .note_files + .extend(self.backfill_tag(tag, &mut note_transport_update.id_by_commitment).await?); + covered.insert(tag); + } + if pruned || backfilled { + note_transport_update.covered_tags = Some(covered); + } + + let cursor = self.store.get_note_transport_cursor().await?; + let note_tags: Vec = + self.store.get_unique_note_tags().await?.into_iter().collect(); + let (note_files, new_cursor) = self + .fetch_transport_notes(cursor, ¬e_tags, &mut note_transport_update.id_by_commitment) + .await?; + note_transport_update.note_files.extend(note_files); + note_transport_update.cursor = Some(new_cursor); + + Ok(note_transport_update) + } + + /// Writes everything [`Client::fetch_note_transport_updates`] returned, in three steps: + /// + /// 1. Imports the fetched notes, which resolves their on-chain state and stores the records. + /// 2. Saves the covered-tag set, when the backfill changed it. + /// 3. Advances the stored note transport cursor, when a page was fetched. + /// + /// The notes are written before the covered-tag set and the cursor, so a crash between them + /// re-fetches instead of skipping notes that were never written. + /// + /// Returns the ids of the imported notes and the details commitments of the records written. + pub(crate) async fn apply_note_transport_update( + &mut self, + update: NoteTransportLayerUpdate, + ) -> Result<(Vec, Vec), ClientError> { + let NoteTransportLayerUpdate { + note_files, + id_by_commitment, + covered_tags, + cursor, + } = update; + + let written = self.import_notes(¬e_files).await?; + let mut imported_ids: Vec = written + .iter() + .filter_map(|commitment| id_by_commitment.get(commitment).copied()) .collect(); - Ok((imported_ids, rcursor)) + if let Some(covered_tags) = covered_tags { + self.save_covered_tags(&covered_tags).await?; + } + + if let Some(cursor) = cursor { + self.store.update_note_transport_cursor(cursor).await?; + } + + imported_ids.sort_unstable(); + imported_ids.dedup(); + + Ok((imported_ids, written)) } /// Drops deliveries of notes a local transaction is consuming; importing them would fail on the @@ -538,6 +635,26 @@ where } } +// NOTE TRANSPORT FETCH +// ================================================================================================ + +/// What the note transport fetch returned, before anything is written. +/// +/// Built by [`Client::fetch_note_transport_updates`] and consumed by +/// [`Client::apply_note_transport_update`]. +#[derive(Default)] +pub(crate) struct NoteTransportLayerUpdate { + /// Notes to import, backfill pages first and then the steady-state page. + note_files: Vec, + /// Note ids by details commitment, taken from the note headers the transport returned. Used to + /// resolve the written records back to ids. + id_by_commitment: BTreeMap, + /// Covered-tag set to persist, `None` when it did not change. + covered_tags: Option>, + /// New global cursor, from the steady-state page. `None` when no page was fetched. + cursor: Option, +} + /// Note transport cursor /// /// Pagination integer used to reduce the number of fetched notes from the note transport network, diff --git a/crates/rust-client/src/sync/block_header.rs b/crates/rust-client/src/sync/block_header.rs index 67498aabcb..bdedcb1755 100644 --- a/crates/rust-client/src/sync/block_header.rs +++ b/crates/rust-client/src/sync/block_header.rs @@ -9,6 +9,7 @@ use miden_protocol::{Felt, Word}; use tracing::warn; use crate::rpc::NodeRpcClient; +use crate::rpc::domain::note::ResolvedSyncNotesBlock; use crate::store::{BlockRelevance, StoreError}; #[cfg(feature = "testing")] use crate::test_utils::mock::MockRpcApi; @@ -128,6 +129,39 @@ impl Client { Ok(Rpo256::hash_elements(&elements)) } + /// Tracks each fetched note block in `partial_mmr` and stores its header together with the + /// authentication nodes that tracking produced. + pub(crate) async fn insert_note_blocks( + &mut self, + blocks: &[ResolvedSyncNotesBlock], + partial_mmr: &mut PartialMmr, + ) -> Result<(), ClientError> { + let mut authenticated_blocks = Vec::with_capacity(blocks.len()); + for block in blocks { + let block_num = block.block_header.block_num(); + // Also skips a block the loop itself just tracked, so one returned twice is stored + // once. + if partial_mmr.is_tracked(block_num.as_usize()) { + continue; + } + + let path_nodes = track_block_in_mmr( + partial_mmr, + block_num, + block.block_header.commitment(), + &block.mmr_path, + )?; + authenticated_blocks.push((block.block_header.clone(), path_nodes)); + } + + for (block_header, path_nodes) in authenticated_blocks { + let nodes = authenticated_block_nodes(&block_header, path_nodes); + self.store.insert_block_header(&block_header, &nodes, true).await?; + } + + Ok(()) + } + // HELPERS // -------------------------------------------------------------------------------------------- diff --git a/crates/rust-client/src/sync/mod.rs b/crates/rust-client/src/sync/mod.rs index d503b2a9f9..1911ce8c64 100644 --- a/crates/rust-client/src/sync/mod.rs +++ b/crates/rust-client/src/sync/mod.rs @@ -57,6 +57,7 @@ //! store. use alloc::collections::BTreeSet; +use alloc::format; use alloc::sync::Arc; use alloc::vec::Vec; use core::cmp::max; @@ -68,7 +69,7 @@ use miden_protocol::note::NoteId; use miden_protocol::transaction::TransactionId; use miden_tx::auth::TransactionAuthenticator; use miden_tx::utils::serde::{Deserializable, DeserializationError, Serializable}; -use tracing::{debug, info}; +use tracing::{debug, info, warn}; use crate::pswap::PswapChainObserver; use crate::store::{NoteFilter, TransactionFilter}; @@ -82,7 +83,8 @@ mod note_observer; pub use note_observer::NoteObserver; mod state_sync; -pub use state_sync::{NoteUpdateAction, OnNoteReceived, StateSync, StateSyncInput}; +pub(crate) use state_sync::block_num_from_forest; +pub use state_sync::{ChainSyncData, NoteUpdateAction, OnNoteReceived, StateSync, StateSyncInput}; mod state_sync_update; pub use state_sync_update::{ @@ -128,23 +130,74 @@ where /// Does **not** fetch private notes from the Note Transport Layer. Use [`Client::sync_state`] /// for the combined sync, or call [`Client::sync_note_transport`] separately. /// - /// Builds the default sync input, runs [`StateSync::sync_state`] (see that method for the - /// detailed pipeline), applies the resulting update to the store, caches the partial MMR, and - /// prunes irrelevant blocks according to the configured cadence. + /// Fetches everything from the node first ([`Client::fetch_chain_updates`] and + /// [`StateSync::fetch_nullifiers`]), then applies the result with + /// [`Client::apply_chain_updates`], which also caches the partial MMR and prunes irrelevant + /// blocks according to the configured cadence. pub async fn sync_chain(&mut self) -> Result { 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. - let note_screener = self.note_screener(); - let state_sync = - StateSync::new(self.rpc_api.clone(), Arc::new(note_screener), self.tx_discard_delta) - .with_note_observer(Arc::new(PswapChainObserver::new(self.store.clone()))); + let state_sync = self.state_sync(); + let mut chain_sync_data = self.fetch_chain_updates(&state_sync).await?; + state_sync.derive_state_updates(&mut chain_sync_data).await?; + state_sync.fetch_nullifiers(&mut chain_sync_data).await?; + + self.apply_chain_updates(&state_sync, chain_sync_data).await + } + + /// Fetches the node's view of everything that changed since the client's chain tip, without + /// storing anything or modifying the partial MMR. + /// + /// Builds the default sync input and runs [`StateSync::fetch_state`]. The state updates must be + /// derived with [`StateSync::derive_state_updates`]. The nullifier check is not part of this: + /// run [`StateSync::fetch_nullifiers`] on the result before applying it, so it can also cover + /// transport-delivered notes another sync path fetched in the same call. + pub async fn fetch_chain_updates( + &self, + state_sync: &StateSync, + ) -> Result { let input = self.build_sync_input().await?; + let block_from = block_num_from_forest(&self.get_current_partial_mmr().await?)?; + state_sync.fetch_state(block_from, input).await + } + + /// Builds the [`StateSync`] driving one chain sync. + /// + /// Each `NoteObserver` owns its own per-sync state, so this must be called once per sync rather + /// than shared; `with_note_observer` just attaches it. + fn state_sync(&self) -> StateSync { + StateSync::new(self.rpc_api.clone(), Arc::new(self.note_screener()), self.tx_discard_delta) + .with_note_observer(Arc::new(PswapChainObserver::new(self.store.clone()))) + } + + /// Verifies fetched chain data against the client's partial MMR and saves the resulting update + /// to the store. + /// + /// [`StateSync::derive_state_updates`] and [`StateSync::fetch_nullifiers`] must have run on the + /// data first. Also caches the partial MMR and prunes irrelevant blocks. + /// + /// # Errors + /// + /// Returns an error if the client no longer starts where the data was fetched from, which means + /// another sync advanced the store in between and the data is stale. + pub async fn apply_chain_updates( + &mut self, + state_sync: &StateSync, + chain_sync_data: ChainSyncData, + ) -> Result { let mut partial_mmr = self.get_current_partial_mmr().await?; - let state_sync_update = state_sync.sync_state(&mut partial_mmr, input).await?; + let block_from = block_num_from_forest(&partial_mmr)?; + if block_from != chain_sync_data.block_from { + return Err(ClientError::ChainValidationError(format!( + "chain sync chain_sync_data starts at block {} but the client is at block {block_from}", + chain_sync_data.block_from + ))); + } + + let state_sync_update = StateSync::build_update(chain_sync_data, &mut partial_mmr)?; let sync_summary: SyncSummary = (&state_sync_update).into(); debug!(sync_summary = ?sync_summary, "Sync summary computed"); @@ -177,42 +230,71 @@ where if !self.is_note_transport_enabled() { return Ok(Vec::new()); } + self.ensure_genesis_in_place().await?; - // Drain any private notes whose previous relay attempt failed. A flush error is logged, not - // propagated: a failing relay must not block the sync, and the entries stay durable for the - // next attempt. - if let Err(err) = self.flush_relay_outbox().await { - tracing::warn!(?err, "relay outbox flush failed during sync; entries retained"); - } - - // Recover historical private notes for any tag added after the global cursor advanced. This - // drains each newly tracked tag from the start, fetching only that tag's own history. - let mut imported_ids = self.backfill_new_tags().await?; - - let cursor = self.store.get_note_transport_cursor().await?; - let note_tags: Vec<_> = self.store.get_unique_note_tags().await?.into_iter().collect(); - let (ids, new_cursor) = self.fetch_transport_notes(cursor, ¬e_tags).await?; - self.store.update_note_transport_cursor(new_cursor).await?; - imported_ids.extend(ids); - - imported_ids.sort_unstable(); - imported_ids.dedup(); - + let note_transport_update = self.fetch_note_transport_updates().await?; + let (imported_ids, _) = self.apply_note_transport_update(note_transport_update).await?; Ok(imported_ids) } - /// Runs the full client sync. + /// Runs the full client sync: private notes from the Note Transport Layer and the client's + /// on-chain state with the Miden node. + /// + /// The NTL and the node are fetched concurrently, and everything that writes runs sequentially + /// afterwards: /// - /// First fetches private notes from the Note Transport Layer (see - /// [`Client::sync_note_transport`]), then syncs the client's on-chain state with the Miden node - /// (see [`Client::sync_chain`]). If note transport is disabled, this is equivalent to - /// [`Client::sync_chain`]. + /// 1. Concurrently: the note transport fetch and [`Client::fetch_chain_updates`]. Only node and + /// NTL calls happen here, which is all that benefits from overlapping. + /// 2. The transport writes, when its fetch succeeded, whose records are then tracked in the + /// chain sync's note updates. + /// 3. [`StateSync::derive_state_updates`], which screens the node's notes against the store — + /// hence after step 2, so a transport-delivered note is recognised rather than discarded — + /// and applies a commitment reported this sync to those records. + /// 4. [`StateSync::fetch_nullifiers`], covering the tracked notes *and* the transport-delivered + /// ones, so a note delivered and consumed in the same window is reported as consumed by this + /// call. + /// 5. The chain update, written last: a nullified transport-delivered note is saved as an + /// update to the row step 2 inserts. /// - /// Fails fast on the first error. Private notes delivered via NTL are imported before the chain - /// sync reads its input set, so their nullifiers are checked in the same call. + /// A transport failure is logged and the chain sync continues without it, leaving the transport + /// cursor for the next call to retry. Before step 2 but the relay outbox, which + /// [`Client::flush_relay_outbox`] persists during the fetch and the next sync retries. pub async fn sync_state(&mut self) -> Result { - let new_private_notes = self.sync_note_transport().await?; - let mut summary = self.sync_chain().await?; + // Both fetch phases need genesis in place, and connecting here means the two concurrent + // futures never race on the RPC client's lazy connect. + self.ensure_genesis_in_place().await?; + self.ensure_rpc_limits_in_place().await?; + + let state_sync = self.state_sync(); + let (note_transport_update, chain_sync_data) = futures::join!( + self.fetch_note_transport_updates(), + self.fetch_chain_updates(&state_sync), + ); + + // An NTL failure does not end the sync + let (new_private_notes, imported_notes) = match note_transport_update { + Ok(note_transport_update) => { + self.apply_note_transport_update(note_transport_update).await? + }, + Err(err) => { + warn!(?err, "note transport fetch failed; syncing the chain without it"); + (Vec::new(), Vec::new()) + }, + }; + + let mut chain_sync_data = chain_sync_data?; + + // The chain sync built its note updates from a store snapshot taken before the import, so + // the imported records are added here. Without them this sync has no record to apply its + // verdicts to, and a note committed within this sync's own block range stays expected. + let imported_notes = + self.get_input_notes(NoteFilter::DetailsCommitments(imported_notes)).await?; + chain_sync_data.note_updates.track_existing_input_notes(imported_notes); + + state_sync.derive_state_updates(&mut chain_sync_data).await?; + state_sync.fetch_nullifiers(&mut chain_sync_data).await?; + + let mut summary = self.apply_chain_updates(&state_sync, chain_sync_data).await?; summary.new_private_notes = new_private_notes; Ok(summary) } diff --git a/crates/rust-client/src/sync/state_sync.rs b/crates/rust-client/src/sync/state_sync.rs index 1f61d8fb45..db5f09c5e7 100644 --- a/crates/rust-client/src/sync/state_sync.rs +++ b/crates/rust-client/src/sync/state_sync.rs @@ -266,23 +266,55 @@ impl StateSync { /// /// Use [`Client::build_sync_input()`](`crate::Client::build_sync_input()`) to build the default /// input, or assemble it manually for custom sync. The `current_partial_mmr` is taken by - /// mutable reference so callers can keep it in memory across syncs. + /// mutable reference so callers can keep it in memory across syncs; it is only modified once + /// every check has passed. /// /// During the sync process, the following steps are performed: /// 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. - /// 5. Process transaction inclusions (commit local txs, record external consumers, discard + /// 3. Screen note inclusions via the configured [`OnNoteReceived`] callback. + /// 4. Process transaction inclusions (commit local txs, record external consumers, discard /// stale/expired txs, commit output notes). + /// 5. Recover the public notes a tracked account consumed but the client never tracked. /// 6. Detect consumed notes via nullifier sync (optional, see /// [`Self::disable_nullifier_sync`]). - /// 7. Track in the MMR the screened blocks that still hold an unspent note. + /// 7. Advance the partial MMR to the chain tip and track the screened blocks that still hold an + /// unspent note. + /// + /// Steps 1-2 are [`Self::fetch_state`], 3-5 [`Self::derive_state_updates`], 6 + /// [`Self::fetch_nullifiers`] and 7 [`Self::build_update`]; each can be driven separately. pub async fn sync_state( &self, current_partial_mmr: &mut PartialMmr, input: StateSyncInput, ) -> Result { + let block_num = block_num_from_forest(current_partial_mmr)?; + + let mut chain_sync_data = self.fetch_state(block_num, input).await?; + self.derive_state_updates(&mut chain_sync_data).await?; + self.fetch_nullifiers(&mut chain_sync_data).await?; + + // Work on a clone so any validation failure leaves `current_partial_mmr` untouched. + let mut working_mmr = current_partial_mmr.clone(); + let update = Self::build_update(chain_sync_data, &mut working_mmr)?; + *current_partial_mmr = working_mmr; + + Ok(update) + } + + /// Fetches the node's view of everything that changed since `block_from`: the MMR delta, the + /// note inclusions, the transactions, and the account states. + /// + /// Every node call that does not depend on note screening happens here, so a caller can run + /// this concurrently with another sync's fetch. Interpreting the response is + /// [`Self::derive_state_updates`]'s, and the nullifier check [`Self::fetch_nullifiers`]'s. Both + /// run afterwards so a caller syncing more than one source can write the other source first, + /// and check nullifiers once across all of them. + pub async fn fetch_state( + &self, + block_from: BlockNumber, + input: StateSyncInput, + ) -> Result { let StateSyncInput { accounts, note_tags, @@ -290,28 +322,25 @@ impl StateSync { output_notes, uncommitted_transactions, } = input; - let block_num = u32::try_from(current_partial_mmr.forest().num_leaves().saturating_sub(1)) - .map_err(|_| ClientError::InvalidPartialMmrForest)? - .into(); let note_tags = Arc::new(note_tags); let account_ids: Vec = accounts.iter().map(AccountHeader::id).collect(); - let mut note_updates = NoteUpdateTracker::new(input_notes, output_notes); - let mut transaction_updates = TransactionUpdateTracker::new(uncommitted_transactions); - let mut partial_blockchain_updates = PartialBlockchainUpdates::default(); + let note_updates = NoteUpdateTracker::new(input_notes, output_notes); + let transaction_updates = TransactionUpdateTracker::new(uncommitted_transactions); let mut account_updates = AccountUpdates::default(); - let Some(sync_data) = self.fetch_sync_data(block_num, &account_ids, ¬e_tags).await? + let Some(sync_data) = self.fetch_sync_data(block_from, &account_ids, ¬e_tags).await? else { // No progress — already at the tip. - return Ok(StateSyncUpdate::from_parts( - block_num, - partial_blockchain_updates, + return Ok(ChainSyncData { + block_from, + advance: None, + superseded_states: Vec::new(), note_updates, transaction_updates, account_updates, - )); + }); }; let FetchedSyncData { @@ -320,7 +349,6 @@ impl StateSync { note_blocks, transactions, } = sync_data; - let chain_tip = chain_tip_header.block_num(); let new_commitments = derive_account_commitments(&transactions); let superseded_states = self @@ -328,44 +356,122 @@ impl StateSync { &mut account_updates, &accounts, &new_commitments, - block_num, + block_from, &chain_tip_header, ) .await?; + Ok(ChainSyncData { + block_from, + advance: Some(ChainAdvance { + chain_tip_header, + mmr_delta, + note_blocks_awaiting_screening: note_blocks, + transactions, + relevant_note_blocks: Vec::new(), + }), + superseded_states, + note_updates, + transaction_updates, + account_updates, + }) + } + + /// Turns the node's raw response into note and transaction updates. + /// + /// Discards the local transactions the node superseded, screens the received notes for + /// relevance, applies the transaction inclusions, and recovers the public notes the tracked + /// accounts consumed, fetching those by id. + pub async fn derive_state_updates( + &self, + chain_sync_data: &mut ChainSyncData, + ) -> Result<(), ClientError> { + let ChainSyncData { + advance, + superseded_states, + note_updates, + transaction_updates, + .. + } = chain_sync_data; + + let Some(advance) = advance.as_mut() else { + return Ok(()); + }; + // Discard the local transactions whose result lost a same-nonce race against the network. - for superseded_state in superseded_states { + for superseded_state in core::mem::take(superseded_states) { transaction_updates.apply_superseded_account_state(superseded_state); } - // Work on a clone so any validation failure leaves `current_partial_mmr` untouched. - let mut working_mmr = current_partial_mmr.clone(); + advance.relevant_note_blocks = self + .screen_note_blocks( + core::mem::take(&mut advance.note_blocks_awaiting_screening), + note_updates, + ) + .await?; - Self::advance_mmr( - mmr_delta, - &chain_tip_header, - &mut working_mmr, - &mut partial_blockchain_updates, - )?; - let relevant_note_blocks = self.screen_note_blocks(note_blocks, &mut note_updates).await?; self.apply_transactions_and_nullifiers( - &chain_tip_header, - &transactions, - &mut note_updates, - &mut transaction_updates, + &advance.chain_tip_header, + &advance.transactions, + note_updates, + transaction_updates, )?; - if self.sync_nullifiers { - self.nullifiers_state_sync( - &mut note_updates, - &mut transaction_updates, - chain_tip, - block_num, - ) - .await?; + self.recover_consumed_public_notes(note_updates, &advance.transactions).await?; + + Ok(()) + } + + /// Verifies the fetched chain data against `partial_mmr` and turns it into the update to apply + /// to the store. + /// + /// It applies the node's delta, checks the resulting peaks against the chain tip header's chain + /// commitment, and tracks the screened note blocks that still hold an unspent note. + pub fn build_update( + chain_sync_data: ChainSyncData, + partial_mmr: &mut PartialMmr, + ) -> Result { + let ChainSyncData { + block_from, + advance, + note_updates, + transaction_updates, + account_updates, + .. + } = chain_sync_data; + + let mut partial_blockchain_updates = PartialBlockchainUpdates::default(); + + let Some(ChainAdvance { + chain_tip_header, + mmr_delta, + note_blocks_awaiting_screening, + relevant_note_blocks, + .. + }) = advance + else { + // No progress — already at the tip. + return Ok(StateSyncUpdate::from_parts( + block_from, + partial_blockchain_updates, + note_updates, + transaction_updates, + account_updates, + )); + }; + // Check the note blocks have been screened before building the update + if !note_blocks_awaiting_screening.is_empty() { + return Err(ClientError::UnscreenedNoteBlocks); } - self.recover_consumed_public_notes(&mut note_updates, &transactions).await?; + let chain_tip = chain_tip_header.block_num(); + + Self::advance_mmr( + mmr_delta, + &chain_tip_header, + partial_mmr, + &mut partial_blockchain_updates, + )?; let blocks_with_unspent_notes: BTreeSet = note_updates.unspent_input_note_block_numbers().collect(); @@ -373,12 +479,10 @@ impl StateSync { Self::validate_and_track_note_blocks( relevant_note_blocks, &blocks_with_unspent_notes, - &mut working_mmr, + partial_mmr, &mut partial_blockchain_updates, )?; - *current_partial_mmr = working_mmr; - Ok(StateSyncUpdate::from_parts( chain_tip, partial_blockchain_updates, @@ -388,6 +492,39 @@ impl StateSync { )) } + /// Checks the node for nullifiers of every note `chain_sync_data` could have consumed. + /// + /// The query covers every note the tracker holds, including any the caller added from another + /// sync path — so a transport-delivered note consumed within one sync is reported as consumed + /// by that same sync. + /// + /// No-op when the nullifier sync is disabled (see [`Self::disable_nullifier_sync`]) or when the + /// node reported no progress, since there is no block range to query. + pub async fn fetch_nullifiers( + &self, + chain_sync_data: &mut ChainSyncData, + ) -> Result<(), ClientError> { + if !self.sync_nullifiers { + return Ok(()); + } + + let Some(chain_tip) = chain_sync_data + .advance + .as_ref() + .map(|advance| advance.chain_tip_header.block_num()) + else { + return Ok(()); + }; + + self.nullifiers_state_sync( + &mut chain_sync_data.note_updates, + &mut chain_sync_data.transaction_updates, + chain_tip, + chain_sync_data.block_from, + ) + .await + } + /// Recovers public notes a watched account consumed, from the `consumed_note_refs` the node /// attaches to its transactions. Fetches the body of each not-yet-tracked note by id and hands /// it to [`NoteUpdateTracker::insert_consumed_public_note`]. Notes the node doesn't return are @@ -1291,9 +1428,54 @@ impl StateSync { } } +// CHAIN SYNC DATA +// ================================================================================================ + +/// The chain data a sync fetched from the node, before any of it has been verified against the +/// client's MMR or to the store. +/// +/// Built by [`StateSync::fetch_state`], extended by [`StateSync::fetch_nullifiers`] and turned into +/// a [`StateSyncUpdate`] by [`StateSync::build_update`]. +pub struct ChainSyncData { + /// The chain tip the sync started from. + pub(crate) block_from: BlockNumber, + /// What the node reported beyond `block_from`, or `None` when the client was already at the + /// chain tip. + advance: Option, + /// Account states the node superseded, to be applied to the transaction updates. + superseded_states: Vec, + /// Notes as the sync found them. A caller that wrote notes of its own after the sync input was + /// built has to track them here, or this sync's verdicts have no record to apply to. + pub(crate) note_updates: NoteUpdateTracker, + transaction_updates: TransactionUpdateTracker, + account_updates: AccountUpdates, +} + +/// The part of a [`ChainSyncData`] that only exists when the node reported progress. +struct ChainAdvance { + /// Header of the chain tip the sync advanced to. + chain_tip_header: BlockHeader, + /// MMR delta from `block_from` to the chain tip, excluding the chain-tip leaf. + mmr_delta: MmrDelta, + /// Note blocks as the node returned them. [`StateSync::derive_state_updates`] drains these into + /// `relevant_note_blocks`, so this is empty by the time the update is built. + note_blocks_awaiting_screening: Vec, + /// Transaction records as the node returned them, read by [`StateSync::derive_state_updates`]. + transactions: Vec, + /// Screened blocks holding a client-relevant note, each with its `sync_notes` MMR path. + relevant_note_blocks: Vec, +} + // HELPERS // ================================================================================================ +/// Returns the block number the given partial MMR is synced to. +pub(crate) fn block_num_from_forest(partial_mmr: &PartialMmr) -> Result { + Ok(u32::try_from(partial_mmr.forest().num_leaves().saturating_sub(1)) + .map_err(|_| ClientError::InvalidPartialMmrForest)? + .into()) +} + /// Groups transaction records by `(account_id, block_num)`. fn group_txs_by_account_block( transaction_records: &[RpcTransactionRecord], diff --git a/crates/testing/miden-client-tests/src/tests/transport.rs b/crates/testing/miden-client-tests/src/tests/transport.rs index 31ee8dc95e..fd203e60a2 100644 --- a/crates/testing/miden-client-tests/src/tests/transport.rs +++ b/crates/testing/miden-client-tests/src/tests/transport.rs @@ -35,6 +35,7 @@ use miden_protocol::asset::{Asset, FungibleAsset}; use miden_protocol::block::BlockNumber; use miden_protocol::crypto::rand::RandomCoin; use miden_protocol::note::{NoteAttachment, NoteAttachmentScheme, NoteType as ProtocolNoteType}; +use miden_protocol::testing::account_id::{ACCOUNT_ID_PRIVATE_FUNGIBLE_FAUCET, ACCOUNT_ID_SENDER}; use miden_protocol::transaction::RawOutputNote; use miden_protocol::utils::serde::Serializable; use miden_protocol::{Felt, Word}; @@ -574,7 +575,7 @@ async fn fetch_private_notes_finds_note_committed_at_sync_height() { let details_bytes = details.to_bytes(); mock_transport_node.write().add_note(*private_note.header(), details_bytes); - // 6. Second sync_state: fetch_transport_notes imports the note, then chain sync runs. The + // 6. Second sync_state: the transport page is imported, then chain sync runs. The // chain scan starts from a lookback window rather than from the sync height, so it still sees // the note at block 1. let summary = client.sync_state().await.unwrap(); @@ -591,6 +592,201 @@ async fn fetch_private_notes_finds_note_committed_at_sync_height() { ); } +/// A private note delivered over the NTL must be committed by the same `sync_state` call that +/// advances past its commitment block. +/// +/// The commitment is learned from two independent sources that only combine through the store: the +/// NTL supplies the note's details, which the transport half writes as an `Expected` record, and +/// the node reports the commitment for the note's tag, which the chain half screens with +/// `NoteScreener::on_note_received`. That screening is a store lookup — a private note carries no +/// details from the node, so a record it cannot find is discarded — which makes the order +/// load-bearing: the transport half must write before the chain half screens. +/// +/// This is the case the lookback in `fetch_private_notes_finds_note_committed_at_sync_height` does +/// not cover. Here the note commits *above* the client's sync height, so the transport half's own +/// commitment check (capped at the stored sync height) cannot see it and the chain half is the only +/// thing that can. The chain sync's note query is a forward-moving window, so a commitment +/// discarded here is never revisited: the record would stay `Expected` forever. +#[tokio::test] +async fn ntl_note_committed_within_the_sync_window_is_committed_by_that_sync() { + // 1. Commit a private note at block 1, then advance the chain past it. + let mut mock_chain_builder = MockChainBuilder::new(); + let mock_account = mock_chain_builder + .add_existing_mock_account(miden_testing::Auth::IncrNonce) + .unwrap(); + + let private_note = NoteBuilder::new( + mock_account.id(), + RandomCoin::new([9, 8, 7, 6].map(Felt::new_unchecked).into()), + ) + .note_type(ProtocolNoteType::Private) + .tag(NoteTag::new(0).into()) + .build() + .unwrap(); + + let spawn_note = + mock_chain_builder.add_spawn_note(std::slice::from_ref(&private_note)).unwrap(); + let mut mock_chain = mock_chain_builder.build().unwrap(); + + let tx = Box::pin( + mock_chain + .build_transaction(MockTransactionInput::AccountId(mock_account.id())) + .unauthenticated_input_note(spawn_note) + .expected_output_notes(vec![RawOutputNote::Full(private_note.clone())]) + .build() + .unwrap() + .execute(), + ) + .await + .unwrap(); + mock_chain.add_pending_executed_transaction(&tx).unwrap(); + mock_chain.prove_next_block().unwrap(); + + for _ in 0..5 { + mock_chain.prove_next_block().unwrap(); + } + + // 2. Build a client that has never synced, so its sync height sits below the note's block. + let mock_transport_node = Arc::new(RwLock::new(MockNoteTransportNode::new())); + + let rpc_api = Arc::new(MockRpcApi::new(mock_chain)); + let transport_client = MockNoteTransportApi::new(mock_transport_node.clone()); + + let mut rng = rand::rng(); + let coin_seed: [u64; 4] = rng.random(); + let rng = RandomCoin::new(coin_seed.map(|v| Felt::new_unchecked(v >> 1)).into()); + + let keystore = FilesystemKeyStore::new(temp_dir()).unwrap(); + + let builder: ClientBuilder = ClientBuilder::new() + .rpc(rpc_api) + .rng(Box::new(rng)) + .sqlite_store(create_test_store_path()) + .authenticator(Arc::new(keystore)) + .tx_discard_delta(None) + .note_transport(Arc::new(transport_client)); + + let mut client = builder.build().await.unwrap(); + client.ensure_genesis_in_place().await.unwrap(); + seed_mock_transaction_encryption_key(&mut client).await; + + client.add_note_tag(NoteTag::new(0)).await.unwrap(); + + let sync_height_before = client.get_sync_height().await.unwrap(); + assert_eq!( + sync_height_before, + BlockNumber::GENESIS, + "the note must commit above the sync height for this test to exercise the chain half" + ); + + // 3. Deliver the note over the NTL before that first sync. + let details_bytes = NoteDetails::from(private_note.clone()).to_bytes(); + mock_transport_node.write().add_note(*private_note.header(), details_bytes); + + // 4. One sync: the transport half receives the details, the chain half reports the commitment + // at block 1, and the window (genesis, tip] is consumed. + client.sync_state().await.unwrap(); + + assert!( + client.get_sync_height().await.unwrap() > BlockNumber::from(1), + "the sync must have advanced past the note's commitment block" + ); + + let committed_notes = client.get_input_notes(NoteFilter::Committed).await.unwrap(); + assert!( + committed_notes.iter().any(|note| note.id() == Some(private_note.id())), + "a delivered note committed inside the synced window must be committed by that sync; \ + leaving it expected strands it, because the chain sync never revisits that block range" + ); +} + +/// A note delivered over the NTL whose nullifier is already on chain must be stored as consumed. +/// +/// Probe for 0xMiden/rust-sdk#2422. +#[tokio::test] +async fn ntl_note_already_spent_below_the_checkpoint_is_not_left_committed() { + let sender_id: AccountId = ACCOUNT_ID_SENDER.try_into().unwrap(); + let faucet_id: AccountId = ACCOUNT_ID_PRIVATE_FUNGIBLE_FAUCET.try_into().unwrap(); + + // 1. Commit a private note to the account, then spend it — both far below the eventual tip. + let mut builder = MockChainBuilder::new(); + let account = builder.add_existing_mock_account(Auth::IncrNonce).unwrap(); + let asset = Asset::Fungible(FungibleAsset::new(faucet_id, 100u64).unwrap()); + let note = builder + .add_p2id_note(sender_id, account.id(), &[asset], ProtocolNoteType::Private) + .unwrap(); + + let mut mock_chain = builder.build().unwrap(); + mock_chain.prove_next_block().unwrap(); // block 1: the note is committed + + let consume_tx = Box::pin( + mock_chain + .build_transaction(MockTransactionInput::Account(account.clone())) + .unauthenticated_input_note(note.clone()) + .build() + .unwrap() + .execute(), + ) + .await + .unwrap(); + mock_chain.add_pending_executed_transaction(&consume_tx).unwrap(); + mock_chain.prove_next_block().unwrap(); // block 2: the nullifier is on chain + + for _ in 0..5 { + mock_chain.prove_next_block().unwrap(); + } + + // 2. A freshly restored client, tracking the note's tag, synced to the tip. + let mock_transport_node = Arc::new(RwLock::new(MockNoteTransportNode::new())); + let rpc_api = Arc::new(MockRpcApi::new(mock_chain)); + let transport_client = MockNoteTransportApi::new(mock_transport_node.clone()); + + let mut rng = rand::rng(); + let coin_seed: [u64; 4] = rng.random(); + let rng = RandomCoin::new(coin_seed.map(|v| Felt::new_unchecked(v >> 1)).into()); + let keystore = FilesystemKeyStore::new(temp_dir()).unwrap(); + + let builder: ClientBuilder = ClientBuilder::new() + .rpc(rpc_api) + .rng(Box::new(rng)) + .sqlite_store(create_test_store_path()) + .authenticator(Arc::new(keystore)) + .tx_discard_delta(None) + .note_transport(Arc::new(transport_client)); + + let mut client = builder.build().await.unwrap(); + client.ensure_genesis_in_place().await.unwrap(); + seed_mock_transaction_encryption_key(&mut client).await; + client.add_note_tag(note.metadata().tag()).await.unwrap(); + + client.sync_state().await.unwrap(); + let checkpoint = client.get_sync_height().await.unwrap(); + assert!( + checkpoint > BlockNumber::from(2), + "the spend must sit below the checkpoint for this test to exercise the gap" + ); + + // 3. The transport now re-serves the note, as it does for a cursor-0 client. + let details_bytes = NoteDetails::from(note.clone()).to_bytes(); + mock_transport_node.write().add_note(*note.header(), details_bytes); + + client.sync_state().await.unwrap(); + + // The import must have happened, otherwise the assertion below passes for the wrong reason. + let all_notes = client.get_input_notes(NoteFilter::All).await.unwrap(); + assert!( + all_notes.iter().any(|n| n.details_commitment() == note.details_commitment()), + "the delivered note should have been imported" + ); + + let committed = client.get_input_notes(NoteFilter::Committed).await.unwrap(); + assert!( + !committed.iter().any(|n| n.id() == Some(note.id())), + "a note whose nullifier is already on chain must not be imported as committed: \ + the forward-only nullifier query never revisits the block that spent it" + ); +} + /// A private note must reach the recipient even when the sender's first relay attempt fails, /// provided the transport later recovers. /// @@ -946,8 +1142,8 @@ async fn transport_fetch_failure_leaves_cursor_for_retry() { .add_note(*note.header(), NoteDetails::from(note.clone()).to_bytes()); faulty.fail_next_n_fetches(2); - recipient.sync_state().await.unwrap_err(); - recipient.sync_state().await.unwrap_err(); + recipient.sync_state().await.unwrap(); + recipient.sync_state().await.unwrap(); assert_eq!(faulty.fetch_attempts(), 2); assert_eq!(recipient.get_input_notes(NoteFilter::All).await.unwrap().len(), 0);