Merge commit 'refs/security-review/pr844' into HEAD

This commit is contained in:
Thomas Durieux committed 2026-10-08 16:25:00 +00:00
commit 243396515e
163 files changed
+16198 -22725

No files matched your search

+57 -33
View File
@@ -1,3 +1,4 @@
import { streamResponse } from "./response-stream";
import { join, basename, dirname } from "path";
import { Response } from "express";
import { Readable } from "stream";
@@ -9,6 +10,8 @@ import { RepositoryStatus } from "./types";
import config from "../config";
import {
anonymizePath,
anonymizePathCompiled,
compileTerms,
hasCustomTermReplacement,
isTextFile,
} from "./anonymize-utils";
@@ -19,6 +22,7 @@ import { IFile } from "./model/files/files.types";
import { FilterQuery } from "mongoose";
import { createLogger, serializeError } from "./logger";
import { githubTokenForStreamer } from "./github-token-context";
import { requestHeaders, startStage } from "./request-monitoring";
const logger = createLogger("anonymized-file");
@@ -168,10 +172,12 @@ export default class AnonymizedFile {
const exactQuery: FilterQuery<IFile> = {
repoId: this.repository.repoId,
path: fileDir,
treeGeneration: this.repository.model?.treeGeneration || { $exists: false },
};
if (filename != "") exactQuery.name = filename;
const exact = await FileModel.findOne(exactQuery);
if (exact) {
if (exact.metadataPending) await this.repository.completeRecoveredFileMetadata(exact);
this._file = exact;
return exact;
}
@@ -203,6 +209,18 @@ export default class AnonymizedFile {
// narrow Mongo query. Fetch the repository's paths and verify them by
// re-applying anonymization. Default XXXX-N masks retain the optimized,
// anchored query.
const compiled = compileTerms(terms);
if (typeof this.repository.findAnonymizedPath === "function") {
const indexed = await this.repository.findAnonymizedPath(this.anonymizedPath);
if (indexed) { this._file = indexed; return indexed; }
if (indexed === null) {
if (anonymizePathCompiled(this.anonymizedPath, compiled) === this.anonymizedPath) {
const recovered = await this.recoverTruncatedFile(fileDir);
if (recovered) { this._file = recovered; return recovered; }
}
throw new AnonymousError("file_not_found", { httpStatus: 404, object: this });
}
}
const candidates = usesCustomReplacement
? await FileModel.find({ repoId: this.repository.repoId }).exec()
: await FileModel.find({
@@ -214,8 +232,9 @@ export default class AnonymizedFile {
for (const candidate of candidates) {
const candidatePath = join(candidate.path, candidate.name);
if (
anonymizePath(candidatePath, terms) == this.anonymizedPath
anonymizePathCompiled(candidatePath, compiled) == this.anonymizedPath
) {
if (candidate.metadataPending) await this.repository.completeRecoveredFileMetadata(candidate);
this._file = candidate;
return candidate;
}
@@ -253,19 +272,24 @@ export default class AnonymizedFile {
const recovered = await source.fetchFileInfoFromPath(this.anonymizedPath);
if (!recovered) return null;
recovered.repoId = this.repository.repoId;
recovered.treeGeneration = this.repository.model.treeGeneration;
logger.info("recovered file from truncated tree", {
repoId: this.repository.repoId,
path: this.anonymizedPath,
});
let cached;
try {
// Cache it so the next request is served from the database.
await FileModel.create(recovered);
cached = await FileModel.create({ ...recovered, metadataPending: true });
} catch (error) {
logger.warn(
"failed to cache recovered file",
serializeError(error as Error)
);
}
// Once persisted, invalidation is required. Keep the pending marker on
// failure so both a subsequent lookup and maintenance retry the work.
if (cached) await this.repository.completeRecoveredFileMetadata(cached);
return recovered;
}
@@ -356,10 +380,13 @@ export default class AnonymizedFile {
// use the streamer service
return got.stream(join(config.STREAMER_ENTRYPOINT, "api"), {
method: "POST",
headers: requestHeaders(),
json: {
token: await githubTokenForStreamer(await this.repository.getToken(), this.repository.model.source.repositoryName),
repoFullName: this.repository.model.source.repositoryName,
commit: this.repository.model.source.commit,
cacheGeneration: `${this.repository.model.treeGeneration || "legacy"}:${this.repository.model.anonymizeDate?.toISOString() || ""}`,
cacheRevision: this.repository.model.contentCacheRevision,
branch: this.repository.model.source.branch,
repoId: this.repository.repoId,
filePath: this.filePath,
@@ -398,22 +425,44 @@ export default class AnonymizedFile {
this.size(),
this.repository.getToken(),
]);
const headersDone = startStage("streamer_headers");
const resStream = got
.stream(join(config.STREAMER_ENTRYPOINT, "api"), {
method: "POST",
headers: requestHeaders(),
json: {
sha,
size,
token: await githubTokenForStreamer(token, this.repository.model.source.repositoryName),
repoFullName: this.repository.model.source.repositoryName,
commit: this.repository.model.source.commit,
cacheGeneration: `${this.repository.model.treeGeneration || "legacy"}:${this.repository.model.anonymizeDate?.toISOString() || ""}`,
cacheRevision: this.repository.model.contentCacheRevision,
branch: this.repository.model.source.branch,
repoId: this.repository.repoId,
filePath: this.filePath,
anonymizerOptions: anonymizer.opt,
},
})
.on("error", (err: Error) => {
});
// Forward Content-Type from the streamer's upstream response.
// got.stream(...).pipe(res) forwards body bytes only — without
// this, the parent response has no Content-Type and the browser
// guesses (text renders as download, images as octet-stream).
resStream.on("response", (upstream: { headers: Record<string, string | string[] | undefined> }) => {
headersDone();
if (res.headersSent) return;
const ct = upstream.headers["content-type"];
if (typeof ct === "string") {
res.contentType(ct);
} else {
const fallback = lookup(this.anonymizedPath);
if (fallback) res.contentType(fallback);
else if (isTextFile(this.anonymizedPath)) res.contentType("text/plain");
}
});
resStream.once("error", () => headersDone(true));
resStream.once("close", () => headersDone());
await streamResponse(resStream, res, (err: Error) => {
const { error } = streamerErrorToAnonymous(
err as Error & {
response?: { statusCode?: number; body?: unknown };
@@ -425,36 +474,9 @@ export default class AnonymizedFile {
);
error.value = this;
handleError(error, res);
});
// Forward Content-Type from the streamer's upstream response.
// got.stream(...).pipe(res) forwards body bytes only — without
// this, the parent response has no Content-Type and the browser
// guesses (text renders as download, images as octet-stream).
resStream.on("response", (upstream: { headers: Record<string, string | string[] | undefined> }) => {
if (res.headersSent) return;
const ct = upstream.headers["content-type"];
if (typeof ct === "string") {
res.contentType(ct);
} else {
const fallback = lookup(this.anonymizedPath);
if (fallback) res.contentType(fallback);
else if (isTextFile(this.anonymizedPath)) res.contentType("text/plain");
}
});
resStream.pipe(res);
// Resolve as soon as the response is fully written rather than
// waiting for the socket to close — keep-alive sockets stay open
// long after the body is delivered, and we don't want to delay
// post-send work like countView() that long.
res.on("finish", () => {
resolve();
});
res.on("close", () => {
resolve();
});
res.on("error", (err) => {
reject(err);
if (res.headersSent && !res.writableEnded) res.destroy();
});
resolve();
return;
}
@@ -480,6 +502,7 @@ export default class AnonymizedFile {
}
});
const content = await this.content();
if (res.destroyed || res.writableEnded) { content.destroy(); anonymizer.destroy(); resolve(); return; }
function handleStreamError(error: Error) {
if (!content.closed && !content.destroyed) {
content.destroy();
@@ -505,6 +528,7 @@ export default class AnonymizedFile {
if (!content.closed && !content.destroyed) {
content.destroy();
}
anonymizer.destroy();
resolve();
});
} catch (error) {
+7 -6
View File
@@ -33,9 +33,7 @@ export default class Conference {
* Expire the conference
*/
async expire() {
await Promise.all(
(await this.repositories()).map(async (conf) => await conf.expire())
);
for (const repo of await this.repositories()) await repo.markExpired(true);
await this.updateStatus("expired");
}
@@ -44,9 +42,12 @@ export default class Conference {
*/
async remove() {
await this.updateStatus("removed");
await Promise.all(
(await this.repositories()).map(async (conf) => await conf.remove())
);
const failures: unknown[] = [];
for (const repo of await this.repositories()) {
try { await repo.remove(); }
catch (error) { failures.push(error); }
}
if (failures.length) throw new AggregateError(failures, "Conference repository removal failed");
}
/**
+55 -10
View File
@@ -1,4 +1,7 @@
import { APP_PROVIDER, appError } from "./github-app";
import { registerGitHubToken } from "./github-token-context";
import { isConnected } from "../server/database";
import { expireEmbeddedContent } from "./content-expiration";
import { APP_PROVIDER, appUserToken } from "./github-app";
import CredentialModel from "./model/credentials/credentials.model";
import { getCredentialToken } from "./credentials";
import { RepositoryStatus } from "./types";
@@ -48,16 +51,39 @@ export default class Gist {
this.owner.model.isNew = false;
}
async getToken() {
if (config.GITHUB_APP_ENABLED && !(await getCredentialToken(this.owner.id)) && await CredentialModel.exists({ ownerId: this.owner.id, provider: APP_PROVIDER })) {
throw appError("github_oauth_required");
private async getAccess() {
if (config.GITHUB_APP_ENABLED && await CredentialModel.exists({ ownerId: this.owner.id, provider: APP_PROVIDER })) {
const ownerId = this.owner.id;
const token = await appUserToken(ownerId);
registerGitHubToken(token, { quotaKey: `app-user:${ownerId}`, renew: () => appUserToken(ownerId) });
return { token, connection: "github-app" };
}
return (await getCredentialToken(this.owner.id, "github", { collection: "anonymizedgists", id: this._model._id })) || config.GITHUB_TOKEN;
return { token: (await getCredentialToken(this.owner.id, "github", { collection: "anonymizedgists", id: this._model._id })) || config.GITHUB_TOKEN,
connection: "oauth" };
}
async getToken() {
return (await this.getAccess()).token;
}
async download() {
const access = await this.getAccess();
try {
await this.downloadContent(access.token);
} catch (error) {
// A stored credential can exist but have been revoked on GitHub.
// Ask the user to reconnect the provider that rejected the credential.
if ((error as { status?: number })?.status === 401) {
throw new AnonymousError(access.connection === "github-app" ? "github_app_reconnect_required" : "github_oauth_required",
{ httpStatus: 403, cause: error as Error });
}
throw error;
}
}
private async downloadContent(token: string) {
logger.info("downloading gist", { gistId: this._model.source.gistId });
const oct = octokit(await this.getToken());
const oct = octokit(token);
const gist_id = this._model.source.gistId;
@@ -131,7 +157,7 @@ export default class Gist {
this._model.options.expirationDate
) {
if (this._model.options.expirationDate <= new Date()) {
await this.expire();
await this.markExpired();
}
}
if (
@@ -217,10 +243,29 @@ export default class Gist {
).exec();
}
async markExpired() {
const now = new Date();
if (isConnected) {
const result = await AnonymizedGistModel.updateOne({ _id: this.model._id,
status: RepositoryStatus.READY, "options.expirationMode": { $ne: "never" },
"options.expirationDate": { $lte: now },
}, { $set: { status: RepositoryStatus.EXPIRING, statusDate: now } }).exec();
if (!result.matchedCount) {
const current = await AnonymizedGistModel.findById(this.model._id).exec();
if (!current) throw new AnonymousError("gist_expired", { object: this, httpStatus: 410 });
this._model = current;
this._gistPayload = undefined;
return;
}
}
this.model.status = RepositoryStatus.EXPIRING;
this.model.statusDate = now;
}
async expire() {
await this.updateStatus(RepositoryStatus.EXPIRING);
await this.resetSate();
await this.updateStatus(RepositoryStatus.EXPIRED);
await expireEmbeddedContent(AnonymizedGistModel, this.model, {
"gist.comments": [], "gist.description": "", "gist.files": [], "gist.ownerLogin": "",
});
}
async remove() {
+8 -2
View File
@@ -11,6 +11,7 @@ import Repository from "./Repository";
import { getCredential, replaceCredential, getCredentialToken } from "./credentials";
import config from "../config";
import { createLogger } from "./logger";
import { measureStage } from "./request-monitoring";
const logger = createLogger("github");
@@ -154,12 +155,12 @@ function ensureRedisGateClient(): Promise<RedisClientType | null> {
socket: {
host: config.REDIS_HOSTNAME,
port: config.REDIS_PORT,
reconnectStrategy: () => false as any,
reconnectStrategy: false,
},
}) as RedisClientType;
c.on("error", () => {
redisGateDisabled = true;
c.disconnect().catch(() => {});
if (c.isOpen) c.destroy();
redisGateReady = null;
});
await c.connect();
@@ -281,6 +282,7 @@ export function octokit(token: string) {
}
throw err;
});
oct.hook.wrap("request", (request, options) => measureStage("upstream", async () => request(options)));
return oct;
}
@@ -308,6 +310,10 @@ export async function checkToken(token: string) {
const checkedRepositoryTokens = new WeakMap<Repository, string>();
export async function getToken(repository: Repository) {
return measureStage("authorization", () => resolveRepositoryToken(repository));
}
async function resolveRepositoryToken(repository: Repository) {
repository.assertNotArchived();
logger.debug("getToken", { repoId: repository.repoId });
if (isConnected && !repository.model.isNew) {
+26 -4
View File
@@ -1,3 +1,5 @@
import { isConnected } from "../server/database";
import { expireEmbeddedContent } from "./content-expiration";
import { boundAppToken } from "./github-app";
import { getCredentialToken } from "./credentials";
import { RepositoryStatus } from "./types";
@@ -121,7 +123,7 @@ export default class PullRequest {
this._model.options.expirationDate
) {
if (this._model.options.expirationDate <= new Date()) {
await this.expire();
await this.markExpired();
}
}
if (
@@ -228,10 +230,30 @@ export default class PullRequest {
/**
* Expire the pullRequest
*/
async markExpired() {
const now = new Date();
if (isConnected) {
const result = await AnonymizedPullRequestModel.updateOne({ _id: this.model._id,
status: RepositoryStatus.READY, "options.expirationMode": { $ne: "never" },
"options.expirationDate": { $lte: now },
}, { $set: { status: RepositoryStatus.EXPIRING, statusDate: now } }).exec();
if (!result.matchedCount) {
const current = await AnonymizedPullRequestModel.findById(this.model._id).exec();
if (!current) throw new AnonymousError("pull_request_expired", { object: this, httpStatus: 410 });
this._model = current;
return;
}
}
this.model.status = RepositoryStatus.EXPIRING;
this.model.statusDate = now;
}
async expire() {
await this.updateStatus(RepositoryStatus.EXPIRING);
await this.resetSate();
await this.updateStatus(RepositoryStatus.EXPIRED);
await expireEmbeddedContent(AnonymizedPullRequestModel, this.model, {
"pullRequest.comments": [], "pullRequest.body": "", "pullRequest.title": "", "pullRequest.diff": "",
"pullRequest.baseRepositoryFullName": "", "pullRequest.headRepositoryFullName": "",
"pullRequest.merged": false, "pullRequest.state": "closed", "pullRequest.draft": false,
}, { "pullRequest.mergedDate": "" });
}
/**
+417 -55
View File
@@ -1,5 +1,8 @@
import AnonymizedPathModel from "./model/anonymized-path";
import storage from "./storage";
import { randomUUID } from "crypto";
import { contentGenerationPrefix, contentRetirementMarker } from "./content-generation";
import { removeTransformedRepositoryCache } from "./transformed-cache";
import { createHash, randomUUID } from "crypto";
import { RepositoryStatus } from "./types";
import { Readable } from "stream";
import * as sha1 from "crypto-js/sha1";
@@ -31,7 +34,36 @@ import AnonymizedRepositoryModel from "./model/anonymizedRepositories/anonymized
import { createLogger, serializeError } from "./logger";
const logger = createLogger("repository");
import { IFile } from "./model/files/files.types";
const pathIndexes = new Map<string, Promise<void>>();
async function buildPathIndex(repoId: string, key: string, terms: string[], generation?: string) {
const cacheKey = `${repoId}:${key}`;
const existing = pathIndexes.get(cacheKey);
if (existing) return existing;
const task = (async () => {
const compiled = compileTerms(terms);
const cursor = FileModel.find({ repoId, treeGeneration: generation || { $exists: false } })
.select("name path sha size").sort({ path: 1, name: 1 }).lean().cursor();
let batch: Parameters<typeof AnonymizedPathModel.bulkWrite>[0] = [];
try {
for await (const file of cursor) {
const name = anonymizePathCompiled(file.name, compiled);
const directory = anonymizePathCompiled(file.path, compiled);
batch.push({ updateOne: { filter: { repoId, key, originalId: file._id }, update: { $set: {
repoId, key, originalId: file._id, name: file.name, path: file.path, sha: file.sha, size: file.size,
anonymousName: name, anonymousDirectory: directory, createdAt: new Date(),
searchName: name.toLowerCase(), searchDirectory: directory.toLowerCase(),
anonymousPath: anonymizePathCompiled(file.path ? `${file.path}/${file.name}` : file.name, compiled),
} }, upsert: true } });
if (batch.length === 250) { await AnonymizedPathModel.bulkWrite(batch); batch = []; }
}
if (batch.length) await AnonymizedPathModel.bulkWrite(batch);
} finally { await cursor.close(); }
})();
// Retain only in-flight work. A subsequent recovery can add file rows.
pathIndexes.set(cacheKey, task);
try { await task; } finally { pathIndexes.delete(cacheKey); }
}
import { IFile, IFileDocument } from "./model/files/files.types";
import AnonymizedFile from "./AnonymizedFile";
import { FilterQuery } from "mongoose";
function anonymizeTreeRecursive(
@@ -87,6 +119,8 @@ export default class Repository {
return new GitHubStream({
repoId: this.repoId,
commit: this.model.source.commit || "HEAD",
cacheGeneration: `${this.model.treeGeneration || "legacy"}:${this.model.anonymizeDate?.toISOString() || ""}`,
cacheRevision: this.model.contentCacheRevision,
organization: ghRepo.owner,
repoName: ghRepo.repo,
getToken: () => this.getToken(),
@@ -133,51 +167,8 @@ export default class Repository {
force: false,
}
): Promise<IFile[]> {
this.assertNotArchived();
await this.ensureFileTree(opt);
const terms = this._model.options.terms || [];
let hasFile = await FileModel.exists({ repoId: this.repoId }).exec();
// Files created by GitHubDownload don't carry a valid 40-char GitHub
// blob SHA. When the source type later switches to GitHubStream the
// stale entries cause blob-API 404s. Detect this by sampling a file
// with a sha and checking its length; force a re-fetch if it doesn't
// look like a GitHub SHA.
if (hasFile && this.source instanceof GitHubStream) {
const sample = await FileModel.findOne(
{ repoId: this.repoId, sha: { $exists: true, $ne: null } },
{ sha: 1 }
).exec();
if (sample?.sha && sample.sha.length !== 40) {
hasFile = null;
}
}
if (!hasFile || opt.force) {
await FileModel.deleteMany({ repoId: this.repoId }).exec();
const source = this.source;
const files = await source.getFiles(opt.progress);
files.forEach((f) => (f.repoId = this.repoId));
await FileModel.insertMany(files);
const sourceWithTruncation = source as unknown as {
truncatedFolderList?: string[];
};
if (Array.isArray(sourceWithTruncation.truncatedFolderList)) {
this._model.truncatedFolders = sourceWithTruncation.truncatedFolderList;
}
this._model.size = { storage: 0, file: 0 };
await this.computeSize();
if (isConnected) {
await AnonymizedRepositoryModel.updateOne(
{ _id: this._model._id },
{
$set: {
truncatedFolders: this._model.truncatedFolders,
size: this._model.size,
},
}
).exec();
}
}
if (
opt.path &&
(opt.path.includes(config.ANONYMIZATION_MASK) ||
@@ -202,6 +193,7 @@ export default class Repository {
const query: FilterQuery<IFile> = {
repoId: this.repoId,
treeGeneration: this.model.treeGeneration || { $exists: false },
};
if (pathQuery !== undefined) {
query.path = pathQuery;
@@ -209,6 +201,268 @@ export default class Repository {
return await FileModel.find(query).exec();
}
private async ensureFileTree(opt: { force?: boolean; progress?: (status: string) => void } = {}) {
this.assertNotArchived();
let hasFile = this.hasEmptyTree() || await FileModel.exists({ repoId: this.repoId, treeGeneration: this.model.treeGeneration || { $exists: false } }).exec();
// Files created by GitHubDownload don't carry a valid 40-char GitHub
// blob SHA. When the source type later switches to GitHubStream the
// stale entries cause blob-API 404s. Detect this by sampling a file
// with a sha and checking its length; force a re-fetch if it doesn't
// look like a GitHub SHA.
if (hasFile && this.source instanceof GitHubStream) {
const sample = await FileModel.findOne(
{ repoId: this.repoId, treeGeneration: this.model.treeGeneration || { $exists: false }, sha: { $exists: true, $ne: null } },
{ sha: 1 }
).exec();
if (sample?.sha && sample.sha.length !== 40) {
hasFile = null;
}
}
if (!hasFile || opt.force) {
const previousGeneration = this.model.treeGeneration;
const previousPrefix = contentGenerationPrefix(`${previousGeneration || "legacy"}:${this.model.anonymizeDate?.toISOString() || ""}`);
const streamed = this.model.source.type !== "Zip";
const source = this.source;
const files = await source.getFiles(opt.progress);
const sourceWithTruncation = source as unknown as { truncatedFolderList?: string[] };
const truncatedFolders = Array.isArray(sourceWithTruncation.truncatedFolderList)
? [...sourceWithTruncation.truncatedFolderList] : [];
const generation = randomUUID();
const buildFilter = { _id: this.model._id,
treeGeneration: previousGeneration || { $exists: false }, ...this.refreshFilter(),
status: this.model.status, statusDate: this.model.statusDate,
"githubAccess.revision": this.model.githubAccess?.revision || { $exists: false },
};
let heartbeat: NodeJS.Timeout | undefined;
if (isConnected) {
// Register before insertion so a crash cannot leave an unknown UUID.
const registered = await AnonymizedRepositoryModel.updateOne(buildFilter, { $push: {
stagedFileTrees: { generation, until: new Date(Date.now() + 15 * 60_000) },
} }).exec();
if (!registered.matchedCount) throw new AnonymousError("repository_changed", { httpStatus: 409 });
heartbeat = setInterval(() => {
const now = new Date();
void AnonymizedRepositoryModel.updateOne({ ...buildFilter,
stagedFileTrees: { $elemMatch: { generation, until: { $gt: now } } },
}, { $set: { "stagedFileTrees.$.until": new Date(now.getTime() + 15 * 60_000) } }).exec().catch(() => {});
}, 30_000);
heartbeat.unref();
}
files.forEach(f => { f.repoId = this.repoId; f.treeGeneration = generation; if (isConnected) f.treeStaged = true; });
try {
await FileModel.insertMany(files);
if (isConnected) {
const activated = await AnonymizedRepositoryModel.updateOne({ _id: this.model._id,
treeGeneration: previousGeneration || { $exists: false }, ...this.refreshFilter(),
status: this.model.status, statusDate: this.model.statusDate,
"githubAccess.revision": this.model.githubAccess?.revision || { $exists: false },
stagedFileTrees: { $elemMatch: { generation, until: { $gt: new Date() } } },
}, { $set: { treeGeneration: generation, truncatedFolders, size: { storage: 0, file: 0 },
...(streamed && !this.model.contentCacheVersion ? { contentCacheVersion: 1, legacyContentCleanupPending: true } : {}),
...(!files.length ? { emptyTreeGeneration: generation } : {}),
}, $unset: { pathIndexKey: "", pathIndexBuiltAt: "", sizeComputedAt: "",
...(files.length ? { emptyTreeGeneration: "" } : {}),
}, $addToSet: { retiredTreeGenerations: previousGeneration || "",
...(streamed ? { retiredContentPrefixes: previousPrefix } : {}),
}, $pull: { stagedFileTrees: { generation } } }).exec();
if (!activated.matchedCount) {
await FileModel.deleteMany({ repoId: this.repoId, treeGeneration: generation }).exec();
throw new AnonymousError("repository_changed", { httpStatus: 409 });
}
}
} finally { clearInterval(heartbeat); }
this.model.treeGeneration = generation;
if (isConnected) {
// A crash here is safe: maintenance recognizes the active generation.
await FileModel.updateMany({ repoId: this.repoId, treeGeneration: generation, treeStaged: true },
{ $unset: { treeStaged: "" } }).exec().catch(error => logger.warn("staged tree completion deferred", serializeError(error)));
}
this.model.emptyTreeGeneration = files.length ? undefined : generation;
this.model.truncatedFolders = truncatedFolders;
if (streamed) this.model.contentCacheVersion = 1;
if (isConnected) {
await this.cleanupRetiredFileTrees().catch(error => logger.warn("retired tree cleanup deferred", serializeError(error)));
} else {
await FileModel.deleteMany({ repoId: this.repoId, treeGeneration: previousGeneration || { $exists: false } }).exec();
}
this._model.size = { storage: 0, file: 0 };
this._model.sizeComputedAt = undefined;
await this.computeSize();
if (isConnected) {
await AnonymizedRepositoryModel.updateOne(
{ _id: this._model._id, treeGeneration: this.model.treeGeneration },
{
$set: {
treeGeneration: this._model.treeGeneration,
size: this._model.size,
},
}
).exec();
}
}
}
/** Activation records retirement atomically; maintenance retries interrupted deletion. */
async cleanupRetiredFileTrees(now = new Date()) {
const stages = await AnonymizedRepositoryModel.findById(this.model._id).select("stagedFileTrees").lean().exec();
for (const stage of stages?.stagedFileTrees || []) {
if (stage.until > now) continue;
await AnonymizedRepositoryModel.updateOne({ _id: this.model._id, treeGeneration: { $ne: stage.generation },
stagedFileTrees: { $elemMatch: { generation: stage.generation, until: { $lte: now } } },
}, { $pull: { stagedFileTrees: { generation: stage.generation } },
$addToSet: { retiredTreeGenerations: stage.generation } }).exec();
}
const current = await AnonymizedRepositoryModel.findById(this.model._id)
.select("treeGeneration retiredTreeGenerations retiredContentPrefixes legacyContentCleanupPending source.type status anonymizeDate").lean().exec();
if (!current) return;
const activePrefix = contentGenerationPrefix(`${current.treeGeneration || "legacy"}:${current.anonymizeDate?.toISOString() || ""}`);
const prefixes = (current.retiredContentPrefixes || []).filter(prefix => prefix !== activePrefix);
for (const prefix of prefixes) {
// Fence producers before removal. A late publisher rechecks this marker
// and removes its own output instead of recreating a retired cache.
await storage.write(this.repoId, contentRetirementMarker(prefix), "retired");
await storage.rm(this.repoId, prefix);
}
if (prefixes.length) await AnonymizedRepositoryModel.updateOne({ _id: current._id },
{ $pull: { retiredContentPrefixes: { $in: prefixes } } }).exec();
if (current.legacyContentCleanupPending && current.status === RepositoryStatus.READY && current.source?.type !== "Zip") {
await storage.removeLegacyContent(this.repoId);
await AnonymizedRepositoryModel.updateOne({ _id: current._id }, { $unset: { legacyContentCleanupPending: "" } }).exec();
}
const retired = (current.retiredTreeGenerations || []).filter(generation => generation !== (current.treeGeneration || ""));
if (retired.length) {
await FileModel.deleteMany({ repoId: this.repoId, $or: [
{ treeGeneration: { $in: retired.filter(Boolean) } },
...(retired.includes("") ? [{ treeGeneration: { $exists: false } }] : []),
] }).exec();
await AnonymizedRepositoryModel.updateOne({ _id: this.model._id },
{ $pull: { retiredTreeGenerations: { $in: retired } } }).exec();
}
await AnonymizedRepositoryModel.updateOne({ _id: this.model._id, "retiredTreeGenerations.0": { $exists: false } },
{ $unset: { retiredTreeGenerations: "" } }).exec();
await AnonymizedRepositoryModel.updateOne({ _id: this.model._id, "retiredContentPrefixes.0": { $exists: false } },
{ $unset: { retiredContentPrefixes: "" } }).exec();
await AnonymizedRepositoryModel.updateOne({ _id: this.model._id, "stagedFileTrees.0": { $exists: false } },
{ $unset: { stagedFileTrees: "" } }).exec();
}
/** Indexed row markers also discover inserts that finish after a build lease is retired. */
static async cleanupStagedFileTree(repoId: string, generation: string, now = new Date()) {
let current = await AnonymizedRepositoryModel.findOne({ repoId }).exec();
if (current && current.treeGeneration !== generation &&
current.stagedFileTrees?.some(stage => stage.generation === generation && stage.until > now)) return;
if (current) {
await new Repository(current).cleanupRetiredFileTrees(now);
current = await AnonymizedRepositoryModel.findById(current._id).exec();
}
const terminal = current?.status && [RepositoryStatus.EXPIRING, RepositoryStatus.EXPIRED,
RepositoryStatus.REMOVING, RepositoryStatus.REMOVED].includes(current.status);
if (current?.treeGeneration === generation && !terminal) {
await FileModel.updateMany({ repoId, treeGeneration: generation, treeStaged: true }, { $unset: { treeStaged: "" } }).exec();
} else if (terminal || !current?.stagedFileTrees?.some(stage => stage.generation === generation && stage.until > now)) {
await FileModel.deleteMany({ repoId, treeGeneration: generation, treeStaged: true }).exec();
}
}
private pathKey() {
return createHash("sha256").update(JSON.stringify([
"paths-v2", config.ANONYMIZATION_MASK, this.options.terms,
this.model.treeGeneration, this.model.fileMetadataRevision, this.model.source.commit, this.model.anonymizeDate,
])).digest("hex");
}
private async indexPaths() {
if (this.status && [RepositoryStatus.EXPIRING, RepositoryStatus.EXPIRED,
RepositoryStatus.REMOVING, RepositoryStatus.REMOVED].includes(this.status)) {
throw new AnonymousError("repository_changed", { httpStatus: 409 });
}
await this.ensureFileTree();
const metadataRevision = this.model.fileMetadataRevision;
const key = this.pathKey();
if (this.model.pathIndexKey !== key || !this.model.pathIndexBuiltAt || this.model.pathIndexBuiltAt.getTime() < Date.now() - 24 * 3600_000) {
try {
await buildPathIndex(this.repoId, key, this.options.terms || [], this.model.treeGeneration);
const builtAt = new Date();
const result = await AnonymizedRepositoryModel.updateOne({ _id: this.model._id,
status: this.model.status, statusDate: this.model.statusDate,
treeGeneration: this.model.treeGeneration || { $exists: false },
fileMetadataRevision: metadataRevision || { $exists: false },
"source.commit": this.model.source.commit || { $exists: false },
"options.terms": this.options.terms || [], anonymizeDate: this.model.anonymizeDate,
}, { $set: { pathIndexKey: key, pathIndexBuiltAt: builtAt } }).exec();
if (!result.matchedCount) throw new AnonymousError("repository_changed", { httpStatus: 409 });
this.model.pathIndexKey = key;
this.model.pathIndexBuiltAt = builtAt;
} catch (error) {
// A reset invalidates this revision. Never delete a valid shared key
// merely because one of its concurrent builders failed.
const current = await AnonymizedRepositoryModel.exists({ _id: this.model._id,
fileMetadataRevision: metadataRevision || { $exists: false },
}).exec();
if (!current) await AnonymizedPathModel.deleteMany({ repoId: this.repoId, key }).exec();
throw error;
}
// TTL retires superseded mappings. An older builder must never delete
// mappings activated by a newer settings or metadata generation.
}
return key;
}
async findAnonymizedPath(path: string): Promise<IFile | null | undefined> {
if (!isConnected) return undefined;
const key = await this.indexPaths();
// Original paths retain priority; collisions between transformed paths
// resolve consistently in original path/name order.
return AnonymizedPathModel.findOne({ repoId: this.repoId, key, anonymousPath: path })
.sort({ path: 1, name: 1 }).lean().exec();
}
async invalidateFileMetadata() {
const revision = randomUUID();
this.model.pathIndexKey = undefined;
this.model.pathIndexBuiltAt = undefined;
this.model.sizeComputedAt = undefined;
this.model.size = { storage: 0, file: 0 };
if (isConnected) {
const result = await AnonymizedRepositoryModel.updateOne({ _id: this.model._id,
treeGeneration: this.model.treeGeneration || { $exists: false },
status: this.model.status, statusDate: this.model.statusDate,
anonymizeDate: this.model.anonymizeDate,
}, { $set: { size: this.model.size, fileMetadataRevision: revision }, $unset: { pathIndexKey: "", pathIndexBuiltAt: "", sizeComputedAt: "" } }).exec();
if (!result.matchedCount) throw new AnonymousError("repository_changed", { httpStatus: 409 });
}
this.model.fileMetadataRevision = revision;
}
async completeRecoveredFileMetadata(file: IFileDocument) {
await this.invalidateFileMetadata();
await FileModel.updateOne({ _id: file._id, metadataPending: true },
{ $unset: { metadataPending: "" } }).exec();
file.metadataPending = undefined;
}
async searchFiles(query: string): Promise<Partial<IFile>[]> {
const q = query.toLowerCase();
if (q.includes("/")) return [];
if (isConnected) {
const key = await this.indexPaths();
const pattern = new RegExp(q.replace(/[.*+?^${}()|[\]\\]/g, "\\$&"));
const rows = await AnonymizedPathModel.find({ repoId: this.repoId, key,
$or: [{ searchName: pattern }, { searchDirectory: pattern }] })
.select("anonymousName anonymousDirectory size").limit(500).lean().exec();
return rows.map(f => ({ name: f.anonymousName, path: f.anonymousDirectory, size: f.size }));
}
const matches: Partial<IFile>[] = [];
for (const f of await this.anonymizedFiles({ includeSha: false, recursive: true })) {
if (f.name?.toLowerCase().includes(q) || (f.path || "").split("/").some(segment => segment.toLowerCase().includes(q))) {
matches.push({ name: f.name, path: f.path, size: f.size });
if (matches.length === 500) break;
}
}
return matches;
}
/**
* Check the status of the repository
*/
@@ -226,7 +480,7 @@ export default class Repository {
this._model.options.expirationDate
) {
if (this._model.options.expirationDate <= new Date()) {
await this.expire();
await this.markExpired();
}
}
if (
@@ -280,7 +534,9 @@ export default class Repository {
generateAnonymizeTransformer(filePath: string) {
return new AnonymizeTransformer({
filePath: filePath,
cacheGeneration: `${this.model.treeGeneration || "legacy"}:${this.model.anonymizeDate?.toISOString() || ""}`,
terms: this.options.terms,
cacheRevision: this.model.contentCacheRevision,
image: this.options.image,
link: this.options.link,
repoId: this.repoId,
@@ -291,7 +547,7 @@ export default class Repository {
async isReady() {
if (this.status !== RepositoryStatus.READY) return false;
if (!(await FileModel.exists({ repoId: this.repoId }).exec())) {
if (!this.hasEmptyTree() && !(await FileModel.exists({ repoId: this.repoId, treeGeneration: this.model.treeGeneration || { $exists: false } }).exec())) {
this.model.status = RepositoryStatus.PREPARING;
await this.updateIfNeeded({ force: true });
return false;
@@ -299,6 +555,10 @@ export default class Repository {
return true;
}
private hasEmptyTree() {
return !!this.model.treeGeneration && this.model.emptyTreeGeneration === this.model.treeGeneration;
}
private refreshToken?: string;
private refreshFilter() {
@@ -578,10 +838,85 @@ export default class Repository {
/**
* Expire the repository
*/
async markExpired(force = false) {
const inactive = [RepositoryStatus.ARCHIVED, RepositoryStatus.REMOVED,
RepositoryStatus.REMOVING, RepositoryStatus.EXPIRED, RepositoryStatus.EXPIRING];
if (force && this.status && inactive.includes(this.status)) return;
const now = new Date();
if (isConnected) {
const result = await AnonymizedRepositoryModel.updateOne({ _id: this.model._id,
status: force ? { $nin: inactive } : RepositoryStatus.READY, ...(force ? {} : { "options.expirationMode": { $ne: "never" },
"options.expirationDate": { $lte: now } }),
}, { $set: { status: RepositoryStatus.EXPIRING, statusDate: now } }).exec();
if (!result.matchedCount) {
if (force) return;
const current = await AnonymizedRepositoryModel.findById(this.model._id).exec();
if (!current) throw new AnonymousError("repository_expired", { object: this, httpStatus: 410 });
this._model = current;
this.assertNotArchived();
return;
}
}
this.model.status = RepositoryStatus.EXPIRING;
this.model.statusDate = now;
}
async expire() {
await this.updateStatus(RepositoryStatus.EXPIRING);
await this.resetSate();
await this.updateStatus(RepositoryStatus.EXPIRED);
if (!isConnected) {
await this.updateStatus(RepositoryStatus.EXPIRING);
await this.resetSate();
await this.updateStatus(RepositoryStatus.EXPIRED);
return;
}
const now = new Date(), token = randomUUID(), revision = randomUUID();
const pending = this.status === RepositoryStatus.EXPIRING || this.status === RepositoryStatus.EXPIRED;
if (!pending && (this.options.expirationMode === "never" || !this.options.expirationDate
|| this.options.expirationDate > now)) return;
if (this.status && [RepositoryStatus.ARCHIVED, RepositoryStatus.REMOVING, RepositoryStatus.REMOVED].includes(this.status)) return;
const claimed = await AnonymizedRepositoryModel.updateOne({ _id: this.model._id,
status: this.model.status, statusDate: this.model.statusDate,
anonymizeDate: this.model.anonymizeDate,
treeGeneration: this.model.treeGeneration || { $exists: false },
fileMetadataRevision: this.model.fileMetadataRevision || { $exists: false },
settingsSavedAt: this.model.settingsSavedAt || { $exists: false },
"options.expirationMode": this.options.expirationMode || { $exists: false },
"options.expirationDate": this.options.expirationDate || { $exists: false },
"githubAccess.revision": this.model.githubAccess?.revision || { $exists: false },
$or: [{ cleanupUntil: { $exists: false } }, { cleanupUntil: { $lte: now } }],
}, { $set: { status: RepositoryStatus.EXPIRING, statusDate: now,
fileMetadataRevision: revision, contentCacheRevision: revision, cleanupToken: token, cleanupUntil: new Date(now.getTime() + 15 * 60_000) },
$unset: { pathIndexKey: "", pathIndexBuiltAt: "" },
}).exec();
if (!claimed.matchedCount) return;
this.model.status = RepositoryStatus.EXPIRING;
this.model.statusDate = now;
this.model.fileMetadataRevision = revision;
this.model.contentCacheRevision = revision;
this.model.pathIndexKey = undefined;
this.model.pathIndexBuiltAt = undefined;
const lease = { _id: this.model._id, status: RepositoryStatus.EXPIRING, cleanupToken: token,
treeGeneration: this.model.treeGeneration || { $exists: false }, fileMetadataRevision: revision };
const heartbeat = setInterval(() => {
void AnonymizedRepositoryModel.updateOne(lease, { $set: { cleanupUntil: new Date(Date.now() + 15 * 60_000) } }).exec().catch(() => {});
}, 30_000);
heartbeat.unref();
try {
// The claimed lifecycle blocks restoration until the entire storage
// root, including empty directories and older cache generations, is gone.
if (!await AnonymizedRepositoryModel.exists(lease).exec()) return;
await removeTransformedRepositoryCache(this.repoId);
await storage.rm(this.repoId);
if (!await AnonymizedRepositoryModel.exists(lease).exec()) return;
await FileModel.deleteMany({ repoId: this.repoId }).exec();
await AnonymizedPathModel.deleteMany({ repoId: this.repoId }).exec();
const result = await AnonymizedRepositoryModel.updateOne(lease, { $set: { status: RepositoryStatus.EXPIRED,
statusDate: new Date(), isReseted: true, size: { storage: 0, file: 0 } },
$unset: { cleanupToken: "", cleanupUntil: "", pathIndexKey: "", pathIndexBuiltAt: "", sizeComputedAt: "", emptyTreeGeneration: "", retiredTreeGenerations: "", retiredContentPrefixes: "", legacyContentCleanupPending: "", stagedFileTrees: "" } }).exec();
if (result.matchedCount) { this.model.status = RepositoryStatus.EXPIRED; this.model.isReseted = true; this.model.size = { storage: 0, file: 0 }; this.model.sizeComputedAt = undefined; this.model.emptyTreeGeneration = undefined; }
} finally {
clearInterval(heartbeat);
await AnonymizedRepositoryModel.updateOne({ _id: this.model._id, cleanupToken: token }, { $unset: { cleanupToken: "", cleanupUntil: "" } }).exec();
}
}
/**
@@ -609,6 +944,19 @@ export default class Repository {
*/
async resetSate(status?: RepositoryStatus, statusMessage?: string) {
this.assertNotArchived();
const revision = randomUUID();
this.model.emptyTreeGeneration = undefined;
if (isConnected) {
const result = await AnonymizedRepositoryModel.updateOne({ _id: this.model._id,
treeGeneration: this.model.treeGeneration || { $exists: false },
}, { $set: { fileMetadataRevision: revision },
$unset: { emptyTreeGeneration: "", pathIndexKey: "", pathIndexBuiltAt: "" },
}).exec();
if (!result.matchedCount) throw new AnonymousError("repository_changed", { httpStatus: 409 });
}
this.model.fileMetadataRevision = revision;
this.model.pathIndexKey = undefined;
this.model.pathIndexBuiltAt = undefined;
// remove attribute
this._model.size = { storage: 0, file: 0 };
if (status) {
@@ -617,6 +965,7 @@ export default class Repository {
// remove cache
await Promise.all([
FileModel.deleteMany({ repoId: this.repoId }).exec(),
AnonymizedPathModel.deleteMany({ repoId: this.repoId }).exec(),
this.removeCache(),
]);
logger.info("reset", { repoId: this._model.repoId });
@@ -629,14 +978,24 @@ export default class Repository {
async removeCache() {
// Archive cleanup is handled by the resumable recovery script. Preserve DB metadata.
if (this.status === RepositoryStatus.ARCHIVED) return;
const cacheRevision = randomUUID();
if (isConnected) {
const result = await AnonymizedRepositoryModel.updateOne({ _id: this.model._id,
treeGeneration: this.model.treeGeneration || { $exists: false },
}, { $set: { contentCacheRevision: cacheRevision } }).exec();
if (!result.matchedCount) throw new AnonymousError("repository_changed", { httpStatus: 409 });
}
this.model.contentCacheRevision = cacheRevision;
await removeTransformedRepositoryCache(this.repoId);
await storage.rm(this.repoId);
this.model.isReseted = true;
this.model.sizeComputedAt = undefined;
this.model.size = { storage: 0, file: 0 };
if (isConnected) {
try {
await AnonymizedRepositoryModel.updateOne(
{ _id: this._model._id },
{ $set: { isReseted: true, size: this._model.size } }
{ $set: { isReseted: true, size: this._model.size }, $unset: { sizeComputedAt: "" } }
).exec();
} catch (error) {
logger.error("removeCache save failed", serializeError(error));
@@ -675,11 +1034,12 @@ export default class Repository {
}> {
if (this.status !== RepositoryStatus.READY)
return { storage: 0, file: 0 };
if (this._model.size.file) return this._model.size;
if (this._model.size.file || this._model.sizeComputedAt) return this._model.size;
const res = await FileModel.aggregate([
{
$match: {
repoId: this.repoId,
treeGeneration: this.model.treeGeneration || { $exists: false },
},
},
{
@@ -694,10 +1054,12 @@ export default class Repository {
storage: res[0]?.storage || 0,
file: res[0]?.file || 0,
};
this._model.sizeComputedAt = new Date();
if (isConnected) {
await AnonymizedRepositoryModel.updateOne(
{ _id: this._model._id },
{ $set: { size: this._model.size } }
{ _id: this._model._id, treeGeneration: this.model.treeGeneration || { $exists: false },
fileMetadataRevision: this.model.fileMetadataRevision || { $exists: false } },
{ $set: { size: this._model.size, sizeComputedAt: this._model.sizeComputedAt } }
).exec();
}
return this._model.size;
+57 -63
View File
@@ -14,6 +14,33 @@ import Gist from "./Gist";
import AnonymizedGistModel from "./model/anonymizedGists/anonymizedGists.model";
import { octokit } from "./GitHubUtils";
import { isConnected } from "../server/database";
async function markListExpired<T extends { _id?: unknown; status?: string; statusDate?: Date;
options: { expirationMode?: string; expirationDate?: Date } }>(model: {
updateMany(filter: Record<string, unknown>, update: Record<string, unknown>): { exec(): Promise<unknown> };
find(filter: Record<string, unknown>): { exec(): Promise<T[]> };
}, rows: T[]): Promise<T[]> {
const now = new Date();
const expired = rows.filter(row => row.status === "ready" && row.options.expirationMode !== "never"
&& row.options.expirationDate && row.options.expirationDate <= now);
if (!expired.length) return rows;
if (isConnected) {
const ids = expired.map(row => row._id);
await model.updateMany({ _id: { $in: ids }, status: "ready",
"options.expirationMode": { $ne: "never" }, "options.expirationDate": { $lte: now },
}, { $set: { status: "expiring", statusDate: now } }).exec();
const current = new Map((await model.find({ _id: { $in: ids } }).exec()).map(row => [String(row._id), row]));
const affected = new Set(ids.map(String));
return rows.flatMap(row => {
if (!affected.has(String(row._id))) return [row];
const fresh = current.get(String(row._id));
return fresh ? [fresh] : [];
});
}
for (const row of expired) { row.status = "expiring"; row.statusDate = now; }
return rows;
}
/**
* Model for a user
*/
@@ -151,6 +178,30 @@ export default class User {
* @returns the list of anonymized repositories
*/
async getRepositories() {
const query = this.repositoryMembership();
const repositories = await AnonymizedRepositoryModel.find(query).exec();
return (await markListExpired<typeof repositories[number]>(AnonymizedRepositoryModel, repositories)).map(d => new Repository(d));
}
/**
* Get the lost of anonymized repositories
* @returns the list of anonymized repositories
*/
async getPullRequests() {
const pullRequests = await AnonymizedPullRequestModel.find({
owner: this.id,
}).exec();
return (await markListExpired<typeof pullRequests[number]>(AnonymizedPullRequestModel, pullRequests)).map(d => new PullRequest(d));
}
/**
* Get the list of anonymized gists
*/
async getGists() {
const gists = await AnonymizedGistModel.find({ owner: this.id }).exec();
return (await markListExpired<typeof gists[number]>(AnonymizedGistModel, gists)).map(d => new Gist(d));
}
repositoryMembership() {
const memberships: Record<string, unknown>[] = [{ owner: this.id }];
const githubId = this.model.externalIDs?.github;
if (githubId) memberships.push({ "coauthors.githubId": githubId });
@@ -160,71 +211,14 @@ export default class User {
$or: [{ githubId: { $exists: false } }, { githubId: null }, { githubId: "" }],
} } });
}
const query = { $or: memberships };
const repositories = (
await AnonymizedRepositoryModel.find(query).exec()
).map((d) => new Repository(d));
const promises = [];
for (const repo of repositories) {
if (
repo.status == "ready" &&
repo.options.expirationMode != "never" &&
repo.options.expirationDate != null &&
repo.options.expirationDate < new Date()
) {
// expire the repository
promises.push(repo.expire());
}
}
await Promise.all(promises);
return repositories;
}
/**
* Get the lost of anonymized repositories
* @returns the list of anonymized repositories
*/
async getPullRequests() {
const pullRequests = (
await AnonymizedPullRequestModel.find({
owner: this.id,
}).exec()
).map((d) => new PullRequest(d));
const promises = [];
for (const repo of pullRequests) {
if (
repo.status == "ready" &&
repo.options.expirationMode != "never" &&
repo.options.expirationDate != null &&
repo.options.expirationDate < new Date()
) {
// expire the repository
promises.push(repo.expire());
}
}
await Promise.all(promises);
return pullRequests;
return { $or: memberships };
}
/**
* Get the list of anonymized gists
*/
async getGists() {
const gists = (
await AnonymizedGistModel.find({ owner: this.id }).exec()
).map((d) => new Gist(d));
const promises = [];
for (const g of gists) {
if (
g.status == "ready" &&
g.options.expirationMode != "never" &&
g.options.expirationDate != null &&
g.options.expirationDate < new Date()
) {
promises.push(g.expire());
}
}
await Promise.all(promises);
return gists;
async quotaRepositories() {
return AnonymizedRepositoryModel.find({ owner: this.id, status: "ready",
$or: [{ "options.expirationMode": "never" }, { "options.expirationDate": { $exists: false } },
{ "options.expirationDate": null }, { "options.expirationDate": { $gt: new Date() } }] })
.select("repoId size sizeComputedAt treeGeneration fileMetadataRevision").lean().exec();
}
get model() {
+106
View File
@@ -0,0 +1,106 @@
import { Worker } from "worker_threads";
import { existsSync } from "fs";
import { join } from "path";
import { startStage } from "./request-monitoring";
import { AsyncResource } from "async_hooks";
const MEMORY_BUDGET = Math.max(128, Number(process.env.ANONYMIZATION_MEMORY_MB) || 512) * 1024 * 1024;
const MAX_WORKERS = Math.min(4, Math.max(1, Number(process.env.ANONYMIZATION_WORKERS) || 1));
interface Job {
message: { input: string; output: string; options: unknown; maxOutput: number; context: { mask: string; hostname: string } };
weight: number;
signal: AbortSignal;
resolve: (changed: boolean) => void;
reject: (error: Error) => void;
timer?: NodeJS.Timeout;
abort?: () => void;
waitingDone: ReturnType<typeof startStage>;
transformDone: () => ReturnType<typeof startStage>;
}
const waiting: Job[] = [];
const idle: Worker[] = [];
let running = 0;
let bytes = 0;
export function anonymizeOnWorker(message: Job["message"], size: number, signal: AbortSignal): Promise<boolean> {
const weight = 64 * 1024 * 1024 + size * 4;
if (weight > MEMORY_BUDGET) return Promise.reject(new Error("anonymization_memory_budget_exceeded"));
if (waiting.length >= 16) return Promise.reject(new Error("anonymization_queue_full"));
return new Promise((resolve, reject) => {
const waitingDone = startStage("worker_wait");
// Bind the stage factory to the submitting request, not the preceding job.
const transformDone = AsyncResource.bind(() => startStage("anonymize"));
const job: Job = { message, weight, signal, resolve, reject, waitingDone, transformDone };
job.abort = () => {
const index = waiting.indexOf(job);
if (index < 0) return;
waiting.splice(index, 1);
clearTimeout(job.timer);
signal.removeEventListener("abort", job.abort!);
waitingDone(true);
reject(new Error("anonymization_cancelled"));
};
if (signal.aborted) { waitingDone(true); return reject(new Error("anonymization_cancelled")); }
signal.addEventListener("abort", job.abort, { once: true });
job.timer = setTimeout(job.abort, 30_000);
waiting.push(job);
pump();
});
}
function pump() {
while (waiting.length && running < MAX_WORKERS && bytes + waiting[0].weight <= MEMORY_BUDGET) {
const job = waiting.shift()!;
job.waitingDone();
const transformDone = job.transformDone();
clearTimeout(job.timer);
job.signal.removeEventListener("abort", job.abort!);
running++;
bytes += job.weight;
const compiled = join(__dirname, "anonymization.worker.js");
let worker: Worker;
try { worker = idle.pop() || new Worker(existsSync(compiled) ? compiled : join(__dirname, "anonymization.worker.ts"), {
execArgv: existsSync(compiled) ? [] : ["-r", require.resolve("ts-node/register/transpile-only")],
resourceLimits: { maxOldGenerationSizeMb: Math.ceil(MEMORY_BUDGET / 1024 / 1024) },
}); } catch (error) { running--; bytes -= job.weight; transformDone(true); job.reject(error as Error); continue; }
worker.ref();
let finished = false;
const finish = (error?: Error, changed = false) => {
if (finished) return;
finished = true;
transformDone(!!error);
clearTimeout(deadline);
job.signal.removeEventListener("abort", cancel);
worker.removeListener("message", message);
worker.removeListener("error", failed);
worker.removeListener("exit", exited);
const release = () => { running--; bytes -= job.weight; pump(); };
// Large jobs can leave substantial unreachable buffers in V8 until GC.
// Release their workers before admitting the next job.
if (error || job.weight >= 96 * 1024 * 1024) {
void worker.terminate().then(release, release);
} else { idle.push(worker); worker.unref(); release(); }
if (error) job.reject(error); else job.resolve(changed);
};
const message = (result: { error?: string; changed?: boolean }) => finish(result.error ? new Error(result.error) : undefined, result.changed);
const failed = (error: Error) => finish(error);
const exited = () => finish(new Error("anonymization_worker_exited"));
const cancel = () => finish(new Error("anonymization_cancelled"));
const deadline = setTimeout(() => finish(new Error("anonymization_deadline_exceeded")), 15_000);
worker.once("message", message);
worker.once("error", failed);
worker.once("exit", exited);
job.signal.addEventListener("abort", cancel, { once: true });
if (job.signal.aborted) cancel(); else worker.postMessage(job.message);
}
}
let spoolBytes = 0;
export function reserveSpool(bytes: number) {
if (spoolBytes + bytes > MEMORY_BUDGET * 2) throw new Error("anonymization_spool_budget_exceeded");
spoolBytes += bytes;
}
export function releaseSpool(bytes: number) { spoolBytes -= bytes; }
export function getAnonymizationPoolStats() {
return { running, waiting: waiting.length, idle: idle.length, reservedBytes: bytes, spoolBytes,
maxWorkers: MAX_WORKERS, maxWaiting: 16, memoryBudgetBytes: MEMORY_BUDGET };
}
+26
View File
@@ -0,0 +1,26 @@
import config from "../config";
import { parentPort } from "worker_threads";
import { readFileSync, writeFileSync } from "fs";
import { ContentAnonimizer } from "./anonymize-utils";
parentPort!.on("message", ({ input, output, options, maxOutput, context }) => {
try {
config.ANONYMIZATION_MASK = context.mask;
config.APP_HOSTNAME = context.hostname;
const original = readFileSync(input);
const text = original.toString("utf8");
const anonymizer = new ContentAnonimizer(options);
// Large RE2 jobs legitimately take more than the synchronous one-second
// limit. The pool also terminates this worker after 15 seconds, including
// input/output and compilation. Native backtracking patterns retain their
// own one-second deadline inside ContentAnonimizer.
const transformed = anonymizer.anonymize(text, 10_000);
if (Buffer.byteLength(transformed, "utf8") > maxOutput) throw new Error("anonymized_output_too_large");
const bytes = transformed === text ? original : Buffer.from(transformed, "utf8");
if (bytes.length > maxOutput) throw new Error("anonymized_output_too_large");
writeFileSync(output, bytes, { flag: "wx" });
parentPort!.postMessage({ changed: anonymizer.wasAnonymized });
} catch (error) {
parentPort!.postMessage({ error: (error as Error).message });
}
});
+237 -50
View File
@@ -1,3 +1,10 @@
import { createHash } from "crypto";
import { transformedFile } from "./transformed-cache";
import { promises as fs, createReadStream, ReadStream } from "fs";
import { tmpdir } from "os";
import { join } from "path";
import { anonymizeOnWorker, reserveSpool, releaseSpool } from "./anonymization-pool";
import { maintainTextSpools } from "./temporary-storage";
import { RE2JS } from "re2js";
import { Script } from "vm";
import { basename } from "path";
@@ -6,6 +13,7 @@ import { isBinaryFileSync } from "isbinaryfile";
import { lookup as lookupMime } from "mime-types";
import config from "../config";
import { TermPresenceIndex } from "./term-presence";
import {
parseTermSpec,
termVariants,
@@ -135,11 +143,20 @@ export class AnonymizeTransformer extends Transform {
anonimizer: ContentAnonimizer;
private wholeChunks: Buffer[] = [];
private wholeBytes = 0;
private digest = createHash("sha256");
private spool?: string;
private reservedSpool = 0;
private output?: ReadStream;
private cancelled = new AbortController();
private spoolOperation?: Promise<void>;
private readonly memoryThreshold = 256 * 1024;
private readonly bufferWholeStream: boolean;
constructor(
readonly opt: {
filePath: string;
cacheGeneration?: string;
cacheRevision?: string;
} & ConstructorParameters<typeof ContentAnonimizer>[0]
) {
super();
@@ -170,12 +187,18 @@ export class AnonymizeTransformer extends Transform {
this.nameVerdict = this.isText;
}
if (this.isText && this.bufferWholeStream) {
this.digest.update(chunk);
this.wholeBytes += chunk.length;
if (this.wholeBytes > config.MAX_FILE_SIZE) {
this.wholeChunks = [];
return callback(new Error(`Text file exceeded ${config.MAX_FILE_SIZE} bytes`));
}
this.wholeChunks.push(chunk);
if (this.wholeBytes <= this.memoryThreshold) this.wholeChunks.push(Buffer.from(chunk));
else {
this.spoolOperation = this.spoolChunk(chunk);
void this.spoolOperation.then(() => callback(), error => callback(error));
return;
}
} else {
this.emit("transform", { isText: this.isText, wasAnonimized: false, chunk });
this.push(chunk);
@@ -183,7 +206,54 @@ export class AnonymizeTransformer extends Transform {
callback();
}
private async spoolChunk(chunk: Buffer) {
const additional = chunk.length + (this.spool ? 0 : this.wholeBytes - chunk.length);
reserveSpool(additional);
this.reservedSpool += additional;
if (!this.spool) {
await maintainTextSpools();
this.spool = await fs.mkdtemp(join(tmpdir(), "anonymous-text-"));
await fs.writeFile(join(this.spool, "input"), Buffer.concat(this.wholeChunks));
this.wholeChunks = [];
}
if (this.cancelled.signal.aborted) throw new Error("anonymization_cancelled");
await fs.appendFile(join(this.spool, "input"), chunk);
}
_read(size: number) {
super._read(size);
this.output?.resume();
}
_destroy(error: Error | null, callback: (error?: Error | null) => void) {
this.cancelled.abort();
this.output?.destroy();
this.wholeChunks = [];
void (async () => {
await this.spoolOperation?.catch(() => {});
try { if (this.spool) await fs.rm(this.spool, { recursive: true, force: true }); }
finally { releaseSpool(this.reservedSpool); this.reservedSpool = 0; }
})().then(() => callback(error), () => callback(error));
}
_flush(callback: (error?: Error) => void) {
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))
.then(changed => {
if (this.destroyed) return;
this.anonimizer.wasAnonymized = changed;
this.output = createReadStream(output);
this.output.on("data", chunk => {
this.emit("transform", { isText: true, wasAnonimized: changed, chunk });
if (!this.push(chunk)) this.output!.pause();
});
this.output.once("error", callback);
this.output.once("end", () => callback());
}, error => callback(error));
return;
}
try {
if (this.nameVerdict === null) this.isText = true;
if (this.wholeBytes) {
@@ -211,9 +281,14 @@ const markdownImageRegex =
interface CompiledTermVariant {
// RE2 for regular patterns; time-limited native fallback for JS extensions.
pattern: RE2JS | RegExp;
// A boundary-free, fixed-width native search can cheaply rule out literal
// terms before RE2 scans a large file. A hit still uses the usual matcher.
// A boundary-free native search for fixed-width patterns can rule out
// absent terms. Candidate replacement retains RE2's boundary semantics.
literalPrefilter?: RegExp;
requiredPrefix?: RegExp;
asciiPattern?: RegExp;
asciiReplacement?: string;
asciiBefore?: boolean;
asciiAfter?: boolean;
before: boolean;
after: boolean;
mask: string;
@@ -242,6 +317,29 @@ function hasCatastrophicBacktracking(src: string): boolean {
return false;
}
// A leading sequence of literal characters and simple letter classes must
// occur in every match. Stop before unfamiliar syntax, drop a quantified last
// atom, and reject alternation entirely. This is only an absence check; RE2
// still decides every replacement and all boundaries.
function requiredPrefix(pattern: string): RegExp | undefined {
if (pattern.includes("|")) return undefined;
const atoms: string[] = [];
let offset = 0;
while (offset < pattern.length) {
const char = String.fromCodePoint(pattern.codePointAt(offset)!);
if (char === "[") {
const end = pattern.indexOf("]", offset + 1);
if (end < 0 || !/^[\p{L}\p{N}]+$/u.test(pattern.slice(offset + 1, end))) break;
atoms.push(pattern.slice(offset, end + 1)); offset = end + 1;
} else {
if (/[\\^$.*+?()[\]{}]/.test(char)) break;
atoms.push(char); offset += char.length;
}
}
if (/[?*+{]/.test(pattern.charAt(offset)) && offset < pattern.length) atoms.pop();
return atoms.length ? new RegExp(atoms.join(""), "iu") : undefined;
}
function compileTerms(terms: string[] | undefined): CompiledTermVariant[] {
if (!terms || terms.length === 0) return [];
@@ -269,11 +367,14 @@ function compileTerms(terms: string[] | undefined): CompiledTermVariant[] {
if (!useAsRegex || hasCatastrophicBacktracking(term)) {
term = term.replace(/[-[\]{}()*+?.,\\^$|#]/g, "\\$&");
}
// Only plain literals qualify: expanding their letters produces fixed
// character classes, never user-controlled backtracking. Do not apply
// this shortcut to regex syntax, including escaped literals.
const isLiteral = !/[\\^$.*+?()[\]{}|]/.test(term);
// Plain text and unquantified dots have fixed width in code points.
// Exclude all other regex syntax so the native search cannot backtrack
// exponentially. This includes dotted names, emails and domains.
const isFixedWidth = !/[\\^$*+?()[\]{}|]/.test(term);
for (const variant of termVariants(term)) {
// Stripping a combining-mark-only term can produce an empty variant.
// Keep zero-width matches on RE2's progress-aware iterator.
const fixedWidth = isFixedWidth && variant.pattern.length > 0;
const bounded = withWordBoundaries(variant.pattern, {
sniffSource: variant.sniff,
unicode: variant.unicode,
@@ -286,9 +387,16 @@ function compileTerms(terms: string[] | undefined): CompiledTermVariant[] {
RE2JS.CASE_INSENSITIVE
);
compiled.push({ pattern, before, after, mask,
requiredPrefix: fixedWidth ? undefined : requiredPrefix(variant.pattern),
asciiPattern: fixedWidth ? new RegExp(bounded.replace(/\./g, "[^\\n]"), "giu") : undefined,
asciiReplacement: fixedWidth ? mask.replace(/\$/g, () => "$$") : undefined,
asciiBefore: !variant.unicode && bounded.startsWith("\\b"),
asciiAfter: !variant.unicode && bounded.endsWith("\\b"),
// RE2 uses Unicode case folding even for the non-diacritic pass.
// Omit boundaries so this search can only rule out absent terms.
literalPrefilter: isLiteral ? new RegExp(variant.pattern, "iu") : undefined });
literalPrefilter: fixedWidth
// RE2's dot excludes only LF; JS dot also excludes CR and U+2028/9.
? new RegExp(variant.pattern.replace(/\./g, "[^\\n]"), "iu") : undefined });
} catch {
// Retain JavaScript-only syntax and large repetition counts under
// the execution deadline; RE2 handles the common case without backtracking.
@@ -303,13 +411,37 @@ function compileTerms(terms: string[] | undefined): CompiledTermVariant[] {
return compiled;
}
interface CompiledTermPlan {
terms: CompiledTermVariant[];
presence: TermPresenceIndex | null;
}
// Every file in a ZIP uses the same rules. Reuse immutable compiled patterns
// across transformers, with bounded retention for unrelated repositories.
const termPlans = new Map<string, CompiledTermPlan>();
function compiledTermPlan(terms: string[] | undefined): CompiledTermPlan {
const key = JSON.stringify([config.ANONYMIZATION_MASK, terms || []]);
const cached = termPlans.get(key);
if (cached) return cached;
const compiled = compileTerms(terms);
const plan = {
terms: compiled,
presence: compiled.length && compiled.length <= 256 && compiled.every(term => term.literalPrefilter)
? new TermPresenceIndex(compiled.map(term => term.literalPrefilter!)) : null,
};
if (key.length <= 8192 && compiled.length <= 256) {
if (termPlans.size >= 64) termPlans.delete(termPlans.keys().next().value!);
termPlans.set(key, plan);
}
return plan;
}
export class ContentAnonimizer {
public wasAnonymized = false;
// Compiled once per instance and reused for every anonymize() call.
// Streamed files invoke anonymize() many times per file (one per chunk),
// so caching here avoids rebuilding regexes on every chunk.
// Per-instance metadata references the bounded cache of compiled rules.
private compiledTerms: CompiledTermVariant[];
private selfLinkRegexes: RegExp[] | null = null;
private termPresence: TermPresenceIndex | null = null;
constructor(
readonly opt: {
@@ -321,7 +453,11 @@ export class ContentAnonimizer {
repoId?: string;
}
) {
this.compiledTerms = compileTerms(opt.terms);
const plan = compiledTermPlan(opt.terms);
// Keep per-instance metadata independent. Regex matching is synchronous;
// native replacement resets lastIndex and RE2 creates a fresh matcher.
this.compiledTerms = plan.terms.map(term => ({ ...term }));
this.termPresence = plan.presence;
if (opt.repoName && opt.branchName) {
const r = escapeRegex(opt.repoName);
const b = escapeRegex(opt.branchName);
@@ -370,31 +506,49 @@ export class ContentAnonimizer {
}
private replaceTerms(content: string): string {
for (const c of this.compiledTerms) {
// remove whole url if it contains the term
if (!this.compiledTerms.length) return content;
let candidates = this.termPresence?.candidates(content);
// On ASCII input both engines have identical fixed-width matching and
// word boundaries. Native replacement avoids allocating one JS match
// object per CSV value. Non-ASCII input keeps RE2 validation below.
let asciiContent = (!candidates || candidates.size > 0) && /^\p{ASCII}*$/u.test(content);
for (let index = 0; index < this.compiledTerms.length; index++) {
if (candidates && !candidates.has(index)) continue;
const c = this.compiledTerms[index];
// An absent term cannot occur inside a URL either.
if (c.literalPrefilter && !c.literalPrefilter.test(content)) continue;
if (c.requiredPrefix && !c.requiredPrefix.test(content)) continue;
const previous = content;
content = content.replace(urlRegex, (match) => {
if (replaceTerm(match, c) !== match) {
if (replaceTerm(match, c, true) !== match) {
this.wasAnonymized = true;
return c.mask;
}
return match;
});
// remove the term in the text
const replaced = replaceTerm(content, c);
// URL replacement may already have inserted a non-ASCII custom mask.
if (content !== previous) asciiContent = asciiContent && /^\p{ASCII}*$/u.test(c.mask);
const replaced = replaceTerm(content, c, asciiContent);
if (replaced !== content) this.wasAnonymized = true;
content = replaced;
// Masks and changed boundaries can introduce later matches. Re-index
// only after a real change, never once per absent term.
if (content !== previous) {
asciiContent = asciiContent && /^\p{ASCII}*$/u.test(c.mask);
candidates = this.termPresence?.candidates(content, index);
}
}
return content;
}
anonymize(content: string): string {
anonymize(content: string, timeoutMs = 1000): string {
return runWithAnonymizationDeadline(() => {
content = this.removeImage(content);
content = this.removeLink(content);
content = this.replaceGitHubSelfLinks(content);
content = this.replaceTerms(content);
return content;
});
}, timeoutMs);
}
}
@@ -433,42 +587,70 @@ function escapeRegex(value: string): string {
// V8 interrupts even a native RegExp that never returns to JavaScript. A
// timeout aborts the operation instead of returning partially anonymized text.
const anonymizationScript = new Script("run()");
function runWithAnonymizationDeadline(run: () => string): string {
return anonymizationScript.runInNewContext({ run }, { timeout: 1000 });
function runWithAnonymizationDeadline(run: () => string, timeoutMs = 1000): string {
return anonymizationScript.runInNewContext({ run }, { timeout: timeoutMs });
}
function replaceTerm(content: string, term: CompiledTermVariant): string {
if (term.literalPrefilter && !term.literalPrefilter.test(content)) return content;
if (term.pattern instanceof RegExp) {
return content.replace(term.pattern, () => term.mask);
function replaceTerm(content: string, term: CompiledTermVariant, asciiContent = false): string {
const pattern = term.pattern;
if (pattern instanceof RegExp) {
// JavaScript-only patterns can backtrack even inside an isolated worker.
// Keep their short limit independent of the whole-file processing budget.
return runWithAnonymizationDeadline(() => content.replace(pattern, () => term.mask));
}
// Fixed-width literal searches run natively. Validate each candidate with
// RE2 on a small window to preserve its case folding and boundary semantics
// without scanning megabytes of notebook outputs in the JS regex engine.
const candidates = term.literalPrefilter
? content.matchAll(new RegExp(term.literalPrefilter.source, "giu"))
: null;
const matcher = candidates ? null : term.pattern.matcher(content);
if (term.requiredPrefix && !term.requiredPrefix.test(content)) return content;
if (term.literalPrefilter && asciiContent && term.asciiPattern) {
return content.replace(term.asciiPattern, term.asciiReplacement!);
}
if (term.literalPrefilter) {
const validated = new Map<string, boolean>();
const candidates = new RegExp(term.literalPrefilter.source, "giu");
const pieces: string[] = [];
let cursor = 0;
let match: RegExpExecArray | null;
while ((match = candidates.exec(content))) {
const candidate = match[0], start = match.index, end = start + candidate.length;
let valid = true;
if (term.before && /[\p{L}\p{N}_]$/u.test(content.slice(Math.max(0, start - 2), start))) valid = false;
if (term.after && /^[\p{L}\p{N}_]/u.test(content.slice(end, end + 2))) valid = false;
// On ASCII candidates the fixed-width native and RE2 searches agree.
// Check RE2's ASCII word boundaries explicitly: JS /iu\b/ also treats
// Kelvin sign and long s as word characters, while RE2 does not.
if (valid && /^\p{ASCII}*$/u.test(candidate)) {
if (term.asciiBefore && asciiWordAt(content, start - 1) === asciiWordAt(content, start)) valid = false;
if (term.asciiAfter && asciiWordAt(content, end - 1) === asciiWordAt(content, end)) valid = false;
} else if (valid) {
// Keep RE2 authoritative for Unicode case folding. Repeated values in
// datasets share small contexts; cap this cache independently of size.
const offset = Math.max(0, start - 2);
const window = content.slice(offset, end + 2);
const key = `${start - offset}:${window}`;
const cached = validated.get(key);
if (cached === undefined) {
const check = (term.pattern as RE2JS).matcher(window);
valid = check.find(start - offset) && check.start() === start - offset && check.end() === end - offset;
if (validated.size < 256) validated.set(key, valid);
} else valid = cached;
}
if (!valid) {
// A boundary-rejected candidate must not hide an overlapping match,
// e.g. the second a-a in "xa-a-a". Advance by one Unicode code point.
candidates.lastIndex = start + (content.codePointAt(start)! > 0xffff ? 2 : 1);
continue;
}
pieces.push(content.slice(cursor, start), term.mask);
cursor = end;
}
if (!pieces.length) return content;
pieces.push(content.slice(cursor));
return pieces.join("");
}
const matcher = pattern.matcher(content);
const pieces: string[] = [];
let cursor = 0;
const matches = function* () {
if (candidates) {
for (const candidate of candidates) {
const start = candidate.index!;
const end = start + candidate[0].length;
const offset = Math.max(0, start - 2);
const check = (term.pattern as RE2JS).matcher(content.slice(offset, end + 2));
if (check.find(start - offset) && check.start() === start - offset && check.end() === end - offset) {
yield { start, end };
}
}
} else {
while (matcher!.find()) yield { start: matcher!.start(), end: matcher!.end() };
}
};
for (const { start, end } of matches()) {
// RE2 has no lookahead. Check the generated Unicode word boundaries
// outside the engine, without executing any user-supplied native regex.
while (matcher.find()) {
const start = matcher.start(), end = matcher.end();
if (term.before && /[\p{L}\p{N}_]$/u.test(content.slice(Math.max(0, start - 2), start))) continue;
if (term.after && /^[\p{L}\p{N}_]/u.test(content.slice(end, end + 2))) continue;
pieces.push(content.slice(cursor, start), term.mask);
@@ -478,3 +660,8 @@ function replaceTerm(content: string, term: CompiledTermVariant): string {
pieces.push(content.slice(cursor));
return pieces.join("");
}
function asciiWordAt(content: string, index: number): boolean {
const code = content.charCodeAt(index);
return code >= 48 && code <= 57 || code >= 65 && code <= 90 || code >= 97 && code <= 122 || code === 95;
}
+17
View File
@@ -0,0 +1,17 @@
import { Archiver, EntryData } from "archiver";
import { Readable } from "stream";
/** Allow only one active entry; compression and the response provide backpressure. */
export function appendArchiveEntry(archive: Archiver, source: Readable, options: EntryData): Promise<void> {
return new Promise((resolve, reject) => {
const cleanup = () => {
archive.removeListener("entry", done); archive.removeListener("error", fail);
archive.removeListener("close", closed); source.removeListener("error", fail);
};
const done = () => { cleanup(); resolve(); };
const fail = (error: Error) => { cleanup(); source.destroy(); reject(error); };
const closed = () => fail(new Error("archive_cancelled"));
archive.once("entry", done); archive.once("error", fail); archive.once("close", closed);
source.once("error", fail);
archive.append(source, options);
});
}
+22
View File
@@ -0,0 +1,22 @@
/** Bounded TTL cache with one calculation per key and retry after failure. */
export class AsyncCache<T> {
private entries = new Map<string, { expires: number; value: Promise<T> }>();
constructor(private ttl: number, private capacity: number, private now = () => Date.now()) {}
get size() { this.prune(); return this.entries.size; }
clear() { this.entries.clear(); }
private prune() {
for (const [key, entry] of this.entries) if (entry.expires <= this.now()) this.entries.delete(key);
}
get(key: string, calculate: () => Promise<T>): Promise<T> {
this.prune();
const cached = this.entries.get(key);
if (cached) return cached.value;
if (this.entries.size >= this.capacity) this.entries.delete(this.entries.keys().next().value!);
const entry = { expires: Infinity, value: Promise.resolve().then(calculate) };
this.entries.set(key, entry);
entry.value.then(() => { entry.expires = this.now() + this.ttl; }, () => {
if (this.entries.get(key) === entry) this.entries.delete(key);
});
return entry.value;
}
}
+83
View File
@@ -0,0 +1,83 @@
import { createClient } from "redis";
import { randomUUID } from "crypto";
import config from "../config";
let connection: ReturnType<typeof createClient> | undefined;
let connecting: Promise<ReturnType<typeof createClient> | undefined> | undefined;
let retryAt = 0;
/** Bound commands even when a connected server stops replying. */
export async function cacheCommand<T>(command: Promise<T>, client: ReturnType<typeof createClient>, timeoutMs = 1000): Promise<T> {
let timer: NodeJS.Timeout | undefined;
try {
return await Promise.race([command, new Promise<never>((_, reject) => {
timer = setTimeout(() => {
retryAt = Date.now() + 30_000;
if (connection === client) connection = undefined;
if (client.isOpen) client.destroy();
reject(new Error("cache_command_timeout"));
}, timeoutMs);
})]);
} finally { clearTimeout(timer); }
}
export async function cacheRedis() {
if (connection?.isReady) return connection;
if (connecting) return connecting;
if (Date.now() < retryAt) return undefined;
connecting = (async () => {
const client = createClient({ disableOfflineQueue: true, socket: {
host: config.REDIS_HOSTNAME, port: config.REDIS_PORT,
connectTimeout: 500, reconnectStrategy: false,
} });
client.on("error", () => {});
try { await cacheCommand(client.connect(), client, 500); connection = client; return client; }
catch { retryAt = Date.now() + 30_000; if (client.isOpen) client.destroy(); return undefined; }
finally { connecting = undefined; }
})();
return connecting;
}
/** A renewable lease. Followers recheck durable output; they never share a Readable. */
export async function coordinatedFill<T>(
key: string, read: () => Promise<T | undefined>, produce: () => Promise<T>,
waitMs = 120_000
): Promise<T> {
const redis = await cacheRedis();
if (!redis) return produce();
const lockKey = `perf:lease:${key}`;
const token = randomUUID();
const deadline = Date.now() + waitMs;
for (;;) {
const value = await read();
if (value !== undefined) return value;
let owned: string | null;
try { owned = await cacheCommand(redis.set(lockKey, token, { NX: true, PX: 30_000 }), redis); }
catch { return produce(); }
if (owned) {
const heartbeat = setInterval(() => {
void cacheCommand(redis.eval("if redis.call('get',KEYS[1]) == ARGV[1] then return redis.call('pexpire',KEYS[1],ARGV[2]) else return 0 end", {
keys: [lockKey], arguments: [token, "30000"],
}), redis).catch(() => {});
}, 10_000);
heartbeat.unref();
try {
// Coordination is optional. A lost heartbeat cannot invalidate output
// that the producer has already completed and verified.
return await produce();
} finally {
clearInterval(heartbeat);
await cacheCommand(redis.eval("if redis.call('get',KEYS[1]) == ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end", {
keys: [lockKey], arguments: [token],
}), redis).catch(() => {});
}
}
if (Date.now() >= deadline) throw new Error("cache_fill_timeout");
await new Promise(resolve => setTimeout(resolve, 100));
}
}
export async function closeCacheRedis() {
const current = connection;
connection = undefined;
retryAt = 0;
if (current?.isOpen) current.destroy();
}
+35
View File
@@ -0,0 +1,35 @@
import { Document, Model } from "mongoose";
import { RepositoryStatus } from "./types";
type ContentSnapshot = Document & {
status?: RepositoryStatus;
statusDate?: Date;
anonymizeDate?: Date;
options: { expirationMode?: string; expirationDate?: Date };
githubAccess?: { revision?: string };
};
// Embedded content and lifecycle state live in one document. Claim and clear
// them in one conditional write so an extension or rebuild cannot interleave.
export async function expireEmbeddedContent<T extends ContentSnapshot>(
model: Model<T>, data: T, fields: Record<string, unknown>, unset: Record<string, ""> = {}
) {
const now = new Date();
const result = await model.updateOne({
_id: data._id,
status: data.status ?? { $exists: false },
statusDate: data.statusDate ?? { $exists: false },
anonymizeDate: data.anonymizeDate ?? { $exists: false },
"options.expirationMode": data.options.expirationMode ?? { $exists: false },
"options.expirationDate": data.options.expirationDate ?? { $exists: false },
"githubAccess.revision": data.githubAccess?.revision ?? { $exists: false },
$and: [
{ status: { $in: [RepositoryStatus.READY, RepositoryStatus.EXPIRING] } },
{ "options.expirationMode": { $ne: "never" } },
{ "options.expirationDate": { $lte: now } },
],
}, { $set: { ...fields, status: RepositoryStatus.EXPIRED, statusDate: now }, $unset: unset }).exec();
if (!result.matchedCount) return;
data.set({ ...fields, status: RepositoryStatus.EXPIRED, statusDate: now });
for (const key of Object.keys(unset)) data.set(key, undefined);
}
+9
View File
@@ -0,0 +1,9 @@
import { createHash } from "crypto";
export function contentGenerationPrefix(generation: string) {
return `__content/${createHash("sha256").update(generation).digest("hex")}`;
}
export function contentRetirementMarker(prefix: string) {
if (!/^__content\/[a-f0-9]{64}$/.test(prefix)) throw new Error("Invalid content generation prefix");
return `__content/__retired/${prefix.slice("__content/".length)}.txt`;
}
+5 -5
View File
@@ -253,23 +253,23 @@ export async function boundAppToken(ownerId: string, binding: RepositoryAccess,
return token;
}
export async function selectRepositoryAccess(ownerId: string, fullName: string, choice?: unknown): Promise<{ token: string; binding: RepositoryAccess }> {
export async function selectRepositoryAccess(ownerId: string, fullName: string, choice?: unknown, repositoryId?: number): Promise<{ token: string; binding: RepositoryAccess; fullName?: string }> {
if (!/^[^/\s]+\/[^/\s]+$/.test(fullName)) throw appError("repo_not_found", 400);
if (choice !== undefined && choice !== "oauth" && choice !== "github-app") throw appError("invalid_connection", 400);
const hasApp = config.GITHUB_APP_ENABLED && await CredentialModel.exists({ ownerId, provider: APP_PROVIDER });
if (choice === "github-app" || (choice === undefined && hasApp)) {
const repo = (await appRepositories(ownerId)).find(r => r.full_name.toLowerCase() === fullName.toLowerCase());
const repo = (await appRepositories(ownerId)).find(r => repositoryId === undefined ? r.full_name.toLowerCase() === fullName.toLowerCase() : r.id === repositoryId);
if (!repo) {
const userToken = await appUserToken(ownerId);
const publicRepo = await githubRequest<GitHubRepositoryInfo>(
`/repos/${fullName.split("/").map(encodeURIComponent).join("/")}`, userToken);
repositoryId === undefined ? `/repos/${fullName.split("/").map(encodeURIComponent).join("/")}` : `/repositories/${repositoryId}`, userToken);
if (publicRepo.private !== false || publicRepo.visibility !== "public" || !Number.isSafeInteger(publicRepo.id)) throw appError("github_app_access_required");
const binding: RepositoryAccess = { kind: "github-app", publicRead: true,
repositoryId: publicRepo.id, revision: randomUUID() };
return { binding, token: await boundAppToken(ownerId, binding, fullName) };
return { binding, fullName: publicRepo.full_name, token: await boundAppToken(ownerId, binding, publicRepo.full_name) };
}
const binding: RepositoryAccess = { kind: "github-app", repositoryId: repo.id, installationId: repo.installationId, revision: randomUUID() };
return { binding, token: await boundAppToken(ownerId, binding, fullName) };
return { binding, fullName: repo.full_name, token: await boundAppToken(ownerId, binding, repo.full_name) };
}
const token = await getCredentialToken(ownerId);
if (!token) throw appError("github_oauth_required");
+1 -1
View File
@@ -74,7 +74,7 @@ function getRedis(): RedisClientType | null {
const c = redisClient;
redisClient = null;
if (c?.isOpen) {
c.disconnect().catch(() => {});
c.destroy();
}
});
redisClient.connect().catch(() => {
+11
View File
@@ -0,0 +1,11 @@
import { IFile } from "./files/files.types";
import { model, Schema } from "mongoose";
const schema = new Schema({ repoId: { type: String, required: true }, key: { type: String, required: true }, originalId: Schema.Types.ObjectId,
name: { type: String, required: true }, path: { type: String, default: "", required: true }, sha: String, size: Number,
anonymousName: String, anonymousDirectory: String, anonymousPath: String,
searchName: String, searchDirectory: String,
createdAt: { type: Date, default: Date.now } });
schema.index({ repoId: 1, key: 1, anonymousPath: 1, path: 1, name: 1 });
schema.index({ repoId: 1, key: 1, originalId: 1 }, { unique: true });
schema.index({ createdAt: 1 }, { expireAfterSeconds: 30 * 24 * 3600 });
export default model<IFile & { key: string; originalId: Schema.Types.ObjectId; anonymousName: string; anonymousDirectory: string; anonymousPath: string; searchName: string; searchDirectory: string; createdAt: Date }>("AnonymizedPath", schema);
@@ -68,4 +68,6 @@ const AnonymizedGistSchema = new Schema({
},
});
AnonymizedGistSchema.index({ status: 1, "options.expirationDate": 1 });
export default AnonymizedGistSchema;
@@ -65,4 +65,6 @@ const AnonymizedPullRequestSchema = new Schema({
},
});
AnonymizedPullRequestSchema.index({ status: 1, "options.expirationDate": 1 });
export default AnonymizedPullRequestSchema;
@@ -48,6 +48,19 @@ const AnonymizedRepositorySchema = new Schema({
repositoryName: String,
accessToken: { type: String, select: false },
},
cleanupToken: { type: String, select: false },
cleanupUntil: { type: Date, select: false },
treeGeneration: String,
stagedFileTrees: { type: [{ generation: String, until: Date }], default: undefined },
retiredTreeGenerations: { type: [String], default: undefined },
retiredContentPrefixes: { type: [String], default: undefined },
legacyContentCleanupPending: Boolean,
contentCacheVersion: Number,
contentCacheRevision: String,
emptyTreeGeneration: String,
fileMetadataRevision: String,
pathIndexKey: String,
pathIndexBuiltAt: Date,
truncatedFolders: {
type: [String],
default: [],
@@ -71,6 +84,7 @@ const AnonymizedRepositorySchema = new Schema({
type: Date,
default: Date.now,
},
sizeComputedAt: Date,
size: {
storage: {
type: Number,
@@ -88,10 +102,17 @@ const AnonymizedRepositorySchema = new Schema({
});
AnonymizedRepositorySchema.index({ "source.repositoryName": 1 });
AnonymizedRepositorySchema.index({ "coauthors.githubId": 1 });
AnonymizedRepositorySchema.index({ retiredTreeGenerations: 1 }, { sparse: true });
AnonymizedRepositorySchema.index({ "stagedFileTrees.until": 1 }, { sparse: true });
AnonymizedRepositorySchema.index({ retiredContentPrefixes: 1 }, { sparse: true });
AnonymizedRepositorySchema.index({ legacyContentCleanupPending: 1, status: 1 },
{ partialFilterExpression: { legacyContentCleanupPending: true } });
AnonymizedRepositorySchema.index({ status: 1, statusDate: 1 });
AnonymizedRepositorySchema.index({ lastView: 1 });
AnonymizedRepositorySchema.index({ anonymizeDate: 1 });
AnonymizedRepositorySchema.index({ status: 1, isReseted: 1, lastView: 1 });
AnonymizedRepositorySchema.index({ status: 1, "options.expirationDate": 1 });
AnonymizedRepositorySchema.index({
status: 1,
isReseted: 1,
@@ -30,6 +30,17 @@ export interface IAnonymizedRepository {
photo?: string;
addedAt?: Date;
}[];
treeGeneration?: string;
stagedFileTrees?: { generation: string; until: Date }[];
retiredTreeGenerations?: string[];
retiredContentPrefixes?: string[];
legacyContentCleanupPending?: boolean;
contentCacheVersion?: number;
contentCacheRevision?: string;
emptyTreeGeneration?: string;
fileMetadataRevision?: string;
pathIndexKey?: string;
pathIndexBuiltAt?: Date;
truncatedFolders: string[];
conference: string;
options: {
@@ -49,6 +60,7 @@ export interface IAnonymizedRepository {
};
pageView: number;
lastView: Date;
sizeComputedAt?: Date;
size: {
storage: number;
file: number;
+5
View File
@@ -0,0 +1,5 @@
import { model, Schema } from "mongoose";
const schema = new Schema({ owner: { type: Schema.Types.ObjectId, required: true },
type: { type: String, required: true }, artifactId: { type: String, required: true }, name: String });
schema.index({ owner: 1, type: 1, artifactId: 1 }, { unique: true });
export default model("DashboardName", schema);
+8
View File
@@ -4,6 +4,9 @@ const FileSchema = new Schema({
name: { type: String, index: true },
path: { type: String, index: true },
repoId: { type: String, index: true },
treeGeneration: String,
treeStaged: Boolean,
metadataPending: Boolean,
sha: {
type: String,
},
@@ -13,6 +16,11 @@ const FileSchema = new Schema({
});
FileSchema.index({ path: 1, repoId: 1 });
FileSchema.index({ metadataPending: 1, repoId: 1, treeGeneration: 1 },
{ partialFilterExpression: { metadataPending: true } });
FileSchema.index({ treeStaged: 1, repoId: 1, treeGeneration: 1 },
{ partialFilterExpression: { treeStaged: true } });
FileSchema.index({ repoId: 1, treeGeneration: 1, path: 1, name: 1 });
FileSchema.index(
{ repoId: 1, size: 1, path: 1 },
{ name: "repoId_1_size_1_path_1" }
+3
View File
@@ -4,6 +4,9 @@ export interface IFile {
name: string;
path: string;
repoId: string;
treeGeneration?: string;
treeStaged?: boolean;
metadataPending?: boolean;
sha?: string;
size?: number;
}
+1
View File
@@ -31,6 +31,7 @@ const UserSchema = new Schema({
},
],
photo: String,
projectNames: { type: Map, of: String },
repositories: [
{
type: String,
+1
View File
@@ -25,6 +25,7 @@ export interface IUser {
default: boolean;
}[];
photo?: string;
projectNames?: Map<string, string>;
repositories?: number[];
default?: {
+216
View File
@@ -0,0 +1,216 @@
import { createClient, RedisClientType } from "redis";
import { hostname } from "os";
import { readdir, readFile, readlink } from "fs/promises";
import { monitorEventLoopDelay, performance } from "perf_hooks";
import config from "../config";
import { AsyncCache } from "./async-cache";
import { activeRequestCount, LATENCY_BUCKETS, Metric, requestMetrics, Service } from "./request-monitoring";
const PREFIX = "performance:v1:";
const RETENTION = 2 * 3600;
let client: RedisClientType | undefined;
let stop: (() => Promise<void>) | undefined;
let flushNow: (() => Promise<void>) | undefined;
const reports = new AsyncCache<unknown>(10000, 2);
const emptyMetric = (metric: string, route: string, method: string): Metric => ({ metric, route, method,
count: 0, sumMs: 0, bytes: 0, aborted: 0, errors: 0, slow: 0, buckets: Array(LATENCY_BUCKETS.length + 1).fill(0) });
async function containerMemoryLimit() {
for (const path of ["/sys/fs/cgroup/memory.max", "/sys/fs/cgroup/memory/memory.limit_in_bytes"]) {
const raw = await readFile(path, "utf8").catch(() => "");
const value = Number(raw.trim());
if (value > 0 && value < Number.MAX_SAFE_INTEGER) return value;
}
return null;
}
async function redisDeadline<T>(work: Promise<T>, redis: RedisClientType): Promise<T> {
let timer: NodeJS.Timeout | undefined;
try {
return await Promise.race([work, new Promise<never>((_resolve, reject) => {
timer = setTimeout(() => {
reject(new Error("performance_monitoring_timeout"));
if (redis.isOpen) {
redis.destroy();
if (client === redis) void redis.connect().catch(() => {});
}
}, 2000); timer.unref();
})]);
} finally { clearTimeout(timer); }
}
export async function flushPerformanceMetrics() { await flushNow?.(); reports.clear(); }
export function percentileUpperBound(buckets: number[], percentile: number): number | null {
const total = buckets.reduce((sum, count) => sum + count, 0);
if (!total) return null;
let count = 0;
for (let i = 0; i < buckets.length; i++) {
count += buckets[i];
if (count >= Math.ceil(total * percentile)) return LATENCY_BUCKETS[i] ?? null;
}
return null;
}
/** PID-owned sockets only: /proc/net also includes healthcheck subprocesses. */
export async function socketSnapshot() {
try {
const descriptors = (await readdir("/proc/self/fd")).slice(0, 4096);
const sockets = new Set<string>();
for (let i = 0; i < descriptors.length; i += 32) {
const links = await Promise.all(descriptors.slice(i, i + 32).map(fd => readlink(`/proc/self/fd/${fd}`).catch(() => "")));
for (const link of links) { const match = /^socket:\[(\d+)\]$/.exec(link); if (match) sockets.add(match[1]); }
}
const states: Record<string, number> = {};
for (const file of ["/proc/self/net/tcp", "/proc/self/net/tcp6"]) {
const table = await readFile(file, "utf8");
for (const line of table.trim().split("\n").slice(1)) {
const columns = line.trim().split(/\s+/);
if (sockets.has(columns[9])) states[columns[3]] = (states[columns[3]] || 0) + 1;
}
}
return { descriptors: sockets.size, established: states["01"] || 0, closeWait: states["08"] || 0,
listening: states["0A"] || 0, truncated: descriptors.length === 4096 };
} catch { return null; }
}
/** One bounded batch every 15s. Redis outages never queue commands or hold HTTP requests. */
export function startPerformanceMonitoring(service: Service, workers: () => unknown = () => null) {
if (stop) return stop;
const redis = createClient({ disableOfflineQueue: true, socket: { host: config.REDIS_HOSTNAME,
port: config.REDIS_PORT, connectTimeout: 2000, reconnectStrategy: (retries: number) => Math.min(5000, 250 * (retries + 1)) } }) as RedisClientType;
client = redis;
redis.on("error", () => {});
void redis.connect().catch(() => {});
const loop = monitorEventLoopDelay({ resolution: 20 }); loop.enable();
let previousUtilization = performance.eventLoopUtilization();
let previousCpu = process.cpuUsage();
let previousCpuAt = performance.now();
let busy = false;
let droppedBatches = 0;
let lastFlushAt: number | null = null;
let stopped = false;
const instance = `${service}:${hostname()}`;
const flush = async () => {
if (busy || stopped) return;
busy = true;
const entries = requestMetrics.drain();
try {
if (!redis.isReady) { droppedBatches += entries.length ? 1 : 0; return; }
const utilization = performance.eventLoopUtilization(previousUtilization);
previousUtilization = performance.eventLoopUtilization();
const cpu = process.cpuUsage(previousCpu);
const cpuAt = performance.now();
const cpuPercent = (cpu.user + cpu.system) / ((cpuAt - previousCpuAt) * 1000) * 100;
previousCpu = process.cpuUsage(); previousCpuAt = cpuAt;
const runtime = { instance, service, sampledAt: Date.now(), uptime: process.uptime(),
cpuPercent, memoryLimitBytes: await containerMemoryLimit(),
memory: process.memoryUsage(), activeRequests: activeRequestCount(), sockets: await socketSnapshot(), workers: workers(),
eventLoop: { utilization: utilization.utilization, p95Ms: loop.percentile(95) / 1e6,
maxMs: Number.isFinite(loop.max) ? loop.max / 1e6 : 0 },
droppedMetrics: requestMetrics.dropped, droppedBatches, lastFlushAt };
loop.reset();
const tx = redis.multi();
const touched = new Set<string>();
for (const { minute, data } of entries) {
const key = `${PREFIX}minute:${service}:${minute}`;
touched.add(key);
const label = JSON.stringify([data.metric, data.route, data.method]);
const fields: Record<string, number> = { count: data.count, sumMs: data.sumMs, bytes: data.bytes,
aborted: data.aborted, errors: data.errors, slow: data.slow };
data.buckets.forEach((count, i) => { fields[`b${i}`] = count; });
for (const [field, value] of Object.entries(fields)) if (value) tx.hIncrBy(key, `${label}|${field}`, value);
}
for (const key of touched) tx.expire(key, RETENTION);
tx.set(`${PREFIX}instance:${instance}`, JSON.stringify(runtime), { EX: 90 });
tx.zAdd(`${PREFIX}instances`, { score: runtime.sampledAt, value: instance });
tx.zRemRangeByScore(`${PREFIX}instances`, 0, runtime.sampledAt - 90000);
tx.expire(`${PREFIX}instances`, 120);
tx.lPush(`${PREFIX}runtime:${service}`, JSON.stringify(runtime));
tx.lTrim(`${PREFIX}runtime:${service}`, 0, service === "api" ? 239 : 959);
tx.expire(`${PREFIX}runtime:${service}`, RETENTION);
await redisDeadline(tx.exec(), redis); lastFlushAt = Date.now();
} catch { droppedBatches++; }
finally { busy = false; }
};
const timer = setInterval(() => { void flush(); }, 15000); timer.unref();
flushNow = flush;
void flush();
stop = async () => {
stopped = true; clearInterval(timer); loop.disable();
if (redis.isOpen) redis.destroy();
client = undefined; stop = undefined; flushNow = undefined; reports.clear();
};
return stop;
}
export async function performanceReport(minutes: number) {
const windowMinutes = minutes === 60 ? 60 : 15;
if (!client?.isReady) return { available: false, windowMinutes, instances: [], routes: [], stages: [] };
const redis = client;
return reports.get(String(windowMinutes), async () => {
const minute = Math.floor(Date.now() / 60000) * 60000;
const groups = new Map<string, Metric & { service: Service }>();
const tx = redis.multi();
const requests: { service: Service; minute: number }[] = [];
for (const service of ["api", "streamer"] as const) {
for (let i = 0; i < windowMinutes; i++) {
requests.push({ service, minute: minute - i * 60000 });
tx.hGetAll(`${PREFIX}minute:${service}:${minute - i * 60000}`);
}
}
const values = await redisDeadline(tx.exec(), redis) as unknown as Record<string, string>[];
const series = new Map<number, { minute: number; requests: number; aborted: number; errors: number; sumMs: number }>();
values.forEach((hash, index) => {
for (const [key, raw] of Object.entries(hash || {})) {
const split = key.lastIndexOf("|");
const label = key.slice(0, split); const field = key.slice(split + 1);
const [metric, route, method] = JSON.parse(label) as string[];
const service = requests[index].service;
const id = `${service}:${label}`;
let row = groups.get(id);
if (!row) { row = { ...emptyMetric(metric, route, method), service }; groups.set(id, row); }
const value = Number(raw) || 0;
if (/^b\d+$/.test(field)) row.buckets[Number(field.slice(1))] += value;
else if (["count", "sumMs", "bytes", "aborted", "errors", "slow"].includes(field)) {
const numericField = field as "count" | "sumMs" | "bytes" | "aborted" | "errors" | "slow";
row[numericField] += value;
}
if (metric === "request" && service === "api") {
const timestamp = requests[index].minute;
const point = series.get(timestamp) || { minute: timestamp, requests: 0, aborted: 0, errors: 0, sumMs: 0 };
if (field === "count") point.requests += value;
if (field === "sumMs") point.sumMs += value;
if (field === "aborted") point.aborted += value;
if (field === "errors") point.errors += value;
series.set(timestamp, point);
}
}
});
const ids = await redisDeadline(redis.zRangeByScore(`${PREFIX}instances`, Date.now() - 90000, "+inf", { LIMIT: { offset: 0, count: 32 } }), redis);
const samples = ids.length ? await redisDeadline(redis.mGet(ids.map(id => `${PREFIX}instance:${id}`)), redis) : [];
const instances = samples.filter((sample): sample is string => !!sample).map(sample => JSON.parse(sample));
const runtimeSeries: { instance: string; service: string; sampledAt: number; rss: number; heapUsed: number; external: number;
cpuPercent: number; loopP95Ms: number; sockets: number | null; closeWait: number | null; waiting: number; reservedBytes: number }[] = [];
for (const service of ["api", "streamer"]) {
const samples = await redisDeadline(redis.lRange(`${PREFIX}runtime:${service}`, 0, service === "api" ? 239 : 959), redis);
for (const raw of samples) {
const sample = JSON.parse(raw);
if (sample.sampledAt >= Date.now() - windowMinutes * 60000) runtimeSeries.push({ service, instance: sample.instance,
sampledAt: sample.sampledAt, rss: sample.memory.rss, heapUsed: sample.memory.heapUsed, external: sample.memory.external,
cpuPercent: sample.cpuPercent, loopP95Ms: sample.eventLoop.p95Ms, sockets: sample.sockets?.descriptors ?? null,
closeWait: sample.sockets?.closeWait ?? null, waiting: sample.workers?.waiting || 0, reservedBytes: sample.workers?.reservedBytes || 0 });
}
}
const rows = [...groups.values()].map(row => ({ ...row, avgMs: row.count ? Math.round(row.sumMs / row.count) : 0,
p95UpperMs: percentileUpperBound(row.buckets, .95), p99UpperMs: percentileUpperBound(row.buckets, .99) }));
return { available: true, generatedAt: Date.now(), windowMinutes, instances,
routes: rows.filter(row => row.metric === "request").map(row => {
const firstByte = rows.find(sample => sample.metric === "first_byte" && sample.service === row.service && sample.route === row.route && sample.method === row.method);
return { ...row, firstByteCount: firstByte?.count || 0, firstByteP95UpperMs: firstByte?.p95UpperMs ?? null };
}).sort((a, b) => b.sumMs - a.sumMs),
stages: rows.filter(row => row.metric !== "request"), series: [...series.values()].sort((a, b) => a.minute - b.minute),
runtimeSeries, latencyBucketsMs: LATENCY_BUCKETS,
percentileNote: "Percentiles are histogram bucket upper bounds; null means no samples or above 60s." };
});
}
+159
View File
@@ -0,0 +1,159 @@
import { AsyncLocalStorage } from "async_hooks";
import { randomUUID } from "crypto";
import { performance } from "perf_hooks";
import { RequestHandler } from "express";
import { createLogger } from "./logger";
export const LATENCY_BUCKETS = [10, 50, 100, 250, 500, 1000, 2500, 5000, 10000, 30000, 60000];
export type Service = "api" | "streamer";
export type Stage = "repository" | "authorization" | "upstream" | "cache_lookup" | "cache_fill" | "worker_wait" | "anonymize" | "streamer_headers";
export interface Metric {
metric: string; route: string; method: string; count: number; sumMs: number;
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 methods = new Set(["GET", "HEAD", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"]);
const staticRoutes = new Set([
"/", "/dashboard", "/anonymize", "/healthcheck", "/api/healthcheck", "/api/stat", "/api/stat/history",
"/api/options", "/api/message", "/api/user", "/api/user/quota", "/api/user/dashboard", "/api/repo",
"/api/anonymize-preview", "/api/admin/overview", "/api/admin/performance", "/api/admin/errors",
"/api/admin/errors/stats", "/api/admin/queues", "/api/admin/queues/metrics", "/github/app/webhook",
]);
/** Fixed labels: never use repository IDs, file names or query strings as metric keys. */
export function requestRoute(url: string, service: Service): string {
const path = url.split("?", 1)[0].replace(/\/+$/, "") || "/";
if (service === "streamer") return ["/api/download", "/api", "/healthcheck"].includes(path) ? path : "other";
if (staticRoutes.has(path)) return path;
if (/^\/api\/repo\//.test(path)) {
const match = path.match(/^\/api\/repo\/[^/]+\/(file|files|zip|options|refresh|coauthors|extend)(?:\/|$)/);
if (match) {
const subroute = match[1] === "files" && /\/(counts|search)$/.exec(path);
return `/api/repo/:repoId/${match[1]}${match[1] === "file" ? "/:path" : subroute ? `/${subroute[1]}` : ""}`;
}
const github = path.match(/^\/api\/repo\/[^/]+\/[^/]+\/(readme|branches)$/);
if (github) return `/api/repo/:owner/:repo/${github[1]}`;
return "/api/repo/:repoId";
}
if (path.startsWith("/w/")) return "/w/:repoId/:path";
if (/^\/(r|repository)\//.test(path)) return "/r/:repoId/:path";
if (/^\/api\/(pr|gist)\//.test(path)) return `/api/${path.split("/")[2]}/:id/:path`;
if (/^\/(script|css|favicon|fonts|partials)\//.test(path)) return "/assets/:path";
if (path.startsWith("/github/")) return "/github/:action";
if (path.startsWith("/api/admin/")) return "/api/admin/:action";
if (path.startsWith("/api/user/")) return "/api/user/:action";
return "other";
}
export class RequestMetrics {
private pending = new Map<string, { minute: number; data: Metric }>();
dropped = 0;
constructor(private capacity = 512, private now = () => Date.now()) {}
get size() { return this.pending.size; }
observe(metric: string, route: string, method: string, ms: number,
detail: { bytes?: number; aborted?: boolean; error?: boolean; slow?: boolean } = {}) {
const minute = Math.floor(this.now() / 60000) * 60000;
const key = JSON.stringify([minute, metric, route, method]);
let entry = this.pending.get(key);
if (!entry) {
if (this.pending.size >= this.capacity) { this.dropped++; return; }
entry = { minute, data: { metric, route, method, count: 0, sumMs: 0, bytes: 0,
aborted: 0, errors: 0, slow: 0, buckets: Array(LATENCY_BUCKETS.length + 1).fill(0) } };
this.pending.set(key, entry);
}
const duration = Math.max(0, Number.isFinite(ms) ? ms : 0);
const data = entry.data;
data.count++; data.sumMs += Math.round(duration); data.bytes += detail.bytes || 0;
data.aborted += Number(!!detail.aborted); data.errors += Number(!!detail.error); data.slow += Number(!!detail.slow);
let bucket = LATENCY_BUCKETS.findIndex(bound => duration <= bound);
if (bucket < 0) bucket = LATENCY_BUCKETS.length;
data.buckets[bucket]++;
}
drain() { const entries = [...this.pending.values()]; this.pending.clear(); return entries; }
}
export const requestMetrics = new RequestMetrics();
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 } : {};
}
export function recordCacheHit(cache: "source" | "transformed", hit: boolean) {
if (context.getStore()) requestMetrics.observe(`${cache}_${hit ? "hit" : "miss"}`, "all", "all", 0);
}
/** Capture the context now: worker admission can resume from another request's callback. */
export function startStage(stage: Stage) {
const started = performance.now();
const trace = context.getStore();
let ended = false;
return (error = false) => {
if (ended) return; ended = true;
if (!trace) return;
const ms = performance.now() - started;
trace.stages[stage] = (trace.stages[stage] || 0) + ms;
requestMetrics.observe(stage, "all", "all", ms, { error });
};
}
export async function measureStage<T>(stage: Stage, work: () => Promise<T>): Promise<T> {
const done = startStage(stage);
try { const value = await work(); done(); return value; }
catch (error) { done(true); throw error; }
}
export function monitorRequests(service: Service, metrics = requestMetrics,
log: Pick<typeof logger, "info" | "warn"> = logger): RequestHandler {
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 route = requestRoute(req.originalUrl, service);
const method = methods.has(req.method) ? req.method : "OTHER";
const started = performance.now();
let firstByteMs: number | undefined;
let bytes = 0;
let ended = false;
activeRequests++;
res.setHeader("X-Request-ID", id);
const writeHead = res.writeHead;
res.writeHead = function (this: typeof res, ...args: unknown[]): typeof res {
firstByteMs ??= performance.now() - started;
return Reflect.apply(writeHead, this, args);
};
const count = (chunk: unknown, encoding: unknown) => {
if (req.method === "HEAD") return;
if (typeof chunk === "string") bytes += Buffer.byteLength(chunk, typeof encoding === "string" ? encoding as BufferEncoding : undefined);
else if (chunk instanceof Uint8Array) bytes += chunk.byteLength;
};
// Preserve callbacks, return values and receiver. Installed before compression,
// these count encoded body bytes without changing stream backpressure.
const write = res.write;
res.write = function (this: typeof res, ...args: unknown[]): boolean {
count(args[0], args[1]);
return Reflect.apply(write, this, args);
};
const end = res.end;
res.end = function (this: typeof res, ...args: unknown[]): typeof res {
count(args[0], args[1]);
return Reflect.apply(end, this, args);
};
const done = () => {
if (ended) return; ended = true;
activeRequests--;
res.removeListener("finish", done); res.removeListener("close", done);
const ms = performance.now() - started;
const aborted = !res.writableFinished;
const transfer = /\/(file|zip)(?:\/|$)/.test(route) || route.startsWith("/w/") || service === "streamer";
const slow = ms > (transfer ? 10000 : 1000);
metrics.observe("request", route, method, ms, { bytes, aborted, error: res.statusCode >= 500, slow });
if (firstByteMs != null) metrics.observe("first_byte", route, method, firstByteMs);
const detail = { requestId: id, service, method, route, status: res.statusCode,
outcome: aborted ? "interrupted" : "completed", ms: Math.round(ms), firstByteMs: firstByteMs == null ? null : Math.round(firstByteMs), bytes, stages: trace.stages };
if (slow || aborted) log.warn(aborted ? "request interrupted" : "slow request", detail);
else if (!route.includes("healthcheck")) log.info("request", detail);
};
res.once("finish", done); res.once("close", done);
context.run(trace, next);
};
}
+24
View File
@@ -0,0 +1,24 @@
import { Readable, Writable } from "stream";
/** Cancel upstream on every response terminal event, including a pre-header close. */
export function streamResponse(source: Readable, response: Writable,
onError: (error: Error) => void): Promise<void> {
return new Promise((resolve, reject) => {
const cleanup = () => {
response.removeListener("close", done);
response.removeListener("finish", done);
response.removeListener("error", failed);
source.removeListener("error", upstreamFailed);
source.unpipe(response);
source.destroy();
};
const done = () => { cleanup(); resolve(); };
const failed = (error: Error) => { cleanup(); reject(error); };
const upstreamFailed = (error: Error) => { onError(error); done(); };
response.once("close", done);
response.once("finish", done);
response.once("error", failed);
source.once("error", upstreamFailed);
if (response.destroyed || response.writableEnded) done(); else source.pipe(response);
});
}
+2
View File
@@ -14,6 +14,8 @@ export interface GitHubBaseData {
organization: string;
repoName: string;
commit: string;
cacheGeneration?: string;
cacheRevision?: string;
}
export default abstract class GitHubBase implements SourceBase {
+2 -3
View File
@@ -1,6 +1,5 @@
import got from "got";
import { Readable } from "stream";
import { OctokitResponse } from "@octokit/types";
import storage from "../storage";
import GitHubBase, {
@@ -18,7 +17,7 @@ export default class GitHubDownload extends GitHubBase {
super(data);
}
public async getZipUrl(): Promise<OctokitResponse<unknown, 302>> {
public async getZipUrl() {
const oct = octokit(await this.data.getToken());
return oct.rest.repos.downloadZipballArchive({
owner: this.data.organization,
@@ -29,7 +28,7 @@ export default class GitHubDownload extends GitHubBase {
}
async download(progress?: (status: string) => void) {
let response: OctokitResponse<unknown, number>;
let response: Awaited<ReturnType<GitHubDownload["getZipUrl"]>>;
try {
response = await this.getZipUrl();
} catch (error) {
+101 -99
View File
@@ -1,3 +1,6 @@
import { contentGenerationPrefix, contentRetirementMarker } from "../content-generation";
import { createHash } from "crypto";
import { coordinatedFill } from "../cache-coordination";
import { githubTokenContext } from "../github-token-context";
import AnonymizedFile from "../AnonymizedFile";
import GitHubBase, {
@@ -11,17 +14,21 @@ import { basename, dirname } from "path";
import * as stream from "stream";
import AnonymousError from "../AnonymousError";
import { FILE_TYPE } from "../storage/Storage";
import { octokit, waitForTokenGate } from "../GitHubUtils";
import FileModel from "../model/files/files.model";
import { IFile } from "../model/files/files.types";
import { createLogger, serializeError } from "../logger";
import { measureStage, recordCacheHit } from "../request-monitoring";
import config from "../../config";
import { isConnected } from "../../server/database";
import AnonymizedRepositoryModel from "../model/anonymizedRepositories/anonymizedRepositories.model";
import { RepositoryStatus } from "../types";
const logger = createLogger("gh-stream");
const GH_API_CONCURRENCY = 6;
const cacheFills = new Map<string, Promise<void>>();
async function pMap<T, R>(
items: T[],
@@ -238,108 +245,94 @@ export default class GitHubStream extends GitHubBase {
const meta = await fileMeta();
const expected: { sha: string; size?: number } =
typeof meta === "string" ? { sha: meta } : meta;
const fileInfo = await storage.exists(repoId, filePath);
if (fileInfo == FILE_TYPE.FILE) {
// If we know the upstream size, validate the cached entry. A cached
// file smaller than the upstream size means a previous fetch was
// truncated — likely a network error during the GitHub fetch left a
// 0-byte or partial blob behind. Treat it as a miss and re-fetch.
// Cached size >= expected is accepted: equal for normal files, and
// larger for Git LFS files where FileModel.size is the pointer's
// size but the cached bytes are the resolved LFS content.
if (expected.size != null && expected.size > 0) {
try {
const stat = await storage.fileInfo(repoId, filePath);
if (stat.size != null && stat.size < expected.size) {
await storage.rm(repoId, filePath);
} else {
return storage.read(repoId, filePath);
}
} catch {
// fall through and re-fetch
const generation = createHash("sha256").update(JSON.stringify([
this.data.commit, expected.sha, this.data.cacheGeneration, this.data.cacheRevision,
])).digest("hex");
const cachePath = this.data.cacheGeneration ? `${contentGenerationPrefix(this.data.cacheGeneration)}/${generation}/${filePath}` : filePath;
const assertActive = async () => {
// Retirement markers live inside the storage root and disappear when
// expiration removes it. The persisted lifecycle also fences late fills.
if (isConnected) {
const current = await AnonymizedRepositoryModel.findOne({ repoId })
.select("status treeGeneration anonymizeDate contentCacheRevision").lean().exec();
if (!current || !current.status || [RepositoryStatus.EXPIRING, RepositoryStatus.EXPIRED,
RepositoryStatus.REMOVING, RepositoryStatus.REMOVED, RepositoryStatus.ARCHIVED].includes(current.status)
|| (this.data.cacheGeneration && `${current.treeGeneration || "legacy"}:${current.anonymizeDate?.toISOString() || ""}` !== this.data.cacheGeneration)
|| this.data.cacheRevision !== current.contentCacheRevision) {
throw new AnonymousError("repository_changed", { httpStatus: 409 });
}
} else {
return storage.read(repoId, filePath);
}
} else if (fileInfo == FILE_TYPE.FOLDER) {
throw new AnonymousError("folder_not_supported", {
httpStatus: 400,
object: filePath,
});
}
// GitHub's blob API rejects blobs larger than 100 MB with HTTP 422.
// Skip the download entirely when the tree already tells us the file is
// over the cap, so we surface a clean `file_too_big` instead of paying
// the round-trip just to translate a 422.
if (!this.data.cacheGeneration) return;
try {
await storage.fileInfo(repoId, contentRetirementMarker(contentGenerationPrefix(this.data.cacheGeneration)));
} catch (error) {
const missing = error as { code?: string; name?: string; $metadata?: { httpStatusCode?: number }; httpStatus?: number };
if (missing.code === "ENOENT" || missing.name === "NotFound" || missing.name === "NoSuchKey" ||
missing.$metadata?.httpStatusCode === 404 || missing.httpStatus === 404) return;
throw error;
}
throw new AnonymousError("repository_changed", { httpStatus: 409 });
};
const key = `${repoId}:${generation}:${filePath}`;
const cached = async () => {
try {
const info = await storage.fileInfo(repoId, cachePath);
if (info.contentType === "application/x-directory") {
throw new AnonymousError("folder_not_supported", { httpStatus: 400 });
}
if (expected.size != null && expected.size > 0 && (info.size == null || info.size < expected.size)) return undefined;
return true;
} catch (error) {
if (error instanceof AnonymousError && error.message === "folder_not_supported") throw error;
return undefined;
}
};
const hit = await measureStage("cache_lookup", cached);
recordCacheHit("source", !!hit);
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,
});
throw new AnonymousError("file_too_big", { httpStatus: 413, object: filePath });
}
const token = await this.data.getToken();
// Try the blob API first, but fall back to the raw URL on recoverable
// blob misses/caps while still preserving LFS pointer handling.
const content = await this.downloadWithFallback(
token,
expected.sha,
filePath
);
// duplicate the stream to write it to the storage
const stream1 = content.pipe(new stream.PassThrough());
const stream2 = content.pipe(new stream.PassThrough());
// Safety net: guarantee an `error` listener exists on both branches
// before any error can be emitted. storage.write attaches its listener
// only after an `await mk(...)`, and the route handler attaches its
// listener after awaiting this function — both leave a window where
// an upstream error would have no listener and escalate to
// uncaughtException, crashing the streamer.
const noop = () => {};
stream1.on("error", noop);
stream2.on("error", noop);
content.on("error", (error) => {
const httpStatus =
(error as { response?: { statusCode?: number } })?.response
?.statusCode ??
(error as { status?: number })?.status ??
(error as { httpStatus?: number })?.httpStatus;
const errCode = (error as { code?: string })?.code;
const isTransient =
!httpStatus &&
(errCode === "ECONNRESET" ||
errCode === "ETIMEDOUT" ||
errCode === "ERR_BODY_PARSE_FAILURE" ||
error.name === "ReadError");
const code =
httpStatus === 422
? "file_too_big"
: httpStatus === 403
? "file_not_accessible"
: isTransient
? "upstream_error"
: "file_not_found";
const wrapped = new AnonymousError(code, {
httpStatus: isTransient ? 502 : httpStatus,
cause: error as Error,
object: filePath,
});
stream1.destroy(wrapped);
stream2.destroy(wrapped);
});
// Fire-and-forget: storage.write logs its own failures inside FileSystem
// (`[fs] write failed`). Swallow the rejection here so an upstream error
// (e.g. GitHub 422 on a too-big blob) doesn't surface as an unhandled
// promise rejection and crash the streamer process.
storage
.write(repoId, filePath, stream1, this.type, expected.size)
.catch(() => {});
return stream2;
let filling = cacheFills.get(key);
if (!filling) {
if (cacheFills.size >= 64) throw new AnonymousError("cache_busy", { httpStatus: 503 });
filling = coordinatedFill(key, cached, async () => {
await assertActive();
const token = await this.data.getToken();
return measureStage("upstream", async () => {
const content = await this.downloadWithFallback(token, expected.sha, filePath);
let count = 0;
const limited = new stream.Transform({ transform(chunk, encoding, callback) {
count += chunk.length;
callback(count > config.MAX_FILE_SIZE ? new AnonymousError("file_too_big", { httpStatus: 413 }) : null, chunk);
} });
content.on("error", error => {
if (error instanceof AnonymousError) { limited.destroy(error); return; }
const status = (error as { response?: { statusCode?: number }; status?: number; httpStatus?: number });
const httpStatus = status.response?.statusCode ?? status.status ?? status.httpStatus;
const code = httpStatus === 403 ? "file_not_accessible"
: httpStatus === 404 ? "file_not_found"
: httpStatus === 422 ? "file_too_big" : "upstream_error";
limited.destroy(new AnonymousError(code, { httpStatus: httpStatus || 502, cause: error, object: filePath }));
});
limited.once("close", () => content.destroy());
limited.on("error", () => {});
content.pipe(limited);
const deadline = setTimeout(() => limited.destroy(new AnonymousError("upstream_error", { httpStatus: 502 })), 120_000);
try {
await storage.write(repoId, cachePath, limited, this.type, expected.size);
try { await assertActive(); }
catch (error) { await storage.rm(repoId, cachePath); throw error; }
}
finally { clearTimeout(deadline); limited.destroy(); content.destroy(); }
return true;
});
}).then(() => {});
cacheFills.set(key, filling);
void filling.finally(() => { if (cacheFills.get(key) === filling) cacheFills.delete(key); }).catch(() => {});
}
await measureStage("cache_fill", () => filling!);
return storage.read(repoId, cachePath);
}
async getFileContent(file: AnonymizedFile): Promise<stream.Readable> {
@@ -444,6 +437,15 @@ export default class GitHubStream extends GitHubBase {
const output: IFile[] = [];
let data;
try {
data = await this.getGHTree(oct, token, sha, count, {
recursive: true,
callback: () => {
if (progress) {
progress("List file: " + count.file);
}
},
});
if (!data.truncated) return this.tree2Tree(data.tree, parentPath);
data = await this.getGHTree(oct, token, sha, count, {
recursive: false,
callback: () => {
+47 -21
View File
@@ -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";
@@ -83,6 +84,7 @@ export default class FileSystem extends StorageBase {
const tmpPath = `${fullPath}.tmp.${process.pid}.${Date.now()}.${Math.random()
.toString(36)
.slice(2, 8)}`;
let sourceFailed = false;
try {
if (typeof data === "string") {
await fs.promises.writeFile(tmpPath, data);
@@ -100,7 +102,10 @@ export default class FileSystem extends StorageBase {
resolve();
}
};
data.on("error", finish);
data.on("error", (error) => {
if (!settled) sourceFailed = true;
finish(error);
});
ws.on("error", finish);
ws.on("finish", () => finish());
data.pipe(ws);
@@ -123,7 +128,8 @@ export default class FileSystem extends StorageBase {
}
await fs.promises.rename(tmpPath, fullPath);
} catch (err) {
logger.error("write failed", serializeError(err));
if (sourceFailed) logger.warn("source stream failed; cache write discarded", serializeError(err));
else logger.error("write failed", serializeError(err));
await fs.promises.rm(tmpPath, { force: true }).catch(() => undefined);
throw err;
}
@@ -139,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);
@@ -238,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;
}
}
+58 -38
View File
@@ -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;
}
}
+3
View File
@@ -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" ? "\\" : "/")
+47
View File
@@ -0,0 +1,47 @@
import { promises as fs } from "fs";
import { join } from "path";
import { tmpdir } from "os";
import { pruneTransformedCache } from "./transformed-cache";
import { createLogger } from "./logger";
const logger = createLogger("temporary-storage");
let nextCleanup = 0;
let cleanup: Promise<void> | undefined;
/** Jobs time out in minutes; an hour-old spool belongs to an interrupted process. */
export async function removeStaleTextSpools(directory = tmpdir(), now = Date.now()) {
const entries = await fs.opendir(directory);
for await (const entry of entries) {
if (!entry.isDirectory() || !entry.name.startsWith("anonymous-text-")) continue;
const path = join(directory, entry.name);
try {
if (now - (await fs.stat(path)).mtimeMs > 3600_000) await fs.rm(path, { recursive: true, force: true });
} catch { /* another process can finish cleanup first */ }
}
}
export function maintainTextSpools(force = false) {
if (cleanup) return cleanup;
if (!force && Date.now() < nextCleanup) return Promise.resolve();
nextCleanup = Date.now() + 60_000;
cleanup = removeStaleTextSpools().catch(() => {}).finally(() => { cleanup = undefined; });
return cleanup;
}
/** Recover on startup and continue while idle, without overlapping scans. */
export async function startTemporaryStorageMaintenance(intervalMs = 60_000) {
let running = false;
const run = async () => {
if (running) return;
running = true;
try {
const results = await Promise.allSettled([maintainTextSpools(true), pruneTransformedCache()]);
for (const result of results) if (result.status === "rejected") {
logger.warn("temporary storage cleanup deferred", { message: (result.reason as Error).message });
}
} finally { running = false; }
};
await run();
const timer = setInterval(() => { void run(); }, intervalMs);
timer.unref();
return () => clearInterval(timer);
}
+67
View File
@@ -0,0 +1,67 @@
/**
* Candidate index for nonempty, capture-free, fixed-width compiler patterns.
* False positives are safe; false negatives are not.
*/
export class TermPresenceIndex {
private groups: { indices: number[]; characters: string[] | null }[] = [];
constructor(private readonly patterns: RegExp[]) {
for (let index = 0; index < patterns.length; index++) {
const characters = firstCharacters(patterns[index].source);
const group = { indices: [index], characters };
// Merge groups whose first characters can overlap under Unicode folding.
// Then a combined search can identify one group without hiding another
// alternative at the same position (Alice versus AliceSmith, for example).
for (let i = this.groups.length - 1; i >= 0; i--) {
const other = this.groups[i];
if (overlap(group.characters, other.characters)) {
group.indices.push(...other.indices);
group.characters = group.characters && other.characters
? [...new Set([...group.characters, ...other.characters])] : null;
this.groups.splice(i, 1);
}
}
this.groups.push(group);
}
}
candidates(content: string, after = -1): Set<number> {
const found = new Set<number>();
const remaining = this.groups.map(group => group.indices.filter(index => index > after))
.filter(indices => indices.length);
let offset = 0;
while (remaining.length) {
// Fixed-width inputs contain no captures or quantifiers. Removing a
// discovered group avoids revisiting millions of identical CSV values.
const search = new RegExp(remaining.map(indices =>
`(${indices.map(index => this.patterns[index].source).join("|")})`).join("|"), "giu");
search.lastIndex = offset;
const match = search.exec(content);
if (!match) break;
const selected = match.slice(1).findIndex(value => value !== undefined);
for (const index of remaining[selected]) found.add(index);
remaining.splice(selected, 1);
// Another group's match may start inside this one. Search from the next
// code point, not from the end of the match, to retain nested candidates.
offset = match.index + (content.codePointAt(match.index)! > 0xffff ? 2 : 1);
}
return found;
}
}
function firstCharacters(source: string): string[] | null {
if (source.startsWith("[")) {
const characters = source.slice(1, source.indexOf("]"));
// Only the compiler's expanded letter classes have an enumerable prefix.
// Dot's [^\n], escapes and any future syntax conservatively join all groups.
return /^[\p{L}\p{N}]+$/u.test(characters) ? Array.from(characters) : null;
}
if (source.startsWith("\\")) return null;
return [Array.from(source)[0]];
}
function overlap(left: string[] | null, right: string[] | null): boolean {
if (!left || !right) return true;
const pattern = new RegExp(`^(?:${left.map(char => char.replace(/[.*+?^${}()|[\]\\]/g, "\\$&")).join("|")})$`, "iu");
return right.some(char => pattern.test(char));
}
+156
View File
@@ -0,0 +1,156 @@
import { promises as fs } from "fs";
import { join } from "path";
import { tmpdir } from "os";
import { createHash, randomUUID } from "crypto";
import config from "../config";
import { coordinatedFill } from "./cache-coordination";
import { isConnected } from "../server/database";
import RepositoryModel from "./model/anonymizedRepositories/anonymizedRepositories.model";
import { RepositoryStatus } from "./types";
import AnonymousError from "./AnonymousError";
import { recordCacheHit } from "./request-monitoring";
const root = join(tmpdir(), "anonymous-transformed-v1");
const TTL = 3600_000;
const DISK_BUDGET = 512 * 1024 * 1024;
const fills = new Map<string, Promise<boolean>>();
interface CacheScope { repoId: string; generation?: string; revision?: string }
async function scopeActive(scope?: CacheScope) {
if (!scope || !isConnected) return true;
const current = await RepositoryModel.findOne({ repoId: scope.repoId })
.select("status treeGeneration anonymizeDate contentCacheRevision").lean().exec();
return !!current?.status && ![RepositoryStatus.EXPIRING, RepositoryStatus.EXPIRED,
RepositoryStatus.REMOVING, RepositoryStatus.REMOVED, RepositoryStatus.ARCHIVED].includes(current.status)
&& (!scope.generation || scope.generation === `${current.treeGeneration || "legacy"}:${current.anonymizeDate?.toISOString() || ""}`)
&& scope.revision === current.contentCacheRevision;
}
async function removeEntry(metadata: string) {
await Promise.all([metadata, metadata.replace(/\.json$/, ".data")].map(path => fs.rm(path, { force: true })));
}
/** The key includes actual content, all rules, generation, and matcher version. */
export async function transformedFile(digest: string, options: unknown, output: string,
produce: () => Promise<boolean>): Promise<boolean> {
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 produceActive = async () => {
await assertActive();
const changed = await produce();
try { await assertActive(); }
catch (error) { await fs.rm(output, { force: true }); throw error; }
return changed;
};
await assertActive();
const key = createHash("sha256").update(JSON.stringify(["matcher-v1", digest, options,
config.ANONYMIZATION_MASK, config.APP_HOSTNAME])).digest("hex");
const data = join(root, `${key}.data`), metadata = join(root, `${key}.json`);
const read = async () => {
try {
const record = JSON.parse(await fs.readFile(metadata, "utf8"));
if (record.version !== 2 || Date.now() - record.created > TTL || (await fs.stat(data)).size !== record.size) return undefined;
return Boolean(record.changed);
} catch { return undefined; }
};
const copyCached = async (changed: boolean) => {
await fs.copyFile(data, output, fs.constants.COPYFILE_EXCL);
try { await assertActive(); } catch (error) { await fs.rm(output, { force: true }); throw error; }
return changed;
};
const hit = await read();
if (hit !== undefined) {
try { const copied = await copyCached(hit); recordCacheHit("transformed", true); return copied; } catch { /* eviction: retry the transformation */ }
}
recordCacheHit("transformed", false);
const pending = fills.get(key);
if (pending) {
await pending.catch(() => {});
const cached = await read();
if (cached !== undefined) { try { return await copyCached(cached); } catch { /* retry */ } }
}
if (fills.size >= 64) return produceActive();
let producedHere = false;
const task = coordinatedFill(`transformed:${key}`, read, async () => {
const changed = await produceActive();
producedHere = true;
const size = (await fs.stat(output)).size;
if (size > DISK_BUDGET / 4) return changed;
const token = randomUUID();
const tempData = `${data}.${token}.tmp`, tempMetadata = `${metadata}.${token}.tmp`;
try {
await fs.mkdir(root, { recursive: true, mode: 0o700 });
await fs.copyFile(output, tempData);
await fs.writeFile(tempMetadata, JSON.stringify({ version: 2, changed, size, created: Date.now(), scope }));
await fs.rename(tempData, data);
await fs.rename(tempMetadata, metadata);
await pruneTransformedCache(false);
} catch { /* A valid private output remains usable when cache storage fails. */
} finally {
await Promise.all([tempData, tempMetadata].map(path => fs.rm(path, { force: true }).catch(() => {})));
}
try { await assertActive(); }
catch (error) { await removeEntry(metadata); await fs.rm(output, { force: true }); throw error; }
return changed;
});
fills.set(key, task);
try {
const changed = await task;
if (producedHere) return changed;
try { return await copyCached(changed); } catch { return produceActive(); }
} finally { if (fills.get(key) === task) fills.delete(key); }
}
export async function pruneTransformedCache(validateLifecycle = true) {
const records = [];
let lifecycleError: Error | undefined;
let entries;
try { entries = await fs.opendir(root); }
catch (error) { if ((error as NodeJS.ErrnoException).code === "ENOENT") return; throw error; }
for await (const entry of entries) {
const name = entry.name;
if (name.endsWith(".tmp") || name.endsWith(".data")) {
const path = join(root, name);
try { if (Date.now() - (await fs.stat(path)).mtimeMs > TTL) await fs.rm(path, { force: true }); } catch { /* another publisher */ }
}
if (!name.endsWith(".json")) continue;
const path = join(root, name);
let record;
try { record = JSON.parse(await fs.readFile(path, "utf8")); }
catch (error) { if ((error as NodeJS.ErrnoException).code !== "ENOENT") await removeEntry(path); continue; }
if (record.version !== 2 || !Number.isFinite(record.created) || !Number.isFinite(record.size)
|| Date.now() - record.created > TTL) { await removeEntry(path); continue; }
if (validateLifecycle && !lifecycleError) {
try { if (!await scopeActive(record.scope)) { await removeEntry(path); continue; } }
catch (error) { lifecycleError = error as Error; }
}
records.push({ path, ...record });
}
records.sort((a, b) => b.created - a.created);
let bytes = 0;
for (let i = 0; i < records.length; i++) {
const record = records[i]; bytes += record.size;
if (i >= 128 || bytes > DISK_BUDGET || Date.now() - record.created > TTL) {
await removeEntry(record.path);
}
}
// Storage age and budget cleanup must finish even when MongoDB is unavailable.
if (lifecycleError) throw lifecycleError;
}
/** Lifecycle changes remove local entries; connected replicas also prune independently. */
export async function removeTransformedRepositoryCache(repoId: string) {
let entries;
try { entries = await fs.opendir(root); }
catch (error) { if ((error as NodeJS.ErrnoException).code === "ENOENT") return; throw error; }
for await (const entry of entries) {
if (!entry.name.endsWith(".json")) continue;
const metadata = join(root, entry.name);
let record;
try { record = JSON.parse(await fs.readFile(metadata, "utf8")); }
catch (error) { if ((error as NodeJS.ErrnoException).code !== "ENOENT") await removeEntry(metadata); continue; }
if (record.version !== 2 || record.scope?.repoId === repoId) await removeEntry(metadata);
}
}
+47 -5
View File
@@ -1,5 +1,5 @@
import got from "got";
import { Readable, Transform } from "stream";
import { Readable, Transform, Writable } from "stream";
import { Parse } from "unzip-stream";
import archiver = require("archiver");
@@ -12,6 +12,7 @@ import {
compileTerms,
} from "./anonymize-utils";
import { createLogger, serializeError } from "./logger";
import { startStage } from "./request-monitoring";
const logger = createLogger("zip-stream");
@@ -67,8 +68,8 @@ function isEntryAllowed(
* Stream the GitHub source zip for a repository, anonymize each entry on the
* fly, and pipe the resulting archive into the provided writable response.
*
* No data is written to local storage — the zip flows GitHub → unzip → per
* file anonymizer → archiver → response.
* Large text entries spool privately before worker anonymization. The parser
* admits the next entry only after the archive output drains to the response.
*/
export async function streamAnonymizedZip(
opt: StreamAnonymizedZipOptions,
@@ -111,20 +112,58 @@ export async function streamAnonymizedZip(
},
});
}
const upstreamDone = startStage("upstream");
const downloadStream = got.stream(response.url);
downloadStream.once("end", () => upstreamDone());
downloadStream.once("error", () => upstreamDone(true));
downloadStream.once("close", () => upstreamDone(!downloadStream.readableEnded));
const archive = archiver("zip", {});
const parser = Parse() as Transform;
const activeStreams = new Set<Readable>();
const compiledTerms = compileTerms(opt.anonymizerOptions.terms || []);
let stopped = false;
let entryWaiting = false;
let cleanupDrain: (() => void) | undefined;
const resumeParser = () => {
if (!stopped && entryWaiting && archive.readableLength === 0 && output.writableLength === 0) {
entryWaiting = false;
parser.resume();
}
};
// Middleware can omit write callbacks. Honor the writable backpressure
// contract instead so queued output stays bounded.
const output = new Writable({
write(chunk, encoding, callback) {
const written = () => {
cleanupDrain?.();
cleanupDrain = undefined;
callback();
setImmediate(resumeParser);
};
if (res.write(chunk, encoding)) written();
else {
// Compression forwards drain subscriptions to its underlying stream.
// Remove the listener from the emitter returned by on(), rather than
// accumulating wrappers on that stream through response.once().
const emitter = res.on("drain", written) as NodeJS.EventEmitter;
cleanupDrain = () => emitter.removeListener("drain", written);
}
},
final(callback) { res.end(); callback(); },
});
output.on("error", error => fail(error));
archive.on("data", () => { if (entryWaiting) setImmediate(resumeParser); });
const stop = () => {
if (stopped) return;
stopped = true;
cleanupDrain?.();
cleanupDrain = undefined;
downloadStream.destroy();
parser.destroy();
for (const stream of activeStreams) stream.destroy();
activeStreams.clear();
output.destroy();
archive.abort();
};
const fail = (error: Error) => {
@@ -145,7 +184,7 @@ export async function streamAnonymizedZip(
res.on("error", (error) => fail(error as Error));
res.on("close", stop);
archive.on("error", fail);
archive.pipe(res);
archive.pipe(output);
downloadStream
.on("error", fail)
@@ -172,6 +211,9 @@ export async function streamAnonymizedZip(
// there to decide whether the entry is text (and therefore should be
// anonymized) vs binary (passthrough). Assigning afterwards leaves
// isText=false for every file, so the zip ships unanonymized.
parser.pause();
const onArchived = () => { entryWaiting = true; setImmediate(resumeParser); };
archive.once("entry", onArchived);
const anonymizer = new AnonymizeTransformer({
...opt.anonymizerOptions,
filePath: entry.path,
@@ -189,7 +231,7 @@ export async function streamAnonymizedZip(
}
})
.on("error", fail)
.on("finish", () => {
.on("end", () => {
if (stopped) return;
void archive.finalize().catch(fail);
});