const test = require("node:test"); const assert = require("node:assert/strict"); const http = require("node:http"); const { PassThrough, Readable } = require("node:stream"); const { StorageObjectNotFoundError, StorageUnavailableError, createStorageService, } = require("../src/storage"); const storageConfig = { endpoint: "http://seaweedfs:8333", region: "us-east-1", bucket: "announcements-v2", accessKeyId: "test-key", secretAccessKey: "test-secret", forcePathStyle: true, }; function fakeClient(handler) { const commands = []; return { commands, async send(command) { commands.push(command); return handler(command, commands.length - 1); }, }; } function commandName(command) { return command.constructor.name; } test("storage initialization probes an existing bucket without creating it", async () => { const client = fakeClient(async () => ({})); const service = createStorageService(storageConfig, { client }); const health = await service.initialize(); assert.equal(health.status, "healthy"); assert.match(health.checkedAt, /T/); assert.equal(health.lastSuccessAt, health.checkedAt); assert.deepEqual(client.commands.map(commandName), ["HeadBucketCommand"]); }); test("storage initialization creates only a missing bucket", async () => { let exists = false; const missing = Object.assign(new Error("missing"), { name: "NoSuchBucket", $metadata: { httpStatusCode: 404 }, }); const client = fakeClient(async (command) => { if (commandName(command) === "HeadBucketCommand" && !exists) throw missing; if (commandName(command) === "CreateBucketCommand") exists = true; return {}; }); const service = createStorageService(storageConfig, { client }); assert.deepEqual(await service.ensureBucket(), { created: true }); assert.deepEqual(client.commands.map(commandName), [ "HeadBucketCommand", "CreateBucketCommand", ]); assert.equal(client.commands[1].input.Bucket, storageConfig.bucket); assert.deepEqual(await service.ensureBucket(), { created: false }); assert.deepEqual(client.commands.map(commandName), [ "HeadBucketCommand", "CreateBucketCommand", "HeadBucketCommand", ]); }); test("storage start performs a nonfatal immediate check and stop releases its client", async () => { let destroyed = false; const client = fakeClient(async () => ({})); client.destroy = () => { destroyed = true; }; const service = createStorageService( { ...storageConfig, healthProbeIntervalMs: 1_000 }, { client }, ); const health = await service.start(); assert.equal(health.status, "healthy"); assert.deepEqual(client.commands.map(commandName), ["HeadBucketCommand"]); service.stop(); assert.equal(destroyed, true); }); test("a create-bucket race is healthy only after the bucket can be verified", async () => { const missing = Object.assign(new Error("missing"), { name: "NoSuchBucket", $metadata: { httpStatusCode: 404 }, }); const conflict = Object.assign(new Error("race"), { name: "BucketAlreadyOwnedByYou", $metadata: { httpStatusCode: 409 }, }); let headCalls = 0; const client = fakeClient(async (command) => { if (commandName(command) === "HeadBucketCommand") { headCalls += 1; if (headCalls === 1) throw missing; return {}; } if (commandName(command) === "CreateBucketCommand") throw conflict; throw new Error(`Unexpected ${commandName(command)}`); }); const service = createStorageService(storageConfig, { client }); assert.deepEqual(await service.ensureBucket(), { created: false }); assert.equal(service.health().status, "healthy"); assert.deepEqual(client.commands.map(commandName), [ "HeadBucketCommand", "CreateBucketCommand", "HeadBucketCommand", ]); }); test("an older failed probe cannot overwrite a newer successful operation", async () => { let rejectProbe; let notifyProbeStarted; const probeStarted = new Promise((resolve) => { notifyProbeStarted = resolve; }); let headCalls = 0; const client = fakeClient(async (command) => { if (commandName(command) === "HeadBucketCommand") { headCalls += 1; if (headCalls === 1) return {}; notifyProbeStarted(); return new Promise((resolve, reject) => { rejectProbe = reject; }); } if (commandName(command) === "PutObjectCommand") return {}; throw new Error(`Unexpected ${commandName(command)}`); }); const service = createStorageService(storageConfig, { client }); await service.initialize(); const staleProbe = service.probe(); await probeStarted; await service.putObject({ key: "assets/id/original/hash", body: Buffer.from("bytes"), contentLength: 5, }); const successfulState = service.health(); rejectProbe(new Error("older request failed")); await staleProbe; assert.equal(successfulState.status, "healthy"); assert.deepEqual(service.health(), successfulState); }); test("temporary storage failure degrades without throwing during initialization and recovers", async () => { let available = false; const warnings = []; const client = fakeClient(async () => { if (!available) throw new Error("connection refused at a private address"); return {}; }); const service = createStorageService(storageConfig, { client, logger: { warn(message) { warnings.push(message); } }, }); const initial = await service.initialize(); assert.equal(initial.status, "degraded"); assert.equal(initial.code, "STORAGE_UNAVAILABLE"); assert.equal(initial.lastSuccessAt, null); assert.deepEqual(warnings, ["Object storage is unavailable during initialization"]); available = true; const recovered = await service.probe(); assert.equal(recovered.status, "healthy"); assert.equal(recovered.lastSuccessAt, recovered.checkedAt); assert.equal(service.health().status, "healthy"); available = false; const degradedAgain = await service.probe(); assert.equal(degradedAgain.status, "degraded"); assert.equal(degradedAgain.lastSuccessAt, recovered.lastSuccessAt); }); test("put, get, head, existence, and delete use the configured private bucket", async () => { const body = Readable.from([Buffer.from("image bytes")]); const client = fakeClient(async (command) => { switch (commandName(command)) { case "PutObjectCommand": return { ETag: '"etag"' }; case "GetObjectCommand": return { Body: body, ContentType: "image/webp", ContentLength: 11, ETag: '"etag"', }; case "HeadObjectCommand": return { ContentType: "image/webp", ContentLength: 11, ETag: '"etag"', }; case "DeleteObjectCommand": return {}; case "HeadBucketCommand": return {}; default: throw new Error(`Unexpected ${commandName(command)}`); } }); const service = createStorageService(storageConfig, { client }); const key = "assets/id/display/hash.webp"; await service.initialize(); client.commands.length = 0; const put = await service.putObject({ key, body: Buffer.from("image bytes"), contentType: "image/webp", contentLength: 11, }); assert.deepEqual(put, { key, etag: '"etag"', versionId: null }); const fetched = await service.getObject(key); assert.equal(fetched.body, body); assert.equal(fetched.contentType, "image/webp"); assert.equal(fetched.contentLength, 11); assert.equal((await service.headObject(key)).etag, '"etag"'); assert.equal(await service.objectExists(key), true); await service.deleteObject(key); for (const command of client.commands) { assert.equal(command.input.Bucket, storageConfig.bucket); assert.equal(command.input.Key, key); } }); test("missing objects are distinct from unavailable storage", async () => { let missing = true; const client = fakeClient(async (command) => { if (commandName(command) === "HeadBucketCommand") return {}; if (missing) { throw Object.assign(new Error("missing"), { name: "NoSuchKey", $metadata: { httpStatusCode: 404 }, }); } throw new Error("network offline"); }); const service = createStorageService(storageConfig, { client }); await service.initialize(); client.commands.length = 0; assert.equal(await service.headObject("assets/id/original/hash"), null); assert.equal(await service.objectExists("assets/id/original/hash"), false); await assert.rejects( service.getObject("assets/id/original/hash"), StorageObjectNotFoundError, ); assert.equal(service.health().status, "healthy"); missing = false; await assert.rejects( service.putObject({ key: "assets/id/original/hash", body: Buffer.alloc(0) }), (error) => error instanceof StorageUnavailableError && error.code === "STORAGE_UNAVAILABLE" && error.statusCode === 503, ); assert.equal(service.health().status, "degraded"); }); test("a missing bucket during object access is a storage outage, not a missing object", async () => { let initialized = false; const client = fakeClient(async (command) => { if (commandName(command) === "HeadBucketCommand" && !initialized) { initialized = true; return {}; } throw Object.assign(new Error("bucket missing"), { name: "NoSuchBucket", $metadata: { httpStatusCode: 404 }, }); }); const service = createStorageService(storageConfig, { client }); await service.initialize(); await assert.rejects( service.getObject("assets/id/original/hash"), StorageUnavailableError, ); assert.equal(service.health().status, "degraded"); }); test("storage rejects unsafe internal object keys before contacting S3", async () => { const client = fakeClient(async () => ({})); const service = createStorageService(storageConfig, { client }); await assert.rejects( service.getObject("../private"), /Storage object key is invalid/, ); await assert.rejects( service.deleteObject("/absolute"), /Storage object key is invalid/, ); assert.equal(client.commands.length, 0); }); test("real HTTP requests time out instead of hanging indefinitely", async (t) => { const server = http.createServer(() => { // Intentionally accept the connection without sending response headers. }); server.listen(0, "127.0.0.1"); await new Promise((resolve) => server.once("listening", resolve)); t.after(async () => { server.closeAllConnections(); await new Promise((resolve) => server.close(resolve)); }); const service = createStorageService({ ...storageConfig, endpoint: `http://127.0.0.1:${server.address().port}`, connectionTimeoutMs: 100, requestTimeoutMs: 100, maxAttempts: 1, }); t.after(() => service.stop()); const startedAt = Date.now(); const health = await service.probe(); assert.equal(health.status, "degraded"); assert.equal(health.code, "STORAGE_UNAVAILABLE"); assert.ok(Date.now() - startedAt < 2_000); }); test("consumer stream cancellation does not falsely degrade storage", async () => { let body = new PassThrough(); const client = fakeClient(async (command) => { if (commandName(command) === "HeadBucketCommand") return {}; if (commandName(command) === "GetObjectCommand") { return { Body: body, ContentLength: 10, ContentType: "image/webp" }; } throw new Error(`Unexpected ${commandName(command)}`); }); const service = createStorageService(storageConfig, { client }); await service.initialize(); const cancelled = await service.getObject("assets/id/display/hash.webp"); cancelled.body.destroy( Object.assign(new Error("consumer closed"), { code: "ERR_STREAM_PREMATURE_CLOSE", }), ); await new Promise((resolve) => setImmediate(resolve)); assert.equal(service.health().status, "healthy"); body = new PassThrough(); const failed = await service.getObject("assets/id/display/hash.webp"); failed.body.destroy(Object.assign(new Error("upstream reset"), { code: "ECONNRESET" })); await new Promise((resolve) => setImmediate(resolve)); assert.equal(service.health().status, "degraded"); }); test("delete treats an already-missing object as an idempotent success", async () => { let missingBucket = false; const client = fakeClient(async (command) => { if (commandName(command) === "HeadBucketCommand") return {}; if (commandName(command) === "DeleteObjectCommand") { throw Object.assign(new Error("missing"), { name: missingBucket ? "NoSuchBucket" : "NoSuchKey", $metadata: { httpStatusCode: 404 }, }); } throw new Error(`Unexpected ${commandName(command)}`); }); const service = createStorageService(storageConfig, { client }); await service.initialize(); await assert.doesNotReject( service.deleteObject("assets/id/original/hash"), ); assert.equal(service.health().status, "healthy"); missingBucket = true; await assert.rejects( service.deleteObject("assets/id/original/hash"), StorageUnavailableError, ); });