Implement checkpoints and catchup

Author: SupernaviX

Cargo.lock

@@ -18,6 +18,7 @@ firefly-server = { path = "../firefly-server" }
 hex = "0.4"
 minicbor = "0.20"
 rand = "0.8"
+rand_chacha = "0.3"
 reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] }
 schemars = "0.8"
 serde = "1"

firefly-cardanoconnect/Cargo.toml

@@ -3,12 +3,13 @@ use dashmap::DashMap;
 use firefly_server::apitypes::{ApiError, ApiResult};
 use tokio::sync::Mutex;
 
-use crate::streams::{Listener, ListenerId, Stream, StreamId};
+use crate::streams::{Listener, ListenerId, Stream, StreamCheckpoint, StreamId};
 
 #[derive(Default)]
 pub struct Persistence {
     all_streams: Mutex<Vec<Stream>>,
     all_listeners: DashMap<StreamId, Vec<Listener>>,
+    all_checkpoints: DashMap<StreamId, StreamCheckpoint>,
 }
 
 impl Persistence {
@@ -45,6 +46,7 @@ impl Persistence {
         streams.retain(|it| it.id != *id);
 
         self.all_listeners.remove(id);
+        self.all_checkpoints.remove(id);
 
         Ok(())
     }
@@ -122,4 +124,25 @@ impl Persistence {
             .collect();
         Ok(listeners)
     }
+
+    pub async fn write_checkpoint(&self, checkpoint: &StreamCheckpoint) -> ApiResult<()> {
+        // check stream existence by checking if we have a (possibly empty) list of listeners for it,
+        // because that's cheaper than locking the streams list
+        if !self.all_listeners.contains_key(&checkpoint.stream_id) {
+            return Err(ApiError::not_found("No stream found with that ID"));
+        }
+        self.all_checkpoints
+            .insert(checkpoint.stream_id.clone(), checkpoint.clone());
+        Ok(())
+    }
+
+    pub async fn read_checkpoint(
+        &self,
+        stream_id: &StreamId,
+    ) -> ApiResult<Option<StreamCheckpoint>> {
+        match self.all_checkpoints.get(stream_id) {
+            Some(checkpoint) => Ok(Some(checkpoint.clone())),
+            None => Ok(None),
+        }
+    }
 }

firefly-cardanoconnect/src/persistence.rs

@@ -39,7 +39,7 @@ impl StreamManager {
             batch_timeout,
         };
         self.persistence.write_stream(&stream).await?;
-        self.mux.handle_stream_write(&stream).await;
+        self.mux.handle_stream_write(&stream).await?;
         Ok(stream)
     }
 
@@ -74,7 +74,7 @@ impl StreamManager {
             stream.batch_timeout = timeout;
         }
         self.persistence.write_stream(&stream).await?;
-        self.mux.handle_stream_write(&stream).await;
+        self.mux.handle_stream_write(&stream).await?;
         Ok(stream)
     }
 
@@ -103,7 +103,7 @@ impl StreamManager {
             stream_id: stream_id.clone(),
         };
         self.persistence.write_listener(&listener).await?;
-        self.mux.handle_listener_write(&listener).await;
+        self.mux.handle_listener_write(&listener).await?;
         Ok(listener)
     }
 
@@ -140,7 +140,7 @@ impl StreamManager {
     ) -> ApiResult<()> {
         self.mux
             .handle_listener_delete(stream_id, listener_id)
-            .await;
+            .await?;
         self.persistence
             .delete_listener(stream_id, listener_id)
             .await?;