diff --git a/dkg/src/orchestrator/actor.rs b/dkg/src/orchestrator/actor.rs
index f815252..060b701 100644
--- a/dkg/src/orchestrator/actor.rs
+++ b/dkg/src/orchestrator/actor.rs
@@ -26,9 +26,14 @@ use commonware_parallel::Strategy;
 use commonware_runtime::{
     buffer::paged::CacheRef,
     spawn_cell,
-    telemetry::metrics::{Gauge, GaugeExt, MetricsExt as _},
+    telemetry::metrics::{
+        histogram::Buckets, CounterFamily, EncodeLabelSet, Gauge, GaugeExt, Histogram,
+        MetricsExt as _,
+    },
     BufferPooler, Clock, ContextCell, Handle, Metrics, Network, Spawner, Storage,
 };
+use commonware_storage::metadata::{self, Metadata};
+use commonware_utils::sequence::U64;
 use commonware_utils::{vec::NonEmptyVec, NZUsize, NZU16};
 use rand::CryptoRng;
 use std::{
@@ -39,6 +44,13 @@ use std::{
 };
 use tracing::{debug, info, warn};
 
+const CLEANUP_WATERMARK_KEY: U64 = U64::new(0);
+
+#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq, EncodeLabelSet)]
+struct CleanupStatusLabel {
+    status: &'static str,
+}
+
 /// Reporter that discards consensus activity.
 pub struct NoopReporter<A> {
     _phantom: PhantomData<A>,
@@ -146,6 +158,10 @@ where
     page_cache_ref: CacheRef,
 
     latest_epoch: Gauge,
+    partition_cleanup_total: CounterFamily<CleanupStatusLabel>,
+    partition_cleanup_watermark: Gauge,
+    partitions_active: Gauge,
+    partition_cleanup_duration: Histogram,
 
     _phantom: PhantomData<L>,
 }
@@ -180,6 +196,23 @@ where
 
         // Register latest_epoch gauge for Grafana integration
         let latest_epoch = context.gauge("latest_epoch", "current epoch");
+        let partition_cleanup_total = context.family(
+            "consensus_partition_cleanup",
+            "Total number of consensus partition cleanup outcomes",
+        );
+        let partition_cleanup_watermark = context.gauge(
+            "consensus_partition_cleanup_watermark",
+            "Next consensus epoch partition requiring cleanup",
+        );
+        let partitions_active = context.gauge(
+            "consensus_partitions_active",
+            "Number of consensus epoch partitions with active engines",
+        );
+        let partition_cleanup_duration = context.histogram(
+            "consensus_partition_cleanup_duration_seconds",
+            "Duration of consensus epoch partition cleanup operations",
+            Buckets::LOCAL,
+        );
 
         (
             Self {
@@ -200,6 +233,10 @@ where
                 recovered_floor: config.recovered_floor,
                 page_cache_ref,
                 latest_epoch,
+                partition_cleanup_total,
+                partition_cleanup_watermark,
+                partitions_active,
+                partition_cleanup_duration,
                 _phantom: PhantomData,
             },
             Mailbox::new(sender),
@@ -268,6 +305,24 @@ where
         // Wait for instructions to transition epochs.
         let epocher = FixedEpocher::new(self.epoch_length);
         let mut engines: BTreeMap<Epoch, Handle<()>> = BTreeMap::new();
+        let cleanup_partition = format!("{}_cleanup-v1", self.partition_prefix);
+        let mut cleanup_metadata = Metadata::<E, U64, U64>::init(
+            self.context.child("partition_cleanup_metadata"),
+            metadata::Config {
+                partition: cleanup_partition,
+                codec_config: (),
+            },
+        )
+        .await
+        .expect("failed to initialize consensus partition cleanup metadata");
+        let mut next_epoch_to_clean = cleanup_metadata
+            .get(&CLEANUP_WATERMARK_KEY)
+            .cloned()
+            .unwrap_or_else(|| U64::new(0));
+        let _ = self
+            .partition_cleanup_watermark
+            .try_set(u64::from(&next_epoch_to_clean));
+        let mut startup_cleanup_done = false;
 
         select_loop! {
             self.context,
@@ -317,6 +372,28 @@ where
                         continue;
                     }
 
+                    if !startup_cleanup_done {
+                        assert!(
+                            u64::from(&next_epoch_to_clean) <= transition.epoch.get(),
+                            "consensus partition cleanup watermark {} is ahead of first entering epoch {}",
+                            u64::from(&next_epoch_to_clean),
+                            transition.epoch,
+                        );
+                        if let Some(previous) = transition.epoch.previous() {
+                            assert!(
+                                !engines.keys().any(|epoch| *epoch <= previous),
+                                "refusing to clean a consensus partition with an active engine"
+                            );
+                            self.cleanup_through(
+                                &mut cleanup_metadata,
+                                &mut next_epoch_to_clean,
+                                previous,
+                            )
+                            .await;
+                        }
+                        startup_cleanup_done = true;
+                    }
+
                     let floor = if self
                         .recovered_floor
                         .as_ref()
@@ -364,6 +441,7 @@ where
                         .await;
                     engines.insert(transition.epoch, handle);
                     let _ = self.latest_epoch.try_set(transition.epoch.get());
+                    let _ = self.partitions_active.try_set(engines.len());
 
                     info!(epoch = %transition.epoch, "entered epoch");
                 }
@@ -374,16 +452,91 @@ where
                         continue;
                     };
                     handle.abort();
+                    // Spawned task handles close their completion channel when aborted.
+                    match handle.await {
+                        Ok(())
+                        | Err(commonware_runtime::Error::Aborted)
+                        | Err(commonware_runtime::Error::Closed) => {}
+                        Err(error) => {
+                            panic!("consensus engine for epoch {epoch} failed while stopping: {error}")
+                        }
+                    }
+                    let _ = self.partitions_active.try_set(engines.len());
 
                     // Unregister the signing scheme for the epoch.
                     assert!(self.provider.unregister(&epoch));
 
+                    if u64::from(&next_epoch_to_clean) <= epoch.get() {
+                        assert!(
+                            !engines.keys().any(|active| *active <= epoch),
+                            "refusing to clean a consensus partition with an active engine"
+                        );
+                        self.cleanup_through(
+                            &mut cleanup_metadata,
+                            &mut next_epoch_to_clean,
+                            epoch,
+                        )
+                        .await;
+                    }
+
                     info!(%epoch, "exited epoch");
                 }
             },
         }
     }
 
+    async fn cleanup_through(
+        &mut self,
+        metadata: &mut Metadata<E, U64, U64>,
+        next_epoch_to_clean: &mut U64,
+        through: Epoch,
+    ) {
+        while u64::from(&*next_epoch_to_clean) <= through.get() {
+            let epoch = Epoch::new(u64::from(&*next_epoch_to_clean));
+            let partition = format!("{}_consensus_{}", self.partition_prefix, epoch);
+            let started = std::time::Instant::now();
+            let status = match self.context.remove(&partition, None).await {
+                Ok(()) => {
+                    info!(%epoch, %partition, "removed retired consensus partition");
+                    "removed"
+                }
+                Err(commonware_runtime::Error::PartitionMissing(_)) => {
+                    info!(%epoch, %partition, "retired consensus partition already missing");
+                    "missing"
+                }
+                Err(error) => {
+                    self.partition_cleanup_total
+                        .get_or_create(&CleanupStatusLabel { status: "failed" })
+                        .inc();
+                    self.partition_cleanup_duration
+                        .observe(started.elapsed().as_secs_f64());
+                    warn!(%epoch, %partition, %error, "failed to remove retired consensus partition");
+                    panic!("failed to remove retired consensus partition {partition}: {error}");
+                }
+            };
+            self.partition_cleanup_total
+                .get_or_create(&CleanupStatusLabel { status })
+                .inc();
+            self.partition_cleanup_duration
+                .observe(started.elapsed().as_secs_f64());
+
+            let next = epoch
+                .get()
+                .checked_add(1)
+                .expect("consensus cleanup watermark overflow");
+            metadata
+                .put_sync(CLEANUP_WATERMARK_KEY, U64::new(next))
+                .await
+                .unwrap_or_else(|error| {
+                    panic!(
+                        "failed to persist consensus cleanup watermark after removing {partition}: {error}"
+                    )
+                });
+            *next_epoch_to_clean = U64::new(next);
+            let _ = self.partition_cleanup_watermark.try_set(next);
+        }
+    }
+
     // Epoch zero uses genesis as its floor; every later epoch is anchored by the
     // last finalized block from the previous epoch.
     fn floor_boundary(epocher: &FixedEpocher, epoch: Epoch) -> Option<Height> {
diff --git a/examples/coins/chain/src/engine.rs b/examples/coins/chain/src/engine.rs
index 00a47dd..2930cbf 100644
--- a/examples/coins/chain/src/engine.rs
+++ b/examples/coins/chain/src/engine.rs
@@ -1,11 +1,9 @@
 use crate::application::{self, Application};
 use crate::execution::NodeHandle;
 use crate::genesis::{genesis_target, state_commitment, ChainGenesis};
+use crate::history::{self, BlocksArchive, FinalizationsArchive};
 use crate::indexer;
-use crate::{
-    Block, EpochProvider, Finalization, Provider, PublicKey, Scheme, Transaction, BLOCKS_PER_EPOCH,
-    NAMESPACE,
-};
+use crate::{Block, EpochProvider, Provider, PublicKey, Scheme, Transaction, NAMESPACE};
 use commonware_broadcast::buffered;
 use commonware_consensus::{
     marshal::{
@@ -13,7 +11,7 @@ use commonware_consensus::{
         core::Actor as MarshalActor,
         resolver,
         standard::{Inline, Standard},
-        store::{Blocks as _, Certificates},
+        store::Certificates,
     },
     simplex::elector::Random,
     types::{Epoch, FixedEpocher, Height, ViewDelta},
@@ -24,7 +22,6 @@ use commonware_cryptography::{
         dkg::feldman_desmedt::Output,
         primitives::{group, variant::MinSig},
     },
-    certificate::Verifier as _,
     ed25519::{self, Batch},
     sha256::Digest,
     BatchVerifier, Digestible, Signer,
@@ -41,12 +38,8 @@ use commonware_runtime::{
     buffer::paged::CacheRef, spawn_cell, BufferPooler, Clock, ContextCell, Handle, Metrics,
     Network, Spawner, Storage, Strategizer,
 };
-use commonware_storage::{
-    archive::{immutable, Identifier as ArchiveIdentifier},
-    metadata::{self, Metadata},
-    queue,
-};
-use commonware_utils::{sequence::U64, union, NZDuration, NZU64};
+use commonware_storage::{archive::{Archive as ArchiveStore, Identifier as ArchiveIdentifier}, metadata::{self, Metadata}, queue};
+use commonware_utils::{sequence::U64, union, NZDuration};
 use futures::lock::Mutex as AsyncMutex;
 use governor::clock::Clock as GClock;
 use nunchi_chain::engine::*;
@@ -61,6 +54,7 @@ use nunchi_mempool::{Mempool, PoolConfig};
 use rand::{CryptoRng, Rng};
 use std::{
     marker::PhantomData,
+    num::NonZeroU64,
     sync::Arc,
     time::{Duration, Instant},
 };
@@ -84,13 +78,12 @@ pub struct Config<B: Blocker<PublicKey = PublicKey>, P: Manager<PublicKey = Publ
     pub blocker: B,
     pub manager: P,
     pub partition_prefix: String,
-    pub blocks_freezer_table_initial_size: u32,
-    pub finalized_freezer_table_initial_size: u32,
     pub signer: ed25519::PrivateKey,
     pub dkg_storage_key: dkg::StorageKey,
     pub output: Output<MinSig, PublicKey>,
     pub share: Option<group::Share>,
     pub peer_config: PeerConfig<PublicKey>,
+    pub epoch_length: NonZeroU64,
     pub leader_timeout: Duration,
     pub certification_timeout: Duration,
     pub strategy: S,
@@ -100,6 +93,7 @@ pub struct Config<B: Blocker<PublicKey = PublicKey>, P: Manager<PublicKey = Publ
     pub pool_config: PoolConfig,
     pub genesis: Option<ChainGenesis>,
     pub indexer: Option<indexer::HttpClient>,
+    pub indexer_spool_limits: indexer::SpoolLimits,
 }
 
 type DkgActor<E, P> = nunchi_chain::DkgActor<E, P, Transaction, ClobExtension>;
@@ -110,8 +104,6 @@ type LimitedStatefulAppMailbox<E> = VerifyLimiter<StatefulAppMailbox<E>>;
 type InlineApp<E> = Inline<E, Scheme, LimitedStatefulAppMailbox<E>, Block, FixedEpocher>;
 type Marshaled<E> = BoxedAutomaton<InlineApp<E>>;
 type SchemeProvider = Provider<Scheme, ed25519::PrivateKey>;
-type FinalizationsArchive<E> = immutable::Archive<E, Digest, Finalization>;
-type BlocksArchive<E> = immutable::Archive<E, Digest, Block>;
 type Marshal<E, S> = MarshalActor<
     E,
     Standard<Block>,
@@ -167,6 +159,7 @@ where
     stateful: StatefulApp<E>,
     stateful_mailbox: StatefulAppMailbox<E>,
     indexer_producer: Option<indexer::Producer>,
+    indexer_producer_handle: Option<Handle<()>>,
     indexer_consumer: Option<IndexerConsumer<E>>,
 }
 
@@ -234,7 +227,7 @@ where
                 max_supported_mode: MAX_SUPPORTED_MODE,
                 namespace: NAMESPACE.to_vec(),
                 storage_protector: dkg::StorageProtector::new(config.dkg_storage_key),
-                epoch_length: BLOCKS_PER_EPOCH,
+                epoch_length: config.epoch_length,
             },
         );
 
@@ -251,88 +244,36 @@ where
         );
 
         let start = Instant::now();
