Implement Phase 2 asset storage pipeline

2026-08-16 19:59:10 -05:00
parent eb19675a5c
commit 722c6cf77d
29 changed files with 5059 additions and 16 deletions
@@ -0,0 +1,397 @@
+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,
+  );
+});