wip: adds scaffold for resolver actor

Author: nathanielc

pipeline/src/aggregator/mod.rs

@@ -870,6 +870,8 @@ actor_envelope! {
     AggregatorRecorder,
     SubscribeSince => SubscribeSinceMsg,
     NewConclusionEvents => NewConclusionEventsMsg,
+    // TODO: Remove this message and use the analogous message on the Resolver.
+    // This way the canonical stream state is provided via the API
     StreamState => StreamStateMsg,
 }
 
@@ -903,13 +905,6 @@ impl Handler<SubscribeSinceMsg> for Aggregator {
         // Execute query to get initial (historical) results
         let query_stream = df.execute_stream().await?;
 
-        // Create subscription stream
-        let subscription_stream = RecordBatchStreamAdapter::new(
-            schemas::event_states(),
-            tokio_stream::wrappers::BroadcastStream::new(subscription)
-                .map_err(|err| exec_datafusion_err!("{err}")),
-        );
-
         // Merge query results with subscription updates
         rows_since(RowsSinceInput {
             session_context: &ctx,
@@ -918,7 +913,11 @@ impl Handler<SubscribeSinceMsg> for Aggregator {
             projection: message.projection,
             filters: message.filters,
             limit: message.limit,
-            subscription: Box::pin(subscription_stream),
+            subscription: Box::pin(RecordBatchStreamAdapter::new(
+                schemas::event_states(),
+                tokio_stream::wrappers::BroadcastStream::new(subscription)
+                    .map_err(|err| exec_datafusion_err!("{err}")),
+            )),
             since: query_stream,
         })
     }

pipeline/src/lib.rs

@@ -14,6 +14,7 @@ mod cache_table;
 pub mod cid_part;
 pub mod cid_string;
 pub mod concluder;
+pub mod resolver;
 mod config;
 pub mod dimension_extract;
 mod metrics;

pipeline/src/metrics.rs

@@ -11,15 +11,24 @@ use prometheus_client::{
 pub struct Metrics {
     pub(crate) message_count: Family<MessageLabels, Counter>,
 
+    pub(crate) concluder_poll_new_events_loop_count: Counter,
+
     pub(crate) aggregator_new_conclusion_events_count: Counter,
 
-    pub(crate) concluder_poll_new_events_loop_count: Counter,
+    pub(crate) resolver_new_event_states_count: Counter,
 }
 impl Metrics {
     /// Register and construct Metrics
     pub fn register(registry: &mut Registry) -> Self {
         let sub_registry = registry.sub_registry_with_prefix("pipeline");
 
+        register!(
+            concluder_poll_new_events_loop_count,
+            "Number of times the loop to poll new conclusion events has run",
+            Counter::default(),
+            sub_registry
+        );
+
         register!(
             message_count,
             "Number of messages delivered to actors",
@@ -32,18 +41,18 @@ impl Metrics {
             Counter::default(),
             sub_registry
         );
-
         register!(
-            concluder_poll_new_events_loop_count,
-            "Number of times the loop to poll new conclusion events has run",
+            resolver_new_event_states_count,
+            "Number of new event states delivered to the resolver",
             Counter::default(),
             sub_registry
         );
 
         Self {
             message_count,
-            aggregator_new_conclusion_events_count,
             concluder_poll_new_events_loop_count,
+            aggregator_new_conclusion_events_count,
+            resolver_new_event_states_count,
         }
     }
 }

pipeline/src/resolver/metrics.rs

@@ -0,0 +1,24 @@
+use ceramic_actor::MessageEvent;
+use ceramic_metrics::Recorder;
+
+use crate::metrics::{MessageLabels, Metrics};
+
+use super::{NewEventStatesMsg, ResolverRecorder, StreamStateMsg};
+
+impl Recorder<MessageEvent<NewEventStatesMsg>> for Metrics {
+    fn record(&self, event: &MessageEvent<NewEventStatesMsg>) {
+        self.message_count
+            .get_or_create(&MessageLabels::from(event))
+            .inc();
+        self.resolver_new_event_states_count
+            .inc_by(event.message.events.num_rows() as u64);
+    }
+}
+impl Recorder<MessageEvent<StreamStateMsg>> for Metrics {
+    fn record(&self, event: &MessageEvent<StreamStateMsg>) {
+        self.message_count
+            .get_or_create(&MessageLabels::from(event))
+            .inc();
+    }
+}
+impl ResolverRecorder for Metrics {}

pipeline/src/resolver/mock.rs

@@ -0,0 +1,80 @@
+//! Provides a mock implmentation of the aggregator actor.
+use async_trait::async_trait;
+use ceramic_actor::{Actor, Handler, Message};
+use mockall::mock;
+use prometheus_client::registry::Registry;
+
+use crate::metrics::Metrics;
+
+use super::{
+    NewEventStatesMsg, Resolver, ResolverActor, ResolverEnvelope, ResolverHandle, StreamStateMsg,
+    SubscribeSinceMsg,
+};
+
+mock! {
+    // mockall does not support multiple methods on the struct with the same name.
+    // This arises when implementing multiple traits that have methods with the same name as is
+    // the case with the [`ceramic_actor::Handler`] trait.
+    //
+    // We add a layer of indirection to get around this limitation.
+    pub Resolver {
+        #[allow(missing_docs)]
+        pub fn handle_subscribe_since(
+            &mut self,
+            message: SubscribeSinceMsg,
+        ) -> <SubscribeSinceMsg as Message>::Result;
+        #[allow(missing_docs)]
+        pub fn handle_new_event_states(
+            &mut self,
+            message: NewEventStatesMsg,
+        ) -> <NewEventStatesMsg as Message>::Result;
+        #[allow(missing_docs)]
+        pub fn handle_stream_state(
+            &mut self,
+            message: StreamStateMsg,
+        ) -> <StreamStateMsg
+            as Message>::Result;
+    }
+}
+
+#[async_trait]
+impl Handler<SubscribeSinceMsg> for MockResolver {
+    async fn handle(
+        &mut self,
+        message: SubscribeSinceMsg,
+    ) -> <SubscribeSinceMsg as Message>::Result {
+        self.handle_subscribe_since(message)
+    }
+}
+
+#[async_trait]
+impl Handler<NewEventStatesMsg> for MockResolver {
+    async fn handle(
+        &mut self,
+        message: NewEventStatesMsg,
+    ) -> <NewEventStatesMsg as Message>::Result {
+        self.handle_new_event_states(message)
+    }
+}
+
+#[async_trait]
+impl Handler<StreamStateMsg> for MockResolver {
+    async fn handle(&mut self, message: StreamStateMsg) -> <StreamStateMsg as Message>::Result {
+        self.handle_stream_state(message)
+    }
+}
+
+impl Actor for MockResolver {
+    type Envelope = ResolverEnvelope;
+}
+impl ResolverActor for MockResolver {}
+
+impl MockResolver {
+    /// Spawn a mock aggregator actor.
+    pub fn spawn(mock_actor: MockResolver) -> ResolverHandle {
+        let metrics = Metrics::register(&mut Registry::default());
+        let (handle, _task_handle) =
+            Resolver::spawn(1_000, mock_actor, metrics, std::future::pending());
+        handle
+    }
+}