Long-poll the desktop OAuth await endpoint

.changeset/oauth-await-longpoll.md

@@ -0,0 +1,10 @@
+---
+"@executor-js/local": patch
+"@executor-js/react": patch
+---
+
+**Desktop OAuth connects finish the moment the provider redirects**
+
+When the desktop app runs an OAuth flow in the system browser, the app learned about completion by polling the local server once a second. The completed result sat in memory while the user watched the "Connecting…" spinner for up to a second more — about half a second wasted on average, on every connect.
+
+The await endpoint now long-polls: the server holds the request open (up to 25 seconds per hold) and answers the instant the flow completes. The client polls one request at a time and reconnects after each answer, so requests never stack. Mixed versions stay compatible in both directions: an old client still gets its answer within one poll of a new server, and a new client against an old server behaves exactly as before.

apps/local/src/oauth-result-store.test.ts

@@ -0,0 +1,222 @@
+import { afterEach, describe, expect, it } from "@effect/vitest";
+// oxlint-disable-next-line executor/no-vitest-import -- boundary: deterministic timer control comes from vitest itself
+import { vi } from "vitest";
+import { OAUTH_POPUP_MESSAGE_TYPE, type OAuthPopupResult } from "@executor-js/sdk";
+
+import {
+  __oauthAwaitHeldWaiterTotalForTests,
+  __oauthAwaitWaiterCountForTests,
+  __resetOAuthResultStoreForTests,
+  consumeOAuthResult,
+  publishOAuthResult,
+  waitForOAuthResult,
+} from "./oauth-result-store";
+
+const sampleResult = (sessionId: string): OAuthPopupResult<unknown> => ({
+  type: OAUTH_POPUP_MESSAGE_TYPE,
+  ok: false,
+  sessionId,
+  error: "access denied",
+});
+
+afterEach(() => {
+  __resetOAuthResultStoreForTests();
+  vi.useRealTimers();
+});
+
+// Waiter registration, publish wake-ups, aborts, and the over-cap instant
+// answer are all synchronous, so every ordering below is exact — no sleeps.
+
+describe("waitForOAuthResult", () => {
+  it("resolves immediately and consumes when the result is already published", async () => {
+    publishOAuthResult(sampleResult("s-ready"));
+
+    const result = await waitForOAuthResult("s-ready", { timeoutMs: 5000 });
+
+    expect(result).toMatchObject({ sessionId: "s-ready" });
+    // One-shot: the wait consumed the entry.
+    expect(consumeOAuthResult("s-ready")).toBeNull();
+    expect(__oauthAwaitWaiterCountForTests("s-ready")).toBe(0);
+  });
+
+  it("resolves a held wait the moment the result is published", async () => {
+    const pending = waitForOAuthResult("s-mid", { timeoutMs: 5000 });
+    expect(__oauthAwaitWaiterCountForTests("s-mid")).toBe(1);
+
+    const publishedAt = Date.now();
+    publishOAuthResult(sampleResult("s-mid"));
+    const result = await pending;
+
+    // Resolved by the publish, not the 5s deadline.
+    expect(Date.now() - publishedAt).toBeLessThan(1000);
+    expect(result).toMatchObject({ sessionId: "s-mid" });
+    expect(consumeOAuthResult("s-mid")).toBeNull();
+    expect(__oauthAwaitWaiterCountForTests("s-mid")).toBe(0);
+  });
+
+  it("returns null at the deadline and removes the waiter", async () => {
+    vi.useFakeTimers();
+    const pending = waitForOAuthResult("s-deadline", { timeoutMs: 25_000 });
+    expect(__oauthAwaitWaiterCountForTests("s-deadline")).toBe(1);
+
+    vi.advanceTimersByTime(25_000);
+
+    expect(await pending).toBeNull();
+    expect(__oauthAwaitWaiterCountForTests("s-deadline")).toBe(0);
+  });
+
+  it("resolves immediately when the signal is already aborted", async () => {
+    const controller = new AbortController();
+    controller.abort();
+
+    const result = await waitForOAuthResult("s-pre-aborted", {
+      timeoutMs: 5000,
+      signal: controller.signal,
+    });
+
+    expect(result).toBeNull();
+    expect(__oauthAwaitWaiterCountForTests("s-pre-aborted")).toBe(0);
+  });
+});
+
+describe("waitForOAuthResult publish/abort races", () => {
+  it("abort settles first; a publish arriving after leaves the result for the next consumer", async () => {
+    const controller = new AbortController();
+    const pending = waitForOAuthResult("s-abort-first", {
+      timeoutMs: 5000,
+      signal: controller.signal,
+    });
+    expect(__oauthAwaitWaiterCountForTests("s-abort-first")).toBe(1);
+
+    controller.abort();
+    expect(await pending).toBeNull();
+    expect(__oauthAwaitWaiterCountForTests("s-abort-first")).toBe(0);
+
+    // The publish lands after the abort settled: the dead waiter must not
+    // consume it — it stays in the store for the client's next poll.
+    publishOAuthResult(sampleResult("s-abort-first"));
+    expect(consumeOAuthResult("s-abort-first")).toMatchObject({ sessionId: "s-abort-first" });
+  });
+
+  it("publish resolves the waiter; an abort firing immediately after does not double-consume", async () => {
+    const controller = new AbortController();
+    const pending = waitForOAuthResult("s-pub-first", {
+      timeoutMs: 5000,
+      signal: controller.signal,
+    });
+    expect(__oauthAwaitWaiterCountForTests("s-pub-first")).toBe(1);
+
+    publishOAuthResult(sampleResult("s-pub-first"));
+    controller.abort();
+
+    expect(await pending).toMatchObject({ sessionId: "s-pub-first" });
+    expect(__oauthAwaitWaiterCountForTests("s-pub-first")).toBe(0);
+
+    // The late abort must not have consumed or dropped anything: a second
+    // publish for the session is still delivered intact.
+    publishOAuthResult(sampleResult("s-pub-first"));
+    expect(consumeOAuthResult("s-pub-first")).toMatchObject({ sessionId: "s-pub-first" });
+  });
+});
+
+describe("waitForOAuthResult publish/timeout races", () => {
+  it("deadline fires first; a publish just after leaves the result consumable", async () => {
+    vi.useFakeTimers();
+    const pending = waitForOAuthResult("s-late-pub", { timeoutMs: 25_000 });
+    expect(__oauthAwaitWaiterCountForTests("s-late-pub")).toBe(1);
+
+    vi.advanceTimersByTime(25_000);
+    expect(await pending).toBeNull();
+    expect(__oauthAwaitWaiterCountForTests("s-late-pub")).toBe(0);
+
+    publishOAuthResult(sampleResult("s-late-pub"));
+    expect(consumeOAuthResult("s-late-pub")).toMatchObject({ sessionId: "s-late-pub" });
+  });
+
+  it("publish resolves the waiter; the stale deadline timer is inert afterwards", async () => {
+    vi.useFakeTimers();
+    const pending = waitForOAuthResult("s-early-pub", { timeoutMs: 25_000 });
+    expect(__oauthAwaitWaiterCountForTests("s-early-pub")).toBe(1);
+
+    publishOAuthResult(sampleResult("s-early-pub"));
+    expect(await pending).toMatchObject({ sessionId: "s-early-pub" });
+    expect(__oauthAwaitWaiterCountForTests("s-early-pub")).toBe(0);
+
+    // Run the clock past the original deadline: the settled waiter's timer
+    // was cleared, so nothing re-fires, re-registers, or consumes again.
+    vi.advanceTimersByTime(25_000);
+    expect(__oauthAwaitWaiterCountForTests("s-early-pub")).toBe(0);
+    publishOAuthResult(sampleResult("s-early-pub"));
+    expect(consumeOAuthResult("s-early-pub")).toMatchObject({ sessionId: "s-early-pub" });
+  });
+});
+
+describe("waitForOAuthResult held-waiter caps", () => {
+  it("holds one waiter per session; stacked requests answer null immediately (old-client behavior)", async () => {
+    // An unpatched desktop client polls every second and would stack ~25
+    // concurrent requests per flow against a holding server. Only the first
+    // may hold; the rest get the pre-long-poll instant "still pending".
+    const held = waitForOAuthResult("s-stack", { timeoutMs: 5000 });
+    expect(__oauthAwaitWaiterCountForTests("s-stack")).toBe(1);
+
+    const stacked = [
+      waitForOAuthResult("s-stack", { timeoutMs: 5000 }),
+      waitForOAuthResult("s-stack", { timeoutMs: 5000 }),
+      waitForOAuthResult("s-stack", { timeoutMs: 5000 }),
+    ];
+    // The stacked requests settle null BEFORE any publish — instant answers.
+    for (const request of stacked) expect(await request).toBeNull();
+    expect(__oauthAwaitWaiterCountForTests("s-stack")).toBe(1);
+    expect(__oauthAwaitHeldWaiterTotalForTests()).toBe(1);
+
+    publishOAuthResult(sampleResult("s-stack"));
+    // Exactly one consumer receives the one-shot result: the held waiter.
+    expect(await held).toMatchObject({ sessionId: "s-stack" });
+    expect(consumeOAuthResult("s-stack")).toBeNull();
+    // No leaked waiters after the flow.
+    expect(__oauthAwaitWaiterCountForTests("s-stack")).toBe(0);
+    expect(__oauthAwaitHeldWaiterTotalForTests()).toBe(0);
+  });
+
+  it("re-arms the per-session hold after the held waiter settles", async () => {
+    const controller = new AbortController();
+    const first = waitForOAuthResult("s-rearm", { timeoutMs: 5000, signal: controller.signal });
+    // Over the per-session cap while the first is held.
+    expect(await waitForOAuthResult("s-rearm", { timeoutMs: 5000 })).toBeNull();
+
+    controller.abort();
+    expect(await first).toBeNull();
+    expect(__oauthAwaitWaiterCountForTests("s-rearm")).toBe(0);
+
+    // The slot is free again: the next request holds and gets the result.
+    const second = waitForOAuthResult("s-rearm", { timeoutMs: 5000 });
+    expect(__oauthAwaitWaiterCountForTests("s-rearm")).toBe(1);
+    publishOAuthResult(sampleResult("s-rearm"));
+    expect(await second).toMatchObject({ sessionId: "s-rearm" });
+    expect(__oauthAwaitWaiterCountForTests("s-rearm")).toBe(0);
+  });
+
+  it("caps total held waiters globally; over-cap sessions answer instantly but stored results still deliver", async () => {
+    const held = Array.from({ length: 64 }, (_, index) =>
+      waitForOAuthResult(`s-global-${index}`, { timeoutMs: 5000 }),
+    );
+    expect(__oauthAwaitHeldWaiterTotalForTests()).toBe(64);
+
+    // A distinct pending session over the global cap answers null instantly
+    // instead of holding.
+    expect(await waitForOAuthResult("s-global-over", { timeoutMs: 5000 })).toBeNull();
+    expect(__oauthAwaitHeldWaiterTotalForTests()).toBe(64);
+
+    // Over-cap behavior matches the pre-long-poll server exactly: a result
+    // already in the store is still consumed and answered immediately.
+    publishOAuthResult(sampleResult("s-global-stored"));
+    expect(await waitForOAuthResult("s-global-stored", { timeoutMs: 5000 })).toMatchObject({
+      sessionId: "s-global-stored",
+    });
+
+    // Settle every held waiter and confirm the registry drains completely.
+    __resetOAuthResultStoreForTests();
+    for (const request of held) expect(await request).toBeNull();
+    expect(__oauthAwaitHeldWaiterTotalForTests()).toBe(0);
+  });
+});

