mirror of
https://github.com/tdurieux/anonymous_github.git
synced 2026-11-14 00:07:00 +01:00
Merge pull request #915 from tdurieux/feat/shared-blob-cache
perf: deduplicate cached files across projects and track cache reuse
This commit is contained in:
31 files changed
+2107
-42
No files matched your search
@@ -4,6 +4,19 @@ import { randomBytes } from "crypto";
|
||||
interface Config {
|
||||
RESEND_API_KEY: string;
|
||||
EMAIL_FROM: string;
|
||||
|
||||
SHARED_CACHE_ENABLED: boolean;
|
||||
SHARED_CACHE_PATH: string;
|
||||
SHARED_CACHE_BYTES: number;
|
||||
SHARED_CACHE_MAX_FILE_BYTES: number;
|
||||
SHARED_CACHE_GZIP: boolean;
|
||||
SHARED_CACHE_GZIP_MIN_SAVING: number;
|
||||
SHARED_CACHE_FREE_BYTES: number;
|
||||
SHARED_CACHE_LOW_WATER: number;
|
||||
SHARED_CACHE_DECISIONS: boolean;
|
||||
SHARED_CACHE_RESULT_BYTES: number;
|
||||
CACHE_MONITORING_ENABLED: boolean;
|
||||
|
||||
GITHUB_APP_ENABLED: boolean;
|
||||
GITHUB_APP_NEW_CONNECTIONS: boolean;
|
||||
GITHUB_OAUTH_ENABLED: boolean;
|
||||
@@ -71,6 +84,19 @@ interface Config {
|
||||
const config: Config = {
|
||||
RESEND_API_KEY: "",
|
||||
EMAIL_FROM: "",
|
||||
|
||||
SHARED_CACHE_ENABLED: false,
|
||||
SHARED_CACHE_PATH: resolve(process.env.FOLDER || resolve(__dirname, "..", "repositories"), ".shared-cache-v1"),
|
||||
SHARED_CACHE_BYTES: 10 * 1024 ** 3,
|
||||
SHARED_CACHE_MAX_FILE_BYTES: 32 * 1024 ** 2,
|
||||
SHARED_CACHE_GZIP: true,
|
||||
SHARED_CACHE_GZIP_MIN_SAVING: 0.10,
|
||||
SHARED_CACHE_FREE_BYTES: 1024 ** 3,
|
||||
SHARED_CACHE_LOW_WATER: 0.9,
|
||||
SHARED_CACHE_DECISIONS: true,
|
||||
SHARED_CACHE_RESULT_BYTES: 512 * 1024 ** 2,
|
||||
CACHE_MONITORING_ENABLED: true,
|
||||
|
||||
GITHUB_APP_ENABLED: false,
|
||||
GITHUB_APP_NEW_CONNECTIONS: true,
|
||||
GITHUB_OAUTH_ENABLED: true,
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import { notifyOwnerAccessProblem } from "./owner-notifications";
|
||||
import { sendSharedContent } from "./cached-response";
|
||||
import { streamResponse } from "./response-stream";
|
||||
import { join, basename, dirname } from "path";
|
||||
import { Response } from "express";
|
||||
@@ -454,7 +455,8 @@ export default class AnonymizedFile {
|
||||
const resStream = got
|
||||
.stream(join(config.STREAMER_ENTRYPOINT, "api"), {
|
||||
method: "POST",
|
||||
headers: requestHeaders(),
|
||||
headers: { ...requestHeaders(), "accept-encoding": res.req?.get?.("accept-encoding") || "identity" },
|
||||
decompress: false,
|
||||
json: {
|
||||
sha,
|
||||
size,
|
||||
@@ -477,6 +479,10 @@ export default class AnonymizedFile {
|
||||
resStream.on("response", (upstream: { headers: Record<string, string | string[] | undefined> }) => {
|
||||
headersDone();
|
||||
if (res.headersSent) return;
|
||||
for (const header of ["content-encoding", "content-length", "vary", "accept-ranges"]) {
|
||||
const value = upstream.headers[header];
|
||||
if (typeof value === "string") res.setHeader(header, value);
|
||||
}
|
||||
const ct = upstream.headers["content-type"];
|
||||
if (typeof ct === "string") {
|
||||
res.contentType(ct);
|
||||
@@ -530,6 +536,7 @@ export default class AnonymizedFile {
|
||||
});
|
||||
const content = await this.content();
|
||||
if (res.destroyed || res.writableEnded) { content.destroy(); anonymizer.destroy(); resolve(); return; }
|
||||
if (await sendSharedContent(content, res, anonymizer.opt)) { anonymizer.destroy(); resolve(); return; }
|
||||
function handleStreamError(error: Error) {
|
||||
if (!content.closed && !content.destroyed) {
|
||||
content.destroy();
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
import { observeCache } from "./cache-observability";
|
||||
import { performance } from "perf_hooks";
|
||||
import { createHash } from "crypto";
|
||||
import { transformedFile } from "./transformed-cache";
|
||||
import { promises as fs, createReadStream, ReadStream } from "fs";
|
||||
@@ -105,7 +107,7 @@ function isTextualMime(mime: string): boolean {
|
||||
// Name-only classification: returns true (known text), false (known binary),
|
||||
// or null when the name alone is inconclusive. The streaming transformer
|
||||
// resolves null by sniffing the first chunk with isbinaryfile.
|
||||
function classifyByName(filePath: string): boolean | null {
|
||||
export function classifyByName(filePath: string): boolean | null {
|
||||
const name = basename(filePath);
|
||||
const extension = name.split(".").reverse()[0].toLowerCase();
|
||||
if (config.additionalExtensions.includes(extension)) return true;
|
||||
@@ -143,6 +145,7 @@ export class AnonymizeTransformer extends Transform {
|
||||
anonimizer: ContentAnonimizer;
|
||||
private wholeChunks: Buffer[] = [];
|
||||
private wholeBytes = 0;
|
||||
private passthroughBytes = 0;
|
||||
private digest = createHash("sha256");
|
||||
private spool?: string;
|
||||
private reservedSpool = 0;
|
||||
@@ -157,6 +160,7 @@ export class AnonymizeTransformer extends Transform {
|
||||
filePath: string;
|
||||
cacheGeneration?: string;
|
||||
cacheRevision?: string;
|
||||
decisionManaged?: boolean;
|
||||
} & ConstructorParameters<typeof ContentAnonimizer>[0]
|
||||
) {
|
||||
super();
|
||||
@@ -200,6 +204,7 @@ export class AnonymizeTransformer extends Transform {
|
||||
return;
|
||||
}
|
||||
} else {
|
||||
this.passthroughBytes += chunk.length;
|
||||
this.emit("transform", { isText: this.isText, wasAnonimized: false, chunk });
|
||||
this.push(chunk);
|
||||
}
|
||||
@@ -237,13 +242,17 @@ export class AnonymizeTransformer extends Transform {
|
||||
}
|
||||
|
||||
_flush(callback: (error?: Error) => void) {
|
||||
if (this.passthroughBytes) observeCache(this.isText ? "inspected_passthrough" : "inspected_binary", this.passthroughBytes);
|
||||
if (this.spool) {
|
||||
const output = join(this.spool, "output");
|
||||
void transformedFile(this.digest.digest("hex"), this.opt, output, () => anonymizeOnWorker({ input: join(this.spool!, "input"), output,
|
||||
options: this.opt, maxOutput: config.MAX_FILE_SIZE * 4, context: { mask: config.ANONYMIZATION_MASK, hostname: config.APP_HOSTNAME } }, this.wholeBytes, this.cancelled.signal))
|
||||
const started = performance.now();
|
||||
const produce = () => anonymizeOnWorker({ input: join(this.spool!, "input"), output,
|
||||
options: this.opt, maxOutput: config.MAX_FILE_SIZE * 4, context: { mask: config.ANONYMIZATION_MASK, hostname: config.APP_HOSTNAME } }, this.wholeBytes, this.cancelled.signal);
|
||||
void (this.opt.decisionManaged ? produce() : transformedFile(this.digest.digest("hex"), this.opt, output, produce))
|
||||
.then(changed => {
|
||||
if (this.destroyed) return;
|
||||
this.anonimizer.wasAnonymized = changed;
|
||||
observeCache(changed ? "inspected_changed" : "inspected_unchanged", this.wholeBytes, performance.now() - started);
|
||||
this.output = createReadStream(output);
|
||||
this.output.on("data", chunk => {
|
||||
this.emit("transform", { isText: true, wasAnonimized: changed, chunk });
|
||||
@@ -260,7 +269,9 @@ export class AnonymizeTransformer extends Transform {
|
||||
const original = Buffer.concat(this.wholeChunks, this.wholeBytes);
|
||||
this.wholeChunks = [];
|
||||
const text = original.toString("utf8");
|
||||
const started = performance.now();
|
||||
const out = this.anonimizer.anonymize(text);
|
||||
observeCache(out === text ? "inspected_unchanged" : "inspected_changed", original.length, performance.now() - started);
|
||||
// Preserve the original encoding when no replacement was needed.
|
||||
const chunk = out === text ? original : Buffer.from(out, "utf8");
|
||||
this.emit("transform", { isText: true, wasAnonimized: this.wasAnonimized, chunk });
|
||||
|
||||
@@ -0,0 +1,423 @@
|
||||
import { DatabaseSync } from "node:sqlite";
|
||||
import { createHash, randomUUID } from "crypto";
|
||||
import { createReadStream, createWriteStream, mkdirSync, unlinkSync, rmdirSync, statfsSync, promises as fs } from "fs";
|
||||
import { join } from "path";
|
||||
import { Readable, Transform } from "stream";
|
||||
import { pipeline } from "stream/promises";
|
||||
import { createGzip, createGunzip } from "zlib";
|
||||
import { performance } from "perf_hooks";
|
||||
import config from "../config";
|
||||
import { observeCache } from "./cache-observability";
|
||||
|
||||
export class CacheAdmissionError extends Error {}
|
||||
export function isCacheContention(error: unknown) {
|
||||
const code = Number((error as { errcode?: number })?.errcode) & 255;
|
||||
return code === 5 || code === 6;
|
||||
}
|
||||
function cacheError(error: unknown): never {
|
||||
if (isCacheContention(error)) throw new CacheAdmissionError("cache_index_busy");
|
||||
throw error;
|
||||
}
|
||||
export interface BlobRecord { digest: string; file: string; size: number; stored: number; allocated: number; encoding: "gzip" | "identity"; accessed: number }
|
||||
export interface BlobProof { gitSha: string; sha256: string; size: number }
|
||||
export interface BlobOptions { root: string; budget: number; maxObject: number; gzip?: boolean; saving?: number; freeBytes?: number; lowWater?: number; name?: string; reservationRoot?: string }
|
||||
export type BlobStream = Readable & { cachedBlob?: { store: BlobCache; record: BlobRecord } };
|
||||
const validDigest = (s: string) => /^(git-[a-f0-9]{40}|sha256-[a-f0-9]{64})$/.test(s);
|
||||
|
||||
/** One local filesystem shared by all processes. SQLite WAL coordinates metadata, fills and reader leases. */
|
||||
export class BlobCache {
|
||||
readonly db: DatabaseSync;
|
||||
readonly name: string;
|
||||
private disk?: DatabaseSync;
|
||||
constructor(readonly options: BlobOptions) {
|
||||
if (!(options.budget > 0 && options.maxObject > 0)) throw new Error("invalid_cache_budget");
|
||||
this.name = options.name || "source";
|
||||
mkdirSync(join(options.root, "tmp"), { recursive: true, mode: 0o700 });
|
||||
this.db = new DatabaseSync(join(options.root, "index.sqlite"));
|
||||
try {
|
||||
this.db.exec(`PRAGMA busy_timeout=5000; PRAGMA journal_mode=WAL;
|
||||
CREATE TABLE IF NOT EXISTS blobs(digest TEXT PRIMARY KEY,file TEXT,size INTEGER,stored INTEGER,encoding TEXT,accessed INTEGER,allocated INTEGER);
|
||||
CREATE INDEX IF NOT EXISTS blobs_lru ON blobs(accessed);
|
||||
CREATE TABLE IF NOT EXISTS aliases(key TEXT PRIMARY KEY,digest TEXT REFERENCES blobs(digest) ON DELETE CASCADE);
|
||||
CREATE INDEX IF NOT EXISTS aliases_digest ON aliases(digest);
|
||||
CREATE TABLE IF NOT EXISTS fills(key TEXT PRIMARY KEY,token TEXT,expires INTEGER,reserved INTEGER);
|
||||
CREATE TABLE IF NOT EXISTS publications(token TEXT PRIMARY KEY,digest TEXT,file TEXT);
|
||||
CREATE TABLE IF NOT EXISTS readers(token TEXT PRIMARY KEY,digest TEXT,expires INTEGER);
|
||||
CREATE INDEX IF NOT EXISTS readers_digest ON readers(digest);
|
||||
CREATE INDEX IF NOT EXISTS readers_expiry ON readers(expires);
|
||||
CREATE TABLE IF NOT EXISTS decisions(key TEXT PRIMARY KEY,digest TEXT,changed INTEGER,accessed INTEGER,expires INTEGER);
|
||||
CREATE INDEX IF NOT EXISTS decisions_age ON decisions(accessed);
|
||||
CREATE INDEX IF NOT EXISTS decisions_expiry ON decisions(expires);
|
||||
CREATE TABLE IF NOT EXISTS maintenance(id INTEGER PRIMARY KEY,nextScan INTEGER);
|
||||
CREATE TABLE IF NOT EXISTS history(key TEXT PRIMARY KEY,seen INTEGER,evicted INTEGER);
|
||||
CREATE INDEX IF NOT EXISTS history_seen ON history(seen);
|
||||
CREATE INDEX IF NOT EXISTS blobs_file ON blobs(file);
|
||||
CREATE TABLE IF NOT EXISTS totals(id INTEGER PRIMARY KEY, bytes INTEGER);
|
||||
CREATE TRIGGER IF NOT EXISTS blob_added AFTER INSERT ON blobs BEGIN UPDATE totals SET bytes=bytes+NEW.allocated WHERE id=1; END;
|
||||
CREATE TRIGGER IF NOT EXISTS blob_removed AFTER DELETE ON blobs BEGIN UPDATE totals SET bytes=bytes-OLD.allocated WHERE id=1; END;
|
||||
PRAGMA foreign_keys=ON; PRAGMA recursive_triggers=ON;`);
|
||||
// Initialize once under a write lock; all connections see data and counters atomically.
|
||||
this.transaction(() => {
|
||||
// Classification is nullable for decisions written by older versions.
|
||||
if (!this.db.prepare("PRAGMA table_info(decisions)").all().some(row => row.name === "isText"))
|
||||
this.db.exec("ALTER TABLE decisions ADD COLUMN isText INTEGER");
|
||||
if (!this.db.prepare("SELECT 1 FROM totals WHERE id=1").get()) this.db.exec("INSERT INTO totals SELECT 1,COALESCE(SUM(allocated),0) FROM blobs");
|
||||
this.db.exec(`CREATE TABLE IF NOT EXISTS summary(id INTEGER PRIMARY KEY,blobs INTEGER,originalBytes INTEGER,storedBytes INTEGER,allocatedBytes INTEGER,activeReaders INTEGER,decisions INTEGER,aliases INTEGER,history INTEGER,reservedBytes INTEGER)`);
|
||||
if (!this.db.prepare("SELECT 1 FROM summary WHERE id=1").get()) {
|
||||
this.db.exec(`INSERT INTO summary SELECT 1,COUNT(*),COALESCE(SUM(size),0),COALESCE(SUM(stored),0),COALESCE(SUM(allocated),0),
|
||||
(SELECT COUNT(*) FROM readers),(SELECT COUNT(*) FROM decisions),(SELECT COUNT(*) FROM aliases),(SELECT COUNT(*) FROM history),(SELECT COALESCE(SUM(reserved),0) FROM fills) FROM blobs`);
|
||||
}
|
||||
for (const [table, columns] of [
|
||||
["blobs", { blobs: "1", originalBytes: "size", storedBytes: "stored", allocatedBytes: "allocated" }],
|
||||
["readers", { activeReaders: "1" }], ["decisions", { decisions: "1" }],
|
||||
["aliases", { aliases: "1" }], ["history", { history: "1" }], ["fills", { reservedBytes: "reserved" }]
|
||||
] as [string, Record<string, string>][]) {
|
||||
const changes = (prefix: string, sign: string) => Object.entries(columns).map(([column, value]) =>
|
||||
`${column}=${column}${sign}${value === "1" ? "1" : prefix + "." + value}`).join(",");
|
||||
this.db.exec(`CREATE TRIGGER IF NOT EXISTS summary_${table}_insert AFTER INSERT ON ${table} BEGIN UPDATE summary SET ${changes("NEW", "+")} WHERE id=1; END;
|
||||
CREATE TRIGGER IF NOT EXISTS summary_${table}_delete AFTER DELETE ON ${table} BEGIN UPDATE summary SET ${changes("OLD", "-")} WHERE id=1; END;`);
|
||||
const values = Object.entries(columns).filter(([, value]) => value !== "1");
|
||||
if (values.length) this.db.exec(`CREATE TRIGGER IF NOT EXISTS summary_${table}_update AFTER UPDATE OF ${values.map(([, value]) => value).join(",")} ON ${table}
|
||||
BEGIN UPDATE summary SET ${values.map(([column, value]) => `${column}=${column}-OLD.${value}+NEW.${value}`).join(",")} WHERE id=1; END;`);
|
||||
}
|
||||
});
|
||||
const reservationRoot = options.reservationRoot || options.root;
|
||||
mkdirSync(reservationRoot, { recursive: true, mode: 0o700 });
|
||||
this.disk = new DatabaseSync(join(reservationRoot, "disk-reservations.sqlite"));
|
||||
this.disk.exec(`PRAGMA busy_timeout=5000; PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL;
|
||||
CREATE TABLE IF NOT EXISTS reservations(token TEXT PRIMARY KEY,bytes INTEGER);
|
||||
PRAGMA busy_timeout=50;`);
|
||||
this.db.exec("PRAGMA synchronous=NORMAL; PRAGMA busy_timeout=50;");
|
||||
} catch (error) { this.disk?.close(); this.db.close(); throw error; }
|
||||
}
|
||||
// Reader leases, history and decisions are disposable after a system failure.
|
||||
// Reservations and publication records use FULL before creating/renaming payloads,
|
||||
// so recovery never releases disk accounting before removing an unindexed file.
|
||||
private transaction<T>(fn: () => T, durable = false): T {
|
||||
if (durable) this.db.exec("PRAGMA synchronous=FULL");
|
||||
try {
|
||||
this.db.exec("BEGIN IMMEDIATE");
|
||||
try { const value = fn(); this.db.exec("COMMIT"); return value; }
|
||||
catch (error) { this.db.exec("ROLLBACK"); throw error; }
|
||||
} finally { if (durable) this.db.exec("PRAGMA synchronous=NORMAL"); }
|
||||
}
|
||||
private trim(table: "aliases" | "history" | "decisions", limit: number, order: string) {
|
||||
const excess = Number(this.db.prepare(`SELECT ${table} n FROM summary WHERE id=1`).get()!.n) - limit;
|
||||
if (excess > 0) this.db.prepare(`DELETE FROM ${table} WHERE rowid IN (SELECT rowid FROM ${table} ORDER BY ${order} LIMIT ?)`).run(excess);
|
||||
}
|
||||
private reserveDisk(token: string, bytes: number) {
|
||||
const disk = this.disk!;
|
||||
disk.exec("BEGIN IMMEDIATE");
|
||||
try {
|
||||
// Take a fresh filesystem sample while holding the shared admission lock.
|
||||
const space = statfsSync(this.options.root);
|
||||
const pending = Number(disk.prepare("SELECT COALESCE(SUM(bytes),0) bytes FROM reservations").get()!.bytes);
|
||||
if (space.bavail * space.bsize < pending + bytes + (this.options.freeBytes || 0))
|
||||
throw new CacheAdmissionError("disk_headroom");
|
||||
disk.prepare("INSERT INTO reservations VALUES(?,?)").run(token, bytes);
|
||||
disk.exec("COMMIT");
|
||||
} catch (error) { disk.exec("ROLLBACK"); throw error; }
|
||||
}
|
||||
private releaseDisk(token: string) { this.disk!.prepare("DELETE FROM reservations WHERE token=?").run(token); }
|
||||
private metric(event: string, bytes = 0, ms = 0) { observeCache(`${this.name}_${event}`, bytes, ms); }
|
||||
private path(record: BlobRecord) {
|
||||
if (!validDigest(record.digest) || !/^[a-f0-9-]+\.(gz|raw)$/.test(record.file)) throw new Error("invalid_blob_record");
|
||||
return join(this.options.root, "blobs", record.digest.substring(record.digest.indexOf("-") + 1, record.digest.indexOf("-") + 3), record.digest, record.file);
|
||||
}
|
||||
private usage() { return Number(this.db.prepare("SELECT allocatedBytes bytes FROM summary WHERE id=1").get()!.bytes); }
|
||||
private reservations() { return Number(this.db.prepare("SELECT reservedBytes bytes FROM summary WHERE id=1").get()!.bytes); }
|
||||
private expire() {
|
||||
const now = Date.now();
|
||||
// A crashed producer's files are removed BEFORE its reservation is released.
|
||||
const dead = this.db.prepare("SELECT token FROM fills WHERE expires<?").all(now);
|
||||
for (const row of dead) {
|
||||
const pending = this.db.prepare("SELECT digest,file FROM publications WHERE token=?").get(row.token);
|
||||
if (pending && !this.db.prepare("SELECT 1 FROM blobs WHERE file=?").get(pending.file)) {
|
||||
try { unlinkSync(this.path(pending as unknown as BlobRecord)); }
|
||||
catch (error) { if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error; }
|
||||
}
|
||||
this.db.prepare("DELETE FROM publications WHERE token=?").run(row.token);
|
||||
for (const suffix of ["raw", "gz"]) {
|
||||
try { unlinkSync(join(this.options.root, "tmp", `${row.token}.${suffix}`)); }
|
||||
catch (error) { if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error; }
|
||||
}
|
||||
this.releaseDisk(String(row.token));
|
||||
}
|
||||
this.db.prepare("DELETE FROM fills WHERE expires<?").run(now);
|
||||
this.db.prepare("DELETE FROM readers WHERE expires<?").run(now);
|
||||
this.db.prepare("DELETE FROM decisions WHERE rowid IN (SELECT rowid FROM decisions WHERE expires<? ORDER BY expires LIMIT 256)").run(now);
|
||||
this.db.prepare("DELETE FROM history WHERE rowid IN (SELECT rowid FROM history WHERE seen<? ORDER BY seen LIMIT 256)").run(now - 7 * 86400000);
|
||||
}
|
||||
private evictTo(target: number) {
|
||||
let used = this.usage();
|
||||
const rows = this.db.prepare(`SELECT * FROM blobs WHERE digest NOT IN (SELECT digest FROM readers)
|
||||
ORDER BY accessed LIMIT 512`).all() as unknown as BlobRecord[];
|
||||
for (const record of rows) {
|
||||
if (used <= target) break;
|
||||
try { unlinkSync(this.path(record)); } catch (error) { if ((error as NodeJS.ErrnoException).code !== "ENOENT") continue; }
|
||||
this.db.prepare("UPDATE history SET evicted=1 WHERE key IN (SELECT key FROM aliases WHERE digest=?)").run(record.digest);
|
||||
this.db.prepare("DELETE FROM blobs WHERE digest=?").run(record.digest);
|
||||
try { rmdirSync(join(this.path(record), "..")); } catch { /* another representation remains */ }
|
||||
used -= record.allocated; this.metric("eviction", record.stored, Date.now() - record.accessed);
|
||||
}
|
||||
}
|
||||
async maintain() {
|
||||
this.transaction(() => { this.expire(); this.evictTo(Math.max(0, this.options.budget - this.reservations())); });
|
||||
const scan = this.transaction(() => {
|
||||
const previous = this.db.prepare("SELECT nextScan FROM maintenance WHERE id=1").get();
|
||||
if (previous && Number(previous.nextScan) > Date.now()) return false;
|
||||
this.db.prepare("INSERT OR REPLACE INTO maintenance VALUES(1,?)").run(Date.now() + 3600000);
|
||||
return true;
|
||||
});
|
||||
if (!scan) return;
|
||||
// Recover publication orphans after crashes. Never remove a recent publisher's file.
|
||||
const walk = async (directory: string): Promise<void> => {
|
||||
let entries; try { entries = await fs.opendir(directory); } catch { return; }
|
||||
for await (const entry of entries) {
|
||||
const p = join(directory, entry.name);
|
||||
if (entry.isDirectory()) { await walk(p); continue; }
|
||||
const stat = await fs.stat(p).catch(() => undefined);
|
||||
if (!stat || Date.now() - stat.mtimeMs < 3600000) continue;
|
||||
const found = this.db.prepare("SELECT 1 FROM blobs WHERE file=?").get(entry.name);
|
||||
if (!found) await fs.unlink(p).catch(() => {});
|
||||
}
|
||||
};
|
||||
await walk(join(this.options.root, "blobs"));
|
||||
}
|
||||
stats() {
|
||||
const data = this.db.prepare("SELECT blobs,originalBytes,storedBytes,allocatedBytes,activeReaders,decisions,reservedBytes FROM summary WHERE id=1").get();
|
||||
return { ...data, budgetBytes: this.options.budget, maxObjectBytes: this.options.maxObject,
|
||||
oldestAccess: this.db.prepare("SELECT accessed FROM blobs ORDER BY accessed LIMIT 1").get()?.accessed ?? null };
|
||||
}
|
||||
async open(digest: string, encoded = false): Promise<BlobStream | undefined> {
|
||||
try { return await this.openReader(digest, encoded); } catch (error) { return cacheError(error); }
|
||||
}
|
||||
private async openReader(digest: string, encoded = false): Promise<BlobStream | undefined> {
|
||||
if (!validDigest(digest)) return;
|
||||
const token = randomUUID();
|
||||
let leaseUntil = Date.now() + 120000;
|
||||
const record = this.transaction(() => {
|
||||
const r = this.db.prepare("SELECT * FROM blobs WHERE digest=?").get(digest) as unknown as BlobRecord | undefined;
|
||||
if (!r) return;
|
||||
this.db.prepare("INSERT INTO readers VALUES(?,?,?)").run(token, digest, leaseUntil);
|
||||
if (Date.now() - r.accessed > 30000) this.db.prepare("UPDATE blobs SET accessed=? WHERE digest=?").run(Date.now(), digest);
|
||||
return r;
|
||||
});
|
||||
if (!record) return;
|
||||
let handle;
|
||||
try {
|
||||
handle = await fs.open(this.path(record), "r");
|
||||
if ((await handle.stat()).size !== record.stored) throw new Error("corrupt_blob_size");
|
||||
} catch {
|
||||
await handle?.close();
|
||||
await fs.unlink(this.path(record)).catch(() => {});
|
||||
this.db.prepare("DELETE FROM readers WHERE token=?").run(token);
|
||||
this.transaction(() => { this.db.prepare("DELETE FROM blobs WHERE digest=?").run(digest); });
|
||||
this.metric("corrupt"); return;
|
||||
}
|
||||
const source = handle.createReadStream();
|
||||
const needsDecode = record.encoding === "gzip" && !encoded;
|
||||
const output = (needsDecode ? createGunzip() : source) as BlobStream;
|
||||
output.cachedBlob = { store: this, record };
|
||||
let start = performance.now();
|
||||
let renewAt = Date.now() + 30000;
|
||||
const heartbeat = setInterval(() => {
|
||||
const now = Date.now();
|
||||
if (now < renewAt) return;
|
||||
if (now >= leaseUntil) { output.destroy(new Error("cache_reader_lease_lost")); return; }
|
||||
try {
|
||||
const renewedUntil = now + 120000;
|
||||
if (!this.db.prepare("UPDATE readers SET expires=? WHERE token=? AND expires>?").run(renewedUntil, token, now).changes) {
|
||||
output.destroy(new Error("cache_reader_lease_lost")); return;
|
||||
}
|
||||
leaseUntil = renewedUntil; renewAt = now + 30000;
|
||||
} catch (error) {
|
||||
if (!isCacheContention(error)) output.destroy(new Error("cache_reader_lease_lost"));
|
||||
// Retry on the next tick while the previously confirmed lease is valid.
|
||||
}
|
||||
}, 1000); heartbeat.unref();
|
||||
let released = false;
|
||||
const release = () => {
|
||||
if (released) return; released = true; clearInterval(heartbeat); source.destroy();
|
||||
try { this.db.prepare("DELETE FROM readers WHERE token=?").run(token); } catch { /* expires */ }
|
||||
};
|
||||
output.once("close", release);
|
||||
if (needsDecode) {
|
||||
// Delay decompression until a consumer asks for data; fast-path callers can close unused streams.
|
||||
const originalRead = output._read.bind(output); let begun = false;
|
||||
output._read = size => { if (!begun) { begun = true; start = performance.now(); source.on("error", e => output.destroy(e)); source.pipe(output as Transform); } originalRead(size); };
|
||||
output.once("end", () => observeCache("decompress", record.size, performance.now() - start));
|
||||
}
|
||||
return output;
|
||||
}
|
||||
async lookup(key: string): Promise<BlobStream | undefined> {
|
||||
const alias = this.db.prepare("SELECT digest FROM aliases WHERE key=?").get(key);
|
||||
return this.open(String(alias?.digest || key));
|
||||
}
|
||||
private accessed(key: string, hit: boolean, bytes = 0) {
|
||||
const previous = this.db.prepare("SELECT * FROM history WHERE key=? AND seen>=?").get(key, Date.now() - 7 * 86400000);
|
||||
this.metric(hit ? "hit" : "miss", bytes);
|
||||
if (!hit) this.metric(previous?.evicted ? "refetch" : previous ? "known_miss" : "first_seen_miss");
|
||||
if (previous) {
|
||||
const age = Date.now() - Number(previous.seen);
|
||||
for (const [label, limit] of [["1h", 3600000], ["1d", 86400000], ["7d", 604800000]] as const) if (age <= limit) this.metric(`reuse_${label}`, bytes);
|
||||
}
|
||||
if (!previous || previous.evicted || Date.now() - Number(previous.seen) >= 30000) this.transaction(() => {
|
||||
this.db.prepare("INSERT INTO history VALUES(?,?,0) ON CONFLICT(key) DO UPDATE SET seen=excluded.seen,evicted=0").run(key, Date.now());
|
||||
this.trim("history", 100000, "seen");
|
||||
});
|
||||
}
|
||||
async getOrFill(key: string, produce: () => Promise<Readable>, verify?: (proof: BlobProof) => Promise<string>, sizeHint?: number): Promise<BlobStream> {
|
||||
try { return await this.fill(key, produce, verify, sizeHint); } catch (error) { return cacheError(error); }
|
||||
}
|
||||
private async fill(key: string, produce: () => Promise<Readable>, verify?: (proof: BlobProof) => Promise<string>, sizeHint?: number): Promise<BlobStream> {
|
||||
if (!/^[a-z0-9-]{1,100}$/.test(key)) throw new Error("invalid_cache_key");
|
||||
let hit = await this.lookup(key);
|
||||
try { this.accessed(key, !!hit, hit?.cachedBlob?.record.size || sizeHint || 0); }
|
||||
catch (error) { hit?.destroy(); throw error; }
|
||||
if (hit) return hit;
|
||||
if (sizeHint != null && sizeHint > this.options.maxObject) { this.metric("bypass", sizeHint); throw new CacheAdmissionError("object_too_large"); }
|
||||
const token = randomUUID(), reservation = this.options.maxObject * 3 + 1024;
|
||||
const deadline = Date.now() + 120000;
|
||||
let leaseUntil = 0;
|
||||
for (;;) {
|
||||
const owned = this.transaction(() => {
|
||||
this.expire();
|
||||
if (this.db.prepare("SELECT 1 FROM fills WHERE key=?").get(key)) return false;
|
||||
const reserved = this.reservations();
|
||||
if (this.usage() + reserved + reservation > this.options.budget) {
|
||||
this.evictTo(Math.max(0, Math.min(this.options.budget * (this.options.lowWater ?? 0.9), this.options.budget - reserved - reservation)));
|
||||
}
|
||||
// Eviction unlinks files: commit its metadata even if admission still fails.
|
||||
if (this.usage() + reserved + reservation > this.options.budget) return null;
|
||||
leaseUntil = Date.now() + 120000;
|
||||
this.db.prepare("INSERT INTO fills VALUES(?,?,?,?)").run(key, token, leaseUntil, reservation); return true;
|
||||
}, true);
|
||||
if (owned === null) throw new CacheAdmissionError("cache_budget_busy");
|
||||
if (owned) break;
|
||||
this.metric("fill_wait");
|
||||
if (Date.now() >= deadline) throw new CacheAdmissionError("fill_timeout");
|
||||
await new Promise(resolve => setTimeout(resolve, 50));
|
||||
hit = await this.lookup(key); if (hit) return hit;
|
||||
}
|
||||
const raw = join(this.options.root, "tmp", `${token}.raw`), gz = join(this.options.root, "tmp", `${token}.gz`);
|
||||
let input: Readable | undefined, file: string | undefined;
|
||||
let renewAt = Date.now() + 30000, leaseLost = false;
|
||||
const loseLease = () => { leaseLost = true; input?.destroy(new Error("cache_fill_lease_lost")); };
|
||||
const heartbeat = setInterval(() => {
|
||||
const now = Date.now();
|
||||
if (leaseLost || now < renewAt) return;
|
||||
if (now >= leaseUntil) { loseLease(); return; }
|
||||
try {
|
||||
const renewedUntil = now + 120000;
|
||||
if (!this.db.prepare("UPDATE fills SET expires=? WHERE key=? AND token=? AND expires>?").run(renewedUntil, key, token, now).changes) {
|
||||
loseLease(); return;
|
||||
}
|
||||
leaseUntil = renewedUntil; renewAt = now + 30000;
|
||||
} catch (error) {
|
||||
if (!isCacheContention(error)) loseLease();
|
||||
// Retry contention on the next tick within the last confirmed lease.
|
||||
}
|
||||
}, 1000); heartbeat.unref();
|
||||
const timeout = setTimeout(() => input?.destroy(new Error("cache_fill_timeout")), 120000); timeout.unref();
|
||||
try {
|
||||
// The pool reservation is already durable before acquiring physical headroom.
|
||||
this.transaction(() => {
|
||||
if (!this.db.prepare("SELECT 1 FROM fills WHERE token=? AND expires>?").get(token, Date.now())) throw new CacheAdmissionError("cache_fill_lease_lost");
|
||||
this.reserveDisk(token, reservation);
|
||||
});
|
||||
// Recheck after ownership; another producer may have published between the initial lookup and lease acquisition.
|
||||
hit = await this.lookup(key); if (hit) return hit;
|
||||
input = await produce();
|
||||
if (leaseLost || Date.now() >= leaseUntil) throw new Error("cache_fill_lease_lost");
|
||||
let size = 0; const sha256 = createHash("sha256");
|
||||
await pipeline(input, new Transform({ transform: (chunk, _encoding, done) => {
|
||||
size += chunk.length; sha256.update(chunk);
|
||||
done(size > this.options.maxObject ? new CacheAdmissionError("object_too_large") : null, chunk);
|
||||
} }), createWriteStream(raw, { flags: "wx", mode: 0o600 }));
|
||||
const git = createHash("sha1").update(`blob ${size}\0`);
|
||||
for await (const chunk of createReadStream(raw)) git.update(chunk);
|
||||
const proof = { gitSha: git.digest("hex"), sha256: sha256.digest("hex"), size };
|
||||
const digest = verify ? await verify(proof) : `sha256-${proof.sha256}`;
|
||||
if (!validDigest(digest)) throw new Error("invalid_verified_digest");
|
||||
let encoding: "gzip" | "identity" = "identity", stored = size, chosen = raw;
|
||||
if (this.options.gzip && size >= 1024) {
|
||||
const started = performance.now();
|
||||
await pipeline(createReadStream(raw), createGzip({ level: 6 }), createWriteStream(gz, { flags: "wx", mode: 0o600 }));
|
||||
const compressed = (await fs.stat(gz)).size;
|
||||
observeCache("compress", size, performance.now() - started);
|
||||
if (compressed <= size * (1 - (this.options.saving ?? 0.1))) { encoding = "gzip"; stored = compressed; chosen = gz; }
|
||||
}
|
||||
const record: BlobRecord = { digest, file: `${token}.${encoding === "gzip" ? "gz" : "raw"}`, size, stored, allocated: 0, encoding, accessed: Date.now() };
|
||||
file = this.path(record); await fs.mkdir(join(file, ".."), { recursive: true, mode: 0o700 });
|
||||
const durable = await fs.open(chosen, "r");
|
||||
try { await durable.sync(); } finally { await durable.close(); }
|
||||
this.transaction(() => { this.db.prepare("INSERT INTO publications VALUES(?,?,?)").run(token, digest, record.file); }, true);
|
||||
await fs.rename(chosen, file);
|
||||
record.allocated = Math.max(4096, (await fs.stat(file)).blocks * 512);
|
||||
let duplicate = false;
|
||||
this.transaction(() => {
|
||||
if (!this.db.prepare("SELECT 1 FROM fills WHERE key=? AND token=? AND expires>?").get(key, token, Date.now())) throw new Error("cache_fill_lease_lost");
|
||||
duplicate = !!this.db.prepare("SELECT 1 FROM blobs WHERE digest=?").get(digest);
|
||||
if (!duplicate) this.db.prepare("INSERT INTO blobs VALUES(?,?,?,?,?,?,?)").run(digest, record.file, size, stored, encoding, Date.now(), record.allocated);
|
||||
this.db.prepare("INSERT INTO aliases VALUES(?,?) ON CONFLICT(key) DO UPDATE SET digest=excluded.digest").run(key, digest);
|
||||
this.trim("aliases", 200000, "rowid");
|
||||
// Bridge publication to open(), including asynchronous duplicate cleanup.
|
||||
this.db.prepare("INSERT INTO readers VALUES(?,?,?)").run(token, digest, Date.now() + 120000);
|
||||
}, true);
|
||||
if (duplicate) { await fs.unlink(file); this.metric("deduplicated", stored); }
|
||||
file = undefined;
|
||||
this.metric("fill", size); this.metric("stored", duplicate ? 0 : stored);
|
||||
const result = await this.open(digest); if (!result) throw new Error("published_blob_missing"); return result;
|
||||
} finally {
|
||||
clearTimeout(timeout); clearInterval(heartbeat); input?.destroy();
|
||||
await Promise.all([raw, gz, ...(file ? [file] : [])].map(p => fs.unlink(p).catch(error => {
|
||||
if (error.code !== "ENOENT") throw error;
|
||||
})));
|
||||
this.releaseFill(key, token);
|
||||
}
|
||||
}
|
||||
private releaseFill(key: string, token: string) {
|
||||
try {
|
||||
this.transaction(() => {
|
||||
this.releaseDisk(token);
|
||||
this.db.prepare("DELETE FROM readers WHERE token=?").run(token);
|
||||
this.db.prepare("DELETE FROM publications WHERE token=?").run(token);
|
||||
this.db.prepare("DELETE FROM fills WHERE key=? AND token=?").run(key, token);
|
||||
});
|
||||
} catch (error) { if (!isCacheContention(error)) throw error; /* Leases expire conservatively. */ }
|
||||
}
|
||||
decision(key: string) {
|
||||
try {
|
||||
const row = this.db.prepare("SELECT * FROM decisions WHERE key=? AND expires>?").get(key, Date.now());
|
||||
if (row) this.db.prepare("UPDATE decisions SET accessed=? WHERE key=? AND accessed<?").run(Date.now(), key, Date.now() - 30000);
|
||||
return row ? { digest: String(row.digest), changed: !!row.changed, isText: row.isText == null ? undefined : !!row.isText } : undefined;
|
||||
} catch (error) { return cacheError(error); }
|
||||
}
|
||||
saveDecision(key: string, digest: string, changed: boolean, isText?: boolean) {
|
||||
try {
|
||||
this.transaction(() => {
|
||||
this.db.prepare("INSERT INTO decisions(key,digest,changed,accessed,expires,isText) VALUES(?,?,?,?,?,?) ON CONFLICT(key) DO UPDATE SET digest=excluded.digest,changed=excluded.changed,accessed=excluded.accessed,expires=excluded.expires,isText=excluded.isText")
|
||||
.run(key, digest, +changed, Date.now(), Date.now() + 7 * 86400000, isText == null ? null : +isText);
|
||||
this.trim("decisions", 100000, "accessed");
|
||||
});
|
||||
} catch (error) { if (!isCacheContention(error)) throw error; /* Optional decision can be recomputed. */ }
|
||||
}
|
||||
close() { this.disk?.close(); this.db.close(); }
|
||||
}
|
||||
let source: BlobCache | undefined, results: BlobCache | undefined;
|
||||
export function sharedBlobCache() {
|
||||
if (!config.SHARED_CACHE_ENABLED || config.STORAGE !== "filesystem") return;
|
||||
try { return source ||= new BlobCache({ reservationRoot: config.SHARED_CACHE_PATH, root: join(config.SHARED_CACHE_PATH, "source"), budget: config.SHARED_CACHE_BYTES,
|
||||
maxObject: config.SHARED_CACHE_MAX_FILE_BYTES, gzip: config.SHARED_CACHE_GZIP, saving: config.SHARED_CACHE_GZIP_MIN_SAVING,
|
||||
freeBytes: config.SHARED_CACHE_FREE_BYTES, lowWater: config.SHARED_CACHE_LOW_WATER });
|
||||
} catch (error) { if (!isCacheContention(error)) throw error; }
|
||||
}
|
||||
export function resultBlobCache() {
|
||||
if (!sharedBlobCache() || !config.SHARED_CACHE_DECISIONS) return;
|
||||
try { return results ||= new BlobCache({ reservationRoot: config.SHARED_CACHE_PATH, root: join(config.SHARED_CACHE_PATH, "results"), budget: config.SHARED_CACHE_RESULT_BYTES,
|
||||
maxObject: Math.min(config.SHARED_CACHE_MAX_FILE_BYTES, Math.floor(config.SHARED_CACHE_RESULT_BYTES / 8)), gzip: config.SHARED_CACHE_GZIP,
|
||||
freeBytes: config.SHARED_CACHE_FREE_BYTES, name: "result" });
|
||||
} catch (error) { if (!isCacheContention(error)) throw error; }
|
||||
}
|
||||
export function closeBlobCaches() { source?.close(); results?.close(); source = results = undefined; }
|
||||
@@ -0,0 +1,76 @@
|
||||
import { hostname } from "os";
|
||||
import { DatabaseSync } from "node:sqlite";
|
||||
import { mkdirSync } from "fs";
|
||||
import { join } from "path";
|
||||
import config from "../config";
|
||||
import { cacheTraffic } from "./request-monitoring";
|
||||
|
||||
export const CACHE_BUCKETS = [0.1, 1, 10, 100, 1000, 10000];
|
||||
type Sample = { count: number; bytes: number; ms: number; buckets: number[] };
|
||||
let db: DatabaseSync | undefined;
|
||||
const pending = new Map<string, Sample>();
|
||||
let dropped = 0;
|
||||
let timer: NodeJS.Timeout | undefined;
|
||||
let lastFlush = 0;
|
||||
function connection() {
|
||||
if (!db) {
|
||||
mkdirSync(config.SHARED_CACHE_PATH, { recursive: true, mode: 0o700 });
|
||||
db = new DatabaseSync(join(config.SHARED_CACHE_PATH, "telemetry.sqlite"));
|
||||
db.exec(`PRAGMA journal_mode=WAL; PRAGMA synchronous=NORMAL; PRAGMA busy_timeout=50;
|
||||
CREATE TABLE IF NOT EXISTS metrics (period INTEGER, bucket INTEGER, name TEXT, traffic TEXT,
|
||||
count INTEGER, bytes INTEGER, ms REAL, b0 INTEGER, b1 INTEGER, b2 INTEGER, b3 INTEGER, b4 INTEGER, b5 INTEGER, b6 INTEGER,
|
||||
PRIMARY KEY(period,bucket,name,traffic));
|
||||
CREATE TABLE IF NOT EXISTS health (id TEXT PRIMARY KEY, updated INTEGER, dropped INTEGER);
|
||||
CREATE INDEX IF NOT EXISTS metric_time ON metrics(period,bucket);`);
|
||||
}
|
||||
return db;
|
||||
}
|
||||
/** Fixed event names only. No paths, terms, user identifiers or blob hashes in metrics. */
|
||||
export function observeCache(name: string, bytes = 0, ms = 0, traffic = cacheTraffic()) {
|
||||
if (!config.CACHE_MONITORING_ENABLED) return;
|
||||
const key = JSON.stringify([Math.floor(Date.now() / 60000) * 60000, name, traffic]);
|
||||
if (!pending.has(key) && pending.size >= 512) { dropped++; return; }
|
||||
const value = pending.get(key) || { count: 0, bytes: 0, ms: 0, buckets: Array(7).fill(0) };
|
||||
value.count++; value.bytes += Math.max(0, bytes); value.ms += Math.max(0, ms);
|
||||
const bucket = CACHE_BUCKETS.findIndex(bound => ms <= bound);
|
||||
value.buckets[bucket < 0 ? 6 : bucket]++;
|
||||
pending.set(key, value);
|
||||
if (!timer) { timer = setInterval(flushCacheMetrics, 5000); timer.unref(); }
|
||||
}
|
||||
export function flushCacheMetrics() {
|
||||
if (!config.CACHE_MONITORING_ENABLED) return;
|
||||
try {
|
||||
const sql = connection();
|
||||
const insert = sql.prepare(`INSERT INTO metrics VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)
|
||||
ON CONFLICT(period,bucket,name,traffic) DO UPDATE SET count=count+excluded.count,bytes=bytes+excluded.bytes,ms=ms+excluded.ms,
|
||||
b0=b0+excluded.b0,b1=b1+excluded.b1,b2=b2+excluded.b2,b3=b3+excluded.b3,b4=b4+excluded.b4,b5=b5+excluded.b5,b6=b6+excluded.b6`);
|
||||
sql.exec("BEGIN IMMEDIATE");
|
||||
try {
|
||||
for (const [key, value] of pending) {
|
||||
const [at, name, traffic] = JSON.parse(key);
|
||||
for (const period of [60000, 3600000]) insert.run(period, Math.floor(at / period) * period, name, traffic,
|
||||
value.count, value.bytes, value.ms, ...value.buckets);
|
||||
}
|
||||
sql.prepare("DELETE FROM metrics WHERE (period=60000 AND bucket<?) OR (period=3600000 AND bucket<?)")
|
||||
.run(Date.now() - 48 * 3600000, Date.now() - 30 * 86400000);
|
||||
sql.prepare("INSERT OR REPLACE INTO health VALUES(?,?,?)").run(`${hostname()}:${process.pid}`, Date.now(), dropped);
|
||||
sql.prepare("DELETE FROM health WHERE updated<?").run(Date.now() - 30 * 86400000);
|
||||
sql.exec("COMMIT"); pending.clear(); lastFlush = Date.now();
|
||||
} catch (error) { sql.exec("ROLLBACK"); throw error; }
|
||||
} catch { dropped += [...pending.values()].reduce((sum, v) => sum + v.count, 0); pending.clear(); }
|
||||
}
|
||||
export function cacheMetricsReport(minutes = 60) {
|
||||
if (!config.CACHE_MONITORING_ENABLED) return { available: false, reason: "Monitoring is disabled", rows: [] };
|
||||
flushCacheMetrics();
|
||||
const windowMinutes = [15, 60, 1440, 10080].includes(minutes) ? minutes : 60;
|
||||
try {
|
||||
const sql = connection();
|
||||
const rows = sql.prepare(`SELECT name,traffic,SUM(count) count,SUM(bytes) bytes,SUM(ms) ms,
|
||||
SUM(b0) b0,SUM(b1) b1,SUM(b2) b2,SUM(b3) b3,SUM(b4) b4,SUM(b5) b5,SUM(b6) b6
|
||||
FROM metrics WHERE period=? AND bucket>=? GROUP BY name,traffic ORDER BY name,traffic`)
|
||||
.all(windowMinutes > 2880 ? 3600000 : 60000, Date.now() - windowMinutes * 60000);
|
||||
const health = sql.prepare("SELECT COALESCE(SUM(dropped),0) dropped FROM health").get();
|
||||
return { available: true, windowMinutes, sampledAt: Date.now(), lastFlush, dropped: health?.dropped, rows, latencyBucketsMs: CACHE_BUCKETS };
|
||||
} catch { return { available: false, reason: "Cache telemetry could not be read", rows: [], dropped }; }
|
||||
}
|
||||
export function closeCacheMetrics() { clearInterval(timer); timer = undefined; flushCacheMetrics(); db?.close(); db = undefined; }
|
||||
@@ -0,0 +1,111 @@
|
||||
import { createGzip } from "zlib";
|
||||
import { createHash } from "crypto";
|
||||
import { Readable, PassThrough } from "stream";
|
||||
import { pipeline } from "stream/promises";
|
||||
import { Response } from "express";
|
||||
import { performance } from "perf_hooks";
|
||||
import config from "../config";
|
||||
import { assertTransformScopeActive } from "./transformed-cache";
|
||||
import { BlobStream, CacheAdmissionError, resultBlobCache } from "./blob-cache";
|
||||
import { AnonymizeTransformer, classifyByName } from "./anonymize-utils";
|
||||
import { observeCache } from "./cache-observability";
|
||||
|
||||
class Unchanged extends Error {}
|
||||
export function decisionKey(digest: string, options: unknown) {
|
||||
return createHash("sha256").update(JSON.stringify(["anonymizer-v2", digest, options,
|
||||
config.ANONYMIZATION_MASK, config.APP_HOSTNAME, config.additionalExtensions])).digest("hex");
|
||||
}
|
||||
/** Returns false when the caller should use the existing streaming pipeline. Authorization precedes this call. */
|
||||
export async function sendSharedContent(content: Readable, res: Response,
|
||||
options: ConstructorParameters<typeof AnonymizeTransformer>[0]): Promise<boolean> {
|
||||
try { return await sendCached(content, res, options); }
|
||||
catch (error) { if (error instanceof CacheAdmissionError) return false; content.destroy(); throw error; }
|
||||
}
|
||||
async function sendCached(content: Readable, res: Response,
|
||||
options: ConstructorParameters<typeof AnonymizeTransformer>[0]): Promise<boolean> {
|
||||
const source = (content as BlobStream).cachedBlob;
|
||||
const results = resultBlobCache();
|
||||
if (!source || !results) return false;
|
||||
const assertActive = async () => {
|
||||
if (res.destroyed) throw new Error("response_cancelled");
|
||||
await assertTransformScopeActive(options);
|
||||
if (res.destroyed) throw new Error("response_cancelled");
|
||||
};
|
||||
const encoding = res.req.acceptsEncodings("gzip", "identity");
|
||||
res.vary("Accept-Encoding");
|
||||
if (!encoding) { content.destroy(); res.status(406).end(); return true; }
|
||||
const key = decisionKey(source.record.digest, options);
|
||||
const namedType = classifyByName(options.filePath);
|
||||
const send = async (store: typeof results, digest: string, isText: boolean, onHit?: () => void) => {
|
||||
const stream = await store.open(digest, encoding === "gzip");
|
||||
if (!stream) return false;
|
||||
try { await assertActive(); } catch (error) { stream.destroy(); throw error; }
|
||||
const record = stream.cachedBlob!.record;
|
||||
onHit?.();
|
||||
content.destroy();
|
||||
if (record.encoding === "gzip" && encoding === "gzip") {
|
||||
res.setHeader("Content-Encoding", "gzip"); res.setHeader("Content-Length", record.stored);
|
||||
} else { res.removeHeader("Content-Encoding"); res.setHeader("Content-Length", record.size); }
|
||||
if (isText && !res.getHeader("Content-Type")) res.type("text/plain");
|
||||
if (isText) res.setHeader("Accept-Ranges", "none");
|
||||
else res.removeHeader("Accept-Ranges");
|
||||
if (record.encoding === "identity" && !res.req.acceptsEncodings("identity")) {
|
||||
res.setHeader("Content-Encoding", "gzip"); res.removeHeader("Content-Length");
|
||||
await pipeline(stream, createGzip(), res);
|
||||
} else await pipeline(stream, res);
|
||||
return true;
|
||||
};
|
||||
const decision = results.decision(key);
|
||||
if (decision && (decision.isText !== undefined || namedType !== null) && await send(decision.changed ? results : source.store, decision.digest, decision.isText ?? namedType!, () => {
|
||||
observeCache("decision_hit", source.record.size);
|
||||
observeCache(decision.changed ? "rewrite_changed_cached" : "rewrite_unchanged_cached", source.record.size);
|
||||
})) return true;
|
||||
observeCache("decision_miss", source.record.size);
|
||||
if (classifyByName(options.filePath) === false) {
|
||||
results.saveDecision(key, source.record.digest, false, false);
|
||||
observeCache("rewrite_binary", source.record.size);
|
||||
return send(source.store, source.record.digest, false);
|
||||
}
|
||||
// Preserve the original source reader as a fallback if result admission fails.
|
||||
let processing: Readable | undefined;
|
||||
let transformer: AnonymizeTransformer | undefined;
|
||||
const started = performance.now();
|
||||
const abort = () => {
|
||||
if (!res.writableFinished) {
|
||||
const error = new Error("response_cancelled");
|
||||
transformer?.destroy(error);
|
||||
processing?.destroy(error);
|
||||
}
|
||||
};
|
||||
res.once("close", abort);
|
||||
try {
|
||||
const output = await results.getOrFill(`render-${key}`, async () => {
|
||||
await assertActive();
|
||||
processing = await source.store.open(source.record.digest);
|
||||
if (!processing) throw new CacheAdmissionError("source_evicted");
|
||||
if (res.destroyed) throw new Error("response_cancelled");
|
||||
const transform = transformer = new AnonymizeTransformer({ ...options, decisionManaged: true });
|
||||
const stream = new PassThrough();
|
||||
void pipeline(processing, transform, stream).catch(error => stream.destroy(error));
|
||||
return stream;
|
||||
}, async proof => {
|
||||
await assertActive();
|
||||
const unchanged = source.record.digest === `git-${proof.gitSha}` || source.record.digest === `sha256-${proof.sha256}`;
|
||||
observeCache(unchanged ? "rewrite_unchanged" : "rewrite_changed", source.record.size, performance.now() - started);
|
||||
if (unchanged) { results.saveDecision(key, source.record.digest, false, !!transformer?.isText); throw new Unchanged(); }
|
||||
return `sha256-${proof.sha256}`;
|
||||
}, source.record.size);
|
||||
const digest = output.cachedBlob!.record.digest;
|
||||
// Keep the result pinned until its HTTP reader has been opened.
|
||||
try {
|
||||
await assertActive();
|
||||
results.saveDecision(key, digest, true, true);
|
||||
if (await send(results, digest, true)) return true;
|
||||
} finally { output.destroy(); }
|
||||
return false;
|
||||
} catch (error) {
|
||||
if (error instanceof Unchanged) return send(source.store, source.record.digest, !!transformer?.isText);
|
||||
if (error instanceof CacheAdmissionError) { observeCache("decision_bypass", source.record.size); return false; }
|
||||
content.destroy(); throw error;
|
||||
} finally { res.removeListener("close", abort); transformer?.destroy(); processing?.destroy(); }
|
||||
}
|
||||
@@ -12,7 +12,7 @@ export interface Metric {
|
||||
bytes: number; aborted: number; errors: number; slow: number; buckets: number[];
|
||||
}
|
||||
const logger = createLogger("requests");
|
||||
const context = new AsyncLocalStorage<{ id: string; service: Service; stages: Partial<Record<Stage, number>> }>();
|
||||
const context = new AsyncLocalStorage<{ id: string; service: Service; traffic?: string; stages: Partial<Record<Stage, number>> }>();
|
||||
const methods = new Set(["GET", "HEAD", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"]);
|
||||
const staticRoutes = new Set([
|
||||
"/", "/dashboard", "/anonymize", "/healthcheck", "/api/healthcheck", "/api/stat", "/api/stat/history",
|
||||
@@ -77,8 +77,9 @@ let activeRequests = 0;
|
||||
export function activeRequestCount() { return activeRequests; }
|
||||
export function requestHeaders(): Record<string, string> {
|
||||
const id = context.getStore()?.id;
|
||||
return id ? { "x-request-id": id } : {};
|
||||
return id ? { "x-request-id": id, "x-cache-traffic": cacheTraffic() } : {};
|
||||
}
|
||||
export function cacheTraffic() { return context.getStore()?.traffic || "background"; }
|
||||
export function recordCacheHit(cache: "source" | "transformed", hit: boolean) {
|
||||
if (context.getStore()) requestMetrics.observe(`${cache}_${hit ? "hit" : "miss"}`, "all", "all", 0);
|
||||
}
|
||||
@@ -107,7 +108,12 @@ export function monitorRequests(service: Service, metrics = requestMetrics,
|
||||
return (req, res, next) => {
|
||||
const forwarded = service === "streamer" ? req.get("x-request-id") : undefined;
|
||||
const id = forwarded && /^[a-f0-9-]{36}$/.test(forwarded) ? forwarded : randomUUID();
|
||||
const trace = { id, service, stages: {} };
|
||||
const forwardedTraffic = service === "streamer" ? req.get("x-cache-traffic") : undefined;
|
||||
const traffic = forwardedTraffic && ["file", "website", "zip", "monitor", "background"].includes(forwardedTraffic)
|
||||
? forwardedTraffic : /uptime|healthcheck|monitor/i.test(req.get("user-agent") || "") ? "monitor"
|
||||
: /\/zip(?:[/?]|$)|\/download(?:[/?]|$)/.test(req.originalUrl) ? "zip"
|
||||
: req.originalUrl.startsWith("/w/") ? "website" : "file";
|
||||
const trace = { id, service, traffic, stages: {} };
|
||||
const route = requestRoute(req.originalUrl, service);
|
||||
const method = methods.has(req.method) ? req.method : "OTHER";
|
||||
const started = performance.now();
|
||||
|
||||
@@ -1,5 +1,7 @@
|
||||
import { withOwnerAccessRecovery } from "../owner-notifications";
|
||||
import { ExpiringMap } from "../expiring-map";
|
||||
import { sharedBlobCache, CacheAdmissionError, BlobProof } from "../blob-cache";
|
||||
import { observeCache } from "../cache-observability";
|
||||
import { contentGenerationPrefix, contentRetirementMarker } from "../content-generation";
|
||||
import { createHash } from "crypto";
|
||||
import { coordinatedFill } from "../cache-coordination";
|
||||
@@ -384,8 +386,92 @@ export default class GitHubStream extends GitHubBase {
|
||||
return undefined;
|
||||
}
|
||||
};
|
||||
const shared = sharedBlobCache();
|
||||
if (shared && /^[a-f0-9]{40}$/.test(expected.sha)) {
|
||||
await assertActive();
|
||||
if (expected.size != null && expected.size > config.MAX_FILE_SIZE) throw new AnonymousError("file_too_big", { httpStatus: 413 });
|
||||
let legacy = false, allowLegacy = true;
|
||||
const download = async (checkMinimum = false) => {
|
||||
const parentToken = await this.data.getToken();
|
||||
const token = expected.upstream ? "" : parentToken;
|
||||
const sourceMiss = cachedContentMiss(this.data, parentToken);
|
||||
if (sourceMiss && sourceMiss !== "file_not_found") throw new AnonymousError(sourceMiss, { httpStatus: 404 });
|
||||
const missKey = createHash("sha256").update(JSON.stringify([
|
||||
key, this.data.organization, this.data.repoName, token,
|
||||
])).digest("hex");
|
||||
const missed = missingContent.get(missKey);
|
||||
if (missed) throw new AnonymousError(missed, { httpStatus: 404 });
|
||||
const normalize = async (error: Error) => {
|
||||
const normalized = downloadError(error, filePath);
|
||||
if (expected.upstream && ["file_not_found", "file_not_accessible"].includes(normalized.message))
|
||||
return new AnonymousError("submodule_not_available", { httpStatus: 404, cause: normalized });
|
||||
if (!expected.upstream && normalized.message === "file_not_found" && normalized.httpStatus === 404) {
|
||||
const code = await classifyContentMiss(this.data, token);
|
||||
missingContent.set(missKey, code, 30_000);
|
||||
if (code !== "file_not_found") return new AnonymousError(code, { httpStatus: 404, cause: normalized });
|
||||
}
|
||||
return normalized;
|
||||
};
|
||||
const input = await origin.downloadWithFallback(token, expected.sha, sourcePath)
|
||||
.catch(async error => { throw await normalize(error); });
|
||||
input.on("error", () => {}); // The iterator below propagates normalized failures.
|
||||
const deadline = setTimeout(() => input.destroy(new AnonymousError("upstream_error", { httpStatus: 502 })), 120000);
|
||||
deadline.unref();
|
||||
const limited = stream.Readable.from((async function* () {
|
||||
let bytes = 0;
|
||||
try {
|
||||
for await (const chunk of input) {
|
||||
bytes += chunk.length;
|
||||
if (bytes > config.MAX_FILE_SIZE) throw new AnonymousError("file_too_big", { httpStatus: 413 });
|
||||
yield chunk;
|
||||
}
|
||||
if (checkMinimum && expected.size != null && expected.size > 0 && bytes < expected.size)
|
||||
throw new AnonymousError("storage_write_size_mismatch", { httpStatus: 502 });
|
||||
} catch (error) { throw await normalize(error as Error); }
|
||||
finally { clearTimeout(deadline); input.destroy(); }
|
||||
})());
|
||||
stopSourceOnDestroy(limited, () => { clearTimeout(deadline); input.destroy(); });
|
||||
return limited;
|
||||
};
|
||||
const produce = async () => {
|
||||
if (allowLegacy && await cached()) { legacy = true; observeCache("source_legacy_promotion"); return storage.read(repoId, cachePath); }
|
||||
legacy = false; return download();
|
||||
};
|
||||
const verify = async (proof: BlobProof) => {
|
||||
if (proof.gitSha === expected.sha) return `git-${proof.gitSha}`;
|
||||
// Resolved LFS bytes do not hash to the Git pointer SHA. Verify both the pointer and its payload.
|
||||
const parentToken = await this.data.getToken();
|
||||
const pointer = origin.downloadFile(expected.upstream ? "" : parentToken, expected.sha);
|
||||
const timer = setTimeout(() => pointer.destroy(new Error("lfs_pointer_timeout")), 30000); timer.unref();
|
||||
let bytes = 0; const chunks: Buffer[] = [];
|
||||
try {
|
||||
for await (const chunk of pointer) { bytes += chunk.length; if (bytes > 1024) throw new Error("invalid_lfs_pointer"); chunks.push(Buffer.from(chunk)); }
|
||||
const raw = Buffer.concat(chunks), text = raw.toString("utf8");
|
||||
const sha = createHash("sha1").update(`blob ${raw.length}\0`).update(raw).digest("hex");
|
||||
if (sha !== expected.sha || !text.startsWith("version https://git-lfs.github.com/spec/v1\n") ||
|
||||
text.match(/^oid sha256:([a-f0-9]{64})$/m)?.[1] !== proof.sha256 ||
|
||||
Number(text.match(/^size (\d+)$/m)?.[1]) !== proof.size) throw new Error("invalid_blob_hash");
|
||||
return `sha256-${proof.sha256}`;
|
||||
} finally { clearTimeout(timer); pointer.destroy(); }
|
||||
};
|
||||
try {
|
||||
let content;
|
||||
try { content = await shared.getOrFill(`git-${expected.sha}`, produce, verify, expected.size); }
|
||||
catch (error) {
|
||||
if (!legacy || error instanceof CacheAdmissionError) throw error;
|
||||
allowLegacy = false;
|
||||
content = await shared.getOrFill(`git-${expected.sha}`, produce, verify, expected.size);
|
||||
}
|
||||
try { await assertActive(); } catch (error) { content.destroy(); throw error; }
|
||||
return content;
|
||||
} catch (error) {
|
||||
if (!(error instanceof CacheAdmissionError)) throw error;
|
||||
observeCache("source_bypass"); await assertActive(); return download(true);
|
||||
}
|
||||
}
|
||||
const hit = await measureStage("cache_lookup", cached);
|
||||
recordCacheHit("source", !!hit);
|
||||
observeCache(hit ? "legacy_source_hit" : "legacy_source_miss", expected.size || 0);
|
||||
if (hit) return storage.read(repoId, cachePath);
|
||||
if (expected.size != null && expected.size > config.MAX_FILE_SIZE) {
|
||||
throw new AnonymousError("file_too_big", { httpStatus: 413, object: filePath });
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { sharedBlobCache, resultBlobCache } from "./blob-cache";
|
||||
import { promises as fs } from "fs";
|
||||
import { join } from "path";
|
||||
import { tmpdir } from "os";
|
||||
@@ -34,7 +35,7 @@ export async function startTemporaryStorageMaintenance(intervalMs = 60_000) {
|
||||
if (running) return;
|
||||
running = true;
|
||||
try {
|
||||
const results = await Promise.allSettled([maintainTextSpools(true), pruneTransformedCache()]);
|
||||
const results = await Promise.allSettled([maintainTextSpools(true), pruneTransformedCache(), sharedBlobCache()?.maintain(), resultBlobCache()?.maintain()]);
|
||||
for (const result of results) if (result.status === "rejected") {
|
||||
logger.warn("temporary storage cleanup deferred", { message: (result.reason as Error).message });
|
||||
}
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { observeCache } from "./cache-observability";
|
||||
import { promises as fs } from "fs";
|
||||
import { join } from "path";
|
||||
import { tmpdir } from "os";
|
||||
@@ -25,6 +26,14 @@ async function scopeActive(scope?: CacheScope) {
|
||||
&& scope.revision === current.contentCacheRevision;
|
||||
}
|
||||
|
||||
/** Shared and legacy result caches must enforce the same repository lifecycle. */
|
||||
export async function assertTransformScopeActive(options: unknown) {
|
||||
const settings = options as { repoId?: string; cacheGeneration?: string; cacheRevision?: string } | undefined;
|
||||
const scope = typeof settings?.repoId === "string"
|
||||
? { repoId: settings.repoId, generation: settings.cacheGeneration, revision: settings.cacheRevision } : undefined;
|
||||
if (!await scopeActive(scope)) throw new AnonymousError("repository_changed", { httpStatus: 409 });
|
||||
}
|
||||
|
||||
async function removeEntry(metadata: string) {
|
||||
await Promise.all([metadata, metadata.replace(/\.json$/, ".data")].map(path => fs.rm(path, { force: true })));
|
||||
}
|
||||
@@ -35,9 +44,7 @@ export async function transformedFile(digest: string, options: unknown, output:
|
||||
const settings = options as { repoId?: string; cacheGeneration?: string; cacheRevision?: string } | undefined;
|
||||
const scope: CacheScope | undefined = typeof settings?.repoId === "string"
|
||||
? { repoId: settings.repoId, generation: settings.cacheGeneration, revision: settings.cacheRevision } : undefined;
|
||||
const assertActive = async () => {
|
||||
if (!await scopeActive(scope)) throw new AnonymousError("repository_changed", { httpStatus: 409 });
|
||||
};
|
||||
const assertActive = () => assertTransformScopeActive(options);
|
||||
const produceActive = async () => {
|
||||
await assertActive();
|
||||
const changed = await produce();
|
||||
@@ -63,9 +70,9 @@ export async function transformedFile(digest: string, options: unknown, output:
|
||||
};
|
||||
const hit = await read();
|
||||
if (hit !== undefined) {
|
||||
try { const copied = await copyCached(hit); recordCacheHit("transformed", true); return copied; } catch { /* eviction: retry the transformation */ }
|
||||
try { const copied = await copyCached(hit); recordCacheHit("transformed", true); observeCache("legacy_result_hit"); return copied; } catch { /* eviction: retry the transformation */ }
|
||||
}
|
||||
recordCacheHit("transformed", false);
|
||||
recordCacheHit("transformed", false); observeCache("legacy_result_miss");
|
||||
const pending = fills.get(key);
|
||||
if (pending) {
|
||||
await pending.catch(() => {});
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
import { cacheMetricsReport } from "../../core/cache-observability";
|
||||
import { sharedBlobCache, resultBlobCache } from "../../core/blob-cache";
|
||||
import { AsyncCache } from "../../core/async-cache";
|
||||
import * as os from "os";
|
||||
import { execFile } from "child_process";
|
||||
@@ -88,6 +90,15 @@ router.get("/performance", async (req, res) => {
|
||||
catch { res.json({ available: false, instances: [], routes: [], stages: [] }); }
|
||||
});
|
||||
|
||||
router.get("/cache", async (req, res) => {
|
||||
res.set("Cache-Control", "no-store");
|
||||
try {
|
||||
res.json({ ...cacheMetricsReport(Number(req.query.minutes)), enabled: config.SHARED_CACHE_ENABLED,
|
||||
filesystemSupported: config.STORAGE === "filesystem", source: sharedBlobCache()?.stats() || null,
|
||||
results: resultBlobCache()?.stats() || null });
|
||||
} catch { res.json({ available: false, reason: "The cache index is unavailable", rows: [] }); }
|
||||
});
|
||||
|
||||
function dashboardCache(
|
||||
_req: express.Request,
|
||||
res: express.Response,
|
||||
|
||||
@@ -146,7 +146,8 @@ router.get(
|
||||
anonymizedPath,
|
||||
repo.model.options
|
||||
);
|
||||
res.header("ETag", etag);
|
||||
if (config.SHARED_CACHE_ENABLED) res.vary("Accept-Encoding");
|
||||
res.header("ETag", config.SHARED_CACHE_ENABLED ? `W/${etag}` : etag);
|
||||
// Force the browser to revalidate every time. The previous 210-day
|
||||
// max-age was keyed only on the upstream sha, so editing the
|
||||
// anonymization term list left old anonymizations cached under the
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { sendSharedContent } from "../core/cached-response";
|
||||
import * as express from "express";
|
||||
import GitHubStream from "../core/source/GitHubStream";
|
||||
import {
|
||||
@@ -99,6 +100,7 @@ router.post("/", async (req: express.Request, res: express.Response) => {
|
||||
} else if (isTextFile(filePath)) {
|
||||
res.contentType("text/plain");
|
||||
}
|
||||
if (await sendSharedContent(content, res, anonymizer.opt)) { anonymizer.destroy(); return; }
|
||||
// Only declare Accept-Ranges: none for text files — they get rewritten on
|
||||
// the fly so byte ranges aren't meaningful. For binary entries the
|
||||
// transformer is a passthrough; let <video>/<audio> fall back to a full
|
||||
|
||||
Reference in new issue
Block a user