fix: aggregator restart missing events

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