Merge pull request #228 from superfly/tx-error

Stop processing changeset on sqlite error
superfly · Jul 2, 2024 · b9e3a0b · b9e3a0b
2 parents dfeec73 + 054834a
commit b9e3a0b
Showing 1 changed file with 15 additions and 21 deletions.
diff --git a/crates/corro-agent/src/agent/util.rs b/crates/corro-agent/src/agent/util.rs
@@ -782,27 +782,21 @@ pub async fn process_multiple_changes(
                         }
                     }
 
-                    let (known, versions) = match process_single_version(
-                        &agent,
-                        &tx,
-                        last_db_version,
-                        change,
-                    ) {
-                        Ok((known, changeset)) => {
-                            let versions = changeset.versions();
-                            if let KnownDbVersion::Current(CurrentVersion { db_version, .. }) =
-                                &known
-                            {
-                                last_db_version = Some(*db_version);
-                                changesets.push((actor_id, changeset, *db_version, src));
-                            }
-                            (known, versions)
-                        }
-                        Err(e) => {
-                            error!(%actor_id, ?versions, "could not process single change: {e}");
-                            continue;
-                        }
-                    };
+                    let (known, changeset) =
+                        process_single_version(&agent, &tx, last_db_version, change).map_err(
+                            |source| ChangeError::Rusqlite {
+                                source,
+                                actor_id: Some(actor_id),
+                                version: None,
+                            },
+                        )?;
+
+                    let versions = changeset.versions();
+                    if let KnownDbVersion::Current(CurrentVersion { db_version, .. }) = &known {
+                        last_db_version = Some(*db_version);
+                        changesets.push((actor_id, changeset, *db_version, src));
+                    }
+
                     debug!(%actor_id, self_actor_id = %agent.actor_id(), ?versions, "got known to insert: {known:?}");
                     known
                 };