Integrate client into chain and add inspector

.github/workflows/publish.yml

@@ -32,3 +32,8 @@ jobs:
       continue-on-error: true
       env:
         CARGO_REGISTRY_TOKEN: ${{ secrets.CARGO_REGISTRY_TOKEN }}
+    - name: Publish inspector
+      run: cargo publish --manifest-path inspector/Cargo.toml
+      continue-on-error: true
+      env:
+        CARGO_REGISTRY_TOKEN: ${{ secrets.CARGO_REGISTRY_TOKEN }}

Cargo.toml

@@ -2,6 +2,7 @@
 members = [
     "chain",
     "client",
+    "inspector",
     "types",
 ]
 resolver = "2"
@@ -30,6 +31,7 @@ tracing = "0.1.41"
 tracing-subscriber = "0.3.19"
 governor = "0.6.3"
 prometheus-client = "0.22.3"
+clap = "4.5.18"
 
 [profile.bench]
 # Because we enable overflow checks in "release," we should benchmark with them.

README.md

@@ -8,6 +8,7 @@
 
 * [chain](./chain/README.md): A minimal blockchain built with the [Commonware Library](https://github.com/commonwarexyz/monorepo).
 * [client](./client/README.md): Client for interacting with `alto`.
+* [inspector](./inspector/README.md): Monitor `alto` activity.
 * [types](./types/README.md): Common types used throughout `alto`.
 
 ## Licensing

chain/Cargo.toml

@@ -12,6 +12,7 @@ documentation = "https://docs.rs/alto-chain"
 
 [dependencies]
 alto-types = { workspace = true }
+alto-client = { workspace = true }
 commonware-consensus = { workspace = true }
 commonware-cryptography = { workspace = true }
 commonware-deployer = { workspace = true }
@@ -30,8 +31,8 @@ tracing = { workspace = true }
 tracing-subscriber = { workspace = true, features = ["fmt", "json"] }
 governor = { workspace = true }
 prometheus-client = { workspace = true }
+clap = { workspace = true }
 sysinfo = "0.33.1"
-clap = "4.5.18"
 axum = "0.8.1"
 uuid = "1.15.1"
 serde = { version = "1.0.218", features = ["derive"] }

chain/README.md

@@ -25,6 +25,12 @@ cargo install commonware-deployer
 cargo run --bin setup -- --peers 10 --bootstrappers 2 --regions us-west-1,us-east-1,eu-west-1,ap-northeast-1,eu-north-1,ap-south-1,sa-east-1,eu-central-1,ap-northeast-2,ap-southeast-2 --instance-type c7g.xlarge --storage-size 10 --storage-class gp3 --worker-threads 4 --message-backlog 16384 --mailbox-size 16384 --dashboard dashboard.json --output assets
 ```
 
+#### [Optional] Upload Consensus Artifacts to Indexer
+
+```bash
+--indexer <indexer URL>
+```
+
 ### Build Validator Binary
 
 #### Build Cross-Platform Compiler

chain/src/actors/application/actor.rs

@@ -4,7 +4,7 @@ use super::{
     Config,
 };
 use crate::actors::syncer;
-use alto_types::{Block, Finalization, Notarization};
+use alto_types::{Block, Finalization, Notarization, Seed};
 use commonware_consensus::threshold_simplex::Prover;
 use commonware_cryptography::{sha256::Digest, Hasher, Sha256};
 use commonware_macros::select;
@@ -215,21 +215,23 @@ impl<R: Rng + Spawner + Metrics + Clock> Actor<R> {
                 }
                 Message::Prepared { proof, payload } => {
                     // Parse the proof
-                    let (view, parent, _, signature, _) =
+                    let (view, parent, _, signature, seed) =
                         self.prover.deserialize_notarization(proof).unwrap();
                     let notarization = Notarization::new(view, parent, payload, signature.into());
+                    let seed = Seed::new(view, seed.into());
 
                     // Send the notarization to the syncer
-                    syncer.notarized(notarization).await;
+                    syncer.notarized(notarization, seed).await;
                 }
                 Message::Finalized { proof, payload } => {
                     // Parse the proof
-                    let (view, parent, _, signature, _) =
+                    let (view, parent, _, signature, seed) =
                         self.prover.deserialize_finalization(proof.clone()).unwrap();
                     let finalization = Finalization::new(view, parent, payload, signature.into());
+                    let seed = Seed::new(view, seed.into());
 
                     // Send the finalization to the syncer
-                    syncer.finalized(finalization).await;
+                    syncer.finalized(finalization, seed).await;
                 }
             }
         }

chain/src/actors/syncer/actor.rs

@@ -10,6 +10,7 @@ use crate::actors::syncer::{
     handler,
     key::{self, MultiIndex, Value},
 };
+use alto_client::Client;
 use alto_types::{Block, Finalization, Finalized, Notarized};
 use commonware_cryptography::{bls12381, ed25519::PublicKey, sha256::Digest};
 use commonware_macros::select;
@@ -51,6 +52,7 @@ pub struct Actor<B: Blob, R: Rng + Spawner + Metrics + Clock + GClock + Storage<
     mailbox_size: usize,
     backfill_quota: Quota,
     activity_timeout: u64,
+    client: Option<Client>,
 
     // Blocks verified stored by view<>digest
     verified: Archive<TwoCap, Digest, B, R>,
@@ -194,6 +196,13 @@ impl<B: Blob, R: Rng + Spawner + Metrics + Clock + GClock + Storage<B>> Actor<B,
 
         // Initialize mailbox
         let (sender, mailbox) = mpsc::channel(config.mailbox_size);
+
+        // Initialize client
+        let mut client = None;
+        if let Some(indexer) = config.indexer {
+            client = Some(Client::new(&indexer, config.identity.into()));
+            info!(indexer, "initialized indexer client");
+        }
         (
             Self {
                 context,
@@ -204,6 +213,7 @@ impl<B: Blob, R: Rng + Spawner + Metrics + Clock + GClock + Storage<B>> Actor<B,
                 mailbox_size: config.mailbox_size,
                 backfill_quota: config.backfill_quota,
                 activity_timeout: config.activity_timeout,
+                client,
 
                 verified: verified_archive,
                 notarized: notarized_archive,
@@ -293,8 +303,6 @@ impl<B: Blob, R: Rng + Spawner + Metrics + Clock + GClock + Storage<B>> Actor<B,
                     0
                 };
 
-                // TODO: need to handle finalization from finalization gaps (may never exist for a height where we finalize a notarize in a later view)
-
                 // Index all finalized blocks
                 //
                 // If using state sync, this is not necessary.
@@ -470,7 +478,21 @@ impl<B: Blob, R: Rng + Spawner + Metrics + Clock + GClock + Storage<B>> Actor<B,
                                 .await
                                 .expect("Failed to insert verified block");
                         }
-                        Message::Notarized { proof } => {
+                        Message::Notarized { proof, seed } => {
+                            // Upload seed to indexer (if available)
+                            if let Some(client) = self.client.as_ref() {
+                                let client = client.clone();
+                                let view = proof.view;
+                                self.context.with_label("indexer").spawn(
+                                    move |_| async move {
+                                        let result = client.seed_upload(seed).await;
+                                        if let Err(e) = result {
+                                            warn!(?e, "failed to upload seed");
+                                        }
+                                        debug!(view, "seed uploaded to indexer");
+                                    });
+                            }
+
                             // Check if in buffer
                             let mut block = None;
                             if let Some(buffered) = buffer.get(&proof.payload) {
@@ -489,12 +511,27 @@ impl<B: Blob, R: Rng + Spawner + Metrics + Clock + GClock + Storage<B>> Actor<B,
                                 let view = proof.view;
                                 let height = block.height;
                                 let digest = proof.payload.clone();
-                                let notarization = Notarized::new(proof, block);
+                                let notarization = Notarized::new(proof.clone(), block.clone());
                                 notarized
                                     .put(view, digest, notarization.serialize().into())
                                     .await
                                     .expect("Failed to insert notarized block");
                                 debug!(view, height, "notarized block stored");
+
+                                // Upload to indexer (if available)
+                                if let Some(client) = self.client.as_ref() {
+                                    let client = client.clone();
+                                    self.context.with_label("indexer").spawn(
+                                        move |_| async move {
+                                            let result = client
+                                                .notarization_upload(proof, block)
+                                                .await;
+                                            if let Err(e) = result {
+                                                warn!(?e, "failed to upload notarization");
+                                            }
+                                            debug!(view, "notarization uploaded to indexer");
+                                    });
+                                }
                                 continue;
                             }
 
@@ -506,7 +543,21 @@ impl<B: Blob, R: Rng + Spawner + Metrics + Clock + GClock + Storage<B>> Actor<B,
                             outstanding_notarize.insert(proof.view);
                             resolver.fetch(MultiIndex::new(Value::Notarized(proof.view))).await;
                         }
-                        Message::Finalized { proof } => {
+                        Message::Finalized { proof, seed } => {
+                            // Upload seed to indexer (if available)
+                            if let Some(client) = self.client.as_ref() {
+                                let client = client.clone();
+                                let view = proof.view;
+                                self.context.with_label("indexer").spawn(
+                                    move |_| async move {
+                                        let result = client.seed_upload(seed).await;
+                                        if let Err(e) = result {
+                                            warn!(?e, "failed to upload seed");
+                                        }
+                                        debug!(view, "seed uploaded to indexer");
+                                    });
+                            }
+
                             // Check if in buffer
                             let mut block = None;
                             if let Some(buffered) = buffer.get(&proof.payload){
@@ -562,6 +613,21 @@ impl<B: Blob, R: Rng + Spawner + Metrics + Clock + GClock + Storage<B>> Actor<B,
 
                                 // Update metrics
                                 self.latest_height.set(height as i64);
+
+                                // Upload to indexer (if available)
+                                if let Some(client) = self.client.as_ref() {
+                                    let client = client.clone();
+                                    self.context.with_label("indexer").spawn(
+                                        move |_| async move {
+                                            let result = client
+                                                .finalization_upload(proof, block)
+                                                .await;
+                                            if let Err(e) = result {
+                                                warn!(?e, "failed to upload finalization");
+                                            }
+                                            debug!(height, "finalization uploaded to indexer");
+                                    });
+                                }
                                 continue;
                             }
 

chain/src/actors/syncer/ingress.rs

@@ -1,4 +1,4 @@
-use alto_types::{Block, Finalization, Notarization};
+use alto_types::{Block, Finalization, Notarization, Seed};
 use commonware_cryptography::sha256::Digest;
 use futures::{
     channel::{mpsc, oneshot},
@@ -21,9 +21,11 @@ pub enum Message {
     },
     Notarized {
         proof: Notarization,
+        seed: Seed,
     },
     Finalized {
         proof: Finalization,
+        seed: Seed,
     },
 }
 
@@ -68,16 +70,16 @@ impl Mailbox {
             .expect("Failed to send lock");
     }
 
-    pub async fn notarized(&mut self, proof: Notarization) {
+    pub async fn notarized(&mut self, proof: Notarization, seed: Seed) {
         self.sender
-            .send(Message::Notarized { proof })
+            .send(Message::Notarized { proof, seed })
             .await
             .expect("Failed to send lock");
     }
 
-    pub async fn finalized(&mut self, proof: Finalization) {
+    pub async fn finalized(&mut self, proof: Finalization, seed: Seed) {
         self.sender
-            .send(Message::Finalized { proof })
+            .send(Message::Finalized { proof, seed })
             .await
             .expect("Failed to send lock");
     }

chain/src/actors/syncer/mod.rs

@@ -29,4 +29,6 @@ pub struct Config {
     pub backfill_quota: Quota,
 
     pub activity_timeout: u64,
+
+    pub indexer: Option<String>,
 }

chain/src/bin/setup.rs

@@ -87,6 +87,12 @@ fn main() {
                 .required(true)
                 .value_parser(value_parser!(String)),
         )
+        .arg(
+            Arg::new("indexer")
+                .long("indexer")
+                .required(false)
+                .value_parser(value_parser!(String)),
+        )
         .get_matches();
 
     // Create logger
@@ -124,7 +130,7 @@ fn main() {
     let (identity, shares) = ops::generate_shares(&mut OsRng, None, peers_u32, threshold);
     info!(
         identity = hex(&poly::public(&identity).serialize()),
-        "generated consensus key"
+        "generated network key"
     );
 
     // Generate instance configurations
@@ -143,6 +149,7 @@ fn main() {
     let worker_threads = *matches.get_one::<usize>("worker-threads").unwrap();
     let message_backlog = *matches.get_one::<usize>("message-backlog").unwrap();
     let mailbox_size = *matches.get_one::<usize>("mailbox-size").unwrap();
+    let indexer = matches.get_one::<String>("indexer").cloned();
     let mut instance_configs = Vec::new();
     let mut peer_configs = Vec::new();
     for (index, scheme) in peer_schemes.iter().enumerate() {
@@ -163,6 +170,8 @@ fn main() {
 
             message_backlog,
             mailbox_size,
+
+            indexer: indexer.clone(),
         };
         peer_configs.push((peer_config_file.clone(), peer_config));
 

chain/src/bin/validator.rs

@@ -195,6 +195,7 @@ fn main() {
             max_fetch_size: MAX_FETCH_SIZE,
             fetch_concurrent: FETCH_CONCURRENT,
             fetch_rate_per_peer: resolver_limit,
+            indexer: config.indexer,
         };
         let engine = engine::Engine::new(context.with_label("engine"), config).await;
 

chain/src/engine.rs

@@ -35,6 +35,8 @@ pub struct Config {
     pub max_fetch_size: usize,
     pub fetch_concurrent: usize,
     pub fetch_rate_per_peer: Quota,
+
+    pub indexer: Option<String>,
 }
 
 pub struct Engine<B: Blob, E: Clock + GClock + Rng + CryptoRng + Spawner + Storage<B> + Metrics> {
@@ -81,6 +83,7 @@ impl<B: Blob, E: Clock + GClock + Rng + CryptoRng + Spawner + Storage<B> + Metri
                 mailbox_size: cfg.mailbox_size,
                 backfill_quota: cfg.backfill_quota,
                 activity_timeout: cfg.activity_timeout,
+                indexer: cfg.indexer,
             },
         )
         .await;