-        let finalizations_by_height = immutable::Archive::init(
-            context.child("finalizations_by_height"),
-            immutable::Config {
-                metadata_partition: format!(
-                    "{}-finalizations-by-height-metadata",
-                    config.partition_prefix
-                ),
-                freezer_table_partition: format!(
-                    "{}-finalizations-by-height-freezer-table",
-                    config.partition_prefix
-                ),
-                freezer_table_initial_size: config.finalized_freezer_table_initial_size,
-                freezer_table_resize_frequency: FREEZER_TABLE_RESIZE_FREQUENCY,
-                freezer_table_resize_chunk_size: FREEZER_TABLE_RESIZE_CHUNK_SIZE,
-                freezer_key_partition: format!(
-                    "{}-finalizations-by-height-freezer-key",
-                    config.partition_prefix
-                ),
-                freezer_key_page_cache: page_cache.clone(),
-                freezer_key_write_buffer: WRITE_BUFFER,
-                freezer_value_partition: format!(
-                    "{}-finalizations-by-height-freezer-value",
-                    config.partition_prefix
-                ),
-                freezer_value_write_buffer: WRITE_BUFFER,
-                freezer_value_target_size: FREEZER_VALUE_TARGET_SIZE,
-                freezer_value_compression: FREEZER_VALUE_COMPRESSION,
-                ordinal_partition: format!(
-                    "{}-finalizations-by-height-ordinal",
-                    config.partition_prefix
-                ),
-                ordinal_write_buffer: WRITE_BUFFER,
-                items_per_section: IMMUTABLE_ITEMS_PER_SECTION,
-                codec_config: Scheme::certificate_codec_config_unbounded(),
-                replay_buffer: REPLAY_BUFFER,
-            },
-        )
-        .await
-        .expect("failed to initialize finalizations by height archive");
-        info!(elapsed = ?start.elapsed(), "restored finalizations by height archive");
-
-        let start = Instant::now();
-        let finalized_blocks = immutable::Archive::init(
-            context.child("finalized_blocks"),
-            immutable::Config {
-                metadata_partition: format!(
-                    "{}-finalized_blocks-metadata",
-                    config.partition_prefix
-                ),
-                freezer_table_partition: format!(
-                    "{}-finalized_blocks-freezer-table",
-                    config.partition_prefix
-                ),
-                freezer_table_initial_size: config.blocks_freezer_table_initial_size,
-                freezer_table_resize_frequency: FREEZER_TABLE_RESIZE_FREQUENCY,
-                freezer_table_resize_chunk_size: FREEZER_TABLE_RESIZE_CHUNK_SIZE,
-                freezer_key_partition: format!(
-                    "{}-finalized_blocks-freezer-key",
-                    config.partition_prefix
-                ),
-                freezer_key_page_cache: page_cache.clone(),
-                freezer_key_write_buffer: WRITE_BUFFER,
-                freezer_value_partition: format!(
-                    "{}-finalized_blocks-freezer-value",
-                    config.partition_prefix
-                ),
-                freezer_value_write_buffer: WRITE_BUFFER,
-                freezer_value_target_size: FREEZER_VALUE_TARGET_SIZE,
-                freezer_value_compression: FREEZER_VALUE_COMPRESSION,
-                ordinal_partition: format!("{}-finalized_blocks-ordinal", config.partition_prefix),
-                ordinal_write_buffer: WRITE_BUFFER,
-                items_per_section: IMMUTABLE_ITEMS_PER_SECTION,
-                codec_config: block_codec_config,
-                replay_buffer: REPLAY_BUFFER,
-            },
+        let opened_history = history::open(
+            &context,
+            &config.partition_prefix,
+            page_cache.clone(),
+            block_codec_config,
         )
         .await
-        .expect("failed to initialize finalized blocks archive");
-        info!(elapsed = ?start.elapsed(), "restored finalized blocks archive");
+        .unwrap_or_else(|error| panic!("failed to initialize bounded finalized history: {error}"));
+        let history::Opened {
+            finalizations: mut finalizations_by_height,
+            blocks: mut finalized_blocks,
+            partitions: history_partitions,
+            marker_status,
+        } = opened_history;
+        info!(
+            elapsed = ?start.elapsed(),
+            format_version = history::FORMAT_VERSION,
+            finalizations_key = %history_partitions.finalizations_key,
+            finalizations_value = %history_partitions.finalizations_value,
+            blocks_key = %history_partitions.blocks_key,
+            blocks_value = %history_partitions.blocks_value,
+            logical_retention = history::LOGICAL_RETENTION,
+            items_per_section = history::PRUNABLE_ITEMS_PER_SECTION.get(),
+            maximum_section_and_cadence_slack = history::MAX_RETAINED_HEIGHTS - history::LOGICAL_RETENTION,
+            ?marker_status,
+            "restored prunable finalized history"
+        );
 
         let recovered_height = Certificates::last_index(&finalizations_by_height);
-        let mut recovered_floor = if let Some(height) = recovered_height {
+        let recovered_floor = if let Some(height) = recovered_height {
             Certificates::get(
                 &finalizations_by_height,
                 ArchiveIdentifier::Index(height.get()),
@@ -425,15 +366,35 @@ where
             .expect("failed to initialize state database for startup probe");
             state.sync_target()
         };
-        if repair_marshal_progress_if_state_is_behind(
+        if let Some(processed_height) = validate_marshal_progress_against_state(
             &context,
             &config.partition_prefix,
             &finalized_blocks,
             &current_state_target,
         )
-        .await
+        .await {
+            validate_history_through_processed(
+                &finalizations_by_height,
+                &finalized_blocks,
+                processed_height,
+            )
+            .await;
+            if let Some(tip) = ArchiveStore::last_index(&finalized_blocks) {
+                let policy_floor = history::policy_floor(tip, history::LOGICAL_RETENTION);
+                let safe_floor = policy_floor.min(processed_height.get());
+                finalizations_by_height
+                    .prune(safe_floor)
+                    .await
+                    .expect("failed to startup-prune finalized certificates");
+                finalized_blocks
+                    .prune(safe_floor)
+                    .await
+                    .expect("failed to startup-prune finalized blocks");
+            }
+        } else if ArchiveStore::last_index(&finalized_blocks).is_some()
+            || ArchiveStore::last_index(&finalizations_by_height).is_some()
         {
-            recovered_floor = None;
+            panic!("finalized history exists without persisted marshal processed height; rebuild or perform verified peer state sync");
         }
         let applied_height = Arc::new(AsyncMutex::new(Height::zero()));
         let app = Application::with_consensus(
@@ -497,7 +458,7 @@ where
             finalized_blocks,
             marshal::Config {
                 provider: provider.clone(),
-                epocher: FixedEpocher::new(BLOCKS_PER_EPOCH),
+                epocher: FixedEpocher::new(config.epoch_length),
                 start: marshal_start,
                 partition_prefix: format!("{}_marshal", config.partition_prefix),
                 mailbox_size: MAILBOX_SIZE,
@@ -551,10 +512,10 @@ where
                 APPLICATION_VERIFY_CONCURRENCY,
             ),
             marshal_mailbox.clone(),
-            FixedEpocher::new(BLOCKS_PER_EPOCH),
+            FixedEpocher::new(config.epoch_length),
         ));
 
-        let (indexer_producer, indexer_pusher, indexer_consumer) =
+        let (indexer_producer, indexer_producer_handle, indexer_pusher, indexer_consumer) =
             if let Some(client) = config.indexer.clone() {
                 let indexer_context = context.child("indexer");
                 let indexer_metrics = indexer::IndexerMetrics::register(&indexer_context);
@@ -562,10 +523,13 @@ where
                 let queue = queue::shared::init(
                     context.child("indexer_queue"),
                     queue::Config {
-                        partition: format!("{}-indexer-finalized-queue", config.partition_prefix),
-                        items_per_section: NZU64!(128),
+                        partition: format!(
+                            "{}-indexer-finalized-payload-queue-v1",
+                            config.partition_prefix
+                        ),
+                        items_per_section: indexer::SPOOL_ITEMS_PER_SECTION,
                         compression: None,
-                        codec_config: (),
+                        codec_config: block_codec_config,
                         page_cache: page_cache.clone(),
                         write_buffer: WRITE_BUFFER,
                     },
@@ -581,14 +545,20 @@ where
                         mailbox_size: MAILBOX_SIZE,
                         backfiller_max_active: commonware_utils::NZUsize!(16),
                         backfiller_retry: Duration::from_millis(500),
+                        spool_limits: config.indexer_spool_limits,
                         metrics: indexer_metrics,
                     },
                 )
                 .await;
-                let (producer, pusher, consumer) = indexer.split();
-                (Some(producer), Some(pusher), Some(consumer))
+                let (producer, producer_handle, pusher, consumer) = indexer.split();
+                (
+                    Some(producer),
+                    Some(producer_handle),
+                    Some(pusher),
+                    Some(consumer),
+                )
             } else {
-                (None, None, None)
+                (None, None, None, None)
             };
 
         let (orchestrator, orchestrator_mailbox) = orchestrator::Actor::new(
@@ -605,7 +575,7 @@ where
                 muxer_size: MAILBOX_SIZE.get(),
                 mailbox_size: MAILBOX_SIZE,
                 partition_prefix: format!("{}_consensus", config.partition_prefix),
-                epoch_length: BLOCKS_PER_EPOCH,
+                epoch_length: config.epoch_length,
                 genesis_digest,
                 recovered_floor,
                 _phantom: PhantomData,
@@ -629,6 +599,7 @@ where
             stateful,
             stateful_mailbox,
             indexer_producer,
+            indexer_producer_handle,
             indexer_consumer,
         };
         (engine, node_handle)
@@ -747,10 +718,13 @@ where
             .mempool
             .start_p2p(self.context.child("mempool"), mempool);
         let indexer_consumer_handle = self.indexer_consumer.map(indexer::Consumer::start);
+        let indexer_producer_handle = self.indexer_producer_handle;
         let clob_handle = self.clob.start_p2p(self.context.child("clob"), clob);
 
         let mut shutdown = self.context.stopped();
-        if let Some(indexer_consumer_handle) = indexer_consumer_handle {
+        if let (Some(indexer_producer_handle), Some(indexer_consumer_handle)) =
+            (indexer_producer_handle, indexer_consumer_handle)
+        {
             commonware_macros::select! {
                 stopped = &mut shutdown => match stopped {
                     Ok(0) => {
@@ -769,6 +743,7 @@ where
                 result = orchestrator_handle => unexpected_exit("orchestrator", result),
                 result = mempool_handle => unexpected_exit("mempool", result),
                 result = clob_handle => unexpected_exit("clob", result),
+                result = indexer_producer_handle => unexpected_exit("indexer_producer", result),
                 result = indexer_consumer_handle => unexpected_exit("indexer_consumer", result),
             }
         } else {
@@ -806,14 +781,13 @@ fn unexpected_exit(
 }
 
 const MARSHAL_PROCESSED_KEY: U64 = U64::new(0xFF);
-const STATE_SYNC_METADATA_SUFFIX: &str = "state_sync_metadata";
 
-async fn repair_marshal_progress_if_state_is_behind<E>(
+async fn validate_marshal_progress_against_state<E>(
     context: &E,
     partition_prefix: &str,
     finalized_blocks: &BlocksArchive<E>,
     current_target: &commonware_storage::qmdb::sync::Target<commonware_storage::mmr::Family, Digest>,
-) -> bool
+) -> Option<Height>
 where
     E: BufferPooler
         + Clock
@@ -837,63 +811,87 @@ where
     )
     .await
     .expect("failed to initialize marshal progress metadata probe");
-    let Some(processed_height) = metadata.get(&MARSHAL_PROCESSED_KEY).copied() else {
-        return false;
-    };
+    let processed_height = metadata.get(&MARSHAL_PROCESSED_KEY).copied()?;
     drop(metadata);
 
-    let Some(processed_block) = finalized_blocks
-        .get(ArchiveIdentifier::Index(processed_height.get()))
-        .await
-        .expect("failed to read processed block for state repair")
-    else {
-        warn!(
-            %processed_height,
-            "marshal progress references a missing finalized block; clearing startup metadata to replay"
-        );
-        remove_partition_if_exists(context, &marshal_metadata_partition).await;
-        clear_disabled_state_sync_metadata(context, partition_prefix).await;
-        return true;
+    let Some(tip) = ArchiveStore::last_index(finalized_blocks) else {
+        panic!("marshal progress references height {processed_height}, but finalized block history is empty; rebuild or perform verified peer state sync");
     };
-    let processed_target = <Application as StatefulApplication<E>>::sync_targets(&processed_block);
-    let current_size = current_target.range.end().as_u64();
-    let processed_size = processed_target.range.end().as_u64();
-    if current_size >= processed_size {
-        return false;
+    let search_end = tip.min(
+        processed_height
+            .get()
+            .saturating_add(MAX_PENDING_ACKS.get() as u64),
+    );
+    for height in processed_height.get()..=search_end {
+        let Some(block) = ArchiveStore::get(
+            finalized_blocks,
+            ArchiveIdentifier::Index(height),
+        )
+            .await
+            .expect("failed to read finalized block while validating QMDB startup target")
+        else {
+            continue;
+        };
+        let target = <Application as StatefulApplication<E>>::sync_targets(&block);
+        if &target == current_target {
+            return Some(processed_height);
+        }
     }
 
-    warn!(
-        %processed_height,
-        current_size,
-        processed_size,
-        "state database is behind marshal progress; clearing startup metadata to replay finalized blocks"
+    panic!(
+        "QMDB committed target does not exactly match canonical finalized history from processed height {} through {}; rebuild or perform verified peer state sync",
+        processed_height,
+        search_end,
     );
-    remove_partition_if_exists(context, &marshal_metadata_partition).await;
-    remove_partition_if_exists(
-        context,
-        &format!("{partition_prefix}{STATE_SYNC_METADATA_SUFFIX}"),
-    )
-    .await;
-    true
 }
 
-async fn clear_disabled_state_sync_metadata<E>(context: &E, partition_prefix: &str)
+async fn validate_history_through_processed<E>(
+    finalizations: &FinalizationsArchive<E>,
+    blocks: &BlocksArchive<E>,
+    processed_height: Height,
+)
 where
-    E: Storage,
+    E: BufferPooler + Metrics + Storage,
 {
-    remove_partition_if_exists(
-        context,
-        &format!("{partition_prefix}{STATE_SYNC_METADATA_SUFFIX}"),
-    )
-    .await;
-}
+    let processed = processed_height.get();
+    let first = match (finalizations.first_index(), blocks.first_index()) {
+        (Some(finalization), Some(block)) => finalization.max(block),
+        _ => panic!(
+            "marshal processed height {processed_height} has incomplete finalized history; rebuild or perform verified peer state sync"
+        ),
+    };
+    if first > processed {
+        panic!(
+            "marshal processed height {processed_height} is below retained finalized history floor {first}; rebuild or perform verified peer state sync"
+        );
+    }
 
-async fn remove_partition_if_exists<E>(context: &E, partition: &str)
-where
-    E: Storage,
-{
-    match context.remove(partition, None).await {
-        Ok(()) | Err(commonware_runtime::Error::PartitionMissing(_)) => {}
-        Err(error) => panic!("failed to remove partition {partition}: {error}"),
+    for height in first..=processed {
+        let certificate = finalizations
+            .get(ArchiveIdentifier::Index(height))
+            .await
+            .expect("failed to validate finalized certificate archive");
+        let block = blocks
+            .get(ArchiveIdentifier::Index(height))
+            .await
+            .expect("failed to validate finalized block archive");
+        if height == processed {
+            assert!(
+                block.is_some(),
+                "marshal processed height {processed_height} references a missing finalized block; rebuild or perform verified peer state sync"
+            );
+        }
+        if let (Some(certificate), Some(block)) = (certificate, block) {
+            assert_eq!(
+                certificate.proposal.payload,
+                block.digest(),
+                "finalized block and certificate conflict at height {height}"
+            );
+            assert_eq!(
+                certificate.proposal.round.epoch(),
+                block.context.round.epoch(),
+                "finalized block and certificate epochs conflict at height {height}"
+            );
+        }
     }
 }
diff --git a/examples/coins/chain/src/history.rs b/examples/coins/chain/src/history.rs
new file mode 100644
index 0000000..6d39607
--- /dev/null
+++ b/examples/coins/chain/src/history.rs
@@ -0,0 +1,466 @@
+//! Bounded finalized-history storage for coins-chain validators.
+
+use crate::{Block, Finalization};
+use commonware_codec::Read;
+use commonware_cryptography::sha256::Digest;
+use commonware_runtime::{buffer::paged::CacheRef, BufferPooler, Clock, Metrics, Spawner, Storage};
+use commonware_storage::{
+    archive::{prunable, Archive as _},
+    metadata::{self, Metadata},
+    translator::EightCap,
+};
+use commonware_utils::{sequence::U64, NZU64};
+use nunchi_chain::engine::{
+    MAX_PENDING_ACKS, PRUNE_MAINTENANCE_INTERVAL, PRUNE_RETAINED_MARSHAL_BLOCKS, REPLAY_BUFFER,
+    WRITE_BUFFER,
+};
+use std::num::NonZeroU64;
+
+pub(crate) const FORMAT_VERSION: u64 = 1;
+pub(crate) const PRUNABLE_ITEMS_PER_SECTION: NonZeroU64 = NZU64!(4_096);
+pub(crate) const LOGICAL_RETENTION: u64 =
+    PRUNE_RETAINED_MARSHAL_BLOCKS as u64 + MAX_PENDING_ACKS.get() as u64 + 1;
+pub(crate) const MAX_RETAINED_HEIGHTS: u64 = LOGICAL_RETENTION
+    + PRUNABLE_ITEMS_PER_SECTION.get()
+    - 1
+    + PRUNE_MAINTENANCE_INTERVAL.get() as u64
+    - 1;
+
+const MARKER_KEY: U64 = U64::new(0);
+
+pub(crate) type FinalizationsArchive<E> =
+    prunable::Archive<EightCap, E, Digest, Finalization>;
+pub(crate) type BlocksArchive<E> = prunable::Archive<EightCap, E, Digest, Block>;
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub(crate) enum MarkerStatus {
+    Created,
+    RecoveredPartialInitialization,
+    Present,
+}
+
+#[derive(Debug, thiserror::Error)]
+pub(crate) enum HistoryError {
+    #[error("legacy finalized-history partition exists: {partition}; stop the devnet and clear the complete validator storage directory before relaunch")]
+    LegacyPartition { partition: String },
+    #[error("history format marker exists but required partition is missing: {partition}; clear the complete devnet validator storage directory before relaunch")]
+    MarkedPartitionMissing { partition: String },
+    #[error("history format marker is missing but {partition} contains finalized data at height {height}; refusing to modify unmarked storage")]
+    NonEmptyUnmarked { partition: String, height: u64 },
+    #[error("invalid history format marker in {partition}: expected version {expected}, found {found:?}")]
+    InvalidMarker {
+        partition: String,
+        expected: u64,
+        found: Option<u64>,
+    },
+    #[error("failed to inspect history partition {partition}: {source}")]
+    Inspect {
+        partition: String,
+        #[source]
+        source: commonware_runtime::Error,
+    },
+    #[error("failed to open history archive partition {partition}: {source}")]
+    Archive {
+        partition: String,
+        #[source]
+        source: commonware_storage::archive::Error,
+    },
+    #[error("failed to access history marker partition {partition}: {source}")]
+    Marker {
+        partition: String,
+        #[source]
+        source: commonware_storage::metadata::Error,
+    },
+    #[error("invalid history retention configuration: {0}")]
+    Retention(&'static str),
+}
+
+#[derive(Clone, Debug)]
+pub(crate) struct Partitions {
+    pub(crate) finalizations_key: String,
+    pub(crate) finalizations_value: String,
+    pub(crate) blocks_key: String,
+    pub(crate) blocks_value: String,
+    pub(crate) metadata: String,
+}
+
+impl Partitions {
+    pub(crate) fn new(prefix: &str) -> Self {
+        Self {
+            finalizations_key: format!(
+                "{prefix}-finalizations-by-height-prunable-v1-key"
+            ),
+            finalizations_value: format!(
+                "{prefix}-finalizations-by-height-prunable-v1-value"
+            ),
+            blocks_key: format!("{prefix}-finalized-blocks-prunable-v1-key"),
+            blocks_value: format!("{prefix}-finalized-blocks-prunable-v1-value"),
+            metadata: format!("{prefix}-history-prunable-v1-metadata"),
+        }
+    }
+
+    fn archives(&self) -> [&str; 4] {
+        [
+            &self.finalizations_key,
+            &self.finalizations_value,
+            &self.blocks_key,
+            &self.blocks_value,
+        ]
+    }
+}
+
+pub(crate) struct Opened<E>
+where
+    E: BufferPooler + Clock + Storage + Metrics,
+{
+    pub(crate) finalizations: FinalizationsArchive<E>,
+    pub(crate) blocks: BlocksArchive<E>,
+    pub(crate) partitions: Partitions,
+    pub(crate) marker_status: MarkerStatus,
+}
+
+pub(crate) const fn policy_floor(tip: u64, retained: u64) -> u64 {
+    tip.saturating_add(1).saturating_sub(retained)
+}
+
+pub(crate) fn validate_production_retention() -> Result<(), HistoryError> {
+    if MAX_PENDING_ACKS.get() != 16 {
+        return Err(HistoryError::Retention(
+            "MAX_PENDING_ACKS changed; update the mandatory rewind assertion",
+        ));
+    }
+    if LOGICAL_RETENTION != 217 {
+        return Err(HistoryError::Retention(
+            "logical finalized-history retention must be 217",
+        ));
+    }
+    if MAX_RETAINED_HEIGHTS != 4_343 {
+        return Err(HistoryError::Retention(
+            "section-granular finalized-history bound must be 4,343",
+        ));
+    }
+    Ok(())
+}
+
+pub(crate) async fn open<E>(
+    context: &E,
+    prefix: &str,
+    page_cache: CacheRef,
+    block_codec_config: <Block as Read>::Cfg,
+) -> Result<Opened<E>, HistoryError>
+where
+    E: BufferPooler + Clock + Spawner + Storage + Metrics,
+{
+    validate_production_retention()?;
+    reject_legacy(context, prefix).await?;
+
+    let partitions = Partitions::new(prefix);
+    let marker_existed = partition_exists(context, &partitions.metadata).await?;
+    let mut archive_existed = [false; 4];
+    for (exists, partition) in archive_existed
+        .iter_mut()
+        .zip(partitions.archives())
+    {
+        *exists = partition_exists(context, partition).await?;
+    }
+    let mut finalizations =
+        init_finalizations(context, &partitions, page_cache.clone()).await?;
+    let mut blocks = init_blocks(
+        context,
+        &partitions,
+        page_cache.clone(),
+        block_codec_config,
+    )
+    .await?;
+
+    let marker_status = if marker_existed {
+        validate_marker(context, &partitions.metadata).await?;
+        let archives_empty = finalizations.first_index().is_none() && blocks.first_index().is_none();
+        let all_lazy = archive_existed.iter().all(|exists| !exists);
+        if !(archive_existed.iter().all(|exists| *exists) || archives_empty && all_lazy) {
+            let partition = partitions
+                .archives()
+                .into_iter()
+                .zip(archive_existed)
+                .find_map(|(partition, exists)| (!exists).then_some(partition))
+                .expect("at least one marked archive partition is missing");
+            return Err(HistoryError::MarkedPartitionMissing {
+                partition: partition.to_string(),
+            });
+        }
+        MarkerStatus::Present
+    } else {
+        if let Some(height) = finalizations.first_index() {
+            return Err(HistoryError::NonEmptyUnmarked {
+                partition: partitions.finalizations_key.clone(),
+                height,
+            });
+        }
+        if let Some(height) = blocks.first_index() {
+            return Err(HistoryError::NonEmptyUnmarked {
+                partition: partitions.blocks_key.clone(),
+                height,
+            });
+        }
+
+        // Ensure all newly-created archive blobs are durable and structurally reopenable before
+        // committing the format marker.
+        finalizations.sync().await.map_err(|source| HistoryError::Archive {
+            partition: partitions.finalizations_key.clone(),
+            source,
+        })?;
+        blocks.sync().await.map_err(|source| HistoryError::Archive {
+            partition: partitions.blocks_key.clone(),
+            source,
+        })?;
+        drop(finalizations);
+        drop(blocks);
+        finalizations = init_finalizations(context, &partitions, page_cache.clone()).await?;
+        blocks = init_blocks(context, &partitions, page_cache, block_codec_config).await?;
+        write_marker(context, &partitions.metadata).await?;
+
+        if archive_existed.into_iter().any(|exists| exists) {
+            MarkerStatus::RecoveredPartialInitialization
+        } else {
+            MarkerStatus::Created
+        }
+    };
+
+    Ok(Opened {
+        finalizations,
+        blocks,
+        partitions,
+        marker_status,
+    })
+}
+
+async fn init_finalizations<E>(
+    context: &E,
+    partitions: &Partitions,
+    page_cache: CacheRef,
+) -> Result<FinalizationsArchive<E>, HistoryError>
+where
+    E: BufferPooler + Clock + Spawner + Storage + Metrics,
+{
+    prunable::Archive::init(
+        context.child("finalizations_by_height"),
+        prunable::Config {
+            translator: EightCap,
+            key_partition: partitions.finalizations_key.clone(),
+            key_page_cache: page_cache,
+            value_partition: partitions.finalizations_value.clone(),
+            compression: Some(3),
+            codec_config: (),
+            items_per_section: PRUNABLE_ITEMS_PER_SECTION,
+            key_write_buffer: WRITE_BUFFER,
+            value_write_buffer: WRITE_BUFFER,
+            replay_buffer: REPLAY_BUFFER,
+        },
+    )
+    .await
+    .map_err(|source| HistoryError::Archive {
+        partition: partitions.finalizations_key.clone(),
+        source,
+    })
+}
+
+async fn init_blocks<E>(
+    context: &E,
+    partitions: &Partitions,
+    page_cache: CacheRef,
+    codec_config: <Block as Read>::Cfg,
+) -> Result<BlocksArchive<E>, HistoryError>
+where
+    E: BufferPooler + Clock + Spawner + Storage + Metrics,
+{
+    prunable::Archive::init(
+        context.child("finalized_blocks"),
+        prunable::Config {
+            translator: EightCap,
+            key_partition: partitions.blocks_key.clone(),
+            key_page_cache: page_cache,
+            value_partition: partitions.blocks_value.clone(),
+            compression: Some(3),
+            codec_config,
+            items_per_section: PRUNABLE_ITEMS_PER_SECTION,
+            key_write_buffer: WRITE_BUFFER,
+            value_write_buffer: WRITE_BUFFER,
+            replay_buffer: REPLAY_BUFFER,
+        },
+    )
+    .await
+    .map_err(|source| HistoryError::Archive {
+        partition: partitions.blocks_key.clone(),
+        source,
+    })
+}
+
+async fn partition_exists<E: Storage>(context: &E, partition: &str) -> Result<bool, HistoryError> {
+    match context.scan(partition).await {
+        Ok(_) => Ok(true),
+        Err(commonware_runtime::Error::PartitionMissing(_)) => Ok(false),
+        Err(source) => Err(HistoryError::Inspect {
+            partition: partition.to_string(),
+            source,
+        }),
+    }
+}
+
+async fn validate_marker<E>(context: &E, partition: &str) -> Result<(), HistoryError>
+where
+    E: BufferPooler + Clock + Storage + Metrics + Spawner,
+{
+    let marker = Metadata::<E, U64, u64>::init(
+        context.child("history_marker"),
+        metadata::Config {
+            partition: partition.to_string(),
+            codec_config: (),
+        },
+    )
+    .await
+    .map_err(|source| HistoryError::Marker {
+        partition: partition.to_string(),
+        source,
+    })?;
+    let found = marker.get(&MARKER_KEY).copied();
+    if found != Some(FORMAT_VERSION) {
+        return Err(HistoryError::InvalidMarker {
+            partition: partition.to_string(),
+            expected: FORMAT_VERSION,
+            found,
+        });
+    }
+    Ok(())
+}
+
+async fn write_marker<E>(context: &E, partition: &str) -> Result<(), HistoryError>
+where
+    E: BufferPooler + Clock + Storage + Metrics + Spawner,
+{
+    let mut marker = Metadata::<E, U64, u64>::init(
+        context.child("history_marker"),
+        metadata::Config {
+            partition: partition.to_string(),
+            codec_config: (),
+        },
+    )
+    .await
+    .map_err(|source| HistoryError::Marker {
+        partition: partition.to_string(),
+        source,
+    })?;
+    marker
+        .put_sync(MARKER_KEY, FORMAT_VERSION)
+        .await
+        .map_err(|source| HistoryError::Marker {
+            partition: partition.to_string(),
+            source,
+        })
+}
+
+async fn reject_legacy<E: Storage>(context: &E, prefix: &str) -> Result<(), HistoryError> {
+    const FINALIZATION_SUFFIXES: [&str; 5] = [
+        "finalizations-by-height-metadata",
+        "finalizations-by-height-freezer-table",
+        "finalizations-by-height-freezer-key",
+        "finalizations-by-height-freezer-value",
+        "finalizations-by-height-ordinal",
+    ];
+    const BLOCK_SUFFIXES: [&str; 5] = [
+        "finalized_blocks-metadata",
+        "finalized_blocks-freezer-table",
+        "finalized_blocks-freezer-key",
+        "finalized_blocks-freezer-value",
+        "finalized_blocks-ordinal",
+    ];
+
+    for suffix in FINALIZATION_SUFFIXES.into_iter().chain(BLOCK_SUFFIXES) {
+        let partition = format!("{prefix}-{suffix}");
+        if partition_exists(context, &partition).await? {
+            return Err(HistoryError::LegacyPartition { partition });
+        }
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use commonware_runtime::{buffer::paged::CacheRef, deterministic, Runner as _};
+    use commonware_utils::{NZU16, NZU32, NZUsize};
+
+    #[test]
+    fn production_retention_arithmetic() {
+        validate_production_retention().unwrap();
+        assert_eq!(policy_floor(216, LOGICAL_RETENTION), 0);
+        assert_eq!(policy_floor(217, LOGICAL_RETENTION), 1);
+        assert_eq!(LOGICAL_RETENTION, 217);
+        assert_eq!(MAX_RETAINED_HEIGHTS, 4_343);
+    }
+
+    #[test]
+    fn fresh_format_marker_is_durable_and_legacy_is_not_created() {
+        deterministic::Runner::default().start(|context| async move {
+            let page_cache = CacheRef::from_pooler(&context, NZU16!(4_096), NZUsize!(32));
+            let first = open(
+                &context,
+                "validator",
+                page_cache.clone(),
+                (NZU32!(1), ()),
+            )
+            .await
+            .expect("initialize fresh history");
+            assert_eq!(first.marker_status, MarkerStatus::Created);
+            assert_eq!(first.finalizations.first_index(), None);
+            assert_eq!(first.blocks.first_index(), None);
+            drop(first);
+
+            let reopened = open(
+                &context,
+                "validator",
+                page_cache,
+                (NZU32!(1), ()),
+            )
+            .await
+            .expect("reopen marked history");
+            assert_eq!(reopened.marker_status, MarkerStatus::Present);
+
+            for suffix in [
+                "finalizations-by-height-metadata",
+                "finalizations-by-height-freezer-table",
+                "finalizations-by-height-freezer-key",
+                "finalizations-by-height-freezer-value",
+                "finalizations-by-height-ordinal",
+                "finalized_blocks-metadata",
+                "finalized_blocks-freezer-table",
+                "finalized_blocks-freezer-key",
+                "finalized_blocks-freezer-value",
+                "finalized_blocks-ordinal",
+            ] {
+                assert!(matches!(
+                    context.scan(&format!("validator-{suffix}")).await,
+                    Err(commonware_runtime::Error::PartitionMissing(_))
+                ));
+            }
+        });
+    }
+
+    #[test]
+    fn refuses_known_legacy_partition() {
+        deterministic::Runner::default().start(|context| async move {
+            context
+                .open("validator-finalized_blocks-ordinal", b"legacy")
+                .await
+                .expect("create legacy partition");
+            let page_cache = CacheRef::from_pooler(&context, NZU16!(4_096), NZUsize!(32));
+            let result = open(
+                &context,
+                "validator",
+                page_cache,
+                (NZU32!(1), ()),
+            )
+            .await;
+            assert!(matches!(result, Err(HistoryError::LegacyPartition { .. })));
+        });
+    }
+}
diff --git a/examples/coins/chain/src/indexer/backfiller/consumer.rs b/examples/coins/chain/src/indexer/backfiller/consumer.rs
index e714eb2..0bd66e8 100644
--- a/examples/coins/chain/src/indexer/backfiller/consumer.rs
+++ b/examples/coins/chain/src/indexer/backfiller/consumer.rs
@@ -1,17 +1,14 @@
-use super::{Decision, Entry, SharedState};
+use super::{
+    producer::{Admission, AdmissionReceiver},
+    Decision, Entry, SharedState,
+};
 use crate::indexer::{
     metrics::{
-        estimated_finalized_bytes, BackfillDecision, BackfillPhase, BackfillResetReason,
-        BackfillUploadGuard, BackfillWaitReason, BlockMetricSource, QueueReadSource, QueueStatus,
-        SharedCacheSource,
+        estimated_finalized_bytes, BackfillDecision, BackfillPhase, BackfillWaitReason,
+        BlockMetricSource, ProducerStatus, QueueReadSource, QueueStatus,
     },
-    Client, IndexerMetrics,
-};
-use crate::{Block, Finalized, Scheme};
-use commonware_consensus::marshal::{
-    core::Mailbox as MarshalMailbox, standard::Standard, Identifier,
+    Client, IndexerMetrics, SpoolLimits,
 };
-use commonware_consensus::types::Height;
 use commonware_cryptography::sha256::Digest;
 use commonware_macros::select_loop;
 use commonware_runtime::{
@@ -22,7 +19,7 @@ use commonware_storage::queue;
 use commonware_utils::futures::{OptionFuture, Pool};
 use std::{
     num::NonZeroUsize,
-    time::{Duration, Instant},
+    time::{Duration, Instant, UNIX_EPOCH},
 };
 use tracing::{debug, warn};
 
@@ -45,47 +42,39 @@ enum Completion {
     Skipped {
         position: u64,
         height: u64,
-    },
-    Stale {
-        reason: BackfillResetReason,
-        first_height: u64,
-        first_digest: Digest,
-        attempts: u64,
-        elapsed: Duration,
+        digest: Digest,
     },
 }
 
 pub(crate) struct Config {
     pub(crate) max_active: NonZeroUsize,
     pub(crate) retry: Duration,
-    pub(crate) missing_finalization_grace: Duration,
-    pub(crate) mismatched_finalization_grace: Duration,
+    pub(crate) spool_limits: SpoolLimits,
 }
 
 pub struct Consumer<E: BufferPooler + Spawner + Clock + Storage + Metrics, C: Client> {
     context: ContextCell<E>,
     client: C,
-    marshal: MarshalMailbox<Scheme, Standard<Block>>,
     metrics: IndexerMetrics,
     upload_results: status::Counter,
     uploads: SharedState,
     writer: queue::Writer<E, Entry>,
     reader: queue::Reader<E, Entry>,
+    admission: AdmissionReceiver,
     active: Pool<Completion>,
     max_active: NonZeroUsize,
     retry: Duration,
-    missing_finalization_grace: Duration,
-    mismatched_finalization_grace: Duration,
+    spool_limits: SpoolLimits,
 }
 
 impl<E: BufferPooler + Spawner + Clock + Storage + Metrics, C: Client> Consumer<E, C> {
     pub fn new(
         context: E,
         client: C,
-        marshal: MarshalMailbox<Scheme, Standard<Block>>,
         metrics: IndexerMetrics,
         uploads: SharedState,
         backfiller: (queue::Writer<E, Entry>, queue::Reader<E, Entry>),
+        admission: AdmissionReceiver,
         config: Config,
     ) -> Self {
         let upload_results = context.register(
@@ -96,25 +85,23 @@ impl<E: BufferPooler + Spawner + Clock + Storage + Metrics, C: Client> Consumer<
         let Config {
             max_active,
             retry,
-            missing_finalization_grace,
-            mismatched_finalization_grace,
+            spool_limits,
         } = config;
         metrics.backfill_configured(max_active.get());
         let (writer, reader) = backfiller;
         Self {
             context: ContextCell::new(context),
             client,
-            marshal,
             metrics,
             upload_results,
             uploads,
             writer,
             reader,
+            admission,
             active: Pool::default(),
             max_active,
             retry,
-            missing_finalization_grace,
-            mismatched_finalization_grace,
+            spool_limits,
         }
     }
 
@@ -127,25 +114,24 @@ impl<E: BufferPooler + Spawner + Clock + Storage + Metrics, C: Client> Consumer<
             self.context,
             on_start => {
                 self.fill_slots().await;
-                if self.active.is_empty() {
-                    let item = Self::recv_queue(&mut self.reader, self.metrics.clone()).await;
-                    let Some((position, entry)) = item else {
-                        warn!("consumer queue closed");
-                        break;
-                    };
-                    self.start_upload(position, entry).await;
-                    continue;
-                }
                 let item = OptionFuture::from(
                     (self.active.len() < self.max_active.get()).then(|| {
                         Self::recv_queue(&mut self.reader, self.metrics.clone())
                     }),
                 );
+                let admission = self.admission.recv();
             },
             on_stopped => {},
             completion = self.active.next_completed() => {
                 self.complete(completion).await;
             },
+            request = admission => {
+                let Some(request) = request else {
+                    warn!("indexer spool admission coordinator closed");
+                    break;
+                };
+                self.admit(request).await;
+            },
             item = item => {
                 match item {
                     Some((position, entry)) => {
@@ -160,6 +146,87 @@ impl<E: BufferPooler + Spawner + Clock + Storage + Metrics, C: Client> Consumer<
         }
     }
 
+    async fn admit(&mut self, admission: Admission) {
+        let Admission { entry, response } = admission;
+        loop {
+            let now_millis: u64 = self
+                .context
+                .current()
+                .duration_since(UNIX_EPOCH)
+                .unwrap_or_default()
+                .as_millis()
+                .try_into()
+                .unwrap_or(u64::MAX);
+            let next = {
+                let uploads = self.uploads.lock();
+                let (entries, bytes) = uploads.spool_usage();
+                uploads.oldest_queued().and_then(
+                    |(digest, position, height, old_bytes, enqueued_at)| {
+                        let age = Duration::from_millis(
+                            now_millis.saturating_sub(enqueued_at),
+                        );
+                        let status = if age >= self.spool_limits.max_age {
+                            Some(ProducerStatus::ExpiredAge)
+                        } else if entries >= self.spool_limits.max_entries {
+                            Some(ProducerStatus::ExpiredEntries)
+                        } else if bytes.saturating_add(entry.encoded_len)
+                            > self.spool_limits.max_bytes
+                        {
+                            Some(ProducerStatus::ExpiredBytes)
+                        } else {
+                            None
+                        };
+                        status.map(|status| {
+                            (digest, position, height, old_bytes, age, status)
+                        })
+                    },
+                )
+            };
+            let Some((digest, position, height, encoded_len, age, status)) = next else {
+                break;
+            };
+
+            // Advancing the floor invalidates positions held by upload tasks. Cancel all of them
+            // and replay the still-live suffix so a stale completion cannot be applied twice.
+            self.active.cancel_all();
+            self.reader
+                .ack_up_to(position.saturating_add(1))
+                .await
+                .unwrap_or_else(|error| panic!("failed to expire indexer spool floor: {error:?}"));
+            let sync_started = Instant::now();
+            self.writer.sync().await.unwrap_or_else(|error| {
+                panic!("failed to durably sync expired indexer spool floor: {error:?}")
+            });
+            self.metrics
+                .queue_synced(QueueStatus::Success, sync_started.elapsed());
+            self.reader.reset().await;
+            let expired = self.uploads.lock().expire_through(position);
+            for (_, _, _) in &expired {
+                self.metrics.producer_recorded(status, Duration::ZERO);
+            }
+            warn!(
+                height,
+                ?digest,
+                encoded_len,
+                ?age,
+                ?status,
+                expired_entries = expired.len(),
+                "terminally expired oldest finalized indexer payload before admission"
+            );
+        }
+
+        self.metrics.queue_entry(entry.height());
+        let position = self.writer.enqueue(entry.clone()).await.unwrap_or_else(|error| {
+            self.metrics.queue_enqueued(QueueStatus::Failure);
+            panic!("failed to enqueue finalized indexer payload: {error:?}")
+        });
+        self.uploads.lock().mark_queued(position, &entry);
+        self.metrics.queue_enqueued(QueueStatus::Success);
+        response
+            .send(())
+            .expect("indexer spool producer stopped before admission completed");
+    }
+
     async fn fill_slots(&mut self) {
         while self.active.len() < self.max_active.get() {
             let item = Self::try_recv_queue(&mut self.reader, self.metrics.clone()).await;
@@ -177,7 +244,7 @@ impl<E: BufferPooler + Spawner + Clock + Storage + Metrics, C: Client> Consumer<
         match reader.recv().await {
             Ok(Some((position, entry))) => {
                 metrics.queue_read(QueueReadSource::Recv, QueueStatus::Success);
-                metrics.queue_entry(entry.height);
+                metrics.queue_entry(entry.height());
                 Some((position, entry))
             }
             Ok(None) => {
@@ -198,7 +265,7 @@ impl<E: BufferPooler + Spawner + Clock + Storage + Metrics, C: Client> Consumer<
         match reader.try_recv().await {
             Ok(Some((position, entry))) => {
                 metrics.queue_read(QueueReadSource::TryRecv, QueueStatus::Success);
-                metrics.queue_entry(entry.height);
+                metrics.queue_entry(entry.height());
                 Some((position, entry))
             }
             Ok(None) => {
@@ -213,7 +280,26 @@ impl<E: BufferPooler + Spawner + Clock + Storage + Metrics, C: Client> Consumer<
     }
 
     async fn start_upload(&mut self, position: u64, entry: Entry) {
-        let Entry { height, digest } = entry;
+        let height = entry.height();
+        let digest = entry.digest();
+        let enqueued_at_millis = entry.enqueued_at_millis;
+        let finalized = entry.finalized;
+        if self.payload_age(enqueued_at_millis) >= self.spool_limits.max_age {
+            self.metrics
+                .producer_recorded(ProducerStatus::ExpiredAge, Duration::ZERO);
+            warn!(
+                height,
+                ?digest,
+                "terminally expiring finalized indexer payload at configured age bound"
+            );
+            self.complete(Completion::Skipped {
+                position,
+                height,
+                digest,
+            })
+            .await;
+            return;
+        }
         let decision = self.uploads.lock().should_upload(&digest);
         self.metrics
             .backfill_decision((&decision).into(), BackfillPhase::Start);
@@ -222,7 +308,11 @@ impl<E: BufferPooler + Spawner + Clock + Storage + Metrics, C: Client> Consumer<
                 let mut upload = self.metrics.start_backfill_upload();
                 upload.skipped();
             }
-            self.complete(Completion::Skipped { position, height })
+            self.complete(Completion::Skipped {
+                position,
+                height,
+                digest,
+            })
                 .await;
             debug!(?digest, "consumer skipping already-uploaded block");
             return;
@@ -235,82 +325,35 @@ impl<E: BufferPooler + Spawner + Clock + Storage + Metrics, C: Client> Consumer<
                 .with_attribute("digest", digest)
                 .with_attribute("height", height);
             let client = self.client.clone();
-            let marshal = self.marshal.clone();
             let metrics = self.metrics.clone();
             let upload_results = self.upload_results.clone();
             let uploads = self.uploads.clone();
             let retry = self.retry;
-            let missing_finalization_grace = self.missing_finalization_grace;
-            let mismatched_finalization_grace = self.mismatched_finalization_grace;
+            let max_age = self.spool_limits.max_age;
             async move {
                 let mut upload = metrics.start_backfill_upload();
-                let Some(block) =
-                    Self::wait_for_uploadable_block(
-                        &context, &marshal, &metrics, &uploads, &mut upload, digest, retry,
-                    )
-                    .await
-                else {
-                    upload.skipped();
-                    debug!(?digest, "skipping previously uploaded block");
-                    return Completion::Skipped { position, height };
-                };
+                metrics.observe_block(BlockMetricSource::ConsumerCached, &finalized.block);
+                upload.hold_block(&finalized.block);
 
-                let mut proof_failure = ProofFailure::default();
-                let finalized = loop {
-                    let Some(proof) = marshal.get_finalization(Height::new(height)).await else {
-                        if let Some(stale) = proof_failure.observe(
-                            &context,
-                            BackfillResetReason::MissingFinalization,
-                            missing_finalization_grace,
-                            height,
-                            digest,
-                        ) {
-                            upload.abandoned();
-                            return stale;
-                        }
-                        warn!(
-                            height,
-                            ?digest,
-                            "consumer could not find finalization, retrying"
-                        );
-                        Self::wait(
-                            &context,
-                            &metrics,
-                            BackfillWaitReason::MissingFinalization,
-                            retry,
-                        )
-                        .await;
-                        continue;
-                    };
-                    if proof.proposal.payload != digest {
-                        if let Some(stale) = proof_failure.observe(
-                            &context,
-                            BackfillResetReason::MismatchedFinalization,
-                            mismatched_finalization_grace,
+                loop {
+                    let now_millis: u64 = context
+                        .current()
+                        .duration_since(UNIX_EPOCH)
+                        .unwrap_or_default()
+                        .as_millis()
+                        .try_into()
+                        .unwrap_or(u64::MAX);
+                    // Backward wall-clock movement produces age zero until the clock catches up.
+                    let age = Duration::from_millis(now_millis.saturating_sub(enqueued_at_millis));
+                    if age >= max_age {
+                        metrics.producer_recorded(ProducerStatus::ExpiredAge, Duration::ZERO);
+                        warn!(height, ?digest, ?age, "terminally expiring active finalized indexer upload at configured age bound");
+                        return Completion::Skipped {
+                            position,
                             height,
                             digest,
-                        ) {
-                            upload.abandoned();
-                            return stale;
-                        }
-                        warn!(
-                            height,
-                            ?digest,
-                            "consumer found mismatched finalization, retrying"
-                        );
-                        Self::wait(
-                            &context,
-                            &metrics,
-                            BackfillWaitReason::MismatchedFinalization,
-                            retry,
-                        )
-                        .await;
-                        continue;
+                        };
                     }
-                    break Finalized::new(proof, block.clone());
-                };
-
-                loop {
                     let decision = {
                         let uploads = uploads.lock();
                         uploads.should_upload(&digest)
@@ -320,7 +363,11 @@ impl<E: BufferPooler + Spawner + Clock + Storage + Metrics, C: Client> Consumer<
                         Decision::Skip => {
                             upload.skipped();
                             debug!(?digest, "skipping previously uploaded block");
-                            return Completion::Skipped { position, height };
+                            return Completion::Skipped {
+                                position,
+                                height,
+                                digest,
+                            };
                         }
                         Decision::Wait => {
                             Self::wait(
@@ -363,86 +410,17 @@ impl<E: BufferPooler + Spawner + Clock + Storage + Metrics, C: Client> Consumer<
         });
     }
 
-    async fn wait_for_uploadable_block(
-        context: &E,
-        marshal: &MarshalMailbox<Scheme, Standard<Block>>,
-        metrics: &IndexerMetrics,
-        uploads: &SharedState,
-        upload: &mut BackfillUploadGuard,
-        digest: Digest,
-        retry: Duration,
-    ) -> Option<Block> {
-        enum NextBlock {
-            AlreadyUploaded,
-            WaitForCertificate,
-            Ready(Box<Block>),
-            FetchFromMarshal,
-        }
-
-        loop {
-            let next = {
-                let uploads = uploads.lock();
-                match uploads.should_upload(&digest) {
-                    Decision::Skip => {
-                        metrics.backfill_decision(
-                            BackfillDecision::Skip,
-                            BackfillPhase::BeforeBlock,
-                        );
-                        NextBlock::AlreadyUploaded
-                    }
-                    Decision::Wait => {
-                        metrics.backfill_decision(
-                            BackfillDecision::Wait,
-                            BackfillPhase::BeforeBlock,
-                        );
-                        NextBlock::WaitForCertificate
-                    }
-                    Decision::Proceed => {
-                        metrics.backfill_decision(
-                            BackfillDecision::Proceed,
-                            BackfillPhase::BeforeBlock,
-                        );
-                        uploads
-                            .cached_block(&digest)
-                            .map(|block| NextBlock::Ready(Box::new(block)))
-                            .unwrap_or(NextBlock::FetchFromMarshal)
-                    }
-                }
-            };
-
-            match next {
-                NextBlock::AlreadyUploaded => return None,
-                NextBlock::WaitForCertificate => {
-                    Self::wait(
-                        context,
-                        metrics,
-                        BackfillWaitReason::CertificateUpload,
-                        retry,
-                    )
-                    .await;
-                }
-                NextBlock::Ready(block) => {
-                    metrics.observe_block(BlockMetricSource::ConsumerCached, &block);
-                    upload.hold_block(&block);
-                    return Some(*block);
-                }
-                NextBlock::FetchFromMarshal => {
-                    if let Some(block) = marshal.get_block(Identifier::Digest(digest)).await {
-                        metrics.observe_block(BlockMetricSource::ConsumerMarshal, &block);
-                        uploads
-                            .lock()
-                            .cache_block(block.clone(), SharedCacheSource::ConsumerMarshal);
-                        upload.hold_block(&block);
-                        return Some(block);
-                    }
-                    warn!(
-                        ?digest,
-                        "consumer could not find block in marshal, retrying"
-                    );
-                    Self::wait(context, metrics, BackfillWaitReason::MissingBlock, retry).await;
-                }
-            }
-        }
+    fn payload_age(&self, enqueued_at_millis: u64) -> Duration {
+        let now_millis: u64 = self
+            .context
+            .current()
+            .duration_since(UNIX_EPOCH)
+            .unwrap_or_default()
+            .as_millis()
+            .try_into()
+            .unwrap_or(u64::MAX);
+        // Backward wall-clock movement produces age zero until the clock catches up.
+        Duration::from_millis(now_millis.saturating_sub(enqueued_at_millis))
     }
 
     async fn wait(
@@ -462,28 +440,22 @@ impl<E: BufferPooler + Spawner + Clock + Storage + Metrics, C: Client> Consumer<
     }
 
     async fn complete(&mut self, completion: Completion) {
-        let (position, height) = match completion {
+        let (position, height, digest) = match completion {
             Completion::Uploaded {
                 position,
                 height,
                 digest,
             } => {
                 self.uploads.lock().mark_uploaded(digest, height);
-                (position, height)
-            }
-            Completion::Skipped { position, height } => (position, height),
-            Completion::Stale {
-                reason,
-                first_height,
-                first_digest,
-                attempts,
-                elapsed,
-            } => {
-                self.reset_stale_queue(reason, first_height, first_digest, attempts, elapsed)
-                    .await;
-                return;
+                (position, height, digest)
             }
+            Completion::Skipped {
+                position,
+                height,
+                digest,
+            } => (position, height, digest),
         };
+        self.uploads.lock().finish_queued(&digest);
 
         let floor = self.reader.ack_floor().await;
         match self.reader.ack(position).await {
@@ -511,109 +483,4 @@ impl<E: BufferPooler + Spawner + Clock + Storage + Metrics, C: Client> Consumer<
         }
     }
 
-    async fn reset_stale_queue(
-        &mut self,
-        reason: BackfillResetReason,
-        first_height: u64,
-        first_digest: Digest,
-        attempts: u64,
-        elapsed: Duration,
-    ) {
-        let queue_size = self.writer.size().await;
-        let ack_floor = self.reader.ack_floor().await;
-        let abandoned_entries = queue_size.saturating_sub(ack_floor);
-        let tip = Self::latest_proof_bearing_tip(self.marshal.clone()).await;
-        let (tip_height, tip_digest) = tip.unwrap_or((0, first_digest));
-        let abandoned_height_span = tip_height.saturating_sub(first_height);
-
-        warn!(
-            ?reason,
-            first_height,
-            ?first_digest,
-            tip_height,
-            ?tip_digest,
-            attempts,
-            ?elapsed,
-            abandoned_entries,
-            abandoned_height_span,
-            "resetting stale durable backfill queue"
-        );
-
-        self.active.cancel_all();
-        match self.reader.ack_up_to(queue_size).await {
-            Ok(()) => self.metrics.queue_acked(QueueStatus::Success),
-            Err(err) => {
-                self.metrics.queue_acked(QueueStatus::Failure);
-                panic!("failed to ack stale finalized queue: {err:?}");
-            }
-        }
-        let sync_started = Instant::now();
-        match self.writer.sync().await {
-            Ok(()) => self
-                .metrics
-                .queue_synced(QueueStatus::Success, sync_started.elapsed()),
-            Err(err) => {
-                self.metrics
-                    .queue_synced(QueueStatus::Failure, sync_started.elapsed());
-                panic!("failed to sync stale finalized queue reset: {err:?}");
-            }
-        }
-        self.reader.reset().await;
-        self.metrics.queue_ack_floor(tip_height);
-        self.metrics.backfill_queue_reset(
-            reason,
-            abandoned_entries,
-            abandoned_height_span,
-        );
-        self.uploads.lock().advance_queue_floor(tip_height);
-        self.uploads.lock().restart_above(tip_height);
-    }
-
-    async fn latest_proof_bearing_tip(
-        marshal: MarshalMailbox<Scheme, Standard<Block>>,
-    ) -> Option<(u64, Digest)> {
-        let (height, _) = marshal.get_info(Identifier::Latest).await?;
-        for height in (0..=height.get()).rev() {
-            if let Some(proof) = marshal.get_finalization(Height::new(height)).await {
-                return Some((height, proof.proposal.payload));
-            }
-        }
-        None
-    }
-}
-
-#[derive(Default)]
-struct ProofFailure {
-    reason: Option<BackfillResetReason>,
-    first_seen: Option<std::time::SystemTime>,
-    attempts: u64,
-}
-
-impl ProofFailure {
-    fn observe<E: Clock>(
-        &mut self,
-        context: &E,
-        reason: BackfillResetReason,
-        grace: Duration,
-        height: u64,
-        digest: Digest,
-    ) -> Option<Completion> {
-        if self.reason != Some(reason) {
-            self.reason = Some(reason);
-            self.first_seen = Some(context.current());
-            self.attempts = 0;
-        }
-        self.attempts = self.attempts.saturating_add(1);
-        let elapsed = context
-            .current()
-            .duration_since(self.first_seen.expect("proof failure start"))
-            .unwrap_or_default();
-        (elapsed >= grace).then_some(Completion::Stale {
-            reason,
-            first_height: height,
-            first_digest: digest,
-            attempts: self.attempts,
-            elapsed,
-        })
-    }
 }
diff --git a/examples/coins/chain/src/indexer/backfiller/producer.rs b/examples/coins/chain/src/indexer/backfiller/producer.rs
index 88b01e4..fccdb5d 100644
--- a/examples/coins/chain/src/indexer/backfiller/producer.rs
+++ b/examples/coins/chain/src/indexer/backfiller/producer.rs
@@ -3,23 +3,32 @@ use crate::{
     indexer::{
         metrics::{
             estimated_block_bytes, BlockMetricSource, ProducerActivity, ProducerStatus,
-            QueueStatus,
         },
-        IndexerMetrics,
+        IndexerMetrics, SpoolLimits,
     },
-    Block,
+    Block, Finalized, Scheme,
 };
 use commonware_actor::{
     mailbox::{self, Overflow, Policy},
     Feedback,
 };
-use commonware_consensus::{marshal::Update, Reporter};
+use commonware_consensus::{
+    marshal::{core::Mailbox as MarshalMailbox, standard::Standard, Update},
+    types::Height,
+    Reporter,
+};
 use commonware_runtime::{
     spawn_cell, BufferPooler, Clock, ContextCell, Handle, Metrics, Spawner, Storage,
 };
-use commonware_storage::queue;
 use commonware_utils::{acknowledgement::Exact, Acknowledgement};
-use std::{collections::VecDeque, num::NonZeroUsize, sync::Arc, time::Instant};
+use commonware_utils::channel::oneshot;
+use std::{
+    collections::VecDeque,
+    num::NonZeroUsize,
+    sync::Arc,
+    time::{Duration, Instant, UNIX_EPOCH},
+};
+use tracing::warn;
 
 #[derive(Clone)]
 pub struct Producer {
@@ -27,6 +36,22 @@ pub struct Producer {
     metrics: IndexerMetrics,
 }
 
+pub(crate) struct Admission {
+    pub(crate) entry: Entry,
+    pub(crate) response: oneshot::Sender<()>,
+}
+
+impl Policy for Admission {
+    type Overflow = VecDeque<Self>;
+
+    fn handle(overflow: &mut Self::Overflow, message: Self) {
+        overflow.push_back(message);
+    }
+}
+
+pub(crate) type AdmissionSender = mailbox::Sender<Admission>;
+pub(crate) type AdmissionReceiver = mailbox::Receiver<Admission>;
+
 struct Message {
     block: Arc<Block>,
     block_estimated_bytes: u64,
@@ -87,8 +112,21 @@ struct Actor<E: BufferPooler + Clock + Storage + Metrics> {
     context: ContextCell<E>,
     uploads: SharedState,
     metrics: IndexerMetrics,
-    writer: queue::Writer<E, Entry>,
+    marshal: MarshalMailbox<Scheme, Standard<Block>>,
+    admission: AdmissionSender,
     receiver: mailbox::Receiver<Message>,
+    retry: Duration,
+    missing_finalization_grace: Duration,
+    mismatched_finalization_grace: Duration,
+    spool_limits: SpoolLimits,
+}
+
+pub(crate) struct Config {
+    pub(crate) mailbox_size: NonZeroUsize,
+    pub(crate) retry: Duration,
+    pub(crate) missing_finalization_grace: Duration,
+    pub(crate) mismatched_finalization_grace: Duration,
+    pub(crate) spool_limits: SpoolLimits,
 }
 
 impl Reporter for Producer {
@@ -123,16 +161,29 @@ impl<E: BufferPooler + Clock + Storage + Metrics + Spawner> Actor<E> {
         context: E,
         uploads: SharedState,
         metrics: IndexerMetrics,
-        writer: queue::Writer<E, Entry>,
-        mailbox_size: NonZeroUsize,
+        marshal: MarshalMailbox<Scheme, Standard<Block>>,
+        admission: AdmissionSender,
+        config: Config,
     ) -> (Self, Producer) {
+        let Config {
+            mailbox_size,
+            retry,
+            missing_finalization_grace,
+            mismatched_finalization_grace,
+            spool_limits,
+        } = config;
         let (sender, receiver) = mailbox::new(context.child("mailbox"), mailbox_size);
         let actor = Self {
             context: ContextCell::new(context),
             uploads,
             metrics: metrics.clone(),
-            writer,
+            marshal,
+            admission,
             receiver,
+            retry,
+            missing_finalization_grace,
+            mismatched_finalization_grace,
+            spool_limits,
         };
         (
             actor,
@@ -158,25 +209,91 @@ impl<E: BufferPooler + Clock + Storage + Metrics + Spawner> Actor<E> {
         let started = Instant::now();
         self.metrics
             .observe_block(BlockMetricSource::ProducerRecord, block);
-        let Some(entry) = self.uploads.lock().record(block) else {
+        let Some(candidate) = self.uploads.lock().record(block) else {
             self.metrics.producer_recorded(
                 ProducerStatus::AlreadyUploaded,
                 started.elapsed(),
             );
             return;
         };
-        self.metrics.queue_entry(entry.height);
-        match self.writer.enqueue(entry).await {
-            Ok(_) => {
-                self.metrics.queue_enqueued(QueueStatus::Success);
-                self.metrics
-                    .producer_recorded(ProducerStatus::Recorded, started.elapsed());
-            }
-            Err(err) => {
-                self.metrics.queue_enqueued(QueueStatus::Failure);
-                panic!("failed to enqueue finalized digest: {err:?}");
+        let first_seen = self.context.current();
+        self.metrics.certificate_request_started();
+        let proof = loop {
+            let Some(proof) = self
+                .marshal
+                .get_finalization(Height::new(candidate.height))
+                .await
+            else {
+                let elapsed = self
+                    .context
+                    .current()
+                    .duration_since(first_seen)
+                    .unwrap_or_default();
+                assert!(
+                    elapsed < self.missing_finalization_grace,
+                    "marshal has no finalization certificate for durable indexer payload at height {} after {:?}",
+                    candidate.height,
+                    elapsed,
+                );
+                warn!(height = candidate.height, ?elapsed, "waiting for finalized certificate before spooling indexer payload");
+                self.context.sleep(self.retry).await;
+                continue;
+            };
+            if proof.proposal.payload != candidate.digest
+                || proof.proposal.round.epoch() != block.context.round.epoch()
+            {
+                let elapsed = self
+                    .context
+                    .current()
+                    .duration_since(first_seen)
+                    .unwrap_or_default();
+                assert!(
+                    elapsed < self.mismatched_finalization_grace,
+                    "marshal finalization certificate conflicts with block at height {} after {:?}",
+                    candidate.height,
+                    elapsed,
+                );
+                warn!(height = candidate.height, ?elapsed, "waiting for matching finalized certificate before spooling indexer payload");
+                self.context.sleep(self.retry).await;
+                continue;
             }
+            break proof;
+        };
+        self.metrics.certificate_request_finished();
+        let enqueued_at_millis = self
+            .context
+            .current()
+            .duration_since(UNIX_EPOCH)
+            .unwrap_or_default()
+            .as_millis()
+            .try_into()
+            .unwrap_or(u64::MAX);
+        let entry = Entry::new(enqueued_at_millis, Finalized::new(proof, block.clone()));
+        let expiry = if entry.encoded_len > self.spool_limits.max_payload_bytes
+            || entry.encoded_len > self.spool_limits.max_bytes
+        {
+            Some(ProducerStatus::ExpiredOversized)
+        } else {
+            None
+        };
+        if let Some(status) = expiry {
+            self.metrics.producer_recorded(status, started.elapsed());
+            warn!(
+                height = entry.height(),
+                digest = ?entry.digest(),
+                encoded_len = entry.encoded_len,
+                ?status,
+                "terminally expiring finalized indexer payload to preserve spool availability bound"
+            );
+            return;
         }
+        let (response, completed) = oneshot::channel();
+        let _ = self.admission.enqueue(Admission { entry, response });
+        completed
+            .await
+            .expect("indexer spool admission coordinator stopped");
+        self.metrics
+            .producer_recorded(ProducerStatus::Recorded, started.elapsed());
     }
 }
 
@@ -184,15 +301,23 @@ pub fn init<E>(
     context: E,
     uploads: SharedState,
     metrics: IndexerMetrics,
-    writer: queue::Writer<E, Entry>,
-    mailbox_size: NonZeroUsize,
-) -> Producer
+    marshal: MarshalMailbox<Scheme, Standard<Block>>,
+    admission: AdmissionSender,
+    config: Config,
+) -> (Producer, Handle<()>)
 where
     E: BufferPooler + Clock + Storage + Metrics + Spawner,
 {
-    let (actor, producer) = Actor::new(context, uploads, metrics, writer, mailbox_size);
-    actor.start();
-    producer
+    let (actor, producer) = Actor::new(
+        context,
+        uploads,
+        metrics,
+        marshal,
+        admission,
+        config,
+    );
+    let handle = actor.start();
+    (producer, handle)
 }
 
 #[cfg(test)]
@@ -200,16 +325,12 @@ mod tests {
     use super::*;
     use crate::{StateCommitment, Transaction, EPOCH};
     use commonware_consensus::types::{Height, Round, View};
-    use commonware_cryptography::{ed25519, Digestible, Hasher, Sha256, Signer};
-    use commonware_runtime::{
-        buffer::paged::CacheRef, deterministic, Runner as _, Supervisor as _,
-    };
+    use commonware_cryptography::{ed25519, Hasher, Sha256, Signer};
+    use commonware_runtime::{deterministic, Runner as _, Supervisor as _};
     use commonware_storage::mmr::Location;
     use commonware_utils::{
-        acknowledgement::Exact, range::NonEmptyRange, sync::Mutex, NZUsize, NZU16, NZU64,
+        acknowledgement::Exact, range::NonEmptyRange, NZUsize,
     };
-    use futures::FutureExt;
-    use std::{sync::Arc, time::Duration};
 
     fn state(height: u64) -> StateCommitment {
         StateCommitment {
@@ -239,60 +360,6 @@ mod tests {
         )
     }
 
-    #[test]
-    fn queues_finalized_block_before_acknowledging() {
-        deterministic::Runner::default().start(|context| async move {
-            let page_cache = CacheRef::from_pooler(&context, NZU16!(4_096), NZUsize!(128));
-            let (writer, mut reader) = queue::shared::init(
-                context.child("queue"),
-                queue::Config {
-                    partition: "indexer-producer-test".to_string(),
-                    items_per_section: NZU64!(16),
-                    compression: None,
-                    codec_config: (),
-                    page_cache,
-                    write_buffer: NZUsize!(1024),
-                },
-            )
-            .await
-            .expect("init queue");
-            let uploads = Arc::new(Mutex::new(crate::indexer::backfiller::State::new()));
-            let metrics = IndexerMetrics::register(&context.child("indexer"));
-            let mut producer = init(context.child("producer"), uploads, metrics, writer, NZUsize!(4));
-            let block = block(2, 2, b"block");
-            let digest = block.digest();
-            let (ack, waiter) = Exact::handle();
-
-            assert!(producer
-                .report(commonware_consensus::marshal::Update::Block(
-                    block.into(),
-                    ack,
-                ))
-                .accepted());
-            commonware_macros::select! {
-                result = waiter.fuse() => result.expect("acknowledged"),
-                _ = context.sleep(Duration::from_secs(1)) => panic!("ack timed out"),
-            }
-
-            let (_, entry) = reader
-                .recv()
-                .await
-                .expect("read queue")
-                .expect("queued entry");
-            assert_eq!(entry.height, 2);
-            assert_eq!(entry.digest, digest);
-
-            let encoded = context.encode();
-            assert!(encoded.contains(
-                "indexer_producer_report_total{activity=\"block\",status=\"enqueued\"} 1",
-            ));
-            assert!(encoded.contains(
-                "indexer_producer_report_total{activity=\"block\",status=\"recorded\"} 1",
-            ));
-            assert!(encoded.contains("indexer_producer_record_duration_seconds_bucket"));
-        });
-    }
-
     #[test]
     fn mailbox_overflow_tracks_retained_block_pressure() {
         deterministic::Runner::default().start(|context| async move {
diff --git a/examples/coins/chain/src/indexer/backfiller/state.rs b/examples/coins/chain/src/indexer/backfiller/state.rs
index 217b70f..e52c24c 100644
--- a/examples/coins/chain/src/indexer/backfiller/state.rs
+++ b/examples/coins/chain/src/indexer/backfiller/state.rs
@@ -5,10 +5,10 @@ use crate::{
         },
         IndexerMetrics,
     },
-    Block,
+    Block, Finalized,
 };
 use bytes::{Buf, BufMut};
-use commonware_codec::{self, FixedSize, Read, Write};
+use commonware_codec::{self, EncodeSize, FixedSize, Read, Write};
 use commonware_cryptography::{sha256::Digest, Digestible};
 use commonware_utils::{sync::Mutex, PrioritySet};
 use std::{collections::BTreeMap, sync::Arc};
@@ -19,30 +19,84 @@ pub enum Decision {
     Proceed,
 }
 
-#[derive(Clone, Copy, PartialEq, Eq)]
-pub struct Entry {
+pub struct Candidate {
     pub height: u64,
     pub digest: Digest,
 }
 
-impl FixedSize for Entry {
-    const SIZE: usize = u64::SIZE + Digest::SIZE;
+const ENTRY_VERSION: u8 = 1;
+
+#[derive(Clone)]
+pub struct Entry {
+    pub version: u8,
+    pub enqueued_at_millis: u64,
+    pub encoded_len: u64,
+    pub finalized: Finalized,
+}
+
+impl Entry {
+    pub fn new(enqueued_at_millis: u64, finalized: Finalized) -> Self {
+        let mut entry = Self {
+            version: ENTRY_VERSION,
+            enqueued_at_millis,
+            encoded_len: 0,
+            finalized,
+        };
+        entry.encoded_len = entry.encode_size() as u64;
+        entry
+    }
+
+    pub fn height(&self) -> u64 {
+        self.finalized.block.height.get()
+    }
+
+    pub fn digest(&self) -> Digest {
+        self.finalized.block.digest()
+    }
 }
 
 impl Write for Entry {
     fn write(&self, buf: &mut impl BufMut) {
-        self.height.write(buf);
-        self.digest.write(buf);
+        self.version.write(buf);
+        self.enqueued_at_millis.write(buf);
+        self.encoded_len.write(buf);
+        self.finalized.write(buf);
     }
 }
 
 impl Read for Entry {
-    type Cfg = ();
+    type Cfg = <Finalized as Read>::Cfg;
+
+    fn read_cfg(buf: &mut impl Buf, cfg: &Self::Cfg) -> Result<Self, commonware_codec::Error> {
+        let version = u8::read_cfg(buf, &())?;
+        if version != ENTRY_VERSION {
+            return Err(commonware_codec::Error::Invalid(
+                "indexer spool entry",
+                "unsupported version",
+            ));
+        }
+        let enqueued_at_millis = u64::read_cfg(buf, &())?;
+        let encoded_len = u64::read_cfg(buf, &())?;
+        let finalized = Finalized::read_cfg(buf, cfg)?;
+        let entry = Self {
+            version,
+            enqueued_at_millis,
+            encoded_len,
+            finalized,
+        };
+        if entry.encode_size() as u64 != encoded_len {
+            return Err(commonware_codec::Error::Invalid(
+                "indexer spool entry",
+                "encoded length mismatch",
+            ));
+        }
+        Ok(entry)
+    }
+}
 
-    fn read_cfg(buf: &mut impl Buf, _: &()) -> Result<Self, commonware_codec::Error> {
-        let height = u64::read_cfg(buf, &())?;
-        let digest = Digest::read_cfg(buf, &())?;
-        Ok(Self { height, digest })
+impl EncodeSize for Entry {
+    fn encode_size(&self) -> usize {
+        u8::SIZE + u64::SIZE + u64::SIZE + self.finalized.encode_size()
     }
 }
 
@@ -55,6 +109,8 @@ pub struct State {
     cached_block_estimated_bytes: u64,
     certificate_uploads: BTreeMap<Digest, usize>,
     certificate_upload_refs: usize,
+    queued: BTreeMap<Digest, (u64, u64, u64, u64)>,
+    queued_bytes: u64,
     metrics: Option<IndexerMetrics>,
 }
 
@@ -74,6 +130,8 @@ impl State {
             cached_block_estimated_bytes: 0,
             certificate_uploads: BTreeMap::new(),
             certificate_upload_refs: 0,
+            queued: BTreeMap::new(),
+            queued_bytes: 0,
             metrics: None,
         }
     }
@@ -89,8 +147,8 @@ impl State {
         self.uploaded.contains(digest)
     }
 
-    pub fn record(&mut self, block: &Block) -> Option<Entry> {
-        let entry = Entry {
+    pub fn record(&mut self, block: &Block) -> Option<Candidate> {
+        let entry = Candidate {
             height: block.height.get(),
             digest: block.digest(),
         };
@@ -99,7 +157,7 @@ impl State {
             self.sync_metrics();
             return None;
         }
-        if self.is_uploaded(&entry.digest) {
+        if self.is_uploaded(&entry.digest) || self.queued(&entry.digest) {
             self.sync_metrics();
             return None;
         }
@@ -108,12 +166,74 @@ impl State {
         Some(entry)
     }
 
+    pub fn queued(&self, digest: &Digest) -> bool {
+        self.queued.contains_key(digest)
+    }
+
+    pub fn mark_queued(&mut self, position: u64, entry: &Entry) {
+        if let Some((_, _, old_bytes, _)) =
+            self.queued
+                .insert(
+                    entry.digest(),
+                    (
+                        position,
+                        entry.height(),
+                        entry.encoded_len,
+                        entry.enqueued_at_millis,
+                    ),
+                )
+        {
+            self.queued_bytes = self.queued_bytes.saturating_sub(old_bytes);
+        }
+        self.queued_bytes = self.queued_bytes.saturating_add(entry.encoded_len);
+        self.sync_metrics();
+    }
+
+    pub fn recover_queued(&mut self, position: u64, entry: &Entry) {
+        self.observe_finalization(entry.height());
+        self.mark_queued(position, entry);
+    }
+
+    pub fn finish_queued(&mut self, digest: &Digest) {
+        if let Some((_, _, bytes, _)) = self.queued.remove(digest) {
+            self.queued_bytes = self.queued_bytes.saturating_sub(bytes);
+        }
+        self.sync_metrics();
+    }
+
+    pub fn spool_usage(&self) -> (u64, u64) {
+        (self.queued.len() as u64, self.queued_bytes)
+    }
+
+    pub fn oldest_queued(&self) -> Option<(Digest, u64, u64, u64, u64)> {
+        self.queued
+            .iter()
+            .min_by_key(|(_, (position, _, _, _))| *position)
+            .map(|(digest, (position, height, bytes, enqueued_at))| {
+                (*digest, *position, *height, *bytes, *enqueued_at)
+            })
+    }
+
+    pub fn expire_through(&mut self, position: u64) -> Vec<(Digest, u64, u64)> {
+        let expired: Vec<_> = self
+            .queued
+            .iter()
+            .filter(|(_, (queued_position, _, _, _))| *queued_position <= position)
+            .map(|(digest, (_, height, bytes, _))| (*digest, *height, *bytes))
+            .collect();
+        for (digest, _, _) in &expired {
+            self.finish_queued(digest);
+        }
+        expired
+    }
+
     pub fn advance_queue_floor(&mut self, height: u64) {
         self.acked_through = self.acked_through.max(height);
         self.prune();
         self.sync_metrics();
     }
 
+    #[cfg(test)]
     pub fn restart_above(&mut self, height: u64) {
         self.restart_watermark = self.restart_watermark.max(height);
         let pruned = self
@@ -145,6 +265,7 @@ impl State {
         self.sync_metrics();
     }
 
+    #[cfg(test)]
     pub fn cached_block(&self, digest: &Digest) -> Option<Block> {
         self.cached_blocks
             .get(digest)
@@ -279,6 +400,20 @@ impl State {
                 uploaded_digests: self.uploaded.len(),
                 latest_finalized_height: self.latest_finalized,
                 acked_through_height: self.acked_through,
+                spool_entries: self.queued.len(),
+                spool_logical_bytes: self.queued_bytes,
+                spool_oldest_height: self
+                    .queued
+                    .values()
+                    .map(|(_, height, _, _)| *height)
+                    .min()
+                    .unwrap_or(0),
+                spool_oldest_enqueued_at_millis: self
+                    .queued
+                    .values()
+                    .map(|(_, _, _, enqueued_at)| *enqueued_at)
+                    .min()
+                    .unwrap_or(0),
             });
         }
     }
@@ -290,7 +425,6 @@ pub type SharedState = Arc<Mutex<State>>;
 mod tests {
     use super::*;
     use crate::{StateCommitment, Transaction, EPOCH};
-    use commonware_codec::{DecodeExt, Encode};
     use commonware_consensus::types::{Height, Round, View};
     use commonware_cryptography::{ed25519, Hasher, Sha256, Signer};
     use commonware_runtime::{deterministic, Metrics as _, Runner as _, Supervisor as _};
@@ -325,20 +459,6 @@ mod tests {
         )
     }
 
-    #[test]
-    fn entry_codec_roundtrips() {
-        let entry = Entry {
-            height: 7,
-            digest: Sha256::hash(b"block"),
-        };
-
-        let encoded = entry.encode();
-        let decoded = Entry::decode(encoded).expect("decode entry");
-
-        assert_eq!(decoded.height, entry.height);
-        assert_eq!(decoded.digest, entry.digest);
-    }
-
     #[test]
     fn record_caches_block_until_uploaded() {
         let mut state = State::new();
diff --git a/examples/coins/chain/src/indexer/metrics.rs b/examples/coins/chain/src/indexer/metrics.rs
index 18d2f08..ca211ea 100644
--- a/examples/coins/chain/src/indexer/metrics.rs
+++ b/examples/coins/chain/src/indexer/metrics.rs
@@ -163,7 +163,6 @@ pub(crate) enum BlockMetricSource {
     ProducerRecord,
     LiveCertificate,
     ConsumerCached,
-    ConsumerMarshal,
 }
 
 impl BlockMetricSource {
@@ -172,7 +171,6 @@ impl BlockMetricSource {
             Self::ProducerRecord => "producer_record",
             Self::LiveCertificate => "live_certificate",
             Self::ConsumerCached => "consumer_cached",
-            Self::ConsumerMarshal => "consumer_marshal",
         }
     }
 }
@@ -192,7 +190,6 @@ impl EncodeLabelValueTrait for BlockMetricSource {
 pub(crate) enum SharedCacheSource {
     ProducerRecord,
     LiveCertificate,
-    ConsumerMarshal,
 }
 
 impl SharedCacheSource {
@@ -200,7 +197,6 @@ impl SharedCacheSource {
         match self {
             Self::ProducerRecord => "producer_record",
             Self::LiveCertificate => "live_certificate",
-            Self::ConsumerMarshal => "consumer_marshal",
         }
     }
 }
@@ -248,7 +244,6 @@ impl EncodeLabelValueTrait for SharedRetentionReason {
 pub(crate) enum BackfillStatus {
     Uploaded,
     Skipped,
-    Abandoned,
 }
 
 impl BackfillStatus {
@@ -256,7 +251,6 @@ impl BackfillStatus {
         match self {
             Self::Uploaded => "uploaded",
             Self::Skipped => "skipped",
-            Self::Abandoned => "abandoned",
         }
     }
 }
@@ -303,7 +297,6 @@ impl EncodeLabelValueTrait for BackfillDecision {
 #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
 pub(crate) enum BackfillPhase {
     Start,
-    BeforeBlock,
     BeforeAttempt,
 }
 
@@ -311,7 +304,6 @@ impl BackfillPhase {
     const fn as_str(self) -> &'static str {
         match self {
             Self::Start => "start",
-            Self::BeforeBlock => "before_block",
             Self::BeforeAttempt => "before_attempt",
         }
     }
@@ -331,9 +323,6 @@ impl EncodeLabelValueTrait for BackfillPhase {
 #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
 pub(crate) enum BackfillWaitReason {
     CertificateUpload,
-    MissingBlock,
-    MissingFinalization,
-    MismatchedFinalization,
     HttpError,
 }
 
@@ -341,40 +330,11 @@ impl BackfillWaitReason {
     const fn as_str(self) -> &'static str {
         match self {
             Self::CertificateUpload => "certificate_upload",
-            Self::MissingBlock => "missing_block",
-            Self::MissingFinalization => "missing_finalization",
-            Self::MismatchedFinalization => "mismatched_finalization",
             Self::HttpError => "http_error",
         }
     }
 }
 
-#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
-pub(crate) enum BackfillResetReason {
-    MissingFinalization,
-    MismatchedFinalization,
-}
-
-impl BackfillResetReason {
-    const fn as_str(self) -> &'static str {
-        match self {
-            Self::MissingFinalization => "missing_finalization",
-            Self::MismatchedFinalization => "mismatched_finalization",
-        }
-    }
-}
-
-impl EncodeLabelValueTrait for BackfillResetReason {
-    fn encode(
-        &self,
-        encoder: &mut commonware_runtime::telemetry::metrics::LabelValueEncoder,
-    ) -> Result<(), fmt::Error> {
-        use fmt::Write as _;
-
-        encoder.write_str(self.as_str())
-    }
-}
-
 impl EncodeLabelValueTrait for BackfillWaitReason {
     fn encode(
         &self,
@@ -475,6 +435,10 @@ pub(crate) enum ProducerStatus {
     Ignored,
     Recorded,
     AlreadyUploaded,
+    ExpiredOversized,
+    ExpiredEntries,
+    ExpiredBytes,
+    ExpiredAge,
 }
 
 impl ProducerStatus {
@@ -485,6 +449,10 @@ impl ProducerStatus {
             Self::Ignored => "ignored",
             Self::Recorded => "recorded",
             Self::AlreadyUploaded => "already_uploaded",
+            Self::ExpiredOversized => "expired_oversized",
+            Self::ExpiredEntries => "expired_entries",
+            Self::ExpiredBytes => "expired_bytes",
+            Self::ExpiredAge => "expired_age",
         }
     }
 }
@@ -538,6 +506,10 @@ pub(crate) struct SharedStateSnapshot {
     pub(crate) uploaded_digests: usize,
     pub(crate) latest_finalized_height: u64,
     pub(crate) acked_through_height: u64,
+    pub(crate) spool_entries: usize,
+    pub(crate) spool_logical_bytes: u64,
+    pub(crate) spool_oldest_height: u64,
+    pub(crate) spool_oldest_enqueued_at_millis: u64,
 }
 
 #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq, EncodeLabelSet)]
@@ -599,11 +571,6 @@ struct BackfillWaitReasonLabel {
     reason: BackfillWaitReason,
 }
 
-#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq, EncodeLabelSet)]
-struct BackfillResetReasonLabel {
-    reason: BackfillResetReason,
-}
-
 #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq, EncodeLabelSet)]
 struct QueueStatusLabel {
     status: QueueStatus,
@@ -666,6 +633,11 @@ pub(crate) struct Inner {
     shared_uploaded_digests: Gauge,
     shared_latest_finalized_height: Gauge,
     shared_acked_through_height: Gauge,
+    spool_entries: Gauge,
+    spool_logical_bytes: Gauge,
+    spool_oldest_height: Gauge,
+    spool_oldest_enqueued_at_millis: Gauge,
+    producer_certificate_requests: Gauge,
     shared_pruned: CounterFamily<SharedRetentionReasonLabel>,
     shared_cache_inserted: CounterFamily<SharedCacheSourceLabel>,
     shared_cache_removed: CounterFamily<SharedRetentionReasonLabel>,
@@ -677,9 +649,6 @@ pub(crate) struct Inner {
     backfill_decision: CounterFamily<BackfillDecisionLabel>,
     backfill_wait_duration: HistogramFamily<BackfillWaitReasonLabel>,
     backfill_retry: CounterFamily<BackfillWaitReasonLabel>,
-    backfill_queue_reset: CounterFamily<BackfillResetReasonLabel>,
-    backfill_queue_reset_abandoned_entries: HistogramFamily<BackfillResetReasonLabel>,
-    backfill_queue_reset_abandoned_height_span: HistogramFamily<BackfillResetReasonLabel>,
     backfill_active_block_estimated_bytes: Gauge,
     backfill_active_body_estimated_bytes: Gauge,
     queue_enqueue: CounterFamily<QueueStatusLabel>,
@@ -817,6 +786,26 @@ impl IndexerMetrics {
                 "shared_acked_through_height",
                 "Queue acknowledgement floor tracked by shared indexer state",
             ),
+            spool_entries: context.gauge(
+                "spool_entries",
+                "Current number of self-contained finalized payloads in the durable spool",
+            ),
+            spool_logical_bytes: context.gauge(
+                "spool_logical_bytes",
+                "Current encoded bytes in self-contained finalized spool payloads",
+            ),
+            spool_oldest_height: context.gauge(
+                "spool_oldest_height",
+                "Oldest finalized height retained in the durable spool",
+            ),
+            spool_oldest_enqueued_at_millis: context.gauge(
+                "spool_oldest_enqueued_at_millis",
+                "Wall-clock Unix millisecond timestamp of the oldest durable spool payload",
+            ),
+            producer_certificate_requests: context.gauge(
+                "spool_certificate_requests",
+                "Current finalized blocks awaiting a marshal certificate before spool admission",
+            ),
             shared_pruned: context.family(
                 "shared_prune",
                 "Total shared indexer state retention removals by reason",
@@ -867,24 +856,6 @@ impl IndexerMetrics {
                 "backfill_retry",
                 "Total durable backfill retries by reason",
             ),
-            backfill_queue_reset: context.family(
-                "backfill_queue_reset",
-                "Total durable backfill queue resets by reason",
-            ),
-            backfill_queue_reset_abandoned_entries: context.register(
-                "backfill_queue_reset_abandoned_entries",
-                "Durable backfill entries abandoned by queue reset",
-                raw::Family::<BackfillResetReasonLabel, raw::Histogram, fn() -> raw::Histogram>::new_with_constructor(
-                    local_histogram,
-                ),
-            ),
-            backfill_queue_reset_abandoned_height_span: context.register(
-                "backfill_queue_reset_abandoned_height_span",
-                "Durable backfill height span abandoned by queue reset",
-                raw::Family::<BackfillResetReasonLabel, raw::Histogram, fn() -> raw::Histogram>::new_with_constructor(
-                    local_histogram,
-                ),
-            ),
             backfill_active_block_estimated_bytes: context.gauge(
                 "backfill_active_block_estimated_bytes",
                 "Current estimated encoded block bytes held by durable backfill upload tasks",
@@ -1074,6 +1045,16 @@ impl IndexerMetrics {
         let _ = self
             .shared_acked_through_height
             .try_set(snapshot.acked_through_height);
+        let _ = self.spool_entries.try_set(snapshot.spool_entries);
+        let _ = self
+            .spool_logical_bytes
+            .try_set(snapshot.spool_logical_bytes);
+        let _ = self
+            .spool_oldest_height
+            .try_set(snapshot.spool_oldest_height);
+        let _ = self
+            .spool_oldest_enqueued_at_millis
+            .try_set(snapshot.spool_oldest_enqueued_at_millis);
         self.queue_ack_floor(snapshot.acked_through_height);
         let lag = snapshot
             .latest_finalized_height
@@ -1081,6 +1062,14 @@ impl IndexerMetrics {
         let _ = self.queue_lag_height.try_set(lag);
     }
 
+    pub(crate) fn certificate_request_started(&self) {
+        self.producer_certificate_requests.inc();
+    }
+
+    pub(crate) fn certificate_request_finished(&self) {
+        self.producer_certificate_requests.dec();
+    }
+
     pub(crate) fn shared_cache_inserted(&self, source: SharedCacheSource) {
         self.shared_cache_inserted
             .get_or_create(&SharedCacheSourceLabel { source })
@@ -1132,22 +1121,6 @@ impl IndexerMetrics {
             .observe(duration.as_secs_f64());
     }
 
-    pub(crate) fn backfill_queue_reset(
-        &self,
-        reason: BackfillResetReason,
-        abandoned_entries: u64,
-        abandoned_height_span: u64,
-    ) {
-        let label = BackfillResetReasonLabel { reason };
-        self.backfill_queue_reset.get_or_create(&label).inc();
-        self.backfill_queue_reset_abandoned_entries
-            .get_or_create(&label)
-            .observe(abandoned_entries as f64);
-        self.backfill_queue_reset_abandoned_height_span
-            .get_or_create(&label)
-            .observe(abandoned_height_span as f64);
-    }
-
     pub(crate) fn start_backfill_body(&self, bytes: u64) -> BackfillBodyGuard {
         let bytes = gauge_u64(bytes);
         self.backfill_active_body_estimated_bytes.inc_by(bytes);
@@ -1403,9 +1376,6 @@ impl BackfillUploadGuard {
         self.status = Some(BackfillStatus::Skipped);
     }
 
-    pub(crate) const fn abandoned(&mut self) {
-        self.status = Some(BackfillStatus::Abandoned);
-    }
 }
 
 impl Drop for BackfillUploadGuard {
diff --git a/examples/coins/chain/src/indexer/mod.rs b/examples/coins/chain/src/indexer/mod.rs
index 2604221..fac82a2 100644
--- a/examples/coins/chain/src/indexer/mod.rs
+++ b/examples/coins/chain/src/indexer/mod.rs
@@ -6,9 +6,9 @@ use commonware_consensus::types::Epoch;
 use commonware_cryptography::bls12381::{
     dkg::feldman_desmedt::Output as DkgOutput, primitives::variant::MinSig,
 };
-use commonware_runtime::{BufferPooler, Clock, Metrics, Spawner, Storage};
+use commonware_runtime::{BufferPooler, Clock, Handle, Metrics, Spawner, Storage};
 use commonware_storage::queue;
-use commonware_utils::sync::Mutex;
+use commonware_utils::{sync::Mutex, NZU64};
 use std::{future::Future, num::NonZeroUsize, sync::Arc, time::Duration};
 use thiserror::Error;
 
@@ -21,7 +21,7 @@ use backfiller::{SharedState, State};
 pub(crate) use metrics::{DkgUploadStatus, IndexerMetrics};
 #[cfg(test)]
 pub(crate) use metrics::{
-    BackfillDecision, BackfillPhase, BackfillResetReason, BackfillWaitReason, BlockMetricSource,
+    BackfillDecision, BackfillPhase, BackfillWaitReason, BlockMetricSource,
     HttpArtifact, LiveUploadArtifact, ProducerActivity, ProducerStatus, QueueReadSource,
     QueueStatus, SharedCacheSource, SharedRetentionReason, SharedStateSnapshot,
 };
@@ -32,6 +32,64 @@ const CONNECT_TIMEOUT: Duration = Duration::from_secs(5);
 const REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
 const MISSING_FINALIZATION_GRACE: Duration = Duration::from_secs(120);
 const MISMATCHED_FINALIZATION_GRACE: Duration = Duration::from_secs(15);
+pub(crate) const SPOOL_ITEMS_PER_SECTION: std::num::NonZeroU64 = NZU64!(128);
+
+/// Availability window for durable finalized uploads while the external indexer is unavailable.
+#[derive(Clone, Copy, Debug)]
+pub struct SpoolLimits {
+    pub max_entries: u64,
+    pub max_bytes: u64,
+    pub max_payload_bytes: u64,
+    pub max_age: Duration,
+}
+
+impl Default for SpoolLimits {
+    fn default() -> Self {
+        Self {
+            max_entries: crate::BLOCKS_PER_EPOCH.get(),
+            max_bytes: 2 * 1024 * 1024 * 1024,
+            max_payload_bytes: 16 * 1024 * 1024,
+            max_age: Duration::from_secs(24 * 60 * 60),
+        }
+    }
+}
+
+impl SpoolLimits {
+    fn validate(self) {
+        assert!(self.max_entries > 0, "indexer spool max_entries must be non-zero");
+        assert!(self.max_bytes > 0, "indexer spool max_bytes must be non-zero");
+        assert!(
+            self.max_payload_bytes > 0 && self.max_payload_bytes <= self.max_bytes,
+            "indexer spool max_payload_bytes must be in 1..=max_bytes"
+        );
+        assert!(!self.max_age.is_zero(), "indexer spool max_age must be non-zero");
+        self.max_encoded_payload_bytes()
+            .expect("indexer spool encoded payload bound overflows u64");
+    }
+
+    pub fn max_encoded_payload_bytes(self) -> Option<u64> {
+        (SPOOL_ITEMS_PER_SECTION.get() - 1)
+            .checked_mul(self.max_payload_bytes)?
+            .checked_add(self.max_bytes)
+    }
+}
+
+#[cfg(test)]
+mod spool_tests {
+    use super::*;
+
+    #[test]
+    fn default_spool_bound_includes_section_slack() {
+        let limits = SpoolLimits::default();
+        assert_eq!(limits.max_entries, crate::BLOCKS_PER_EPOCH.get());
+        assert_eq!(limits.max_bytes, 2 * 1024 * 1024 * 1024);
+        assert_eq!(limits.max_age, Duration::from_secs(24 * 60 * 60));
+        assert_eq!(
+            limits.max_encoded_payload_bytes(),
+            Some(limits.max_bytes + 127 * limits.max_payload_bytes)
+        );
+    }
+}
 
 /// Errors returned by the HTTP indexer client.
 #[derive(Debug, Error)]
@@ -184,11 +242,13 @@ pub(crate) struct Config {
     pub(crate) mailbox_size: NonZeroUsize,
     pub(crate) backfiller_max_active: NonZeroUsize,
     pub(crate) backfiller_retry: Duration,
+    pub(crate) spool_limits: SpoolLimits,
     pub(crate) metrics: IndexerMetrics,
 }
 
 pub(crate) struct Indexer<E: BufferPooler + Spawner + Clock + Storage + Metrics, C: Client> {
     producer: Producer,
+    producer_handle: Handle<()>,
     pusher: Pusher<E, C>,
     consumer: Consumer<E, C>,
 }
@@ -205,8 +265,10 @@ impl<E: BufferPooler + Spawner + Clock + Storage + Metrics, C: Client> Indexer<E
             mailbox_size,
             backfiller_max_active,
             backfiller_retry,
+            spool_limits,
             metrics,
         } = config;
+        spool_limits.validate();
         let uploads: SharedState = Arc::new(Mutex::new(State::with_metrics(metrics.clone())));
         let pusher = Pusher::new(
             context.child("pusher"),
@@ -215,42 +277,65 @@ impl<E: BufferPooler + Spawner + Clock + Storage + Metrics, C: Client> Indexer<E
             uploads.clone(),
             metrics.clone(),
         );
-        let (writer, reader) = backfiller;
-        let producer = backfiller::producer::init(
+        let (writer, mut reader) = backfiller;
+        loop {
+            let Some((position, entry)) = reader
+                .try_recv()
+                .await
+                .expect("failed to reconstruct durable indexer spool")
+            else {
+                break;
+            };
+            uploads.lock().recover_queued(position, &entry);
+        }
+        reader.reset().await;
+        let (admission_sender, admission_receiver) = commonware_actor::mailbox::new(
+            context.child("admission_mailbox"),
+            mailbox_size,
+        );
+        let (producer, producer_handle) = backfiller::producer::init(
             context.child("producer"),
             uploads.clone(),
             metrics.clone(),
-            writer.clone(),
-            mailbox_size,
+            marshal.clone(),
+            admission_sender,
+            backfiller::producer::Config {
+                mailbox_size,
+                retry: backfiller_retry,
+                missing_finalization_grace: MISSING_FINALIZATION_GRACE,
+                mismatched_finalization_grace: MISMATCHED_FINALIZATION_GRACE,
+                spool_limits,
+            },
         );
         let consumer = Consumer::new(
             context.child("consumer"),
             client,
-            marshal,
             metrics,
             uploads,
             (writer, reader),
+            admission_receiver,
             backfiller::consumer::Config {
                 max_active: backfiller_max_active,
                 retry: backfiller_retry,
-                missing_finalization_grace: MISSING_FINALIZATION_GRACE,
-                mismatched_finalization_grace: MISMATCHED_FINALIZATION_GRACE,
+                spool_limits,
             },
         );
 
         Self {
             producer,
+            producer_handle,
             pusher,
             consumer,
         }
     }
 
-    pub(crate) fn split(self) -> (Producer, Pusher<E, C>, Consumer<E, C>) {
+    pub(crate) fn split(self) -> (Producer, Handle<()>, Pusher<E, C>, Consumer<E, C>) {
         let Self {
             producer,
+            producer_handle,
             pusher,
             consumer,
         } = self;
-        (producer, pusher, consumer)
+        (producer, producer_handle, pusher, consumer)
     }
 }
diff --git a/examples/coins/chain/src/lib.rs b/examples/coins/chain/src/lib.rs
index b034460..359f7f8 100644
--- a/examples/coins/chain/src/lib.rs
+++ b/examples/coins/chain/src/lib.rs
@@ -19,6 +19,7 @@ pub mod application;
 pub mod engine;
 pub mod execution;
 pub mod genesis;
+pub(crate) mod history;
 pub mod indexer;
 pub mod rpc;
 pub mod runtime;
diff --git a/examples/coins/chain/src/testnet.rs b/examples/coins/chain/src/testnet.rs
index a1fc1ea..feaf62a 100644
--- a/examples/coins/chain/src/testnet.rs
+++ b/examples/coins/chain/src/testnet.rs
@@ -10,7 +10,7 @@ use crate::{
     channels,
     engine::{Config as EngineConfig, Engine},
     genesis::{ChainGenesis, GenesisError},
-    indexer, rpc, PublicKey, NAMESPACE,
+    indexer, rpc, PublicKey, BLOCKS_PER_EPOCH, NAMESPACE,
 };
 use commonware_codec::{Decode, DecodeExt, Encode, EncodeSize};
 use commonware_consensus::marshal;
@@ -29,7 +29,7 @@ use commonware_p2p::{
 use commonware_runtime::{
     tokio, Clock as _, Handle, Runner as _, Spawner as _, Strategizer as _, Supervisor as _,
 };
-use commonware_utils::{ordered::Set, N3f1, NZUsize, NZU32};
+use commonware_utils::{ordered::Set, Hostname, N3f1, NZUsize, NZU32};
 use governor::Quota;
 use nunchi_dkg::{
     ContinueOnUpdate, PeerConfig, Storage as DkgStorage, StorageKey, StorageProtector,
@@ -48,7 +48,6 @@ use std::{
 };
 use tracing::{info, warn, Level};
 
-const FREEZER_TABLE_INITIAL_SIZE: u32 = 2u32.pow(14);
 const DEFAULT_MAX_BLOCK_TRANSACTIONS: usize = 4_096;
 const DEFAULT_MAX_MESSAGE_SIZE: u32 = 1024 * 1024;
 const DEFAULT_CHANNEL_BACKLOG: usize = 1024;
@@ -124,7 +123,12 @@ pub struct NodeConfig {
     pub share: String,
     pub peer_config: PeerConfig<PublicKey>,
     pub listen_address: SocketAddr,
-    pub dialable_address: SocketAddr,
+    /// Address advertised to peers. IP literals use socket syntax (with brackets around IPv6),
+    /// while DNS names use `hostname:port` without a URL scheme or path. DNS is resolved again
+    /// whenever a disconnected peer retries this node, so operators should use a stable hostname
+    /// and an appropriate TTL.
+    #[serde(with = "ingress_serde")]
+    pub dialable_address: Ingress,
     pub rpc_address: SocketAddr,
     pub metrics_address: SocketAddr,
     pub bootstrappers: Vec<BootstrapperConfig>,
@@ -132,6 +136,18 @@ pub struct NodeConfig {
     pub genesis_path: Option<PathBuf>,
     #[serde(default)]
     pub indexer_url: Option<String>,
+    /// Maximum self-contained finalized payloads retained while the indexer is unavailable.
+    #[serde(default = "default_indexer_spool_max_entries")]
+    pub indexer_spool_max_entries: u64,
+    /// Maximum logical encoded payload bytes retained by the indexer spool.
+    #[serde(default = "default_indexer_spool_max_bytes")]
+    pub indexer_spool_max_bytes: u64,
+    /// Maximum encoded size accepted for one finalized payload.
+    #[serde(default = "default_indexer_spool_max_payload_bytes")]
+    pub indexer_spool_max_payload_bytes: u64,
+    /// Maximum payload age before visible terminal expiry.
+    #[serde(default = "default_indexer_spool_max_age_seconds")]
+    pub indexer_spool_max_age_seconds: u64,
     pub consensus: ConsensusConfig,
     pub networking: NetworkConfig,
     /// Enable one-time peer QMDB state sync for a fresh joining node.
@@ -152,10 +168,83 @@ impl NodeConfig {
     }
 }
 
+fn default_indexer_spool_max_entries() -> u64 {
+    indexer::SpoolLimits::default().max_entries
+}
+
+fn default_indexer_spool_max_bytes() -> u64 {
+    indexer::SpoolLimits::default().max_bytes
+}
+
+fn default_indexer_spool_max_payload_bytes() -> u64 {
+    indexer::SpoolLimits::default().max_payload_bytes
+}
+
+fn default_indexer_spool_max_age_seconds() -> u64 {
+    indexer::SpoolLimits::default().max_age.as_secs()
+}
+
 #[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct BootstrapperConfig {
     pub public_key: String,
-    pub address: SocketAddr,
+    /// Address used to dial the bootstrapper. IP literals use socket syntax (with brackets around
+    /// IPv6), while DNS names use `hostname:port` without a URL scheme or path. DNS is resolved on
+    /// each reconnect attempt rather than continuously while a connection is healthy.
+    #[serde(with = "ingress_serde")]
+    pub address: Ingress,
+}
+
+mod ingress_serde {
+    use super::*;
+    use serde::{de::Error as _, Deserializer, Serializer};
+
+    pub fn deserialize<'de, D>(deserializer: D) -> Result<Ingress, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        let value = String::deserialize(deserializer)?;
+        if let Ok(address) = SocketAddr::from_str(&value) {
+            return Ok(Ingress::Socket(address));
+        }
+        if value.contains("://") {
+            return Err(D::Error::custom(
+                "peer address must not contain a URL scheme or path",
+            ));
+        }
+
+        let (host, port) = value
+            .rsplit_once(':')
+            .ok_or_else(|| D::Error::custom("peer address is missing a port"))?;
+        if host.is_empty() {
+            return Err(D::Error::custom("peer address is missing a host"));
+        }
+        if port.is_empty() {
+            return Err(D::Error::custom("peer address is missing a port"));
+        }
+        if host.contains(':') {
+            return Err(D::Error::custom(
+                "IPv6 peer addresses must use bracketed socket syntax",
+            ));
+        }
+        let port = port
+            .parse::<u16>()
+            .map_err(|error| D::Error::custom(format!("invalid peer address port: {error}")))?;
+        let host = Hostname::new(host)
+            .map_err(|error| D::Error::custom(format!("invalid peer address hostname: {error}")))?;
+        Ok(Ingress::Dns { host, port })
+    }
+
+    pub fn serialize<S>(ingress: &Ingress, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        match ingress {
+            Ingress::Socket(address) => serializer.serialize_str(&address.to_string()),
+            Ingress::Dns { host, port } => {
+                serializer.serialize_str(&format!("{host}:{port}"))
+            }
+        }
+    }
 }
 
 #[derive(Clone, Debug, Serialize, Deserialize)]
@@ -284,14 +373,14 @@ pub fn generate_local_testnet(config: LocalTestnetConfig) -> Result<LocalTestnet
         }
         let config_path = config.base_data_dir.join(format!("{name}.toml"));
         let listen_address = SocketAddr::new(config.bind_ip, port);
-        let dialable_address = SocketAddr::new(
+        let dialable_address = Ingress::Socket(SocketAddr::new(
             config
                 .public_ips
                 .as_ref()
                 .map(|hosts| hosts[index])
                 .unwrap_or(config.bind_ip),
             port,
-        );
+        ));
         let bootstrappers = participants
             .iter()
             .enumerate()
@@ -299,14 +388,14 @@ pub fn generate_local_testnet(config: LocalTestnetConfig) -> Result<LocalTestnet
             .map(|(candidate, public_key)| {
                 Ok(BootstrapperConfig {
                     public_key: encode(public_key),
-                    address: SocketAddr::new(
+                    address: Ingress::Socket(SocketAddr::new(
                         config
                             .public_ips
                             .as_ref()
                             .map(|hosts| hosts[candidate])
                             .unwrap_or(config.bind_ip),
                         config.base_port + u16::try_from(candidate)?,
-                    ),
+                    )),
                 })
             })
             .collect::<Result<Vec<_>, Error>>()?;
@@ -328,6 +417,10 @@ pub fn generate_local_testnet(config: LocalTestnetConfig) -> Result<LocalTestnet
             storage_dir: storage_dir.clone(),
             genesis_path: config.genesis_path.clone(),
             indexer_url: config.indexer_url.clone(),
+            indexer_spool_max_entries: default_indexer_spool_max_entries(),
+            indexer_spool_max_bytes: default_indexer_spool_max_bytes(),
+            indexer_spool_max_payload_bytes: default_indexer_spool_max_payload_bytes(),
+            indexer_spool_max_age_seconds: default_indexer_spool_max_age_seconds(),
             consensus: ConsensusConfig::default(),
             networking: NetworkConfig::default(),
             state_sync: false,
@@ -477,7 +570,7 @@ async fn start_node(
         .map(|bootstrapper| {
             Ok((
                 decode_unit::<PublicKey>(&bootstrapper.public_key, "bootstrapper.public_key")?,
-                Ingress::from(bootstrapper.address),
+                bootstrapper.address.clone(),
             ))
         })
         .collect::<Result<Vec<_>, Error>>()?;
@@ -486,6 +579,7 @@ async fn start_node(
         node = %config.name,
         public_key = %public_key,
         listen = %config.listen_address,
+        dialable = ?config.dialable_address,
         rpc = %config.rpc_address,
         metrics = %config.metrics_address,
         "starting coins-chain validator"
@@ -495,7 +589,7 @@ async fn start_node(
         private_key.clone(),
         NAMESPACE,
         config.listen_address,
-        config.dialable_address,
+        config.dialable_address.clone(),
         bootstrappers,
         config.networking.max_message_size,
     );
@@ -541,13 +635,12 @@ async fn start_node(
         blocker: oracle.clone(),
         manager: oracle.clone(),
         partition_prefix: config.name.clone(),
-        blocks_freezer_table_initial_size: FREEZER_TABLE_INITIAL_SIZE,
-        finalized_freezer_table_initial_size: FREEZER_TABLE_INITIAL_SIZE,
         signer: private_key,
         dkg_storage_key,
         output,
         share: Some(share),
         peer_config: config.peer_config.clone(),
+        epoch_length: BLOCKS_PER_EPOCH,
         leader_timeout: Duration::from_millis(config.consensus.leader_timeout_ms),
         certification_timeout: Duration::from_millis(config.consensus.certification_timeout_ms),
         strategy: context
@@ -557,6 +650,12 @@ async fn start_node(
         pool_config: PoolConfig::default(),
         genesis: read_genesis(config.genesis_path.as_ref())?,
         indexer: indexer_client.map(|(client, _metrics)| client),
+        indexer_spool_limits: indexer::SpoolLimits {
+            max_entries: config.indexer_spool_max_entries,
+            max_bytes: config.indexer_spool_max_bytes,
+            max_payload_bytes: config.indexer_spool_max_payload_bytes,
+            max_age: Duration::from_secs(config.indexer_spool_max_age_seconds),
+        },
     };
     let dkg_callback: Box<dyn UpdateCallBack<MinSig, PublicKey>> = ContinueOnUpdate::boxed();
 
diff --git a/examples/coins/chain/src/tests/indexer.rs b/examples/coins/chain/src/tests/indexer.rs
index b26e49e..d26399b 100644
--- a/examples/coins/chain/src/tests/indexer.rs
+++ b/examples/coins/chain/src/tests/indexer.rs
@@ -1,6 +1,6 @@
 use crate::{
     indexer::{
-        BackfillDecision, BackfillPhase, BackfillResetReason, BackfillWaitReason,
+        BackfillDecision, BackfillPhase, BackfillWaitReason,
         BlockMetricSource, DkgUploadStatus, HttpArtifact, IndexerMetrics, LiveUploadArtifact,
         ProducerActivity, ProducerStatus, QueueReadSource, QueueStatus, SharedCacheSource,
         SharedRetentionReason,
@@ -346,10 +346,13 @@ fn shared_state_metrics_use_expected_labels() {
             uploaded_digests: 4,
             latest_finalized_height: 8,
             acked_through_height: 6,
+            spool_entries: 5,
+            spool_logical_bytes: 512,
+            spool_oldest_height: 2,
+            spool_oldest_enqueued_at_millis: 1_000,
         });
         metrics.shared_cache_inserted(SharedCacheSource::ProducerRecord);
         metrics.shared_cache_inserted(SharedCacheSource::LiveCertificate);
-        metrics.shared_cache_inserted(SharedCacheSource::ConsumerMarshal);
         metrics.shared_cache_removed(SharedRetentionReason::Uploaded);
         metrics.shared_cache_removed(SharedRetentionReason::CertificateFinished);
         metrics.shared_cache_removed(SharedRetentionReason::Pruned);
@@ -373,9 +376,6 @@ fn shared_state_metrics_use_expected_labels() {
         assert!(encoded.contains(
             "indexer_shared_cache_insert_total{source=\"live_certificate\"} 1",
         ));
-        assert!(encoded.contains(
-            "indexer_shared_cache_insert_total{source=\"consumer_marshal\"} 1",
-        ));
         assert!(encoded.contains(
             "indexer_shared_cache_remove_total{reason=\"uploaded\"} 1",
         ));
@@ -402,7 +402,6 @@ fn block_metrics_use_expected_sources() {
         metrics.observe_block(BlockMetricSource::ProducerRecord, &block);
         metrics.observe_block(BlockMetricSource::LiveCertificate, &block);
         metrics.observe_block(BlockMetricSource::ConsumerCached, &block);
-        metrics.observe_block(BlockMetricSource::ConsumerMarshal, &block);
 
         let encoded = context.encode();
         assert!(encoded.contains("indexer_block_estimated_bytes_bucket"));
@@ -410,7 +409,6 @@ fn block_metrics_use_expected_sources() {
         assert!(encoded.contains("source=\"producer_record\""));
         assert!(encoded.contains("source=\"live_certificate\""));
         assert!(encoded.contains("source=\"consumer_cached\""));
-        assert!(encoded.contains("source=\"consumer_marshal\""));
     });
 }
 
@@ -433,20 +431,11 @@ fn backfill_metrics_use_expected_labels() {
             upload.skipped();
         }
         metrics.backfill_decision(BackfillDecision::Skip, BackfillPhase::Start);
-        metrics.backfill_decision(BackfillDecision::Wait, BackfillPhase::BeforeBlock);
         metrics.backfill_decision(BackfillDecision::Proceed, BackfillPhase::BeforeAttempt);
         metrics.backfill_retry(BackfillWaitReason::CertificateUpload);
-        metrics.backfill_retry(BackfillWaitReason::MissingBlock);
-        metrics.backfill_retry(BackfillWaitReason::MissingFinalization);
-        metrics.backfill_retry(BackfillWaitReason::MismatchedFinalization);
         metrics.backfill_retry(BackfillWaitReason::HttpError);
         metrics.backfill_waited(BackfillWaitReason::CertificateUpload, Duration::from_millis(1));
-        metrics.backfill_waited(BackfillWaitReason::MissingBlock, Duration::from_millis(1));
-        metrics.backfill_waited(BackfillWaitReason::MissingFinalization, Duration::from_millis(1));
-        metrics.backfill_waited(BackfillWaitReason::MismatchedFinalization, Duration::from_millis(1));
         metrics.backfill_waited(BackfillWaitReason::HttpError, Duration::from_millis(1));
-        metrics.backfill_queue_reset(BackfillResetReason::MissingFinalization, 32, 31);
-        metrics.backfill_queue_reset(BackfillResetReason::MismatchedFinalization, 2, 1);
         metrics.queue_enqueued(QueueStatus::Success);
         metrics.queue_enqueued(QueueStatus::Failure);
         metrics.queue_acked(QueueStatus::Success);
@@ -486,9 +475,6 @@ fn backfill_metrics_use_expected_labels() {
         assert!(encoded.contains(
             "indexer_backfill_decision_total{decision=\"skip\",phase=\"start\"} 1",
         ));
-        assert!(encoded.contains(
-            "indexer_backfill_decision_total{decision=\"wait\",phase=\"before_block\"} 1",
-        ));
         assert!(encoded.contains(
             "indexer_backfill_decision_total{decision=\"proceed\",phase=\"before_attempt\"} 1",
         ));
@@ -496,20 +482,7 @@ fn backfill_metrics_use_expected_labels() {
         assert!(encoded.contains(
             "indexer_backfill_retry_total{reason=\"certificate_upload\"} 1",
         ));
-        assert!(encoded.contains("indexer_backfill_retry_total{reason=\"missing_block\"} 1"));
-        assert!(encoded
-            .contains("indexer_backfill_retry_total{reason=\"missing_finalization\"} 1"));
-        assert!(encoded
-            .contains("indexer_backfill_retry_total{reason=\"mismatched_finalization\"} 1"));
         assert!(encoded.contains("indexer_backfill_retry_total{reason=\"http_error\"} 1"));
-        assert!(encoded.contains(
-            "indexer_backfill_queue_reset_total{reason=\"missing_finalization\"} 1",
-        ));
-        assert!(encoded.contains(
-            "indexer_backfill_queue_reset_total{reason=\"mismatched_finalization\"} 1",
-        ));
-        assert!(encoded.contains("indexer_backfill_queue_reset_abandoned_entries_bucket"));
-        assert!(encoded.contains("indexer_backfill_queue_reset_abandoned_height_span_bucket"));
         assert!(encoded.contains("indexer_backfill_active_block_estimated_bytes 0"));
         assert!(encoded.contains("indexer_backfill_active_body_estimated_bytes 0"));
         assert!(encoded.contains("indexer_queue_enqueue_total{status=\"success\"} 1"));
diff --git a/examples/coins/chain/src/tests/testnet.rs b/examples/coins/chain/src/tests/testnet.rs
index ac240e4..cd701fb 100644
--- a/examples/coins/chain/src/tests/testnet.rs
+++ b/examples/coins/chain/src/tests/testnet.rs
@@ -5,16 +5,24 @@
 //! as `narae` consume. [`run_node`] boots a single validator from one of those configs on the
 //! tokio runtime with authenticated peer discovery, and serves the aggregated JSON-RPC module.
 
-use commonware_cryptography::{bls12381::primitives::group, ed25519};
+use commonware_cryptography::{bls12381::primitives::group, ed25519, Signer as _};
+use commonware_macros::select;
+use commonware_p2p::{
+    authenticated::discovery::{self, Network},
+    Ingress, Manager, Receiver as _, Recipients, Sender as _,
+};
+use commonware_runtime::{deterministic, Clock as _, Quota, Runner as _, Supervisor as _};
+use commonware_utils::{ordered::Set, Hostname, NZU32};
 use std::collections::HashSet;
 use std::{
     fs,
-    net::{IpAddr, Ipv4Addr},
+    net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr},
     num::NonZeroU32,
     path::PathBuf,
+    time::Duration,
 };
 
-use crate::testnet::*;
+use crate::{testnet::*, NAMESPACE};
 
 #[test]
 fn generated_testnet_has_unique_ports_dirs_and_complete_peer_sets() {
@@ -124,17 +132,20 @@ fn generated_testnet_can_advertise_remote_hosts() {
 
     let first = NodeConfig::read(&manifest.nodes[0].config_path).expect("read first config");
     let second = NodeConfig::read(&manifest.nodes[1].config_path).expect("read second config");
+    let first_raw = fs::read_to_string(&manifest.nodes[0].config_path).expect("read first config");
 
     assert_eq!(first.listen_address.ip(), IpAddr::V4(Ipv4Addr::UNSPECIFIED));
-    assert_eq!(first.dialable_address.ip(), public_ips[0]);
-    assert_eq!(first.bootstrappers[0].address.ip(), public_ips[1]);
+    assert_eq!(first.dialable_address.ip(), Some(public_ips[0]));
+    assert_eq!(first.bootstrappers[0].address.ip(), Some(public_ips[1]));
+    assert!(first_raw.contains("dialable_address = \"192.0.2.10:30000\""));
+    assert!(first_raw.contains("address = \"192.0.2.11:30001\""));
     assert_eq!(first.storage_dir, storage_dir);
     assert_eq!(
         first.indexer_url.as_deref(),
         Some("https://indexer.example.com/coins-chain")
     );
-    assert_eq!(second.dialable_address.ip(), public_ips[1]);
-    assert_eq!(second.bootstrappers[0].address.ip(), public_ips[0]);
+    assert_eq!(second.dialable_address.ip(), Some(public_ips[1]));
+    assert_eq!(second.bootstrappers[0].address.ip(), Some(public_ips[0]));
     assert_eq!(
         second.indexer_url.as_deref(),
         Some("https://indexer.example.com/coins-chain")
@@ -142,3 +153,454 @@ fn generated_testnet_can_advertise_remote_hosts() {
 
     let _ = fs::remove_dir_all(dir);
 }
+
+#[test]
+fn peer_addresses_round_trip_dns_and_ipv6_as_strings() {
+    let dir = std::env::temp_dir().join(format!(
+        "coins-chain-address-round-trip-{}",
+        std::process::id()
+    ));
+    let _ = fs::remove_dir_all(&dir);
+    let manifest = generate_local_testnet(LocalTestnetConfig {
+        validators: 2,
+        base_port: 32_000,
+        base_rpc_port: 33_000,
+        base_metrics_port: 34_000,
+        base_data_dir: dir.clone(),
+        bind_ip: IpAddr::V4(Ipv4Addr::LOCALHOST),
+        public_ips: None,
+        storage_dir: None,
+        genesis_path: None,
+        indexer_url: None,
+        seed: 9,
+    })
+    .expect("generate testnet");
+    let path = &manifest.nodes[0].config_path;
+    let mut config = NodeConfig::read(path).expect("read generated config");
+    config.dialable_address = Ingress::Dns {
+        host: Hostname::new("validator-0.example.com").unwrap(),
+        port: 30_000,
+    };
+    config.bootstrappers[0].address = Ingress::Dns {
+        host: Hostname::new("validator-1.example.com").unwrap(),
+        port: 30_001,
+    };
+    config.write(path).expect("write DNS config");
+
+    let raw = fs::read_to_string(path).expect("read DNS config");
+    assert!(raw.contains("dialable_address = \"validator-0.example.com:30000\""));
+    assert!(raw.contains("address = \"validator-1.example.com:30001\""));
+    let decoded = NodeConfig::read(path).expect("read DNS config");
+    assert_eq!(decoded.dialable_address, config.dialable_address);
+    assert_eq!(
+        decoded.bootstrappers[0].address,
+        config.bootstrappers[0].address
+    );
+
+    config.dialable_address = Ingress::Socket(SocketAddr::new(
+        IpAddr::V6("2001:db8::10".parse::<Ipv6Addr>().unwrap()),
+        30_000,
+    ));
+    config.write(path).expect("write IPv6 config");
+    let raw = fs::read_to_string(path).expect("read IPv6 config");
+    assert!(raw.contains("dialable_address = \"[2001:db8::10]:30000\""));
+    assert_eq!(
+        NodeConfig::read(path)
+            .expect("read IPv6 config")
+            .dialable_address,
+        config.dialable_address
+    );
+
+    let _ = fs::remove_dir_all(dir);
+}
+
+#[test]
+fn invalid_peer_addresses_are_rejected_through_node_config_read() {
+    let dir = std::env::temp_dir().join(format!(
+        "coins-chain-invalid-addresses-{}",
+        std::process::id()
+    ));
+    let _ = fs::remove_dir_all(&dir);
+    let manifest = generate_local_testnet(LocalTestnetConfig {
+        validators: 1,
+        base_port: 35_000,
+        base_rpc_port: 36_000,
+        base_metrics_port: 37_000,
+        base_data_dir: dir.clone(),
+        bind_ip: IpAddr::V4(Ipv4Addr::LOCALHOST),
+        public_ips: None,
+        storage_dir: None,
+        genesis_path: None,
+        indexer_url: None,
+        seed: 10,
+    })
+    .expect("generate testnet");
+    let valid_path = &manifest.nodes[0].config_path;
+    let valid = fs::read_to_string(valid_path).expect("read generated config");
+    let valid_address = "dialable_address = \"127.0.0.1:35000\"";
+    assert!(valid.contains(valid_address));
+
+    let cases = [
+        ("validator.example.com", "missing a port"),
+        ("validator.example.com:not-a-port", "invalid peer address port"),
+        ("validator.example.com:65536", "invalid peer address port"),
+        ("-validator.example.com:30000", "invalid peer address hostname"),
+        (
+            "https://validator.example.com:30000",
+            "must not contain a URL scheme or path",
+        ),
+        ("2001:db8::10:30000", "must use bracketed socket syntax"),
+    ];
+    for (index, (address, expected)) in cases.into_iter().enumerate() {
+        let path = dir.join(format!("invalid-{index}.toml"));
+        let invalid = valid.replace(
+            valid_address,
+            &format!("dialable_address = \"{address}\""),
+        );
+        fs::write(&path, invalid).expect("write invalid config");
+        let error = NodeConfig::read(&path).expect_err("invalid address should fail");
+        assert!(
+            error.to_string().contains(expected),
+            "unexpected error for {address}: {error}"
+        );
+    }
+
+    let _ = fs::remove_dir_all(dir);
+}
+
+fn parse_dns_address_through_node_config(address: &str, seed: u64) -> Ingress {
+    let dir = std::env::temp_dir().join(format!(
+        "coins-chain-parse-dns-{seed}-{}",
+        std::process::id()
+    ));
+    let _ = fs::remove_dir_all(&dir);
+    let manifest = generate_local_testnet(LocalTestnetConfig {
+        validators: 1,
+        base_port: 38_000,
+        base_rpc_port: 38_100,
+        base_metrics_port: 38_200,
+        base_data_dir: dir.clone(),
+        bind_ip: IpAddr::V4(Ipv4Addr::LOCALHOST),
+        public_ips: None,
+        storage_dir: None,
+        genesis_path: None,
+        indexer_url: None,
+        seed,
+    })
+    .expect("generate parser fixture");
+    let path = &manifest.nodes[0].config_path;
+    let valid = fs::read_to_string(path).expect("read parser fixture");
+    let dns = valid.replace(
+        "dialable_address = \"127.0.0.1:38000\"",
+        &format!("dialable_address = \"{address}\""),
+    );
+    fs::write(path, dns).expect("write parser fixture");
+    let ingress = NodeConfig::read(path)
+        .expect("parse DNS address")
+        .dialable_address;
+    let _ = fs::remove_dir_all(dir);
+    ingress
+}
+
+fn reconnect_config(
+    key: ed25519::PrivateKey,
+    listen: SocketAddr,
+    dialable: Ingress,
+    bootstrappers: Vec<(ed25519::PublicKey, Ingress)>,
+) -> discovery::Config<ed25519::PrivateKey> {
+    let mut config = discovery::Config::local(
+        key,
+        NAMESPACE,
+        listen,
+        dialable,
+        bootstrappers,
+        1024 * 1024,
+    );
+    config.peer_connection_cooldown = Duration::from_millis(100);
+    config.dial_frequency = Duration::from_millis(50);
+    config.gossip_bit_vec_frequency = Duration::from_millis(100);
+    config
+}
+
+fn run_dns_bootstrapper_redeployment(seed: u64, dns: Ingress) -> String {
+    let runner = deterministic::Runner::new(
+        deterministic::Config::new()
+            .with_seed(seed)
+            .with_timeout(Some(Duration::from_secs(20))),
+    );
+    runner.start(|context| async move {
+        let key_a = ed25519::PrivateKey::from_seed(100);
+        let key_b = ed25519::PrivateKey::from_seed(101);
+        let public_a = key_a.public_key();
+        let public_b = key_b.public_key();
+        let peers = Set::try_from(vec![public_a.clone(), public_b.clone()]).unwrap();
+        let socket_a = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 10)), 39_000);
+        let socket_b1 = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 11)), 39_000);
+        let socket_b2 = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 12)), 39_000);
+        let unreachable_a = Ingress::Socket(SocketAddr::new(
+            IpAddr::V4(Ipv4Addr::new(127, 0, 0, 250)),
+            49_000,
+        ));
+        context.resolver_register("validator-b-bootstrap.test", Some(vec![socket_b1.ip()]));
+
+        let config_b = reconnect_config(key_b.clone(), socket_b1, dns.clone(), vec![]);
+        let (mut network_b, mut oracle_b) =
+            Network::new(context.child("b_initial").child("network"), config_b);
+        oracle_b.track(0, peers.clone());
+        let (mut sender_b, mut receiver_b) =
+            network_b.register(0, Quota::per_second(NZU32!(100)), 128);
+        let handle_b = network_b.start();
+
+        let config_a = reconnect_config(
+            key_a,
+            socket_a,
+            unreachable_a,
+            vec![(public_b.clone(), dns.clone())],
+        );
+        assert!(config_a.allow_dns);
+        let (mut network_a, mut oracle_a) =
+            Network::new(context.child("a").child("network"), config_a);
+        oracle_a.track(0, peers.clone());
+        let (mut sender_a, mut receiver_a) =
+            network_a.register(0, Quota::per_second(NZU32!(100)), 128);
+        let _handle_a = network_a.start();
+
+        let initial_exchange = async {
+            loop {
+                sender_a.send(
+                    Recipients::One(public_b.clone()),
+                    b"a-before".to_vec(),
+                    true,
+                );
+                select! {
+                    result = receiver_b.recv() => {
+                        let (from, message) = result.unwrap();
+                        assert_eq!(from, public_a);
+                        assert_eq!(message.as_ref(), b"a-before");
+                        break;
+                    },
+                    _ = context.sleep(Duration::from_millis(50)) => {},
+                }
+            }
+            loop {
+                sender_b.send(
+                    Recipients::One(public_a.clone()),
+                    b"b-before".to_vec(),
+                    true,
+                );
+                select! {
+                    result = receiver_a.recv() => {
+                        let (from, message) = result.unwrap();
+                        assert_eq!(from, public_b);
+                        assert_eq!(message.as_ref(), b"b-before");
+                        break;
+                    },
+                    _ = context.sleep(Duration::from_millis(50)) => {},
+                }
+            }
+        };
+        select! {
+            _ = initial_exchange => {},
+            _ = context.sleep(Duration::from_secs(5)) => panic!("initial DNS connection timed out"),
+        }
+
+        handle_b.abort();
+        drop(sender_b);
+        drop(receiver_b);
+        context.resolver_register("validator-b-bootstrap.test", Some(vec![socket_b2.ip()]));
+
+        // The restarted peer has no bootstrappers and A advertises an unreachable address, so B
+        // cannot initiate the replacement connection.
+        let config_b = reconnect_config(key_b, socket_b2, dns, vec![]);
+        let (mut network_b, mut oracle_b) =
+            Network::new(context.child("b_restarted").child("network"), config_b);
+        oracle_b.track(0, peers);
+        let (_sender_b, mut receiver_b) =
+            network_b.register(0, Quota::per_second(NZU32!(100)), 128);
+        let _handle_b = network_b.start();
+
+        let reconnected = async {
+            loop {
+                sender_a.send(
+                    Recipients::One(public_b.clone()),
+                    b"a-after".to_vec(),
+                    true,
+                );
+                select! {
+                    result = receiver_b.recv() => {
+                        let (from, message) = result.unwrap();
+                        assert_eq!(from, public_a);
+                        assert_eq!(message.as_ref(), b"a-after");
+                        break;
+                    },
+                    _ = context.sleep(Duration::from_millis(50)) => {},
+                }
+            }
+        };
+        select! {
+            _ = reconnected => {},
+            _ = context.sleep(Duration::from_secs(5)) => panic!("DNS bootstrapper did not reconnect"),
+        }
+        context.auditor().state()
+    })
+}
+
+#[test]
+fn dns_bootstrapper_is_resolved_again_after_redeployment() {
+    let dns = parse_dns_address_through_node_config("validator-b-bootstrap.test:39000", 11);
+    let first = run_dns_bootstrapper_redeployment(42, dns.clone());
+    let second = run_dns_bootstrapper_redeployment(42, dns);
+    assert_eq!(first, second);
+}
+
+fn run_discovered_dns_redeployment(seed: u64, dns: Ingress) -> String {
+    let runner = deterministic::Runner::new(
+        deterministic::Config::new()
+            .with_seed(seed)
+            .with_timeout(Some(Duration::from_secs(20))),
+    );
+    runner.start(|context| async move {
+        let key_a = ed25519::PrivateKey::from_seed(200);
+        let key_b = ed25519::PrivateKey::from_seed(201);
+        let key_c = ed25519::PrivateKey::from_seed(202);
+        let public_a = key_a.public_key();
+        let public_b = key_b.public_key();
+        let public_c = key_c.public_key();
+        let peers = Set::try_from(vec![
+            public_a.clone(),
+            public_b.clone(),
+            public_c.clone(),
+        ])
+        .unwrap();
+        let socket_a = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 1, 10)), 40_000);
+        let socket_b1 = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 1, 11)), 40_000);
+        let socket_b2 = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 1, 12)), 40_000);
+        let socket_c = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 1, 13)), 40_000);
+        let unreachable_a = Ingress::Socket(SocketAddr::new(
+            IpAddr::V4(Ipv4Addr::new(127, 0, 1, 250)),
+            50_000,
+        ));
+        context.resolver_register("validator-b-discovered.test", Some(vec![socket_b1.ip()]));
+
+        let config_c = reconnect_config(key_c, socket_c, socket_c.into(), vec![]);
+        let (mut network_c, mut oracle_c) =
+            Network::new(context.child("c").child("network"), config_c);
+        oracle_c.track(0, peers.clone());
+        let (_sender_c, _receiver_c) =
+            network_c.register(0, Quota::per_second(NZU32!(100)), 128);
+        let _handle_c = network_c.start();
+
+        let config_b = reconnect_config(
+            key_b.clone(),
+            socket_b1,
+            dns.clone(),
+            vec![(public_c.clone(), socket_c.into())],
+        );
+        let (mut network_b, mut oracle_b) =
+            Network::new(context.child("b_initial").child("network"), config_b);
+        oracle_b.track(0, peers.clone());
+        let (mut sender_b, mut receiver_b) =
+            network_b.register(0, Quota::per_second(NZU32!(100)), 128);
+        let handle_b = network_b.start();
+
+        // A only knows C initially. It must learn B's DNS ingress through discovery.
+        let config_a = reconnect_config(
+            key_a,
+            socket_a,
+            unreachable_a,
+            vec![(public_c, socket_c.into())],
+        );
+        let (mut network_a, mut oracle_a) =
+            Network::new(context.child("a").child("network"), config_a);
+        oracle_a.track(0, peers.clone());
+        let (mut sender_a, mut receiver_a) =
+            network_a.register(0, Quota::per_second(NZU32!(100)), 128);
+        let _handle_a = network_a.start();
+
+        let learned_and_connected = async {
+            loop {
+                sender_a.send(
+                    Recipients::One(public_b.clone()),
+                    b"a-before".to_vec(),
+                    true,
+                );
+                select! {
+                    result = receiver_b.recv() => {
+                        let (from, message) = result.unwrap();
+                        assert_eq!(from, public_a);
+                        assert_eq!(message.as_ref(), b"a-before");
+                        break;
+                    },
+                    _ = context.sleep(Duration::from_millis(50)) => {},
+                }
+            }
+            loop {
+                sender_b.send(
+                    Recipients::One(public_a.clone()),
+                    b"b-before".to_vec(),
+                    true,
+                );
+                select! {
+                    result = receiver_a.recv() => {
+                        let (from, message) = result.unwrap();
+                        assert_eq!(from, public_b);
+                        assert_eq!(message.as_ref(), b"b-before");
+                        break;
+                    },
+                    _ = context.sleep(Duration::from_millis(50)) => {},
+                }
+            }
+        };
+        select! {
+            _ = learned_and_connected => {},
+            _ = context.sleep(Duration::from_secs(7)) => panic!("A did not discover B's DNS ingress"),
+        }
+
+        handle_b.abort();
+        drop(sender_b);
+        drop(receiver_b);
+        context.resolver_register("validator-b-discovered.test", Some(vec![socket_b2.ip()]));
+
+        // B has no bootstrapper after restart and A's advertised address is unreachable. A must
+        // retain the discovered hostname, resolve it again, and dial B at its replacement IP.
+        let config_b = reconnect_config(key_b, socket_b2, dns, vec![]);
+        let (mut network_b, mut oracle_b) =
+            Network::new(context.child("b_restarted").child("network"), config_b);
+        oracle_b.track(0, peers);
+        let (_sender_b, mut receiver_b) =
+            network_b.register(0, Quota::per_second(NZU32!(100)), 128);
+        let _handle_b = network_b.start();
+
+        let reconnected = async {
+            loop {
+                sender_a.send(
+                    Recipients::One(public_b.clone()),
+                    b"a-after".to_vec(),
+                    true,
+                );
+                select! {
+                    result = receiver_b.recv() => {
+                        let (from, message) = result.unwrap();
+                        assert_eq!(from, public_a);
+                        assert_eq!(message.as_ref(), b"a-after");
+                        break;
+                    },
+                    _ = context.sleep(Duration::from_millis(50)) => {},
+                }
+            }
+        };
+        select! {
+            _ = reconnected => {},
+            _ = context.sleep(Duration::from_secs(7)) => panic!("discovered DNS peer did not reconnect"),
+        }
+        context.auditor().state()
+    })
+}
+
+#[test]
+fn discovered_dns_address_is_resolved_again_after_redeployment() {
+    let dns = parse_dns_address_through_node_config("validator-b-discovered.test:40000", 12);
+    let first = run_discovered_dns_redeployment(84, dns.clone());
+    let second = run_discovered_dns_redeployment(84, dns);
+    assert_eq!(first, second);
+}
diff --git a/examples/coins/chain/tests/coins.rs b/examples/coins/chain/tests/coins.rs
index 9051289..da9ec09 100644
--- a/examples/coins/chain/tests/coins.rs
+++ b/examples/coins/chain/tests/coins.rs
@@ -8,7 +8,8 @@ use commonware_cryptography::Signer as _;
 use commonware_cryptography::{Hasher, Sha256};
 use commonware_macros::{select, test_traced};
 use commonware_p2p::simulated::Link;
-use commonware_runtime::{deterministic, Clock, Runner as _};
+use commonware_runtime::{deterministic, Clock, Runner as _, Spawner as _, Supervisor as _};
+use commonware_utils::NZU64;
 use nunchi_authority::{
     proposal_id, AuthorityOperation, MultisigPolicy, RegistryChange,
     Transaction as AuthorityTransaction,
@@ -104,6 +105,56 @@ fn reaches_height_100() {
     });
 }
 
+#[test_traced]
+fn crosses_epoch_boundary_and_reclaims_retired_partition() {
+    with_large_stack(|| {
+        let executor = deterministic::Runner::timed(Duration::from_secs(60));
+        executor.start(|mut context| async move {
+            let cfg = ValidatorConfig {
+                epoch_length: NZU64!(20),
+                ..ValidatorConfig::default()
+            };
+            let mut network = TestNetworkBuilder::new(VALIDATORS)
+                .with_initial_link(reliable_link())
+                .with_validator_config(cfg)
+                .build(&mut context)
+                .await;
+            network.start_all().await;
+
+            let validator_task_prefix = "validator";
+            assert!(network.running_tasks(validator_task_prefix) > 0);
+
+            network.run_until_height(25).await;
+
+            network.assert_validator_metric("engine_orchestrator_latest_epoch", &[], 1);
+            network.assert_validator_metric(
+                "engine_orchestrator_consensus_partitions_active",
+                &[],
+                1,
+            );
+            network.assert_validator_metric(
+                "engine_orchestrator_consensus_partition_cleanup_watermark",
+                &[],
+                1,
+            );
+            network.assert_validator_metric(
+                "engine_orchestrator_consensus_partition_cleanup_total",
+                &[("status", "removed")],
+                1,
+            );
+            network.assert_consensus_partition_missing(0).await;
+            assert!(network.running_tasks(validator_task_prefix) > 0);
+
+            let shutdown = network.context().child("shutdown");
+            shutdown
+                .stop(0, Some(Duration::from_secs(10)))
+                .await
+                .expect("validator tasks should stop cleanly");
+            assert_eq!(network.running_tasks(validator_task_prefix), 0);
+        });
+    });
+}
+
 #[test_traced]
 fn state_syncs_late_validator() {
     with_large_stack(|| {
@@ -154,6 +205,7 @@ fn recovers_unclean_shutdown() {
                 let cfg = ValidatorConfig {
                     leader_timeout: Duration::from_millis(250),
                     certification_timeout: Duration::from_millis(500),
+                    ..ValidatorConfig::default()
                 };
 
                 let wait =
diff --git a/examples/coins/chain/tests/common/network.rs b/examples/coins/chain/tests/common/network.rs
index 04495cb..21d473f 100644
--- a/examples/coins/chain/tests/common/network.rs
+++ b/examples/coins/chain/tests/common/network.rs
@@ -15,7 +15,7 @@ use commonware_p2p::{
 use commonware_parallel::Sequential;
 use commonware_runtime::{
     deterministic::{self, Runner},
-    Clock, Metrics, Runner as _, Supervisor,
+    Clock, Error as RuntimeError, Metrics, Runner as _, Storage, Supervisor,
 };
 use commonware_utils::{
     ordered::{Map, Set},
@@ -27,7 +27,7 @@ use nunchi_coins::{Address, Ledger};
 use nunchi_coins_chain::{
     engine::{Config, Engine},
     execution::NodeHandle,
-    PublicKey, Transaction,
+    PublicKey, Transaction, BLOCKS_PER_EPOCH,
 };
 use nunchi_common::QmdbReader;
 use nunchi_dkg::{ContinueOnUpdate, PeerConfig};
@@ -35,10 +35,10 @@ use nunchi_mempool::{MempoolHandle, PoolConfig};
 use nunchi_oracle::OracleLedger;
 use std::{
     collections::{HashMap, HashSet},
+    num::NonZeroU64,
     time::Duration,
 };
 
-const FREEZER_TABLE_INITIAL_SIZE: u32 = 2u32.pow(14); // 1MB
 const TEST_QUOTA: Quota = Quota::per_second(NZU32!(u32::MAX));
 const MAX_BLOCK_TRANSACTIONS: usize = 256;
 
@@ -109,6 +109,7 @@ pub(crate) fn lossy_link() -> Link {
 
 #[derive(Clone)]
 pub(crate) struct ValidatorConfig {
+    pub(crate) epoch_length: NonZeroU64,
     pub(crate) leader_timeout: Duration,
     pub(crate) certification_timeout: Duration,
 }
@@ -116,6 +117,7 @@ pub(crate) struct ValidatorConfig {
 impl Default for ValidatorConfig {
     fn default() -> Self {
         Self {
+            epoch_length: BLOCKS_PER_EPOCH,
             leader_timeout: Duration::from_secs(1),
             certification_timeout: Duration::from_secs(2),
         }
@@ -356,6 +358,83 @@ impl TestNetwork<'_> {
         }
     }
 
+    pub(crate) fn assert_validator_metric(
+        &self,
+        suffix: &str,
+        required_labels: &[(&str, &str)],
+        expected_value: u64,
+    ) {
+        let expected = self.started_validator_ids();
+        let mut observed = HashSet::new();
+        let metrics = self.context.encode();
+        for line in metrics.lines() {
+            let Some((metric, labels, value)) = validator_metric_sample(line) else {
+                continue;
+            };
+            if !metric.ends_with(suffix)
+                || !required_labels
+                    .iter()
+                    .all(|(name, expected)| metric_label(labels, name) == Some(*expected))
+            {
+                continue;
+            }
+            let Some(id) = metric_label(labels, "id") else {
+                continue;
+            };
+            if !expected.contains(id) {
+                continue;
+            }
+            assert_eq!(
+                value.parse::<u64>().unwrap(),
+                expected_value,
+                "unexpected {suffix} value for {id}: {line}",
+            );
+            assert!(
+                observed.insert(id.to_string()),
+                "duplicate {suffix} for {id}"
+            );
+        }
+        assert_eq!(
+            observed, expected,
+            "missing {suffix} samples in metrics: {metrics}",
+        );
+    }
+
+    pub(crate) async fn assert_consensus_partition_missing(&self, epoch: u64) {
+        for signer in &self.private_keys {
+            let public_key = signer.public_key();
+            if self.registrations.contains_key(&public_key) {
+                continue;
+            }
+            let partition = format!("validator_{public_key}_consensus_consensus_{epoch}");
+            let result = self.context.scan(&partition).await;
+            assert!(
+                matches!(
+                    result,
+                    Err(RuntimeError::PartitionMissing(ref missing)) if missing == &partition
+                ),
+                "retired consensus partition {partition} still exists: {result:?}",
+            );
+        }
+    }
+
+    pub(crate) fn running_tasks(&self, prefix: &str) -> usize {
+        self.context
+            .encode()
+            .lines()
+            .filter_map(|line| {
+                if !line.starts_with("runtime_tasks_running{") {
+                    return None;
+                }
+                let name = line.split("name=\"").nth(1)?.split('"').next()?;
+                if !name.starts_with(prefix) {
+                    return None;
+                }
+                line.rsplit(' ').next()?.parse::<usize>().ok()
+            })
+            .sum()
+    }
+
     fn started_validator_ids(&self) -> HashSet<String> {
         self.private_keys
             .iter()
@@ -512,13 +591,12 @@ async fn start_validator(
         blocker: oracle.control(public_key.clone()),
         manager: oracle.manager(),
         partition_prefix: uid.clone(),
-        blocks_freezer_table_initial_size: FREEZER_TABLE_INITIAL_SIZE,
-        finalized_freezer_table_initial_size: FREEZER_TABLE_INITIAL_SIZE,
         signer: signer.clone(),
         dkg_storage_key: [9u8; 32],
         output,
         share: Some(share),
         peer_config,
+        epoch_length: cfg.epoch_length,
         leader_timeout: cfg.leader_timeout,
         certification_timeout: cfg.certification_timeout,
         strategy: Sequential,
@@ -527,6 +605,7 @@ async fn start_validator(
         pool_config: PoolConfig::default(),
         genesis: None,
         indexer: None,
+        indexer_spool_limits: Default::default(),
     };
 
     let validator_context = context.child("validator").with_attribute("id", &uid);