Implement PeerNetworkInterface

Cargo.toml

@@ -11,6 +11,7 @@ members = [
     "modules/mithril_snapshot_fetcher",     # Mithril snapshot fetcher
     "modules/snapshot_bootstrapper",        # Bootstrap state from a ledger snapshot
     "modules/upstream_chain_fetcher",       # Upstream chain fetcher
+    "modules/peer_network_interface",       # Multi-peer network interface
     "modules/block_unpacker",               # Block to transaction unpacker
     "modules/tx_unpacker",                  # Tx to UTXO unpacker
     "modules/utxo_state",                   # UTXO state

common/src/genesis_values.rs

@@ -6,6 +6,7 @@ const MAINNET_SHELLEY_GENESIS_HASH: &str =
     "1a3be38bcbb7911969283716ad7aa550250226b76a61fc51cc9a9a35d9276d81";
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+#[serde(rename_all = "kebab-case")]
 pub struct GenesisValues {
     pub byron_timestamp: u64,
     pub shelley_epoch: u64,

common/src/lib.rs

@@ -22,6 +22,7 @@ pub mod snapshot;
 pub mod stake_addresses;
 pub mod state_history;
 pub mod types;
+pub mod upstream_cache;
 pub mod validation;
 
 // Flattened re-exports

modules/upstream_chain_fetcher/src/upstream_cache.rs → common/src/upstream_cache.rs

@@ -1,9 +1,9 @@
-use acropolis_common::{messages::RawBlockMessage, BlockInfo};
-use anyhow::{anyhow, bail, Result};
+use crate::{messages::RawBlockMessage, BlockInfo};
+use anyhow::{anyhow, bail, Context, Result};
 use std::{
     fs::File,
     io::{BufReader, Write},
-    path::Path,
+    path::{Path, PathBuf},
     sync::Arc,
 };
 
@@ -19,26 +19,26 @@ pub trait Storage {
 }
 
 pub struct FileStorage {
-    path: String,
+    path: PathBuf,
 }
 
 impl FileStorage {
-    pub fn new(path: &str) -> Self {
-        Self {
-            path: path.to_string(),
-        }
+    pub fn new<P: AsRef<Path>>(path: P) -> Result<Self> {
+        let path = path.as_ref().to_path_buf();
+        std::fs::create_dir_all(&path)?;
+        Ok(Self { path })
     }
 
-    fn get_file_name(&self, chunk_no: usize) -> String {
-        format!("{}/chunk-{chunk_no}.json", self.path)
+    fn get_file_name(&self, chunk_no: usize) -> PathBuf {
+        self.path.join(format!("chunk-{chunk_no}.json"))
     }
 }
 
 pub type UpstreamCache = UpstreamCacheImpl<FileStorage>;
 
 impl UpstreamCache {
-    pub fn new(path: &str) -> Self {
-        UpstreamCache::new_impl(FileStorage::new(path))
+    pub fn new<P: AsRef<Path>>(path: P) -> Result<Self> {
+        Ok(UpstreamCache::new_impl(FileStorage::new(path)?))
     }
 }
 
@@ -124,7 +124,9 @@ impl<S: Storage> UpstreamCacheImpl<S> {
 
     pub fn write_record(&mut self, record: &UpstreamCacheRecord) -> Result<()> {
         self.chunk_cached.push(record.clone());
-        self.storage.write_chunk(self.current_chunk, &self.chunk_cached)?;
+        self.storage
+            .write_chunk(self.current_chunk, &self.chunk_cached)
+            .context("could not write cache record")?;
 
         self.current_record += 1;
         if self.current_record >= self.density {
@@ -139,34 +141,33 @@ impl<S: Storage> UpstreamCacheImpl<S> {
 
 impl Storage for FileStorage {
     fn read_chunk(&mut self, chunk_no: usize) -> Result<Vec<UpstreamCacheRecord>> {
-        let name = self.get_file_name(chunk_no);
-        let path = Path::new(&name);
+        let path = self.get_file_name(chunk_no);
         if !path.try_exists()? {
             return Ok(vec![]);
         }
 
-        let file = File::open(&name)?;
+        let file = File::open(&path)?;
         let reader = BufReader::new(file);
-        match serde_json::from_reader::<BufReader<std::fs::File>, Vec<UpstreamCacheRecord>>(reader)
-        {
-            Ok(res) => Ok(res.clone()),
-            Err(err) => Err(anyhow!(
-                "Error reading upstream cache chunk JSON from {name}: '{err}'"
-            )),
-        }
+        serde_json::from_reader(reader).with_context(|| {
+            format!(
+                "Error reading upstream cache chunk JSON from {}",
+                path.display()
+            )
+        })
     }
 
     fn write_chunk(&mut self, chunk_no: usize, data: &[UpstreamCacheRecord]) -> Result<()> {
-        let mut file = File::create(self.get_file_name(chunk_no))?;
-        file.write_all(serde_json::to_string(data)?.as_bytes())?;
+        let mut file =
+            File::create(self.get_file_name(chunk_no)).context("could not write chunk")?;
+        file.write_all(&serde_json::to_vec(data)?)?;
         Ok(())
     }
 }
 
 #[cfg(test)]
 mod test {
     use crate::upstream_cache::{Storage, UpstreamCacheImpl, UpstreamCacheRecord};
-    use acropolis_common::{messages::RawBlockMessage, BlockHash, BlockInfo, BlockStatus, Era};
+    use crate::{messages::RawBlockMessage, BlockHash, BlockInfo, BlockStatus, Era};
     use anyhow::Result;
     use std::{collections::HashMap, sync::Arc};
 

modules/genesis_bootstrapper/build.rs

@@ -62,6 +62,16 @@ async fn main() -> Result<()> {
             "https://book.world.dev.cardano.org/environments/mainnet/shelley-genesis.json",
             "mainnet-shelley-genesis.json",
         ),
+        download(
+            &client,
+            "https://book.world.dev.cardano.org/environments/preview/byron-genesis.json",
+            "preview-byron-genesis.json",
+        ),
+        download(
+            &client,
+            "https://book.world.dev.cardano.org/environments/preview/shelley-genesis.json",
+            "preview-shelley-genesis.json",
+        ),
         download(
             &client,
             "https://raw.githubusercontent.com/Hornan7/SanchoNet-Tutorials/refs/heads/main/genesis/byron-genesis.json",

modules/genesis_bootstrapper/src/genesis_bootstrapper.rs

@@ -32,6 +32,9 @@ const DEFAULT_NETWORK_NAME: &str = "mainnet";
 const MAINNET_BYRON_GENESIS: &[u8] = include_bytes!("../downloads/mainnet-byron-genesis.json");
 const MAINNET_SHELLEY_GENESIS: &[u8] = include_bytes!("../downloads/mainnet-shelley-genesis.json");
 const MAINNET_SHELLEY_START_EPOCH: u64 = 208;
+const PREVIEW_BYRON_GENESIS: &[u8] = include_bytes!("../downloads/preview-byron-genesis.json");
+const PREVIEW_SHELLEY_GENESIS: &[u8] = include_bytes!("../downloads/preview-shelley-genesis.json");
+const PREVIEW_SHELLEY_START_EPOCH: u64 = 0;
 const SANCHONET_BYRON_GENESIS: &[u8] = include_bytes!("../downloads/sanchonet-byron-genesis.json");
 const SANCHONET_SHELLEY_GENESIS: &[u8] =
     include_bytes!("../downloads/sanchonet-shelley-genesis.json");
@@ -101,6 +104,11 @@ impl GenesisBootstrapper {
                             MAINNET_SHELLEY_GENESIS,
                             MAINNET_SHELLEY_START_EPOCH,
                         ),
+                        "preview" => (
+                            PREVIEW_BYRON_GENESIS,
+                            PREVIEW_SHELLEY_GENESIS,
+                            PREVIEW_SHELLEY_START_EPOCH,
+                        ),
                         "sanchonet" => (
                             SANCHONET_BYRON_GENESIS,
                             SANCHONET_SHELLEY_GENESIS,

modules/peer_network_interface/Cargo.toml

@@ -0,0 +1,24 @@
+# Acropolis upstream chain fetcher module
+
+[package]
+name = "acropolis_module_peer_network_interface"
+version = "0.2.0"
+edition = "2024"
+authors = ["Simon Gellis <simon@sundae.fi>"]
+description = "Multiplexed chain fetcher Caryatid module for Acropolis"
+license = "Apache-2.0"
+
+[dependencies]
+acropolis_common = { path = "../../common" }
+
+caryatid_sdk = { workspace = true }
+
+anyhow = { workspace = true }
+config = { workspace = true }
+pallas = { workspace = true }
+serde = { workspace = true, features = ["rc"] }
+tokio = { workspace = true }
+tracing = { workspace = true }
+
+[lib]
+path = "src/peer_network_interface.rs"

modules/peer_network_interface/NOTES.md

@@ -0,0 +1,18 @@
+# Architecture
+
+This module uses an event-queue-based architecture. A `NetworkManager` is responsible for creating a set of `PeerConnection`s and sending commands to them. Each `PeerConnection` maintains a connection to a single peer; it responds to commands from the `NetworkManager`, and emits events to an event queue. The `NetworkManager` reads from that queue to decide which chain to follow. When blocks from the preferred chain have been fetched, it publishes those blocks to the message bus.
+
+This module requests the body for every block announced by any chain, from the first chain which announced it. When it has the body for the next block announced, it will publish it to the message bus.
+
+```mermaid
+graph LR
+    EQ[Event Queue]-->NM[NetworkManager]
+    subgraph Peers
+        P1[PeerConnection 1]
+        P2[PeerConnection 2]
+        P3[PeerConnection 3]
+    end
+    NM -->|RequestBlock</br>FindIntersect| P1 & P2 & P3
+    Peers -->|ChainSync<br/>BlockFetched<br/>Disconnect|EQ
+    NM -->|BlockAvailable| MB[Message Bus]
+```

modules/peer_network_interface/README.md

@@ -0,0 +1,18 @@
+# Peer network interface module
+
+The peer network interface module uses the ChainSync and BlockFetch protocols to fetch blocks from one of several upstream sources. It chooses one peer to treat as the "preferred" chain to follow, but will gracefully switch which peer it follows during network issues.
+
+It can either run independently, from the origin or current tip, or
+be triggered by a Mithril snapshot event (the default) where it starts from
+where the snapshot left off, and follows the chain from there.
+
+Rollbacks are handled by signalling in the block data - it is downstream
+subscribers' responsibility to deal with the effects of this.
+
+## Configuration
+
+See [./config.default.toml](./config.default.toml) for the available configuration options and their default values.
+
+## Messages
+
+This module publishes "raw block messages" to the configured `block-topic`. Each message includes the raw bytes composing the header and body of a block. The module follows the head of one chain at any given time, though that chain may switch during runtime. If that chain reports a rollback (or if this module switches to a different chain), the next message it emits will be the new head of the chain and have the status `RolledBack`.

modules/peer_network_interface/config.default.toml

@@ -0,0 +1,24 @@
+# The topic to publish blocks on
+block-topic = "cardano.block.available"
+# The topic to wait for when sync-point is "snapshot"
+snapshot-completion-topic = "cardano.snapshot.complete"
+# The topic to wait for when listening for genesis values from another module
+genesis-completion-topic = "cardano.sequence.bootstrapped"
+
+# Upstream node connections
+node-addresses = [
+    "backbone.cardano.iog.io:3001",
+    "backbone.mainnet.cardanofoundation.org:3001",
+    "backbone.mainnet.emurgornd.com:3001",
+]
+# The network magic for the chain to connect to
+magic-number = 764824073
+
+# The initial point to start syncing from. Options:
+# - "origin": sync from the very start of the chain
+# - "tip": sync from the very end of the chain
+# - "cache": replay messages from a local filesystem cache, then sync from the point right after that cache.
+# - "snapshot": wait for another module to restore from a snapshot, then sync from the point right after that snapshot.
+sync-point = "snapshot"
+# The cache dir to use when sync-point is "cache"
+cache-dir = "upstream-cache"