bug: flush aggregator cache on shutdown

actor/src/lib.rs

@@ -61,6 +61,16 @@ pub trait Actor {
     }
 }
 
+/// Optional trait for actors that need to perform cleanup operations during shutdown.
+/// Only actors that need cleanup should implement this trait.
+#[async_trait]
+pub trait Shutdown {
+    /// Called when the actor is shutting down to allow for cleanup operations.
+    /// This method is called after all pending messages have been processed
+    /// but before the actor loop exits.
+    async fn shutdown(&mut self);
+}
+
 /// Wrapper of any T with its tracing span context.
 #[derive(Debug)]
 pub struct Traced<T> {

actor/src/macros.rs

@@ -1,19 +1,75 @@
 /// Constructs an envelope enumeration that contains all messages for an actor.
 ///
+/// # Usage
+///
+/// Basic usage (no shutdown support):
+/// ```ignore
+/// actor_envelope! {
+///     MyEnvelope,
+///     MyActor,
+///     MyRecorder,
+///     Message1 => Msg1Type,
+///     Message2 => Msg2Type,
+/// }
+/// ```
+///
+/// With shutdown support (actor must implement `Shutdown` trait):
+/// ```ignore
+/// actor_envelope! {
+///     MyEnvelope,
+///     MyActor,
+///     MyRecorder,
+///     with_shutdown,
+///     Message1 => Msg1Type,
+///     Message2 => Msg2Type,
+/// }
+/// ```
+///
 /// The first identifier is the name of the enum.
 /// The second identifier is the name of a trait specific to the actor.
 /// The third identifier is the name of a trait for recording message events.
+/// Optional `with_shutdown` parameter adds shutdown support for actors that implement the Shutdown trait.
 /// The remaining pairs are the variants of the envelope indicating the messages the actor handles.
 ///
 /// The constructed actor trait is a union of the [`crate::Handler`] traits for each message along with the [`crate::Actor`] trait.
 ///
 /// The constructed recorder trait is a union of the [`ceramic_metrics::Recorder`] traits for each message.
 #[macro_export]
 macro_rules! actor_envelope {
+    // Version with shutdown support
+    (
+        $enum_name:ident,
+        $actor_trait:ident,
+        $recorder_trait:ident,
+        with_shutdown,
+        $(
+            $variant_name:ident => $message_type:ty,
+        )*
+    ) => {
+        $crate::actor_envelope_impl!($enum_name, $actor_trait, $recorder_trait, with_shutdown, $($variant_name => $message_type,)*);
+    };
+
+    // Default version without shutdown
+    (
+        $enum_name:ident,
+        $actor_trait:ident,
+        $recorder_trait:ident,
+        $(
+            $variant_name:ident => $message_type:ty,
+        )*
+    ) => {
+        $crate::actor_envelope_impl!($enum_name, $actor_trait, $recorder_trait, without_shutdown, $($variant_name => $message_type,)*);
+    };
+}
+
+/// Internal implementation macro that handles both with/without shutdown cases
+#[macro_export]
+macro_rules! actor_envelope_impl {
     (
         $enum_name:ident,
         $actor_trait:ident,
         $recorder_trait:ident,
+        $shutdown_mode:ident,
         $(
             $variant_name:ident => $message_type:ty,
         )*
@@ -51,11 +107,7 @@ macro_rules! actor_envelope {
                 }
             }
         }
-        #[doc = std::stringify!($actor_trait)]
-        #[doc = " is an [`ceramic_actor::Actor`] and [`ceramic_actor::Handler`] for each message type in the actor envelope "]
-        #[doc = stringify!($enum_name)]
-        #[doc = "."]
-        pub trait $actor_trait : $crate::Actor<Envelope = $enum_name> $( + $crate::Handler<$message_type> )* + ::std::marker::Send + 'static { }
+        $crate::actor_trait_def!($actor_trait, $enum_name, $shutdown_mode, $($message_type,)*);
 
         #[doc = std::stringify!($recorder_trait)]
         #[doc = " is an [`ceramic_metrics::Recorder`] for each message type in the actor envelope "]
@@ -64,6 +116,7 @@ macro_rules! actor_envelope {
         pub trait $recorder_trait : $(::ceramic_metrics::Recorder<$crate::MessageEvent<$message_type>> +)*
             ::std::fmt::Debug + ::std::marker::Send + ::std::marker::Sync + 'static { }
 
+
         impl $enum_name {
             /// Runs the actor handling messages as they arrive.
             pub async fn run<A>(mut actor: A, mut receiver: $crate::Receiver<A::Envelope>, mut shutdown: impl ::std::future::Future<Output=()> + ::std::marker::Send + 'static)
@@ -111,6 +164,7 @@ macro_rules! actor_envelope {
                         }
                     }
                 }
+                $crate::actor_shutdown_call!($shutdown_mode, actor);
             }
         }
         $(
@@ -132,3 +186,33 @@ macro_rules! actor_envelope {
         )*
     };
 }
