fix(client): connectvity caching

OMG, I'm so dumb.

Author: dpc

crates/rostra-client/src/client.rs

@@ -34,9 +34,9 @@ use tracing::{debug, info, trace, warn};
 use super::{RRECORD_HEAD_KEY, RRECORD_P2P_KEY, get_rrecord_typed};
 use crate::LOG_TARGET;
 use crate::error::{
-    ActivateResult, ActivateSnafu, IdResolveError, IdResolveResult, IdSecretReadResult,
-    InitIrohClientSnafu, InitPkarrClientSnafu, InitResult, InvalidIdSnafu, IoSnafu,
-    MissingTicketSnafu, ParsingSnafu, PkarrResolveSnafu, PostResult, RRecordSnafu,
+    ActivateResult, ActivateSnafu, ConnectResult, IdResolveError, IdResolveResult,
+    IdSecretReadResult, InitIrohClientSnafu, InitPkarrClientSnafu, InitResult, InvalidIdSnafu,
+    IoSnafu, MissingTicketSnafu, ParsingSnafu, PkarrResolveSnafu, PostResult, RRecordSnafu,
     SecretMismatchSnafu,
 };
 use crate::id::{CompactTicket, IdPublishedData, IdResolvedData};
@@ -227,7 +227,7 @@ impl ClientRef<'_> {
     /// Returns a cached connection if available, otherwise creates a new one.
     /// This is more efficient than `connect_uncached` when making repeated
     /// connections to the same peer.
-    pub async fn connect_cached(&self, id: RostraId) -> Option<Connection> {
+    pub async fn connect_cached(&self, id: RostraId) -> ConnectResult<Connection> {
         self.connection_cache.get_or_connect(self, id).await
     }
 }

crates/rostra-client/src/connection_cache.rs

@@ -7,10 +7,11 @@ use tokio::sync::{Mutex, OnceCell};
 use tracing::trace;
 
 use crate::ClientRef;
+use crate::error::ConnectResult;
 
 const LOG_TARGET: &str = "rostra-client::connection-cache";
 
-type LazySharedConnection = Arc<OnceCell<Option<Connection>>>;
+type LazySharedConnection = Arc<OnceCell<Connection>>;
 
 #[derive(Clone)]
 pub struct ConnectionCache {
@@ -30,14 +31,18 @@ impl ConnectionCache {
         }
     }
 
-    pub async fn get_or_connect(&self, client: &ClientRef<'_>, id: RostraId) -> Option<Connection> {
+    pub async fn get_or_connect(
+        &self,
+        client: &ClientRef<'_>,
+        id: RostraId,
+    ) -> ConnectResult<Connection> {
         let mut pool_lock = self.connections.lock().await;
 
         let entry_arc = pool_lock
             .entry(id)
             .and_modify(|entry_arc| {
                 // Check if existing connection is disconnected and remove it
-                if let Some(Some(existing_conn)) = entry_arc.get()
+                if let Some(existing_conn) = entry_arc.get()
                     && existing_conn.is_closed() {
                         trace!(target: LOG_TARGET, %id, "Existing connection is disconnected, removing from pool");
                         *entry_arc = Arc::new(OnceCell::new());
@@ -50,21 +55,21 @@ impl ConnectionCache {
         drop(pool_lock);
 
         let result = entry_arc
-            .get_or_init(|| async {
+            .get_or_try_init(|| async {
                 trace!(target: LOG_TARGET, %id, "Creating new connection");
                 match client.connect_uncached(id).await {
                     Ok(conn) => {
                         trace!(target: LOG_TARGET, %id, "Connection successful");
-                        Some(conn)
+                        Ok(conn)
                     }
                     Err(err) => {
                         trace!(target: LOG_TARGET, %id, err = %err, "Connection failed");
-                        None
+                        Err(err)
                     }
                 }
             })
             .await;
 
-        result.clone()
+        result.cloned()
     }
 }

crates/rostra-client/src/task/followee_head_checker.rs

@@ -160,7 +160,7 @@ impl FolloweeHeadChecker {
         id: RostraId,
         connections: &ConnectionCache,
     ) -> BoxedErrorResult<Option<ShortEventId>> {
-        let Some(conn) = connections.get_or_connect(client, id).await else {
+        let Ok(conn) = connections.get_or_connect(client, id).await else {
             return Ok(None);
         };
 
@@ -239,8 +239,19 @@ impl FolloweeHeadChecker {
             };
 
             // ConnectionCache handles its own interior mutability
-            let Some(conn) = connections.get_or_connect(&client, *follower_id).await else {
-                continue;
+            let conn = match connections.get_or_connect(&client, *follower_id).await {
+                Ok(conn) => conn,
+                Err(err) => {
+                    debug!(target: LOG_TARGET,
+                        rostra_id = %rostra_id,
+                        head = %head,
+                        follower_id = %follower_id,
+                        err = %err.fmt_compact(),
+                        "Could not connect to fetch updates"
+                    );
+
+                    continue;
+                }
             };
 
             debug!(target: LOG_TARGET,

crates/rostra-client/src/task/missing_event_fetcher.rs

@@ -88,7 +88,7 @@ impl MissingEventFetcher {
                     follower_id = %follower_id,
                     "Looking for a missing events from"
                 );
-                let Some(conn) = connections.get_or_connect(&client, *follower_id).await else {
+                let Ok(conn) = connections.get_or_connect(&client, *follower_id).await else {
                     debug!(
                         target:  LOG_TARGET,
                         author_id = %author_id,

crates/rostra-client/src/util/rpc.rs

@@ -51,7 +51,8 @@ pub async fn get_event_content_from_followers(
 
                     let conn = (connections_cache
                         .get_or_connect(&client_ref, follower_id)
-                        .await)?;
+                        .await)
+                        .ok()?;
 
                     debug!(target: LOG_TARGET,
                         author_id = %author_id.to_short(),