From cd8dec9f928c0fe7bc26295cce8d44c3ae43f041 Mon Sep 17 00:00:00 2001 From: KOVACS Krisztian Date: Mon, 17 Aug 2026 18:39:20 +0200 Subject: [PATCH 1/3] feat(rpc): add streaming account vault sync v2 --- crates/rpc/src/server/api.rs | 1 + .../src/server/api/sync_account_vault_v2.rs | 174 ++++++++++++++++++ crates/rpc/src/tests.rs | 101 +++++++++- crates/store/src/db/mod.rs | 37 +++- .../store/src/db/models/queries/accounts.rs | 72 +++++++- crates/store/src/db/tests.rs | 109 +++++++++++ crates/store/src/lib.rs | 37 +++- crates/store/src/state/view/sync.rs | 26 ++- docs/external/src/rpc/index.md | 16 +- docs/external/src/rpc/public-api.md | 7 +- proto/proto/rpc.proto | 16 ++ 11 files changed, 582 insertions(+), 14 deletions(-) create mode 100644 crates/rpc/src/server/api/sync_account_vault_v2.rs diff --git a/crates/rpc/src/server/api.rs b/crates/rpc/src/server/api.rs index ccac7f6e72..9ee962996d 100644 --- a/crates/rpc/src/server/api.rs +++ b/crates/rpc/src/server/api.rs @@ -90,6 +90,7 @@ mod submit_proven_tx_batch; mod subscription; mod sync_account_storage_maps; mod sync_account_vault; +mod sync_account_vault_v2; mod sync_chain_mmr; mod sync_notes; mod sync_nullifiers; diff --git a/crates/rpc/src/server/api/sync_account_vault_v2.rs b/crates/rpc/src/server/api/sync_account_vault_v2.rs new file mode 100644 index 0000000000..52ce0ab4e0 --- /dev/null +++ b/crates/rpc/src/server/api/sync_account_vault_v2.rs @@ -0,0 +1,174 @@ +use std::num::NonZeroUsize; +use std::ops::RangeInclusive; +use std::time::Duration; + +use miden_node_proto::decode::{read_account_id, read_block_range}; +use miden_node_proto::generated as proto; +use miden_node_store::{AccountVaultValue, AccountVaultValuesPage, StateView}; +use miden_node_utils::tracing::{miden_instrument, miden_span_record}; +use miden_protocol::Word; +use miden_protocol::account::AccountId; +use miden_protocol::block::BlockNumber; +use tokio::sync::mpsc; +use tokio::sync::mpsc::error::SendTimeoutError; +use tokio_stream::wrappers::ReceiverStream; +use tonic::Status; +use tracing::Instrument; + +use super::{ + RpcInvalidBlockRange, + RpcService, + database_error_to_status, + invalid_block_range_to_status, +}; +use crate::{COMPONENT, LOG_TARGET}; + +/// Database rows fetched per page. This bounds internal work and memory, not encoded response size. +const DB_PAGE_SIZE: NonZeroUsize = NonZeroUsize::new(256).unwrap(); +/// Stream items buffered before backpressure pauses the database producer. +const STREAM_BUFFER_SIZE: usize = 32; +/// Maximum time a stream producer waits for a stalled client to accept one update. +const SEND_TIMEOUT: Duration = Duration::from_secs(10); + +type Input = (AccountId, RangeInclusive); + +#[tonic::async_trait] +impl proto::server::rpc_api::SyncAccountVaultV2 for RpcService { + type Input = Input; + type Item = AccountVaultValue; + type ItemStream = ReceiverStream>; + + fn decode(request: proto::rpc::SyncAccountVaultV2Request) -> tonic::Result { + let account_id = + read_account_id::(request.account_id)?; + let range = read_block_range::(request.block_range, "SyncAccountVaultV2Request")?; + let block_range = range + .into_inclusive_range::() + .map_err(invalid_block_range_to_status)?; + + Ok((account_id, block_range)) + } + + fn encode(item: Self::Item) -> tonic::Result { + let vault_key: Word = item.vault_key.into(); + Ok(proto::rpc::AccountVaultUpdate { + vault_key: Some(vault_key.into()), + asset: item.asset.map(Into::into), + block_num: item.block_num.as_u32(), + }) + } + + #[miden_instrument( + target = COMPONENT, + name = "sync_account_vault_v2", + err, + )] + async fn handle( + &self, + (account_id, block_range): Self::Input, + _metadata: &tonic::metadata::MetadataMap, + _extensions: &tonic::codegen::http::Extensions, + ) -> tonic::Result { + miden_span_record!( + account.id = %account_id, + block_range.from = %block_range.start(), + block_range.to = %block_range.end(), + ); + + tracing::debug!(target: LOG_TARGET, "Streaming account vault updates"); + + if !account_id.is_public() { + return Err(Status::invalid_argument(format!("account {account_id} is not public"))); + } + + // Keep this view for the finite stream's lifetime. Besides fixing the chain-tip view used + // for validation, this pins the history generation so pruning cannot remove rows between + // internal database pages. Cancellation and the bounded send timeout release the view if + // the client stops consuming the stream. + let view = self.state.view(); + let first_page = view + .sync_account_vault_v2_page(account_id, block_range.clone(), None, DB_PAGE_SIZE) + .await + .map_err(|err| database_error_to_status(&err))?; + + // Reserve a slot for a terminal error so a full data buffer cannot turn a timeout or + // database failure into an apparently successful end-of-stream. + let (tx, rx) = mpsc::channel(STREAM_BUFFER_SIZE + 1); + let terminal_permit = tx + .clone() + .try_reserve_owned() + .expect("a newly created vault sync channel must have capacity"); + VaultSyncProducer { + view, + account_id, + block_range, + page: first_page, + tx, + terminal_permit: Some(terminal_permit), + } + .spawn(); + + Ok(ReceiverStream::new(rx)) + } +} + +struct VaultSyncProducer { + view: StateView, + account_id: AccountId, + block_range: RangeInclusive, + page: AccountVaultValuesPage, + tx: mpsc::Sender>, + terminal_permit: Option>>, +} + +impl VaultSyncProducer { + fn spawn(self) { + tokio::spawn(self.run().instrument(tracing::Span::current())); + } + + async fn run(mut self) { + loop { + let next_cursor = self.page.next_cursor.take(); + for value in std::mem::take(&mut self.page.values) { + match self.tx.send_timeout(Ok(value), SEND_TIMEOUT).await { + Ok(()) => {}, + Err(SendTimeoutError::Closed(_)) => return, + Err(SendTimeoutError::Timeout(_)) => { + self.send_terminal_error(Status::deadline_exceeded( + "account vault sync client stopped consuming updates", + )); + return; + }, + } + } + + let Some(cursor) = next_cursor else { + return; + }; + + self.page = match self + .view + .sync_account_vault_v2_page( + self.account_id, + self.block_range.clone(), + Some(cursor), + DB_PAGE_SIZE, + ) + .await + { + Ok(page) => page, + Err(err) => { + self.send_terminal_error(database_error_to_status(&err)); + return; + }, + }; + } + } + + fn send_terminal_error(&mut self, status: Status) { + self.terminal_permit + .take() + .expect("terminal permit is consumed at most once") + .send(Err(status)); + } +} diff --git a/crates/rpc/src/tests.rs b/crates/rpc/src/tests.rs index 9b5bfc4b70..a7947daa69 100644 --- a/crates/rpc/src/tests.rs +++ b/crates/rpc/src/tests.rs @@ -43,6 +43,12 @@ use miden_protocol::account::{ AccountUpdateDetails, AssetCallbackFlag, }; +use miden_protocol::asset::{Asset, FungibleAsset}; +use miden_protocol::block::BlockNumber; +use miden_protocol::testing::account_id::{ + ACCOUNT_ID_PUBLIC_FUNGIBLE_FAUCET, + ACCOUNT_ID_PUBLIC_FUNGIBLE_FAUCET_1, +}; use miden_protocol::testing::noop_auth_component::NoopAuthComponent; use miden_protocol::transaction::{ProvenTransaction, TxAccountUpdate}; use miden_protocol::utils::serde::Serializable; @@ -1190,12 +1196,16 @@ async fn get_limits_endpoint() { QueryParamNoteTagLimit::LIMIT ); - // SyncAccountVault and SyncAccountStorageMaps accept a singular account_id, not a repeated + // The account vault and storage-map endpoints accept a singular account_id, not a repeated // list, so they do not have list parameter limits. assert!( !limits.endpoints.contains_key("SyncAccountVault"), "SyncAccountVault should not have list parameter limits" ); + assert!( + !limits.endpoints.contains_key("SyncAccountVaultV2"), + "SyncAccountVaultV2 should not have list parameter limits" + ); assert!( !limits.endpoints.contains_key("SyncAccountStorageMaps"), "SyncAccountStorageMaps should not have list parameter limits" @@ -1341,6 +1351,15 @@ async fn sync_endpoints_reject_block_to_beyond_chain_tip() { .expect_err("sync_account_vault should reject block_to beyond chain tip"); assert_beyond_tip(&status, "sync_account_vault"); + let status = rpc_client + .sync_account_vault_v2(proto::rpc::SyncAccountVaultV2Request { + block_range: block_range(), + account_id: account_id(), + }) + .await + .expect_err("sync_account_vault_v2 should reject block_to beyond chain tip"); + assert_beyond_tip(&status, "sync_account_vault_v2"); + let status = rpc_client .sync_transactions(proto::rpc::SyncTransactionsRequest { block_range: block_range(), @@ -1350,3 +1369,83 @@ async fn sync_endpoints_reject_block_to_beyond_chain_tip() { .expect_err("sync_transactions should reject block_to beyond chain tip"); assert_beyond_tip(&status, "sync_transactions"); } + +#[tokio::test] +async fn sync_account_vault_v2_validates_requests_and_completes_empty_stream() { + let (mut rpc_client, _rpc_addr, _store) = start_rpc().await; + let public_account = AccountId::dummy( + [0; 15], + AccountIdVersion::Version1, + AccountType::Public, + AssetCallbackFlag::Disabled, + ); + + let status = rpc_client + .sync_account_vault_v2(proto::rpc::SyncAccountVaultV2Request { + block_range: None, + account_id: Some(public_account.into()), + }) + .await + .expect_err("sync_account_vault_v2 should require a block range"); + assert_eq!(status.code(), tonic::Code::InvalidArgument); + + let private_account = AccountId::dummy( + [1; 15], + AccountIdVersion::Version1, + AccountType::Private, + AssetCallbackFlag::Disabled, + ); + let status = rpc_client + .sync_account_vault_v2(proto::rpc::SyncAccountVaultV2Request { + block_range: Some(proto::rpc::BlockRange { block_from: 0, block_to: 0 }), + account_id: Some(private_account.into()), + }) + .await + .expect_err("sync_account_vault_v2 should reject private accounts"); + assert_eq!(status.code(), tonic::Code::InvalidArgument); + + let mut stream = rpc_client + .sync_account_vault_v2(proto::rpc::SyncAccountVaultV2Request { + block_range: Some(proto::rpc::BlockRange { block_from: 0, block_to: 0 }), + account_id: Some(public_account.into()), + }) + .await + .expect("sync_account_vault_v2 should accept a public account at the chain tip") + .into_inner(); + assert_eq!(stream.message().await.expect("stream should complete successfully"), None); +} + +#[tokio::test] +async fn sync_account_vault_v2_streams_squashed_updates() { + let (mut rpc_client, _rpc_addr, store) = start_rpc().await; + let account_id = AccountId::try_from(ACCOUNT_ID_PUBLIC_FUNGIBLE_FAUCET).unwrap(); + let other_faucet = AccountId::try_from(ACCOUNT_ID_PUBLIC_FUNGIBLE_FAUCET_1).unwrap(); + let asset_a = Asset::Fungible(FungibleAsset::new(account_id, 100).unwrap()); + let asset_b = Asset::Fungible(FungibleAsset::new(other_faucet, 200).unwrap()); + miden_node_store::test_support::seed_account_vault( + &store.data_directory_path().join("miden-store.sqlite3"), + account_id, + BlockNumber::GENESIS, + &[(asset_a.id(), Some(asset_a)), (asset_b.id(), Some(asset_b))], + ); + + let mut stream = rpc_client + .sync_account_vault_v2(proto::rpc::SyncAccountVaultV2Request { + block_range: Some(proto::rpc::BlockRange { block_from: 0, block_to: 0 }), + account_id: Some(account_id.into()), + }) + .await + .expect("sync_account_vault_v2 should return a stream") + .into_inner(); + + let mut assets = Vec::new(); + while let Some(update) = stream.message().await.expect("stream should complete successfully") { + assert_eq!(update.block_num, 0); + assets.push(Asset::try_from(update.asset.expect("seeded values are additions")).unwrap()); + } + assets.sort_by_key(Asset::id); + + let mut expected = vec![asset_a, asset_b]; + expected.sort_by_key(Asset::id); + assert_eq!(assets, expected); +} diff --git a/crates/store/src/db/mod.rs b/crates/store/src/db/mod.rs index 2155eeaf9c..684812470e 100644 --- a/crates/store/src/db/mod.rs +++ b/crates/store/src/db/mod.rs @@ -120,7 +120,7 @@ impl DerefMut for Db { /// Describes the value of an asset for an account ID at `block_num` specifically. /// /// If `asset` is `None`, the asset was removed. -#[derive(Debug, Clone)] +#[derive(Debug, Clone, PartialEq, Eq)] pub struct AccountVaultValue { pub block_num: BlockNumber, pub vault_key: AssetId, @@ -128,6 +128,20 @@ pub struct AccountVaultValue { pub asset: Option, } +/// Stable cursor used to read a squashed account-vault delta in bounded database pages. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct AccountVaultCursor { + pub(crate) block_num: BlockNumber, + pub(crate) vault_key: AssetId, +} + +/// A bounded page of squashed account-vault updates. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct AccountVaultValuesPage { + pub values: Vec, + pub next_cursor: Option, +} + impl AccountVaultValue { pub fn from_raw_row(row: (i64, Vec, Option>)) -> Result { let (block_num, vault_key, asset) = row; @@ -784,6 +798,27 @@ impl Db { .await } + /// Selects one final update per vault key changed in `block_range`. + pub async fn select_account_vault_updates_v2( + &self, + account_id: AccountId, + block_range: ScopedBlockRange, + cursor: Option, + page_size: NonZeroUsize, + ) -> Result { + let block_range = block_range.into_inner(); + self.transact("account vault sync v2", move |conn| { + queries::select_account_vault_updates_v2( + conn, + account_id, + block_range, + cursor, + page_size, + ) + }) + .await + } + /// Returns the script for a note by its root. pub async fn select_note_script_by_root(&self, root: Word) -> Result> { self.transact("note script by root", move |conn| { diff --git a/crates/store/src/db/models/queries/accounts.rs b/crates/store/src/db/models/queries/accounts.rs index 24d6442245..13a6ba1fb3 100644 --- a/crates/store/src/db/models/queries/accounts.rs +++ b/crates/store/src/db/models/queries/accounts.rs @@ -52,7 +52,7 @@ use crate::COMPONENT; use crate::db::models::conv::{SqlTypeConvert, nonce_to_raw_sql, raw_sql_to_nonce}; #[cfg(test)] use crate::db::models::vec_raw_try_into; -use crate::db::{AccountVaultValue, schema}; +use crate::db::{AccountVaultCursor, AccountVaultValue, AccountVaultValuesPage, schema}; use crate::errors::DatabaseError; mod at_block; @@ -575,6 +575,76 @@ pub(crate) fn select_account_vault_assets( Ok((last_block_included, values)) } +/// Selects a bounded page containing the final update at `block_range.end()` for every vault key +/// changed within the inclusive block range. +/// +/// Vault rows are valid in `[block_num, valid_until)`. Requiring `valid_until > block_to` removes +/// intermediate updates while retaining a historical value that was superseded after the target. +/// Results use the table's `(account_id, block_num, vault_key)` primary-key order so they can be +/// continued with a stable keyset cursor without response-size accounting. +pub(crate) fn select_account_vault_updates_v2( + conn: &mut SqliteConnection, + account_id: AccountId, + block_range: RangeInclusive, + cursor: Option, + page_size: NonZeroUsize, +) -> Result { + use schema::account_vault_assets as t; + + if !account_id.is_public() { + return Err(DatabaseError::AccountNotPublic(account_id)); + } + + if block_range.is_empty() { + return Err(DatabaseError::InvalidBlockRange { + from: *block_range.start(), + to: *block_range.end(), + }); + } + + let block_from = block_range.start().to_raw_sql(); + let block_to = block_range.end().to_raw_sql(); + let mut query = SelectDsl::select(t::table, (t::block_num, t::vault_key, t::asset)) + .filter(t::account_id.eq(account_id.to_bytes())) + .filter(t::block_num.ge(block_from)) + .filter(t::block_num.le(block_to)) + .filter(t::valid_until.gt(block_to)) + .into_boxed(); + + if let Some(cursor) = cursor { + let cursor_block = cursor.block_num.to_raw_sql(); + let cursor_key: Word = cursor.vault_key.into(); + query = query.filter( + t::block_num + .gt(cursor_block) + .or(t::block_num.eq(cursor_block).and(t::vault_key.gt(cursor_key.to_bytes()))), + ); + } + + let limit = page_size.get(); + let query_limit = i64::try_from(limit.saturating_add(1)).expect("page size fits within i64"); + let mut raw = query + .order((t::block_num.asc(), t::vault_key.asc())) + .limit(query_limit) + .load::<(i64, Vec, Option>)>(conn)?; + + let has_more = raw.len() > limit; + raw.truncate(limit); + let values = raw + .into_iter() + .map(AccountVaultValue::from_raw_row) + .collect::, DatabaseError>>()?; + let next_cursor = has_more.then(|| { + let last = values.last().expect("a page with more rows cannot be empty"); + AccountVaultCursor { + block_num: last.block_num, + vault_key: last.vault_key, + } + }); + + Ok(AccountVaultValuesPage { values, next_cursor }) +} + /// Query vault assets at a specific block by finding the most recent update for each `vault_key`. /// /// Selects, per vault key, the row whose validity interval covers `block_num`: diff --git a/crates/store/src/db/tests.rs b/crates/store/src/db/tests.rs index 9dad5cc135..9b04cc50ae 100644 --- a/crates/store/src/db/tests.rs +++ b/crates/store/src/db/tests.rs @@ -1,3 +1,4 @@ +use std::num::NonZeroUsize; use std::sync::{Arc, Mutex}; use assert_matches::assert_matches; @@ -459,6 +460,114 @@ fn sync_account_vault_basic_validation() { assert_eq!(vault_key_1_asset.unwrap().asset, Some(updated_fungible_asset_1)); } +#[test] +#[miden_node_test_macro::enable_logging] +fn sync_account_vault_v2_returns_one_target_value_per_changed_key() { + let mut conn = create_db(); + let account_id = AccountId::try_from(ACCOUNT_ID_PUBLIC_FUNGIBLE_FAUCET).unwrap(); + let blocks: Vec = (1..=6).map(BlockNumber::from).collect(); + + for block in &blocks { + create_block(&mut conn, *block); + queries::upsert_accounts( + &mut conn, + &[mock_block_account_update(account_id, 0)], + *block, + &queries::PrecomputedPublicAccountStates::new(), + ) + .unwrap(); + } + + let faucet_a = account_id; + let faucet_b = AccountId::try_from(ACCOUNT_ID_PUBLIC_FUNGIBLE_FAUCET_1).unwrap(); + let faucet_c = AccountId::try_from(ACCOUNT_ID_PUBLIC_FUNGIBLE_FAUCET_2).unwrap(); + let faucet_d = AccountId::try_from(ACCOUNT_ID_PUBLIC_FUNGIBLE_FAUCET_3).unwrap(); + + let asset_a_1 = Asset::Fungible(FungibleAsset::new(faucet_a, 100).unwrap()); + let asset_a_3 = Asset::Fungible(FungibleAsset::new(faucet_a, 300).unwrap()); + let asset_a_6 = Asset::Fungible(FungibleAsset::new(faucet_a, 600).unwrap()); + let asset_b_2 = Asset::Fungible(FungibleAsset::new(faucet_b, 200).unwrap()); + let asset_c_1 = Asset::Fungible(FungibleAsset::new(faucet_c, 100).unwrap()); + let asset_d_4 = Asset::Fungible(FungibleAsset::new(faucet_d, 400).unwrap()); + + for (block, asset) in [ + (blocks[0], asset_a_1), + (blocks[0], asset_c_1), + (blocks[1], asset_b_2), + (blocks[2], asset_a_3), + (blocks[3], asset_d_4), + ] { + queries::insert_account_vault_asset(&mut conn, account_id, block, asset.id(), Some(asset)) + .unwrap(); + } + + // Remove D at the inclusive target and update A after the target. The V2 query must return D's + // tombstone and A's block-3 value, whose validity interval is finite but covers block 5. + queries::insert_account_vault_asset(&mut conn, account_id, blocks[4], asset_d_4.id(), None) + .unwrap(); + queries::insert_account_vault_asset( + &mut conn, + account_id, + blocks[5], + asset_a_6.id(), + Some(asset_a_6), + ) + .unwrap(); + + let range = blocks[1]..=blocks[4]; + let page_size = NonZeroUsize::new(1).unwrap(); + let mut cursor = None; + let mut values = Vec::new(); + loop { + let page = queries::select_account_vault_updates_v2( + &mut conn, + account_id, + range.clone(), + cursor, + page_size, + ) + .unwrap(); + values.extend(page.values); + let Some(next_cursor) = page.next_cursor else { + break; + }; + cursor = Some(next_cursor); + } + + assert_eq!(values.len(), 3); + assert_eq!(values[0].block_num, blocks[1]); + assert_eq!(values[0].vault_key, asset_b_2.id()); + assert_eq!(values[0].asset, Some(asset_b_2)); + assert_eq!(values[1].block_num, blocks[2]); + assert_eq!(values[1].vault_key, asset_a_3.id()); + assert_eq!(values[1].asset, Some(asset_a_3)); + assert_eq!(values[2].block_num, blocks[4]); + assert_eq!(values[2].vault_key, asset_d_4.id()); + assert_eq!(values[2].asset, None); + + // C changed before the inclusive range and A's block-1/block-6 values lie outside it. + assert!(values.iter().all(|value| value.vault_key != asset_c_1.id())); + + let invalid = queries::select_account_vault_updates_v2( + &mut conn, + account_id, + blocks[4]..=blocks[1], + None, + page_size, + ); + assert_matches!(invalid, Err(DatabaseError::InvalidBlockRange { .. })); + + let private_account = AccountId::try_from(ACCOUNT_ID_PRIVATE_SENDER).unwrap(); + let private = queries::select_account_vault_updates_v2( + &mut conn, + private_account, + range, + None, + page_size, + ); + assert_matches!(private, Err(DatabaseError::AccountNotPublic(id)) if id == private_account); +} + #[test] #[miden_node_test_macro::enable_logging] fn select_nullifiers_by_prefix_works() { diff --git a/crates/store/src/lib.rs b/crates/store/src/lib.rs index ec40d3fe3b..2d27b70395 100644 --- a/crates/store/src/lib.rs +++ b/crates/store/src/lib.rs @@ -16,7 +16,9 @@ pub use data_directory::DataDirectory; pub use db::models::conv::SqlTypeConvert; pub use db::models::queries::StorageMapValuesPage; pub use db::{ + AccountVaultCursor, AccountVaultValue, + AccountVaultValuesPage, DatabaseOptions, Db, NoteRecord, @@ -73,9 +75,14 @@ pub mod test_support { use diesel::prelude::*; use miden_protocol::Word; use miden_protocol::account::AccountId; + use miden_protocol::asset::{Asset, AssetId}; use miden_protocol::block::BlockNumber; - use crate::db::models::queries::{AccountRowInsert, NetworkAccountType}; + use crate::db::models::queries::{ + AccountRowInsert, + NetworkAccountType, + insert_account_vault_asset, + }; use crate::db::schema; /// Opens a fresh connection to the store's SQLite database and inserts a private @@ -101,6 +108,34 @@ pub mod test_support { .execute(&mut conn) .expect("insert network account row"); } + + /// Inserts a public account row and vault values for downstream RPC integration tests. + pub fn seed_account_vault( + db_path: &Path, + account_id: AccountId, + block_num: BlockNumber, + values: &[(AssetId, Option)], + ) { + let mut conn = SqliteConnection::establish(db_path.to_str().expect("db path is utf-8")) + .expect("connect to store sqlite"); + + let row = AccountRowInsert::new_private( + account_id, + NetworkAccountType::None, + Word::default(), + block_num, + block_num, + ); + diesel::insert_into(schema::accounts::table) + .values(&row) + .execute(&mut conn) + .expect("insert public test account row"); + + for (vault_key, asset) in values { + insert_account_vault_asset(&mut conn, account_id, block_num, *vault_key, *asset) + .expect("insert test account vault value"); + } + } } // CONSTANTS diff --git a/crates/store/src/state/view/sync.rs b/crates/store/src/state/view/sync.rs index 603c078beb..7e454ce6ac 100644 --- a/crates/store/src/state/view/sync.rs +++ b/crates/store/src/state/view/sync.rs @@ -1,3 +1,4 @@ +use std::num::NonZeroUsize; use std::ops::RangeInclusive; use miden_node_utils::tracing::miden_instrument; @@ -8,7 +9,13 @@ use miden_protocol::crypto::merkle::mmr::{Forest, MmrDelta, MmrProof}; use super::StateView; use crate::COMPONENT; use crate::db::models::queries::StorageMapValuesPage; -use crate::db::{AccountVaultValue, NoteSyncUpdate, NullifierInfo}; +use crate::db::{ + AccountVaultCursor, + AccountVaultValue, + AccountVaultValuesPage, + NoteSyncUpdate, + NullifierInfo, +}; use crate::errors::{DatabaseError, NoteSyncError, StateSyncError}; // STATE SYNCHRONIZATION ENDPOINTS @@ -169,6 +176,23 @@ impl StateView { self.db.get_account_vault_sync(account_id, block_range).await } + /// Returns a bounded page with one final update per vault key changed in a block range. + /// + /// Returns [`RangeBeyondTip`](crate::errors::RangeBeyondTip) if the range extends beyond this + /// view's chain tip. + pub async fn sync_account_vault_v2_page( + &self, + account_id: AccountId, + block_range: RangeInclusive, + cursor: Option, + page_size: NonZeroUsize, + ) -> Result { + let block_range = self.scope_range(block_range)?; + self.db + .select_account_vault_updates_v2(account_id, block_range, cursor, page_size) + .await + } + /// Returns storage map values for syncing within a block range. /// /// Returns [`RangeBeyondTip`](crate::errors::RangeBeyondTip) if the range extends beyond this diff --git a/docs/external/src/rpc/index.md b/docs/external/src/rpc/index.md index 0da68d6a8a..49bfa19804 100644 --- a/docs/external/src/rpc/index.md +++ b/docs/external/src/rpc/index.md @@ -46,14 +46,14 @@ The RPC server supports: ## Endpoint Groups -| Group | Methods | -| ---------------------- | --------------------------------------------------------------------------------------------------------------- | -| Status and limits | `Status`, `GetLimits` | -| State queries | `GetAccount`, `GetBlockByNumber`, `GetBlockHeaderByNumber`, `GetNotesById`, `GetNoteScriptByRoot` | -| Transaction submission | `GetTransactionEncryptionKey`, `SubmitProvenTx`, `SubmitProvenTxBatch` | -| State synchronization | `SyncTransactions`, `SyncNotes`, `SyncNullifiers`, `SyncAccountVault`, `SyncAccountStorageMaps`, `SyncChainMmr` | -| Block streaming | `BlockSubscription`, `ProofSubscription` | -| Network note debugging | `GetNetworkNoteStatus` | +| Group | Methods | +| ---------------------- | ------------------------------------------------------------------------------------------------------------------------------------- | +| Status and limits | `Status`, `GetLimits` | +| State queries | `GetAccount`, `GetBlockByNumber`, `GetBlockHeaderByNumber`, `GetNotesById`, `GetNoteScriptByRoot` | +| Transaction submission | `GetTransactionEncryptionKey`, `SubmitProvenTx`, `SubmitProvenTxBatch` | +| State synchronization | `SyncTransactions`, `SyncNotes`, `SyncNullifiers`, `SyncAccountVault`, `SyncAccountVaultV2`, `SyncAccountStorageMaps`, `SyncChainMmr` | +| Block streaming | `BlockSubscription`, `ProofSubscription` | +| Network note debugging | `GetNetworkNoteStatus` | See [Public RPC](/rpc/public-api) for endpoint summaries, [Subscriptions](/rpc/subscriptions) for stream semantics, and [Errors and Limits](/rpc/errors-and-limits) for request limits, content negotiation, and method-specific error codes. diff --git a/docs/external/src/rpc/public-api.md b/docs/external/src/rpc/public-api.md index f2c73c72e4..be2543e768 100644 --- a/docs/external/src/rpc/public-api.md +++ b/docs/external/src/rpc/public-api.md @@ -71,7 +71,8 @@ codes returned in gRPC status details. | `SyncTransactions` | Returns transaction records for specified accounts within a block range. | | `SyncNotes` | Returns note metadata and inclusion proofs for matching note tags within a block range. | | `SyncNullifiers` | Returns nullifiers matching specified 16-bit prefixes within a block range. | -| `SyncAccountVault` | Returns public account vault updates within a block range. | +| `SyncAccountVault` | Returns historical public account vault updates within a block range. | +| `SyncAccountVaultV2` | Streams one target-state vault update per key changed within an inclusive block range. | | `SyncAccountStorageMaps` | Returns public account storage map updates within a block range. | | `SyncChainMmr` | Returns MMR delta information needed to synchronize the chain MMR. | @@ -84,6 +85,10 @@ Use `GetLimits` to discover the maximum request sizes accepted by the node befor | `BlockSubscription` | Streams committed blocks from `block_from`, replaying history before live blocks. | | `ProofSubscription` | Streams block proofs from `block_from`, replaying existing proofs before live proofs. | +`SyncAccountVaultV2` is a finite server stream. A client whose state includes block `C` and which is synchronizing to +block `N` requests the inclusive range `[C + 1, N]`. An OK end-of-stream marks the result complete; a non-OK termination +must be discarded and retried. + These streams are the primary mechanism full nodes use to replicate chain data from an upstream source. They are also useful for indexers, explorers, and other services that need an append-only view of network progress. diff --git a/proto/proto/rpc.proto b/proto/proto/rpc.proto index 66f9468f45..faea6a1ad8 100644 --- a/proto/proto/rpc.proto +++ b/proto/proto/rpc.proto @@ -88,6 +88,9 @@ service Api { // Returns account vault updates for specified account within a block range. rpc SyncAccountVault(SyncAccountVaultRequest) returns (SyncAccountVaultResponse) {} + // Streams the final account vault value for every key changed within a block range. + rpc SyncAccountVaultV2(SyncAccountVaultV2Request) returns (stream AccountVaultUpdate) {} + // Returns storage map updates for specified account and storage slots within a block range. rpc SyncAccountStorageMaps(SyncAccountStorageMapsRequest) returns (SyncAccountStorageMapsResponse) {} @@ -460,6 +463,19 @@ message SyncAccountVaultRequest { account.AccountId account_id = 2; } +// Account vault synchronization request with squashed, server-streaming responses. +message SyncAccountVaultV2Request { + // Inclusive range of vault updates to consider. + // + // A client whose state includes block C and which is synchronizing to block N should request + // [C + 1, N]. The server emits at most one update per vault key: the value of that key at + // `block_to`, provided the key changed within this range. + BlockRange block_range = 1; + + // Public account whose asset vault should be synchronized. + account.AccountId account_id = 2; +} + message SyncAccountVaultResponse { // Pagination information. PaginationInfo pagination_info = 1; From 9d757fd06d4d8d93dea915bf03e1f8f536b19151 Mon Sep 17 00:00:00 2001 From: KOVACS Krisztian Date: Tue, 18 Aug 2026 11:46:24 +0200 Subject: [PATCH 2/3] fix(rpc): fix review comments Validate target block is in the retained account-history window and don't pin a view for the lifetime of the response stream: return a BlockPruned error instead so that the client can recover. --- crates/rpc/src/server/api.rs | 15 ++- .../src/server/api/sync_account_vault_v2.rs | 26 +++-- .../store/src/db/models/queries/accounts.rs | 24 +++- crates/store/src/db/tests.rs | 105 ++++++++++++++++++ crates/store/src/errors.rs | 5 + crates/store/src/state/view/sync.rs | 3 +- docs/external/src/rpc/public-api.md | 4 +- proto/proto/rpc.proto | 3 +- 8 files changed, 168 insertions(+), 17 deletions(-) diff --git a/crates/rpc/src/server/api.rs b/crates/rpc/src/server/api.rs index 9ee962996d..9272114ff2 100644 --- a/crates/rpc/src/server/api.rs +++ b/crates/rpc/src/server/api.rs @@ -292,7 +292,9 @@ fn database_error_to_status(err: &DatabaseError) -> Status { | DatabaseError::AccountsNotFoundInDb(_) | DatabaseError::AccountNotPublic(_) => Status::not_found(message), DatabaseError::TransactionPageExceedsPayloadLimit { .. } => Status::out_of_range(message), - DatabaseError::RangeBeyondTip(_) => Status::invalid_argument(message), + DatabaseError::RangeBeyondTip(_) | DatabaseError::BlockPruned { .. } => { + Status::invalid_argument(message) + }, _ => Status::internal(message), } } @@ -356,6 +358,7 @@ static RPC_LIMITS: LazyLock = LazyLock::new(|| { #[cfg(test)] mod tests { use miden_node_proto::generated::server::rpc_api::GetLimits; + use miden_protocol::block::BlockNumber; use super::*; @@ -363,4 +366,14 @@ mod tests { fn get_limits_decodes_unit_request() { assert_eq!(RpcService::decode(()).unwrap(), ()); } + + #[test] + fn block_pruned_database_error_is_invalid_argument() { + let status = database_error_to_status(&DatabaseError::BlockPruned { + block_num: BlockNumber::from(49), + oldest_available: BlockNumber::from(50), + }); + + assert_eq!(status.code(), tonic::Code::InvalidArgument); + } } diff --git a/crates/rpc/src/server/api/sync_account_vault_v2.rs b/crates/rpc/src/server/api/sync_account_vault_v2.rs index 52ce0ab4e0..ac0d667ed4 100644 --- a/crates/rpc/src/server/api/sync_account_vault_v2.rs +++ b/crates/rpc/src/server/api/sync_account_vault_v2.rs @@ -1,10 +1,11 @@ use std::num::NonZeroUsize; use std::ops::RangeInclusive; +use std::sync::Arc; use std::time::Duration; use miden_node_proto::decode::{read_account_id, read_block_range}; use miden_node_proto::generated as proto; -use miden_node_store::{AccountVaultValue, AccountVaultValuesPage, StateView}; +use miden_node_store::{AccountVaultValue, AccountVaultValuesPage, State}; use miden_node_utils::tracing::{miden_instrument, miden_span_record}; use miden_protocol::Word; use miden_protocol::account::AccountId; @@ -30,11 +31,11 @@ const STREAM_BUFFER_SIZE: usize = 32; /// Maximum time a stream producer waits for a stalled client to accept one update. const SEND_TIMEOUT: Duration = Duration::from_secs(10); -type Input = (AccountId, RangeInclusive); +type RequestInput = (AccountId, RangeInclusive); #[tonic::async_trait] impl proto::server::rpc_api::SyncAccountVaultV2 for RpcService { - type Input = Input; + type Input = RequestInput; type Item = AccountVaultValue; type ItemStream = ReceiverStream>; @@ -81,12 +82,12 @@ impl proto::server::rpc_api::SyncAccountVaultV2 for RpcService { return Err(Status::invalid_argument(format!("account {account_id} is not public"))); } - // Keep this view for the finite stream's lifetime. Besides fixing the chain-tip view used - // for validation, this pins the history generation so pruning cannot remove rows between - // internal database pages. Cancellation and the bounded send timeout release the view if - // the client stops consuming the stream. - let view = self.state.view(); - let first_page = view + // Fetch the first page before establishing the stream so request validation failures are + // returned as the initial RPC status. Each page uses its own short-lived state view; the + // stream must not let a client pin a snapshot generation for its entire lifetime. + let first_page = self + .state + .view() .sync_account_vault_v2_page(account_id, block_range.clone(), None, DB_PAGE_SIZE) .await .map_err(|err| database_error_to_status(&err))?; @@ -99,7 +100,7 @@ impl proto::server::rpc_api::SyncAccountVaultV2 for RpcService { .try_reserve_owned() .expect("a newly created vault sync channel must have capacity"); VaultSyncProducer { - view, + state: Arc::clone(&self.state), account_id, block_range, page: first_page, @@ -113,7 +114,7 @@ impl proto::server::rpc_api::SyncAccountVaultV2 for RpcService { } struct VaultSyncProducer { - view: StateView, + state: Arc, account_id: AccountId, block_range: RangeInclusive, page: AccountVaultValuesPage, @@ -147,7 +148,8 @@ impl VaultSyncProducer { }; self.page = match self - .view + .state + .view() .sync_account_vault_v2_page( self.account_id, self.block_range.clone(), diff --git a/crates/store/src/db/models/queries/accounts.rs b/crates/store/src/db/models/queries/accounts.rs index 13a6ba1fb3..4127e5eb51 100644 --- a/crates/store/src/db/models/queries/accounts.rs +++ b/crates/store/src/db/models/queries/accounts.rs @@ -2,6 +2,7 @@ use std::collections::{BTreeMap, HashMap, HashSet}; use std::num::NonZeroUsize; use std::ops::RangeInclusive; +use diesel::dsl::max; use diesel::prelude::{Queryable, QueryableByName}; use diesel::query_dsl::methods::SelectDsl; use diesel::sqlite::Sqlite; @@ -602,8 +603,9 @@ pub(crate) fn select_account_vault_updates_v2( }); } + let target_block = *block_range.end(); let block_from = block_range.start().to_raw_sql(); - let block_to = block_range.end().to_raw_sql(); + let block_to = target_block.to_raw_sql(); let mut query = SelectDsl::select(t::table, (t::block_num, t::vault_key, t::asset)) .filter(t::account_id.eq(account_id.to_bytes())) .filter(t::block_num.ge(block_from)) @@ -628,6 +630,26 @@ pub(crate) fn select_account_vault_updates_v2( .limit(query_limit) .load::<(i64, Vec, Option>)>(conn)?; + // Check the retention horizon after reading the page, within the same transaction. This ensures + // the page and chain tip come from one SQLite snapshot: if pruning has already made the target + // incomplete, discard the page instead of returning an apparently complete delta. + let chain_tip = + SelectDsl::select(schema::block_headers::table, max(schema::block_headers::block_num)) + .get_result::>(conn)? + .ok_or_else(|| { + DatabaseError::DataCorrupted("block headers table is empty".to_owned()) + })?; + let chain_tip = BlockNumber::from_raw_sql(chain_tip)?; + let oldest_available = chain_tip + .checked_sub(HISTORICAL_BLOCK_RETENTION) + .unwrap_or(BlockNumber::GENESIS); + if target_block < oldest_available { + return Err(DatabaseError::BlockPruned { + block_num: target_block, + oldest_available, + }); + } + let has_more = raw.len() > limit; raw.truncate(limit); let values = raw diff --git a/crates/store/src/db/tests.rs b/crates/store/src/db/tests.rs index 9b04cc50ae..a9eca29941 100644 --- a/crates/store/src/db/tests.rs +++ b/crates/store/src/db/tests.rs @@ -568,6 +568,111 @@ fn sync_account_vault_v2_returns_one_target_value_per_changed_key() { assert_matches!(private, Err(DatabaseError::AccountNotPublic(id)) if id == private_account); } +#[test] +#[miden_node_test_macro::enable_logging] +fn sync_account_vault_v2_rejects_targets_below_pruning_horizon_between_pages() { + let mut conn = create_db(); + let account_id = AccountId::try_from(ACCOUNT_ID_PUBLIC_FUNGIBLE_FAUCET).unwrap(); + let other_faucet = AccountId::try_from(ACCOUNT_ID_PUBLIC_FUNGIBLE_FAUCET_1).unwrap(); + let target = BlockNumber::from(5); + + for block in 1..=target.as_u32() { + let block = BlockNumber::from(block); + create_block(&mut conn, block); + queries::upsert_accounts( + &mut conn, + &[mock_block_account_update(account_id, 0)], + block, + &queries::PrecomputedPublicAccountStates::new(), + ) + .unwrap(); + } + + let asset_a_at_target = Asset::Fungible(FungibleAsset::new(account_id, 300).unwrap()); + let asset_b = Asset::Fungible(FungibleAsset::new(other_faucet, 200).unwrap()); + queries::insert_account_vault_asset( + &mut conn, + account_id, + BlockNumber::from(2), + asset_b.id(), + Some(asset_b), + ) + .unwrap(); + queries::insert_account_vault_asset( + &mut conn, + account_id, + BlockNumber::from(3), + asset_a_at_target.id(), + Some(asset_a_at_target), + ) + .unwrap(); + + let page_size = NonZeroUsize::new(1).unwrap(); + let first_page = queries::select_account_vault_updates_v2( + &mut conn, + account_id, + BlockNumber::from(1)..=target, + None, + page_size, + ) + .unwrap(); + assert_eq!(first_page.values.len(), 1); + let cursor = first_page.next_cursor.expect("two updates require another page"); + + // Supersede A immediately after the target, then advance far enough that the target falls one + // block below the pruning horizon. Pruning removes A's value at the target, so the next page + // must fail instead of silently omitting it. + let oldest_available = target + 1; + let chain_tip = oldest_available + HISTORICAL_BLOCK_RETENTION; + for block in oldest_available.as_u32()..=chain_tip.as_u32() { + let block = BlockNumber::from(block); + create_block(&mut conn, block); + queries::upsert_accounts( + &mut conn, + &[mock_block_account_update(account_id, 0)], + block, + &queries::PrecomputedPublicAccountStates::new(), + ) + .unwrap(); + } + let asset_a_after_target = Asset::Fungible(FungibleAsset::new(account_id, 600).unwrap()); + queries::insert_account_vault_asset( + &mut conn, + account_id, + oldest_available, + asset_a_after_target.id(), + Some(asset_a_after_target), + ) + .unwrap(); + queries::prune_history(&mut conn, chain_tip).unwrap(); + + let next_page = queries::select_account_vault_updates_v2( + &mut conn, + account_id, + BlockNumber::from(1)..=target, + Some(cursor), + page_size, + ); + assert_matches!( + next_page, + Err(DatabaseError::BlockPruned { + block_num, + oldest_available: cutoff, + }) if block_num == target && cutoff == oldest_available + ); + + // The cutoff block itself remains queryable because pruning keeps rows whose validity extends + // beyond it. + let boundary = queries::select_account_vault_updates_v2( + &mut conn, + account_id, + BlockNumber::from(1)..=oldest_available, + None, + page_size, + ); + assert!(boundary.is_ok(), "the pruning cutoff should remain queryable"); +} + #[test] #[miden_node_test_macro::enable_logging] fn select_nullifiers_by_prefix_works() { diff --git a/crates/store/src/errors.rs b/crates/store/src/errors.rs index fb1f41843b..8a730ffbda 100644 --- a/crates/store/src/errors.rs +++ b/crates/store/src/errors.rs @@ -96,6 +96,11 @@ pub enum DatabaseError { AccountNotPublic(AccountId), #[error("invalid block parameters: block_from ({from}) > block_to ({to})")] InvalidBlockRange { from: BlockNumber, to: BlockNumber }, + #[error("block {block_num} has been pruned; the oldest available block is {oldest_available}")] + BlockPruned { + block_num: BlockNumber, + oldest_available: BlockNumber, + }, #[error( "transactions for block {block_num} would exceed maximum response size, \ use a stricter filter to reduce the number of transactions returned" diff --git a/crates/store/src/state/view/sync.rs b/crates/store/src/state/view/sync.rs index 7e454ce6ac..37b0dca9e9 100644 --- a/crates/store/src/state/view/sync.rs +++ b/crates/store/src/state/view/sync.rs @@ -179,7 +179,8 @@ impl StateView { /// Returns a bounded page with one final update per vault key changed in a block range. /// /// Returns [`RangeBeyondTip`](crate::errors::RangeBeyondTip) if the range extends beyond this - /// view's chain tip. + /// view's chain tip. Returns [`DatabaseError::BlockPruned`] if the range targets a block older + /// than the retained account history. pub async fn sync_account_vault_v2_page( &self, account_id: AccountId, diff --git a/docs/external/src/rpc/public-api.md b/docs/external/src/rpc/public-api.md index be2543e768..20f11616db 100644 --- a/docs/external/src/rpc/public-api.md +++ b/docs/external/src/rpc/public-api.md @@ -87,7 +87,9 @@ Use `GetLimits` to discover the maximum request sizes accepted by the node befor `SyncAccountVaultV2` is a finite server stream. A client whose state includes block `C` and which is synchronizing to block `N` requests the inclusive range `[C + 1, N]`. An OK end-of-stream marks the result complete; a non-OK termination -must be discarded and retried. +must be discarded and retried. The target `N` must remain within the server's retained account-history window, but `C` +may be older. If the target crosses the pruning horizon before all pages are read, the server terminates the stream with +`INVALID_ARGUMENT`; retry against a newer target. These streams are the primary mechanism full nodes use to replicate chain data from an upstream source. They are also useful for indexers, explorers, and other services that need an append-only view of network progress. diff --git a/proto/proto/rpc.proto b/proto/proto/rpc.proto index faea6a1ad8..790797c3e9 100644 --- a/proto/proto/rpc.proto +++ b/proto/proto/rpc.proto @@ -469,7 +469,8 @@ message SyncAccountVaultV2Request { // // A client whose state includes block C and which is synchronizing to block N should request // [C + 1, N]. The server emits at most one update per vault key: the value of that key at - // `block_to`, provided the key changed within this range. + // `block_to`, provided the key changed within this range. `block_to` must be within the + // server's retained account-history window; `block_from` may be older. BlockRange block_range = 1; // Public account whose asset vault should be synchronized. From 05d7f82eacf825d1a0ce76332d0e5e69595665e7 Mon Sep 17 00:00:00 2001 From: KOVACS Krisztian Date: Wed, 19 Aug 2026 12:09:01 +0200 Subject: [PATCH 3/3] fix(store): move retention horizon check before page load --- .../store/src/db/models/queries/accounts.rs | 41 ++++++++++--------- 1 file changed, 21 insertions(+), 20 deletions(-) diff --git a/crates/store/src/db/models/queries/accounts.rs b/crates/store/src/db/models/queries/accounts.rs index 95e4ed7396..7ef06e62ae 100644 --- a/crates/store/src/db/models/queries/accounts.rs +++ b/crates/store/src/db/models/queries/accounts.rs @@ -606,6 +606,27 @@ pub(crate) fn select_account_vault_updates_v2( let target_block = *block_range.end(); let block_from = block_range.start().to_raw_sql(); let block_to = target_block.to_raw_sql(); + + // Check the retention horizon before loading the page. This read establishes the SQLite + // transaction's snapshot, so the page query below observes the same chain tip and pruning + // state. Rejecting here avoids loading a page that would be discarded as incomplete. + let chain_tip = + SelectDsl::select(schema::block_headers::table, max(schema::block_headers::block_num)) + .get_result::>(conn)? + .ok_or_else(|| { + DatabaseError::DataCorrupted("block headers table is empty".to_owned()) + })?; + let chain_tip = BlockNumber::from_raw_sql(chain_tip)?; + let oldest_available = chain_tip + .checked_sub(HISTORICAL_BLOCK_RETENTION) + .unwrap_or(BlockNumber::GENESIS); + if target_block < oldest_available { + return Err(DatabaseError::BlockPruned { + block_num: target_block, + oldest_available, + }); + } + let mut query = SelectDsl::select(t::table, (t::block_num, t::vault_key, t::asset)) .filter(t::account_id.eq(account_id.to_bytes())) .filter(t::block_num.ge(block_from)) @@ -630,26 +651,6 @@ pub(crate) fn select_account_vault_updates_v2( .limit(query_limit) .load::<(i64, Vec, Option>)>(conn)?; - // Check the retention horizon after reading the page, within the same transaction. This ensures - // the page and chain tip come from one SQLite snapshot: if pruning has already made the target - // incomplete, discard the page instead of returning an apparently complete delta. - let chain_tip = - SelectDsl::select(schema::block_headers::table, max(schema::block_headers::block_num)) - .get_result::>(conn)? - .ok_or_else(|| { - DatabaseError::DataCorrupted("block headers table is empty".to_owned()) - })?; - let chain_tip = BlockNumber::from_raw_sql(chain_tip)?; - let oldest_available = chain_tip - .checked_sub(HISTORICAL_BLOCK_RETENTION) - .unwrap_or(BlockNumber::GENESIS); - if target_block < oldest_available { - return Err(DatabaseError::BlockPruned { - block_num: target_block, - oldest_available, - }); - } - let has_more = raw.len() > limit; raw.truncate(limit); let values = raw