diff --git a/core/src/storage/s3-provider.ts b/core/src/storage/s3-provider.ts index ef01b84..61f1dd7 100644 --- a/core/src/storage/s3-provider.ts +++ b/core/src/storage/s3-provider.ts @@ -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 = {}, +): 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({ diff --git a/core/tests/module-storage.integration.ts b/core/tests/module-storage.integration.ts index 4ac6f3a..d493c60 100644 --- a/core/tests/module-storage.integration.ts +++ b/core/tests/module-storage.integration.ts @@ -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; - 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(" { - const shared = provider(); + // Genuinely needs buckets of its own: the claim is that two modules with the same logical name + // land in two different physical buckets, which a shared bucket could not express. const prefix = disposablePrefix(); const announcements = createModuleStorage({ bucketPrefix: prefix, logicalBucketIds: ["assets"], moduleId: "announcements", - provider: shared, + provider: sharedProvider(), }); const video = createModuleStorage({ bucketPrefix: prefix, logicalBucketIds: ["assets"], moduleId: "video", - provider: shared, + provider: sharedProvider(), }); await announcements.bucket("assets").put("same-key.txt", "from-announcements"); @@ -281,44 +308,32 @@ describe.sequential("live S3-compatible module storage", () => { }); it("accepts the longest bucket name the mapping can produce", async () => { - const shared = provider(); - const prefix = disposablePrefix(); const moduleId = `m${"o".repeat(62)}`; - const bucket = physicalBucketName(prefix, moduleId, `b${"u".repeat(46)}`); + const bucket = physicalBucketName(runPrefix, moduleId, `b${"u".repeat(46)}`); expect(bucket.length).toBeLessThanOrEqual(63); // Proven against the real server rather than against the documented rule. - await expect(shared.ensureBucket(bucket)).resolves.toBeUndefined(); + await expect(sharedProvider().ensureBucket(bucket)).resolves.toBeUndefined(); }); it("supports nested object keys, which modules genuinely use", 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("2026/easter/slides/welcome.png", "x", { contentType: "image/png" }); - expect((await bucket.list({ prefix: "2026/easter/" })).objects.map((o) => o.key)).toEqual([ - "2026/easter/slides/welcome.png", - ]); + await bucket.put(`${namespace}/2026/easter/slides/welcome.png`, "x", { + contentType: "image/png", + }); + expect( + (await bucket.list({ prefix: `${namespace}/2026/easter/` })).objects.map((o) => o.key), + ).toEqual([`${namespace}/2026/easter/slides/welcome.png`]); }); it("probes a reachable provider, and reports an unreachable one as provider-unavailable", async () => { - await expect(provider().probe()).resolves.toBeUndefined(); + await expect(sharedProvider().probe()).resolves.toBeUndefined(); const unreachable = createS3StorageProvider( - createStorageProviderConfig( - { - accessKey: "unused", - // A port nothing listens on: a real connection failure rather than a simulated one. - endpoint: "http://127.0.0.1:1", - secretKey: "unused", - }, - { environment: "development" }, - )!, + // A port nothing listens on: a real connection failure rather than a simulated one. + providerConfig("http://127.0.0.1:1"), ); try { let caught: unknown; @@ -336,14 +351,63 @@ describe.sequential("live S3-compatible module storage", () => { } }); + it("gives up on a provider that accepts the connection and then goes silent", async () => { + // A refused connection fails fast on its own; a provider that *answers* and then stalls is the + // failure with no natural end, and the one that has to be bounded deliberately. A server which + // accepts sockets and never writes a byte reproduces it exactly. + const accepted = new Set(); + const blackHole: Server = createServer((socket) => { + // Deliberately never responds and never closes; only remembered so teardown can cut it off. + accepted.add(socket); + socket.on("close", () => accepted.delete(socket)); + }); + await new Promise((resolve) => blackHole.listen(0, "127.0.0.1", resolve)); + const address = blackHole.address(); + if (address === null || typeof address === "string") { + throw new IntegrationHarnessError("The stalled-provider server did not bind a port."); + } + + const stalled = createS3StorageProvider(providerConfig(`http://127.0.0.1:${address.port}`), { + // The production defaults are deliberately patient; this proves the mechanism, not the number. + maxAttempts: 2, + socketIdleTimeoutMs: 250, + }); + try { + const startedAt = Date.now(); + let caught: unknown; + try { + await stalled.probe(); + } catch (error) { + caught = error; + } + const elapsed = Date.now() - startedAt; + + expect(caught).toBeInstanceOf(ModuleStorageError); + expect((caught as ModuleStorageError).code).toBe("STORAGE_PROVIDER_UNAVAILABLE"); + // The point of the assertion is that it *ended*. Without a socket deadline the SDK waits + // forever and only the test runner's own timeout stops it, which is what a module's request + // handler would do in production with no runner to rescue it. + expect(elapsed).toBeLessThan(5_000); + } finally { + await stalled.close().catch(() => undefined); + // `close()` alone only stops new connections and then waits for established ones to end, + // and the SDK's keep-alive agent holds this one open indefinitely — the never-finished socket + // is the whole point of the server. Cut them off so teardown cannot become its own hang. + for (const socket of accepted) { + socket.destroy(); + } + await new Promise((resolve) => blackHole.close(() => resolve())); + } + }); + it("never deletes a bucket, even when every object in it is removed", async () => { - const shared = provider(); - const prefix = disposablePrefix(); + // Needs a bucket of its own: the claim is about a bucket whose *every* object is gone, which + // cannot be asserted inside a bucket other tests are also writing to. const bucket = createModuleStorage({ - bucketPrefix: prefix, + bucketPrefix: disposablePrefix(), logicalBucketIds: ["assets"], moduleId: "announcements", - provider: shared, + provider: sharedProvider(), }).bucket("assets"); await bucket.put("only.txt", "x"); diff --git a/infrastructure/compose.yaml b/infrastructure/compose.yaml index dc18b5c..4a2935e 100644 --- a/infrastructure/compose.yaml +++ b/infrastructure/compose.yaml @@ -24,7 +24,14 @@ services: object-storage: image: chrislusf/seaweedfs:4.28 restart: unless-stopped - command: ["mini", "-dir=/data"] + # Every S3 bucket becomes its own SeaweedFS collection, and each collection claims a block of + # volumes from a budget the server derives as free-disk ÷ volume-size. Left to auto-configure, + # `mini` picks a production-shaped ~1GB volume, so a machine with a modest amount of free disk + # is left with room for only a handful of buckets — and an exhausted budget is not reported as a + # failed write, it simply stalls. Pinning a small volume size suits the object sizes this + # development and test service actually holds and makes the bucket budget depend on the host's + # disk far less sharply. It caps neither total stored bytes nor individual object size. + command: ["mini", "-dir=/data", "-master.volumeSizeLimitMB=32"] environment: AWS_ACCESS_KEY_ID: ${S3_ACCESS_KEY:?S3_ACCESS_KEY is required} AWS_SECRET_ACCESS_KEY: ${S3_SECRET_KEY:?S3_SECRET_KEY is required}