fix: aggregator restart missing events

.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.

anchor-service/src/anchor.rs

@@ -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()
     }
 

core/src/peer.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,

flight/tests/server.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,
         )

pipeline/Cargo.toml

@@ -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/src/aggregator/model_instance_patch.rs

@@ -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/model_patch.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();
                             }
                         };

sql/src/sqlite.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 })
     }

tests/hermetic/src/cli/tester.rs

@@ -786,12 +786,12 @@ fn job_volumes() -> Vec<Volume> {
 fn job_init() -> Container {
     Container {
         name: "init-tests".to_owned(),
-        image: Some("stedolan/jq".to_owned()),
+        image: Some("alpine:latest".to_owned()),
         image_pull_policy: Some("IfNotPresent".to_owned()),
         command: Some(vec![
-            "/bin/bash".to_owned(),
+            "/bin/sh".to_owned(),
             "-c".to_owned(),
-            "/network/process-peers.sh".to_owned(),
+            "apk add --no-cache bash jq curl && /network/process-peers.sh".to_owned(),
         ]),
         env: Some(vec![
             EnvVar {