fix: make apply patch not fail the pipeline (#690)

With this change events with bad patches or data will just end up with
null states instead of preventing the entire pipeline from processing
more data.

Author: nathanielc

pipeline/src/aggregator/model_instance_patch.rs

@@ -17,6 +17,7 @@ use datafusion::{
     },
 };
 use json_patch::PatchOperation;
+use tracing::warn;
 
 use super::{bytes_value_at, u32_value_at, EventDataContainer};
 
@@ -218,9 +219,17 @@ impl PartitionEvaluator for CeramicPatchEvaluator {
                     } else {
                         let patch = patches.value(i);
                         resolved_patches.append_value(patch);
-                        let (data, model_version) = Self::apply_patch(patch, previous_state)?;
-                        new_states.append_value(data);
-                        model_versions.append_option(model_version.map(|mv| mv.to_bytes()));
+                        match Self::apply_patch(patch, previous_state) {
+                            Ok((data, model_version)) => {
+                                new_states.append_value(data);
+                                model_versions.append_option(model_version.map(|mv| mv.to_bytes()));
+                            }
+                            Err(err) => {
+                                warn!(%err, event_cid=?Cid::read_bytes(event_cids.value(i)), "failed to apply patch to model instance event");
+                                new_states.append_null();
+                                model_versions.append_null();
+                            }
+                        };
                     }
                 } else {
                     // Unreachable when data is well formed.

pipeline/src/aggregator/model_patch.rs

@@ -5,6 +5,7 @@ use arrow::{
     datatypes::DataType,
 };
 use arrow_schema::Field;
+use cid::Cid;
 use datafusion::{
     common::{
         cast::{as_binary_array, as_uint32_array},
@@ -16,6 +17,7 @@ use datafusion::{
     },
 };
 use json_patch::PatchOperation;
+use tracing::warn;
 
 use super::{bytes_value_at, u32_value_at, EventDataContainer};
 
@@ -168,10 +170,15 @@ impl PartitionEvaluator for CeramicPatchEvaluator {
                         resolved_previous_states.append_value(&state);
                     } else {
                         resolved_previous_states.append_value(previous_state);
-                        new_states.append_value(CeramicPatchEvaluator::apply_patch(
-                            patches.value(i),
-                            previous_state,
-                        )?);
+                        match Self::apply_patch(patches.value(i), previous_state) {
+                            Ok(data) => {
+                                new_states.append_value(data);
+                            }
+                            Err(err) => {
+                                warn!(%err, event_cid=?Cid::read_bytes(event_cids.value(i)), "failed to apply patch to model event");
+                                new_states.append_null();
+                            }
+                        };
                     }
                 } else {
                     // Unreachable when data is well formed.