Stream patches to the front end from edda; try to reduce nats write load; priority queue for MV builds

[jhelwig]

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.

Try to reduce nats load by only inserting FrontEndObjects if they're not already in the KV store

This doesn't reduce the amount of information sent over the wrire to nats on an insert, but nats should be able to short-circuit and avoid doing any write in the case where the object has already been generated for any reason.

With the interface provided by the nats KV store, our options for only inserting if the key doesn't already exist looked to be:

• Send the (potentially) new thing over the wire and watching for a failure (create behaves this way).
• Try to get first (and receive the full value).
• Iterate over all of the keys to see if the key is already there.

Trying the insert using create and ignoring the "already exists" error kind seems to be the least-bad of the options.

Use a priority queue to determine order of MV builds in edda

When streaming patches out to the front end, it would be helpful for the front end's reactivity to be able to send lists early so it can display skeletons for individual items it hasn't gotten the patch information for yet.

Right now, there are two "priorities":

• Lists
• Details (everything else)

By default an MV will use the "Details" priority. If the build_priority annotation is provided for the MV struct definition, then it will use whatever is specified in the annotation.

This also re-structures how MV build tasks are spawned so that we can get everything queued before letting the priority queue handle things from there, to ensure the higher-priority items are always built first. Rather than having to explicitly check if the Change's entity_kind is the same as the MV's trigger_entity, this check has been abstracted away into the MaterializedViewInventoryItem and is checked during the initial queue population. This means that whenever we go to spawn an MV build task, we already know that the change is relevant for that specific MV kind.

Make streamed patches from edda a config setting instead of always on

Rather than always sending the patches for the front end both as individual patches when each is built, and as a single batch after all have been built, there is now a SI_EDDA__STREAMING_PATCHES setting to toggle between the two behaviors. The default behavior is to send the patches as a single batch, making the newer streaming behavior opt-in for development & testing.

[github-actions]

Dependency Review

✅ No vulnerabilities or OpenSSF Scorecard issues found.

Scanned Files

None

[jhelwig]

An example of one of the streamed patches on the websocket:

{"workspaceId":"01GWAJW3PQ0S1DVBN76XH9DBHV","changeSetId":"01JXN8KGNG2CK01ETFH33VBNJE","kind":"Component","id":"01JXN8MRMF6AZ50CTMXPKT741Y","fromChecksum":"616b523fb65078f0063ad308e6e7cf1b","toChecksum":"214531f044b7b66cc11976eb8fea6847","patch":[{"op":"replace","path":"/resourceDiff/current","value":"{\n  \"si\": {\n    \"name\": \"StackFleetAssociation\",\n    \"type\": \"component\",\n    \"color\": \"#FF9900\"\n  },\n  \"domain\": {\n    \"extra\": {\n      \"Region\": \"us-east-1\",\n      \"AwsResourceType\": \"AWS::AppStream::StackFleetAssociation\",\n      \"PropUsageMap\": \"{\\\"createOnly\\\":[],\\\"updatable\\\":[\\\"StackName\\\",\\\"FleetName\\\"],\\\"secrets\\\":[]}\"\n    }\n  }\n}"},{"op":"replace","path":"/resourceDiff/diff","value":"+{\n+  \"si\": {\n+    \"name\": \"StackFleetAssociation\",\n+    \"type\": \"component\",\n+    \"color\": \"#FF9900\"\n+  },\n+  \"domain\": {\n+    \"extra\": {\n+      \"Region\": \"us-east-1\",\n+      \"AwsResourceType\": \"AWS::AppStream::StackFleetAssociation\",\n+      \"PropUsageMap\": \"{\\\"createOnly\\\":[],\\\"updatable\\\":[\\\"StackName\\\",\\\"FleetName\\\"],\\\"secrets\\\":[]}\"\n+    }\n+  }\n+}"}],"messageKind":"StreamingPatch"}

Human readable-ish version:

{
  "workspaceId": "01GWAJW3PQ0S1DVBN76XH9DBHV",
  "changeSetId": "01JXN8KGNG2CK01ETFH33VBNJE",
  "kind": "Component",
  "id": "01JXN8MRMF6AZ50CTMXPKT741Y",
  "fromChecksum": "616b523fb65078f0063ad308e6e7cf1b",
  "toChecksum": "214531f044b7b66cc11976eb8fea6847",
  "patch": [
    {
      "op": "replace",
      "path": "/resourceDiff/current",
      "value": "{\n  \"si\": {\n    \"name\": \"StackFleetAssociation\",\n    \"type\": \"component\",\n    \"color\": \"#FF9900\"\n  },\n  \"domain\": {\n    \"extra\": {\n      \"Region\": \"us-east-1\",\n      \"AwsResourceType\": \"AWS::AppStream::StackFleetAssociation\",\n      \"PropUsageMap\": \"{\\\"createOnly\\\":[],\\\"updatable\\\":[\\\"StackName\\\",\\\"FleetName\\\"],\\\"secrets\\\":[]}\"\n    }\n  }\n}"
    },
    {
      "op": "replace",
      "path": "/resourceDiff/diff",
      "value": "+{\n+  \"si\": {\n+    \"name\": \"StackFleetAssociation\",\n+    \"type\": \"component\",\n+    \"color\": \"#FF9900\"\n+  },\n+  \"domain\": {\n+    \"extra\": {\n+      \"Region\": \"us-east-1\",\n+      \"AwsResourceType\": \"AWS::AppStream::StackFleetAssociation\",\n+      \"PropUsageMap\": \"{\\\"createOnly\\\":[],\\\"updatable\\\":[\\\"StackName\\\",\\\"FleetName\\\"],\\\"secrets\\\":[]}\"\n+    }\n+  }\n+}"
    }
  ],
  "messageKind": "StreamingPatch"
}

[jobelenus]

@jhelwig thank you for the example payload. Is that id field the "placeholder" (e.g. that value will be in the "end" message for the index update with the resulting index checksum?)

[jhelwig]

The id in the sample payload is the ID of the atom/MV. At the time the streaming patch is sent, we don't yet know what the index checksum will be.

[jhelwig]

The "placeholder"/"staging" index checksum should probably be the change set id since we mostly treat the checksum as an opaque string, and it will never overlap with a "real" checksum.

[fnichol]

I like your approach to send both the incremental and the batch as we can migrate/experiment with both approaches.

Looks good!

[fnichol]

👍🏽