fix: stream order (#411)

Author: thantos

apps/tests/aws-runtime/test/test-service.ts

@@ -607,14 +607,20 @@ export const counterCollectionOrderByN = counterCollection.index(
 const entityEvent = event<{ id: string }>("entityEvent");
 const entitySignal = signal("entitySignal");
 const entitySignal2 = signal<{ n: number }>("entitySignal2");
+const entityOrderSignal = signal<{ n: number }>("entityOrderSignal");
 
 export const counterWatcher = counter.stream(
   "counterWatcher",
-  { operations: ["remove"], includeOld: true },
+  { operations: ["modify", "remove"], includeOld: true },
   async (item) => {
     console.log(item);
-    const { n } = item.oldValue!;
-    await entitySignal2.sendSignal(item.key.id, { n: n + 1 });
+    if (item.operation === "remove") {
+      const { n } = item.oldValue!;
+      await entitySignal2.sendSignal(item.key.id, { n: n + 1 });
+    } else if (item.newValue.namespace === "default") {
+      const { n } = item.newValue;
+      await entityOrderSignal.sendSignal(item.key.id, { n });
+    }
   }
 );
 
@@ -658,12 +664,31 @@ export const entityTask = task(
   "entityTask",
   async (_, { execution: { id } }) => {
     const value = await counter.get(["default", id]);
+    // send 4 updates to force multiple update events at the same time and test stream ordering
     await counter.put({
       namespace: "default",
       id,
       n: (value?.n ?? 0) + 1,
       optional: undefined,
     });
+    await counter.put({
+      namespace: "default",
+      id,
+      n: (value?.n ?? 0) + 2,
+      optional: undefined,
+    });
+    await counter.put({
+      namespace: "default",
+      id,
+      n: (value?.n ?? 0) + 3,
+      optional: undefined,
+    });
+    await counter.put({
+      namespace: "default",
+      id,
+      n: (value?.n ?? 0) + 4,
+      optional: undefined,
+    });
   }
 );
 
@@ -748,15 +773,34 @@ export const entityIndexTask = task(
 export const entityWorkflow = workflow(
   "entityWorkflow",
   async (_, { execution: { id } }) => {
+    let orderValue = 0;
+    let orderError: string | undefined;
+    // this signal handler is intended to test that the counter stream is executed in order.
+    // it sends a signal each time the "default" name space counter is updated
+    // and we want to check if the order is always increasing
+    // and then validate the order value against the final value at the end.
+    entityOrderSignal.onSignal(({ n }) => {
+      if (n > orderValue) {
+        console.log(`Ordered Value: ${orderValue} ${n}`);
+        orderValue = n;
+      } else {
+        orderError = `order value ${n} is less than or equal to previous value ${orderValue}, the stream handler executed out of order!`;
+      }
+    });
+
+    // default: 1
     await counter.put({ namespace: "default", id, n: 1, optional: undefined });
     await counter.put({
       namespace: "different",
       id,
       n: 1,
       optional: undefined,
     });
+    // default: 2 after insert "different"
     await entitySignal.expectSignal();
+    // default: 6 after entity task (adds 4)
     await entityTask();
+    // default: 7 after event subscription
     await Promise.all([entityEvent.emit({ id }), entitySignal.expectSignal()]);
     try {
       // will fail
@@ -769,6 +813,7 @@ export const entityWorkflow = workflow(
     }
     const { value: entityValue, version } =
       (await counter.getWithMetadata(["default", id])) ?? {};
+    // default: 8 after get with metadata and successful assertion
     await counter.put(
       { namespace: "default", id, n: entityValue!.n + 1, optional: undefined },
       { expectedVersion: version }
@@ -782,6 +827,9 @@ export const entityWorkflow = workflow(
       },
     ]);
 
+    // default: 9 after transact write
+    const finalValue = await counter.get(["default", id]);
+
     const result0 = await entityIndexTask();
 
     // send deletion, to be picked up by the stream
@@ -793,6 +841,22 @@ export const entityWorkflow = workflow(
       entitySignal2.expectSignal(),
     ]);
 
+    if (
+      orderError ||
+      // wait 30 seconds for the order value to match the final value.
+      !(await condition(
+        { timeout: duration(300, "seconds") },
+        () => finalValue?.n === orderValue
+      ))
+    ) {
+      if (orderError) {
+        throw new Error(orderError);
+      }
+      throw new Error(
+        `Order handler never received the final value! have: ${orderValue} expected: ${finalValue?.n}`
+      );
+    }
+
     /**
      * Testing sort keys and query
      */

apps/tests/aws-runtime/test/tester.test.ts

@@ -132,7 +132,7 @@ eventualRuntimeTestHarness(
           },
           {
             namespace: "default",
-            n: 6,
+            n: 9,
           },
           {
             namespace: "another2",
@@ -150,7 +150,7 @@ eventualRuntimeTestHarness(
           },
           {
             namespace: "default",
-            n: 6,
+            n: 9,
           },
           {
             namespace: "another",
@@ -168,7 +168,7 @@ eventualRuntimeTestHarness(
           },
           {
             namespace: "default",
-            n: 6,
+            n: 9,
           },
           {
             namespace: "different",
@@ -184,13 +184,13 @@ eventualRuntimeTestHarness(
         betweenN: [
           {
             namespace: "default",
-            n: 6,
+            n: 9,
           },
         ],
         greaterThanN: [
           {
             namespace: "default",
-            n: 6,
+            n: 9,
           },
           {
             namespace: "another",
@@ -204,7 +204,7 @@ eventualRuntimeTestHarness(
           },
           {
             namespace: "default",
-            n: 6,
+            n: 9,
           },
         ],
         sparse: [
@@ -224,7 +224,7 @@ eventualRuntimeTestHarness(
           },
         ],
       },
-      { n: 7 },
+      { n: 10 },
       [
         { counterNumber: 1, n: 1 },
         { counterNumber: 2, n: 2 },

packages/@eventual/core-runtime/src/handlers/entity-stream-worker.ts

@@ -1,6 +1,7 @@
 import type { EntityStreamContext, EntityStreamItem } from "@eventual/core";
 import { ServiceType, getEventualResource } from "@eventual/core/internal";
 import { serviceTypeScope } from "../service-type.js";
+import { normalizeCompositeKey } from "../stores/entity-store.js";
 import { getLazy, promiseAllSettledPartitioned } from "../utils.js";
 import { registerWorkerIntrinsics, type WorkerIntrinsicDeps } from "./utils.js";
 
@@ -23,12 +24,10 @@ export function createEntityStreamWorker(
 
   return async (entityName, streamName, items) =>
     serviceTypeScope(ServiceType.EntityStreamWorker, async () => {
-      const streamHandler = getEventualResource(
-        "Entity",
-        entityName
-      )?.streams.find((s) => s.name === streamName);
+      const entity = getEventualResource("Entity", entityName);
+      const streamHandler = entity?.streams.find((s) => s.name === streamName);
 
-      if (!streamHandler) {
+      if (!entity || !streamHandler) {
         throw new Error(`Stream handler ${streamName} does not exist`);
       }
 
@@ -45,20 +44,34 @@ export function createEntityStreamWorker(
 
         return { failedItemIds: result?.failedItemIds ?? [] };
       } else {
+        const itemsByKey: Record<string, EntityStreamItem<any>[]> = {};
+        items.forEach((item) => {
+          const normalizedKey = normalizeCompositeKey(entity, item.key);
+          const serializedKey = JSON.stringify(normalizedKey);
+          (itemsByKey[serializedKey] ??= []).push(item);
+        });
+
         const results = await promiseAllSettledPartitioned(
-          items,
-          async (item) => {
-            return await streamHandler.handler(item, context);
+          Object.entries(itemsByKey),
+          async ([, itemGroup]) => {
+            for (const i in itemGroup) {
+              const item = itemGroup[i]!;
+              try {
+                const result = await streamHandler.handler(item, context);
+                // if the handler doesn't fail and doesn't return false, continue
+                if (result !== false) {
+                  continue;
+                }
+              } catch {}
+              // if the handler fails or returns false, return the rest of the items
+              return itemGroup.slice(Number(i)).map((i) => i.id);
+            }
+            return [];
           }
         );
 
         return {
-          failedItemIds: [
-            ...results.rejected.map(([e]) => e.id),
-            ...results.fulfilled
-              .filter(([, r]) => r === false)
-              .map(([e]) => e.id),
-          ],
+          failedItemIds: results.fulfilled.flatMap((s) => s[1]),
         };
       }
     });