diff --git a/CHANGELOG.md b/CHANGELOG.md index c05e5e3d..ae980b97 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,6 +17,10 @@ published version with a date and open a fresh empty `[Unreleased]` above it. - See `docs/migration/file-native-writeback.md` for the new read/edit/create/delete contract, schema discovery flow, writeback status surface, and per-adapter draft filename examples. - See `docs/migration/new-json-callers.md` for the Cloud and demo caller scan that identifies `new.json` write paths to migrate. +### Fixed + +- Direct HTTP write admission now honors `Retry-After` up to 30 seconds for `workspace_busy` / `write_admission_limit` responses in the SDK's existing four-attempt retry layer, while preserving the prior two-second cap for other retryable responses. Implicit defaults are three seconds for receipt polling and 90 seconds for admission; an explicit `writebackTimeoutMs` bounds each phase independently, with `0` leaving admission unbounded and receipt polling disabled. + ### Added - GitHub pull indexes now surface `merged` and `mergedAt` across webhook, direct, and bulk ingestion paths so time-windowed consumers can identify merged pull requests without opening every `meta.json`. diff --git a/packages/core/src/vfs-client/index.ts b/packages/core/src/vfs-client/index.ts index 87e0cf54..3d380fca 100644 --- a/packages/core/src/vfs-client/index.ts +++ b/packages/core/src/vfs-client/index.ts @@ -85,6 +85,33 @@ export class RelayfileWritebackPendingError extends RelayfileWritebackError { } } +export interface RelayfileWritebackAdmissionTimeoutErrorOptions { + provider: string; + operation: string; + path: string; + timeoutMs: number; +} + +/** Direct HTTP admission never minted an op before the caller's admission deadline. */ +export class RelayfileWritebackAdmissionTimeoutError extends RelayfileWritebackError { + readonly path: string; + readonly timeoutMs: number; + + constructor(options: RelayfileWritebackAdmissionTimeoutErrorOptions) { + super({ + provider: options.provider, + operation: options.operation, + cause: new Error( + `writeback_admission_timeout: no operation admitted for ${options.path} after ${options.timeoutMs}ms` + ), + retryable: true + }); + this.name = "RelayfileWritebackAdmissionTimeoutError"; + this.path = options.path; + this.timeoutMs = options.timeoutMs; + } +} + export interface RelayfileWritebackTerminalErrorOptions { provider: string; operation: string; @@ -148,7 +175,12 @@ export interface IntegrationClientOptions { /** * Max wait, in ms, for the Relayfile writeback worker to emit a receipt onto * the just-written draft. Defaults to 3000ms. `0` means fire-and-forget — the - * client returns immediately without a receipt. + * client returns immediately without a receipt. In direct HTTP mode, an + * explicit value also bounds write admission as an independent phase. When + * omitted, receipt waiting defaults to 3s while admission defaults to 90s. + * Advertised delays are honored while they fit inside that admission budget; + * after three consecutive 30s delays, the deadline wins at t+90s before a + * fourth request. */ writebackTimeoutMs?: number; /** Poll interval while waiting for a receipt. Default 250ms. */ @@ -197,6 +229,9 @@ export interface WritebackResult { } const DEFAULT_WRITEBACK_TIMEOUT_MS = 3_000; +const SDK_LEGACY_RETRY_MAX_DELAY_MS = 2_000; +const WORKSPACE_BUSY_RETRY_MAX_DELAY_MS = 30_000; +const DEFAULT_WRITEBACK_ADMISSION_TIMEOUT_MS = WORKSPACE_BUSY_RETRY_MAX_DELAY_MS * 3; function isRecord(value: unknown): value is Record { return typeof value === "object" && value !== null && !Array.isArray(value); @@ -207,6 +242,85 @@ function nonEmpty(value: string | undefined): string | undefined { return trimmed ? trimmed : undefined; } +function isWorkspaceBusyAdmission(value: unknown): boolean { + if (!isRecord(value) || value.code !== "workspace_busy") return false; + const reason = value.reason ?? (isRecord(value.details) ? value.details.reason : undefined); + return reason === "write_admission_limit"; +} + +function isWorkspaceBusyAdmissionError(value: unknown): value is RelayFileApiError { + return value instanceof RelayFileApiError && value.status === 429 && isWorkspaceBusyAdmission(value); +} + +function retryAfterDelayMs(value: string): number | undefined { + const trimmed = value.trim(); + if (/^\d+$/.test(trimmed)) { + const seconds = Number(trimmed); + if (Number.isFinite(seconds)) return seconds * 1_000; + } + const timestamp = Date.parse(trimmed); + return Number.isNaN(timestamp) ? undefined : Math.max(0, timestamp - Date.now()); +} + +async function responseIsWorkspaceBusyAdmission(response: Response): Promise { + if (response.status !== 429) return false; + try { + return isWorkspaceBusyAdmission(await response.clone().json()); + } catch { + return false; + } +} + +function responseWithRetryAfter(response: Response, retryAfter: string): Response { + const headers = new Headers(response.headers); + headers.set("Retry-After", retryAfter); + return new Response(response.body, { + status: response.status, + statusText: response.statusText, + headers + }); +} + +function isDirectWriteAdmissionRequest(input: RequestInfo | URL, init?: RequestInit): boolean { + const method = (init?.method ?? (input instanceof Request ? input.method : "GET")).toUpperCase(); + const url = input instanceof Request ? input.url : String(input); + try { + return method === "PUT" && new URL(url, "http://relayfile.invalid").pathname.endsWith("/fs/file"); + } catch { + return false; + } +} + +/** + * The SDK owns the only retry loop. Raising its max delay lets workspace write + * admission honor Retry-After; this adapter keeps the SDK's previous 2s cap for + * every other retryable response so unrelated 429/5xx behavior does not move. + */ +function directRetryFetch(fetchImpl: typeof fetch): typeof fetch { + return async (input, init) => { + const response = await fetchImpl(input, init); + if (response.status !== 429 && (response.status < 500 || response.status > 599)) { + return response; + } + const retryAfter = response.headers.get("Retry-After"); + if (!retryAfter) return response; + const delayMs = retryAfterDelayMs(retryAfter); + if ( + isDirectWriteAdmissionRequest(input, init) && + (await responseIsWorkspaceBusyAdmission(response)) + ) { + return delayMs === undefined + ? response + : responseWithRetryAfter(response, String(Math.ceil(delayMs / 1_000))); + } + + if (delayMs === undefined || delayMs <= SDK_LEGACY_RETRY_MAX_DELAY_MS) { + return response; + } + return responseWithRetryAfter(response, String(SDK_LEGACY_RETRY_MAX_DELAY_MS / 1_000)); + }; +} + function mountRootCandidate(value: string | undefined): string | undefined { const trimmed = value?.trim(); return trimmed ? trimmed : undefined; @@ -304,12 +418,16 @@ function directClientConfig( nonEmpty(process.env.RELAYFILE_WORKSPACE) ?? nonEmpty(process.env.RELAY_WORKSPACE_ID); if (!baseUrl || !token || !workspaceId) return undefined; + const fetchImpl = directRetryFetch(client.fetchImpl ?? globalThis.fetch); return { workspaceId, relayfile: new RelayFileClient({ baseUrl, token, - fetchImpl: client.fetchImpl + fetchImpl, + retry: { + maxDelayMs: WORKSPACE_BUSY_RETRY_MAX_DELAY_MS + } }) }; } @@ -423,30 +541,53 @@ async function writeJsonFileViaRelayfileApi( body: unknown, direct: { workspaceId: string; relayfile: RelayFileClient } ): Promise { - const queued = await direct.relayfile.writeFile({ - workspaceId: direct.workspaceId, - path: relayPath, - baseRevision: "*", - contentType: "application/json", - content: `${JSON.stringify(body, null, 2)}\n` - }); - if (queued.writeback && !nonEmpty(queued.opId)) { - throw new RelayfileWritebackReceiptError({ - provider, - operation, - opId: "(missing)", - reason: "queued writeback response did not include opId" + const timeoutMs = client.writebackTimeoutMs ?? DEFAULT_WRITEBACK_ADMISSION_TIMEOUT_MS; + const controller = timeoutMs > 0 ? new AbortController() : undefined; + const deadlineTimer = controller + ? setTimeout(() => controller.abort(), timeoutMs) + : undefined; + let admitted = false; + try { + const queued = await direct.relayfile.writeFile({ + workspaceId: direct.workspaceId, + path: relayPath, + baseRevision: "*", + contentType: "application/json", + content: `${JSON.stringify(body, null, 2)}\n`, + signal: controller?.signal }); + admitted = true; + if (deadlineTimer !== undefined) clearTimeout(deadlineTimer); + if (queued.writeback && !nonEmpty(queued.opId)) { + throw new RelayfileWritebackReceiptError({ + provider, + operation, + opId: "(missing)", + reason: "queued writeback response did not include opId" + }); + } + const receipt = queued.opId + ? await waitForOperationReceipt(client, provider, operation, direct, queued.opId, relayPath) + : undefined; + return { + path: relayPath, + absolutePath: relayPath, + opId: queued.opId, + ...(receipt ? { receipt } : {}) + }; + } catch (error) { + if (controller?.signal.aborted && !admitted && !(error instanceof RelayfileWritebackError)) { + throw new RelayfileWritebackAdmissionTimeoutError({ + provider, + operation, + path: relayPath, + timeoutMs + }); + } + throw error; + } finally { + if (deadlineTimer !== undefined) clearTimeout(deadlineTimer); } - const receipt = queued.opId - ? await waitForOperationReceipt(client, provider, operation, direct, queued.opId, relayPath) - : undefined; - return { - path: relayPath, - absolutePath: relayPath, - opId: queued.opId, - ...(receipt ? { receipt } : {}) - }; } function toAbsolutePath(client: IntegrationClientOptions, relayPath: string): string { @@ -612,7 +753,12 @@ export async function writeJsonFile( if (cause instanceof RelayfileWritebackError) { throw cause; } - throw new RelayfileWritebackError({ provider, operation, cause, retryable: false }); + throw new RelayfileWritebackError({ + provider, + operation, + cause, + retryable: isWorkspaceBusyAdmissionError(cause) + }); } } diff --git a/packages/core/tests/vfs-client/vfs-client.test.ts b/packages/core/tests/vfs-client/vfs-client.test.ts index 89f5c221..39830a8c 100644 --- a/packages/core/tests/vfs-client/vfs-client.test.ts +++ b/packages/core/tests/vfs-client/vfs-client.test.ts @@ -3,7 +3,9 @@ import { mkdtemp, mkdir, readFile, readdir, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import path from "node:path"; import test from "node:test"; +import { RelayFileApiError } from "@relayfile/sdk"; import { + RelayfileWritebackAdmissionTimeoutError, RelayfileWritebackError, RelayfileWritebackPendingError, RelayfileWritebackReceiptError, @@ -22,6 +24,25 @@ async function mount(): Promise<{ root: string; opts: { relayfileMountRoot: stri return { root, opts: { relayfileMountRoot: root, writebackTimeoutMs: 0 } }; } +async function withImmediateTimeouts(fn: (delays: number[]) => Promise): Promise { + const delays: number[] = []; + const realSetTimeout = globalThis.setTimeout; + globalThis.setTimeout = (( + callback: (...args: unknown[]) => void, + delay?: number, + ...args: unknown[] + ) => { + delays.push(delay ?? 0); + queueMicrotask(() => callback(...args)); + return 0 as unknown as ReturnType; + }) as unknown as typeof setTimeout; + try { + return await fn(delays); + } finally { + globalThis.setTimeout = realSetTimeout; + } +} + test("writeJsonFile drops a draft atomically under the mount path", async () => { const { root, opts } = await mount(); const rel = `/linear/issues/${encodeSegment("ISS-1")}/comments/${draftFile("comment")}`; @@ -206,6 +227,427 @@ test("writeJsonFile can wait on direct Relayfile op providerResult instead of mo assert.match(requests[1].url, /\/v1\/workspaces\/rw_7ccfea89\/ops\/op_slack_1$/); }); +test("writeJsonFile direct mode honors workspace_busy Retry-After in one four-attempt SDK retry layer", async () => { + const requests: Array<{ url: string; body: BodyInit | null | undefined }> = []; + let attempts = 0; + const fetchImpl: typeof fetch = async (input, init = {}) => { + attempts += 1; + requests.push({ url: String(input), body: init.body }); + if (attempts < 4) { + return Response.json( + { + code: "workspace_busy", + message: "workspace write path is busy; retry after the advertised delay", + reason: "write_admission_limit", + retryAfterSeconds: 5 + }, + { status: 429, headers: { "Retry-After": "5" } } + ); + } + return Response.json({ status: "queued", targetRevision: "rev_1" }); + }; + + const result = await withImmediateTimeouts(async (delays) => { + const write = await writeJsonFile( + { + relayfileBaseUrl: "https://relayfile.example.test", + relayfileApiToken: "test-token", + workspaceId: "rw_busy", + fetchImpl, + writebackTimeoutMs: 0 + }, + "slack", + "post", + "/slack/channels/C1/messages/relayfile-writeback--busy.json", + { text: "hello", idempotencyKey: "tick:delivery-1:1" } + ); + assert.deepEqual(delays, [5_000, 5_000, 5_000]); + return write; + }); + + assert.equal(result.opId, undefined); + assert.equal(requests.length, 4, "the SDK layer must own exactly four total attempts"); + assert.equal(new Set(requests.map((request) => request.url)).size, 1); + assert.equal(new Set(requests.map((request) => String(request.body))).size, 1); + assert.equal(requests.some((request) => request.url.includes("/ops/")), false); + assert.match(String(requests[0].body), /tick:delivery-1:1/); +}); + +test("writeJsonFile direct mode marks only exhausted workspace_busy admission as retryable", async () => { + let attempts = 0; + const fetchImpl: typeof fetch = async () => { + attempts += 1; + return Response.json( + { + code: "workspace_busy", + message: "workspace write path is busy", + reason: "write_admission_limit" + }, + { status: 429, headers: { "Retry-After": "5" } } + ); + }; + + await withImmediateTimeouts(async (delays) => { + await assert.rejects( + () => + writeJsonFile( + { + relayfileBaseUrl: "https://relayfile.example.test", + relayfileApiToken: "test-token", + workspaceId: "rw_busy_exhausted", + fetchImpl, + writebackTimeoutMs: 0 + }, + "slack", + "post", + "/slack/channels/C1/messages/relayfile-writeback--busy-exhausted.json", + { text: "hello" } + ), + (error: unknown) => + error instanceof RelayfileWritebackError && + error.retryable && + error.cause instanceof RelayFileApiError && + error.cause.status === 429 && + error.cause.code === "workspace_busy" && + error.cause.details?.reason === "write_admission_limit" + ); + assert.deepEqual(delays, [5_000, 5_000, 5_000]); + }); + assert.equal(attempts, 4); +}); + +test("writeJsonFile direct mode preserves the two-second cap for workspace_busy with another reason", async () => { + let attempts = 0; + const fetchImpl: typeof fetch = async () => { + attempts += 1; + return Response.json( + { code: "workspace_busy", reason: "another_limit", message: "ordinary rate limit" }, + { status: 429, headers: { "Retry-After": "5" } } + ); + }; + + await withImmediateTimeouts(async (delays) => { + await assert.rejects( + () => + writeJsonFile( + { + relayfileBaseUrl: "https://relayfile.example.test", + relayfileApiToken: "test-token", + workspaceId: "rw_rate_limited", + fetchImpl, + writebackTimeoutMs: 0 + }, + "slack", + "post", + "/slack/channels/C1/messages/relayfile-writeback--rate-limited.json", + { text: "hello" } + ), + RelayfileWritebackError + ); + assert.deepEqual(delays, [2_000, 2_000, 2_000]); + }); + assert.equal(attempts, 4); +}); + +test("writeJsonFile direct mode preserves the existing two-second cap for 5xx retries", async () => { + let attempts = 0; + const fetchImpl: typeof fetch = async () => { + attempts += 1; + return Response.json( + { code: "service_unavailable", message: "try later" }, + { status: 503, headers: { "Retry-After": "5" } } + ); + }; + + await withImmediateTimeouts(async (delays) => { + await assert.rejects( + () => + writeJsonFile( + { + relayfileBaseUrl: "https://relayfile.example.test", + relayfileApiToken: "test-token", + workspaceId: "rw_unavailable", + fetchImpl, + writebackTimeoutMs: 0 + }, + "slack", + "post", + "/slack/channels/C1/messages/relayfile-writeback--unavailable.json", + { text: "hello" } + ), + RelayfileWritebackError + ); + assert.deepEqual(delays, [2_000, 2_000, 2_000]); + }); + assert.equal(attempts, 4); +}); + +test("writeJsonFile direct mode preserves the SDK backoff schedule without Retry-After", async () => { + let attempts = 0; + const fetchImpl: typeof fetch = async () => { + attempts += 1; + return Response.json( + { code: "rate_limited", message: "retry without an advertised delay" }, + { status: 429 } + ); + }; + const realRandom = Math.random; + Math.random = () => 0.5; + try { + await withImmediateTimeouts(async (delays) => { + await assert.rejects( + () => + writeJsonFile( + { + relayfileBaseUrl: "https://relayfile.example.test", + relayfileApiToken: "test-token", + workspaceId: "rw_no_retry_after", + fetchImpl, + writebackTimeoutMs: 0 + }, + "slack", + "post", + "/slack/channels/C1/messages/relayfile-writeback--no-retry-after.json", + { text: "hello" } + ), + RelayfileWritebackError + ); + assert.deepEqual(delays, [100, 200, 400]); + }); + } finally { + Math.random = realRandom; + } + assert.equal(attempts, 4); +}); + +test("writeJsonFile direct mode parses a digit-leading Retry-After date as a date", async () => { + let attempts = 0; + const fetchImpl: typeof fetch = async () => { + attempts += 1; + return Response.json( + { + code: "workspace_busy", + message: "workspace write path is busy", + reason: "write_admission_limit" + }, + { status: 429, headers: { "Retry-After": "1 Jan 1970 00:00:00 GMT" } } + ); + }; + + await withImmediateTimeouts(async (delays) => { + await assert.rejects( + () => + writeJsonFile( + { + relayfileBaseUrl: "https://relayfile.example.test", + relayfileApiToken: "test-token", + workspaceId: "rw_date_retry_after", + fetchImpl, + writebackTimeoutMs: 0 + }, + "slack", + "post", + "/slack/channels/C1/messages/relayfile-writeback--date-retry-after.json", + { text: "hello" } + ), + RelayfileWritebackError + ); + assert.deepEqual(delays, []); + }); + assert.equal(attempts, 4); +}); + +test("writeJsonFile direct admission deadline aborts an advertised retry without an orphan attempt", async () => { + let attempts = 0; + const fetchImpl: typeof fetch = async () => { + attempts += 1; + return Response.json( + { + code: "workspace_busy", + message: "workspace write path is busy; retry after the advertised delay", + reason: "write_admission_limit", + retryAfterSeconds: 30 + }, + { status: 429, headers: { "Retry-After": "30" } } + ); + }; + const deadlineHandle = 20 as unknown as ReturnType; + const retryHandle = 30_000 as unknown as ReturnType; + const clearedHandles: Array> = []; + let deadlineCallback: (() => void) | undefined; + const realSetTimeout = globalThis.setTimeout; + const realClearTimeout = globalThis.clearTimeout; + globalThis.setTimeout = (( + callback: (...args: unknown[]) => void, + delay?: number, + ...args: unknown[] + ) => { + if (delay === 20) { + deadlineCallback = () => callback(...args); + return deadlineHandle; + } + assert.equal(delay, 30_000); + queueMicrotask(() => deadlineCallback?.()); + return retryHandle; + }) as unknown as typeof setTimeout; + globalThis.clearTimeout = ((handle: ReturnType) => { + clearedHandles.push(handle); + }) as typeof clearTimeout; + try { + await assert.rejects( + () => + writeJsonFile( + { + relayfileBaseUrl: "https://relayfile.example.test", + relayfileApiToken: "test-token", + workspaceId: "rw_deadline", + fetchImpl, + writebackTimeoutMs: 20 + }, + "slack", + "post", + "/slack/channels/C1/messages/relayfile-writeback--deadline.json", + { text: "hello" } + ), + (error: unknown) => + error instanceof RelayfileWritebackAdmissionTimeoutError && + error.retryable && + /writeback_admission_timeout/.test(error.message) + ); + } finally { + globalThis.setTimeout = realSetTimeout; + globalThis.clearTimeout = realClearTimeout; + } + + assert.equal(attempts, 1); + assert.equal( + clearedHandles.includes(retryHandle), + true, + "the SDK retry timer must be canceled when the client deadline wins" + ); +}); + +test("writeJsonFile direct mode uses a 90s admission default when receipt timeout is omitted", async () => { + let attempts = 0; + const fetchImpl: typeof fetch = async () => { + attempts += 1; + if (attempts === 1) { + return Response.json( + { + code: "workspace_busy", + message: "workspace write path is busy", + reason: "write_admission_limit" + }, + { status: 429, headers: { "Retry-After": "5" } } + ); + } + return Response.json({ status: "queued", targetRevision: "rev_default_admitted" }); + }; + const delays: number[] = []; + const deadlineHandle = 90_000 as unknown as ReturnType; + let deadlineCleared = false; + const realSetTimeout = globalThis.setTimeout; + const realClearTimeout = globalThis.clearTimeout; + globalThis.setTimeout = (( + callback: (...args: unknown[]) => void, + delay?: number, + ...args: unknown[] + ) => { + delays.push(delay ?? 0); + if (delay === 90_000) return deadlineHandle; + queueMicrotask(() => callback(...args)); + return 5_000 as unknown as ReturnType; + }) as unknown as typeof setTimeout; + globalThis.clearTimeout = ((handle: ReturnType) => { + if (handle === deadlineHandle) deadlineCleared = true; + }) as typeof clearTimeout; + try { + const result = await writeJsonFile( + { + relayfileBaseUrl: "https://relayfile.example.test", + relayfileApiToken: "test-token", + workspaceId: "rw_default_admission", + fetchImpl + }, + "slack", + "post", + "/slack/channels/C1/messages/relayfile-writeback--default-admission.json", + { text: "hello" } + ); + assert.equal(result.opId, undefined); + } finally { + globalThis.setTimeout = realSetTimeout; + globalThis.clearTimeout = realClearTimeout; + } + assert.equal(attempts, 2); + assert.deepEqual(delays, [90_000, 5_000]); + assert.equal(deadlineCleared, true); +}); + +test("writeJsonFile direct mode bounds repeated 30s admission delays at the 90s default", async () => { + let attempts = 0; + const fetchImpl: typeof fetch = async () => { + attempts += 1; + return Response.json( + { + code: "workspace_busy", + message: "workspace write path is busy", + reason: "write_admission_limit" + }, + { status: 429, headers: { "Retry-After": "30" } } + ); + }; + const delays: number[] = []; + let deadlineCallback: (() => void) | undefined; + let retryTimers = 0; + const realSetTimeout = globalThis.setTimeout; + const realClearTimeout = globalThis.clearTimeout; + globalThis.setTimeout = (( + callback: (...args: unknown[]) => void, + delay?: number, + ...args: unknown[] + ) => { + delays.push(delay ?? 0); + if (delay === 90_000) { + deadlineCallback = () => callback(...args); + return 90_000 as unknown as ReturnType; + } + retryTimers += 1; + if (retryTimers < 3) { + queueMicrotask(() => callback(...args)); + } else { + queueMicrotask(() => deadlineCallback?.()); + } + return retryTimers as unknown as ReturnType; + }) as unknown as typeof setTimeout; + globalThis.clearTimeout = (() => {}) as typeof clearTimeout; + try { + await assert.rejects( + () => + writeJsonFile( + { + relayfileBaseUrl: "https://relayfile.example.test", + relayfileApiToken: "test-token", + workspaceId: "rw_default_admission_bound", + fetchImpl + }, + "slack", + "post", + "/slack/channels/C1/messages/relayfile-writeback--default-admission-bound.json", + { text: "hello" } + ), + RelayfileWritebackAdmissionTimeoutError + ); + } finally { + globalThis.setTimeout = realSetTimeout; + globalThis.clearTimeout = realClearTimeout; + } + assert.equal(attempts, 3); + assert.deepEqual(delays, [90_000, 30_000, 30_000, 30_000]); + await new Promise((resolve) => setTimeout(resolve, 20)); + assert.equal(attempts, 3, "the default admission deadline must not leave an orphan fourth attempt"); +}); + test("writeJsonFile direct mode reports a pending op as retryable writeback_pending", async () => { const fetchImpl: typeof fetch = async (input) => { const url = String(input);