perf: bound request work and resource usage (#866)

* perf: bound request work and resource usage

* fix: address performance review regressions

* fix: guard content expiration and dashboard state

* fix: tolerate cache outages and retire derived paths

* fix: revalidate expiration claims and status pagination

* fix: retry interrupted repository metadata cleanup

* fix: retire obsolete content cache storage

* fix: activate tree metadata and recheck expiration claims

* fix: fence cache publication across expiration

* fix: recover staged trees and enforce cache lifecycle cleanup

* fix: prune expired transforms during database outages

* fix: finish conference removals and reload expiration candidates

src/core/storage/FileSystem.ts

@@ -1,7 +1,8 @@
+import { appendArchiveEntry } from "../archive-entry";
 import config from "../../config";
 import * as fs from "fs";
 import { Extract } from "unzip-stream";
-import { join, basename, dirname } from "path";
+import { join, dirname } from "path";
 import { Response } from "express";
 import { Readable, pipeline, Transform } from "stream";
 import * as archiver from "archiver";
@@ -144,6 +145,24 @@ export default class FileSystem extends StorageBase {
     });
   }
 
+  async removeLegacyContent(repoId: string): Promise<void> {
+    let directory: fs.Dir;
+    try { directory = await fs.promises.opendir(join(config.FOLDER, this.repoPath(repoId))); }
+    catch (error) {
+      if ((error as NodeJS.ErrnoException).code === "ENOENT") return;
+      throw error;
+    }
+    for await (const entry of directory) {
+      if (entry.name !== "__content" || !entry.isDirectory()) { await this.rm(repoId, entry.name); continue; }
+      const content = await fs.promises.opendir(join(config.FOLDER, this.repoPath(repoId), "__content"));
+      for await (const generation of content) {
+        if (generation.name !== "__retired" && !/^[a-f0-9]{64}$/.test(generation.name)) {
+          await this.rm(repoId, join("__content", generation.name));
+        }
+      }
+    }
+  }
+
   /** @override */
   async mk(repoId: string, dir: string = ""): Promise<void> {
     this.assertSafePath(dir);
@@ -243,26 +262,28 @@ export default class FileSystem extends StorageBase {
   ) {
     this.assertSafePath(dir);
     const archive = archiver(opt?.format || "zip", {});
-    const fullPath = join(config.FOLDER, this.repoPath(repoId), dir);
-
-    await this.listFiles(repoId, dir, {
-      onEntry: async (file) => {
-        let rs: Readable = await this.read(repoId, file.path);
+    let active: Readable | undefined;
+    let stopped = false;
+    archive.once("close", () => { stopped = true; active?.destroy(); });
+    void (async () => {
+      for (const file of await this.listFiles(repoId, dir)) {
+        if (stopped || file.size == null) continue;
+        const filename = join(file.path, file.name);
+        const source = await this.read(repoId, filename);
+        if (stopped) { source.destroy(); return; }
+        active = source;
+        let output = source;
         if (opt?.fileTransformer) {
-          const src = rs;
-          const transformer = opt.fileTransformer(file.path);
-          src.on("error", (err) => transformer.destroy(err));
-          rs = src.pipe(transformer);
+          const transformer = opt.fileTransformer(filename);
+          source.on("error", error => transformer.destroy(error));
+          transformer.once("close", () => source.destroy());
+          output = source.pipe(transformer);
+          active = output;
         }
-        const f = file.path.replace(fullPath, "");
-        archive.append(rs, {
-          name: basename(f),
-          prefix: dirname(f),
-        });
-      },
-    }).then(() => {
-      archive.finalize();
-    });
+        await appendArchiveEntry(archive, output, { name: filename.slice(dir ? dir.length + 1 : 0) });
+      }
+      if (!stopped) await archive.finalize();
+    })().catch(error => archive.destroy(error));
     return archive;
   }
 }

src/core/storage/S3.ts

@@ -1,3 +1,4 @@
+import { appendArchiveEntry } from "../archive-entry";
 import {
   GetObjectCommand,
   ListObjectsV2CommandOutput,
@@ -7,7 +8,7 @@ import {
 import { Upload } from "@aws-sdk/lib-storage";
 import { NodeHttpHandler } from "@smithy/node-http-handler";
 import config from "../../config";
-import { pipeline, Readable, Transform, PassThrough } from "stream";
+import { pipeline, Readable, Transform } from "stream";
 import ArchiveStreamToS3 from "decompress-stream-to-s3";
 import { Response } from "express";
 import { lookup } from "mime-types";
@@ -32,23 +33,30 @@ export default class S3Storage extends StorageBase {
       });
   }
 
