wip: add concluder todos

Author: nathanielc

event-svc/src/event/service.rs

@@ -396,6 +396,11 @@ impl EventService {
                     init,
                     previous: vec![*time_event.prev()],
                     order: delivered as u64,
+                    before: todo!(),
+                    chain_id: todo!(),
+                    tx_hash: todo!(),
+                    tx_type: todo!(),
+                    root: todo!(),
                 }))
             }
             ceramic_event::unvalidated::Event::Signed(signed_event) => {

pipeline/src/concluder/event.rs

@@ -2,6 +2,7 @@ use std::sync::Arc;
 
 use anyhow::{anyhow, Result};
 use arrow::datatypes::{DataType, Field, Int32Type};
+use ceramic_core::ssi::caip2;
 use ceramic_core::METAMODEL_STREAM_ID;
 use ceramic_event::{unvalidated, StreamId, StreamIdType};
 use cid::Cid;
@@ -110,7 +111,34 @@ pub struct ConclusionTime {
     pub init: ConclusionInit,
     /// Ordered list of previous events this event references.
     pub previous: Vec<Cid>,
-    //TODO Add temporal conclusions, i.e the block timestamp of this event
+
+    // TODO figure out how to populate these values.
+    // Once populated we need to update the conclusion_events table schema to include them
+    // Then we need to preserve them in the event_states table
+    // Finally in the resolver we can use this information to determine a canonical tip for a
+    // stream.
+    //
+    // Challenges: The anchor proof information is gather when validating the event but then
+    // forgotten. It likely needs to be persisteted into a new sqlite table and then the conclusion
+    // feed call needs to join against that table to produce these fields.
+    //
+    // How do we populate that table for existing data?
+    // Likely need some kind of explicit migration command that can be run to populate the table.
+    //
+    /// It is known that the time event existed before this time as a Unix timestamp in seconds.
+    pub before: u64,
+
+    // Only before should be needed for conflict resolution but it may be nice to preserve these
+    // other values as well?
+    //
+    /// Chain ID where this time event was anchored.
+    pub chain_id: String,
+    /// Transaction hash that anchored this time event.
+    pub tx_hash: String,
+    /// Transaction type (TODO: not sure what this is).
+    pub tx_type: String,
+    /// Root cid of the proof.
+    pub root: Cid,
 }
 
 impl<'a> TryFrom<&'a unvalidated::Event<Ipld>> for ConclusionInit {

pipeline/src/concluder/metrics.rs

@@ -3,7 +3,7 @@ use ceramic_metrics::Recorder;
 
 use crate::metrics::{MessageLabels, Metrics};
 
-use super::{ConcluderRecorder, EventsSinceMsg, NewEventsMsg};
+use super::{ConcluderRecorder, NewEventsMsg};
 
 impl Recorder<MessageEvent<NewEventsMsg>> for Metrics {
     fn record(&self, event: &MessageEvent<NewEventsMsg>) {
@@ -12,11 +12,4 @@ impl Recorder<MessageEvent<NewEventsMsg>> for Metrics {
             .inc();
     }
 }
-impl Recorder<MessageEvent<EventsSinceMsg>> for Metrics {
-    fn record(&self, event: &MessageEvent<EventsSinceMsg>) {
-        self.message_count
-            .get_or_create(&MessageLabels::from(event))
-            .inc();
-    }
-}
 impl ConcluderRecorder for Metrics {}

pipeline/src/concluder/mock.rs

@@ -7,8 +7,7 @@ use prometheus_client::registry::Registry;
 use crate::metrics::Metrics;
 
 use super::{
-    Concluder, ConcluderActor, ConcluderEnvelope, ConcluderHandle, EventsSinceMsg, NewEventsMsg,
-    SubscribeSinceMsg,
+    Concluder, ConcluderActor, ConcluderEnvelope, ConcluderHandle, NewEventsMsg, SubscribeSinceMsg,
 };
 
 mock! {
@@ -28,11 +27,6 @@ mock! {
             &mut self,
             message: SubscribeSinceMsg,
         ) -> <SubscribeSinceMsg as Message>::Result;
-        #[allow(missing_docs)]
-        pub fn handle_events_since(
-            &mut self,
-            message: EventsSinceMsg,
-        ) -> <EventsSinceMsg as Message>::Result;
     }
 }
 
@@ -53,13 +47,6 @@ impl Handler<SubscribeSinceMsg> for MockConcluder {
     }
 }
 
-#[async_trait]
-impl Handler<EventsSinceMsg> for MockConcluder {
-    async fn handle(&mut self, message: EventsSinceMsg) -> <EventsSinceMsg as Message>::Result {
-        self.handle_events_since(message)
-    }
-}
-
 impl Actor for MockConcluder {
     type Envelope = ConcluderEnvelope;
 }

pipeline/src/concluder/mod.rs

@@ -169,7 +169,6 @@ actor_envelope! {
     ConcluderRecorder,
     NewEvents => NewEventsMsg,
     SubscribeSince => SubscribeSinceMsg,
-    EventsSince => EventsSinceMsg,
 }
 
 /// Notify actor of new events
@@ -329,23 +328,6 @@ async fn events_since(
     Ok(conclusion_events.execute_stream().await?)
 }
 
-/// Request the events since a highwater mark
-#[derive(Debug)]
-pub struct EventsSinceMsg {
-    /// Optional filters to apply to the query
-    pub filters: Vec<Expr>,
-}
-impl Message for EventsSinceMsg {
-    type Result = anyhow::Result<SendableRecordBatchStream>;
-}
-
-#[async_trait]
-impl Handler<EventsSinceMsg> for Concluder {
-    async fn handle(&mut self, message: EventsSinceMsg) -> <EventsSinceMsg as Message>::Result {
-        events_since(&self.ctx, Some(message.filters), None).await
-    }
-}
-
 #[async_trait]
 impl FeedTableSource for ConcluderHandle {
     fn schema(&self) -> SchemaRef {

pipeline/src/lib.rs

@@ -14,10 +14,10 @@ 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;
+pub mod resolver;
 pub mod schemas;
 mod since;
 pub mod stream_id_string;

pipeline/src/resolver/mod.rs

@@ -35,7 +35,7 @@ use datafusion::{
     functions_aggregate::expr_fn::last_value,
     logical_expr::{col, lit, ExprFunctionExt as _},
     physical_plan::stream::RecordBatchStreamAdapter,
-    prelude::{wildcard, SessionContext},
+    prelude::{wildcard, Expr, SessionContext},
 };
 use futures::TryStreamExt as _;
 use shutdown::{Shutdown, ShutdownSignal};
@@ -47,7 +47,7 @@ use crate::{
     cache_table::CacheTable,
     metrics::Metrics,
     schemas,
-    since::{rows_since, FeedTable, FeedTableSource},
+    since::{gt_expression, rows_since, FeedTable, FeedTableSource, RowsSinceInput},
     PipelineContext, Result, SessionContextRef,
 };
 // Use the SubscribeSinceMsg so its clear its a message for this actor
@@ -310,7 +310,7 @@ async fn aggregator_subscription(
     let mut rx = aggregator
         .send(SubscribeSinceMsg {
             projection: None,
-            offset,
+            filters: offset.map(|o| vec![gt_expression("event_state_order", o)]),
             limit: None,
         })
         .await??;
@@ -354,40 +354,42 @@ impl Handler<SubscribeSinceMsg> for Resolver {
     ) -> <SubscribeSinceMsg as Message>::Result {
         let subscription = self.broadcast_tx.subscribe();
         let ctx = self.ctx.clone();
-        rows_since(
-            schemas::stream_states(),
-            "stream_state_order",
-            message.projection,
-            message.offset,
-            message.limit,
-            Box::pin(RecordBatchStreamAdapter::new(
+        rows_since(RowsSinceInput {
+            session_context: &ctx,
+            schema: schemas::stream_states(),
+            order_col: "stream_state_order",
+            projection: message.projection,
+            filters: message.filters.clone(),
+            limit: message.limit,
+            subscription: Box::pin(RecordBatchStreamAdapter::new(
                 schemas::stream_states(),
                 tokio_stream::wrappers::BroadcastStream::new(subscription)
                     .map_err(|err| exec_datafusion_err!("{err}")),
             )),
             // Future Optimization can be to send the projection and limit into the events_since call.
-            stream_states_since(&ctx, message.offset).await?,
-        )
+            since: stream_states_since(&ctx, message.filters).await?,
+        })
     }
 }
 
 async fn stream_states_since(
     ctx: &SessionContext,
-    offset: Option<u64>,
+    filters: Option<Vec<Expr>>,
 ) -> Result<SendableRecordBatchStream> {
     let mut stream_states = ctx
         .table(STREAM_STATES_TABLE)
         .await?
         .select(vec![wildcard()])?
         // Do not return the partition columns
-        .drop_columns(&["stream_cid_partition"])?;
-    if let Some(offset) = offset {
-        stream_states = stream_states.filter(col("stream_state_order").gt(lit(offset)))?;
+        .drop_columns(&["stream_cid_partition"])?
+        .sort(vec![col("stream_state_order").sort(true, true)])?;
+
+    if let Some(filters) = filters {
+        for filter in filters {
+            stream_states = stream_states.filter(filter)?;
+        }
     }
-    Ok(stream_states
-        .sort(vec![col("stream_state_order").sort(true, true)])?
-        .execute_stream()
-        .await?)
+    Ok(stream_states.execute_stream().await?)
 }
 
 /// Inform the resolver about new event states.
@@ -581,13 +583,13 @@ impl FeedTableSource for ResolverHandle {
     async fn subscribe_since(
         &self,
         projection: Option<Vec<usize>>,
-        offset: Option<u64>,
+        filters: Option<Vec<Expr>>,
         limit: Option<usize>,
     ) -> anyhow::Result<SendableRecordBatchStream> {
         Ok(self
             .send(SubscribeSinceMsg {
                 projection,
-                offset,
+                filters,
                 limit,
             })
             .await??)