apps/local/src/oauth-result-store.ts

@@ -28,6 +28,37 @@ const RESULT_TTL_MS = 10 * 60 * 1000; // 10 minutes — long enough for slow MFA
 
 const store = new Map<string, StoredResult>();
 
+/**
+ * Long-poll waiters, keyed by sessionId. Each entry is the wake callback for
+ * the single held `/api/oauth/await/:sessionId` request for that session.
+ * `publishOAuthResult` wakes the waiter, which runs `consumeOAuthResult` and
+ * answers with the result.
+ *
+ * Held waiters are bounded — each pins a connection, a registry entry, and a
+ * deadline timer, and the route only needs a bearer, so unbounded holds are a
+ * DoS surface. At most one waiter is held per session (the patched client
+ * polls sequentially; an unpatched client polling every second would
+ * otherwise stack ~25 holds per flow against a 25s deadline), and at most
+ * `MAX_HELD_WAITERS_TOTAL` across all sessions. Over either bound,
+ * `waitForOAuthResult` degrades to the pre-long-poll behavior: it answers the
+ * current store value immediately (null = still pending) instead of holding,
+ * so over-cap callers see exactly what an old server would have sent.
+ */
+const MAX_HELD_WAITERS_TOTAL = 64;
+
+const waiters = new Map<string, () => void>();
+
+const removeWaiter = (sessionId: string, wake: () => void): void => {
+  if (waiters.get(sessionId) === wake) waiters.delete(sessionId);
+};
+
+const wakeWaiter = (sessionId: string): void => {
+  const wake = waiters.get(sessionId);
+  if (!wake) return;
+  waiters.delete(sessionId);
+  wake();
+};
+
 const cleanupExpired = (now: number) => {
   for (const [sessionId, entry] of store) {
     if (entry.expiresAt < now) store.delete(sessionId);
@@ -44,6 +75,7 @@ export const publishOAuthResult = (result: AnyResult): void => {
   const now = Date.now();
   cleanupExpired(now);
   store.set(sessionId, { result, expiresAt: now + RESULT_TTL_MS });
+  wakeWaiter(sessionId);
 };
 
 /**
@@ -59,7 +91,57 @@ export const consumeOAuthResult = (sessionId: string): AnyResult | null => {
   return entry.result;
 };
 
-/** Test-only — clears the entire store between tests. */
+/**
+ * Long-poll for a result. Consumes and resolves immediately when a result
+ * is already stored; otherwise holds until `publishOAuthResult` fires for
+ * the sessionId, the deadline elapses, or `signal` aborts (client gone).
+ * The latter two resolve `null` — the same "still pending" answer an
+ * immediate poll gives — so the caller's retry loop keeps working. A
+ * waiter that times out or aborts is always removed from the registry.
+ *
+ * Holding is bounded (see the waiter registry above): when the session
+ * already has a held waiter, or the global held-waiter ceiling is reached,
+ * this answers `null` immediately instead of holding.
+ */
+export const waitForOAuthResult = (
+  sessionId: string,
+  opts: { readonly timeoutMs: number; readonly signal?: AbortSignal },
+): Promise<AnyResult | null> => {
+  const immediate = consumeOAuthResult(sessionId);
+  if (immediate !== null) return Promise.resolve(immediate);
+  if (opts.timeoutMs <= 0 || opts.signal?.aborted === true) return Promise.resolve(null);
+  if (waiters.has(sessionId) || waiters.size >= MAX_HELD_WAITERS_TOTAL) {
+    return Promise.resolve(null);
+  }
+
+  return new Promise((resolve) => {
+    let done = false;
+    const finish = (result: AnyResult | null) => {
+      if (done) return;
+      done = true;
+      clearTimeout(timer);
+      opts.signal?.removeEventListener("abort", onAbort);
+      removeWaiter(sessionId, wake);
+      resolve(result);
+    };
+    const wake = () => finish(consumeOAuthResult(sessionId));
+    const onAbort = () => finish(null);
+    const timer = setTimeout(() => finish(null), opts.timeoutMs);
+
+    waiters.set(sessionId, wake);
+    opts.signal?.addEventListener("abort", onAbort, { once: true });
+  });
+};
+
+/** Test-only — clears the store and resolves any held waiters as pending. */
 export const __resetOAuthResultStoreForTests = (): void => {
   store.clear();
+  for (const sessionId of [...waiters.keys()]) wakeWaiter(sessionId);
 };
+
+/** Test-only — number of held long-poll waiters for a sessionId (0 or 1). */
+export const __oauthAwaitWaiterCountForTests = (sessionId: string): number =>
+  waiters.has(sessionId) ? 1 : 0;
+
+/** Test-only — total held long-poll waiters across all sessions. */
+export const __oauthAwaitHeldWaiterTotalForTests = (): number => waiters.size;