Add Reliable Indexer (Attempt 2) (#184)

* spike

* fmt

* nit

* fmt

* add fail certs test

* add more coverage

* spike

* progress

* add metrics

* metrics tests

* spike

* progress

* cleanup

* simplify

* more work

* add some comments

* fmt

* nits

* more handling

* remove token

* nits

* push

* fmt

* nits

* nits

* progress

* flatten

* nit

* nit

* record

* nits

* nits

* comments

* provide constructor

* cleanup

* still complex

* progress

* remove duplicate metric

* dedup

* nit

* cleanup visibility

* cleanup

* nits

* recorder

* producer/consumer

* naming cleanup

* nits

* nits

* simplify

* nits

* nits

* nits

* queue entry

* remove useless labels

* nit

* cleanup

* nits

* nit

* nit

* nit

* naming

* backfiller

* slim down

* remove useless sync

* update dashboard

* add retriable fetch

* fmt

* nits

* nit

* remove raw

* add jitter

* nit

* add resolver

* consensus.rs

* nits

* nit

* flatten

* add better retry behavior

* simpler name

* nit

* nits

* max active

* nits

* nit

* nits

* cleanup consumer

* nit

* nits

* nit

Author: patrick-ogrady

Cargo.lock

@@ -44,7 +44,7 @@ futures-util = "0.3.31"
 tracing = "0.1.41"
 tracing-subscriber = "0.3.19"
 governor = "0.10.2"
-prometheus-client = "0.22.3"
+prometheus-client = "0.24.0"
 clap = "4.5.18"
 tokio = "1.41.0"
 axum = "0.8.8"

Cargo.toml

@@ -33,3 +33,4 @@ tracing = { workspace = true }
 tracing-subscriber = { workspace = true, features = ["fmt", "json"] }
 governor = { workspace = true }
 serde = { version = "1.0.218", features = ["derive"] }
+prometheus-client = { workspace = true }

chain/Cargo.toml

@@ -1,3 +1,4 @@
+use crate::indexer;
 use alto_types::{Block, Context, Scheme, EPOCH};
 use commonware_consensus::{
     marshal::{
@@ -7,8 +8,8 @@ use commonware_consensus::{
     types::{Height, Round, View},
     Heightable, Reporter,
 };
-use commonware_cryptography::{ed25519, sha256, Digest, Digestible, Hasher, Sha256, Signer};
-use commonware_runtime::{Clock, Metrics, Spawner};
+use commonware_cryptography::{ed25519, sha256, Digest as _, Digestible, Hasher, Sha256, Signer};
+use commonware_runtime::{Clock, Metrics, Spawner, Storage};
 use commonware_utils::{Acknowledgement, SystemTimeExt};
 use futures::StreamExt;
 use rand::Rng;
@@ -22,11 +23,12 @@ const GENESIS: &[u8] = b"commonware is neat";
 const SYNCHRONY_BOUND: u64 = 500;
 
 #[derive(Clone)]
-pub struct Application {
+pub struct Application<E: Clock + Storage + Metrics> {
     genesis: Arc<Block>,
+    backfiller: Option<indexer::Producer<E>>,
 }
 
-impl Application {
+impl<E: Clock + Storage + Metrics> Application<E> {
     pub fn new() -> Self {
         let genesis_context = Context {
             round: Round::new(EPOCH, View::zero()),
@@ -36,19 +38,25 @@ impl Application {
         let genesis = Block::new(genesis_context, Sha256::hash(GENESIS), Height::zero(), 0);
         Self {
             genesis: Arc::new(genesis),
+            backfiller: None,
         }
     }
+
+    pub(crate) fn with_backfiller(mut self, backfiller: indexer::Producer<E>) -> Self {
+        self.backfiller = Some(backfiller);
+        self
+    }
 }
 
-impl Default for Application {
+impl<E: Clock + Storage + Metrics> Default for Application<E> {
     fn default() -> Self {
         Self::new()
     }
 }
 
-impl<E> commonware_consensus::Application<E> for Application
+impl<E: Clock + Storage + Metrics> commonware_consensus::Application<E> for Application<E>
 where
-    E: Rng + Spawner + Metrics + Clock,
+    E: Rng + Spawner + Metrics + Clock + Storage,
 {
     type SigningScheme = Scheme;
     type Context = Context;
@@ -65,7 +73,7 @@ where
     ) -> Option<Self::Block> {
         let parent = ancestry.next().await?;
 
-        // Create a new block
+        // Create a new block.
         let mut current = runtime_context.current().epoch_millis();
         if current <= parent.timestamp {
             current = parent.timestamp + 1;
@@ -80,9 +88,9 @@ where
     }
 }
 
-impl<E> commonware_consensus::VerifyingApplication<E> for Application
+impl<E: Clock + Storage + Metrics> commonware_consensus::VerifyingApplication<E> for Application<E>
 where
-    E: Rng + Spawner + Metrics + Clock,
+    E: Rng + Spawner + Metrics + Clock + Storage,
 {
     async fn verify<A: BlockProvider<Block = Self::Block>>(
         &mut self,
@@ -96,7 +104,7 @@ where
             return false;
         };
 
-        // Verify the block
+        // Verify the block.
         if block.timestamp <= parent.timestamp {
             return false;
         }
@@ -108,16 +116,22 @@ where
         // The height and digest invariants are enforced in `Marshaled`:
         // - The block height must be one greater than the parent's height.
         // - The block's parent digest must match the parent's digest.
-
         true
     }
 }
 
-impl Reporter for Application {
+impl<E: Clock + Storage + Metrics> Reporter for Application<E> {
     type Activity = Update<Block>;
 
     async fn report(&mut self, activity: Self::Activity) {
         if let Update::Block(block, ack_rx) = activity {
+            // Cache the finalized block in memory and enqueue its digest
+            // before acking so the consumer can recover it across restarts.
+            if let Some(backfiller) = &self.backfiller {
+                backfiller.record(&block).await;
+            }
+
+            // Acknowledge the block.
             info!(height = %block.height(), "finalized block");
             ack_rx.acknowledge();
         }

chain/src/application.rs

@@ -1,6 +1,6 @@
 use crate::{
     application::Application,
-    indexer::{self, Indexer},
+    indexer::{self, Client},
 };
 use alto_types::{Activity, Block, Finalization, Scheme, EPOCH, EPOCH_LENGTH, NAMESPACE};
 use commonware_broadcast::buffered;
@@ -28,7 +28,7 @@ use commonware_runtime::{
     buffer::paged::CacheRef, spawn_cell, BufferPooler, Clock, ContextCell, Handle, Metrics,
     Spawner, Storage, ThreadPooler,
 };
-use commonware_storage::archive::immutable;
+use commonware_storage::{archive::immutable, queue};
 use commonware_utils::channel::mpsc;
 use commonware_utils::{ordered::Set, NZU16};
 use commonware_utils::{NZUsize, NZU64};
@@ -37,19 +37,20 @@ use governor::clock::Clock as GClock;
 use governor::Quota;
 use rand::{CryptoRng, Rng};
 use std::{
-    num::NonZero,
+    num::{NonZero, NonZeroUsize},
     time::{Duration, Instant},
 };
 use tracing::{error, info, warn};
 
 /// Reporter type for [simplex::Engine].
-type Reporter<E, I> =
-    Reporters<Activity, MarshalMailbox<Scheme, Standard<Block>>, Option<indexer::Pusher<E, I>>>;
+type Reporter<E, C> =
+    Reporters<Activity, MarshalMailbox<Scheme, Standard<Block>>, Option<indexer::Pusher<E, C>>>;
 
 /// To better support peers near tip during network instability, we multiply
 /// the consensus activity timeout by this factor.
 const SYNCER_ACTIVITY_TIMEOUT_MULTIPLIER: u64 = 10;
 const PRUNABLE_ITEMS_PER_SECTION: NonZero<u64> = NZU64!(4_096);
+const QUEUE_ITEMS_PER_SECTION: NonZero<u64> = NZU64!(128);
 const IMMUTABLE_ITEMS_PER_SECTION: NonZero<u64> = NZU64!(262_144);
 const FREEZER_TABLE_RESIZE_FREQUENCY: u8 = 4;
 const FREEZER_TABLE_RESIZE_CHUNK_SIZE: u32 = 2u32.pow(16); // 3MB
@@ -66,7 +67,7 @@ const MAX_PENDING_ACKS: NonZero<usize> = NZUsize!(16);
 pub struct Config<
     B: Blocker<PublicKey = PublicKey>,
     P: Provider<PublicKey = PublicKey>,
-    I: Indexer,
+    C: Client,
     S: Strategy,
 > {
     pub blocker: B,
@@ -91,23 +92,25 @@ pub struct Config<
     pub max_fetch_size: usize,
     pub fetch_concurrent: usize,
     pub fetch_rate_per_peer: Quota,
+    pub backfiller_max_active: NonZeroUsize,
+    pub backfiller_retry: Duration,
 
     pub strategy: S,
 
-    pub indexer: Option<I>,
+    pub indexer: Option<C>,
 }
 
-type Marshaled<E> = Deferred<E, Scheme, Application, Block, FixedEpocher>;
+type Marshaled<E> = Deferred<E, Scheme, Application<E>, Block, FixedEpocher>;
 
 /// The engine that drives the [Application].
 #[allow(clippy::type_complexity)]
-pub struct Engine<E, B, P, S, I>
+pub struct Engine<E, B, P, S, C>
 where
     E: BufferPooler + Clock + GClock + Rng + CryptoRng + Spawner + Storage + Metrics,
     B: Blocker<PublicKey = PublicKey>,
     P: Provider<PublicKey = PublicKey>,
     S: Strategy,
-    I: Indexer,
+    C: Client,
 {
     context: ContextCell<E>,
 
@@ -125,19 +128,21 @@ where
     marshaled: Marshaled<E>,
 
     consensus:
-        Consensus<E, Scheme, Random, B, Digest, Marshaled<E>, Marshaled<E>, Reporter<E, I>, S>,
+        Consensus<E, Scheme, Random, B, Digest, Marshaled<E>, Marshaled<E>, Reporter<E, C>, S>,
+
+    consumer: Option<indexer::Consumer<E, C>>,
 }
 
-impl<E, B, P, S, I> Engine<E, B, P, S, I>
+impl<E, B, P, S, C> Engine<E, B, P, S, C>
 where
     E: BufferPooler + Clock + GClock + Rng + CryptoRng + Spawner + ThreadPooler + Storage + Metrics,
     B: Blocker<PublicKey = PublicKey>,
     P: Provider<PublicKey = PublicKey>,
     S: Strategy,
-    I: Indexer,
+    C: Client,
 {
     /// Create a new [Engine].
-    pub async fn new(context: E, cfg: Config<B, P, I, S>) -> Self {
+    pub async fn new(context: E, cfg: Config<B, P, C, S>) -> Self {
         // Create the buffer
         let (buffer, buffer_mailbox) = buffered::Engine::new(
             context.with_label("buffer"),
@@ -266,27 +271,49 @@ where
         )
         .await;
 
+        // Create the reporter and, when an indexer is configured, a backfill
+        // queue of finalized digests so block uploads can resume after
+        // restarts.
+        let (app, pusher, consumer) = if let Some(indexer) = cfg.indexer {
+            let queue = queue::shared::init(
+                context.with_label("queue"),
+                queue::Config {
+                    partition: format!("{}-finalized-queue", cfg.partition_prefix),
+                    items_per_section: QUEUE_ITEMS_PER_SECTION,
+                    compression: None,
+                    codec_config: (),
+                    page_cache: page_cache.clone(),
+                    write_buffer: WRITE_BUFFER,
+                },
+            )
+            .await
+            .expect("failed to initialize finalized queue");
+            let indexer = indexer::Indexer::new(
+                context.with_label("indexer"),
+                indexer,
+                marshal_mailbox.clone(),
+                queue,
+                cfg.backfiller_max_active,
+                cfg.backfiller_retry,
+            )
+            .await;
+            let (producer, pusher, consumer) = indexer.split();
+            let app = Application::new().with_backfiller(producer);
+            (app, Some(pusher), Some(consumer))
+        } else {
+            (Application::new(), None, None)
+        };
+
         // Create the application
-        let app = Application::new();
         let marshaled = Marshaled::new(
             context.with_label("marshaled"),
             app,
             marshal_mailbox.clone(),
             epocher,
         );
 
-        // Create the reporter
-        let reporter = (
-            marshal_mailbox.clone(),
-            cfg.indexer.map(|indexer| {
-                indexer::Pusher::new(
-                    context.with_label("indexer"),
-                    indexer,
-                    marshal_mailbox.clone(),
-                )
-            }),
-        )
-            .into();
+        // Create the reporter.
+        let reporter = (marshal_mailbox.clone(), pusher).into();
 
         // Create the consensus engine
         let consensus = Consensus::new(
@@ -324,6 +351,8 @@ where
             marshal,
             marshaled,
             consensus,
+
+            consumer,
         }
     }
 
@@ -391,14 +420,22 @@ where
             .marshal
             .start(self.marshaled, self.buffer_mailbox, marshal);
 
+        // Start draining queued block uploads before consensus so recovered work
+        // resumes immediately on startup.
+        let consumer_handle = self.consumer.map(indexer::Consumer::start);
+
         // Start consensus
         //
         // We start the application prior to consensus to ensure we can handle enqueued events from consensus (otherwise
         // restart could block).
         let consensus_handle = self.consensus.start(pending, recovered, resolver);
 
         // Wait for any actor to finish
-        if let Err(e) = try_join_all(vec![buffer_handle, marshal_handle, consensus_handle]).await {
+        let mut handles: Vec<Handle<()>> = vec![buffer_handle, marshal_handle, consensus_handle];
+        if let Some(h) = consumer_handle {
+            handles.push(h);
+        }
+        if let Err(e) = try_join_all(handles).await {
             error!(?e, "engine failed");
         } else {
             warn!("engine stopped");