Merge pull request #918 from tdurieux/fix/large-repository-heap-cleanup

fix: bound large repository tree memory and cleanup
This commit is contained in:
Thomas Durieux authored and GitHub committed 2026-11-03 09:32:21 +01:00
commit 99a2bff1ba
18 files changed
+748 -92

No files matched your search

+26
View File
@@ -126,6 +126,32 @@ checking the repository's source snapshot. Application rate limits remain in
force; 429 replies now include `rate_limited`, `resetAt`, and `Retry-After` so
clients can wait for the actual store reset.
## Large repository metadata
GitHub tree preparation consumes pages of plain metadata and persists at most
500 entries per insert. Preparation does not read the finished tree back into
memory. Traversal rejects more than 100,000 queued folders with
`repository_tree_too_large` and HTTP 413. Folder listings and folder-count responses reject results above 10,000
entries with `file_listing_too_large` and HTTP 413. Narrower folder reads and
the bounded filename search remain available. This limit also applies to
recursive internal listing calls.
A preparation reset switches tree generation and retires the old file rows and
active path-index key atomically. It does not wait for metadata deletion before
enqueueing the replacement. Maintenance discovers these retirement markers and
retries interrupted deletion. Activation performs at most ten batches per
retired metadata collection before deferring the remainder to maintenance.
Each metadata delete selects at most 500 IDs, rechecks its filter during the
write, and yields between batches. Expiration and removal also check lifecycle
ownership between batches. `file-tree` logs report deleted rows and elapsed
time as `metadata cleanup complete` or `metadata cleanup deferred`.
Run `npm run migrate:performance` when upgrading to add the sparse
`retiredPathIndexKeys` maintenance index without dropping existing indexes.
The constrained-heap regression fixture can be run without network or MongoDB:
`node --max-old-space-size=256 test/fixtures/large-repository-tree.js`.
## Shared blob cache
See [Shared blob cache](shared-cache.md) for configuration, storage and quota
+2
View File
@@ -76,6 +76,8 @@
"file_not_found": "The requested file is not found.",
"file_not_accessible": "The requested file is not accessible.",
"file_not_supported": "The file type is not supported. Anonymous GitHub cannot handle it.",
"repository_tree_too_large": "The repository has too many pending folders to process safely. Share a smaller repository.",
"file_listing_too_large": "This folder has too many entries to list. Open a more specific folder or use file search.",
"file_too_big": "The file size exceeds the limit of Anonymous GitHub.",
"is_folder": "The path points to a folder.",
"folder_not_supported": "The path points to a folder. Please select a file.",
+131 -51
View File
@@ -1,3 +1,5 @@
import { FILE_TREE_BATCH_SIZE, FILE_LISTING_LIMIT, deleteRowsInBatches } from "./file-tree";
import { SourceBase } from "./source/Source";
import { notifyOwnerAccessProblem, withOwnerAccessRecovery } from "./owner-notifications";
import AnonymizedPathModel from "./model/anonymized-path";
import storage from "./storage";
@@ -36,6 +38,24 @@ import AnonymizedRepositoryModel from "./model/anonymizedRepositories/anonymized
import { createLogger, serializeError } from "./logger";
const logger = createLogger("repository");
const snapshotValue = <T>(value: T | undefined) => value === undefined ? { $exists: false } : value;
/** Capture and retire the stored key while clearing it, including concurrent index publication. */
async function updateRetiringPathIndex(
filter: () => FilterQuery<IAnonymizedRepositoryDocument>,
update: UpdateQuery<IAnonymizedRepositoryDocument>
): Promise<{ matchedCount: number }> {
for (let attempt = 0; attempt < 5; attempt++) {
const current = await AnonymizedRepositoryModel.findOne(filter()).select("pathIndexKey").lean().exec();
if (!current) return { matchedCount: 0 };
const result = await AnonymizedRepositoryModel.updateOne({ ...filter(),
pathIndexKey: current.pathIndexKey === undefined ? { $exists: false } : current.pathIndexKey,
}, { ...update, ...(current.pathIndexKey ? { $addToSet: {
...update.$addToSet, retiredPathIndexKeys: current.pathIndexKey,
} } : {}) }).exec();
if (result.matchedCount) return result;
}
return { matchedCount: 0 };
}
const pathIndexes = new Map<string, Promise<void>>();
async function buildPathIndex(repoId: string, key: string, terms: string[], generation?: string) {
const cacheKey = `${repoId}:${key}`;
@@ -67,7 +87,7 @@ async function buildPathIndex(repoId: string, key: string, terms: string[], gene
}
import { IFile, IFileDocument } from "./model/files/files.types";
import AnonymizedFile from "./AnonymizedFile";
import { FilterQuery } from "mongoose";
import { FilterQuery, UpdateQuery } from "mongoose";
function anonymizeTreeRecursive(
tree: IFile[],
terms: string[],
@@ -200,7 +220,12 @@ export default class Repository {
if (pathQuery !== undefined) {
query.path = pathQuery;
}
return await FileModel.find(query).exec();
const files = await FileModel.find(query).select("name path repoId size sha upstream submodule")
.limit(FILE_LISTING_LIMIT + 1).lean().exec();
if (files.length > FILE_LISTING_LIMIT) {
throw new AnonymousError("file_listing_too_large", { httpStatus: 413 });
}
return files;
}
async submoduleMounts(): Promise<SubmoduleMount[]> {
@@ -238,18 +263,9 @@ export default class Repository {
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 readFiles = () => source.getFiles(opt.progress);
const files = await (streamed ? withOwnerAccessRecovery(this.owner.id,
{ kind: "repository", id: String(this.model._id) }, readFiles,
{ "source.repositoryName": this.model.source.repositoryName, "source.commit": this.model.source.commit,
"githubAccess.revision": this.model.githubAccess?.revision ?? { $exists: false } }) : readFiles()).catch(error => {
void notifyOwnerAccessProblem(this.owner.id, error, { kind: "repository", id: String(this.model._id) });
throw error;
});
const sourceWithTruncation = source as unknown as { truncatedFolderList?: string[] };
const truncatedFolders = Array.isArray(sourceWithTruncation.truncatedFolderList)
? [...sourceWithTruncation.truncatedFolderList] : [];
const source: SourceBase & { truncatedFolderList?: string[] } = this.source;
let fileCount = 0;
let truncatedFolders: string[] = [];
const generation = randomUUID();
const buildFilter = { _id: this.model._id,
treeGeneration: previousGeneration || { $exists: false }, ...this.refreshFilter(),
@@ -271,25 +287,48 @@ export default class Repository {
}, 30_000);
heartbeat.unref();
}
files.forEach(f => { f.repoId = this.repoId; f.treeGeneration = generation; if (isConnected) f.treeStaged = true; });
const readFiles = async () => {
let batch: IFile[] = [];
const insertBatch = async () => {
if (isConnected && !await AnonymizedRepositoryModel.exists({ ...buildFilter,
stagedFileTrees: { $elemMatch: { generation, until: { $gt: new Date() } } },
}).exec()) throw new AnonymousError("repository_changed", { httpStatus: 409 });
await FileModel.insertMany(batch); batch = [];
};
const files = source.iterateFiles ? source.iterateFiles(opt.progress) : await source.getFiles(opt.progress);
for await (const file of files) {
const row = file instanceof FileModel ? file.toObject() : file;
batch.push({ ...row, repoId: this.repoId, treeGeneration: generation, ...(isConnected ? { treeStaged: true } : {}) });
fileCount++;
if (batch.length === FILE_TREE_BATCH_SIZE) await insertBatch();
}
if (batch.length) await insertBatch();
truncatedFolders = Array.isArray(source.truncatedFolderList) ? [...source.truncatedFolderList] : [];
};
try {
await FileModel.insertMany(files);
await (streamed ? withOwnerAccessRecovery(this.owner.id,
{ kind: "repository", id: String(this.model._id) }, readFiles,
{ "source.repositoryName": this.model.source.repositoryName, "source.commit": this.model.source.commit,
"githubAccess.revision": this.model.githubAccess?.revision ?? { $exists: false } }) : readFiles()).catch(error => {
void notifyOwnerAccessProblem(this.owner.id, error, { kind: "repository", id: String(this.model._id) });
throw error;
});
if (isConnected) {
const activated = await AnonymizedRepositoryModel.updateOne({ _id: this.model._id,
const activated = await updateRetiringPathIndex(() => ({ _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 },
}), { $set: { treeGeneration: generation, truncatedFolders, size: { storage: 0, file: 0 },
...(streamed && !this.model.contentCacheVersion ? { contentCacheVersion: 1, legacyContentCleanupPending: true } : {}),
...(!files.length ? { emptyTreeGeneration: generation } : {}),
...(!fileCount ? { emptyTreeGeneration: generation } : {}),
}, $unset: { pathIndexKey: "", pathIndexBuiltAt: "", sizeComputedAt: "",
...(files.length ? { emptyTreeGeneration: "" } : {}),
...(fileCount ? { emptyTreeGeneration: "" } : {}),
}, $addToSet: { retiredTreeGenerations: previousGeneration || "",
...(streamed ? { retiredContentPrefixes: previousPrefix } : {}),
}, $pull: { stagedFileTrees: { generation } } }).exec();
}, $pull: { stagedFileTrees: { generation } } });
if (!activated.matchedCount) {
await FileModel.deleteMany({ repoId: this.repoId, treeGeneration: generation }).exec();
await deleteRowsInBatches(FileModel.collection, { repoId: this.repoId, treeGeneration: generation });
throw new AnonymousError("repository_changed", { httpStatus: 409 });
}
}
@@ -300,11 +339,11 @@ export default class Repository {
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.emptyTreeGeneration = fileCount ? 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)));
await this.cleanupRetiredFileTrees(new Date(), 10).catch(error => logger.warn("retired tree cleanup deferred", serializeError(error)));
} else {
await FileModel.deleteMany({ repoId: this.repoId, treeGeneration: previousGeneration || { $exists: false } }).exec();
}
@@ -327,7 +366,7 @@ export default class Repository {
}
/** Activation records retirement atomically; maintenance retries interrupted deletion. */
async cleanupRetiredFileTrees(now = new Date()) {
async cleanupRetiredFileTrees(now = new Date(), maxBatches = Infinity) {
const stages = await AnonymizedRepositoryModel.findById(this.model._id).select("stagedFileTrees").lean().exec();
for (const stage of stages?.stagedFileTrees || []) {
if (stage.until > now) continue;
@@ -337,7 +376,7 @@ export default class Repository {
$addToSet: { retiredTreeGenerations: stage.generation } }).exec();
}
const current = await AnonymizedRepositoryModel.findById(this.model._id)
.select("treeGeneration retiredTreeGenerations retiredContentPrefixes legacyContentCleanupPending source.type status anonymizeDate").lean().exec();
.select("treeGeneration retiredTreeGenerations retiredContentPrefixes retiredPathIndexKeys pathIndexKey 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);
@@ -355,13 +394,21 @@ export default class Repository {
}
const retired = (current.retiredTreeGenerations || []).filter(generation => generation !== (current.treeGeneration || ""));
if (retired.length) {
await FileModel.deleteMany({ repoId: this.repoId, $or: [
const complete = await deleteRowsInBatches(FileModel.collection, { repoId: this.repoId, $or: [
{ treeGeneration: { $in: retired.filter(Boolean) } },
...(retired.includes("") ? [{ treeGeneration: { $exists: false } }] : []),
] }).exec();
await AnonymizedRepositoryModel.updateOne({ _id: this.model._id },
] }, { maxBatches });
if (complete) await AnonymizedRepositoryModel.updateOne({ _id: this.model._id },
{ $pull: { retiredTreeGenerations: { $in: retired } } }).exec();
}
const retiredKeys = (current.retiredPathIndexKeys || []).filter(key => key !== current.pathIndexKey);
if (retiredKeys.length && await deleteRowsInBatches(AnonymizedPathModel.collection,
{ repoId: this.repoId, key: { $in: retiredKeys } }, { maxBatches })) {
await AnonymizedRepositoryModel.updateOne({ _id: this.model._id },
{ $pull: { retiredPathIndexKeys: { $in: retiredKeys } } }).exec();
}
await AnonymizedRepositoryModel.updateOne({ _id: this.model._id, "retiredPathIndexKeys.0": { $exists: false } },
{ $unset: { retiredPathIndexKeys: "" } }).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 } },
@@ -384,7 +431,7 @@ export default class Repository {
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();
await deleteRowsInBatches(FileModel.collection, { repoId, treeGeneration: generation, treeStaged: true });
}
}
@@ -423,7 +470,7 @@ export default class Repository {
const current = await AnonymizedRepositoryModel.exists({ _id: this.model._id,
fileMetadataRevision: metadataRevision || { $exists: false },
}).exec();
if (!current) await AnonymizedPathModel.deleteMany({ repoId: this.repoId, key }).exec();
if (!current) await deleteRowsInBatches(AnonymizedPathModel.collection, { repoId: this.repoId, key });
throw error;
}
// TTL retires superseded mappings. An older builder must never delete
@@ -802,11 +849,7 @@ export default class Repository {
}
this.model.increment();
await this.updateStatus(RepositoryStatus.DOWNLOAD);
await this.files({
force: false,
progress,
recursive: false,
});
await this.ensureFileTree({ progress });
// Previously inserted a dummy {path:"", name:"", size:0} FileModel
// here for empty repos "to avoid errors" — but that record collides
// with the special-case in AnonymizedFile.getFileInfo for the empty
@@ -972,11 +1015,12 @@ export default class Repository {
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 current = async () => !!await AnonymizedRepositoryModel.exists(lease).exec();
if (!await deleteRowsInBatches(FileModel.collection, { repoId: this.repoId }, { current })) return;
if (!await deleteRowsInBatches(AnonymizedPathModel.collection, { repoId: this.repoId }, { current })) return;
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();
$unset: { cleanupToken: "", cleanupUntil: "", pathIndexKey: "", pathIndexBuiltAt: "", sizeComputedAt: "", emptyTreeGeneration: "", retiredTreeGenerations: "", retiredPathIndexKeys: "", 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);
@@ -984,6 +1028,19 @@ export default class Repository {
}
}
/** Fields that identify the cached source and its current lifecycle owner. */
private fileSnapshotFilter() {
return {
treeGeneration: snapshotValue(this.model.treeGeneration),
status: snapshotValue(this.model.status), statusDate: snapshotValue(this.model.statusDate),
anonymizeDate: snapshotValue(this.model.anonymizeDate),
fileMetadataRevision: snapshotValue(this.model.fileMetadataRevision),
"source.type": snapshotValue(this.model.source.type), "source.repositoryName": snapshotValue(this.model.source.repositoryName),
"source.commit": snapshotValue(this.model.source.commit), "source.branch": snapshotValue(this.model.source.branch),
"githubAccess.revision": snapshotValue(this.model.githubAccess?.revision),
};
}
/**
* Remove the repository
*/
@@ -991,12 +1048,15 @@ export default class Repository {
if (expected) {
// Claim the lifecycle before deleting files; migration rejects REMOVING.
this.assertNotArchived();
const statusDate = new Date();
const result = await AnonymizedRepositoryModel.updateOne({ _id: this.model._id,
status: this.model.status,
"githubAccess.revision": expected.accessRevision || { $exists: false },
}, { $set: { status: RepositoryStatus.REMOVING, statusDate: new Date() } });
...this.fileSnapshotFilter(), ...this.refreshFilter(),
settingsSavedAt: snapshotValue(this.model.settingsSavedAt),
"githubAccess.revision": snapshotValue(expected.accessRevision),
}, { $set: { status: RepositoryStatus.REMOVING, statusDate } });
if (!result.matchedCount) throw new AnonymousError("connection_changed", { httpStatus: 409 });
this.model.status = RepositoryStatus.REMOVING;
this.model.statusDate = statusDate;
} else {
await this.updateStatus(RepositoryStatus.REMOVING);
}
@@ -1010,15 +1070,28 @@ export default class Repository {
async resetSate(status?: RepositoryStatus, statusMessage?: string) {
this.assertNotArchived();
const revision = randomUUID();
this.model.emptyTreeGeneration = undefined;
const defer = isConnected && (status || this.status) === RepositoryStatus.PREPARING;
const previousGeneration = this.model.treeGeneration;
const generation = defer ? randomUUID() : previousGeneration;
const refreshToken = this.refreshToken;
// Options-only saves apply at read time and leave this source snapshot valid.
// They must not cancel a queued worker without scheduling replacement work.
const snapshot = this.fileSnapshotFilter();
if (isConnected) {
const result = await AnonymizedRepositoryModel.updateOne({ _id: this.model._id,
treeGeneration: this.model.treeGeneration || { $exists: false },
}, { $set: { fileMetadataRevision: revision },
const result = await updateRetiringPathIndex(() => ({ _id: this.model._id,
...snapshot,
...(refreshToken ? { refreshToken, refreshUntil: { $gt: new Date() } } : {}),
}), { $set: { fileMetadataRevision: revision, ...(defer ? { treeGeneration: generation } : {}) },
...(defer ? { $addToSet: { retiredTreeGenerations: previousGeneration || "",
} } : {}),
$unset: { emptyTreeGeneration: "", pathIndexKey: "", pathIndexBuiltAt: "" },
}).exec();
if (!result.matchedCount) throw new AnonymousError("repository_changed", { httpStatus: 409 });
});
this.checkRefreshWrite(result);
if (!result.matchedCount) throw new AnonymousError(this.protectLifecycle ? "repository_job_cancelled" : "repository_changed",
{ httpStatus: this.protectLifecycle ? 410 : 409 });
}
this.model.emptyTreeGeneration = undefined;
this.model.treeGeneration = generation;
this.model.fileMetadataRevision = revision;
this.model.pathIndexKey = undefined;
this.model.pathIndexBuiltAt = undefined;
@@ -1027,10 +1100,17 @@ export default class Repository {
if (status) {
await this.updateStatus(status, statusMessage);
}
// remove cache
const current = async () => !isConnected || !!await AnonymizedRepositoryModel.exists({ _id: this.model._id,
treeGeneration: generation || { $exists: false }, fileMetadataRevision: revision, status: this.model.status,
}).exec();
const clearRows = async (collection: Parameters<typeof deleteRowsInBatches>[0]) => {
if (!await deleteRowsInBatches(collection, { repoId: this.repoId }, { current })) {
throw new AnonymousError("repository_changed", { httpStatus: 409 });
}
};
// A preparation reset retires metadata durably; maintenance performs its deletion.
await Promise.all([
FileModel.deleteMany({ repoId: this.repoId }).exec(),
AnonymizedPathModel.deleteMany({ repoId: this.repoId }).exec(),
...(!defer ? [clearRows(FileModel.collection), clearRows(AnonymizedPathModel.collection)] : []),
this.removeCache(),
]);
logger.info("reset", { repoId: this._model.repoId });
+32
View File
@@ -0,0 +1,32 @@
import { mongo } from "mongoose";
import { setImmediate } from "timers/promises";
import { createLogger } from "./logger";
export const FILE_TREE_BATCH_SIZE = 500;
// A folder response must fit in memory alongside other API requests.
export const FILE_LISTING_LIMIT = 10000;
const logger = createLogger("file-tree");
/** Keep each Mongo write bounded, and let other work run between batches. */
export async function deleteRowsInBatches(
collection: Pick<mongo.Collection, "find" | "deleteMany">,
filter: mongo.Filter<mongo.Document> & { repoId: string },
options: { maxBatches?: number; current?: () => Promise<boolean> } = {}
): Promise<boolean> {
const started = Date.now();
let deleted = 0;
for (let batch = 0; batch < (options.maxBatches ?? Infinity); batch++) {
if (options.current && !await options.current()) return false;
const rows = await collection.find(filter, { projection: { _id: 1 }, limit: FILE_TREE_BATCH_SIZE }).toArray();
if (!rows.length) {
if (deleted) logger.info("metadata cleanup complete", { repoId: filter.repoId, deleted, ms: Date.now() - started });
return true;
}
if (options.current && !await options.current()) return false;
const result = await collection.deleteMany({ ...filter, _id: { $in: rows.map(row => row._id) } }, { hint: "_id_" });
deleted += result.deletedCount;
await setImmediate();
}
logger.info("metadata cleanup deferred", { repoId: filter.repoId, deleted, ms: Date.now() - started });
return false;
}
@@ -59,6 +59,7 @@ const AnonymizedRepositorySchema = new Schema({
treeGeneration: String,
stagedFileTrees: { type: [{ generation: String, until: Date }], default: undefined },
retiredTreeGenerations: { type: [String], default: undefined },
retiredPathIndexKeys: { type: [String], default: undefined },
retiredContentPrefixes: { type: [String], default: undefined },
legacyContentCleanupPending: Boolean,
contentCacheVersion: Number,
@@ -110,6 +111,7 @@ const AnonymizedRepositorySchema = new Schema({
AnonymizedRepositorySchema.index({ "source.repositoryName": 1 });
AnonymizedRepositorySchema.index({ "coauthors.githubId": 1 });
AnonymizedRepositorySchema.index({ retiredTreeGenerations: 1 }, { sparse: true });
AnonymizedRepositorySchema.index({ retiredPathIndexKeys: 1 }, { sparse: true });
AnonymizedRepositorySchema.index({ "stagedFileTrees.until": 1 }, { sparse: true });
AnonymizedRepositorySchema.index({ retiredContentPrefixes: 1 }, { sparse: true });
AnonymizedRepositorySchema.index({ legacyContentCleanupPending: 1, status: 1 },
@@ -34,6 +34,7 @@ export interface IAnonymizedRepository {
treeGeneration?: string;
stagedFileTrees?: { generation: string; until: Date }[];
retiredTreeGenerations?: string[];
retiredPathIndexKeys?: string[];
retiredContentPrefixes?: string[];
legacyContentCleanupPending?: boolean;
contentCacheVersion?: number;
+28 -13
View File
@@ -19,8 +19,7 @@ import { basename, dirname } from "path";
import * as stream from "stream";
import AnonymousError from "../AnonymousError";
import { octokit, waitForTokenGate, RateLimitDelayError } from "../GitHubUtils";
import FileModel from "../model/files/files.model";
import { expandSubmodules, validUpstream } from "./github-submodules";
import { expandSubmodules, validUpstream, MAX_SUBMODULE_ENTRIES } from "./github-submodules";
import { IFile, UpstreamFile } from "../model/files/files.types";
import { createLogger, serializeError } from "../logger";
import { measureStage, recordCacheHit } from "../request-monitoring";
@@ -34,6 +33,7 @@ import { cachedContentMiss, classifyContentMiss } from "./github-content-miss";
const logger = createLogger("gh-stream");
const GH_API_CONCURRENCY = 6;
const MAX_PENDING_TREE_FOLDERS = 100000;
const cacheFills = new Map<string, Promise<void>>();
const missingContent = new ExpiringMap<string>(2048);
@@ -558,8 +558,17 @@ export default class GitHubStream extends GitHubBase {
}
async getFiles(progress?: (status: string) => void) {
const files: IFile[] = [];
for await (const file of this.iterateFiles(progress)) files.push(file);
return files;
}
async *iterateFiles(progress?: (status: string) => void): AsyncGenerator<IFile> {
this._truncatedFolders = [];
return expandSubmodules(await this.getTruncatedTree(this.data.commit, progress), this.data);
const budget = { modules: 0, remaining: MAX_SUBMODULE_ENTRIES };
for await (const batch of this.iterateTreeBatches(this.data.commit, progress)) {
yield* await expandSubmodules(batch, this.data, budget);
}
}
/**
@@ -663,7 +672,7 @@ export default class GitHubStream extends GitHubBase {
}
}
private async getTruncatedTree(
private async *iterateTreeBatches(
sha: string,
progress?: (status: string) => void,
parentPath: string = ""
@@ -674,21 +683,23 @@ export default class GitHubStream extends GitHubBase {
request: 0,
file: 0,
};
const output: IFile[] = [];
let data = await this.getGHTree(oct, token, sha, count, {
recursive: true,
callback: () => { progress?.("List file: " + count.file); },
});
if (!data.truncated) return this.tree2Tree(data.tree, parentPath);
if (!data.truncated) { yield this.tree2Tree(data.tree, parentPath); return; }
data = await this.getGHTree(oct, token, sha, count, {
recursive: false,
callback: () => { progress?.("List file: " + count.file); },
});
if (data.truncated) this._truncatedFolders.push(parentPath);
output.push(...this.tree2Tree(data.tree, parentPath));
yield this.tree2Tree(data.tree, parentPath);
const subtrees: { sha: string; parentPath: string }[] = [];
for (const file of data.tree) {
if (file.type == "tree" && file.path && file.sha) {
if (subtrees.length >= MAX_PENDING_TREE_FOLDERS) {
throw new AnonymousError("repository_tree_too_large", { httpStatus: 413 });
}
subtrees.push({
sha: file.sha,
parentPath: path.join(parentPath, file.path),
@@ -741,11 +752,15 @@ export default class GitHubStream extends GitHubBase {
GH_API_CONCURRENCY
);
for (const result of batchResults) {
output.push(...result.files);
queue.push(...result.children);
yield result.files;
for (const child of result.children) {
if (queue.length >= MAX_PENDING_TREE_FOLDERS) {
throw new AnonymousError("repository_tree_too_large", { httpStatus: 413 });
}
queue.push(child);
}
}
}
return output;
}
private tree2Tree(
@@ -765,14 +780,14 @@ export default class GitHubStream extends GitHubBase {
if (pathFile === ".") {
pathFile = "";
}
return new FileModel({
return {
name: basename(fullPath),
path: pathFile,
repoId: this.data.repoId,
size: elem.size,
...(elem.type === "commit" ? { submodule: "unsupported" } : {}),
...(elem.type === "commit" ? { submodule: "unsupported" as const } : {}),
sha: elem.sha,
});
};
});
}
}
+3
View File
@@ -21,4 +21,7 @@ export interface SourceBase {
* Get all the files from a specific source
*/
getFiles(progress?: (status: string) => void): Promise<IFile[]>;
/** Read large trees without retaining every entry. */
iterateFiles?(progress?: (status: string) => void): AsyncIterable<IFile>;
}
+6 -6
View File
@@ -36,12 +36,12 @@ export function submoduleRepository(url: unknown, parent: GitHubBaseData): { org
}
/** Resolve only direct public GitHub dependencies. Never send the parent token to a child. */
export async function expandSubmodules(files: IFile[], data: GitHubBaseData): Promise<IFile[]> {
export async function expandSubmodules(files: IFile[], data: GitHubBaseData,
budget = { modules: 0, remaining: MAX_SUBMODULE_ENTRIES }): Promise<IFile[]> {
const output = files.filter(file => file.name !== ".gitmodules");
const modules = output.filter(file => file.submodule);
let remaining = MAX_SUBMODULE_ENTRIES;
for (const [index, mount] of modules.entries()) {
if (index >= MAX_SUBMODULES) { mount.submodule = "limit"; continue; }
for (const mount of modules) {
if (budget.modules++ >= MAX_SUBMODULES) { mount.submodule = "limit"; continue; }
const mountPath = [mount.path, mount.name].filter(Boolean).join("/");
if (!safeRepositoryPath(mountPath) || !mount.sha || !/^[a-f0-9]{40}$/.test(mount.sha)) continue;
const token = await data.getToken();
@@ -59,10 +59,10 @@ export async function expandSubmodules(files: IFile[], data: GitHubBaseData): Pr
const child = await octokit("").git.getTree({
owner: repository.organization, repo: repository.repoName, tree_sha: mount.sha, recursive: "1",
});
if (child.data.truncated || child.data.tree.length > remaining) { mount.submodule = "limit"; continue; }
if (child.data.truncated || child.data.tree.length > budget.remaining) { mount.submodule = "limit"; continue; }
const entries = child.data.tree;
if (entries.some(entry => !safeRepositoryPath(entry.path))) { mount.submodule = "unsupported"; continue; }
remaining -= entries.length;
budget.remaining -= entries.length;
mount.submodule = "included";
mount.upstream = { ...repository, commit: mount.sha, path: "" };
for (const entry of entries) {
@@ -17,6 +17,7 @@ async function main() {
await File.collection.createIndex({ metadataPending: 1, repoId: 1, treeGeneration: 1 },
{ partialFilterExpression: { metadataPending: true } });
await Repository.collection.createIndex({ retiredTreeGenerations: 1 }, { sparse: true });
await Repository.collection.createIndex({ retiredPathIndexKeys: 1 }, { sparse: true });
await Repository.collection.createIndex({ "stagedFileTrees.until": 1 }, { sparse: true });
await File.collection.createIndex({ treeStaged: 1, repoId: 1, treeGeneration: 1 },
{ partialFilterExpression: { treeStaged: true } });
+13 -8
View File
@@ -529,11 +529,14 @@ router.post(
const sourceChanged = reconnecting || hasRepositorySourceChanged(repo.model, repoUpdate);
const previousAccessRevision = repo.model.githubAccess?.revision;
updateRepoModel(repo.model, repoUpdate);
const reactivating = shouldReactivateInactiveRepository(repo.model);
// Cleanup must use the persisted snapshot. Keep proposed settings separate
// until removal has finished under the original source and access fences.
const proposed = new AnonymizedRepositoryModel(repo.model.toObject());
updateRepoModel(proposed, repoUpdate);
const reactivating = shouldReactivateInactiveRepository(proposed);
if (reactivating) {
repo.model.anonymizeDate = new Date();
proposed.anonymizeDate = new Date();
}
if (sourceChanged) {
@@ -570,15 +573,17 @@ router.post(
const commit = await repository.getCommitInfo(repoUpdate.source.commit, {
accessToken: sourceAccess.token,
});
if (reconnecting) repo.model.source.commitDate = commit.commit.committer?.date
if (reconnecting) proposed.source.commitDate = commit.commit.committer?.date
? new Date(commit.commit.committer.date) : undefined;
repo.model.githubAccess = { ...sourceAccess.binding, revision: randomUUID() };
repo.model.source.repositoryId = repository.model.id;
repo.model.source.repositoryName =
proposed.githubAccess = { ...sourceAccess.binding, revision: randomUUID() };
proposed.source.repositoryId = repository.model.id;
proposed.source.repositoryName =
repository.fullName || repoUpdate.fullName;
repo.model.anonymizeDate = new Date();
proposed.anonymizeDate = new Date();
await repo.remove({ accessRevision: previousAccessRevision });
}
repo.model.set({ options: proposed.options, source: proposed.source,
githubAccess: proposed.githubAccess, anonymizeDate: proposed.anonymizeDate });
const removeRepoFromConference = async (conferenceID: string) => {
const conf = await ConferenceModel.findOne({
+15 -5
View File
@@ -1,3 +1,4 @@
import { FILE_LISTING_LIMIT } from "../../core/file-tree";
import { repositoryFailureStatus } from "../../core/repository-failure-status";
import { anonymizePath } from "../../core/anonymize-utils";
import * as express from "express";
@@ -217,21 +218,30 @@ router.get(
{ $match: { repoId, treeGeneration: repo.model.treeGeneration || { $exists: false }, size: { $ne: null } } },
{ $project: { _id: 0, path: 1 } },
{ $group: { _id: "$path", count: { $sum: 1 } } },
{ $limit: FILE_LISTING_LIMIT + 1 },
]).exec();
if (results.length > FILE_LISTING_LIMIT) throw new AnonymousError("file_listing_too_large", { httpStatus: 413 });
const directCounts: Record<string, number> = {};
const directCounts: Record<string, number> = Object.create(null);
for (const r of results) {
directCounts[r._id ?? ""] = r.count;
}
const folderCounts: Record<string, number> = {};
const folderCounts: Record<string, number> = Object.create(null);
let folders = 0;
const addCount = (folder: string, count: number) => {
if (!Object.hasOwn(folderCounts, folder) && ++folders > FILE_LISTING_LIMIT) {
throw new AnonymousError("file_listing_too_large", { httpStatus: 413 });
}
folderCounts[folder] = (folderCounts[folder] || 0) + count;
};
for (const [folder, count] of Object.entries(directCounts)) {
let p = folder;
folderCounts[p] = (folderCounts[p] || 0) + count;
addCount(p, count);
while (p) {
const idx = p.lastIndexOf("/");
p = idx >= 0 ? p.substring(0, idx) : "";
folderCounts[p] = (folderCounts[p] || 0) + count;
addCount(p, count);
if (!p) break;
}
}
@@ -242,7 +252,7 @@ router.get(
"../../core/anonymize-utils"
);
const compiled = compileTerms(terms);
const anonymized: Record<string, number> = {};
const anonymized: Record<string, number> = Object.create(null);
for (const [folder, count] of Object.entries(folderCounts)) {
const anonFolder = anonymizePathCompiled(folder, compiled);
anonymized[anonFolder] = (anonymized[anonFolder] || 0) + count;
+1 -1
View File
@@ -156,7 +156,7 @@ export async function runRepositoryStatusCheck(now = new Date()) {
await expireContent(AnonymizedPullRequestModel, data => new PullRequest(data), now);
const retiredCursor = AnonymizedRepositoryModel.find({ $or: [
{ retiredTreeGenerations: { $exists: true } }, { retiredContentPrefixes: { $exists: true } },
{ retiredTreeGenerations: { $exists: true } }, { retiredPathIndexKeys: { $exists: true } }, { retiredContentPrefixes: { $exists: true } },
{ legacyContentCleanupPending: true, status: RepositoryStatus.READY },
{ "stagedFileTrees.until": { $lte: now } },
] }).cursor();
+27
View File
@@ -0,0 +1,27 @@
// Synthetic metadata only. No network or database is used by this process.
const assert = require("node:assert/strict");
require("ts-node/register/transpile-only");
const GitHubStream = require("../../src/core/source/GitHubStream").default;
const source = new GitHubStream({ repoId: "heap-fixture", organization: "fixture", repoName: "large",
commit: "head", getToken: () => "fixture-token" });
const pageSize = 10000, pages = Number(process.env.TREE_FIXTURE_PAGES || 200);
source.getGHTree = async (_oct, _token, sha, _count, options) => {
if (sha === "head") return options.recursive
? { truncated: true, tree: [] }
: { truncated: false, tree: Array.from({ length: pages }, (_, i) => ({ path: `folder${i}`, type: "tree", sha: String(i) })) };
return { truncated: false, tree: Array.from({ length: pageSize }, (_, i) => ({
path: `file${i}.txt`, type: "blob", sha: "a".repeat(40), size: i,
})) };
};
(async () => {
let count = 0, bytes = 0;
const files = source.iterateFiles ? source.iterateFiles() : await source.getFiles();
for await (const file of files) {
assert.equal(file.repoId, "heap-fixture");
if (file.size !== undefined) { bytes += file.size; assert.ok(file.path.startsWith("folder")); }
count++;
}
assert.equal(count, pages * (pageSize + 1));
assert.equal(bytes, pages * pageSize * (pageSize - 1) / 2);
console.log(JSON.stringify({ count, bytes, peakRssMiB: process.resourceUsage().maxRSS / 1024 }));
})().catch(error => { console.error(error); process.exitCode = 1; });
+448
View File
@@ -0,0 +1,448 @@
const { expect } = require("chai");
const { execFile } = require("node:child_process");
const { promisify } = require("node:util");
const fs = require("node:fs/promises"), path = require("node:path"), os = require("node:os");
require("ts-node/register/transpile-only");
const mongoose = require("mongoose");
const db = require("../src/server/database");
const Repository = require("../src/core/Repository").default;
const Repo = require("../src/core/model/anonymizedRepositories/anonymizedRepositories.model").default;
const File = require("../src/core/model/files/files.model").default;
const Path = require("../src/core/model/anonymized-path").default;
const { FILE_TREE_BATCH_SIZE, FILE_LISTING_LIMIT, deleteRowsInBatches } = require("../src/core/file-tree");
const GitHubStream = require("../src/core/source/GitHubStream").default;
const { expandSubmodules, MAX_SUBMODULES } = require("../src/core/source/github-submodules");
const github = require("../src/core/GitHubUtils");
const config = require("../src/config").default;
describe("large repository tree allocation", function () {
this.timeout(120000);
it("reads two million entries within a 256 MiB V8 heap", async () => {
const { stdout } = await promisify(execFile)(process.execPath,
["--max-old-space-size=256", path.join(__dirname, "fixtures/large-repository-tree.js")],
{ timeout: 110000, maxBuffer: 262144, env: { ...process.env, NODE_ENV: "test", SESSION_SECRET: "large-tree-test" } });
expect(JSON.parse(stdout).count).to.equal(2000200);
});
it("does not fetch child trees before the consumer advances past the parent page", async () => {
const source = new GitHubStream({ repoId: "backpressure", organization: "fixture", repoName: "repo", commit: "head", getToken: () => "fixture" });
let calls = 0;
source.getGHTree = async (_oct, _token, _sha, _count, options) => {
calls++;
return options.recursive ? { truncated: true, tree: [] }
: { tree: [{ path: "folder", type: "tree", sha: "a".repeat(40) }] };
};
const iterator = source.iterateFiles();
expect((await iterator.next()).value.name).to.equal("folder");
expect(calls).to.equal(2);
await iterator.return();
expect(calls).to.equal(2);
});
it("rejects an oversized pending-folder queue before fetching its children", async () => {
const source = new GitHubStream({ repoId: "wide", organization: "fixture", repoName: "repo", commit: "head", getToken: () => "fixture" });
let calls = 0;
source.getGHTree = async (_oct, _token, _sha, _count, options) => {
calls++;
return options.recursive ? { truncated: true, tree: [] }
: { tree: Array.from({ length: 100001 }, (_, i) => ({ path: `folder${i}`, type: "tree", sha: String(i) })) };
};
const error = await (async () => { for await (const file of source.iterateFiles()) void file; })().then(() => null, error => error);
expect(error.message).to.equal("repository_tree_too_large"); expect(error.httpStatus).to.equal(413); expect(calls).to.equal(2);
});
it("applies the submodule budget across separate tree pages", async () => {
const octokit = github.octokit, gate = github.waitForTokenGate;
const budget = { modules: 0, remaining: 10000 }, sha = "a".repeat(40);
let calls = 0;
github.waitForTokenGate = async () => {};
github.octokit = () => ({ repos: { getContent: async () => {
calls++; return { data: { submodule_git_url: "https://github.com/child/public.git", sha } };
} }, git: { getTree: async () => ({ data: { tree: [], truncated: false } }) } });
const data = { repoId: "modules", organization: "fixture", repoName: "repo", commit: "head", getToken: () => "fixture" };
try {
for (let i = 0; i <= MAX_SUBMODULES; i++) {
const [mount] = await expandSubmodules([{ name: `child${i}`, path: "vendor", sha, submodule: "unsupported" }], data, budget);
expect(mount.submodule).to.equal(i < MAX_SUBMODULES ? "included" : "limit");
}
expect(calls).to.equal(MAX_SUBMODULES);
} finally { github.octokit = octokit; github.waitForTokenGate = gate; }
});
});
const uri = process.env.PERFORMANCE_MONGO_URI;
(uri ? describe : describe.skip)("large repository persistence and cleanup", function () {
this.timeout(60000);
let previousFolder, folder;
before(async () => {
if (!/\/perf_test_[A-Za-z0-9_-]+(?:\?|$)/.test(uri)) throw Error("An isolated perf_test_ database is required");
await mongoose.connect(uri);
db.isConnected = true;
previousFolder = config.FOLDER;
folder = await fs.mkdtemp(path.join(os.tmpdir(), "large-tree-tests-")); config.FOLDER = folder;
await Promise.all([Repo, File, Path].map(model => model.createIndexes()));
});
after(async () => {
db.isConnected = false;
await mongoose.connection.dropDatabase(); await mongoose.disconnect();
config.FOLDER = previousFolder; await fs.rm(folder, { recursive: true, force: true });
});
beforeEach(async () => { await Promise.all([Repo, File, Path].map(model => model.deleteMany({}))); });
const owner = new mongoose.Types.ObjectId();
const repository = (extra = {}) => Repo.create({ repoId: "large", owner, status: "preparing", statusDate: new Date(),
source: { type: "GitHubStream", repositoryName: "fixture/repo", commit: "a".repeat(40) }, ...extra });
async function rows(count, extra = {}) {
for (let offset = 0; offset < count; offset += FILE_TREE_BATCH_SIZE) {
await File.collection.insertMany(Array.from({ length: Math.min(count - offset, FILE_TREE_BATCH_SIZE) }, (_, i) => ({
repoId: "large", treeGeneration: "old", name: `file${offset + i}.txt`, path: "", size: 1, sha: "b".repeat(40),
...(typeof extra === "function" ? extra(offset + i) : extra),
})));
}
}
it("persists streamed input in bounded batches without reading a listing after preparation", async () => {
const row = await repository(), repo = new Repository(row), insert = File.insertMany;
let produced = 0, persisted = 0, calls = 0;
Object.defineProperty(repo, "source", { get: () => ({
getFiles: () => { throw Error("unbounded source read"); },
async *iterateFiles() {
for (let i = 0; i < 1503; i++) { produced++; yield { name: `file${i}.txt`, path: "", size: i, sha: "b".repeat(40) }; }
},
}) });
repo.files = () => { throw Error("preparation reloaded the tree"); };
File.insertMany = async function (batch, ...args) {
expect(batch.length).to.be.at.most(FILE_TREE_BATCH_SIZE);
expect(produced).to.equal(persisted + batch.length);
const model = await Repo.findById(row._id);
expect(model.stagedFileTrees.some(stage => stage.generation === batch[0].treeGeneration)).to.equal(true);
const result = await insert.call(this, batch, ...args); persisted += batch.length; calls++; return result;
};
try { await repo.anonymize(); } finally { File.insertMany = insert; }
expect(calls).to.equal(4); expect(persisted).to.equal(1503);
expect(await File.countDocuments({ repoId: row.repoId, treeGeneration: repo.model.treeGeneration })).to.equal(1503);
expect((await Repo.findById(row._id)).status).to.equal("ready");
expect(await File.countDocuments({ treeStaged: true })).to.equal(0);
});
it("retires preparation metadata without deleting it in the request, and resumes interrupted cleanup", async () => {
const row = await repository({ treeGeneration: "old", pathIndexKey: "old-key" }), repo = new Repository(row);
await rows(1201);
await File.create({ repoId: "unrelated", treeGeneration: "old", name: "keep.txt", path: "" });
await Path.collection.insertOne({ repoId: row.repoId, key: "old-key", anonymousPath: "old.txt" });
const fileDelete = File.collection.deleteMany, pathDelete = Path.collection.deleteMany;
File.collection.deleteMany = Path.collection.deleteMany = () => { throw Error("foreground metadata deletion"); };
try { await repo.resetSate("preparing"); }
finally { File.collection.deleteMany = fileDelete; Path.collection.deleteMany = pathDelete; }
let current = await Repo.findById(row._id);
expect(current.treeGeneration).not.to.equal("old"); expect(current.retiredTreeGenerations).to.deep.equal(["old"]);
expect(current.retiredPathIndexKeys).to.deep.equal(["old-key"]);
expect(await File.countDocuments({ repoId: row.repoId })).to.equal(1201);
await File.create({ repoId: row.repoId, treeGeneration: current.treeGeneration, name: "active.txt", path: "" });
await Path.collection.insertOne({ repoId: row.repoId, key: "active-key", anonymousPath: "active.txt" });
await Repo.updateOne({ _id: row._id }, { $set: { pathIndexKey: "active-key" } });
let batches = 0;
File.collection.deleteMany = async function (filter, ...args) {
expect(filter._id.$in.length).to.be.at.most(FILE_TREE_BATCH_SIZE);
if (++batches === 2) throw Error("interrupted cleanup");
return fileDelete.call(this, filter, ...args);
};
try { await repo.cleanupRetiredFileTrees().then(() => { throw Error("expected interruption"); }, error => expect(error.message).to.equal("interrupted cleanup")); }
finally { File.collection.deleteMany = fileDelete; }
expect((await Repo.findById(row._id)).retiredTreeGenerations).to.deep.equal(["old"]);
await repo.cleanupRetiredFileTrees();
expect((await File.find({ repoId: row.repoId })).map(file => file.name)).to.deep.equal(["active.txt"]);
expect((await Path.find({ repoId: row.repoId })).map(file => file.key)).to.deep.equal(["active-key"]);
expect(await File.countDocuments({ repoId: "unrelated" })).to.equal(1);
current = await Repo.findById(row._id);
expect(current.retiredTreeGenerations).to.equal(undefined); expect(current.retiredPathIndexKeys).to.equal(undefined);
});
it("stops ingestion between batches when another tree replaces its build", async () => {
const row = await repository({ treeGeneration: "old" }), repo = new Repository(row), insert = File.insertMany;
let produced = 0, calls = 0, closed = false;
Object.defineProperty(repo, "source", { get: () => ({ getFiles: () => { throw Error("unbounded read"); },
async *iterateFiles() {
try { for (let i = 0; i < 1500; i++) { produced++; yield { name: `file${i}.txt`, path: "", size: i }; } }
finally { closed = true; }
},
}) });
File.insertMany = async function (...args) {
const result = await insert.apply(this, args); calls++;
await Repo.updateOne({ _id: row._id }, { $set: { treeGeneration: "replacement" } });
await File.create({ repoId: row.repoId, treeGeneration: "replacement", name: "keep.txt", path: "" });
return result;
};
let error;
try { error = await repo.anonymize().then(() => null, error => error); } finally { File.insertMany = insert; }
expect(error.message).to.equal("repository_changed"); expect(calls).to.equal(1); expect(produced).to.equal(1000); expect(closed).to.equal(true);
expect(await File.countDocuments({ repoId: row.repoId, treeGeneration: "replacement" })).to.equal(1);
expect(await File.countDocuments({ repoId: row.repoId, treeStaged: true })).to.equal(500);
});
it("retires an index published between reading and clearing the persisted key", async () => {
const row = await repository({ status: "ready", treeGeneration: "old" }), resetter = new Repository(row);
await rows(2);
const searcher = new Repository(await Repo.findById(row._id)), updateOne = Repo.updateOne;
let attempts = 0, publishedKey;
Repo.updateOne = function (filter, update, ...args) {
const query = updateOne.call(this, filter, update, ...args);
if (update.$unset && Object.hasOwn(update.$unset, "pathIndexKey") && Object.hasOwn(filter, "pathIndexKey")) {
const exec = query.exec;
query.exec = async function (...execArgs) {
if (++attempts === 1) {
expect(await searcher.searchFiles("file")).to.have.length(2);
publishedKey = (await Repo.findById(row._id)).pathIndexKey;
}
return exec.apply(this, execArgs);
};
}
return query;
};
try { await resetter.resetSate("preparing"); } finally { Repo.updateOne = updateOne; }
expect(attempts).to.equal(2);
expect((await Repo.findById(row._id)).retiredPathIndexKeys).to.deep.equal([publishedKey]);
await resetter.cleanupRetiredFileTrees();
expect(await Path.countDocuments({ repoId: row.repoId })).to.equal(0);
});
it("retires a path index published after the activating worker loaded its model", async () => {
const row = await repository({ status: "ready", treeGeneration: "old" }), worker = new Repository(row);
await rows(2);
const searcher = new Repository(await Repo.findById(row._id));
Object.defineProperty(worker, "source", { get: () => ({ getFiles: async () => {
expect(await searcher.searchFiles("file")).to.have.length(2);
expect((await Repo.findById(row._id)).pathIndexKey).to.be.a("string");
return [{ name: "replacement.txt", path: "", size: 1, sha: "c".repeat(40) }];
} }) });
await worker.files({ force: true });
expect(await Path.countDocuments({ repoId: row.repoId })).to.equal(0);
expect((await File.find({ repoId: row.repoId })).map(file => file.name)).to.deep.equal(["replacement.txt"]);
});
const changedSnapshots = [
...["archived", "removing", "removed", "expiring", "expired", "ready"].map(status => [status, { status, statusDate: new Date(1) }]),
["status timestamp", { statusDate: new Date(1) }],
["source commit", { "source.commit": "c".repeat(40) }],
["source repository", { "source.repositoryName": "other/repo" }],
["source branch", { "source.branch": "replacement" }],
["source type", { "source.type": "Zip" }],
["anonymization snapshot", { anonymizeDate: new Date(1) }],
["metadata revision", { fileMetadataRevision: "replacement" }],
["access revision", { "githubAccess.revision": "replacement" }],
];
for (const [label, replacement] of changedSnapshots) it(`cancels preparation before retiring metadata after a changed ${label}`, async () => {
const row = await repository({ status: "download", treeGeneration: "old", emptyTreeGeneration: "old", pathIndexKey: "old-key", fileMetadataRevision: "before" });
const worker = new Repository(row); worker.protectLifecycle = true;
worker.removeCache = () => { throw Error("cancelled reset touched storage"); };
await rows(1); await Path.collection.insertOne({ repoId: row.repoId, key: "old-key", anonymousPath: "keep.txt" });
await Repo.updateOne({ _id: row._id }, { $set: replacement });
const before = await Repo.findById(row._id).lean();
const error = await worker.resetSate("preparing").then(() => null, error => error);
expect(error.message).to.equal("repository_job_cancelled");
const after = await Repo.findById(row._id).lean();
for (const field of ["status", "statusDate", "treeGeneration", "emptyTreeGeneration", "fileMetadataRevision", "pathIndexKey", "retiredTreeGenerations", "retiredPathIndexKeys"]) {
expect(after[field], field).to.deep.equal(before[field]);
}
expect(worker.model.treeGeneration).to.equal("old"); expect(worker.model.emptyTreeGeneration).to.equal("old");
await new Repository(await Repo.findById(row._id)).cleanupRetiredFileTrees();
expect(await File.countDocuments({ repoId: row.repoId, treeGeneration: "old" })).to.equal(1);
expect(await Path.countDocuments({ repoId: row.repoId, key: "old-key" })).to.equal(1);
});
for (const replaced of [false, true]) it(`rejects an ${replaced ? "replaced" : "expired"} refresh lease before retirement`, async () => {
const row = await repository({ status: "download", treeGeneration: "old" }), worker = new Repository(row);
worker.protectLifecycle = true; worker.refreshToken = "original-lease";
await Repo.updateOne({ _id: row._id }, { $set: { refreshToken: replaced ? "new-lease" : "original-lease",
refreshUntil: new Date(Date.now() + (replaced ? 60000 : -60000)) } });
const error = await worker.resetSate("preparing").then(() => null, error => error);
expect(error.message).to.equal("invalid_status");
const current = await Repo.findById(row._id);
expect(current.treeGeneration).to.equal("old"); expect(current.retiredTreeGenerations).to.equal(undefined);
});
it("rejects archival between reading the path-index key and committing retirement", async () => {
const row = await repository({ status: "download", treeGeneration: "old", pathIndexKey: "preserved" }), worker = new Repository(row);
worker.protectLifecycle = true;
const updateOne = Repo.updateOne; let attempted = false;
Repo.updateOne = function (filter, update, ...args) {
const query = updateOne.call(this, filter, update, ...args);
if (update.$unset && Object.hasOwn(update.$unset, "pathIndexKey")) {
const exec = query.exec;
query.exec = async function (...execArgs) {
attempted = true;
await updateOne.call(Repo, { _id: row._id }, { $set: { status: "archived" } }).exec();
return exec.apply(this, execArgs);
};
}
return query;
};
let error;
try { error = await worker.resetSate("preparing").then(() => null, error => error); } finally { Repo.updateOne = updateOne; }
expect(attempted).to.equal(true); expect(error.message).to.equal("repository_job_cancelled");
const current = await Repo.findById(row._id);
expect(current.status).to.equal("archived"); expect(current.treeGeneration).to.equal("old");
expect(current.pathIndexKey).to.equal("preserved"); expect(current.retiredTreeGenerations).to.equal(undefined);
});
it("can remove a repository using the timestamp from its guarded access claim", async () => {
const row = await repository({ status: "ready", treeGeneration: "old" }), repo = new Repository(row);
await rows(1);
await repo.remove({});
expect((await Repo.findById(row._id)).status).to.equal("removed");
expect(await File.countDocuments({ repoId: row.repoId })).to.equal(0);
});
for (const reconnecting of [false, true]) it(`saves a ${reconnecting ? "reconnected repository" : "changed source"} through real removal`, async () => {
const row = await repository({ status: "ready", treeGeneration: "old", pathIndexKey: "old-key",
githubAccess: { kind: "oauth", revision: "old-access" }, anonymizeDate: new Date(1),
options: { terms: ["before"], expirationMode: "never" } }), repo = new Repository(row);
await rows(2); await Path.collection.insertOne({ repoId: row.repoId, key: "old-key", anonymousPath: "old.txt" });
const utils = require("../src/server/routes/route-utils"), app = require("../src/core/github-app");
const source = require("../src/core/source/GitHubRepository"), queue = require("../src/queue");
const router = require("../src/server/routes/repository-private").default;
const handler = router.stack.find(layer => layer.route?.path === "/:repoId/" && layer.route.methods.post).route.stack[0].handle;
const restores = [], jobs = []; let response, cleanupSnapshot, removals = 0;
const stub = (object, key, value) => { const original = object[key]; restores.push(() => { object[key] = original; }); object[key] = value; };
stub(utils, "getRepo", async () => repo);
stub(utils, "getUser", async () => ({ id: repo.owner.id, model: { id: repo.owner.id } }));
stub(utils, "handleError", error => { throw error; });
stub(app, "selectRepositoryAccess", async () => ({ token: "fixture", binding: { kind: "oauth" } }));
stub(source, "getRepositoryFromGitHub", async () => ({ id: "gh_123", model: { id: "new-source" }, fullName: "fixture/repo",
branches: async () => [{ name: "replacement" }], getCommitInfo: async () => ({ commit: { committer: { date: "2026-11-01" } } }) }));
stub(queue, "downloadQueue", { add: async (...args) => { jobs.push(args); } });
const remove = repo.remove;
stub(repo, "remove", async function (...args) {
removals++;
cleanupSnapshot = { commit: this.model.source.commit, accessRevision: this.model.githubAccess.revision,
anonymizeDate: this.model.anonymizeDate };
return remove.apply(this, args);
});
try {
await handler({ body: { repoId: row.repoId, fullName: "fixture/repo", terms: ["after"],
source: { branch: "replacement", commit: "b".repeat(40) }, options: { expirationMode: "never" },
...(reconnecting ? { reconnectRepositoryId: "gh_123" } : {}) } }, { json: value => { response = value; } });
} finally { while (restores.length) restores.pop()(); }
expect(removals).to.equal(1); expect(response).to.deep.equal({ status: "preparing" });
expect(cleanupSnapshot).to.deep.equal({ commit: "a".repeat(40), accessRevision: "old-access", anonymizeDate: new Date(1) });
const saved = await Repo.findById(row._id);
expect(saved.status).to.equal("preparing"); expect(saved.source.commit).to.equal("b".repeat(40));
expect(saved.source.branch).to.equal("replacement"); expect(saved.source.repositoryId).to.equal("new-source");
expect(saved.githubAccess.revision).not.to.equal("old-access"); expect(saved.anonymizeDate.getTime()).to.be.greaterThan(1);
expect(saved.options.terms).to.deep.equal(["after"]);
expect(await File.countDocuments({ repoId: row.repoId })).to.equal(0); expect(await Path.countDocuments({ repoId: row.repoId })).to.equal(0);
expect(jobs).to.have.length(1); expect(jobs[0][1]).to.deep.equal({ repoId: row.repoId });
});
for (const [label, change] of [...changedSnapshots, ["settings revision", { settingsSavedAt: new Date(1) }]] ) {
it(`rejects removal before claiming the lifecycle after a changed ${label}`, async () => {
const row = await repository({ status: "ready", treeGeneration: "old", pathIndexKey: "old-key" }), stale = new Repository(row);
await rows(1); await Repo.updateOne({ _id: row._id }, { $set: change });
const before = await Repo.findById(row._id).lean();
stale.removeCache = () => { throw Error("stale removal touched storage"); };
const error = await stale.remove({}).then(() => null, error => error);
expect(error.message).to.equal("connection_changed");
expect(await Repo.findById(row._id).lean()).to.deep.equal(before);
expect(await File.countDocuments({ repoId: row.repoId })).to.equal(1);
});
}
it("finishes preparation after an accepted options-only save with the same source", async () => {
const row = await repository({ status: "preparing", treeGeneration: "old", anonymizeDate: new Date(1),
source: { type: "GitHubStream", repositoryName: "fixture/repo", branch: "main", commit: "a".repeat(40) },
options: { terms: ["before"], expirationMode: "never" } });
const worker = new Repository(row), editor = new Repository(await Repo.findById(row._id));
await rows(1);
Object.defineProperty(worker, "source", { get: () => ({ async *iterateFiles() {
yield { name: "replacement.txt", path: "", size: 2, sha: "b".repeat(40) };
} }) });
const utils = require("../src/server/routes/route-utils"), queue = require("../src/queue");
const router = require("../src/server/routes/repository-private").default;
const handler = router.stack.find(layer => layer.route?.path === "/:repoId/" && layer.route.methods.post).route.stack[0].handle;
const restores = [], jobs = []; let savedSettingsAt;
const stub = (object, key, value) => { const original = object[key]; restores.push(() => { object[key] = original; }); object[key] = value; };
stub(utils, "getRepo", async () => editor);
stub(utils, "getUser", async () => ({ id: editor.owner.id, model: { id: editor.owner.id } }));
stub(utils, "handleError", error => { throw error; });
stub(queue, "downloadQueue", { add: async (...args) => { jobs.push(args); } });
stub(db, "connect", async () => {});
stub(db, "getRepository", async () => {
// The worker snapshot was loaded before this accepted owner edit.
await handler({ body: { repoId: row.repoId, fullName: "fixture/repo", terms: ["after"],
source: { branch: "main", commit: "a".repeat(40) }, options: { expirationMode: "never" } } }, { json() {} });
savedSettingsAt = (await Repo.findById(row._id)).settingsSavedAt;
return worker;
});
stub(github, "getToken", async () => "fixture"); stub(github, "getRedisGateResetAt", async () => 0);
try {
await require("../src/queue/processes/downloadRepository").default({ data: { repoId: row.repoId }, updateProgress: async () => {} });
} finally { while (restores.length) restores.pop()(); }
const saved = await Repo.findById(row._id);
expect(saved.status).to.equal("ready"); expect(saved.options.terms).to.deep.equal(["after"]);
expect(saved.settingsSavedAt).to.deep.equal(savedSettingsAt); expect(savedSettingsAt).to.be.instanceOf(Date);
expect(jobs).to.have.length(0);
expect((await File.find({ repoId: row.repoId })).map(file => file.name)).to.deep.equal(["replacement.txt"]);
});
it("retains retirement markers when foreground cleanup reaches its batch budget", async () => {
const row = await repository({ treeGeneration: "active", retiredTreeGenerations: ["old"] }), repo = new Repository(row);
await rows(501);
await repo.cleanupRetiredFileTrees(new Date(), 1);
expect(await File.countDocuments({ repoId: row.repoId })).to.equal(1);
expect((await Repo.findById(row._id)).retiredTreeGenerations).to.deep.equal(["old"]);
await repo.cleanupRetiredFileTrees();
expect(await File.countDocuments({ repoId: row.repoId })).to.equal(0);
});
it("stops expiration deletion when its lifecycle claim changes between batches", async () => {
const row = await repository({ status: "expiring", treeGeneration: "old" }), repo = new Repository(row);
await rows(1001);
const original = File.collection.deleteMany; let calls = 0;
File.collection.deleteMany = async function (...args) {
const result = await original.apply(this, args); calls++;
await Repo.updateOne({ _id: row._id }, { $set: { status: "ready", treeGeneration: "replacement", fileMetadataRevision: "replacement" } });
await File.create({ repoId: row.repoId, treeGeneration: "replacement", name: "keep.txt", path: "" });
return result;
};
try { await repo.expire(); } finally { File.collection.deleteMany = original; }
expect(calls).to.equal(1); expect((await Repo.findById(row._id)).status).to.equal("ready");
expect(await File.countDocuments({ repoId: row.repoId, treeGeneration: "replacement" })).to.equal(1);
expect(await File.countDocuments({ repoId: row.repoId, treeGeneration: "old" })).to.equal(501);
});
it("rejects oversized listings before hydration while allowing narrower folder reads", async () => {
const row = await repository({ status: "ready", treeGeneration: "old" }), repo = new Repository(row);
await rows(FILE_LISTING_LIMIT + 1);
await File.create({ repoId: row.repoId, treeGeneration: "old", name: "small.txt", path: "specific", size: 2, sha: "c".repeat(40) });
const error = await repo.files({ recursive: false }).then(() => null, error => error);
expect(error.message).to.equal("file_listing_too_large"); expect(error.httpStatus).to.equal(413);
const files = await repo.files({ recursive: false, path: "specific" });
expect(files).to.have.length(1); expect(files[0].name).to.equal("small.txt");
expect(files[0]).not.to.be.instanceOf(mongoose.Document);
});
it("preserves rows that stop matching the deletion filter after selection", async () => {
await rows(2);
let checks = 0;
const complete = await deleteRowsInBatches(File.collection, { repoId: "large", treeGeneration: "old" }, {
current: async () => {
if (++checks === 2) await File.updateMany({ repoId: "large" }, { $set: { treeGeneration: "replacement" } });
return true;
},
});
expect(complete).to.equal(true); expect(await File.countDocuments({ repoId: "large" })).to.equal(2);
});
for (const ancestors of [false, true]) it(`bounds ${ancestors ? "ancestor expansion" : "direct folders"} in folder-count responses`, async () => {
const row = await repository({ status: "ready", treeGeneration: "old" }), repo = new Repository(row);
await rows(ancestors ? 3000 : FILE_LISTING_LIMIT + 1, i => ({ path: ancestors ? `folder${i}/a/b/c` : `folder${i}` }));
const utils = require("../src/server/routes/route-utils"), getRepo = utils.getRepo, handleError = utils.handleError;
const router = require("../src/server/routes/repository-public").default;
const handler = router.stack.find(layer => layer.route?.path === "/:repoId/files/counts").route.stack.at(-1).handle;
utils.getRepo = async () => repo; utils.handleError = error => { throw error; };
try {
const error = await handler({ params: { repoId: row.repoId } }, { header() {}, json() { throw Error("unbounded counts returned"); } })
.then(() => null, error => error);
expect(error.message).to.equal("file_listing_too_large"); expect(error.httpStatus).to.equal(413);
} finally { utils.getRepo = getRepo; utils.handleError = handleError; }
});
});
+8 -5
View File
@@ -128,6 +128,7 @@ const uri = process.env.PERFORMANCE_MONGO_URI;
await Gist.collection.createIndex({ statusMessage: 1 }, { name: "keep_existing_index" });
for (const model of [Repo, Gist, PR]) await model.collection.dropIndex("status_1_options.expirationDate_1");
await Repo.collection.dropIndex("retiredTreeGenerations_1");
await Repo.collection.dropIndex("retiredPathIndexKeys_1");
await Repo.collection.dropIndex("stagedFileTrees.until_1");
await File.collection.dropIndex("treeStaged_1_repoId_1_treeGeneration_1");
await File.collection.dropIndex("metadataPending_1_repoId_1_treeGeneration_1");
@@ -143,6 +144,7 @@ const uri = process.env.PERFORMANCE_MONGO_URI;
}
expect((await Gist.collection.indexes()).map(index => index.name)).to.include("keep_existing_index");
expect((await Repo.collection.indexes()).find(index => index.name === "retiredTreeGenerations_1").sparse).to.equal(true);
expect((await Repo.collection.indexes()).find(index => index.name === "retiredPathIndexKeys_1").sparse).to.equal(true);
expect((await Repo.collection.indexes()).find(index => index.name === "stagedFileTrees.until_1").sparse).to.equal(true);
expect((await File.collection.indexes()).find(index => index.name === "treeStaged_1_repoId_1_treeGeneration_1").partialFilterExpression).to.deep.equal({ treeStaged: true });
expect((await File.collection.indexes()).find(index => index.name === "metadataPending_1_repoId_1_treeGeneration_1").partialFilterExpression).to.deep.equal({ metadataPending: true });
@@ -1013,10 +1015,10 @@ const uri = process.env.PERFORMANCE_MONGO_URI;
treeGeneration: previousGeneration, source: { type: "GitHubStream" }, options: { expirationMode: "never" } });
await File.create({ repoId: model.repoId, treeGeneration: previousGeneration, path: "private", name: "old.txt", size: 1 });
await File.create({ repoId: "unrelated", treeGeneration: previousGeneration, path: "", name: "keep.txt", size: 1 });
const repo = new Repository(model), deleteMany = File.deleteMany;
const repo = new Repository(model), deleteMany = File.collection.deleteMany;
Object.defineProperty(repo, "source", { get: () => ({ getFiles: async () => [{ path: "", name: "active.txt", size: 2 }] }) });
File.deleteMany = function (filter, ...args) {
if (filter.repoId === model.repoId && filter.$or) return { exec: async () => { throw Error("cleanup unavailable"); } };
File.collection.deleteMany = async function (filter, ...args) {
if (filter.repoId === model.repoId && filter.$or) throw Error("cleanup unavailable");
return deleteMany.call(this, filter, ...args);
};
try {
@@ -1027,7 +1029,7 @@ const uri = process.env.PERFORMANCE_MONGO_URI;
await require("../src/server/schedule").runRepositoryStatusCheck();
expect((await Repo.findById(model._id)).retiredTreeGenerations).to.have.length(1);
expect(await File.countDocuments({ repoId: model.repoId })).to.equal(2);
} finally { File.deleteMany = deleteMany; }
} finally { File.collection.deleteMany = deleteMany; }
await require("../src/server/schedule").runRepositoryStatusCheck();
expect((await Repo.findById(model._id)).retiredTreeGenerations).to.equal(undefined);
expect((await File.find({ repoId: model.repoId })).map(file => file.name)).to.deep.equal(["active.txt"]);
@@ -1039,13 +1041,14 @@ const uri = process.env.PERFORMANCE_MONGO_URI;
await Repo.insertMany(Array.from({ length: 2000 }, (_, i) => ({ repoId: `clean-${i}` })));
await File.insertMany(Array.from({ length: 2000 }, (_, i) => ({ repoId: `clean-${i}`, path: "", name: "file.txt" })));
await Repo.create({ repoId: "retired", treeGeneration: "active", retiredTreeGenerations: ["old"] });
await Repo.create({ repoId: "retired-paths", retiredPathIndexKeys: ["old-key"] });
await Repo.create({ repoId: "retired-content", retiredContentPrefixes: ["__content/old"] });
await Repo.create({ repoId: "legacy-content", legacyContentCleanupPending: true, status: "ready" });
await Repo.create({ repoId: "staged", stagedFileTrees: [{ generation: "abandoned", until: new Date(0) }] });
await File.create({ repoId: "pending", path: "", name: "pending.txt", metadataPending: true });
await File.create({ repoId: "staged", treeGeneration: "abandoned", treeStaged: true, path: "private", name: "staged.txt" });
for (const [model, query] of [[Repo, { $or: [
{ retiredTreeGenerations: { $exists: true } }, { retiredContentPrefixes: { $exists: true } },
{ retiredTreeGenerations: { $exists: true } }, { retiredPathIndexKeys: { $exists: true } }, { retiredContentPrefixes: { $exists: true } },
{ legacyContentCleanupPending: true, status: "ready" },
{ "stagedFileTrees.until": { $lte: new Date() } },
] }], [File, { metadataPending: true }], [File, { treeStaged: true }]]) {
+2 -2
View File
@@ -117,7 +117,7 @@ describe("performance regressions", function () {
calls.push({ sha, recursive: options.recursive });
return { truncated: false, tree: Array.from({ length: 20 }, (_, i) => ({ path: `folder${i}/file.ts`, type: "blob", sha: "abc" })) };
};
const files = await source.getTruncatedTree("head");
const files = await source.getFiles();
expect(files).to.have.length(20); expect(calls).to.deep.equal([{ sha: "head", recursive: true }]);
});
@@ -157,7 +157,7 @@ describe("performance regressions", function () {
if (sha === "head") return { truncated: false, tree: [{ path: "src", type: "tree", sha: "src" }] };
return { truncated: false, tree: [{ path: "file.ts", type: "blob", sha: "abc" }] };
};
const files = await source.getTruncatedTree("head");
const files = await source.getFiles();
expect(files.map(file => `${file.path}/${file.name}`)).to.deep.equal(["/src", "src/file.ts"]);
expect(calls).to.deep.equal(["head:true", "head:false", "src:true"]);
});
+2 -1
View File
@@ -206,7 +206,8 @@ describe("production regressions", function () {
it("retains truncation warnings from the source instance that fetched the tree", async function () {
const FileModel = require("../src/core/model/files/files.model").default;
stub(db, "isConnected", false);
for (const method of ["exists", "deleteMany", "find"]) stub(FileModel, method, () => ({ exec: async () => method === "find" ? [] : null }));
for (const method of ["exists", "deleteMany"]) stub(FileModel, method, () => ({ exec: async () => null }));
stub(FileModel, "find", () => ({ select() { return this; }, limit() { return this; }, lean() { return this; }, exec: async () => [] }));
stub(FileModel, "insertMany", async () => []);
const repo = new Repository(new RepoModel({ repoId: "repo", options: {} }));
repo.computeSize = async () => {};