fix: handle cache errors more gracefully

SupernaviX · SupernaviX · commit 4647b72d52f8 · 2025-11-04T13:33:51.000-05:00
diff --git a/common/src/upstream_cache.rs b/common/src/upstream_cache.rs
@@ -23,10 +23,10 @@ pub struct FileStorage {
 }
 
 impl FileStorage {
-    pub fn new<P: AsRef<Path>>(path: P) -> Self {
-        Self {
-            path: path.as_ref().to_path_buf(),
-        }
+    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) -> PathBuf {
@@ -37,8 +37,8 @@ impl FileStorage {
 pub type UpstreamCache = UpstreamCacheImpl<FileStorage>;
 
 impl UpstreamCache {
-    pub fn new<P: AsRef<Path>>(path: P) -> 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,7 @@ 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 {
@@ -155,7 +155,7 @@ impl Storage for FileStorage {
     }
 
     fn write_chunk(&mut self, chunk_no: usize, data: &[UpstreamCacheRecord]) -> Result<()> {
-        let mut file = File::create(self.get_file_name(chunk_no))?;
+        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(())
     }
diff --git a/modules/peer_network_interface/config.default.toml b/modules/peer_network_interface/config.default.toml
@@ -4,6 +4,8 @@ genesis-completion-topic = "cardano.sequence.bootstrapped"
 
 node-addresses = [
     "backbone.cardano.iog.io:3001",
+    "backbone.mainnet.cardanofoundation.org:3001",
+    "backbone.mainnet.emurgornd.com:3001",
 ]
 magic-number = 764824073
 
diff --git a/modules/peer_network_interface/src/network.rs b/modules/peer_network_interface/src/network.rs
@@ -58,7 +58,7 @@ impl NetworkManager {
         while let Some(event) = self.events.recv().await {
             match event {
                 NetworkEvent::PeerUpdate { peer, event } => {
-                    let maybe_publish_blocks = self.handle_peer_update(peer, event)?;
+                    let maybe_publish_blocks = self.handle_peer_update(peer, event);
                     if maybe_publish_blocks {
                         self.publish_blocks().await?;
                     }
@@ -124,7 +124,7 @@ impl NetworkManager {
     // or when publishing messages to other modules. This avoids deadlock; if our event queue
     // is full and this method is blocked on writing to it, the queue can never drain.
     // Returns true if we might have new events to publish downstream.
-    fn handle_peer_update(&mut self, peer: PeerId, event: PeerEvent) -> Result<bool> {
+    fn handle_peer_update(&mut self, peer: PeerId, event: PeerEvent) -> bool {
         let is_preferred = self.preferred_upstream.is_some_and(|id| id == peer);
         match event {
             PeerEvent::ChainSync(PeerChainSyncEvent::RollForward(header)) => {
@@ -148,12 +148,12 @@ impl NetworkManager {
                                 break; // only fetch from one
                             }
                         }
-                        Ok(false)
+                        false
                     }
                     BlockStatus::Fetched(_) => {
                         // If chainsync has requested a block which we've already fetched,
                         // we might be able to publish one or more.
-                        Ok(is_preferred)
+                        is_preferred
                     }
                 }
             }
@@ -184,18 +184,18 @@ impl NetworkManager {
                         }
                     }
                 }
-                Ok(false)
+                false
             }
             PeerEvent::BlockFetched(fetched) => {
                 let Some(block) = self.blocks.get_mut(&fetched.hash) else {
-                    return Ok(false);
+                    return false;
                 };
                 block.set_body(&fetched.body);
-                Ok(true)
+                true
             }
             PeerEvent::Disconnected => {
                 self.handle_disconnect(peer);
-                Ok(false)
+                false
             }
         }
     }
diff --git a/modules/peer_network_interface/src/peer_network_interface.rs b/modules/peer_network_interface/src/peer_network_interface.rs
@@ -3,18 +3,16 @@ mod connection;
 mod network;
 
 use acropolis_common::{
-    BlockInfo, BlockStatus,
-    genesis_values::GenesisValues,
-    messages::{CardanoMessage, Message, RawBlockMessage},
-    upstream_cache::{UpstreamCache, UpstreamCacheRecord},
+    BlockInfo, BlockStatus, genesis_values::GenesisValues, messages::{CardanoMessage, Message, RawBlockMessage}, upstream_cache::{UpstreamCache, UpstreamCacheRecord}
 };
 use anyhow::{Result, bail};
 use caryatid_sdk::{Context, Module, Subscription, module};
 use config::Config;
 use pallas::network::miniprotocols::Point;
 use tokio::sync::mpsc;
+use tracing::{error, warn};
 
-use std::{sync::Arc, time::Duration};
+use std::{path::Path, sync::Arc, time::Duration};
 
 use crate::{
     configuration::{InterfaceConfig, SyncPoint},
@@ -53,19 +51,17 @@ impl PeerNetworkInterface {
             };
 
             let mut upstream_cache = None;
-            let mut cache_sync_point = None;
+            let mut cache_sync_point = Point::Origin;
             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?;
+                match Self::init_cache(&cfg.cache_dir, &cfg.block_topic, &context).await {
+                    Ok((cache, sync_point)) => {
+                        upstream_cache = Some(cache);
+                        cache_sync_point = sync_point;
+                    }
+                    Err(e) => {
+                        warn!("could not initialize upstream cache: {e:#}");
+                    }
                 }
-                upstream_cache = Some(cache);
             }
 
             let sink = BlockSink {
@@ -83,28 +79,57 @@ impl PeerNetworkInterface {
 
             match cfg.sync_point {
                 SyncPoint::Origin => manager.sync_to_point(Point::Origin),
-                SyncPoint::Tip => manager.sync_to_tip().await?,
-                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::Tip => {
+                    if let Err(error) = manager.sync_to_tip().await {
+                        warn!("could not sync to tip: {error:#}");
+                        return;
+                    }
                 }
+                SyncPoint::Cache => manager.sync_to_point(cache_sync_point),
                 SyncPoint::Snapshot => {
                     let mut subscription =
                         snapshot_complete.expect("Snapshot topic subscription missing");
-                    let point = Self::wait_snapshot_completion(&mut subscription).await?;
-                    manager.sync_to_point(point);
+                    match Self::wait_snapshot_completion(&mut subscription).await {
+                        Ok(point) => manager.sync_to_point(point),
+                        Err(error) => {
+                            warn!("snapshot restoration never completed: {error:#}");
+                            return;                            
+                        }
+                    }
                 }
             }
 
-            manager.run().await
+            if let Err(err) = manager.run().await {
+                error!("chain sync failed: {err:#}");
+            }
         });
 
         Ok(())
     }
 
+    async fn init_cache(
+        cache_dir: &Path,
+        block_topic: &str,
+        context: &Context<Message>,
+    ) -> Result<(UpstreamCache, Point)> {
+        let mut cache = UpstreamCache::new(cache_dir)?;
+        let mut cache_sync_point = None;
+        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(block_topic, message).await?;
+        }
+        let sync_point = match cache_sync_point {
+            None => Point::Origin,
+            Some((slot, hash)) => Point::Specific(slot, hash.to_vec()),
+        };
+        Ok((cache, sync_point))
+    }
+
     async fn wait_genesis_completion(
         subscription: &mut Box<dyn Subscription<Message>>,
     ) -> Result<GenesisValues> {
diff --git a/modules/upstream_chain_fetcher/src/upstream_chain_fetcher.rs b/modules/upstream_chain_fetcher/src/upstream_chain_fetcher.rs
@@ -227,7 +227,7 @@ impl UpstreamChainFetcher {
                 Self::sync_to_point(cfg, None, Point::Origin).await?;
             }
             SyncPoint::Cache => {
-                let mut upstream_cache = UpstreamCache::new(&cfg.cache_dir);
+                let mut upstream_cache = UpstreamCache::new(&cfg.cache_dir)?;
                 let point = match Self::read_cache(cfg.clone(), &mut upstream_cache).await? {
                     None => Point::Origin,
                     Some(blk) => Point::Specific(blk.slot, blk.hash.to_vec()),
diff --git a/processes/omnibus/.gitignore b/processes/omnibus/.gitignore
@@ -3,6 +3,7 @@ sled-immutable-utxos
 fjall-blocks
 fjall-immutable-utxos
 cache
+upstream-cache
 
 # DB files
 *_db
diff --git a/processes/omnibus/omnibus-local.toml b/processes/omnibus/omnibus-local.toml
@@ -1,7 +1,7 @@
 # Top-level configuration for Acropolis omnibus process
 
 [module.genesis-bootstrapper]
-network-name = "preview" # "sanchonet", "mainnet"
+network-name = "preview" # "sanchonet", "preview", "mainnet"
 
 #[module.mithril-snapshot-fetcher]
 #Turned off with SanchoNet

Original file line number	Diff line number	Diff line change
`@@ -23,10 +23,10 @@ pub struct FileStorage {`
`23`	`23`	`}`
`24`	`24`
`25`	`25`	`impl FileStorage {`
`26`		`- pub fn new<P: AsRef<Path>>(path: P) -> Self {`
`27`		`- Self {`
`28`		`- path: path.as_ref().to_path_buf(),`
`29`		`- }`
	`26`	`+ pub fn new<P: AsRef<Path>>(path: P) -> Result<Self> {`
	`27`	`+ let path = path.as_ref().to_path_buf();`
	`28`	`+ std::fs::create_dir_all(&path)?;`
	`29`	`+ Ok(Self { path })`
`30`	`30`	`}`
`31`	`31`
`32`	`32`	`fn get_file_name(&self, chunk_no: usize) -> PathBuf {`
`@@ -37,8 +37,8 @@ impl FileStorage {`
`37`	`37`	`pub type UpstreamCache = UpstreamCacheImpl<FileStorage>;`
`38`	`38`
`39`	`39`	`impl UpstreamCache {`
`40`		`- pub fn new<P: AsRef<Path>>(path: P) -> Self {`
`41`		`- UpstreamCache::new_impl(FileStorage::new(path))`
	`40`	`+ pub fn new<P: AsRef<Path>>(path: P) -> Result<Self> {`
	`41`	`+ Ok(UpstreamCache::new_impl(FileStorage::new(path)?))`
`42`	`42`	`}`
`43`	`43`	`}`
`44`	`44`
`@@ -124,7 +124,7 @@ impl<S: Storage> UpstreamCacheImpl<S> {`
`124`	`124`
`125`	`125`	`pub fn write_record(&mut self, record: &UpstreamCacheRecord) -> Result<()> {`
`126`	`126`	`self.chunk_cached.push(record.clone());`
`127`		`- self.storage.write_chunk(self.current_chunk, &self.chunk_cached)?;`
	`127`	`+ self.storage.write_chunk(self.current_chunk, &self.chunk_cached).context("could not write cache record")?;`
`128`	`128`
`129`	`129`	`self.current_record += 1;`
`130`	`130`	`if self.current_record >= self.density {`
`@@ -155,7 +155,7 @@ impl Storage for FileStorage {`
`155`	`155`	`}`
`156`	`156`
`157`	`157`	`fn write_chunk(&mut self, chunk_no: usize, data: &[UpstreamCacheRecord]) -> Result<()> {`
`158`		`- let mut file = File::create(self.get_file_name(chunk_no))?;`
	`158`	`+ let mut file = File::create(self.get_file_name(chunk_no)).context("could not write chunk")?;`
`159`	`159`	`file.write_all(&serde_json::to_vec(data)?)?;`
`160`	`160`	`Ok(())`
`161`	`161`	`}`
Original file line number	Diff line number	Diff line change
`@@ -4,6 +4,8 @@ genesis-completion-topic = "cardano.sequence.bootstrapped"`
`4`	`4`
`5`	`5`	`node-addresses = [`
`6`	`6`	`"backbone.cardano.iog.io:3001",`
	`7`	`+ "backbone.mainnet.cardanofoundation.org:3001",`
	`8`	`+ "backbone.mainnet.emurgornd.com:3001",`
`7`	`9`	`]`
`8`	`10`	`magic-number = 764824073`
`9`	`11`
Original file line number	Diff line number	Diff line change
`@@ -58,7 +58,7 @@ impl NetworkManager {`
`58`	`58`	`while let Some(event) = self.events.recv().await {`
`59`	`59`	`match event {`
`60`	`60`	`NetworkEvent::PeerUpdate { peer, event } => {`
`61`		`- let maybe_publish_blocks = self.handle_peer_update(peer, event)?;`
	`61`	`+ let maybe_publish_blocks = self.handle_peer_update(peer, event);`
`62`	`62`	`if maybe_publish_blocks {`
`63`	`63`	`self.publish_blocks().await?;`
`64`	`64`	`}`
`@@ -124,7 +124,7 @@ impl NetworkManager {`
`124`	`124`	`// or when publishing messages to other modules. This avoids deadlock; if our event queue`
`125`	`125`	`// is full and this method is blocked on writing to it, the queue can never drain.`
`126`	`126`	`// Returns true if we might have new events to publish downstream.`
`127`		`- fn handle_peer_update(&mut self, peer: PeerId, event: PeerEvent) -> Result<bool> {`
	`127`	`+ fn handle_peer_update(&mut self, peer: PeerId, event: PeerEvent) -> bool {`
`128`	`128`	`let is_preferred = self.preferred_upstream.is_some_and(\|id\| id == peer);`
`129`	`129`	`match event {`
`130`	`130`	`PeerEvent::ChainSync(PeerChainSyncEvent::RollForward(header)) => {`
`@@ -148,12 +148,12 @@ impl NetworkManager {`
`148`	`148`	`break; // only fetch from one`
`149`	`149`	`}`
`150`	`150`	`}`
`151`		`- Ok(false)`
	`151`	`+ false`
`152`	`152`	`}`
`153`	`153`	`BlockStatus::Fetched(_) => {`
`154`	`154`	`// If chainsync has requested a block which we've already fetched,`
`155`	`155`	`// we might be able to publish one or more.`
`156`		`- Ok(is_preferred)`
	`156`	`+ is_preferred`
`157`	`157`	`}`
`158`	`158`	`}`
`159`	`159`	`}`
`@@ -184,18 +184,18 @@ impl NetworkManager {`
`184`	`184`	`}`
`185`	`185`	`}`
`186`	`186`	`}`
`187`		`- Ok(false)`
	`187`	`+ false`
`188`	`188`	`}`
`189`	`189`	`PeerEvent::BlockFetched(fetched) => {`
`190`	`190`	`let Some(block) = self.blocks.get_mut(&fetched.hash) else {`
`191`		`- return Ok(false);`
	`191`	`+ return false;`
`192`	`192`	`};`
`193`	`193`	`block.set_body(&fetched.body);`
`194`		`- Ok(true)`
	`194`	`+ true`
`195`	`195`	`}`
`196`	`196`	`PeerEvent::Disconnected => {`
`197`	`197`	`self.handle_disconnect(peer);`
`198`		`- Ok(false)`
	`198`	`+ false`
`199`	`199`	`}`
`200`	`200`	`}`
`201`	`201`	`}`
Original file line number	Diff line number	Diff line change
`@@ -227,7 +227,7 @@ impl UpstreamChainFetcher {`
`227`	`227`	`Self::sync_to_point(cfg, None, Point::Origin).await?;`
`228`	`228`	`}`
`229`	`229`	`SyncPoint::Cache => {`
`230`		`- let mut upstream_cache = UpstreamCache::new(&cfg.cache_dir);`
	`230`	`+ let mut upstream_cache = UpstreamCache::new(&cfg.cache_dir)?;`
`231`	`231`	`let point = match Self::read_cache(cfg.clone(), &mut upstream_cache).await? {`
`232`	`232`	`None => Point::Origin,`
`233`	`233`	`Some(blk) => Point::Specific(blk.slot, blk.hash.to_vec()),`