refactor(rust-client): make NTL and chain syncing concurrent

CHANGELOG.md

@@ -6,12 +6,18 @@
 
 * [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)).
 
+
 ### 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)).
 
+### Enhancements
+
+* [rust] The chain sync and the note transport sync are each split into a fetch phase and a store phase, so every NTL and RPC call happens before the first store write. `Client::sync_state` runs the two fetch phases concurrently and applies both sets of updates afterwards, instead of running a full note transport sync before the chain sync ([#2453](https://github.com/0xMiden/rust-sdk/pull/2453)).
+
 ## 0.16.0 (2026-09-07)
 
 ### Breaking Changes

crates/rust-client/src/errors.rs

@@ -244,6 +244,8 @@ pub enum ClientError {
     /// `From<MyFeatureError> for ClientError` returning `Observer(Box::new(err))`.
     #[error(transparent)]
     Observer(Box<dyn core::error::Error + Send + Sync + 'static>),
+    #[error("expected note blocks to be screened before state sync update is built")]
+    UnscreenedNoteBlocks,
 }
 
 // OBSERVER FAN-OUT

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, SyncedNote};
 use crate::rpc::{NoteContentFetch, RpcError};
 use crate::store::input_note_states::ExpectedNoteState;
 use crate::store::{InputNoteRecord, InputNoteState, NoteFilter};
@@ -68,6 +68,18 @@ where
         &mut self,
         note_files: &[NoteFile],
     ) -> Result<Vec<NoteDetailsCommitment>, ClientError> {
+        let records = self.import_note_records(note_files).await?;
+        Ok(records.iter().map(InputNoteRecord::details_commitment).collect())
+    }
+
+    /// Imports `note_files` and returns the records written to the store.
+    ///
+    /// A caller that must act on the imported notes in the same operation needs the records, not
+    /// just their commitments: a sync uses them to extend its own note updates.
+    pub(crate) async fn import_note_records(
+        &mut self,
+        note_files: &[NoteFile],
+    ) -> Result<Vec<InputNoteRecord>, ClientError> {
         self.ensure_genesis_in_place().await?;
 
         // Deduplicate the incoming files, keeping note IDs and details commitments in separate
@@ -128,7 +140,7 @@ where
                         previous_note,
                         details,
                         sync_hint.after_block_num(),
-                        Some(sync_hint.tag()),
+                        sync_hint.tag(),
                     ));
                 },
                 NoteFile::Committed { note, proof } => {
@@ -156,20 +168,19 @@ where
             imported_notes.extend(notes_by_proof);
         }
 
-        let mut imported_commitments = Vec::with_capacity(imported_notes.len());
-        for note in imported_notes {
-            let details_commitment = note.details_commitment();
+        for note in &imported_notes {
+            // A record still expected needs its tag tracked so a later sync finds it. A committed
+            // one is no longer in that state, so it is skipped here.
             if let InputNoteState::Expected(ExpectedNoteState { tag: Some(tag), .. }) = note.state()
             {
                 self.store
-                    .add_note_tag(NoteTagRecord::with_note_source(*tag, details_commitment))
+                    .add_note_tag(NoteTagRecord::with_note_source(*tag, note.details_commitment()))
                     .await?;
             }
-            self.store.upsert_input_notes(&[note]).await?;
-            imported_commitments.push(details_commitment);
         }
+        self.store.upsert_input_notes(&imported_notes).await?;
 
-        Ok(imported_commitments)
+        Ok(imported_notes)
     }
 
     // HELPERS
@@ -342,36 +353,74 @@ where
     /// the new information changed them.
     async fn import_note_records_by_details(
         &mut self,
-        requested_notes: Vec<(Option<InputNoteRecord>, NoteDetails, BlockNumber, Option<NoteTag>)>,
+        requested_notes: Vec<NoteImportRequest>,
     ) -> Result<Vec<InputNoteRecord>, ClientError> {
         let mut lowest_request_block: BlockNumber = u32::MAX.into();
-        let mut note_requests = vec![];
+        let mut sync_tags = BTreeSet::new();
+        let mut requested_commitments = Vec::with_capacity(requested_notes.len());
         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);
