diff --git a/server/src/__tests__/storage-s3-transient-retry.test.ts b/server/src/__tests__/storage-s3-transient-retry.test.ts new file mode 100644 index 00000000000..01bedcc6753 --- /dev/null +++ b/server/src/__tests__/storage-s3-transient-retry.test.ts @@ -0,0 +1,193 @@ +import { createServer, type IncomingMessage, type ServerResponse } from "node:http"; +import type { AddressInfo } from "node:net"; +import { mkdtemp, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { createS3StorageProvider } from "../storage/s3-provider.js"; + +type Seen = { method: string; range: string | null; ifMatch: string | null; body: string }; +const seen: Seen[] = []; +let respond: (req: IncomingMessage, res: ServerResponse, body: string, index: number) => void = (_req, res) => res.end(); + +const server = createServer((req, res) => { + const chunks: Buffer[] = []; + req.on("data", chunk => chunks.push(chunk as Buffer)); + req.on("end", () => { + const body = Buffer.concat(chunks).toString("utf8"); + seen.push({ method: req.method ?? "", range: req.headers.range ?? null, + ifMatch: (req.headers["if-match"] as string | undefined) ?? null, body }); + respond(req, res, body, seen.length - 1); + }); +}); +let endpoint = ""; +let dir = ""; +const savedEnv = { ...process.env }; + +beforeAll(async () => { + await new Promise(resolve => server.listen(0, "127.0.0.1", resolve)); + endpoint = `http://127.0.0.1:${(server.address() as AddressInfo).port}`; + process.env.AWS_ACCESS_KEY_ID = "fixture-access-key"; + process.env.AWS_SECRET_ACCESS_KEY = "fixture-secret-key"; + delete process.env.AWS_SESSION_TOKEN; + delete process.env.AWS_PROFILE; + dir = await mkdtemp(join(tmpdir(), "s3-transient-")); +}); + +afterAll(async () => { + process.env = savedEnv; + await rm(dir, { recursive: true, force: true }); + await new Promise(resolve => server.close(() => resolve())); +}); + +beforeEach(() => { seen.length = 0; }); + +function provider() { + return createS3StorageProvider({ bucket: "bucket", region: "us-east-1", endpoint, forcePathStyle: true }); +} + +async function readAll(stream: NodeJS.ReadableStream) { + const chunks: Buffer[] = []; + for await (const chunk of stream) chunks.push(Buffer.from(chunk as Buffer)); + return Buffer.concat(chunks).toString("utf8"); +} + +async function sourceFile(content: string) { + const file = join(dir, `${seen.length}-${Math.random().toString(16).slice(2)}.bin`); + await writeFile(file, content); + return file; +} + +const SERVER_ERROR = ` +InternalErrorWe encountered an internal error.req-500`; + +describe("S3 transient failures on saved files", () => { + it.each([ + ["an internal error", (res: ServerResponse) => { res.writeHead(500, { "content-type": "application/xml" }); res.end(SERVER_ERROR); }], + ["a connection reset", (res: ServerResponse) => { res.socket?.destroy(); }], + ])("resends a small file upload after %s by reading the file again", async (_label, fail) => { + const content = "saved workspace archive bytes"; + respond = (_req, res, _body, index) => { + if (index === 0) { fail(res); return; } + res.writeHead(200, { etag: '"saved"' }); + res.end(); + }; + + await expect(provider().putObject({ objectKey: "company/workspace.pcws", body: Buffer.alloc(0), + contentType: "application/octet-stream", contentLength: content.length, sourceFile: await sourceFile(content) })) + .resolves.toEqual({ integrity: "server_verified_md5" }); + + expect(seen.map(entry => entry.method)).toEqual(["PUT", "PUT"]); + expect(seen[1]!.body).toBe(content); + }); + + it("does not resend a small upload the store refused for good", async () => { + respond = (_req, res) => { + res.writeHead(403, { "content-type": "application/xml" }); + res.end(`AccessDenieddenied`); + }; + + await expect(provider().putObject({ objectKey: "company/workspace.pcws", body: Buffer.alloc(0), + contentType: "application/octet-stream", contentLength: 5, sourceFile: await sourceFile("bytes") })) + .rejects.toMatchObject({ name: "AccessDenied" }); + expect(seen.map(entry => entry.method)).toEqual(["PUT"]); + }); + + it("keeps a download complete when its connection breaks after the last byte", async () => { + const content = "0123456789abcdefghij"; + respond = (_req, res) => { + res.writeHead(200, { "content-length": String(content.length + 5), etag: '"v1"' }); + res.write(content); + setTimeout(() => res.socket?.destroy(), 20); + }; + + const object = await provider().getObject({ objectKey: "company/restore.pcws", range: { start: 0, end: content.length - 1 } }); + + expect(await readAll(object.stream)).toBe(content); + expect(seen.map(entry => entry.method)).toEqual(["GET"]); + }); + + it("resumes a download whose body is cut off from the byte where it stopped", async () => { + const content = "0123456789abcdefghij"; + respond = (req, res, _body, index) => { + if (index === 0) { + res.writeHead(200, { "content-length": String(content.length), etag: '"v1"' }); + res.write(content.slice(0, 8)); + setTimeout(() => res.socket?.destroy(), 20); + return; + } + const [, from, to] = /bytes=(\d+)-(\d*)/.exec(req.headers.range ?? "")!; + const end = to ? Number(to) : content.length - 1; + res.writeHead(206, { "content-length": String(end - Number(from) + 1), etag: '"v1"', + "content-range": `bytes ${from}-${end}/${content.length}` }); + res.end(content.slice(Number(from), end + 1)); + }; + + const object = await provider().getObject({ objectKey: "company/restore.pcws" }); + + expect(await readAll(object.stream)).toBe(content); + expect(seen.map(entry => [entry.method, entry.range, entry.ifMatch])).toEqual([ + ["GET", null, null], ["GET", "bytes=8-19", '"v1"']]); + }); + + it("closes a stalled download's connection when its reader gives up", async () => { + let closed!: () => void; + const connectionClosed = new Promise(resolve => { closed = resolve; }); + respond = (_req, res) => { + res.writeHead(200, { "content-length": "20", etag: '"v1"' }); + res.write("01234567"); + res.on("close", () => closed()); + }; + + const object = await provider().getObject({ objectKey: "company/restore.pcws" }); + const reader = object.stream[Symbol.asyncIterator](); + expect(String((await reader.next()).value)).toBe("01234567"); + object.stream.destroy(); + + await expect(Promise.race([connectionClosed.then(() => "closed"), + new Promise(resolve => setTimeout(() => resolve("still open"), 2_000))])).resolves.toBe("closed"); + }); + + it("closes a stalled resume request when its reader gives up", async () => { + let closed!: () => void; + const resumeClosed = new Promise(resolve => { closed = resolve; }); + respond = (_req, res, _body, index) => { + if (index === 0) { + res.writeHead(200, { "content-length": "20", etag: '"v1"' }); + res.write("01234567"); + setTimeout(() => res.socket?.destroy(), 20); + return; + } + res.on("close", () => closed()); + }; + + const object = await provider().getObject({ objectKey: "company/restore.pcws" }); + const reader = object.stream[Symbol.asyncIterator](); + expect(String((await reader.next()).value)).toBe("01234567"); + void reader.next().catch(() => undefined); + await vi.waitFor(() => expect(seen).toHaveLength(2), { timeout: 5_000 }); + object.stream.destroy(); + + await expect(Promise.race([resumeClosed.then(() => "closed"), + new Promise(resolve => setTimeout(() => resolve("still open"), 2_000))])).resolves.toBe("closed"); + }); + + it("fails instead of mixing bytes when the object changed before the download resumed", async () => { + const content = "0123456789abcdefghij"; + respond = (_req, res, _body, index) => { + if (index === 0) { + res.writeHead(200, { "content-length": String(content.length), etag: '"v1"' }); + res.write(content.slice(0, 8)); + setTimeout(() => res.socket?.destroy(), 20); + return; + } + res.writeHead(412, { "content-type": "application/xml" }); + res.end(`PreconditionFailedchanged`); + }; + + const object = await provider().getObject({ objectKey: "company/restore.pcws" }); + + await expect(readAll(object.stream)).rejects.toMatchObject({ name: "PreconditionFailed" }); + expect(seen.map(entry => entry.ifMatch)).toEqual([null, '"v1"']); + }); +}); diff --git a/server/src/storage/s3-multipart.ts b/server/src/storage/s3-multipart.ts index 8a18d035c59..74992e5f1a8 100644 --- a/server/src/storage/s3-multipart.ts +++ b/server/src/storage/s3-multipart.ts @@ -147,6 +147,43 @@ function isRequestRateLimited(error: unknown) { return name === "SlowDown" || $metadata?.httpStatusCode === 503 || $metadata?.httpStatusCode === 429; } +const TRANSIENT_FILE_REQUEST_STATUS = new Set([408, 429, 500, 502, 503, 504]); +const TRANSIENT_FILE_REQUEST_CODES = new Set(["ECONNRESET", "ECONNREFUSED", "ECONNABORTED", "ETIMEDOUT", "EPIPE", "EAI_AGAIN", + "ENOTFOUND", "ERR_STREAM_PREMATURE_CLOSE", "UND_ERR_SOCKET"]); +const TRANSIENT_FILE_REQUEST_NAMES = new Set(["RequestTimeout", "TimeoutError", "NetworkingError", "InternalError", "ServiceUnavailable"]); + +function isTransientFileRequestFailure(error: unknown) { + if (!error || typeof error !== "object") return false; + const { name, code, $metadata } = error as { name?: unknown; code?: unknown; $metadata?: { httpStatusCode?: unknown } }; + const status = $metadata?.httpStatusCode; + if (typeof status === "number") return TRANSIENT_FILE_REQUEST_STATUS.has(status); + return (typeof code === "string" && TRANSIENT_FILE_REQUEST_CODES.has(code)) + || (typeof name === "string" && TRANSIENT_FILE_REQUEST_NAMES.has(name)); +} + +export async function sendFileRequestWithRetry(signal: AbortSignal | undefined, send: () => Promise): Promise { + for (let failures = 0, throttles = 0; ;) { + signal?.throwIfAborted(); + try { + return await send(); + } catch (error) { + if (signal?.aborted) throw error; + if (isRequestRateLimited(error)) { + recordStorageThrottle(); + if (++throttles >= THROTTLED_PART_ATTEMPTS) throw error; + const ceiling = Math.min(THROTTLE_BACKOFF_MAX_MS, THROTTLE_BACKOFF_BASE_MS * 2 ** (throttles - 1)); + const wait = Math.floor(Math.random() * ceiling); + recordStorageBackoff(wait); + await delay(wait, undefined, { signal }); + continue; + } + if (!isTransientFileRequestFailure(error) || ++failures >= FILE_PART_ATTEMPTS) throw error; + recordStorageBackoff(failures * 500); + await delay(failures * 500, undefined, { signal }); + } + } +} + /** Base64 MD5 of an inclusive byte range, as the Content-MD5 header expects. */ export async function fileRangeMd5(file: string, start: number, end: number, signal?: AbortSignal) { const digest = createHash("md5"); diff --git a/server/src/storage/s3-provider.ts b/server/src/storage/s3-provider.ts index 9e553280d7d..294afb83a6e 100644 --- a/server/src/storage/s3-provider.ts +++ b/server/src/storage/s3-provider.ts @@ -7,7 +7,7 @@ import { ListObjectsV2Command, ListObjectVersionsCommand, } from "@aws-sdk/client-s3"; -import { assertSourceFileSize, fileRangeMd5, putS3Multipart, putS3MultipartFile } from "./s3-multipart.js"; +import { assertSourceFileSize, fileRangeMd5, putS3Multipart, putS3MultipartFile, sendFileRequestWithRetry } from "./s3-multipart.js"; import { createReadStream } from "node:fs"; import { createHash } from "node:crypto"; import { recordS3BodyFailure, withS3FailureDiagnostics } from "./s3-failure.js"; @@ -66,6 +66,59 @@ async function toReadableStream(body: unknown): Promise { throw unprocessable("Unsupported S3 body stream type"); } +function observeBodyFailures(stream: Readable, response: unknown) { + stream.on("error", function recordBodyFailure(error) { + recordS3BodyFailure(error, response); + if (stream.listenerCount("error") === 1) throw error; + }); + return stream; +} + +const BODY_RESUME_ATTEMPTS = 3; +const TRANSIENT_BODY_ERRORS = new Set(["ECONNRESET", "ETIMEDOUT", "EPIPE", "ECONNABORTED", "ERR_STREAM_PREMATURE_CLOSE", "UND_ERR_SOCKET"]); + +function isTransientBodyError(error: unknown) { + if (!error || typeof error !== "object") return false; + const { code, message } = error as { code?: unknown; message?: unknown }; + return (typeof code === "string" && TRANSIENT_BODY_ERRORS.has(code)) + || (typeof message === "string" && /socket hang up|aborted|premature close/i.test(message)); +} + +function resumableBody(first: Readable, reopen: (offset: number, signal: AbortSignal) => Promise, length: number, signal?: AbortSignal) { + let current = first, closed = false; + const reading = new AbortController(); + const resumeSignal = signal ? AbortSignal.any([signal, reading.signal]) : reading.signal; + async function* chunks() { + let offset = 0, resumes = 0; + while (true) { + try { + for await (const chunk of current) { + const bytes = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk as Uint8Array); + offset += bytes.length; + yield bytes; + } + return; + } catch (error) { + current.destroy(); + if (offset >= length) return; + if (closed || signal?.aborted || resumes >= BODY_RESUME_ATTEMPTS || !isTransientBodyError(error)) throw error; + resumes += 1; + current = await reopen(offset, resumeSignal); + if (closed) { current.destroy(); return; } + } + } + } + const stream = Readable.from(chunks(), { objectMode: false }); + const destroyWrapper = stream._destroy.bind(stream); + stream._destroy = (error, callback) => { + closed = true; + reading.abort(error ?? new Error("download_closed")); + current.destroy(); + destroyWrapper(error, callback); + }; + return stream; +} + function toDate(value: Date | undefined): Date | undefined { return value instanceof Date ? value : undefined; } @@ -164,13 +217,15 @@ export function createS3StorageProvider(config: S3ProviderConfig): StorageProvid if (input.body instanceof Readable) input.body.destroy(); await assertSourceFileSize(sourceFile, input.contentLength); const contentMd5 = await fileRangeMd5(sourceFile, 0, input.contentLength - 1, input.signal); - const body = createReadStream(sourceFile); - try { - await withS3FailureDiagnostics("put_object", () => fileClient.send(new PutObjectCommand({ - Bucket: bucket, Key: key, Body: body, ContentType: input.contentType, - ContentLength: input.contentLength, ContentMD5: contentMd5, - }), { abortSignal: input.signal })); - } finally { body.destroy(); } + await sendFileRequestWithRetry(input.signal, async () => { + const body = createReadStream(sourceFile); + try { + await withS3FailureDiagnostics("put_object", () => fileClient.send(new PutObjectCommand({ + Bucket: bucket, Key: key, Body: body, ContentType: input.contentType, + ContentLength: input.contentLength, ContentMD5: contentMd5, + }), { abortSignal: input.signal })); + } finally { body.destroy(); } + }); return { integrity: "server_verified_md5" } as const; } if (input.contentLength > 16 * 1024 * 1024) { @@ -201,13 +256,20 @@ export function createS3StorageProvider(config: S3ProviderConfig): StorageProvid { abortSignal: input.signal }, )); - const stream = await toReadableStream(output.Body); - // Observe, do not handle: a consumer without an error listener still - // sees the failure exactly as it would without this one. - stream.on("error", function recordBodyFailure(error) { - recordS3BodyFailure(error, output); - if (stream.listenerCount("error") === 1) throw error; - }); + const first = observeBodyFailures(await toReadableStream(output.Body), output); + const etag = output.ETag; + const start = input.range?.start ?? 0; + const end = input.range?.end ?? (typeof output.ContentLength === "number" ? output.ContentLength - 1 : undefined); + const reopen = async (offset: number, resumeSignal: AbortSignal) => { + const resumed = await withS3FailureDiagnostics("get_object", () => client.send(new GetObjectCommand({ + Bucket: bucket, Key: key, IfMatch: etag, + Range: `bytes=${start + offset}-${end ?? ""}`, + }), { abortSignal: resumeSignal })); + return observeBodyFailures(await toReadableStream(resumed.Body), resumed); + }; + const stream = etag && end !== undefined + ? resumableBody(first, reopen, end - start + 1, input.signal) + : first; if (input.signal) addAbortSignal(input.signal, stream); return { stream,