Fix realtime re-subscribing stale data issue (#2162)

* Fix realtime re-subscribing stale data issue

Fixes an issue with realtime when re-subscribing to a run, that would temporarily display stale data and the changes. Now when re-subscribing to a run only the latest changes will be vended

* removed logs

Author: ericallam

.changeset/wicked-ads-walk.md

@@ -0,0 +1,6 @@
+---
+"@trigger.dev/react-hooks": patch
+"@trigger.dev/core": patch
+---
+
+Fixes an issue with realtime when re-subscribing to a run, that would temporarily display stale data and the changes. Now when re-subscribing to a run only the latest changes will be vended

packages/core/src/v3/apiClient/stream.ts

@@ -114,6 +114,8 @@ class ReadableShapeStream<T extends Row<unknown> = Row> {
       },
     });
 
+    let updatedKeys = new Set<string>();
+
     // Create the transformed stream that processes messages and emits complete rows
     this.#changeStream = createAsyncIterableStream(source, {
       transform: (messages, controller) => {
@@ -122,7 +124,7 @@ class ReadableShapeStream<T extends Row<unknown> = Row> {
         }
 
         try {
-          const updatedKeys = new Set<string>();
+          let isUpToDate = false;
 
           for (const message of messages) {
             if (isChangeMessage(message)) {
@@ -147,18 +149,23 @@ class ReadableShapeStream<T extends Row<unknown> = Row> {
               if (message.headers.control === "must-refetch") {
                 this.#currentState.clear();
                 this.#error = false;
+              } else if (message.headers.control === "up-to-date") {
+                isUpToDate = true;
               }
             }
           }
 
           // Now enqueue only one updated row per key, after all messages have been processed.
-          if (!this.#isStreamClosed) {
+          // If the stream is not up to date, we don't want to enqueue any rows.
+          if (!this.#isStreamClosed && isUpToDate) {
             for (const key of updatedKeys) {
               const finalRow = this.#currentState.get(key);
               if (finalRow) {
                 controller.enqueue(finalRow);
               }
             }
+
+            updatedKeys.clear();
           }
         } catch (error) {
           console.error("Error processing stream messages:", error);

references/hello-world/src/trigger/realtime.ts

@@ -1,5 +1,6 @@
 import { logger, runs, task } from "@trigger.dev/sdk";
 import { helloWorldTask } from "./example.js";
+import { setTimeout } from "timers/promises";
 
 export const realtimeByTagsTask = task({
   id: "realtime-by-tags",
@@ -32,3 +33,29 @@ export const realtimeByTagsTask = task({
     };
   },
 });
+
+export const realtimeUpToDateTask = task({
+  id: "realtime-up-to-date",
+  run: async ({ runId }: { runId?: string }) => {
+    if (!runId) {
+      const handle = await helloWorldTask.trigger(
+        { hello: "world" },
+        {
+          tags: ["hello-world", "realtime"],
+        }
+      );
+
+      runId = handle.id;
+    }
+
+    logger.info("runId", { runId });
+
+    for await (const run of runs.subscribeToRun(runId, { stopOnCompletion: true })) {
+      logger.info("run", { run });
+    }
+
+    return {
+      message: "Hello, world!",
+    };
+  },
+});