//! Shared marshal fixtures for stateful actor tests. use super::mocks::{TestBlock, TestScheme, TestVariant}; use commonware_actor::Feedback; use commonware_consensus::{ Heightable as _, Reporter, marshal::{ self, Update, core::{Actor as MarshalActor, Floor, Mailbox as MarshalMailbox}, resolver::handler, }, simplex::types::{Finalization, Finalize, Proposal}, types::{Epoch, FixedEpocher, Round, View, ViewDelta}, }; use commonware_cryptography::{ Digestible as _, certificate::{ConstantProvider, mocks::Fixture}, ed25519, sha256::Digest as Sha256Digest, }; use commonware_parallel::Sequential; use commonware_resolver::{Fetch, Resolver, TargetedResolver}; use commonware_runtime::{Handle, Supervisor as _, buffer::paged::CacheRef, deterministic}; use commonware_storage::{ archive::{Archive as _, immutable, prunable}, translator::TwoCap, }; use commonware_utils::{ NZU16, NZU64, NZUsize, acknowledgement::{Acknowledgement as _, Exact}, non_empty, sync::Mutex, vec::NonEmptyVec, }; use std::{num::NonZeroUsize, sync::Arc}; #[derive(Clone)] struct FixtureReporter { acknowledge: bool, pending: Arc>>, } impl Reporter for FixtureReporter { type Activity = Update; fn report(&mut self, activity: Self::Activity) -> Feedback { if let Update::Block(_, ack) = activity { if self.acknowledge { ack.acknowledge(); } else { self.pending.lock().push(ack); } } Feedback::Ok } } #[derive(Clone, Copy)] enum Dispatch { Stopped, Acknowledge, Hold, } struct Options<'a> { seed: Option<(&'a TestBlock, Finalization)>, block: Option<&'a TestBlock>, floor: Option>, max_pending_acks: NonZeroUsize, dispatch: Dispatch, } /// Backfill resolver for a started marshal fixture: its archives are pre-seeded, so /// every fetch is ignored. #[derive(Clone)] pub(crate) struct IgnoreResolver; impl Resolver for IgnoreResolver { type Key = handler::Key; type Subscriber = handler::Annotation; fn fetch(&mut self, _key: F) -> Feedback where F: Into> + Send, { Feedback::Ok } fn fetch_all(&mut self, _keys: Vec) -> Feedback where F: Into> + Send, { Feedback::Ok } fn retain( &mut self, _predicate: impl Fn(&Self::Key, &Self::Subscriber) -> bool + Send + 'static, ) -> Feedback { Feedback::Ok } } impl TargetedResolver for IgnoreResolver { type PublicKey = ed25519::PublicKey; fn fetch_targeted( &mut self, _fetch: impl Into> + Send, _targets: NonEmptyVec, ) -> Feedback { Feedback::Ok } fn fetch_all_targeted(&mut self, _keys: Vec<(F, NonEmptyVec)>) -> Feedback where F: Into> + Send, { Feedback::Ok } } /// Builds a finalization for `view` over `payload`, signed by every scheme in `fixture`. pub(crate) fn finalization( fixture: &Fixture, view: u64, payload: Sha256Digest, ) -> Finalization { let proposal = Proposal { round: Round::new(Epoch::zero(), View::new(view)), parent: View::new(view.saturating_sub(1)), payload, }; let finalizes = fixture .schemes .iter() .map(|scheme| Finalize::sign(scheme, proposal.clone()).expect("sign finalize")) .collect::>(); Finalization::from_finalizes( &fixture.verifier, non_empty![@finalizes.iter()], &Sequential, ) .expect("recover finalization") } fn archive_config(page_cache: CacheRef, partition: &str) -> immutable::Config<()> { immutable::Config { metadata_partition: format!("{partition}-metadata"), freezer_table_partition: format!("{partition}-freezer-table"), freezer_table_initial_size: 4, freezer_table_resize_frequency: 2, freezer_table_resize_chunk_size: 2, freezer_key_partition: format!("{partition}-freezer-key"), freezer_key_page_cache: page_cache, freezer_value_partition: format!("{partition}-freezer-value"), freezer_value_target_size: 128, freezer_value_compression: None, ordinal_partition: format!("{partition}-ordinal"), items_per_section: NZU64!(4), codec_config: (), replay_buffer: NZUsize!(64), freezer_key_write_buffer: NZUsize!(64), freezer_value_write_buffer: NZUsize!(64), ordinal_write_buffer: NZUsize!(64), } } fn prunable_archive_config(page_cache: CacheRef, partition: &str) -> prunable::Config { prunable::Config { translator: TwoCap, metadata_partition: format!("{partition}-metadata"), key_partition: format!("{partition}-key"), key_page_cache: page_cache, value_partition: format!("{partition}-value"), compression: None, codec_config: (), items_per_section: NZU64!(4), key_write_buffer: NZUsize!(64), value_write_buffer: NZUsize!(64), replay_buffer: NZUsize!(64), } } /// A marshal actor fixture over initialized finalization and block archives. pub(crate) struct MarshalFixture { pub(crate) mailbox: MarshalMailbox, pub(crate) floor: Floor, /// Keeps the fixture alive: the unstarted actor (its mailbox dead-letters once this /// drops) or the started actor's resolver handler and task handle. pub(crate) guards: Box, } impl MarshalFixture { /// Aborts a started fixture so its durable storage can be reopened. pub(crate) fn abort(self) { let guards = self .guards .downcast::<(handler::Handler, Handle<()>)>() .unwrap_or_else(|_| panic!("marshal fixture was not started")); let (_, handle) = *guards; handle.abort(); } } /// Initializes a marshal actor whose finalization archive is pre-seeded with `seed`. /// /// When `start` is set, the actor runs with a reporter that acknowledges every dispatched /// block and a resolver that ignores every fetch, so `get_finalization` serves the seeded /// finalization without any peer fetching. Otherwise the actor is returned unstarted /// inside [`MarshalFixture::guards`]. pub(crate) async fn marshal_fixture( context: deterministic::Context, prefix: &str, scheme: TestScheme, seed: Option<(&TestBlock, Finalization)>, max_pending_acks: NonZeroUsize, start: bool, ) -> MarshalFixture { let dispatch = if start { Dispatch::Acknowledge } else { Dispatch::Stopped }; marshal_fixture_inner( context, prefix, scheme, Options { seed, block: None, floor: None, max_pending_acks, dispatch, }, ) .await } /// Initializes a marshal actor whose finalized-block archive is pre-seeded with `block`. pub(crate) async fn marshal_fixture_with_finalized_block( context: deterministic::Context, prefix: &str, scheme: TestScheme, block: &TestBlock, max_pending_acks: NonZeroUsize, start: bool, ) -> MarshalFixture { let dispatch = if start { Dispatch::Acknowledge } else { Dispatch::Stopped }; marshal_fixture_inner( context, prefix, scheme, Options { seed: None, block: Some(block), floor: None, max_pending_acks, dispatch, }, ) .await } /// Initializes a started marshal actor with `block` as its floor anchor and freezes the /// anchor acknowledgement so the durable processed height remains at its predecessor. pub(crate) async fn marshal_fixture_with_floor( context: deterministic::Context, prefix: &str, scheme: TestScheme, block: &TestBlock, finalization: Finalization, max_pending_acks: NonZeroUsize, ) -> MarshalFixture { marshal_fixture_inner( context, prefix, scheme, Options { seed: None, block: Some(block), floor: Some(finalization), max_pending_acks, dispatch: Dispatch::Hold, }, ) .await } /// Initializes a started marshal actor over prunable finalized archives. /// /// When provided, `block` is pre-seeded. Dispatched blocks are held open unless /// `acknowledge` is set. pub(crate) async fn prunable_marshal_fixture( context: deterministic::Context, prefix: &str, scheme: TestScheme, block: Option<&TestBlock>, floor: Option>, max_pending_acks: NonZeroUsize, acknowledge: bool, ) -> MarshalFixture { let options = Options { seed: None, block, floor, max_pending_acks, dispatch: if acknowledge { Dispatch::Acknowledge } else { Dispatch::Hold }, }; let page_cache = CacheRef::from_pooler(&context, NZU16!(1024), NZUsize!(8)); let finalizations_by_height = prunable::Archive::init( context.child("finalizations_by_height"), prunable_archive_config(page_cache.clone(), &format!("{prefix}-finalizations")), ) .await .expect("failed to initialize finalizations archive"); let mut finalized_blocks = prunable::Archive::init( context.child("finalized_blocks"), prunable_archive_config(page_cache.clone(), &format!("{prefix}-blocks")), ) .await .expect("failed to initialize blocks archive"); if let Some(block) = options.block { finalized_blocks = finalized_blocks .put(block.height().get(), block.digest(), block.clone()) .await .expect("failed to seed finalized block") .sync() .await .expect("failed to sync finalized blocks archive"); } start_marshal_fixture( context, prefix, scheme, options, page_cache, finalizations_by_height, finalized_blocks, ) .await } async fn marshal_fixture_inner( context: deterministic::Context, prefix: &str, scheme: TestScheme, mut options: Options<'_>, ) -> MarshalFixture { let page_cache = CacheRef::from_pooler(&context, NZU16!(1024), NZUsize!(8)); let mut finalizations_by_height = immutable::Archive::init( context.child("finalizations_by_height"), archive_config(page_cache.clone(), &format!("{prefix}-finalizations")), ) .await .expect("failed to initialize finalizations archive"); let mut finalized_blocks = immutable::Archive::init( context.child("finalized_blocks"), archive_config(page_cache.clone(), &format!("{prefix}-blocks")), ) .await .expect("failed to initialize blocks archive"); if let Some(block) = options.block { finalized_blocks = finalized_blocks .put(block.height().get(), block.digest(), block.clone()) .await .expect("failed to seed finalized block") .sync() .await .expect("failed to sync finalized blocks archive"); } if let Some((block, finalization)) = options.seed.take() { finalizations_by_height = finalizations_by_height .put(block.height().get(), block.digest(), finalization) .await .expect("failed to seed finalization") .sync() .await .expect("failed to sync finalizations archive"); } start_marshal_fixture( context, prefix, scheme, options, page_cache, finalizations_by_height, finalized_blocks, ) .await } async fn start_marshal_fixture( context: deterministic::Context, prefix: &str, scheme: TestScheme, options: Options<'_>, page_cache: CacheRef, finalizations_by_height: FC, finalized_blocks: FB, ) -> MarshalFixture where FC: marshal::store::Certificates< BlockDigest = Sha256Digest, Commitment = Sha256Digest, Scheme = TestScheme, >, FB: marshal::store::Blocks, { let provider = ConstantProvider::new(scheme); let (actor, mailbox, floor) = MarshalActor::<_, TestVariant, _, _, _, _, _>::init( context.child("marshal_actor"), finalizations_by_height, finalized_blocks, marshal::Config { provider, epocher: FixedEpocher::new(NZU64!(u64::MAX)), start: options.floor.map_or_else( || marshal::Start::Genesis(TestBlock::new(0, 0)), marshal::Start::Floor, ), partition_prefix: format!("{prefix}-marshal"), mailbox_size: NZUsize!(8), view_retention: ViewDelta::new(1), prunable_items_per_section: NZU64!(4), page_cache, replay_buffer: NZUsize!(64), key_write_buffer: NZUsize!(64), value_write_buffer: NZUsize!(64), block_codec_config: (), max_repair: NZUsize!(1), max_pending_acks: options.max_pending_acks, strategy: Sequential, }, ) .await; if matches!(options.dispatch, Dispatch::Stopped) { return MarshalFixture { mailbox, floor, guards: Box::new(actor), }; } let (resolver_receiver, resolver_handler) = handler::init(context.child("resolver_handler"), NZUsize!(8)); let reporter = FixtureReporter { acknowledge: matches!(options.dispatch, Dispatch::Acknowledge), pending: Arc::new(Mutex::new(Vec::new())), }; let handle = actor.start_unbuffered(reporter, (resolver_receiver, IgnoreResolver)); MarshalFixture { mailbox, floor, guards: Box::new((resolver_handler, handle)), } }