Merge pull request #47 from caido/ef-fix-early-event

Fix an issue where fast events would get missed

Author: Sytten

.github/workflows/sdk-client-tests.yml

@@ -25,6 +25,7 @@ jobs:
         caido-version:
           - 0.56.0
           - 0.57.0
+          - 0.58.0
 
     services:
       caido:

packages/sdk-client/src/sdks/replay.spec.ts

@@ -0,0 +1,92 @@
+import { describe, expect, it, vi } from "vitest";
+import { makeSubject, toAsyncIterable } from "wonka";
+
+import { ReplaySDK } from "./replay.js";
+
+import type { GraphQLClient } from "@/graphql/index.js";
+import { Version } from "@/version.js";
+
+type FinishedEvent = {
+  finishedTask: {
+    task: {
+      __typename: "ReplayTask";
+      id: string;
+      createdAt: string;
+      replayEntry: { id: string };
+    };
+    status: "DONE";
+    error: null;
+  };
+};
+
+describe("ReplaySDK.send", () => {
+  it("resolves when the task finishes before the start mutation returns", async () => {
+    const subject = makeSubject<FinishedEvent>();
+
+    const task = {
+      __typename: "ReplayTask" as const,
+      id: "task-1",
+      createdAt: "2026-01-01T00:00:00.000Z",
+      replayEntry: { id: "entry-1" },
+    };
+
+    const graphql = {
+      subscribe: vi.fn(() => toAsyncIterable(subject.source)),
+      mutation: vi.fn(async () => {
+        // Fast-finishing task: finished event arrives before mutation resolves
+        subject.next({
+          finishedTask: {
+            task,
+            status: "DONE",
+            error: null,
+          },
+        });
+        subject.complete();
+
+        return {
+          startReplayTask: {
+            task,
+            error: null,
+          },
+        };
+      }),
+      query: vi.fn(async () => ({
+        replayEntry: {
+          id: "entry-1",
+          createdAt: "2026-01-01T00:00:00.000Z",
+          error: null,
+          raw: undefined,
+          connection: {
+            __typename: "ConnectionInfo",
+            host: "example.com",
+            port: 80,
+            isTLS: false,
+            SNI: null,
+          },
+          request: null,
+          session: { id: "session-1" },
+          settings: { placeholders: [] },
+        },
+      })),
+    } as unknown as GraphQLClient;
+
+    const replay = new ReplaySDK(graphql, Version.of("0.56.0"));
+
+    const result = await replay.send("session-1", {
+      raw: "GET / HTTP/1.1\r\nHost: example.com\r\n\r\n",
+      connection: {
+        host: "example.com",
+        port: 80,
+        isTLS: false,
+      },
+    });
+
+    expect(result.status).toBe("DONE");
+    expect(result.entry.id).toBe("entry-1");
+    expect(graphql.subscribe).toHaveBeenCalled();
+    expect(graphql.mutation).toHaveBeenCalled();
+    expect(
+      vi.mocked(graphql.subscribe).mock.invocationCallOrder[0]!,
+    ).toBeLessThan(vi.mocked(graphql.mutation).mock.invocationCallOrder[0]!);
+  });
+});

packages/sdk-client/src/sdks/replay.ts

@@ -20,6 +20,7 @@ import {
   type ReplaySendResult,
   TransportVersion,
 } from "@/types/index.js";
+import { bufferAsyncIterable } from "@/utils/asyncIterable.js";
 import { handleGraphQLError } from "@/utils/errors.js";
 import { isAbsent, isPresent } from "@/utils/optional.js";
 import type { Version } from "@/version.js";
@@ -143,19 +144,20 @@ export class ReplaySDK {
       });
     }
 
