feat(sdk): add iterator resume on transient errors

Author: zubeyralmaho

docs/GETTING_STARTED.md

@@ -96,7 +96,9 @@ try {
   for await (const event of client.iterateJobEvents({
     jobId: "22222222-2222-2222-2222-222222222222",
     limit: 10,
-    maxEvents: 50
+    maxEvents: 50,
+    resumeOnTransientError: true,
+    maxResumeAttempts: 3
   })) {
     console.log(event.kind, event.at);
   }

typescript/packages/sdk/src/index.ts

@@ -99,7 +99,10 @@ export type GetJobEventsAllPagesRequest = Omit<GetJobEventsRequest, "before"> &
   onPage?: (page: JobEventsResponse, pageIndex: number) => void;
 };
 
-export type IterateJobEventsRequest = GetJobEventsAllPagesRequest;
+export type IterateJobEventsRequest = GetJobEventsAllPagesRequest & {
+  resumeOnTransientError?: boolean;
+  maxResumeAttempts?: number;
+};
 
 export type JobEvent = {
   at: string;
@@ -285,20 +288,45 @@ export class SpatiadClient {
       return;
     }
 
+    const resumeOnTransientError = request.resumeOnTransientError ?? false;
+    const maxResumeAttempts = Math.max(0, request.maxResumeAttempts ?? 3);
     let cursor: string | undefined;
     let yielded = 0;
+    let resumeAttempts = 0;
 
     for (let page = 0; page < maxPages; page += 1) {
-      const current = await this.getJobEvents({
-        jobId: request.jobId,
-        limit: request.limit,
-        cursor,
-        kinds: request.kinds,
-        dispatcherToken: request.dispatcherToken,
-        dispatcherAuthMode: request.dispatcherAuthMode,
-        signal: request.signal,
-        retry: request.retry
-      });
+      let current: JobEventsResponse;
+      try {
+        current = await this.getJobEvents({
+          jobId: request.jobId,
+          limit: request.limit,
+          cursor,
+          kinds: request.kinds,
+          dispatcherToken: request.dispatcherToken,
+          dispatcherAuthMode: request.dispatcherAuthMode,
+          signal: request.signal,
+          retry: request.retry
+        });
+      } catch (error) {
+        if (
+          resumeOnTransientError
+          && error instanceof SpatiadApiError
+          && error.retryable
+          && resumeAttempts < maxResumeAttempts
+        ) {
+          resumeAttempts += 1;
+          const backoffBase = Math.max(0, request.retry?.backoffMs ?? 150);
+          const backoffMax = Math.max(backoffBase, request.retry?.maxBackoffMs ?? 2000);
+          const waitMs = Math.min(backoffMax, backoffBase * (2 ** (resumeAttempts - 1)));
+          await waitWithSignal(waitMs, request.signal);
+          page -= 1;
+          continue;
+        }
+
+        throw error;
+      }
+
+      resumeAttempts = 0;
 
       request.onPage?.(current, page);
 

typescript/packages/sdk/test/client.test.js

@@ -531,3 +531,79 @@ test("iterateJobEvents respects maxEvents", async () => {
     globalThis.fetch = originalFetch;
   }
 });
+
+test("iterateJobEvents can resume on transient api errors", async () => {
+  const originalFetch = globalThis.fetch;
+  let calls = 0;
+
+  globalThis.fetch = async () => {
+    calls += 1;
+    if (calls === 1) {
+      return makeJsonResponse(503, {
+        error: "service_unavailable",
+        message: "temporary outage"
+      });
+    }
+
+    return makeJsonResponse(200, {
+      job_id: "job-iter-recover",
+      events: [
+        { at: "2026-03-20T10:00:00Z", kind: "offer_created", offer_id: "o1", driver_id: "d1", status: "pending" }
+      ],
+      next_cursor: null,
+      next_before_cursor: null
+    });
+  };
+
+  try {
+    const client = new SpatiadClient("http://localhost:3000");
+    const streamed = [];
+
+    for await (const event of client.iterateJobEvents({
+      jobId: "job-iter-recover",
+      resumeOnTransientError: true,
+      maxResumeAttempts: 1,
+      retry: { maxAttempts: 1, backoffMs: 1, retryOnStatuses: [503] }
+    })) {
+      streamed.push(event.kind);
+    }
+
+    assert.deepEqual(streamed, ["offer_created"]);
+    assert.equal(calls, 2);
+  } finally {
+    globalThis.fetch = originalFetch;
+  }
+});
+
+test("iterateJobEvents fails without resumeOnTransientError", async () => {
+  const originalFetch = globalThis.fetch;
+
+  globalThis.fetch = async () =>
+    makeJsonResponse(503, {
+      error: "service_unavailable",
+      message: "temporary outage"
+    });
+
+  try {
+    const client = new SpatiadClient("http://localhost:3000");
+
+    await assert.rejects(
+      async () => {
+        for await (const _event of client.iterateJobEvents({
+          jobId: "job-iter-fail",
+          retry: { maxAttempts: 1, retryOnStatuses: [503] }
+        })) {
+          // no-op
+        }
+      },
+      (error) => {
+        assert.ok(error instanceof SpatiadApiError);
+        assert.equal(error.status, 503);
+        assert.equal(error.retryable, true);
+        return true;
+      }
+    );
+  } finally {
+    globalThis.fetch = originalFetch;
+  }
+});