+            sync_tags.insert(*tag);
+            requested_commitments.push(details.commitment());
+            lowest_request_block = lowest_request_block.min(*after_block_num);
+        }
+        let blocks = self.sync_expected_notes(lowest_request_block, &sync_tags).await?;
+
+        // An expected note has no metadata and thus no `NoteId`, so each returned note is matched
+        // to its request by rebuilding the id from the committed metadata. Only the blocks holding
+        // a match are kept: the rest hold notes under the same tag that answer no request.
+        let mut committed_notes_data = BTreeMap::new();
+        let mut matched_blocks = Vec::new();
+        for block in blocks {
+            let mut block_matched = false;
+            for (note_id, sync_note) in &block.notes {
+                let metadata = sync_note.committed.metadata();
+                let Some(commitment) = requested_commitments
+                    .iter()
+                    .find(|commitment| NoteId::new(**commitment, metadata) == *note_id)
+                else {
+                    continue;
+                };
+
+                committed_notes_data
+                    .insert(*commitment, (sync_note.clone(), block.block_header.clone()));
+                block_matched = true;
+            }
+
+            if block_matched {
+                matched_blocks.push(block);
             }
         }
-        let mut committed_notes_data =
-            self.sync_expected_notes(lowest_request_block, note_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(matched_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(&note_record.details_commitment())
+            let Some((
+                SyncedNote {
+                    committed: committed_note, attachments, ..
+                },
+                block_header,
+            )) = committed_notes_data.remove(&note_record.details_commitment())
             else {
                 note_records.push(note_record);
                 continue;
@@ -380,10 +429,6 @@ where
             // 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 metadata = *committed_note.metadata();
             let mut note_changed = note_record
                 .inclusion_proof_received(committed_note.inclusion_proof().clone(), metadata)?;
@@ -409,82 +454,121 @@ 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 every record in `note_records` whose nullifier is already on chain as consumed.
+    ///
+    /// The query starts at the lowest block that committed one of these notes, so it also covers a
+    /// note spent below the client's checkpoint. A sync only queries nullifiers from its own
+    /// checkpoint forward and would never revisit that block.
+    ///
+    /// 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?;
+
+        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) {
+                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<BTreeMap<NoteDetailsCommitment, SyncedNote>, ClientError> {
-        let sync_tags: BTreeSet<NoteTag> = expected_notes.iter().map(|(_, tag)| *tag).collect();
-
-        let mut matched_notes = BTreeMap::new();
+        sync_tags: &BTreeSet<NoteTag>,
+    ) -> Result<Vec<ResolvedSyncNotesBlock>, ClientError> {
         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
+        let mut blocks = self
             .rpc_api
             .sync_notes_with_content(
                 request_block_num,
                 current_block_num,
-                &sync_tags,
+                sync_tags,
                 NoteContentFetch::AttachmentsOnly,
             )
             .await
             .map_err(ClientError::RpcError)?;
 
-        for block in blocks {
+        blocks.retain_mut(|block| {
             if block.block_header.block_num() > current_block_num {
-                break;
+                return false;
             }
 
-            for sync_note in block.notes.into_values() {
-                let committed = &sync_note.committed;
+            // A note carries its own commit height in its inclusion proof, which is a separate
+            // field from the block header checked above. Authenticating the note later looks that
+            // height up in the partial MMR, so a height beyond our synced view has to be dropped
+            // here rather than trusted.
+            block
+                .notes
+                .retain(|_, sync_note| sync_note.committed.block_num() <= current_block_num);
 
-                // The note carries its own commit height in its inclusion proof, which is a
-                // separate field from the block header checked above. Authenticating the note later
-                // looks that height up in the partial MMR, so a height beyond our synced view has
-                // to be dropped here rather than trusted.
-                if committed.block_num() > current_block_num {
-                    continue;
-                }
-
-                let Some((commitment, _)) = expected_notes.iter().find(|(commitment, _)| {
-                    NoteId::new(*commitment, committed.metadata()) == *committed.note_id()
-                }) else {
-                    continue;
-                };
+            !block.notes.is_empty()
+        });
 
-                matched_notes.insert(*commitment, sync_note);
-            }
-        }
-
-        Ok(matched_notes)
+        Ok(blocks)
     }
 }
 
+// EXPECTED NOTE IMPORT
+// ================================================================================================
+
+/// A note to import: 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 NoteImportRequest = (Option<InputNoteRecord>, 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()
     {

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<Item = InputNoteRecord>,
+    ) {
+        for note in notes {
+            if self.input_notes.contains_key(&note.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