-    // Start the replay task
-    const result = await this.graphql.mutation(LatestStartReplayTaskDocument, {
-      sessionId: sessionId as string,
-    });
-
-    const payload = result.startReplayTask;
-    if (isPresent(payload.error)) {
-      handleGraphQLError(payload.error);
-    }
-    const task = new ReplayTask(this.graphql, payload.task!);
+    return this.waitForReplayTask(async () => {
+      const result = await this.graphql.mutation(
+        LatestStartReplayTaskDocument,
+        {
+          sessionId: sessionId as string,
+        },
+      );
 
-    // Wait for task to finish
-    return this.waitForReplayTask(task);
+      const payload = result.startReplayTask;
+      if (isPresent(payload.error)) {
+        handleGraphQLError(payload.error);
+      }
+      return new ReplayTask(this.graphql, payload.task!);
+    });
   }
 
   private async sendV056(
@@ -183,25 +185,35 @@ export class ReplaySDK {
       },
     } satisfies StartReplayTaskInput;
 
-    const result = await this.graphql.mutation(V056StartReplayTaskDocument, {
-      sessionId: sessionId as string,
-      input,
+    return this.waitForReplayTask(async () => {
+      const result = await this.graphql.mutation(V056StartReplayTaskDocument, {
+        sessionId: sessionId as string,
+        input,
+      });
+
+      const payload = result.startReplayTask;
+      if (isPresent(payload.error)) {
+        handleGraphQLError(payload.error);
+      }
+      return new ReplayTask(this.graphql, payload.task!);
     });
+  }
 
-    const payload = result.startReplayTask;
-    if (isPresent(payload.error)) {
-      handleGraphQLError(payload.error);
-    }
-    const task = new ReplayTask(this.graphql, payload.task!);
+  /**
+   * Open the finished-task subscription before running `start`, then resolve
+   * when the started task finishes.
+   */
+  private async waitForReplayTask(
+    start: () => Promise<ReplayTask>,
+  ): Promise<ReplaySendResult> {
+    const finished = bufferAsyncIterable(this.tasks.finished());
+    const task = await start();
 
-    // Wait for task to finish
-    return this.waitForReplayTask(task);
-  }
+    for await (const result of finished) {
+      if (result.task.id !== task.id) {
+        continue;
+      }
 
-  private async waitForReplayTask(task: ReplayTask): Promise<ReplaySendResult> {
-    for await (const result of this.tasks.finished(
-      (finished) => finished.task.id === task.id,
-    )) {
       const entry = await this.entries.get(task.replayEntryId);
       if (isAbsent(entry)) {
         throw new OtherUserError("INTERNAL", "Replay entry not found");

packages/sdk-client/src/sdks/task.ts

@@ -42,21 +42,23 @@ export class TaskSDK {
   }
 
   finished(
-    filter: (taskResult: TaskResult) => boolean,
+    filter?: (taskResult: TaskResult) => boolean,
   ): AsyncIterable<TaskResult> {
-    return filterAsyncIterable(
-      filter,
-      mapAsyncIterable((event) => {
-        const task = new Task(this.graphql, event.finishedTask.task);
-        return {
-          task,
-          status: event.finishedTask.status as TaskStatus,
-          error: event.finishedTask.error
-            ? { code: event.finishedTask.error.code }
-            : undefined,
-        };
-      }, this.graphql.subscribe(FinishedTaskDocument)),
-    );
+    const results = mapAsyncIterable((event) => {
+      const task = new Task(this.graphql, event.finishedTask.task);
+      return {
+        task,
+        status: event.finishedTask.status as TaskStatus,
+        error: event.finishedTask.error
+          ? { code: event.finishedTask.error.code }
+          : undefined,
+      };
+    }, this.graphql.subscribe(FinishedTaskDocument));
+
+    if (filter === undefined) {
+      return results;
+    }
+    return filterAsyncIterable(filter, results);
   }
 }
 

packages/sdk-client/src/utils/asyncIterable.ts

@@ -18,3 +18,24 @@ export async function* filterAsyncIterable<T>(
     }
   }
 }
+
+export function bufferAsyncIterable<T>(
+  source: AsyncIterable<T>,
+): AsyncIterable<T> {
+  const iterator = source[Symbol.asyncIterator]();
+  const first = iterator.next();
+
+  return {
+    async *[Symbol.asyncIterator]() {
+      try {
+        let result = await first;
+        while (result.done !== true) {
+          yield result.value;
+          result = await iterator.next();
+        }
+      } finally {
+        await iterator.return?.(undefined);
+      }
+    },
+  };
+}