mod data_store;
mod error;
mod staged_smt;
use alloc::boxed::Box;
use alloc::collections::{BTreeMap, BTreeSet};
use alloc::sync::Arc;
use alloc::vec::Vec;
pub(crate) use data_store::InMemoryBatchDataStore;
pub use error::BatchBuilderError;
use miden_protocol::MIN_PROOF_SECURITY_LEVEL;
use miden_protocol::account::AccountId;
use miden_protocol::batch::{ProposedBatch, ProvenBatch};
use miden_protocol::block::{BlockHeader, BlockNumber};
use miden_protocol::note::NoteId;
use miden_protocol::transaction::{PartialBlockchain, ProvenTransaction, TransactionId};
use miden_tx::auth::TransactionAuthenticator;
use miden_tx_batch::{BatchExecutor, LocalBatchProver};
use crate::rpc::RpcError;
use crate::rpc::encryption::seal_transaction_inputs;
use crate::store::data_store::{ClientDataStore, build_partial_mmr_with_paths};
use crate::transaction::{
TransactionRequest,
TransactionResult,
TransactionStoreUpdate,
ensure_account_allowed,
validate_executed_transaction,
};
use crate::{Client, ClientError};
#[derive(Debug, Clone)]
pub struct ProvenBatchSubmission {
proven_batch: ProvenBatch,
proposed_batch: Box<ProposedBatch>,
tx_results: Vec<TransactionResult>,
}
impl ProvenBatchSubmission {
pub fn transaction_count(&self) -> usize {
self.tx_results.len()
}
pub fn transaction_ids(&self) -> impl Iterator<Item = TransactionId> + '_ {
self.tx_results.iter().map(|tx_result| tx_result.executed_transaction().id())
}
}
pub(crate) struct PushedTx {
pub(crate) proven_tx: Arc<ProvenTransaction>,
pub(crate) tx_result: TransactionResult,
}
pub struct BatchBuilder<'c, AUTH> {
pub(crate) client: &'c mut Client<AUTH>,
pub(crate) data_store: InMemoryBatchDataStore,
pub(crate) pushed_txs: Vec<PushedTx>,
pub(crate) consumed_input_notes: BTreeSet<NoteId>,
}
impl<AUTH> BatchBuilder<'_, AUTH> {
pub fn len(&self) -> usize {
self.pushed_txs.len()
}
pub fn is_empty(&self) -> bool {
self.pushed_txs.is_empty()
}
}
impl<AUTH> Client<AUTH>
where
AUTH: TransactionAuthenticator + Sync + 'static,
{
pub fn new_transaction_batch(&mut self) -> BatchBuilder<'_, AUTH> {
let inner_data_store = ClientDataStore::new(self.store.clone(), self.rpc_api.clone());
BatchBuilder {
client: self,
data_store: InMemoryBatchDataStore::new(inner_data_store),
pushed_txs: Vec::new(),
consumed_input_notes: BTreeSet::new(),
}
}
pub async fn retry_proven_batch(
&mut self,
submission: &ProvenBatchSubmission,
) -> Result<BlockNumber, ClientError> {
self.send_and_apply_proven_batch(submission).await
}
async fn send_and_apply_proven_batch(
&mut self,
submission: &ProvenBatchSubmission,
) -> Result<BlockNumber, ClientError> {
let key = self.transaction_encryption_key().await?;
let sealed_inputs = submission
.tx_results
.iter()
.map(|tx_result| {
let executed = tx_result.executed_transaction();
seal_transaction_inputs(&mut self.rng, &key, executed.id(), executed.tx_inputs())
})
.collect::<Result<Vec<_>, _>>()?;
let result = self
.rpc_api
.submit_proven_batch(
&submission.proven_batch,
&submission.proposed_batch,
sealed_inputs,
)
.await;
if let Err(err) = &result {
self.forget_stale_transaction_encryption_key(err).await;
}
let block_num = result.map_err(|err| promote_indeterminate_submission(err, submission))?;
let mut updates: Vec<TransactionStoreUpdate> =
Vec::with_capacity(submission.transaction_count());
for tx_result in &submission.tx_results {
let update = self.get_transaction_store_update(tx_result, block_num).await.map_err(
|source| BatchBuilderError::BatchSubmittedButUpdateBuildFailed {
block_num,
source,
},
)?;
updates.push(update);
}
if let Err(source) = self.store.apply_transaction_batch(updates).await {
return Err(ClientError::from(BatchBuilderError::BatchSubmittedButApplyFailed {
block_num,
source,
}));
}
Ok(block_num)
}
}
impl<AUTH> BatchBuilder<'_, AUTH>
where
AUTH: TransactionAuthenticator + Sync + 'static,
{
pub async fn submit(self) -> Result<BlockNumber, ClientError> {
let ref_block_num = self
.pushed_txs
.iter()
.map(|p| p.proven_tx.ref_block_num())
.max()
.ok_or(BatchBuilderError::Empty)?;
let lower_refs: BTreeSet<BlockNumber> = self
.pushed_txs
.iter()
.map(|p| p.proven_tx.ref_block_num())
.filter(|&r| r < ref_block_num)
.collect();
let account_ids: BTreeSet<AccountId> =
self.pushed_txs.iter().map(|p| p.proven_tx.account_id()).collect();
for account_id in account_ids {
if self.client.is_allowlist_gated(account_id).await? {
ensure_account_allowed(
account_id,
self.client.is_account_allowed(account_id).await,
)?;
}
}
let store = self.client.store.clone();
let (ref_block_header, _) = store
.get_block_header_by_num(ref_block_num)
.await
.map_err(ClientError::StoreError)?
.ok_or_else(|| {
ClientError::StoreError(crate::store::StoreError::BlockHeaderNotFound(
ref_block_num,
))
})?;
let fetched =
store.get_block_headers(&lower_refs).await.map_err(ClientError::StoreError)?;
let authenticated_blocks: Vec<BlockHeader> =
fetched.into_iter().map(|(header, _)| header).collect();
let fetched_nums: BTreeSet<BlockNumber> =
authenticated_blocks.iter().map(BlockHeader::block_num).collect();
if let Some(&missing) = lower_refs.difference(&fetched_nums).next() {
return Err(ClientError::StoreError(crate::store::StoreError::BlockHeaderNotFound(
missing,
)));
}
let current_peaks =
store.get_current_blockchain_peaks().await.map_err(ClientError::StoreError)?;
let partial_mmr =
build_partial_mmr_with_paths(&store, current_peaks, &authenticated_blocks).await?;
let partial_blockchain = PartialBlockchain::new(partial_mmr, authenticated_blocks)?;
let len = self.pushed_txs.len();
let mut proven_txs: Vec<Arc<ProvenTransaction>> = Vec::with_capacity(len);
let mut tx_results: Vec<TransactionResult> = Vec::with_capacity(len);
for pushed in self.pushed_txs {
proven_txs.push(pushed.proven_tx);
tx_results.push(pushed.tx_result);
}
let unauthenticated_note_proofs = BTreeMap::new();
let proposed_batch = ProposedBatch::new(
proven_txs,
ref_block_header,
partial_blockchain,
unauthenticated_note_proofs,
MIN_PROOF_SECURITY_LEVEL,
)?;
let executed_batch = BatchExecutor::new().execute(proposed_batch.clone())?;
let proven_batch =
LocalBatchProver::new(miden_tx::Prover::default()).prove(executed_batch)?;
let submission = ProvenBatchSubmission {
proven_batch,
proposed_batch: Box::new(proposed_batch),
tx_results,
};
let block_num = self.client.send_and_apply_proven_batch(&submission).await?;
Ok(block_num)
}
pub async fn push(
&mut self,
account_id: AccountId,
req: TransactionRequest,
) -> Result<&mut Self, ClientError> {
for note_id in req.input_note_ids() {
if self.consumed_input_notes.contains(¬e_id) {
return Err(ClientError::from(BatchBuilderError::DuplicateInputNote(note_id)));
}
}
let tx_result =
Box::pin(execute_transaction_for_batch(self.client, &self.data_store, account_id, req))
.await?;
let proven_tx = self.client.prove_transaction(&tx_result).await?;
self.data_store
.apply_executed_transaction(tx_result.executed_transaction())
.await?;
for note in tx_result.consumed_notes().iter() {
self.consumed_input_notes.insert(note.id());
}
self.pushed_txs.push(PushedTx {
proven_tx: Arc::new(proven_tx),
tx_result,
});
Ok(self)
}
}
async fn execute_transaction_for_batch<AUTH>(
client: &Client<AUTH>,
data_store: &InMemoryBatchDataStore,
account_id: AccountId,
transaction_request: TransactionRequest,
) -> Result<TransactionResult, ClientError>
where
AUTH: TransactionAuthenticator + Sync + 'static,
{
let account_reader = client.account_reader(account_id);
if account_reader.status().await?.is_locked() {
return Err(ClientError::AccountLocked(account_id));
}
let account = match data_store.cached_account(account_id) {
Some(account) => account,
None => account_reader.partial_account().await?,
};
let prep = client.prepare_transaction_for_batch(&account, transaction_request).await?;
data_store.register_note_scripts(prep.output_note_scripts());
for fpi_account in &prep.foreign_account_inputs {
data_store.mast_store().load_account_code(fpi_account.code());
}
data_store.register_foreign_account_inputs(prep.foreign_account_inputs);
data_store.mast_store().load_account_code(account.code());
let mut notes = prep.notes;
if prep.ignore_invalid_notes {
notes = client
.get_valid_input_notes(
data_store,
account_id,
prep.block_num,
notes,
prep.tx_args.clone(),
)
.await?;
}
let executed_transaction = client
.build_executor(data_store)?
.execute_transaction(account_id, prep.block_num, notes, prep.tx_args)
.await?;
validate_executed_transaction(&executed_transaction, &prep.output_recipients)?;
TransactionResult::new(executed_transaction, prep.future_notes)
}
fn promote_indeterminate_submission(
err: RpcError,
submission: &ProvenBatchSubmission,
) -> ClientError {
if !err.is_indeterminate_submission() {
return ClientError::RpcError(err);
}
BatchBuilderError::BatchSubmissionOutcomeUnknown {
submission: Box::new(submission.clone()),
source: err,
}
.into()
}