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

@@ -22,9 +22,17 @@ cargo install commonware-deployer
 ### Create Deployer Artifacts
 
 ```bash
-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
+cargo run --bin setup -- generate --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] Configure Indexer Upload
+
+```bash
+cargo run --bin setup -- indexer --dir assets --url <indexer URL>
+```
+
+_The indexer URL is configured separately because it is typically only known after the threshold key is generated (derived in `setup generate`)._
+
 ### 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,7 +10,9 @@ use crate::actors::syncer::{
     handler,
     key::{self, MultiIndex, Value},
 };
+use alto_client::Client;
 use alto_types::{Block, Finalization, Finalized, Notarized};
+use bytes::Bytes;
 use commonware_cryptography::{bls12381, ed25519::PublicKey, sha256::Digest};
 use commonware_macros::select;
 use commonware_p2p::{utils::requester, Receiver, Recipients, Sender};
@@ -51,6 +53,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 +197,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 +214,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 +304,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 +479,22 @@ 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;
+                                let seed = seed.serialize().into();
+                                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) {
@@ -490,11 +514,27 @@ impl<B: Blob, R: Rng + Spawner + Metrics + Clock + GClock + Storage<B>> Actor<B,
                                 let height = block.height;
                                 let digest = proof.payload.clone();
                                 let notarization = Notarized::new(proof, block);
+                                let notarization: Bytes = notarization.serialize().into();
                                 notarized
-                                    .put(view, digest, notarization.serialize().into())
+                                    .put(view, digest, notarization.clone())
                                     .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(notarization)
+                                                .await;
+                                            if let Err(e) = result {
+                                                warn!(?e, "failed to upload notarization");
+                                            }
+                                            debug!(view, "notarization uploaded to indexer");
+                                    });
+                                }
                                 continue;
                             }
 
@@ -506,7 +546,22 @@ 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;
+                                let seed = seed.serialize().into();
+                                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 +617,22 @@ 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();
+                                    let finalization = Finalized::new(proof, block).serialize().into();
+                                    self.context.with_label("indexer").spawn(
+                                        move |_| async move {
+                                            let result = client
+                                                .finalization_upload(finalization)
+                                                .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>,
 }