fix: bound provider exchanges and stop the storage suite starving its server
The live storage suite failed in CI with three failures and a 30s runner timeout while passing locally in under three seconds. Two independent defects produced that, one of them a real runtime defect. Provider calls had no deadline. `S3Client` was built with no `requestHandler` and no `maxAttempts`, and the SDK's `connectionTimeout` and `socketTimeout` both default to 0 — disabled. A provider that accepts a connection and then goes silent therefore held the caller open forever, bounded in the suite only by vitest and in production by nothing at all. ADR 0030 promises such a failure reaches a module as the transient `STORAGE_PROVIDER_UNAVAILABLE` it retries through; `translateStorageError` already mapped the SDK's `TimeoutError` onto that code, but the branch was unreachable because nothing ever timed out. The bound is on socket idleness rather than total elapsed time so a module streaming a large object is never cut off for being big, only for going quiet. The suite provisioned a fresh bucket per test. Every S3 bucket becomes its own SeaweedFS collection claiming a block of volumes from a budget the server derives from free disk, and `mini` auto-configures a production-shaped ~1GB volume — so the ceiling is small and varies with the runner. Twelve buckets per run exhausted it, and an exhausted budget is not a failed write: the master retries allocation internally and never replies, which is the hang CI saw. Tests now share one bucket and isolate by key namespace, which exercises prefix scoping at the same time; bucket identity is provisioned only where it is the thing under test. Pinning a small volume size makes the budget depend far less sharply on the host's disk. Verified by starving a server on purpose: the same exhaustion that used to hang until the runner killed it now fails in ~15s with a structured error and no runner timeout, so the failure is bounded rather than hidden. Ten consecutive runs pass in ~2.1s each, using 28 volumes per run against a 5212 budget where the old suite used 84 against 161. Co-Authored-By: Claude Opus 5 <[email protected]>
This commit is contained in:
co-authored by
Claude Opus 5
parent
b9ca0d560d
commit
af4a8009fb
@@ -12,8 +12,8 @@ import {
|
||||
import {
|
||||
type ModuleStorageBody,
|
||||
ModuleStorageError,
|
||||
type ModuleStorageListOptions,
|
||||
type ModuleStorageListing,
|
||||
type ModuleStorageListOptions,
|
||||
type ModuleStorageObject,
|
||||
type ModuleStorageObjectMetadata,
|
||||
type ModuleStoragePutOptions,
|
||||
@@ -135,12 +135,66 @@ function toRequestBody(body: ModuleStorageBody): Uint8Array | string | Readable
|
||||
return body;
|
||||
}
|
||||
|
||||
export function createS3StorageProvider(config: StorageProviderConfig): StorageProvider {
|
||||
/**
|
||||
* How long a single exchange with the provider may take before it is abandoned.
|
||||
*
|
||||
* These exist because the SDK's own defaults are *no deadline at all*: `connectionTimeout` and
|
||||
* `socketTimeout` both default to `0`, so a provider that accepts a connection and then goes silent
|
||||
* holds the calling module's request handler open forever. That is not a hypothetical failure —
|
||||
* an S3-compatible server which has exhausted its storage budget can answer `PutObject` by retrying
|
||||
* allocation internally and never replying at all, which is exactly this shape.
|
||||
*
|
||||
* ADR 0030 promises that a provider failure reaches a module as the transient
|
||||
* `STORAGE_PROVIDER_UNAVAILABLE` it is expected to retry through. That promise is unkeepable
|
||||
* without a deadline: `translateStorageError` already maps the SDK's `TimeoutError` onto that code,
|
||||
* but the branch was unreachable while nothing ever timed out.
|
||||
*
|
||||
* The bound is deliberately on socket *idleness* rather than on total elapsed time. A module
|
||||
* streaming a large object must never be cut off for being big, only for going quiet — an idle
|
||||
* timer resets whenever bytes move, whereas a total-elapsed bound would truncate a healthy large
|
||||
* transfer. A healthy exchange settles in milliseconds against host-local storage and in well under
|
||||
* a second against a remote one, so five seconds without a single byte moving already means
|
||||
* something is wrong. Three attempts keep the SDK's resilience to a genuine blip while capping a
|
||||
* wholly unresponsive provider at roughly fifteen seconds — long enough never to fire spuriously,
|
||||
* short enough that a module's request handler still answers its caller.
|
||||
*/
|
||||
export interface S3TransportBounds {
|
||||
/** Milliseconds allowed to establish the connection before the attempt is abandoned. */
|
||||
readonly connectionTimeoutMs: number;
|
||||
/** Milliseconds the socket may sit idle mid-exchange; resets whenever bytes move. */
|
||||
readonly socketIdleTimeoutMs: number;
|
||||
/** Total attempts, including the first, for a failure the SDK classifies as retryable. */
|
||||
readonly maxAttempts: number;
|
||||
}
|
||||
|
||||
const DEFAULT_TRANSPORT_BOUNDS: S3TransportBounds = Object.freeze({
|
||||
connectionTimeoutMs: 3_000,
|
||||
maxAttempts: 3,
|
||||
socketIdleTimeoutMs: 5_000,
|
||||
});
|
||||
|
||||
export function createS3StorageProvider(
|
||||
config: StorageProviderConfig,
|
||||
/** Tightened only by the live provider suite, so it can prove the deadline actually fires without
|
||||
* spending the production budget waiting for it. Hosts always take the defaults. */
|
||||
bounds: Partial<S3TransportBounds> = {},
|
||||
): StorageProvider {
|
||||
const { connectionTimeoutMs, maxAttempts, socketIdleTimeoutMs } = {
|
||||
...DEFAULT_TRANSPORT_BOUNDS,
|
||||
...bounds,
|
||||
};
|
||||
const client = new S3Client({
|
||||
credentials: { accessKeyId: config.accessKey, secretAccessKey: config.secretKey },
|
||||
endpoint: config.endpoint,
|
||||
forcePathStyle: config.forcePathStyle,
|
||||
maxAttempts,
|
||||
region: config.region,
|
||||
// Not `requestTimeout`: that bounds total elapsed request-and-response time, which would cap a
|
||||
// legitimately large upload or download by size rather than by health.
|
||||
requestHandler: {
|
||||
connectionTimeout: connectionTimeoutMs,
|
||||
socketTimeout: socketIdleTimeoutMs,
|
||||
},
|
||||
});
|
||||
|
||||
return Object.freeze({
|
||||
|
||||
@@ -1,5 +1,10 @@
|
||||
import { randomBytes } from "node:crypto";
|
||||
import { ModuleStorageError } from "@ekklesia/contracts";
|
||||
import { createServer, type Server, type Socket } from "node:net";
|
||||
import {
|
||||
type ModuleStorage,
|
||||
type ModuleStorageBucket,
|
||||
ModuleStorageError,
|
||||
} from "@ekklesia/contracts";
|
||||
import { afterAll, describe, expect, it } from "vitest";
|
||||
import { physicalBucketName } from "../src/storage/binding.js";
|
||||
import { createStorageProviderConfig } from "../src/storage/config.js";
|
||||
@@ -21,6 +26,22 @@ import { createS3StorageProvider } from "../src/storage/s3-provider.js";
|
||||
* (ADR 0004) and because the repository's Compose file already pins and provisions it, so this adds
|
||||
* no new CI infrastructure. Nothing here is SeaweedFS-specific in the *contract* — the same suite
|
||||
* would pass against MinIO or AWS S3, which is the portability claim being tested.
|
||||
*
|
||||
* ## Why this suite is frugal with buckets
|
||||
*
|
||||
* **A bucket is a scarce resource on a real server, so tests share one.** On the pinned SeaweedFS
|
||||
* image every bucket becomes its own storage *collection*, and each collection immediately claims a
|
||||
* block of volumes out of a budget the server derives from free disk. A suite that provisioned a
|
||||
* fresh bucket per test therefore consumed that budget in proportion to its own size, and once the
|
||||
* budget was gone the server did not fail the write — it accepted the connection and retried
|
||||
* allocation internally, forever. That is environment-dependent by construction: it passed on a
|
||||
* roomy disk and hung on a smaller one, which is exactly how it reached CI.
|
||||
*
|
||||
* So bucket identity is provisioned only where bucket identity is the thing under test — module
|
||||
* isolation, the longest representable name, and the never-delete invariant. Everything else shares
|
||||
* one bucket and isolates by **key namespace**, which tests prefix scoping honestly at the same
|
||||
* time. Keep it that way: a new test should reach for `sharedBucket()` and a `keyNamespace()`
|
||||
* unless it genuinely needs a bucket of its own.
|
||||
*/
|
||||
const endpointEnvironment = "EKKLESIA_TEST_STORAGE_ENDPOINT";
|
||||
const accessKeyEnvironment = "EKKLESIA_TEST_STORAGE_ACCESS_KEY";
|
||||
@@ -34,8 +55,7 @@ class IntegrationHarnessError extends Error {
|
||||
}
|
||||
}
|
||||
|
||||
function providerConfig() {
|
||||
const endpoint = process.env[endpointEnvironment];
|
||||
function providerConfig(endpoint = process.env[endpointEnvironment]) {
|
||||
if (typeof endpoint !== "string" || endpoint.trim() === "") {
|
||||
throw new IntegrationHarnessError("The storage integration endpoint is not configured.");
|
||||
}
|
||||
@@ -69,6 +89,39 @@ function provider(): StorageProvider {
|
||||
return created;
|
||||
}
|
||||
|
||||
/** The prefix every shared-bucket test binds through. */
|
||||
const runPrefix = disposablePrefix();
|
||||
|
||||
let cachedProvider: StorageProvider | undefined;
|
||||
|
||||
/** One client for the whole file. Each `S3Client` carries its own connection pool, so building one
|
||||
* per test spent sockets to prove nothing. */
|
||||
function sharedProvider(): StorageProvider {
|
||||
cachedProvider ??= provider();
|
||||
return cachedProvider;
|
||||
}
|
||||
|
||||
let cachedStorage: ModuleStorage | undefined;
|
||||
|
||||
function sharedBucket(): ModuleStorageBucket {
|
||||
cachedStorage ??= createModuleStorage({
|
||||
bucketPrefix: runPrefix,
|
||||
logicalBucketIds: ["announcement-assets"],
|
||||
moduleId: "announcements",
|
||||
provider: sharedProvider(),
|
||||
});
|
||||
return cachedStorage.bucket("announcement-assets");
|
||||
}
|
||||
|
||||
let namespaceCounter = 0;
|
||||
|
||||
/** A key namespace private to one test. `describe.sequential` means these never interleave, so a
|
||||
* counter is both sufficient and readable in a failure message. */
|
||||
function keyNamespace(): string {
|
||||
namespaceCounter += 1;
|
||||
return `case-${namespaceCounter}`;
|
||||
}
|
||||
|
||||
afterAll(async () => {
|
||||
for (const open of openProviders.splice(0)) {
|
||||
await open.close().catch(() => undefined);
|
||||
@@ -77,33 +130,26 @@ afterAll(async () => {
|
||||
|
||||
describe.sequential("live S3-compatible module storage", () => {
|
||||
it("creates a module's bucket idempotently, across both provider dialects", async () => {
|
||||
const shared = provider();
|
||||
const prefix = disposablePrefix();
|
||||
const bucket = physicalBucketName(prefix, "announcements", "announcement-assets");
|
||||
const bucket = physicalBucketName(runPrefix, "announcements", "announcement-assets");
|
||||
|
||||
await shared.ensureBucket(bucket);
|
||||
await sharedProvider().ensureBucket(bucket);
|
||||
// The assertion this whole suite exists for: the pinned SeaweedFS image answers a repeat
|
||||
// `CreateBucket` with `BucketAlreadyExists` rather than AWS's `BucketAlreadyOwnedByYou`, and the
|
||||
// provider must treat both as success. Every restart depends on this.
|
||||
await expect(shared.ensureBucket(bucket)).resolves.toBeUndefined();
|
||||
await expect(shared.ensureBucket(bucket)).resolves.toBeUndefined();
|
||||
await expect(sharedProvider().ensureBucket(bucket)).resolves.toBeUndefined();
|
||||
await expect(sharedProvider().ensureBucket(bucket)).resolves.toBeUndefined();
|
||||
});
|
||||
|
||||
it("round-trips an object through the scoped facade with its content type preserved", async () => {
|
||||
const prefix = disposablePrefix();
|
||||
const storage = createModuleStorage({
|
||||
bucketPrefix: prefix,
|
||||
logicalBucketIds: ["announcement-assets"],
|
||||
moduleId: "announcements",
|
||||
provider: provider(),
|
||||
});
|
||||
const bucket = storage.bucket("announcement-assets");
|
||||
const bucket = sharedBucket();
|
||||
const namespace = keyNamespace();
|
||||
const key = `${namespace}/slides/welcome.png`;
|
||||
const png = new Uint8Array([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a]);
|
||||
|
||||
const written = await bucket.put("slides/welcome.png", png, { contentType: "image/png" });
|
||||
expect(written.key).toBe("slides/welcome.png");
|
||||
const written = await bucket.put(key, png, { contentType: "image/png" });
|
||||
expect(written.key).toBe(key);
|
||||
|
||||
const object = await bucket.get("slides/welcome.png");
|
||||
const object = await bucket.get(key);
|
||||
expect(object.contentType).toBe("image/png");
|
||||
expect(object.contentLength).toBe(png.byteLength);
|
||||
// Unquoted for the caller — providers return a quoted ETag, and every module learning to strip
|
||||
@@ -113,37 +159,27 @@ describe.sequential("live S3-compatible module storage", () => {
|
||||
});
|
||||
|
||||
it("streams a body without buffering, when the length is supplied", async () => {
|
||||
const prefix = disposablePrefix();
|
||||
const bucket = createModuleStorage({
|
||||
bucketPrefix: prefix,
|
||||
logicalBucketIds: ["assets"],
|
||||
moduleId: "media",
|
||||
provider: provider(),
|
||||
}).bucket("assets");
|
||||
const bucket = sharedBucket();
|
||||
const key = `${keyNamespace()}/large.bin`;
|
||||
const payload = "x".repeat(64 * 1024);
|
||||
const stream = new Blob([payload]).stream() as ReadableStream<Uint8Array>;
|
||||
|
||||
await bucket.put("large.bin", stream, {
|
||||
await bucket.put(key, stream, {
|
||||
contentLength: payload.length,
|
||||
contentType: "application/octet-stream",
|
||||
});
|
||||
|
||||
const object = await bucket.get("large.bin");
|
||||
const object = await bucket.get(key);
|
||||
expect(object.contentLength).toBe(payload.length);
|
||||
expect(await new Response(object.body).text()).toBe(payload);
|
||||
});
|
||||
|
||||
it("returns a Web stream a module can hand straight to a Response", async () => {
|
||||
const prefix = disposablePrefix();
|
||||
const bucket = createModuleStorage({
|
||||
bucketPrefix: prefix,
|
||||
logicalBucketIds: ["assets"],
|
||||
moduleId: "announcements",
|
||||
provider: provider(),
|
||||
}).bucket("assets");
|
||||
await bucket.put("welcome.png", new Uint8Array([1, 2, 3]), { contentType: "image/png" });
|
||||
const bucket = sharedBucket();
|
||||
const key = `${keyNamespace()}/welcome.png`;
|
||||
await bucket.put(key, new Uint8Array([1, 2, 3]), { contentType: "image/png" });
|
||||
|
||||
const object = await bucket.get("welcome.png");
|
||||
const object = await bucket.get(key);
|
||||
expect(object.body).toBeInstanceOf(ReadableStream);
|
||||
// The browser-delivery shape this pass settles on: browser → module route → scoped storage,
|
||||
// with no S3 knowledge or credential anywhere near the browser.
|
||||
@@ -155,110 +191,101 @@ describe.sequential("live S3-compatible module storage", () => {
|
||||
});
|
||||
|
||||
it("heads, lists with a prefix, and deletes", async () => {
|
||||
const prefix = disposablePrefix();
|
||||
const bucket = createModuleStorage({
|
||||
bucketPrefix: prefix,
|
||||
logicalBucketIds: ["assets"],
|
||||
moduleId: "announcements",
|
||||
provider: provider(),
|
||||
}).bucket("assets");
|
||||
const bucket = sharedBucket();
|
||||
const namespace = keyNamespace();
|
||||
|
||||
await bucket.put("slides/a.png", "a", { contentType: "image/png" });
|
||||
await bucket.put("slides/b.png", "b", { contentType: "image/png" });
|
||||
await bucket.put("other/c.txt", "c");
|
||||
await bucket.put(`${namespace}/slides/a.png`, "a", { contentType: "image/png" });
|
||||
await bucket.put(`${namespace}/slides/b.png`, "b", { contentType: "image/png" });
|
||||
await bucket.put(`${namespace}/other/c.txt`, "c");
|
||||
|
||||
const metadata = await bucket.head("slides/a.png");
|
||||
const metadata = await bucket.head(`${namespace}/slides/a.png`);
|
||||
expect(metadata.contentType).toBe("image/png");
|
||||
expect(metadata.contentLength).toBe(1);
|
||||
expect(metadata.lastModified).toBeInstanceOf(Date);
|
||||
|
||||
const listing = await bucket.list({ prefix: "slides/" });
|
||||
expect(listing.objects.map((object) => object.key)).toEqual(["slides/a.png", "slides/b.png"]);
|
||||
// The sibling `other/` key proves the prefix genuinely filters rather than the bucket happening
|
||||
// to hold only matching keys.
|
||||
const listing = await bucket.list({ prefix: `${namespace}/slides/` });
|
||||
expect(listing.objects.map((object) => object.key)).toEqual([
|
||||
`${namespace}/slides/a.png`,
|
||||
`${namespace}/slides/b.png`,
|
||||
]);
|
||||
expect(listing.objects[0]?.contentLength).toBe(1);
|
||||
// A complete listing carries no cursor, so an absent cursor always means "done".
|
||||
expect(listing.cursor).toBeUndefined();
|
||||
|
||||
await bucket.delete("slides/a.png");
|
||||
expect((await bucket.list({ prefix: "slides/" })).objects.map((o) => o.key)).toEqual([
|
||||
"slides/b.png",
|
||||
]);
|
||||
await bucket.delete(`${namespace}/slides/a.png`);
|
||||
expect(
|
||||
(await bucket.list({ prefix: `${namespace}/slides/` })).objects.map((o) => o.key),
|
||||
).toEqual([`${namespace}/slides/b.png`]);
|
||||
});
|
||||
|
||||
it("paginates a listing rather than silently truncating it", async () => {
|
||||
const prefix = disposablePrefix();
|
||||
const bucket = createModuleStorage({
|
||||
bucketPrefix: prefix,
|
||||
logicalBucketIds: ["assets"],
|
||||
moduleId: "paged",
|
||||
provider: provider(),
|
||||
}).bucket("assets");
|
||||
const bucket = sharedBucket();
|
||||
const namespace = keyNamespace();
|
||||
for (const index of [0, 1, 2, 3, 4]) {
|
||||
await bucket.put(`object-${index}.txt`, String(index));
|
||||
await bucket.put(`${namespace}/object-${index}.txt`, String(index));
|
||||
}
|
||||
|
||||
const first = await bucket.list({ limit: 2 });
|
||||
const first = await bucket.list({ limit: 2, prefix: `${namespace}/` });
|
||||
expect(first.objects).toHaveLength(2);
|
||||
expect(first.cursor).toBeDefined();
|
||||
|
||||
const keys = [...first.objects.map((object) => object.key)];
|
||||
let cursor = first.cursor;
|
||||
while (cursor !== undefined) {
|
||||
const page = await bucket.list({ cursor, limit: 2 });
|
||||
const page = await bucket.list({ cursor, limit: 2, prefix: `${namespace}/` });
|
||||
keys.push(...page.objects.map((object) => object.key));
|
||||
cursor = page.cursor;
|
||||
}
|
||||
|
||||
expect(keys.sort()).toEqual([
|
||||
"object-0.txt",
|
||||
"object-1.txt",
|
||||
"object-2.txt",
|
||||
"object-3.txt",
|
||||
"object-4.txt",
|
||||
`${namespace}/object-0.txt`,
|
||||
`${namespace}/object-1.txt`,
|
||||
`${namespace}/object-2.txt`,
|
||||
`${namespace}/object-3.txt`,
|
||||
`${namespace}/object-4.txt`,
|
||||
]);
|
||||
});
|
||||
|
||||
it("maps a missing object to a structured not-found error, never a provider exception", async () => {
|
||||
const prefix = disposablePrefix();
|
||||
const bucket = createModuleStorage({
|
||||
bucketPrefix: prefix,
|
||||
logicalBucketIds: ["assets"],
|
||||
moduleId: "announcements",
|
||||
provider: provider(),
|
||||
}).bucket("assets");
|
||||
await bucket.put("present.txt", "x");
|
||||
const bucket = sharedBucket();
|
||||
const namespace = keyNamespace();
|
||||
await bucket.put(`${namespace}/present.txt`, "x");
|
||||
|
||||
let caught: unknown;
|
||||
try {
|
||||
await bucket.get("absent.txt");
|
||||
await bucket.get(`${namespace}/absent.txt`);
|
||||
} catch (error) {
|
||||
caught = error;
|
||||
}
|
||||
|
||||
expect(caught).toBeInstanceOf(ModuleStorageError);
|
||||
expect((caught as ModuleStorageError).code).toBe("STORAGE_OBJECT_NOT_FOUND");
|
||||
expect((caught as ModuleStorageError).key).toBe("absent.txt");
|
||||
expect((caught as ModuleStorageError).key).toBe(`${namespace}/absent.txt`);
|
||||
// Nothing from the provider survives translation: no endpoint, no bucket, no request id, no XML.
|
||||
const rendered = `${String(caught)} ${JSON.stringify(caught)}`;
|
||||
expect(rendered).not.toContain("http");
|
||||
expect(rendered).not.toContain(prefix);
|
||||
expect(rendered).not.toContain(runPrefix);
|
||||
expect(rendered).not.toContain("<?xml");
|
||||
expect(rendered).not.toContain(process.env[accessKeyEnvironment] ?? " | ||||