feat: integrate upstream cache

Author: SupernaviX

common/src/lib.rs

@@ -21,6 +21,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 renamed to 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,25 +19,25 @@ pub trait Storage {
 }
 
 pub struct FileStorage {
-    path: String,
+    path: PathBuf,
 }
 
 impl FileStorage {
-    pub fn new(path: &str) -> Self {
+    pub fn new<P: AsRef<Path>>(path: P) -> Self {
         Self {
-            path: path.to_string(),
+            path: path.as_ref().to_path_buf(),
         }
     }
 
-    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 {
+    pub fn new<P: AsRef<Path>>(path: P) -> Self {
         UpstreamCache::new_impl(FileStorage::new(path))
     }
 }
@@ -139,34 +139,32 @@ 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())?;
+        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/peer_network_interface/src/configuration.rs

@@ -22,7 +22,6 @@ pub struct InterfaceConfig {
     pub genesis_completion_topic: String,
     pub node_addresses: Vec<String>,
     pub magic_number: u64,
-    #[expect(unused)]
     pub cache_dir: PathBuf,
     #[serde(flatten)]
     pub genesis_values: Option<GenesisValues>,

modules/peer_network_interface/src/connection.rs

@@ -201,22 +201,11 @@ impl PeerConnectionWorker {
         let hdr_tag = header.byron_prefix.map(|p| p.0);
         let hdr_variant = header.variant;
         let hdr = MultiEraHeader::decode(hdr_variant, hdr_tag, &header.cbor)?;
-        let era = match hdr {
-            MultiEraHeader::EpochBoundary(_) => return Ok(None),
-            MultiEraHeader::Byron(_) => Era::Byron,
-            MultiEraHeader::ShelleyCompatible(_) => match hdr_variant {
-                1 => Era::Shelley,
-                2 => Era::Allegra,
-                3 => Era::Mary,
-                4 => Era::Alonzo,
-                x => bail!("Impossible header variant {x} for ShelleyCompatible (TPraos)"),
-            },
-            MultiEraHeader::BabbageCompatible(_) => match hdr_variant {
-                5 => Era::Babbage,
-                6 => Era::Conway,
-                x => bail!("Impossible header variant {x} for BabbageCompatible (Praos)"),
-            },
-        };
+        if hdr.as_eb().is_some() {
+            // skip EpochBoundary blocks
+            return Ok(None);
+        }
+        let era = Era::try_from(hdr_variant)?;
         Ok(Some(Header {
             hash: BlockHash::new(*hdr.hash()),
             slot: hdr.slot(),

modules/peer_network_interface/src/peer_network_interface.rs

@@ -6,6 +6,7 @@ use acropolis_common::{
     BlockInfo, BlockStatus,
     genesis_values::GenesisValues,
     messages::{CardanoMessage, Message, RawBlockMessage},
+    upstream_cache::{UpstreamCache, UpstreamCacheRecord},
 };
 use anyhow::{Result, bail};
 use caryatid_sdk::{Context, Module, Subscription, module};
@@ -50,10 +51,28 @@ impl PeerNetworkInterface {
             } else {
                 cfg.genesis_values.expect("genesis values not found")
             };
+
+            let mut upstream_cache = None;
+            let mut cache_sync_point = None;
+            if cfg.sync_point == SyncPoint::Cache {
+                let mut cache = UpstreamCache::new(&cfg.cache_dir);
+                cache.start_reading()?;
+                while let Some(record) = cache.read_record()? {
+                    cache_sync_point = Some((record.id.slot, record.id.hash));
+                    let message = Arc::new(Message::Cardano((
+                        record.id,
+                        CardanoMessage::BlockAvailable(Arc::unwrap_or_clone(record.message)),
+                    )));
+                    context.message_bus.publish(&cfg.block_topic, message).await?;
+                }
+                upstream_cache = Some(cache);
+            }
+
             let sink = BlockSink {
                 context,
                 topic: cfg.block_topic,
                 genesis_values,
+                upstream_cache,
             };
 
             let mut manager =
@@ -65,7 +84,13 @@ impl PeerNetworkInterface {
             match cfg.sync_point {
                 SyncPoint::Origin => manager.sync_to_point(Point::Origin),
                 SyncPoint::Tip => manager.sync_to_tip().await?,
-                SyncPoint::Cache => unimplemented!(),
+                SyncPoint::Cache => {
+                    let point = match cache_sync_point {
+                        Some((slot, hash)) => Point::Specific(slot, hash.to_vec()),
+                        None => Point::Origin,
+                    };
+                    manager.sync_to_point(point);
+                }
                 SyncPoint::Snapshot => {
                     let mut subscription =
                         snapshot_complete.expect("Snapshot topic subscription missing");
@@ -109,15 +134,31 @@ struct BlockSink {
     context: Arc<Context<Message>>,
     topic: String,
     genesis_values: GenesisValues,
+    upstream_cache: Option<UpstreamCache>,
 }
 impl BlockSink {
-    pub async fn announce(&self, header: &Header, body: &[u8], rolled_back: bool) -> Result<()> {
+    pub async fn announce(
+        &mut self,
+        header: &Header,
+        body: &[u8],
+        rolled_back: bool,
+    ) -> Result<()> {
         let info = self.make_block_info(header, rolled_back);
-        let available = CardanoMessage::BlockAvailable(RawBlockMessage {
+        let raw_block = RawBlockMessage {
             header: header.bytes.clone(),
             body: body.to_vec(),
-        });
-        let message = Arc::new(Message::Cardano((info, available)));
+        };
+        if let Some(cache) = self.upstream_cache.as_mut() {
+            let record = UpstreamCacheRecord {
+                id: info.clone(),
+                message: Arc::new(raw_block.clone()),
+            };
+            cache.write_record(&record)?;
+        }
+        let message = Arc::new(Message::Cardano((
+            info,
+            CardanoMessage::BlockAvailable(raw_block),
+        )));
         self.context.publish(&self.topic, message).await
     }
 

modules/upstream_chain_fetcher/src/body_fetcher.rs

@@ -1,7 +1,11 @@
 //! Acropolis Miniprotocols module for Caryatid
 //! Multi-connection, block body fetching part of the client (in separate thread).
 
-use acropolis_common::{messages::RawBlockMessage, BlockHash, BlockInfo, BlockStatus, Era};
+use acropolis_common::{
+    messages::RawBlockMessage,
+    upstream_cache::{UpstreamCache, UpstreamCacheRecord},
+    BlockHash, BlockInfo, BlockStatus, Era,
+};
 use anyhow::{bail, Result};
 use crossbeam::channel::{Receiver, TryRecvError};
 use pallas::{
@@ -15,7 +19,6 @@ use std::{sync::Arc, time::Duration};
 use tokio::{sync::Mutex, time::sleep};
 use tracing::{debug, error, info};
 
-use crate::upstream_cache::{UpstreamCache, UpstreamCacheRecord};
 use crate::{
     utils,
     utils::{

modules/upstream_chain_fetcher/src/upstream_chain_fetcher.rs

@@ -4,6 +4,7 @@
 use acropolis_common::{
     genesis_values::GenesisValues,
     messages::{CardanoMessage, Message},
+    upstream_cache::{UpstreamCache, UpstreamCacheRecord},
     BlockInfo,
 };
 use anyhow::{anyhow, bail, Result};
@@ -24,12 +25,10 @@ use tokio::{sync::Mutex, time::sleep};
 use tracing::{debug, error, info};
 
 mod body_fetcher;
-mod upstream_cache;
 mod utils;
 
 use crate::utils::FetchResult;
 use body_fetcher::BodyFetcher;
-use upstream_cache::{UpstreamCache, UpstreamCacheRecord};
 use utils::{FetcherConfig, SyncPoint};
 
 const MAX_BODY_FETCHER_CHANNEL_LENGTH: usize = 100;