+
+/// Helper macro to define actor trait with or without shutdown requirement
+#[macro_export]
+macro_rules! actor_trait_def {
+    ($actor_trait:ident, $enum_name:ident, with_shutdown, $($message_type:ty,)*) => {
+        #[doc = std::stringify!($actor_trait)]
+        #[doc = " is an [`ceramic_actor::Actor`] and [`ceramic_actor::Handler`] for each message type in the actor envelope "]
+        #[doc = stringify!($enum_name)]
+        #[doc = ". This actor must also implement [`ceramic_actor::Shutdown`]."]
+        pub trait $actor_trait : $crate::Actor<Envelope = $enum_name> $( + $crate::Handler<$message_type> )* + $crate::Shutdown + ::std::marker::Send + 'static { }
+    };
+    ($actor_trait:ident, $enum_name:ident, without_shutdown, $($message_type:ty,)*) => {
+        #[doc = std::stringify!($actor_trait)]
+        #[doc = " is an [`ceramic_actor::Actor`] and [`ceramic_actor::Handler`] for each message type in the actor envelope "]
+        #[doc = stringify!($enum_name)]
+        #[doc = "."]
+        pub trait $actor_trait : $crate::Actor<Envelope = $enum_name> $( + $crate::Handler<$message_type> )* + ::std::marker::Send + 'static { }
+    };
+}
+
+/// Helper macro to conditionally call shutdown
+#[macro_export]
+macro_rules! actor_shutdown_call {
+    (with_shutdown, $actor:ident) => {
+        $actor.shutdown().await;
+    };
+    (without_shutdown, $actor:ident) => {
+        // No shutdown call for actors that don't implement shutdown
+    };
+}

one/src/daemon.rs

@@ -1,7 +1,7 @@
 use std::{path::PathBuf, time::Duration};
 
 use crate::{
-    default_directory, handle_signals, http, http_metrics, metrics, network::Ipfs, DBOpts,
+    default_directory, handle_signals, http, http_metrics, metrics, network::Ipfs, startup, DBOpts,
     DBOptsExperimental, Info, LogOpts, Network,
 };
 use anyhow::{anyhow, bail, Result};
@@ -274,6 +274,16 @@ pub struct DaemonOpts {
         default_value = "file://./pipeline"
     )]
     object_store_url: url::Url,
