Initial sketch of streaming patches to the front end from edda

The idea with streaming patches to the front end is that it will be able
to start using the new MVs as soon as they are generated, before _all_
of the data is ready, and there will be a message with the MvIndex patch
(message not created/sent yet) that will let the front end know that the
stream of patches has finished, and which objects it should have by the
end (fetching any it missed the patch for).

Right now, these streaming patches are sent in addition to the existing
patch batches, but once we start using the streamed patches in the front
end, we'll want to stop creating & sending the batched patches as it
will be purely redundant from the streamed patches.

We will also want to figure out if the existing `IndexUpdate` message
should be augmented to include both the from-checksum & patch for the
MvIndex, or if we want an entirely new message to handle this.

Author: jhelwig

lib/edda-server/src/change_set_processor_task/materialized_view.rs

@@ -56,6 +56,7 @@ use si_frontend_mv_types::{
             IndexUpdate,
             ObjectPatch,
             PatchBatch,
+            StreamingPatch,
             UpdateMeta,
         },
     },
@@ -311,6 +312,7 @@ pub async fn build_all_mv_for_change_set(
         ctx,
         frigg,
         parallel_build_limit,
+        edda_updates,
         ctx.workspace_pk()?,
         ctx.change_set_id(),
         &changes,
@@ -427,6 +429,7 @@ pub async fn build_mv_for_changes_in_change_set(
         ctx,
         frigg,
         parallel_build_limit,
+        edda_updates,
         workspace_id,
         change_set_id,
         changes,
@@ -542,6 +545,7 @@ async fn build_mv_inner(
     ctx: &DalContext,
     frigg: &FriggStore,
     parallel_build_limit: usize,
+    edda_updates: &EddaUpdates,
     workspace_pk: si_id::WorkspacePk,
     change_set_id: ChangeSetId,
     changes: &[Change],
@@ -615,13 +619,30 @@ async fn build_mv_inner(
 
             match execution_result {
                 Ok((maybe_patch, maybe_frontend_object)) => {
+                    // We need to make sure the frontend object is inserted into the store first so that
+                    // a client can directly fetch it without racing against the object's insertion if the
+                    // client does not already have the base object to apply the streaming patch to.
+                    if let Some(frontend_object) = maybe_frontend_object {
+                        frigg.insert_object(workspace_pk, &frontend_object).await?;
+                        frontend_objects.push(frontend_object);
+                    }
                     if let Some(patch) = maybe_patch {
+                        let streaming_patch = StreamingPatch::new(
+                            workspace_pk,
+                            change_set_id,
+                            kind,
+                            mv_id,
+                            patch.from_checksum.clone(),
+                            patch.to_checksum.clone(),
+                            patch.patch.clone(),
+                        );
+                        edda_updates
+                            .publish_streaming_patch(streaming_patch)
+                            .await?;
+
                         debug!("Patch!: {:?}", patch);
                         patches.push(patch);
                     }
-                    if let Some(frontend_object) = maybe_frontend_object {
-                        frontend_objects.push(frontend_object);
-                    }
                     build_count += 1;
                     if build_duration > build_max_elapsed {
                         build_max_elapsed = build_duration;
@@ -636,10 +657,6 @@ async fn build_mv_inner(
         }
     }
 
-    frigg
-        .insert_objects(workspace_pk, frontend_objects.iter())
-        .await?;
-
     Ok((
         frontend_objects,
         patches,

lib/edda-server/src/updates.rs

@@ -15,6 +15,7 @@ use si_data_nats::{
 use si_frontend_mv_types::object::patch::{
     IndexUpdate,
     PatchBatch,
+    StreamingPatch,
 };
 use si_id::WorkspacePk;
 use telemetry::prelude::*;
@@ -79,6 +80,30 @@ impl EddaUpdates {
         .await
     }
 
+    #[instrument(
+        name = "edda_updates.publish_streaming_patch",
+        level = "debug",
+        skip_all,
+        fields()
+    )]
+    pub(crate) async fn publish_streaming_patch(
+        &self,
+        streaming_patch: StreamingPatch,
+    ) -> Result<()> {
+        let mut id_buf = WorkspacePk::array_to_str_buf();
+
+        self.serialize_compress_publish(
+            subject::update_for(
+                self.subject_prefix.as_deref(),
+                streaming_patch.workspace_id.array_to_str(&mut id_buf),
+                streaming_patch.message_kind(),
+            ),
+            streaming_patch,
+            true,
+        )
+        .await
+    }
+
     #[instrument(
         name = "edda_updates.publish_index_update",
         level = "debug",

lib/si-frontend-mv-types-rs/src/object/patch.rs

@@ -4,8 +4,11 @@ use si_id::{
     WorkspacePk,
 };
 
-const PATCH_BATCH_KIND: &str = "PatchMessage";
+use crate::reference::ReferenceKind;
+
 const INDEX_UPDATE_KIND: &str = "IndexUpdate";
+const PATCH_BATCH_KIND: &str = "PatchMessage";
+const STREAMING_PATCH_MESSAGE_KIND: &str = "StreamingPatch";
 
 #[derive(Debug, Clone, Serialize)]
 #[serde(rename_all = "camelCase")]
@@ -21,6 +24,63 @@ pub struct UpdateMeta {
     pub from_index_checksum: String,
 }
 
+#[derive(Debug, Clone, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct StreamingPatch {
+    /// The workspace this patch is targeting.
+    pub workspace_id: WorkspacePk,
+    /// The change set this patch is targeting.
+    pub change_set_id: ChangeSetId,
+    /// The MV kind this patch is targeting.
+    pub kind: String,
+    /// The ID of the object this patch is targeting.
+    pub id: String,
+    /// The original checksum of the object before this patch.
+    ///
+    /// Checksum of `"0"` means this is a new object that must be created.
+    pub from_checksum: String,
+    /// The new checksum of the object after this patch.
+    ///
+    /// Checksum of `"0"` means this is an existing object that must be removed
+    pub to_checksum: String,
+    /// The JSON patch to apply to the object.
+    ///
+    /// If neither of `from_checksum`, and `to_checksum` are all `0`, this field
+    /// contains the JSON Patch document to apply to the version of the object with
+    /// a checksum matching `from_checksum` that will result in a version with the
+    /// checksum matching `to_checksum`.
+    pub patch: json_patch::Patch,
+    /// The message kind for the front end.
+    message_kind: &'static str,
+}
+
+impl StreamingPatch {
+    pub fn new(
+        workspace_id: WorkspacePk,
+        change_set_id: ChangeSetId,
+        kind: ReferenceKind,
+        id: String,
+        from_checksum: String,
+        to_checksum: String,
+        patch: json_patch::Patch,
+    ) -> Self {
+        Self {
+            workspace_id,
+            change_set_id,
+            kind: kind.to_string(),
+            id,
+            from_checksum,
+            to_checksum,
+            patch,
+            message_kind: STREAMING_PATCH_MESSAGE_KIND,
+        }
+    }
+
+    pub fn message_kind(&self) -> &'static str {
+        self.message_kind
+    }
+}
+
 #[derive(Debug, Clone, Serialize)]
 #[serde(rename_all = "camelCase")]
 pub struct PatchBatch {