+  private clients = new Map<number, S3>();
+
   private client(timeout = 10000) {
+    const existing = this.clients.get(timeout);
+    if (existing) return existing;
     if (!config.S3_CLIENT_ID) throw new Error("S3_CLIENT_ID not set");
     if (!config.S3_CLIENT_SECRET) throw new Error("S3_CLIENT_SECRET not set");
     if (!config.S3_REGION) throw new Error("S3_REGION not set");
     if (!config.S3_ENDPOINT) throw new Error("S3_ENDPOINT not set");
-    return new S3({
+    const client = new S3({
       credentials: {
         accessKeyId: config.S3_CLIENT_ID,
         secretAccessKey: config.S3_CLIENT_SECRET,
       },
       region: config.S3_REGION,
       endpoint: config.S3_ENDPOINT,
+      forcePathStyle: true,
       requestHandler: new NodeHttpHandler({
         requestTimeout: timeout,
         connectionTimeout: timeout,
       }),
     });
+    this.clients.set(timeout, client);
+    return client;
   }
 
   /** @override */
@@ -82,6 +90,18 @@ export default class S3Storage extends StorageBase {
     if (!config.S3_BUCKET) throw new Error("S3_BUCKET not set");
     this.assertSafePath(dir);
     const prefix = join(this.repoPath(repoId), dir).replace(/\/$/, "");
+    await this.removeMatching(prefix, key => key === prefix || key.startsWith(prefix + "/"));
+  }
+
+  async removeLegacyContent(repoId: string): Promise<void> {
+    if (!config.S3_BUCKET) throw new Error("S3_BUCKET not set");
+    const prefix = this.repoPath(repoId);
+    await this.removeMatching(prefix, key => key.startsWith(prefix) &&
+      !/^__content\/(?:[a-f0-9]{64}|__retired)(?:\/|$)/.test(key.slice(prefix.length)));
+  }
+
+  private async removeMatching(prefix: string, shouldRemove: (key: string) => boolean) {
+    if (!config.S3_BUCKET) throw new Error("S3_BUCKET not set");
     const client = this.client(200000);
     let continuationToken: string | undefined;
     do {
@@ -92,7 +112,7 @@ export default class S3Storage extends StorageBase {
         ContinuationToken: continuationToken,
       });
       const objects = (data.Contents || [])
-        .filter(({ Key }) => Key === prefix || Key?.startsWith(prefix + "/"))
+        .filter(({ Key }) => Key && shouldRemove(Key))
         .map(({ Key }) => ({ Key: Key! }));
       if (objects.length) {
         const result = await client.deleteObjects({
@@ -212,11 +232,14 @@ export default class S3Storage extends StorageBase {
     let bytesWritten = -1;
     if (typeof expectedSize === "number" && expectedSize > 0 && typeof data !== "string") {
       bytesWritten = 0;
-      const counter = new PassThrough();
-      counter.on("data", (chunk: Buffer) => {
-        bytesWritten += chunk.length;
+      const counter = new Transform({
+        transform(chunk, _encoding, callback) { bytesWritten += chunk.length; callback(null, chunk); },
+        flush(callback) {
+          callback(bytesWritten < expectedSize! ? new AnonymousError("storage_write_size_mismatch", { httpStatus: 502 }) : undefined);
+        },
       });
       data.on("error", (err: Error) => counter.destroy(err));
+      counter.once("close", () => (data as Readable).destroy());
       data.pipe(counter);
       body = counter;
     }
@@ -351,39 +374,36 @@ export default class S3Storage extends StorageBase {
     const archive = archiver(opt?.format || "zip", {});
     if (dir && dir[dir.length - 1] != "/") dir = dir + "/";
 
-    let req: ListObjectsV2CommandOutput;
-    let nextContinuationToken: string | undefined;
-    do {
-      req = await this.client(30000).listObjectsV2({
-        Bucket: config.S3_BUCKET,
-        Prefix: join(this.repoPath(repoId), dir),
-        MaxKeys: 250,
-        ContinuationToken: nextContinuationToken,
-      });
-
-      nextContinuationToken = req.NextContinuationToken;
-      for (const f of req.Contents || []) {
-        if (!f.Key) continue;
-        const filename = basename(f.Key);
-        const prefix = dirname(
-          f.Key.replace(join(this.repoPath(repoId), dir), "")
-        );
-
-        let rs: Readable = await this.read(repoId, f.Key);
-        if (opt?.fileTransformer) {
-          const src = rs;
-          const transformer = opt.fileTransformer(f.Key);
-          src.on("error", (err) => transformer.destroy(err));
-          rs = src.pipe(transformer);
+    let active: Readable | undefined;
+    let stopped = false;
+    archive.once("close", () => { stopped = true; active?.destroy(); });
+    void (async () => {
+      let next: string | undefined;
+      do {
+        const listing = await this.client(30000).listObjectsV2({ Bucket: config.S3_BUCKET!,
+          Prefix: join(this.repoPath(repoId), dir), MaxKeys: 250, ContinuationToken: next });
+        next = listing.IsTruncated ? listing.NextContinuationToken : undefined;
+        for (const file of listing.Contents || []) {
+          if (stopped) return;
+          if (!file.Key || file.Key.endsWith("/")) continue;
+          const filename = file.Key.slice(this.repoPath(repoId).length);
+          const source = await this.read(repoId, filename);
+          if (stopped) { source.destroy(); return; }
+          active = source;
+          let output = source;
+          if (opt?.fileTransformer) {
+            const transformer = opt.fileTransformer(filename);
+            source.on("error", error => transformer.destroy(error));
+            transformer.once("close", () => source.destroy());
+            output = source.pipe(transformer);
+            active = output;
+          }
+          await appendArchiveEntry(archive, output, { name: filename.slice(dir.length) });
         }
-
-        archive.append(rs, {
-          name: filename,
-          prefix,
-        });
-      }
-    } while (req && req.Contents?.length && req.IsTruncated);
-    archive.finalize();
+        if (listing.IsTruncated && !next) throw new Error("Invalid S3 continuation token");
+      } while (next && !stopped);
+      if (!stopped) await archive.finalize();
+    })().catch(error => archive.destroy(error));
     return archive;
   }
 }

src/core/storage/Storage.ts

@@ -120,6 +120,9 @@ export default abstract class StorageBase {
    */
   abstract mk(repoId: string, dir: string): Promise<void>;
 
+  /** Remove unversioned stream caches, preserving the managed content namespace. */
+  abstract removeLegacyContent(repoId: string): Promise<void>;
+
   repoPath(repoId: string) {
     return (
       join(repoId, "original") + (process.platform === "win32" ? "\\" : "/")