+
+    /// Comma-separated list of stream IDs or model IDs to register as initial interests on startup.
+    /// The node will track these streams and sync their data.
+    #[arg(
+        long,
+        use_value_delimiter = true,
+        value_delimiter = ',',
+        env = "CERAMIC_ONE_INITIAL_INTERESTS"
+    )]
+    initial_interests: Vec<String>,
 }
 
 fn spawn_database_optimizer(
@@ -447,6 +457,10 @@ pub async fn run(opts: DaemonOpts) -> Result<()> {
     let node_key = NodeKey::try_from_dir(opts.p2p_key_dir).await?;
     let node_id = node_key.id();
 
+    // Process initial interests if provided
+    startup::process_initial_interests(&opts.initial_interests, &interest_svc, &network, &node_id)
+        .await?;
+
     // Register metrics for all components
     let recon_metrics = MetricsHandle::register(recon::Metrics::register);
     let peer_svc_store_metrics =

one/src/lib.rs

@@ -8,6 +8,7 @@ mod metrics;
 mod migrations;
 mod network;
 mod query;
+mod startup;
 
 use anyhow::{anyhow, bail, Result};
 use ceramic_core::ssi::caip2::ChainId;

one/src/startup.rs

@@ -0,0 +1,186 @@
+//! Startup utilities for the Ceramic One daemon.
+
+use anyhow::{anyhow, Result};
+use ceramic_api::InterestService as InterestServiceTrait;
+use ceramic_core::{EventId, Interest, NodeId, StreamId};
+use ceramic_interest_svc::InterestService;
+use std::str::FromStr;
+use std::sync::Arc;
+use tracing::{debug, info, warn};
+
+/// Process initial interests by registering them with the interest service
+pub async fn process_initial_interests(
+    initial_interests: &[String],
+    interest_svc: &Arc<InterestService>,
+    network: &ceramic_core::Network,
+    node_id: &NodeId,
+) -> Result<()> {
+    if initial_interests.is_empty() {
+        return Ok(());
+    }
+
+    info!("Processing {} initial interests", initial_interests.len());
+
+    for stream_id_str in initial_interests {
+        let stream_id_str = stream_id_str.trim();
+        if stream_id_str.is_empty() {
+            continue;
+        }
+
+        // Validate that the model stream ID is parseable
+        let _stream_id = StreamId::from_str(stream_id_str)
+            .map_err(|e| anyhow!("Invalid model ID '{}': {}", stream_id_str, e))?;
+
+        // Create an interest for the "model" separator key covering the full range for this stream
+        // This follows the same pattern as the API endpoint /ceramic/interests/model/{stream_id}
+        let stream_id_bytes = multibase::decode(stream_id_str)
+            .map_err(|e| anyhow!("Failed to decode stream ID '{}': {}", stream_id_str, e))?
+            .1;
+        let start = EventId::builder()
+            .with_network(network)
+            .with_sep("model", &stream_id_bytes)
+            .with_min_controller()
+            .with_min_init()
+            .with_min_event()
+            .build_fencepost();
+        let stop = EventId::builder()
+            .with_network(network)
+            .with_sep("model", &stream_id_bytes)
+            .with_max_controller()
+            .with_max_init()
+            .with_max_event()
+            .build_fencepost();
+
+        let interest = Interest::builder()
+            .with_sep_key("model")
+            .with_peer_id(&node_id.peer_id())
+            .with_range((start.as_slice(), stop.as_slice()))
+            .with_not_after(0)
+            .build();
+
+        match interest_svc.insert(interest).await {
+            Ok(was_inserted) => {
+                if was_inserted {
+                    info!(
+                        "Successfully registered initial interest for stream: {}",
+                        stream_id_str
+                    );
+                } else {
+                    debug!(
+                        "Interest for stream {} was already registered",
+                        stream_id_str
+                    );
+                }
+            }
+            Err(e) => {
+                warn!(
+                    "Failed to register initial interest for stream {}: {}",
+                    stream_id_str, e
+                );
+            }
+        }
+    }
+
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use ceramic_core::Network;
+    use ceramic_sql::sqlite::SqlitePool;
+    use std::sync::Arc;
+
+    #[tokio::test]
+    async fn test_process_initial_interests_empty() {
+        let pool = SqlitePool::connect_in_memory().await.unwrap();
+        let interest_svc = Arc::new(InterestService::new(pool));
+        let network = Network::InMemory;
+        let node_key = ceramic_core::NodeKey::random();
+        let node_id = node_key.id();
+
+        // Test with empty interests
+        let result = process_initial_interests(&[], &interest_svc, &network, &node_id).await;
+        assert!(result.is_ok());
+    }
+
+    #[tokio::test]
+    async fn test_process_initial_interests_invalid_stream_id() {
+        let pool = SqlitePool::connect_in_memory().await.unwrap();
+        let interest_svc = Arc::new(InterestService::new(pool));
+        let network = Network::InMemory;
+        let node_key = ceramic_core::NodeKey::random();
+        let node_id = node_key.id();
+
+        // Invalid stream ID - tests daemon-specific stream ID parsing and error handling
+        let stream_ids = vec!["invalid-stream-id".to_string()];
+
+        let result =
+            process_initial_interests(&stream_ids, &interest_svc, &network, &node_id).await;
+        assert!(result.is_err());
+
+        // Verify the error message contains information about the invalid stream ID
+        let error_msg = result.unwrap_err().to_string();
+        assert!(error_msg.contains("invalid-stream-id"));
+    }
+
+    #[tokio::test]
+    async fn test_process_initial_interests_mixed_valid_invalid() {
+        let pool = SqlitePool::connect_in_memory().await.unwrap();
+        let interest_svc = Arc::new(InterestService::new(pool));
+        let network = Network::InMemory;
+        let node_key = ceramic_core::NodeKey::random();
+        let node_id = node_key.id();
+
+        // Mix of valid and invalid stream IDs - tests early failure behavior
+        let stream_ids = vec![
+            "k2t6wz4ylx0qr6v7dvbczbxqy7pqjb0879qx930c1e27gacg3r8sllonqt4xx9".to_string(), // Valid
+            "invalid-stream-id".to_string(),                                              // Invalid
+        ];
+
+        let result =
+            process_initial_interests(&stream_ids, &interest_svc, &network, &node_id).await;
+        // Should fail on the first invalid ID
+        assert!(result.is_err());
+    }
+
+    #[tokio::test]
+    async fn test_process_initial_interests_whitespace_handling() {
+        let pool = SqlitePool::connect_in_memory().await.unwrap();
+        let interest_svc = Arc::new(InterestService::new(pool));
+        let network = Network::InMemory;
+        let node_key = ceramic_core::NodeKey::random();
+        let node_id = node_key.id();
+
+        // Test daemon-specific input sanitization logic
+        let stream_ids = vec![
+            "  k2t6wz4ylx0qr6v7dvbczbxqy7pqjb0879qx930c1e27gacg3r8sllonqt4xx9  ".to_string(), // Valid with whitespace
+            "".to_string(),    // Empty string
+            "   ".to_string(), // Only whitespace
+        ];
+
+        let result =
+            process_initial_interests(&stream_ids, &interest_svc, &network, &node_id).await;
+        // Should succeed and handle whitespace correctly
+        assert!(result.is_ok());
+    }
+
+    #[tokio::test]
+    async fn test_process_initial_interests_multibase_decoding() {
+        let pool = SqlitePool::connect_in_memory().await.unwrap();
+        let interest_svc = Arc::new(InterestService::new(pool));
+        let network = Network::InMemory;
+        let node_key = ceramic_core::NodeKey::random();
+        let node_id = node_key.id();
+
+        // Test daemon-specific multibase decoding logic
+        let stream_ids = vec![
+            "k2t6wz4ylx0qr6v7dvbczbxqy7pqjb0879qx930c1e27gacg3r8sllonqt4xx9".to_string(), // Valid multibase-encoded stream ID
+        ];
+
+        let result =
+            process_initial_interests(&stream_ids, &interest_svc, &network, &node_id).await;
+        // Should succeed with proper multibase decoding
+        assert!(result.is_ok());
+    }
+}