diff --git a/docs/performance-monitoring.md b/docs/performance-monitoring.md index f6336a4..7e828d2 100644 --- a/docs/performance-monitoring.md +++ b/docs/performance-monitoring.md @@ -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 diff --git a/public/i18n/locale-en.json b/public/i18n/locale-en.json index 938a4fd..8bb8010 100644 --- a/public/i18n/locale-en.json +++ b/public/i18n/locale-en.json @@ -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.", diff --git a/src/core/Repository.ts b/src/core/Repository.ts index e9c1a1f..8835774 100644 --- a/src/core/Repository.ts +++ b/src/core/Repository.ts @@ -1,4 +1,6 @@ import { allowance, reserveRepository } from "./billing/quotas"; +import { FILE_TREE_BATCH_SIZE, FILE_LISTING_LIMIT, deleteRowsInBatches } from "./file-tree"; +import { SourceBase, sourceStorageSize } from "./source/Source"; import { notifyOwnerAccessProblem, withOwnerAccessRecovery } from "./owner-notifications"; import AnonymizedPathModel from "./model/anonymized-path"; import storage from "./storage"; @@ -37,6 +39,24 @@ import AnonymizedRepositoryModel from "./model/anonymizedRepositories/anonymized import { createLogger, serializeError } from "./logger"; const logger = createLogger("repository"); +const snapshotValue = (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, + update: UpdateQuery +): 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>(); async function buildPathIndex(repoId: string, key: string, terms: string[], generation?: string) { const cacheKey = `${repoId}:${key}`; @@ -68,7 +88,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[], @@ -201,7 +221,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 { @@ -239,19 +264,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; - }); - if (this.model.billingManaged) await reserveRepository(String(this.model.owner), this.repoId, files.reduce((sum, file) => sum + (file.size || 0), 0), this.model.conference); - 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, storageBytes = 0; + let truncatedFolders: string[] = []; const generation = randomUUID(); const buildFilter = { _id: this.model._id, treeGeneration: previousGeneration || { $exists: false }, ...this.refreshFilter(), @@ -273,25 +288,50 @@ 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++; + storageBytes += file.size || 0; + 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 (this.model.billingManaged) await reserveRepository(String(this.model.owner), this.repoId, storageBytes, this.model.conference); 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 }); } } @@ -302,11 +342,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(); } @@ -329,7 +369,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; @@ -339,7 +379,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); @@ -357,13 +397,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 } }, @@ -386,7 +434,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 }); } } @@ -425,7 +473,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 @@ -747,10 +795,10 @@ export default class Repository { const candidateModel = new AnonymizedRepositoryModel(this.model.toObject()); candidateModel.source.commit = newCommit; const candidate = new Repository(candidateModel); - const files = await candidate.source.getFiles(); + const storageBytes = await sourceStorageSize(candidate.source); try { await reserveRepository(String(this.model.owner), this.repoId, - files.reduce((sum, file) => sum + (file.size || 0), 0), this.model.conference); + storageBytes, this.model.conference); } catch (error) { if (!opt?.force && error instanceof Error && error.message === "quota_exceeded") return; throw error; @@ -822,11 +870,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 @@ -992,11 +1036,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); @@ -1004,6 +1049,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 */ @@ -1011,12 +1069,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); } @@ -1031,15 +1092,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; @@ -1048,10 +1122,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[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 }); diff --git a/src/core/file-tree.ts b/src/core/file-tree.ts new file mode 100644 index 0000000..d9d8165 --- /dev/null +++ b/src/core/file-tree.ts @@ -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, + filter: mongo.Filter & { repoId: string }, + options: { maxBatches?: number; current?: () => Promise } = {} +): Promise { + 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; +} diff --git a/src/core/model/anonymizedRepositories/anonymizedRepositories.schema.ts b/src/core/model/anonymizedRepositories/anonymizedRepositories.schema.ts index 58be801..c552c00 100644 --- a/src/core/model/anonymizedRepositories/anonymizedRepositories.schema.ts +++ b/src/core/model/anonymizedRepositories/anonymizedRepositories.schema.ts @@ -61,6 +61,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, @@ -112,6 +113,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 }, diff --git a/src/core/model/anonymizedRepositories/anonymizedRepositories.types.ts b/src/core/model/anonymizedRepositories/anonymizedRepositories.types.ts index 28095f9..951eb23 100644 --- a/src/core/model/anonymizedRepositories/anonymizedRepositories.types.ts +++ b/src/core/model/anonymizedRepositories/anonymizedRepositories.types.ts @@ -36,6 +36,7 @@ export interface IAnonymizedRepository { treeGeneration?: string; stagedFileTrees?: { generation: string; until: Date }[]; retiredTreeGenerations?: string[]; + retiredPathIndexKeys?: string[]; retiredContentPrefixes?: string[]; legacyContentCleanupPending?: boolean; contentCacheVersion?: number; diff --git a/src/core/source/GitHubStream.ts b/src/core/source/GitHubStream.ts index 1817c13..b53f6d8 100644 --- a/src/core/source/GitHubStream.ts +++ b/src/core/source/GitHubStream.ts @@ -20,8 +20,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"; @@ -35,6 +34,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>(); const missingContent = new ExpiringMap(2048); @@ -583,8 +583,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 { 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); + } } /** @@ -688,7 +697,7 @@ export default class GitHubStream extends GitHubBase { } } - private async getTruncatedTree( + private async *iterateTreeBatches( sha: string, progress?: (status: string) => void, parentPath: string = "" @@ -699,21 +708,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), @@ -766,11 +777,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( @@ -790,14 +805,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, - }); + }; }); } } diff --git a/src/core/source/Source.ts b/src/core/source/Source.ts index f78a9ea..012d6c9 100644 --- a/src/core/source/Source.ts +++ b/src/core/source/Source.ts @@ -21,4 +21,15 @@ export interface SourceBase { * Get all the files from a specific source */ getFiles(progress?: (status: string) => void): Promise; + + /** Read large trees without retaining every entry. */ + iterateFiles?(progress?: (status: string) => void): AsyncIterable; +} + +/** Sum admission size without materializing a large source tree. */ +export async function sourceStorageSize(source: SourceBase): Promise { + let bytes = 0; + const files = source.iterateFiles ? source.iterateFiles() : await source.getFiles(); + for await (const file of files) bytes += file.size || 0; + return bytes; } diff --git a/src/core/source/github-submodules.ts b/src/core/source/github-submodules.ts index 16477da..55d136d 100644 --- a/src/core/source/github-submodules.ts +++ b/src/core/source/github-submodules.ts @@ -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 { +export async function expandSubmodules(files: IFile[], data: GitHubBaseData, + budget = { modules: 0, remaining: MAX_SUBMODULE_ENTRIES }): Promise { 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) { diff --git a/src/scripts/create-performance-indexes.ts b/src/scripts/create-performance-indexes.ts index 949e14b..d8e89c1 100644 --- a/src/scripts/create-performance-indexes.ts +++ b/src/scripts/create-performance-indexes.ts @@ -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 } }); diff --git a/src/server/routes/repository-private.ts b/src/server/routes/repository-private.ts index be40886..3d06d2f 100644 --- a/src/server/routes/repository-private.ts +++ b/src/server/routes/repository-private.ts @@ -1,3 +1,5 @@ +import Repository from "../../core/Repository"; +import { sourceStorageSize } from "../../core/source/Source"; import { neverExpirationDeadline, validateBillingExpiration } from "../../core/billing/expiration"; import config from "../../config"; import { assertConferenceAdmission, reserveRepository } from "../../core/billing/quotas"; @@ -535,19 +537,22 @@ router.post( const sourceChanged = reconnecting || hasRepositorySourceChanged(repo.model, repoUpdate); const previousAccessRevision = repo.model.githubAccess?.revision; + // Keep proposed settings separate until cleanup passes the persisted fences. + const proposed = new AnonymizedRepositoryModel(repo.model.toObject()); if (repo.model.billingManaged) validateBillingExpiration(repoUpdate.options); if (config.BILLING_ENABLED && repoUpdate.options.expirationMode === "never" && (repo.model.options.expirationMode !== "never" || repo.model.billingNeverExpiresUntil)) { - repo.model.billingNeverExpiresUntil = await neverExpirationDeadline(String(repo.model.owner), + proposed.billingNeverExpiresUntil = await neverExpirationDeadline(String(repo.model.owner), repo.model.options.expirationMode === "never" ? repo.model.billingNeverExpiresUntil : undefined); } else if (repoUpdate.options.expirationMode !== "never") { - repo.model.billingNeverExpiresUntil = undefined; + proposed.billingNeverExpiresUntil = undefined; } - updateRepoModel(repo.model, repoUpdate); - const reactivating = shouldReactivateInactiveRepository(repo.model); + updateRepoModel(proposed, repoUpdate); + const reactivating = shouldReactivateInactiveRepository(proposed); + if (reactivating) { - repo.model.anonymizeDate = new Date(); + proposed.anonymizeDate = new Date(); } const needsAdmission = sourceChanged || reactivating || repoUpdate.conference !== repo.model.conference; @@ -606,20 +611,22 @@ 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(); if (repo.model.billingManaged) { - const files = await repo.source.getFiles(); - await reserveRepository(String(repo.model.owner), repo.repoId, - files.reduce((sum, file) => sum + (file.size || 0), 0), repoUpdate.conference); + const storageBytes = await sourceStorageSize(new Repository(proposed).source); + await reserveRepository(String(repo.model.owner), repo.repoId, storageBytes, repoUpdate.conference); } await repo.remove({ accessRevision: previousAccessRevision, replacement: true }); } + repo.model.set({ options: proposed.options, source: proposed.source, + githubAccess: proposed.githubAccess, anonymizeDate: proposed.anonymizeDate, + billingNeverExpiresUntil: proposed.billingNeverExpiresUntil }); const removeRepoFromConference = async (conferenceID: string) => { const conf = await ConferenceModel.findOne({ diff --git a/src/server/routes/repository-public.ts b/src/server/routes/repository-public.ts index 3011e6e..dc74786 100644 --- a/src/server/routes/repository-public.ts +++ b/src/server/routes/repository-public.ts @@ -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"; @@ -218,21 +219,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 = {}; + const directCounts: Record = Object.create(null); for (const r of results) { directCounts[r._id ?? ""] = r.count; } - const folderCounts: Record = {}; + const folderCounts: Record = 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; } } @@ -243,7 +253,7 @@ router.get( "../../core/anonymize-utils" ); const compiled = compileTerms(terms); - const anonymized: Record = {}; + const anonymized: Record = Object.create(null); for (const [folder, count] of Object.entries(folderCounts)) { const anonFolder = anonymizePathCompiled(folder, compiled); anonymized[anonFolder] = (anonymized[anonFolder] || 0) + count; diff --git a/src/server/schedule.ts b/src/server/schedule.ts index ebbd3ce..af1f04d 100644 --- a/src/server/schedule.ts +++ b/src/server/schedule.ts @@ -158,7 +158,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(); diff --git a/test/fixtures/large-repository-tree.js b/test/fixtures/large-repository-tree.js new file mode 100644 index 0000000..b181367 --- /dev/null +++ b/test/fixtures/large-repository-tree.js @@ -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; }); diff --git a/test/large-repository-tree.test.js b/test/large-repository-tree.test.js new file mode 100644 index 0000000..8d6e379 --- /dev/null +++ b/test/large-repository-tree.test.js @@ -0,0 +1,479 @@ +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("sums billing admission sizes using the streaming interface", async () => { + const { sourceStorageSize } = require("../src/core/source/Source"); + const size = await sourceStorageSize({ + getFiles: () => { throw Error("unbounded admission read"); }, + async *iterateFiles() { for (let i = 0; i < 10001; i++) yield { size: 3 }; }, + }); + expect(size).to.equal(30003); + expect(await sourceStorageSize({ getFiles: async () => [{ size: 2 }, {}] })).to.equal(2); + }); + + 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("rejects billed streamed trees before publishing their generation", async () => { + const quotas = require("../src/core/billing/quotas"); + const reserve = quotas.reserveRepository; + const row = await repository({ treeGeneration: "old", billingManaged: true }); + const repo = new Repository(row); + await rows(1); + Object.defineProperty(repo, "source", { get: () => ({ + getFiles: () => { throw Error("unbounded source read"); }, + async *iterateFiles() { yield { name: "replacement", path: "", size: 100, sha: "b".repeat(40) }; }, + }) }); + quotas.reserveRepository = async (_owner, _repoId, bytes) => { + expect(bytes).to.equal(100); throw Error("quota_exceeded"); + }; + try { + const error = await repo.ensureFileTree({ force: true }).then(() => null, error => error); + expect(error?.message).to.equal("quota_exceeded"); + expect((await Repo.findById(row._id)).treeGeneration).to.equal("old"); + expect(await File.countDocuments({ treeGeneration: "old" })).to.equal(1); + } finally { quotas.reserveRepository = reserve; } + }); + + 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; } + }); +}); diff --git a/test/performance-integration.test.js b/test/performance-integration.test.js index 8fa69d8..b71ac06 100644 --- a/test/performance-integration.test.js +++ b/test/performance-integration.test.js @@ -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 }]]) { diff --git a/test/performance-regressions.test.js b/test/performance-regressions.test.js index 1c19db4..73198dd 100644 --- a/test/performance-regressions.test.js +++ b/test/performance-regressions.test.js @@ -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"]); }); diff --git a/test/production-regressions.test.js b/test/production-regressions.test.js index c58664c..d5b1c5c 100644 --- a/test/production-regressions.test.js +++ b/test/production-regressions.test.js @@ -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 () => {}; diff --git a/test/repository-activity.test.js b/test/repository-activity.test.js index 44c2be9..0555cac 100644 --- a/test/repository-activity.test.js +++ b/test/repository-activity.test.js @@ -93,7 +93,7 @@ describe("repository activity dates", () => { stub(quotas, "assertConferenceAdmission", async () => {}); stub(access, "selectRepositoryAccess", async () => ({ token: "test", binding: { kind: "oauth" } })); stub(github, "getRepositoryFromGitHub", async () => ({ model: { id: "gh_1" }, fullName: "owner/repo", getCommitInfo: async () => ({ commit: {} }) })); - stub(Stream.prototype, "getFiles", async function () { expect(this.data.commit).to.equal("abcdef456"); return [{ size: 500 }]; }); + stub(Stream.prototype, "iterateFiles", async function* () { expect(this.data.commit).to.equal("abcdef456"); yield { size: 500 }; }); stub(quotas, "reserveRepository", async (_owner, _repo, bytes) => { expect(bytes).to.equal(500); throw Error("quota_exceeded"); }); repo.remove = async () => { throw Error("must preserve snapshot"); }; let error; diff --git a/test/repository-refresh.test.js b/test/repository-refresh.test.js index db06f51..3e63db9 100644 --- a/test/repository-refresh.test.js +++ b/test/repository-refresh.test.js @@ -104,8 +104,8 @@ describe("repository refresh and restoration", () => { expect(bytes).to.equal(999); throw new Error("quota_exceeded"); }); - stub(Stream.prototype, "getFiles", async function () { - expect(this.data.commit).to.equal("new-sha"); return [{ size: 999 }]; + stub(Stream.prototype, "iterateFiles", async function* () { + expect(this.data.commit).to.equal("new-sha"); yield { size: 999 }; }); repo.getToken = async () => "test-token"; stub(github, "getRepositoryFromGitHub", async () => ({ fullName: "owner/repo", model: {},