pub mod errors;
pub mod generated;
#[cfg(feature = "tonic")]
pub mod grpc;
use alloc::boxed::Box;
use alloc::collections::{BTreeMap, BTreeSet};
use alloc::string::String;
use alloc::sync::Arc;
use alloc::vec::Vec;
use miden_protocol::address::Address;
use miden_protocol::block::BlockNumber;
use miden_protocol::note::{
Note,
NoteDetails,
NoteDetailsCommitment,
NoteHeader,
NoteId,
NoteInclusionProof,
NoteTag,
};
use miden_protocol::utils::serde::Serializable;
use miden_tx::auth::TransactionAuthenticator;
use miden_tx::utils::serde::{
ByteReader,
ByteWriter,
Deserializable,
DeserializationError,
SliceReader,
};
pub use self::errors::NoteTransportError;
use crate::note::{NoteFile, NoteSyncHint};
use crate::store::{NoteFilter, SettingScope};
use crate::sync::NoteTagSource;
use crate::{Client, ClientError};
pub const NOTE_TRANSPORT_MAINNET_ENDPOINT: &str = "https://transport.mainnet.miden.io";
pub const NOTE_TRANSPORT_TESTNET_ENDPOINT: &str = "https://transport.miden.io";
pub const NOTE_TRANSPORT_DEVNET_ENDPOINT: &str = "https://transport.devnet.miden.io";
#[deprecated(since = "0.17.1", note = "note transport does not read this setting")]
pub const NOTE_TRANSPORT_CURSOR_STORE_SETTING: &str = "note_transport_cursor";
#[deprecated(
since = "0.18.0",
note = "note transport stores one cursor under NOTE_TRANSPORT_SYNC_STATE_KEY"
)]
pub const NOTE_TRANSPORT_CURSORS_KEY: &str = "note_transport_cursors";
pub const NOTE_TRANSPORT_SYNC_STATE_KEY: &str = "note_transport_sync_state";
const MAX_NOTE_TAGS_PER_TRANSPORT_REQUEST: usize = 128;
#[deprecated(since = "0.17.1", note = "note transport no longer keeps per-tag backfill state")]
pub const NOTE_TRANSPORT_COVERED_TAGS_KEY: &str = "note_transport_covered_tags";
impl<AUTH> Client<AUTH> {
pub const MAX_NOTE_TAGS_PER_TRANSPORT_REQUEST: usize = MAX_NOTE_TAGS_PER_TRANSPORT_REQUEST;
#[deprecated(
since = "0.17.1",
note = "use MAX_NOTE_TAGS_PER_TRANSPORT_REQUEST; account tags are no longer limited"
)]
pub const MAX_ACCOUNT_TAGS: usize = Self::MAX_NOTE_TAGS_PER_TRANSPORT_REQUEST;
pub fn is_note_transport_enabled(&self) -> bool {
self.note_transport_api.is_some()
}
pub(crate) fn get_note_transport_api(
&self,
) -> Result<Arc<dyn NoteTransportClient>, NoteTransportError> {
self.note_transport_api.clone().ok_or(NoteTransportError::Disabled)
}
pub async fn send_private_note_with_proof(
&mut self,
note: Note,
address: &Address,
inclusion_proof: NoteInclusionProof,
) -> Result<(), ClientError> {
let api = self.get_note_transport_api()?;
let _ = address;
api.send_note_with_proof(TransportNote::from(note), inclusion_proof).await?;
Ok(())
}
#[allow(deprecated)]
async fn load_note_transport_sync_state(
&self,
) -> Result<Option<NoteTransportSyncState>, ClientError> {
let get = |key: &str| self.store.get_setting(SettingScope::Client, key.into());
let state = if let Some(bytes) = get(NOTE_TRANSPORT_SYNC_STATE_KEY).await? {
NoteTransportSyncState::read_from_bytes(&bytes).map(Some)
} else if let Some(bytes) = get(NOTE_TRANSPORT_CURSORS_KEY).await? {
BTreeMap::<NoteTag, Option<NoteTransportCursor>>::read_from_bytes(&bytes)
.map(migrate_tag_cursors)
} else {
Ok(None)
};
Ok(state
.inspect_err(|err| tracing::warn!(?err, "resetting unreadable note transport state"))
.unwrap_or_default())
}
}
impl<AUTH> Client<AUTH>
where
AUTH: TransactionAuthenticator + Sync + 'static,
{
#[deprecated(since = "0.17.1", note = "note transport no longer performs per-tag backfills")]
pub const MAX_BACKFILL_TAGS_PER_SYNC: usize = 64;
pub async fn fetch_private_notes(&mut self) -> Result<(), ClientError> {
self.ensure_genesis_in_place().await?;
let mut update = self.fetch_transport_notes_in_chunks().await?;
let fetch_error = update.fetch_error.take();
self.apply_note_transport_update(update).await?;
if let Some(error) = fetch_error {
return Err(error);
}
Ok(())
}
async fn screen_transport_notes(
&self,
notes: &mut Vec<(NoteId, Note, Option<BlockNumber>)>,
) -> Result<(), ClientError> {
let account_tags = self.tracked_account_tags().await?;
let notes_to_screen: Vec<Note> = notes
.iter()
.filter(|(_, note, _)| account_tags.contains(¬e.metadata().tag()))
.map(|(_, note, _)| note.clone())
.collect();
let consumable = self.note_screener().get_batch_consumability(¬es_to_screen).await?;
notes.retain(|(_, note, _)| {
!account_tags.contains(¬e.metadata().tag()) || consumable.contains_key(¬e.id())
});
Ok(())
}
async fn tracked_account_tags(&self) -> Result<BTreeSet<NoteTag>, ClientError> {
let tags = self
.store
.get_note_tags()
.await?
.into_iter()
.filter(|record| matches!(record.source, NoteTagSource::Account(_)))
.map(|record| record.tag)
.collect();
Ok(tags)
}
async fn fetch_transport_notes_in_chunks(
&self,
) -> Result<NoteTransportLayerUpdate, ClientError> {
let api = self.get_note_transport_api()?;
let (start, stored_tags) = self.load_note_transport_sync_state().await?.unzip();
let (covered, added): (Vec<_>, Vec<_>) = self
.store
.get_unique_note_tags()
.await?
.into_iter()
.partition(|tag| stored_tags.as_ref().is_some_and(|tags| tags.contains(tag)));
let mut update = NoteTransportLayerUpdate::default();
let mut notes = Vec::new();
let mut sync_to = None;
let mut reached = BTreeSet::new();
let chunks = covered
.chunks(MAX_NOTE_TAGS_PER_TRANSPORT_REQUEST)
.map(|chunk| (start, chunk))
.chain(added.chunks(MAX_NOTE_TAGS_PER_TRANSPORT_REQUEST).map(|chunk| (None, chunk)));
for (start, chunk) in chunks {
match fetch_transport_chunk(api.as_ref(), chunk, start, &mut sync_to, &mut notes).await
{
Ok(()) => reached.extend(chunk),
Err(error) => {
update.fetch_error.get_or_insert(error.into());
},
}
}
if let Some(cursor) = sync_to
&& covered.iter().all(|tag| reached.contains(tag))
{
update.sync_state = Some((cursor, reached));
}
update.id_by_commitment =
notes.iter().map(|(id, note, _)| (note.details_commitment(), *id)).collect();
update.note_files = self.prepare_transport_notes(notes).await?;
Ok(update)
}
async fn prepare_transport_notes(
&self,
mut notes: Vec<(NoteId, Note, Option<BlockNumber>)>,
) -> Result<Vec<NoteFile>, ClientError> {
const NOTE_LOOKBACK_BLOCKS: u32 = 20;
if notes.is_empty() {
return Ok(Vec::new());
}
self.drop_notes_resolved_locally(&mut notes).await?;
Box::pin(self.screen_transport_notes(&mut notes)).await?;
let sync_height = self.get_sync_height().await?;
let fallback_after_block_num =
BlockNumber::from(sync_height.as_u32().saturating_sub(NOTE_LOOKBACK_BLOCKS));
let mut note_files = Vec::with_capacity(notes.len());
for (_, note, block_hint) in notes {
let tag = note.metadata().tag();
let after_block_num = block_hint.unwrap_or(fallback_after_block_num);
note_files.push(NoteFile::ExpectedNote {
details: note.into(),
sync_hint: NoteSyncHint::new(after_block_num, tag),
});
}
Ok(note_files)
}
pub(crate) async fn fetch_note_transport_updates(
&self,
) -> Result<NoteTransportLayerUpdate, ClientError> {
if !self.is_note_transport_enabled() {
return Ok(NoteTransportLayerUpdate::default());
}
self.fetch_transport_notes_in_chunks().await
}
pub(crate) async fn apply_note_transport_update(
&mut self,
update: NoteTransportLayerUpdate,
) -> Result<(Vec<NoteId>, Vec<NoteDetailsCommitment>), ClientError> {
let NoteTransportLayerUpdate {
note_files,
id_by_commitment,
sync_state,
fetch_error,
} = update;
let written = self.import_notes(¬e_files).await?;
let mut imported_ids: Vec<NoteId> = written
.iter()
.filter_map(|commitment| id_by_commitment.get(commitment).copied())
.collect();
if let Some(state) = sync_state {
let key = String::from(NOTE_TRANSPORT_SYNC_STATE_KEY);
self.store.set_setting(SettingScope::Client, key, state.to_bytes()).await?;
}
if let Some(error) = fetch_error {
tracing::warn!(
?error,
"note transport fetch failed; the next fetch reads the tags again"
);
}
imported_ids.sort_unstable();
imported_ids.dedup();
Ok((imported_ids, written))
}
async fn drop_notes_resolved_locally(
&self,
notes: &mut Vec<(NoteId, Note, Option<BlockNumber>)>,
) -> Result<(), ClientError> {
let commitments = notes.iter().map(|(_, note, _)| note.details_commitment()).collect();
let records: BTreeMap<_, _> = self
.get_input_notes(NoteFilter::DetailsCommitments(commitments))
.await?
.into_iter()
.map(|record| (record.details_commitment(), record))
.collect();
notes.retain(|(id, note, _)| {
let Some(record) = records.get(¬e.details_commitment()) else {
return true;
};
if record.is_processing() {
tracing::warn!(%id, "skipping delivery of a note being consumed locally");
return false;
}
!((record.is_committed() || record.is_consumed()) && record.id() == Some(*id))
});
Ok(())
}
}
#[derive(Default)]
pub(crate) struct NoteTransportLayerUpdate {
note_files: Vec<NoteFile>,
id_by_commitment: BTreeMap<NoteDetailsCommitment, NoteId>,
sync_state: Option<NoteTransportSyncState>,
pub(crate) fetch_error: Option<ClientError>,
}
type NoteTransportSyncState = (NoteTransportCursor, BTreeSet<NoteTag>);
fn migrate_tag_cursors(
cursors: BTreeMap<NoteTag, Option<NoteTransportCursor>>,
) -> Option<NoteTransportSyncState> {
let cursors: BTreeMap<_, _> =
cursors.into_iter().filter_map(|(tag, cursor)| Some((tag, cursor?))).collect();
let cursor = *cursors.values().min_by_key(|cursor| cursor.sequence)?;
cursors
.values()
.all(|other| other.nonce == cursor.nonce)
.then(|| (cursor, cursors.into_keys().collect()))
}
async fn fetch_transport_chunk(
api: &dyn NoteTransportClient,
tags: &[NoteTag],
start: Option<NoteTransportCursor>,
sync_to: &mut Option<NoteTransportCursor>,
notes: &mut Vec<(NoteId, Note, Option<BlockNumber>)>,
) -> Result<(), NoteTransportError> {
let mut cursor = start;
loop {
let page = api.fetch_notes_page(tags, cursor, *sync_to).await?;
notes.extend(validate_transport_page(&page, tags, cursor, *sync_to)?);
sync_to.get_or_insert(page.sync_to);
if !page.has_more {
return Ok(());
}
cursor = Some(page.cursor);
}
}
fn validate_transport_page(
page: &NoteTransportPage,
tags: &[NoteTag],
request_cursor: Option<NoteTransportCursor>,
request_sync_to: Option<NoteTransportCursor>,
) -> Result<Vec<(NoteId, Note, Option<BlockNumber>)>, NoteTransportError> {
let NoteTransportCursor { nonce, sequence } = page.cursor;
if request_sync_to.is_some_and(|bound| bound != page.sync_to)
|| nonce != page.sync_to.nonce
|| sequence > page.sync_to.sequence
{
return Err(NoteTransportError::Network(String::from(
"fetch response does not match the sync bound",
)));
}
if request_cursor.is_some_and(|request| {
nonce == request.nonce
&& (sequence < request.sequence
|| (!page.notes.is_empty() && sequence == request.sequence))
}) || (!page.notes.is_empty() && sequence == 0)
|| (page.has_more && page.notes.is_empty())
{
return Err(NoteTransportError::Network(String::from(
"fetch response has invalid pagination progress",
)));
}
let mut notes = Vec::with_capacity(page.notes.len());
for info in &page.notes {
let note = rejoin_note(&info.header, &info.details_bytes)?;
if !tags.contains(¬e.metadata().tag()) {
return Err(NoteTransportError::UnrequestedTag(note.metadata().tag()));
}
notes.push((info.header.id(), note, info.block_hint));
}
Ok(notes)
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct NoteTransportCursor {
pub nonce: u64,
pub sequence: u64,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct TransportNote {
header: NoteHeader,
details: NoteDetails,
}
impl TransportNote {
pub fn new(header: NoteHeader, details: NoteDetails) -> Result<Self, NoteTransportError> {
validate_note_parts(&header, &details)?;
Ok(Self { header, details })
}
pub fn header(&self) -> &NoteHeader {
&self.header
}
pub fn details(&self) -> &NoteDetails {
&self.details
}
pub fn into_parts(self) -> (NoteHeader, NoteDetails) {
(self.header, self.details)
}
}
impl From<Note> for TransportNote {
fn from(note: Note) -> Self {
let header = *note.header();
let details = NoteDetails::from(note);
Self { header, details }
}
}
#[cfg_attr(not(target_arch = "wasm32"), async_trait::async_trait)]
#[cfg_attr(target_arch = "wasm32", async_trait::async_trait(?Send))]
pub trait NoteTransportClient: Send + Sync {
async fn send_note_with_proof(
&self,
note: TransportNote,
inclusion_proof: NoteInclusionProof,
) -> Result<(), NoteTransportError>;
async fn fetch_notes_page(
&self,
tags: &[NoteTag],
cursor: Option<NoteTransportCursor>,
sync_to: Option<NoteTransportCursor>,
) -> Result<NoteTransportPage, NoteTransportError>;
}
pub struct NoteTransportPage {
pub notes: Vec<NoteInfo>,
pub cursor: NoteTransportCursor,
pub has_more: bool,
pub sync_to: NoteTransportCursor,
}
#[derive(Debug, Clone)]
pub struct NoteInfo {
pub header: NoteHeader,
pub details_bytes: Vec<u8>,
pub block_hint: Option<BlockNumber>,
}
impl NoteInfo {
pub fn new(header: NoteHeader, details_bytes: Vec<u8>) -> Self {
Self { header, details_bytes, block_hint: None }
}
}
impl Serializable for TransportNote {
fn write_into<W: ByteWriter>(&self, target: &mut W) {
self.header.write_into(target);
self.details.to_bytes().write_into(target);
}
}
impl Deserializable for TransportNote {
fn read_from<R: ByteReader>(source: &mut R) -> Result<Self, DeserializationError> {
let header = NoteHeader::read_from(source)?;
let details_bytes = Vec::<u8>::read_from(source)?;
let details = NoteDetails::read_from_bytes(&details_bytes)?;
Self::new(header, details)
.map_err(|error| DeserializationError::InvalidValue(format!("{error}")))
}
}
impl Serializable for NoteInfo {
fn write_into<W: ByteWriter>(&self, target: &mut W) {
self.header.write_into(target);
self.details_bytes.write_into(target);
self.block_hint.write_into(target);
}
}
impl Deserializable for NoteInfo {
fn read_from<R: ByteReader>(source: &mut R) -> Result<Self, DeserializationError> {
let header = NoteHeader::read_from(source)?;
let details_bytes = Vec::<u8>::read_from(source)?;
let block_hint = Option::<BlockNumber>::read_from(source)?;
Ok(NoteInfo { header, details_bytes, block_hint })
}
}
impl Serializable for NoteTransportCursor {
fn write_into<W: ByteWriter>(&self, target: &mut W) {
(self.nonce, self.sequence).write_into(target);
}
}
impl Deserializable for NoteTransportCursor {
fn read_from<R: ByteReader>(source: &mut R) -> Result<Self, DeserializationError> {
let (nonce, sequence) = <(u64, u64)>::read_from(source)?;
Ok(Self { nonce, sequence })
}
}
fn rejoin_note(header: &NoteHeader, details_bytes: &[u8]) -> Result<Note, NoteTransportError> {
let mut reader = SliceReader::new(details_bytes);
let details = NoteDetails::read_from(&mut reader)?;
validate_note_parts(header, &details)?;
let partial_metadata = *header.metadata().partial_metadata();
Ok(Note::new(
details.assets().clone(),
partial_metadata,
details.recipient().clone(),
))
}
pub(crate) fn validate_note_parts(
header: &NoteHeader,
details: &NoteDetails,
) -> Result<(), NoteTransportError> {
let header_commitment = header.details_commitment();
let details_commitment = details.commitment();
if header_commitment != details_commitment {
return Err(NoteTransportError::NoteDetailsMismatch {
header: header_commitment,
details: details_commitment,
});
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
fn cursor(sequence: u64) -> NoteTransportCursor {
NoteTransportCursor { nonce: 1, sequence }
}
fn empty_page(cursor: NoteTransportCursor, has_more: bool) -> NoteTransportPage {
NoteTransportPage {
notes: Vec::new(),
cursor,
has_more,
sync_to: self::cursor(20),
}
}
#[test]
fn transport_rejects_invalid_pagination() {
for page in [empty_page(cursor(9), false), empty_page(cursor(10), true)] {
assert!(validate_transport_page(&page, &[], Some(cursor(10)), None).is_err());
}
}
#[test]
fn transport_rejects_page_outside_sync_bound() {
let other_database = NoteTransportCursor { nonce: 2, sequence: 5 };
for (page, request_sync_to) in [
(empty_page(cursor(10), false), Some(cursor(21))),
(empty_page(cursor(21), false), None),
(empty_page(other_database, false), None),
] {
assert!(
validate_transport_page(&page, &[], Some(cursor(10)), request_sync_to).is_err()
);
}
}
#[test]
fn transport_accepts_pages_within_sync_bound() {
let page = empty_page(cursor(10), false);
assert!(validate_transport_page(&page, &[], Some(cursor(10)), Some(cursor(20))).is_ok());
let other_database = NoteTransportCursor { nonce: 2, sequence: 50 };
let page = empty_page(cursor(3), false);
assert!(validate_transport_page(&page, &[], Some(other_database), None).is_ok());
}
#[test]
fn transport_reads_legacy_tag_cursors() {
let tag = NoteTag::new(1);
let legacy = BTreeMap::from([(tag, Some((1u64, 4u64)))]).to_bytes();
let cursors =
BTreeMap::<NoteTag, Option<NoteTransportCursor>>::read_from_bytes(&legacy).unwrap();
assert_eq!(cursors, BTreeMap::from([(tag, Some(cursor(4)))]));
}
#[test]
fn transport_migrates_tag_cursors_to_lowest_cursor() {
let [first, second, third] = [1, 2, 3].map(NoteTag::new);
let cursors =
BTreeMap::from([(first, Some(cursor(7))), (second, Some(cursor(4))), (third, None)]);
let state = migrate_tag_cursors(cursors);
assert_eq!(state, Some((cursor(4), BTreeSet::from([first, second]))));
}
#[test]
fn transport_migration_drops_cursors_from_different_databases() {
let other_database = NoteTransportCursor { nonce: 2, sequence: 4 };
let cursors = BTreeMap::from([
(NoteTag::new(1), Some(cursor(7))),
(NoteTag::new(2), Some(other_database)),
]);
assert_eq!(migrate_tag_cursors(cursors), None);
assert_eq!(migrate_tag_cursors(BTreeMap::new()), None);
}
}