fix: random things like feed queries, query CLI, migration logging, deps (#708)

* chore: update flight sql deps

* fix: check migration model filter earlier

this avoids lots of errors e.g. "fatal error: finding capability block" for events we don't want to review

* fix: adjust query CLI to display streaming results

querying the _feed tables without a limit will now print all the data retrieved and then wait for updates. previously it would just hang doing nothing.

* test: modify tests so that events pass validation

* fix: index -> conclusion_event_order

this column was renamed but the predicate was not correctly updated so the feed queries always started from the beginning

* fix: revert predicate pushdown change

* fix: update ordering and push down filters

we now pass predicates down into the pipeline so that _feed queries work better.
if conclusion_event_order is included in the predicate, we will skip reading all events
and start from the offset (we could potentially push the expr down but it's more involved).
in general, WHERE clause predicates should be respected by _feed queries now. I had to adjust
the ordering used as we were conflating conclusion_event_order and event_state_order, so we now
use the conclusion_event_order as the offset, and apply predicates afterward to filter the stream
data returned. afaict, the adjustment to use >= instead of > is required now, and doesn't cause
issues as we expect we may get the same event in both streams and reduce to one correctly.

* chore: point to right sdk docs

* fix: improve stream/feed queries to support filters

previously, we only supported highwater mark and tried to filter after
the fact. now, we create two streams, the first using the filters to
find all existing events, the second is all new events that are filtered
before being pushed to the consumers.

* chore: Option vec instead of allocating empty vec for filters

* chore: fix clippy lints

* fix: return error instead of simply logging when expression can't be evaluated

* chore: update sccache version

* fix: use SessionContext in feed to use execution props

* chore: clippy

* fix: remove TODO

---------

Co-authored-by: Mohsin Zaidi <mohsinrzaidi@gmail.com>

Author: dav1do

.github/workflows/build-and-test.yml

@@ -24,7 +24,7 @@ jobs:
         prefix-key: v0
         # Cache only the cargo registry
         cache-targets: false
-    - uses: mozilla-actions/sccache-action@v0.0.3
+    - uses: mozilla-actions/sccache-action@v0.0.9
     - name: git file permission config
       run: git config --global --add safe.directory '*'
     - name: Check fmt
@@ -52,7 +52,7 @@ jobs:
         prefix-key: v0
         # Cache only the cargo registry
         cache-targets: false
-    - uses: mozilla-actions/sccache-action@v0.0.3
+    - uses: mozilla-actions/sccache-action@v0.0.9
     - name: Run tests
       run: make test
   build:
@@ -69,7 +69,7 @@ jobs:
         prefix-key: v0
         # Cache only the cargo registry
         cache-targets: false
-    - uses: mozilla-actions/sccache-action@v0.0.3
+    - uses: mozilla-actions/sccache-action@v0.0.9
     -
       name: Set up Docker Buildx
       uses: docker/setup-buildx-action@v2
@@ -108,7 +108,7 @@ jobs:
         prefix-key: v0
         # Cache only the cargo registry
         cache-targets: false
-    - uses: mozilla-actions/sccache-action@v0.0.3
+    - uses: mozilla-actions/sccache-action@v0.0.9
     - name: Check fmt
       run: make -C tests check-fmt
     - name: Check clippy
@@ -171,7 +171,7 @@ jobs:
         prefix-key: v0
         # Cache only the cargo registry
         cache-targets: false
-    - uses: mozilla-actions/sccache-action@v0.0.3
+    - uses: mozilla-actions/sccache-action@v0.0.9
     - name: Build Tester
       run: make -C tests BUILD_PROFILE=release driver
     - uses: actions/upload-artifact@master

Cargo.lock

@@ -269,9 +269,3 @@ repository = "https://github.com/3box/rust-ceramic"
 inherits = "release"
 debug = true
 strip = "none"
-
-[patch."https://github.com/datafusion-contrib/datafusion-flight-sql-server.git"]
-datafusion-flight-sql-server = { git = "https://github.com/nathanielc/datafusion-flight-sql-server.git", branch = "chore/datafusion-45" }
-
-[patch.crates-io]
-datafusion-federation = { git = "https://github.com/nathanielc/datafusion-federation.git", branch = "chore/datafusion-45" }

Cargo.toml

@@ -14,7 +14,7 @@ use recon::ReconItem;
 use serde::Deserialize;
 use thiserror::Error;
 use tokio::{fs::File, io::AsyncWriteExt as _};
-use tracing::{debug, error, info, instrument, Level};
+use tracing::{debug, error, info, instrument, trace, Level};
 
 use crate::{
     event::{BlockStore, DeliverableRequirement},
@@ -217,6 +217,13 @@ impl<'a, S: BlockStore> Migrator<'a, S> {
                 payload.header().controllers()[0].clone(),
                 cid,
             );
+            if self.is_filtered_out(&event_builder) {
+                trace!(
+                    event_cid=%event_builder.event_cid,
+                    "skipping unreferenced init payload due to model filter"
+                );
+                return Ok(());
+            }
             let event = unvalidated::init::Event::new(payload);
             let event: unvalidated::Event<Ipld> = unvalidated::Event::from(Box::new(event));
             self.validate_build_and_push(event_builder, event, &model)
@@ -284,6 +291,11 @@ impl<'a, S: BlockStore> Migrator<'a, S> {
             }
         };
 
+        if self.is_filtered_out(&event_builder) {
+            trace!(event_cid=%event_builder.event_cid, "skipping signed event due to model filter");
+            return Ok(());
+        }
+
         let model = ModelContext::from(event_builder.sep.as_slice());
 
         let mut capability = None;
@@ -370,6 +382,12 @@ impl<'a, S: BlockStore> Migrator<'a, S> {
             unvalidated::RawEvent::Unsigned(payload) => Ok(payload),
         }
     }
+
+    /// Returns true if the event should be skipped due to the model (sep) filter included
+    fn is_filtered_out(&self, event_builder: &EventBuilder) -> bool {
+        !self.sep_filter.is_empty() && !self.sep_filter.contains(&event_builder.sep)
+    }
+
     // Find and add all blocks related to this time event
     #[instrument(skip(self, event), ret(level = Level::DEBUG))]
     async fn process_time_event(
@@ -386,6 +404,10 @@ impl<'a, S: BlockStore> Migrator<'a, S> {
             init_payload.header().controllers()[0].clone(),
             init,
         );
+        if self.is_filtered_out(&event_builder) {
+            trace!(event_cid=%event_builder.event_cid, "skipping time event due to model filter");
+            return Ok(());
+        }
         let proof_id = event.proof();
         let data = self
             .load_block(&proof_id)
@@ -440,40 +462,38 @@ impl<'a, S: BlockStore> Migrator<'a, S> {
         event: unvalidated::Event<Ipld>,
         model: &ModelContext,
     ) -> Result<()> {
-        if self.sep_filter.is_empty() || self.sep_filter.contains(&event_builder.sep) {
-            match &event {
-                unvalidated::Event::Time(event) => {
-                    if let Some(supported_chains) = &self.supported_chains {
-                        let chain_id = event.proof().chain_id().to_owned();
-                        if !supported_chains.contains(&chain_id) {
-                            return Err(anyhow!("event has unsupported chain: {chain_id}"))
-                                .with_model_context(model);
-                        }
+        match &event {
+            unvalidated::Event::Time(event) => {
+                if let Some(supported_chains) = &self.supported_chains {
+                    let chain_id = event.proof().chain_id().to_owned();
+                    if !supported_chains.contains(&chain_id) {
+                        return Err(anyhow!("event has unsupported chain: {chain_id}"))
+                            .with_model_context(model);
                     }
                 }
-                unvalidated::Event::Signed(event) => {
-                    if self.validate_signatures {
-                        event
-                            .verify_signature(
-                                Some(&event_builder.controller),
-                                &VerifyJwsOpts {
-                                    at_time: AtTime::SkipTimeChecks,
-                                    ..Default::default()
-                                },
-                            )
-                            .await
-                            .with_model_context(model)?;
-                    }
+            }
+            unvalidated::Event::Signed(event) => {
+                if self.validate_signatures {
+                    event
+                        .verify_signature(
+                            Some(&event_builder.controller),
+                            &VerifyJwsOpts {
+                                at_time: AtTime::SkipTimeChecks,
+                                ..Default::default()
+                            },
+                        )
+                        .await
+                        .with_model_context(model)?;
                 }
-                unvalidated::Event::Unsigned(_) => {}
-            };
-            self.batch.push(
-                event_builder
-                    .build(&self.network, event)
-                    .await
-                    .with_model_context(model)?,
-            );
-        }
+            }
+            unvalidated::Event::Unsigned(_) => {}
+        };
+        self.batch.push(
+            event_builder
+                .build(&self.network, event)
+                .await
+                .with_model_context(model)?,
+        );
         Ok(())
     }
 }

event-svc/src/event/migration.rs

@@ -199,7 +199,7 @@ impl EventAccess {
         I: Iterator<Item = &'a EventInsertable>,
     {
         let mut inserted = Vec::new();
-        let mut tx = self.pool.begin_tx().await.map_err(Error::from)?;
+        let mut tx = self.pool.begin_tx().await?;
 
         for item in to_add {
             let new_key = self
@@ -224,7 +224,7 @@ impl EventAccess {
                 self.mark_ready_to_deliver(&mut tx, item.cid()).await?;
             }
         }
-        tx.commit().await.map_err(Error::from)?;
+        tx.commit().await?;
         let res = InsertResult::new(inserted);
 
         Ok(res)

event-svc/src/store/sql/access/event.rs

@@ -149,8 +149,7 @@ impl EventInsertable {
         let mut idx = 0;
         let mut blocks = vec![];
         while let Some((cid, data)) = reader.next_block().await.map_err(Error::new_app)? {
-            let ebr = EventBlockRaw::try_new(&self.cid, idx, roots.contains(&cid), cid, data)
-                .map_err(Error::from)?;
+            let ebr = EventBlockRaw::try_new(&self.cid, idx, roots.contains(&cid), cid, data)?;
             blocks.push(ebr);
             idx += 1;
         }

event-svc/src/store/sql/entities/event.rs

@@ -140,7 +140,7 @@ fn events(start_index: u64, highwater_mark: u64, limit: usize) -> Vec<Conclusion
     };
     vec![
         ConclusionEvent::Data(ConclusionData {
-            order: start_index + 0,
+            order: start_index,
             event_cid: model_stream_id.cid,
             init: ConclusionInit {
                 stream_cid: model_stream_id.cid,
@@ -364,7 +364,7 @@ async fn event_states_simple() -> Result<()> {
         let mut sub = aggregator
             .send(SubscribeSinceMsg {
                 projection: None,
-                offset: None,
+                filters: None,
                 // We know there are only four events, query all of them
                 limit: Some(4),
             })

flight/tests/server.rs

@@ -57,9 +57,9 @@ impl CeramicOneInterest {
 impl CeramicOneInterest {
     /// Insert a single interest into the database.
     pub async fn insert(pool: &SqlitePool, key: &Interest) -> Result<bool> {
-        let mut tx = pool.begin_tx().await.map_err(Error::from)?;
+        let mut tx = pool.begin_tx().await?;
         let new_key = CeramicOneInterest::insert_tx(&mut tx, key).await?;
-        tx.commit().await.map_err(Error::from)?;
+        tx.commit().await?;
         Ok(new_key)
     }
 
@@ -72,14 +72,14 @@ impl CeramicOneInterest {
             0 => Ok(InsertResult::new(0)),
             _ => {
                 let mut results = 0;
-                let mut tx = pool.begin_tx().await.map_err(Error::from)?;
+                let mut tx = pool.begin_tx().await?;
 
                 for item in items.iter() {
                     CeramicOneInterest::insert_tx(&mut tx, item)
                         .await?
                         .then(|| results += 1);
                 }
-                tx.commit().await.map_err(Error::from)?;
+                tx.commit().await?;
                 Ok(InsertResult::new(results))
             }
         }

interest-svc/src/store/sql/access/interest.rs

@@ -293,7 +293,7 @@ async fn get_eth_rpc_providers(
                 let provider_chain = provider.chain_id();
                 if network
                     .supported_chain_ids()
-                    .map_or(true, |ids| ids.contains(provider_chain))
+                    .is_none_or(|ids| ids.contains(provider_chain))
                 {
                     info!(
                         "Using ethereum rpc provider for chain: {} with url: {}",