diff --git a/packages/void/src/cli/client.ts b/packages/void/src/cli/client.ts index 5d9fe5d82..762e1e879 100644 --- a/packages/void/src/cli/client.ts +++ b/packages/void/src/cli/client.ts @@ -170,6 +170,23 @@ export function isExpiredTokenError(error: unknown): boolean { return false; } +/** + * True when `fetch` gave up waiting for response headers (undici's + * `HeadersTimeoutError`, 300 seconds by default). The request reached the + * server, which is still working on it. + */ +export function isHeadersTimeoutError(error: unknown): boolean { + let current = error; + for (let depth = 0; depth < 6 && current !== null && typeof current === 'object'; depth++) { + const { code, name } = current as { code?: unknown; name?: unknown }; + if (code === 'UND_ERR_HEADERS_TIMEOUT' || name === 'HeadersTimeoutError') { + return true; + } + current = (current as { cause?: unknown }).cause; + } + return false; +} + function isPreflightRejection(error: unknown): boolean { return ( error instanceof PlatformApiError && @@ -577,6 +594,10 @@ export type ManifestOnlyDeployPayload = DeployUploadPayload; * - Only `fetch` rejections (`TypeError` — no HTTP response) are retried. * A Response of any status means the server answered; surface it * verbatim rather than replay a request it may have accepted. + * - A headers timeout is retried too. Every attempt sends the same + * idempotency key, so once an earlier request has created its deployment, + * a receipt-aware platform answers the retry with that deployment's + * receipt and the CLI follows it to the real result. * - A rejected fetch does NOT prove the request never arrived — the * connection may have dropped after delivery. Arbitration is left to * the server's precheck, the only atomic (project-locked) verdict: if @@ -590,6 +611,11 @@ export type ManifestOnlyDeployPayload = DeployUploadPayload; */ const DEPLOY_POST_MAX_RETRIES = 3; const DEPLOY_POST_BACKOFF_MS = [1000, 5000, 10000]; +/** + * Every transport failure is retried, so a `TypeError` thrown by the deploy + * POST always comes after this many attempts. + */ +export const DEPLOY_POST_ATTEMPTS: number = DEPLOY_POST_MAX_RETRIES + 1; const DEPLOY_OBSERVATION_TIMEOUT_MS = 30_000; const DEPLOY_OBSERVATION_DEADLINE_MS = 30 * 60_000; const DEPLOY_OBSERVATION_BACKOFF_MS = [1000, 2000, 5000, 10_000, 30_000]; diff --git a/packages/void/src/cli/deploy-errors.ts b/packages/void/src/cli/deploy-errors.ts index 7196f626a..4fad9581e 100644 --- a/packages/void/src/cli/deploy-errors.ts +++ b/packages/void/src/cli/deploy-errors.ts @@ -1,4 +1,10 @@ -import { isCliOutdatedError, parsePlatformErrorBody, PlatformApiError } from './client.ts'; +import { + DEPLOY_POST_ATTEMPTS, + isCliOutdatedError, + isHeadersTimeoutError, + parsePlatformErrorBody, + PlatformApiError, +} from './client.ts'; import { platformErrorMessage } from './platform-errors.ts'; const BODY_SNIPPET_MAX = 200; @@ -27,6 +33,31 @@ function appendTrailingLines( return appendLogLine(appendDeploymentLine(message, deploymentId), logPath); } +/** + * Describe the deepest `cause` of an error in one short line, e.g. + * `getaddrinfo ENOTFOUND api.example.com` under `fetch failed`. + */ +function describeRootCause(err: Error): string | null { + let root: unknown = err; + for (let depth = 0; depth < 6; depth++) { + const cause = (root as { cause?: unknown }).cause; + if (cause === null || typeof cause !== 'object') { + break; + } + root = cause; + } + if (root === err) { + return null; + } + const { code, message } = root as { code?: unknown; message?: unknown }; + const text = typeof message === 'string' ? message.trim() : ''; + const codeText = typeof code === 'string' ? code : ''; + if (!text) { + return codeText || null; + } + return codeText && !text.includes(codeText) ? `${codeText}: ${text}` : text; +} + /** * Format a server-emitted `event: 'error'` payload with an optional deployment id. */ @@ -56,14 +87,18 @@ export function formatConnectionLostMessage( * Convert an exception thrown during the deploy NDJSON stream into a * user-facing `Error` with a readable message. * - * Three cases: + * Four cases: * 1. `PlatformApiError` with 409 + a JSON body carrying `{error, deployment}` * — surface the server-supplied error and prefer the body's deployment id. * 2. Any other `PlatformApiError` — retain the client's actionable message and * structured error fields; opaque provider failures use a trimmed body snippet. - * 3. Anything else — preserve the original message under a `Connection lost - * during deploy: ` prefix so the caller can still tell network, - * parse, and transport errors apart. + * 3. A headers timeout on the last deploy POST attempt — the platform sent + * no response in time. Explain that the deploy may still finish and how + * to check and retry. + * 4. Anything else — preserve the original message, plus its root cause when + * there is one, under a `Connection lost during deploy: ` + * prefix so the caller can still tell network, parse, and transport + * errors apart. */ export function formatStreamDeployError( err: unknown, @@ -98,10 +133,23 @@ export function formatStreamDeployError( return err; } + if (isHeadersTimeoutError(err)) { + const lines = [ + `deploy: The Void platform did not respond while preparing the deploy. Tried ${DEPLOY_POST_ATTEMPTS} times; the last try waited 5 minutes.`, + 'It may still finish. Check `void project status`, then run `void deploy` again to retry.', + ]; + if (logPath) { + lines.push('If this keeps happening, share the detailed log below.'); + } + return new Error(appendTrailingLines(lines.join('\n'), deploymentId, logPath)); + } + const original = err instanceof Error ? err.message : String(err); + const rootCause = err instanceof Error ? describeRootCause(err) : null; + const detail = rootCause && rootCause !== original ? ` (${rootCause})` : ''; return new Error( appendTrailingLines( - `deploy: Connection lost during deploy: '${original}'.`, + `deploy: Connection lost during deploy: '${original}'${detail}.`, deploymentId, logPath, ), diff --git a/packages/void/test/unit/direct-r2-client.test.ts b/packages/void/test/unit/direct-r2-client.test.ts index ad4619e8b..404499c4f 100644 --- a/packages/void/test/unit/direct-r2-client.test.ts +++ b/packages/void/test/unit/direct-r2-client.test.ts @@ -1,5 +1,9 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vite-plus/test'; -import { PlatformClient, type ManifestOnlyDeployPayload } from '../../src/cli/client.ts'; +import { + DEPLOY_POST_ATTEMPTS, + PlatformClient, + type ManifestOnlyDeployPayload, +} from '../../src/cli/client.ts'; import type { DeployManifest } from '../../src/cli/types.ts'; import { withPlatformDiscovery } from '../helpers/platform-discovery.ts'; @@ -296,7 +300,9 @@ describe('PlatformClient.finalizeDeploy', () => { posts++; if (posts === 1) { return Promise.reject( - new TypeError('fetch failed', { cause: new Error('read ECONNRESET') }), + new TypeError('fetch failed', { + cause: Object.assign(new Error('read ECONNRESET'), { code: 'ECONNRESET' }), + }), ); } return Promise.resolve( @@ -450,6 +456,85 @@ describe('PlatformClient.finalizeDeploy', () => { expect(fetchMock).toHaveBeenCalledTimes(1); }); + function headersTimeoutError() { + const cause = Object.assign(new Error('Headers Timeout Error'), { + name: 'HeadersTimeoutError', + code: 'UND_ERR_HEADERS_TIMEOUT', + }); + return { cause, error: new TypeError('fetch failed', { cause }) }; + } + + const idempotencyKeys = () => + deployPostCalls().map( + ([, init]) => (init.headers as Record)['x-void-deploy-idempotency-key'], + ); + + it('retries a headers timeout with the same idempotency key, then rethrows it', async () => { + const { cause, error } = headersTimeoutError(); + fetchMock.mockRejectedValue(error); + + vi.useFakeTimers(); + const cliLog = makeCliLog(); + const client = new PlatformClient(token, { apiUrl, cliLog }); + try { + const assertion = expect( + client.finalizeDeploy(projectId, makePayload()).next(), + ).rejects.toMatchObject({ message: 'fetch failed', cause }); + await vi.runAllTimersAsync(); + await assertion; + + expect(deployPostCalls()).toHaveLength(4); + expect(DEPLOY_POST_ATTEMPTS).toBe(4); + expect(new Set(idempotencyKeys()).size).toBe(1); + const retries = cliLog.warn.mock.calls.filter(([msg]) => msg === 'deploy_fetch_retry'); + expect(retries.map(([, fields]) => fields?.delayMs)).toEqual([1000, 5000, 10000]); + expect(cliLog.error).toHaveBeenCalledWith( + 'deploy_fetch_rejected', + expect.any(TypeError), + expect.objectContaining({ attempts: 4 }), + ); + } finally { + vi.useRealTimers(); + } + }); + + it('follows the existing deployment receipt returned to a retry after a headers timeout', async () => { + const done = { + event: 'done', + deployment: 'dpl_existing', + url: 'https://x.void.app', + urls: [], + assets: 1, + workers: 1, + seq: 3, + }; + fetchMock + .mockRejectedValueOnce(headersTimeoutError().error) + .mockResolvedValueOnce( + ndjsonStreamResponse([ + { event: 'start', deploymentId: 'dpl_existing', seq: 1 }, + { event: 'worker_upload', seq: 2 }, + done, + ]), + ); + + vi.useFakeTimers(); + try { + const deploy = collectDeploy(); + await vi.runAllTimersAsync(); + expect(await deploy).toEqual([ + { event: 'start', deploymentId: 'dpl_existing' }, + { event: 'worker_upload', seq: 2 }, + done, + ]); + expect(deployPostCalls()).toHaveLength(2); + const [first, second] = idempotencyKeys(); + expect(second).toBe(first); + } finally { + vi.useRealTimers(); + } + }); + it('does not retry HTTP error responses — the server answered', async () => { fetchMock.mockResolvedValue(new Response('busted', { status: 503 })); const client = makeClient(); diff --git a/packages/void/test/unit/stream-deploy-errors.test.ts b/packages/void/test/unit/stream-deploy-errors.test.ts index 028120b0f..1b6d51e26 100644 --- a/packages/void/test/unit/stream-deploy-errors.test.ts +++ b/packages/void/test/unit/stream-deploy-errors.test.ts @@ -63,6 +63,71 @@ describe('formatStreamDeployError', () => { expect(formatted.message).toContain('fetch failed'); }); + it('explains a headers timeout after the last retry in plain words and keeps the log line', () => { + const err = new TypeError('fetch failed', { + cause: Object.assign(new Error('Headers Timeout Error'), { + name: 'HeadersTimeoutError', + code: 'UND_ERR_HEADERS_TIMEOUT', + }), + }); + + const formatted = formatStreamDeployError(err, null, '/tmp/deploy.jsonl'); + + expect(formatted.message).toBe( + [ + 'deploy: The Void platform did not respond while preparing the deploy. Tried 4 times; the last try waited 5 minutes.', + 'It may still finish. Check `void project status`, then run `void deploy` again to retry.', + 'If this keeps happening, share the detailed log below.', + 'Detailed log: /tmp/deploy.jsonl', + ].join('\n'), + ); + }); + + it('omits the log hint for a headers timeout when no log was written', () => { + const err = new TypeError('fetch failed', { + cause: Object.assign(new Error('Headers Timeout Error'), { + code: 'UND_ERR_HEADERS_TIMEOUT', + }), + }); + + const formatted = formatStreamDeployError(err, null, null); + + expect(formatted.message).not.toContain('log'); + expect(formatted.message).toContain('void project status'); + }); + + it('adds the root cause of a fetch failure', () => { + const err = new TypeError('fetch failed', { + cause: Object.assign(new Error('getaddrinfo ENOTFOUND api.example.com'), { + code: 'ENOTFOUND', + }), + }); + + const formatted = formatStreamDeployError(err, null, '/tmp/deploy.jsonl'); + + expect(formatted.message).toBe( + "deploy: Connection lost during deploy: 'fetch failed' (getaddrinfo ENOTFOUND api.example.com).\n" + + 'Detailed log: /tmp/deploy.jsonl', + ); + }); + + it('uses the cause code when the root cause has no message', () => { + const refused = Object.assign(new AggregateError([], ''), { code: 'ECONNREFUSED' }); + const closed = Object.assign(new Error('other side closed'), { + name: 'SocketError', + code: 'UND_ERR_SOCKET', + }); + + expect( + formatStreamDeployError(new TypeError('fetch failed', { cause: refused }), null).message, + ).toBe("deploy: Connection lost during deploy: 'fetch failed' (ECONNREFUSED)."); + expect( + formatStreamDeployError(new TypeError('fetch failed', { cause: closed }), null).message, + ).toBe( + "deploy: Connection lost during deploy: 'fetch failed' (UND_ERR_SOCKET: other side closed).", + ); + }); + it('includes the deployment id for generic mid-stream errors when known', () => { const err = new Error('reader canceled'); diff --git a/platform/packages/api/src/deployment-execution.ts b/platform/packages/api/src/deployment-execution.ts index 243bec27d..3af95735f 100644 --- a/platform/packages/api/src/deployment-execution.ts +++ b/platform/packages/api/src/deployment-execution.ts @@ -1,5 +1,6 @@ import { and, eq, getTableColumns, lte, ne, or, sql } from 'drizzle-orm'; import { paginatedR2Delete } from './assets.ts'; +import { R2_CONCURRENCY } from './concurrency.ts'; import { createDb, type Db, type DbOrTx } from './db.ts'; import type { DeploymentInput, DeploymentOperation } from './deployment-operation-types.ts'; import type { ApiEnv } from './env.ts'; @@ -109,14 +110,43 @@ export async function prepareDeploymentInput( }; delete input.body.workerFiles; try { + // Hash one module at a time so at most one extra copy of module bytes is + // alive beside the decoded Blobs; a near-cap request already sits close + // to the isolate memory limit. + const modules: DeploymentInput['modules'] = []; for (let index = 0; index < files.length; index++) { const { name, content } = files[index]!; - const bytes = await content.arrayBuffer(); - const digest = await sha256Hex(bytes); - const key = `${prefix}${index}`; - await r2.put(key, bytes); - input.modules.push({ name, key, digest, type: content.type }); + const digest = await sha256Hex(await content.arrayBuffer()); + modules.push({ name, key: `${prefix}${index}`, digest, type: content.type }); } + // The request has not answered yet, so upload the Blobs themselves in + // parallel. Workers let at most six R2 calls wait for a response at once + // (R2_CONCURRENCY); a larger pool would only queue inside the runtime. + // After the first failure no new put starts, and every started put + // settles before cleanup so none lands under the prefix after it was + // deleted. + let next = 0; + let failed = false; + const workers = await Promise.allSettled( + Array.from({ length: Math.min(R2_CONCURRENCY, files.length) }, async () => { + while (!failed && next < files.length) { + const index = next++; + try { + await r2.put(modules[index]!.key, files[index]!.content); + } catch (error) { + failed = true; + throw error; + } + } + }), + ); + const rejected = workers.find( + (result): result is PromiseRejectedResult => result.status === 'rejected', + ); + if (rejected) { + throw rejected.reason; + } + input.modules = modules; const identity = JSON.stringify({ body: input.body, modules: input.modules.map(({ name, digest, type }) => ({ name, digest, type })), diff --git a/platform/packages/api/test/integration/api-email.test.ts b/platform/packages/api/test/integration/api-email.test.ts index 5847ebca9..abafabb71 100644 --- a/platform/packages/api/test/integration/api-email.test.ts +++ b/platform/packages/api/test/integration/api-email.test.ts @@ -530,20 +530,27 @@ describe('POST /projects/:id/email/destinations', () => { }); it('rate-limits to 5 adds per minute', async () => { - mockCfApi(); - const user = await createTestUser(env.db); - const project = await createTestProject(env.db, user.id); + // Keep all six requests in one minute bucket even when the test crosses + // a wall-clock minute. + vi.useFakeTimers({ toFake: ['Date'], now: Date.now() }); + try { + mockCfApi(); + const user = await createTestUser(env.db); + const project = await createTestProject(env.db, user.id); + + for (let i = 0; i < 5; i++) { + const res = await addDestination(project.id, user.jwt, `r${i}@example.com`); + expect(res.status).toBe(201); + } - for (let i = 0; i < 5; i++) { - const res = await addDestination(project.id, user.jwt, `r${i}@example.com`); - expect(res.status).toBe(201); + const res = await addDestination(project.id, user.jwt, 'r5@example.com'); + expect(res.status).toBe(429); + expect((await res.json()) as { error: string }).toEqual({ + error: expect.stringMatching(/per minute|limit/i), + }); + } finally { + vi.useRealTimers(); } - - const res = await addDestination(project.id, user.jwt, 'r5@example.com'); - expect(res.status).toBe(429); - expect((await res.json()) as { error: string }).toEqual({ - error: expect.stringMatching(/per minute|limit/i), - }); }); it('rate-limits to 50 adds per day', async () => { diff --git a/platform/packages/api/test/integration/helpers.ts b/platform/packages/api/test/integration/helpers.ts index 241d7862c..86c747aef 100644 --- a/platform/packages/api/test/integration/helpers.ts +++ b/platform/packages/api/test/integration/helpers.ts @@ -318,7 +318,7 @@ class MemoryR2Bucket { async put( key: string, - value: ArrayBuffer | ArrayBufferView | ReadableStream | string, + value: ArrayBuffer | ArrayBufferView | ReadableStream | Blob | string, options?: { httpMetadata?: unknown; onlyIf?: { etagMatches?: string; etagDoesNotMatch?: string }; @@ -337,6 +337,8 @@ class MemoryR2Bucket { } else if (value instanceof ReadableStream) { const bytes = await readStreamToBytes(value); body = toArrayBuffer(bytes); + } else if (value instanceof Blob) { + body = await value.arrayBuffer(); } else { body = toArrayBuffer(value); } diff --git a/platform/packages/api/test/integration/native-form-browser.ts b/platform/packages/api/test/integration/native-form-browser.ts index ddd30e59d..3bd8246db 100644 --- a/platform/packages/api/test/integration/native-form-browser.ts +++ b/platform/packages/api/test/integration/native-form-browser.ts @@ -1,7 +1,19 @@ -import { chromium } from 'playwright'; +import { chromium, type Browser } from 'playwright'; +import { afterAll } from 'vite-plus/test'; import { app } from '../../src/index.ts'; import type { TestEnv } from './helpers.ts'; +// One browser serves every submission in a test file; each submission gets its +// own context, so cookies and page state stay isolated. Launching and closing +// Chromium per submission costs more than the form exchange itself on a busy +// machine. +let browser: Promise | undefined; +afterAll(async () => { + const launched = browser; + browser = undefined; + await (await launched?.catch(() => undefined))?.close(); +}); + /** Exercise the rendered native form locally, without provider or public HTTP. */ export async function nativeFormSubmission(input: { env: TestEnv; @@ -12,13 +24,13 @@ export async function nativeFormSubmission(input: { button: string; noReferrer?: boolean; }) { - const browser = await chromium.launch(); + browser ??= chromium.launch(); + const context = await (await browser).newContext({ javaScriptEnabled: false }); let pagePolicy: string | null = null; let submission: | { origin: string | undefined; referrer: string | undefined; status: number } | undefined; try { - const context = await browser.newContext({ javaScriptEnabled: false }); const split = input.cookie.indexOf('='); await context.addCookies([ { @@ -104,6 +116,6 @@ export async function nativeFormSubmission(input: { } return { pagePolicy, submission }; } finally { - await browser.close(); + await context.close(); } } diff --git a/platform/packages/api/test/unit/deployment-input-staging.test.ts b/platform/packages/api/test/unit/deployment-input-staging.test.ts new file mode 100644 index 000000000..976b5cc76 --- /dev/null +++ b/platform/packages/api/test/unit/deployment-input-staging.test.ts @@ -0,0 +1,165 @@ +import { describe, expect, it } from 'vite-plus/test'; +import { R2_CONCURRENCY } from '../../src/concurrency.ts'; +import { prepareDeploymentInput } from '../../src/deployment-execution.ts'; + +type Put = { key: string; release: (error?: Error) => void }; + +/** Holds every put open until the test releases it. */ +class GatedR2 { + readonly objects = new Map(); + readonly pending: Array = []; + readonly started: Array = []; + readonly values = new Map(); + inFlight = 0; + maxInFlight = 0; + deletedWhileInFlight = false; + + async put(key: string, value: Blob | string): Promise { + this.started.push(key); + this.values.set(key, value); + this.inFlight++; + this.maxInFlight = Math.max(this.maxInFlight, this.inFlight); + try { + await new Promise((resolve, reject) => { + this.pending.push({ key, release: (error) => (error ? reject(error) : resolve()) }); + }); + } finally { + this.inFlight--; + } + this.objects.set(key, typeof value === 'string' ? value : await value.text()); + return {}; + } + + async list({ prefix }: { prefix: string }) { + return { + objects: [...this.objects.keys()] + .filter((key) => key.startsWith(prefix)) + .map((key) => ({ + key, + })), + truncated: false, + }; + } + + async delete(keys: string | Array) { + if (this.inFlight > 0) { + this.deletedWhileInFlight = true; + } + for (const key of Array.isArray(keys) ? keys : [keys]) { + this.objects.delete(key); + } + } + + /** Release the oldest pending put once one is waiting. */ + async releaseNext(error?: Error): Promise { + while (this.pending.length === 0) { + await new Promise((resolve) => setTimeout(resolve, 0)); + } + const put = this.pending.shift()!; + put.release(error); + return put.key; + } +} + +async function waitFor(condition: () => boolean): Promise { + for (let attempt = 0; !condition(); attempt++) { + if (attempt > 1000) { + throw new Error('condition was not reached'); + } + await new Promise((resolve) => setTimeout(resolve, 0)); + } +} + +function files(count: number) { + return Array.from({ length: count }, (_, index) => ({ + name: `module-${index}.js`, + content: new Blob([`export default ${index};`], { type: 'application/javascript+module' }), + })); +} + +describe('prepareDeploymentInput', () => { + it('stages modules with bounded concurrency, in input order, and writes the receipt last', async () => { + const r2 = new GatedR2(); + const count = R2_CONCURRENCY * 3 + 1; + const modules = files(count); + const staging = prepareDeploymentInput( + r2 as unknown as R2Bucket, + 'prj_test', + { manifest: { type: 'void' }, workerFiles: { ignored: true } }, + modules, + ); + // Keep the pool full, and release puts out of start order: newest first. + for (let released = 0; released < count; released++) { + await waitFor(() => r2.pending.length === Math.min(R2_CONCURRENCY, count - released)); + expect(r2.started.some((key) => key.endsWith('receipt'))).toBe(false); + r2.pending.pop()!.release(); + } + expect(await r2.releaseNext()).toMatch(/receipt$/); + const { input, prefix, digest } = await staging; + + expect(r2.maxInFlight).toBe(R2_CONCURRENCY); + expect(r2.started).toHaveLength(count + 1); + expect(r2.started.at(-1)).toBe(`${prefix}receipt`); + expect(input.body).not.toHaveProperty('workerFiles'); + expect(input.modules.map(({ name, key }) => ({ name, key }))).toEqual( + Array.from({ length: count }, (_, index) => ({ + name: `module-${index}.js`, + key: `${prefix}${index}`, + })), + ); + expect(r2.objects.get(`${prefix}${count - 1}`)).toBe(`export default ${count - 1};`); + // Each put sends the decoded Blob itself rather than a buffered copy. + modules.forEach(({ content }, index) => + expect(r2.values.get(`${prefix}${index}`)).toBe(content), + ); + expect(digest).toMatch(/^[0-9a-f]{64}$/); + + // The identity digest depends only on the content, not on completion order. + const sequential = new GatedR2(); + const again = prepareDeploymentInput( + sequential as unknown as R2Bucket, + 'prj_test', + { manifest: { type: 'void' } }, + files(count), + ); + for (let released = 0; released <= count; released++) { + await sequential.releaseNext(); + } + expect((await again).digest).toBe(digest); + }); + + it('rejects on the first failed put, starts no further puts, and cleans up after in-flight puts settle', async () => { + const r2 = new GatedR2(); + const count = R2_CONCURRENCY * 4; + const staging = prepareDeploymentInput( + r2 as unknown as R2Bucket, + 'prj_test', + { manifest: { type: 'void' } }, + files(count), + ); + const outcome = staging.then( + () => null, + (error: unknown) => error, + ); + await waitFor(() => r2.pending.length === R2_CONCURRENCY); + await r2.releaseNext(); + await waitFor(() => r2.pending.length === R2_CONCURRENCY); + await r2.releaseNext(new Error('R2 put failed')); + const startedAtFailure = r2.started.length; + expect(startedAtFailure).toBe(R2_CONCURRENCY + 1); + // Settle the puts that were already running when the failure happened. + while (r2.pending.length > 0) { + r2.pending.shift()!.release(); + await new Promise((resolve) => setTimeout(resolve, 0)); + } + + const error = await outcome; + expect(error).toBeInstanceOf(Error); + expect((error as Error).message).toBe('R2 put failed'); + expect(r2.started).toHaveLength(startedAtFailure); + expect(r2.started.length).toBeLessThan(count); + expect(r2.started.some((key) => key.endsWith('receipt'))).toBe(false); + expect(r2.deletedWhileInFlight).toBe(false); + expect(r2.objects.size).toBe(0); + }); +});