Merge pull request #394 from input-output-hk/whankinsiv/peer-network-dynamic-sync

feat: Add dynamic sync mode to peer_network_interface

Author: SupernaviX

Cargo.lock

@@ -35,6 +35,7 @@ members = [
     "processes/replayer",           # All-inclusive process to replay messages
     "processes/golden_tests",       # All-inclusive golden tests process
     "processes/tx_submitter_cli",   # CLI wrapper for TX submitter
+    "processes/indexer",            # Minimal example indexer
 ]
 resolver = "2"
 

Cargo.toml

@@ -29,7 +29,6 @@ regex = "1"
 serde = { workspace = true, features = ["rc"] }
 serde_json = { workspace = true }
 serde_with = { workspace = true, features = ["base64"] }
-tempfile = "3"
 tokio = { workspace = true }
 tracing = { workspace = true }
 futures = "0.3.31"
@@ -40,8 +39,11 @@ rayon = "1.11.0"
 cryptoxide = "0.5.1"
 thiserror = "2.0.17"
 sha2 = "0.10.8"
+
+[dev-dependencies]
 caryatid_process = { workspace = true }
 config = { workspace = true }
+tempfile = "3"
 
 [lib]
 crate-type = ["rlib"]

common/Cargo.toml

@@ -0,0 +1,6 @@
+use crate::{BlockHash, Slot};
+
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+pub enum ChainSyncCommand {
+    FindIntersect { slot: Slot, hash: BlockHash },
+}

common/src/commands/chain_sync.rs

@@ -1 +1,2 @@
+pub mod chain_sync;
 pub mod transactions;

common/src/commands/mod.rs

@@ -3,6 +3,7 @@
 // We don't use these messages in the acropolis_common crate itself
 #![allow(dead_code)]
 
+use crate::commands::chain_sync::ChainSyncCommand;
 use crate::commands::transactions::{TransactionsCommand, TransactionsCommandResponse};
 use crate::genesis_values::GenesisValues;
 use crate::ledger_state::SPOState;
@@ -453,6 +454,7 @@ pub enum StateQueryResponse {
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub enum Command {
     Transactions(TransactionsCommand),
+    ChainSync(ChainSyncCommand),
 }
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]

common/src/messages.rs

@@ -0,0 +1,22 @@
+# Acropolis indexer module
+
+[package]
+name = "acropolis_module_indexer"
+version = "0.1.0"
+edition = "2021"
+authors = ["William Hankins <william@sundae.fi>"]
+description = "Core indexer logic"
+license = "Apache-2.0"
+
+[dependencies]
+acropolis_common = { path = "../../common" }
+
+caryatid_sdk = { workspace = true }
+
+anyhow = { workspace = true }
+config = { workspace = true }
+serde = { workspace = true, features = ["rc"] }
+tracing = { workspace = true }
+
+[lib]
+path = "src/indexer.rs"

modules/indexer/Cargo.toml

@@ -0,0 +1,2 @@
+# The topic to publish sync commands on
+sync-command-topic = "cardano.sync.command"

modules/indexer/config.default.toml

@@ -0,0 +1,21 @@
+use anyhow::Result;
+use config::Config;
+
+#[derive(serde::Deserialize)]
+#[serde(rename_all = "kebab-case")]
+pub struct IndexerConfig {
+    pub sync_command_topic: String,
+}
+
+impl IndexerConfig {
+    pub fn try_load(config: &Config) -> Result<Self> {
+        let full_config = Config::builder()
+            .add_source(config::File::from_str(
+                include_str!("../config.default.toml"),
+                config::FileFormat::Toml,
+            ))
+            .add_source(config.clone())
+            .build()?;
+        Ok(full_config.try_deserialize()?)
+    }
+}

modules/indexer/src/configuration.rs

@@ -0,0 +1,67 @@
+//! Acropolis indexer module for Caryatid
+mod configuration;
+
+use acropolis_common::{
+    commands::chain_sync::ChainSyncCommand,
+    hash::Hash,
+    messages::{Command, Message},
+};
+use anyhow::Result;
+use caryatid_sdk::{module, Context};
+use config::Config;
+use std::{str::FromStr, sync::Arc};
+use tracing::info;
+
+use crate::configuration::IndexerConfig;
+
+/// Indexer module
+#[module(
+    message_type(Message),
+    name = "indexer",
+    description = "Core indexer module for indexer process"
+)]
+pub struct Indexer;
+
+impl Indexer {
+    /// Async initialisation
+    pub async fn init(&self, context: Arc<Context<Message>>, config: Arc<Config>) -> Result<()> {
+        let cfg = IndexerConfig::try_load(&config)?;
+        info!(
+            "Creating sync command publisher on '{}'",
+            cfg.sync_command_topic
+        );
+
+        let ctx = context.clone();
+
+        // This is a placeholder to test dynamic sync
+        context.run(async move {
+            let example = ChainSyncCommand::FindIntersect {
+                slot: 4492799,
+                hash: Hash::from_str(
+                    "f8084c61b6a238acec985b59310b6ecec49c0ab8352249afd7268da5cff2a457",
+                )
+                .expect("Valid hash"),
+            };
+
+            // Initial sync message (This will be read from config for first sync and from DB on subsequent runs)
+            ctx.message_bus
+                .publish(
+                    &cfg.sync_command_topic,
+                    Arc::new(Message::Command(Command::ChainSync(example.clone()))),
+                )
+                .await
+                .unwrap();
+
+            // Simulate a later sync command to reset sync point to where we started
+
+            ctx.message_bus
+                .publish(
+                    &cfg.sync_command_topic,
+                    Arc::new(Message::Command(Command::ChainSync(example))),
+                )
+                .await
+                .unwrap();
+        });
+        Ok(())
+    }
+}