fix: aggregator restart missing events (#737)

* chore: deal with rustc 1.89 mismatched_lifetime_syntaxes warnings

* fix: aggregator idempotency

restart from the beginning of conclusion events to avoid skipping on restarts. need to figure out why memory table contains conclusion_event_order that is less than the data written to event_states still, and hopefully restart where we left off. for now, we just ignore things we've seen before.

* fix: ensure conclusion event batch is written flushed from cache as a whole

we split the batch into models/mids and were writing both pieces to disk separately. The entire batch is orded by conclusion_event_order, but each group could overlap. So we could flush the cache for the first write and then we'd have a cache that was "going backward" in conclusion_event_order. On restart, those events were missed and events failed to aggregate correctly

* test: fix comment and modify test to better replicate ordering bug

* chore(ci): add fmt/clippy

* chore(ci): use newer jq image

* chore(ci): use alpine image for jq

* chore(ci): install bash in init container

---------

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

Author: dav1do

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

@@ -101,6 +101,8 @@ jobs:
       with:
         package_json_file: tests/suite/package.json
     - uses: dtolnay/rust-toolchain@stable
+      with:
+        components: rustfmt, clippy
     - uses: Swatinem/rust-cache@v2
       with:
         # The prefix cache key, this can be changed to start a new cache manually.

Cargo.lock

@@ -57,7 +57,7 @@ impl MerkleNodes {
     }
 
     /// Return an iterator over the MerkleNodes
-    pub fn iter(&self) -> indexmap::map::Iter<Cid, MerkleNode> {
+    pub fn iter(&self) -> indexmap::map::Iter<'_, Cid, MerkleNode> {
         self.nodes.iter()
     }
 

anchor-service/src/anchor.rs

@@ -224,7 +224,7 @@ impl Builder<WithExpiration> {
         PeerKey(format!("{:0>11}", self.state.expiration))
     }
     /// Set the peer id. Note, a NodeKey is required so the [`PeerEntry`] can be signed.
-    pub fn with_id(self, id: &NodeKey) -> Builder<WithId> {
+    pub fn with_id(self, id: &NodeKey) -> Builder<WithId<'_>> {
         Builder {
             state: WithId {
                 node_key: id,

core/src/peer.rs

@@ -397,7 +397,7 @@ async fn event_states_simple() -> Result<()> {
                 data::varchar as data,
                 before,
                 chain_id
-            FROM event_states LIMIT 4"#
+            FROM event_states order by event_state_order LIMIT 4"#
                 .to_string(),
             None,
         )

flight/tests/server.rs

@@ -52,3 +52,5 @@ url.workspace = true
 mockall.workspace = true
 test-log.workspace = true
 tracing-subscriber.workspace = true
+rand.workspace = true
+multihash-codetable.workspace = true

pipeline/Cargo.toml

@@ -226,6 +226,7 @@ impl PartitionEvaluator for CeramicPatchEvaluator {
                             }
                             Err(err) => {
                                 warn!(%err, event_cid=?Cid::read_bytes(event_cids.value(i)), "failed to apply patch to model instance event");
+                                tracing::debug!(%previous_height, %num_rows, patch=?String::from_utf8_lossy(patches.value(i)), previous_state=?String::from_utf8_lossy(previous_state), "failed to apply patch to model instance event");
                                 new_states.append_null();
                                 model_versions.append_null();
                             }

pipeline/src/aggregator/mod.rs

@@ -176,6 +176,7 @@ impl PartitionEvaluator for CeramicPatchEvaluator {
                             }
                             Err(err) => {
                                 warn!(%err, event_cid=?Cid::read_bytes(event_cids.value(i)), "failed to apply patch to model event");
+                                tracing::debug!(%previous_height, %num_rows, patch=?String::from_utf8_lossy(patches.value(i)), previous_state=?String::from_utf8_lossy(previous_state), "failed to apply patch to model event");
                                 new_states.append_null();
                             }
                         };

pipeline/src/aggregator/model_instance_patch.rs

@@ -192,7 +192,7 @@ impl SqlitePool {
 
     /// Begin a transaction. The transaction must be committed by calling `commit_tx`.
     /// Will be rolled back on drop if not committed.
-    pub async fn begin_tx(&self) -> Result<SqliteTransaction> {
+    pub async fn begin_tx(&self) -> Result<SqliteTransaction<'_>> {
         let tx = self.writer.begin().await?;
         Ok(SqliteTransaction { tx })
     }