mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 12:37:55 +01:00
fix(storage): retry S3 reads that the provider times out with HTTP 408 (#19039)
Mark HTTP 408 on GET/HEAD as retryable inside the SDK retry middleware so the standard backoff, retry quota and three-attempt limit apply. Writes are unchanged. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
68db7ed9e3
commit
de82f93443
3 files changed
+139
No files matched your search
@@ -0,0 +1,99 @@
|
||||
import { createServer, type IncomingMessage, type ServerResponse } from "node:http";
|
||||
import type { AddressInfo } from "node:net";
|
||||
import { afterAll, beforeAll, beforeEach, describe, expect, it } from "vitest";
|
||||
import { storageFailureDiagnosticsFor } from "../storage/failure-diagnostics.js";
|
||||
import { createS3StorageProvider } from "../storage/s3-provider.js";
|
||||
|
||||
// The body an S3-compatible gateway sends when it gives up on a request.
|
||||
const REQUEST_CANCELED = `<?xml version="1.0" encoding="UTF-8"?>
|
||||
<Error><Code>RequestCanceled</Code><Message>Request is canceled.</Message><RequestId>req-408</RequestId></Error>`;
|
||||
|
||||
let failuresLeft = 0;
|
||||
const requests: string[] = [];
|
||||
const server = createServer((req: IncomingMessage, res: ServerResponse) => {
|
||||
req.resume();
|
||||
req.on("end", () => {
|
||||
requests.push(req.method ?? "");
|
||||
if (failuresLeft > 0) {
|
||||
failuresLeft -= 1;
|
||||
res.writeHead(408, { "content-type": "application/xml", "x-amz-request-id": "req-408" });
|
||||
res.end(req.method === "HEAD" ? undefined : REQUEST_CANCELED);
|
||||
return;
|
||||
}
|
||||
res.writeHead(200, { "content-type": "text/plain", "content-length": "5", etag: '"etag"' });
|
||||
res.end(req.method === "GET" ? "hello" : undefined);
|
||||
});
|
||||
});
|
||||
let endpoint = "";
|
||||
const savedEnv = { ...process.env };
|
||||
|
||||
beforeAll(async () => {
|
||||
await new Promise<void>((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;
|
||||
});
|
||||
|
||||
afterAll(async () => {
|
||||
process.env = savedEnv;
|
||||
await new Promise<void>((resolve) => server.close(() => resolve()));
|
||||
});
|
||||
|
||||
beforeEach(() => {
|
||||
failuresLeft = 0;
|
||||
requests.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");
|
||||
}
|
||||
|
||||
describe("S3 request timeout retries", () => {
|
||||
it("retries a read that the provider answered with 408 RequestCanceled", async () => {
|
||||
failuresLeft = 1;
|
||||
const object = await provider().getObject({ objectKey: "company/asset.png" });
|
||||
|
||||
expect(await readAll(object.stream)).toBe("hello");
|
||||
expect(requests).toEqual(["GET", "GET"]);
|
||||
});
|
||||
|
||||
it("retries a HEAD request that timed out", async () => {
|
||||
failuresLeft = 1;
|
||||
await expect(provider().headObject({ objectKey: "company/asset.png" })).resolves.toMatchObject({ exists: true });
|
||||
expect(requests).toEqual(["HEAD", "HEAD"]);
|
||||
});
|
||||
|
||||
it("stops at the SDK attempt limit and still reports the failed read", async () => {
|
||||
failuresLeft = 10;
|
||||
const error = await provider().getObject({ objectKey: "company/asset.png" }).catch((err: unknown) => err);
|
||||
|
||||
expect(error).toMatchObject({ name: "RequestCanceled" });
|
||||
expect(requests).toEqual(["GET", "GET", "GET"]);
|
||||
expect(storageFailureDiagnosticsFor(error)).toEqual({
|
||||
provider: "s3",
|
||||
operation: "get_object",
|
||||
errorCode: "RequestCanceled",
|
||||
httpStatus: 408,
|
||||
requestId: "req-408",
|
||||
attempts: 3,
|
||||
});
|
||||
});
|
||||
|
||||
it("does not repeat a write that timed out", async () => {
|
||||
failuresLeft = 1;
|
||||
const error = await provider()
|
||||
.putObject({ objectKey: "k", body: Buffer.from("abc"), contentType: "text/plain", contentLength: 3 })
|
||||
.catch((err: unknown) => err);
|
||||
|
||||
expect(error).toMatchObject({ name: "RequestCanceled" });
|
||||
expect(requests).toEqual(["PUT"]);
|
||||
});
|
||||
});
|
||||
@@ -10,6 +10,7 @@ import {
|
||||
import { putS3Multipart } from "./s3-multipart.js";
|
||||
import { recordS3BodyFailure, withS3FailureDiagnostics } from "./s3-failure.js";
|
||||
import { addAbortSignal, Readable } from "node:stream";
|
||||
import { retryS3RequestTimeouts } from "./s3-request-timeout-retry.js";
|
||||
import type { StorageProvider, GetObjectResult, HeadObjectResult } from "./types.js";
|
||||
import { notFound, unprocessable } from "../errors.js";
|
||||
|
||||
@@ -79,6 +80,7 @@ export function createS3StorageProvider(config: S3ProviderConfig): StorageProvid
|
||||
endpoint: config.endpoint,
|
||||
forcePathStyle: Boolean(config.forcePathStyle),
|
||||
});
|
||||
retryS3RequestTimeouts(client);
|
||||
|
||||
async function listExactVersions(key: string, signal?: AbortSignal) {
|
||||
const versions = new Set<string>(), cursors = new Set<string>();
|
||||
|
||||
@@ -0,0 +1,38 @@
|
||||
import type { S3Client } from "@aws-sdk/client-s3";
|
||||
|
||||
// Methods that S3 defines as safe to repeat without a request body.
|
||||
const REPEATABLE_METHODS = new Set(["GET", "HEAD"]);
|
||||
|
||||
function isRepeatableRequestTimeout(error: unknown, method: unknown): boolean {
|
||||
if (typeof method !== "string" || !REPEATABLE_METHODS.has(method.toUpperCase())) return false;
|
||||
if (!error || typeof error !== "object") return false;
|
||||
const { $metadata, $retryable } = error as { $metadata?: { httpStatusCode?: unknown }; $retryable?: unknown };
|
||||
return $retryable === undefined && $metadata?.httpStatusCode === 408;
|
||||
}
|
||||
|
||||
/**
|
||||
* S3-compatible gateways can answer a read that they stopped waiting on with
|
||||
* HTTP 408 (for example `RequestCanceled`). The SDK's standard strategy retries
|
||||
* only 5xx, throttling and named timeout codes, so one slow read failed the
|
||||
* caller at once. Mark 408 on GET and HEAD as transient. The SDK then applies its
|
||||
* usual backoff, retry quota and attempt limit. Writes keep their current
|
||||
* behavior because a repeated upload needs a replayable body.
|
||||
*/
|
||||
export function retryS3RequestTimeouts(client: S3Client): void {
|
||||
client.middlewareStack.add(
|
||||
(next) => async (args) => {
|
||||
try {
|
||||
return await next(args);
|
||||
} catch (error) {
|
||||
// `finalizeRequest` runs after the SDK built the HTTP request.
|
||||
if (isRepeatableRequestTimeout(error, (args.request as { method?: unknown } | undefined)?.method)) {
|
||||
(error as { $retryable?: { throttling: boolean } }).$retryable = { throttling: false };
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
},
|
||||
// Inside the retry middleware (finalizeRequest, priority high): it sees the
|
||||
// trait before it classifies the error.
|
||||
{ step: "finalizeRequest", priority: "low", name: "paperclipS3RequestTimeoutRetry" },
|
||||
);
|
||||
}
|
||||
Reference in new issue
Block a user