From 35c3b1804b968578f6ab446d197f38b6fa947493 Mon Sep 17 00:00:00 2001 From: Thomas Durieux <5577568+tdurieux@users.noreply.github.com> Date: Thu, 20 Aug 2026 00:52:19 -1000 Subject: [PATCH] Fix repository cleanup after removal and expiration (#782) --- src/core/AnonymizedFile.ts | 10 +- src/core/Repository.ts | 14 +++ src/queue/index.ts | 107 ++++++++++++++++- src/queue/processes/removeRepository.ts | 30 +++-- src/server/index.ts | 13 +- src/server/routes/repository-private.ts | 11 +- src/server/schedule.ts | 153 +++++++++++++++++------- test/backend-reliability.test.js | 101 ++++++++++++++++ 8 files changed, 370 insertions(+), 69 deletions(-) diff --git a/src/core/AnonymizedFile.ts b/src/core/AnonymizedFile.ts index b0a8239..45d7b46 100644 --- a/src/core/AnonymizedFile.ts +++ b/src/core/AnonymizedFile.ts @@ -331,11 +331,11 @@ export default class AnonymizedFile { }); } const content = await this.repository.source?.getFileContent(this); - if ( - !this.repository.model.isReseted || - this.repository.status != RepositoryStatus.READY - ) { - this.repository.model.isReseted = false; + const cacheWasReset = this.repository.model.isReseted; + if (cacheWasReset) { + await this.repository.markCachePresent(); + } + if (cacheWasReset || this.repository.status != RepositoryStatus.READY) { await this.repository.updateStatus(RepositoryStatus.READY); } return content; diff --git a/src/core/Repository.ts b/src/core/Repository.ts index 3965230..53dd195 100644 --- a/src/core/Repository.ts +++ b/src/core/Repository.ts @@ -543,6 +543,20 @@ export default class Repository { } } + /** + * Record that repository content has been cached again after a reset. + */ + async markCachePresent() { + if (!this.model.isReseted) return; + this.model.isReseted = false; + if (isConnected) { + await AnonymizedRepositoryModel.updateOne( + { _id: this._model._id }, + { $set: { isReseted: false } } + ).exec(); + } + } + /** * Compute the size of the repository in term of storage and number of files. * diff --git a/src/queue/index.ts b/src/queue/index.ts index a5ae75d..8efff2a 100644 --- a/src/queue/index.ts +++ b/src/queue/index.ts @@ -1,4 +1,4 @@ -import { Queue, Worker } from "bullmq"; +import { JobsOptions, Queue, Worker } from "bullmq"; import config from "../config"; import AnonymizedRepositoryModel from "../core/model/anonymizedRepositories/anonymizedRepositories.model"; import { RepositoryStatus } from "../core/types"; @@ -22,6 +22,22 @@ const IN_FLIGHT_STATUSES: RepositoryStatus[] = [ RepositoryStatus.DOWNLOAD, ]; +const LIVE_JOB_STATES = new Set([ + "active", + "waiting", + "delayed", + "prioritized", + "waiting-children", +]); + +const REMOVAL_JOB_OPTIONS: JobsOptions = { + attempts: 3, + backoff: { type: "exponential", delay: 1000 }, + removeOnComplete: true, + // Keep a bounded failure history so operators can inspect and retry jobs. + removeOnFail: { count: 1000 }, +}; + async function markErrorIfInFlight(repoId: string, message: string) { try { await AnonymizedRepositoryModel.updateOne( @@ -44,6 +60,26 @@ async function markErrorIfInFlight(repoId: string, message: string) { } } +async function markErrorIfRemoving(repoId: string, message: string) { + try { + await AnonymizedRepositoryModel.updateOne( + { repoId, status: RepositoryStatus.REMOVING }, + { + $set: { + status: RepositoryStatus.ERROR, + statusDate: new Date(), + statusMessage: message || "removal_failed", + }, + } + ).exec(); + } catch (e) { + logger.error("markErrorIfRemoving failed", { + ...serializeError(e), + repoId, + }); + } +} + /** * Recover repositories left in an in-flight status (preparing/queue/download) * with no live BullMQ job — typically caused by a worker process crash or @@ -84,6 +120,58 @@ export let cacheQueue: Queue; export let removeQueue: Queue; export let downloadQueue: Queue; +type RemovalQueue = Pick, "add" | "getJob">; + +/** + * Add an idempotent repository-removal job. A live job wins; a terminal job + * with the same stable id is replaced so a retry is not silently discarded. + */ +export async function addRemovalJob( + repoId: string, + queue: RemovalQueue = removeQueue +): Promise { + const jobId = `repo-${repoId}`; + const existing = await queue.getJob(jobId); + if (existing) { + const state = await existing.getState(); + if (LIVE_JOB_STATES.has(state)) return false; + await existing.remove().catch(() => undefined); + } + await queue.add(repoId, { repoId }, { ...REMOVAL_JOB_OPTIONS, jobId }); + return true; +} + +/** + * Requeue removals whose database status survived but whose BullMQ job did + * not, for example after a crash between updating MongoDB and queueing Redis. + * Repository removal is idempotent, so replaying a terminal job is safe. + */ +export async function recoverStuckRemoving() { + if (!removeQueue) return; + try { + const stuck = await AnonymizedRepositoryModel.find( + { status: RepositoryStatus.REMOVING }, + { repoId: 1 } + ).lean(); + for (const doc of stuck) { + try { + const queued = await addRemovalJob(doc.repoId); + if (queued) { + logger.info("requeued interrupted removal", { repoId: doc.repoId }); + } + } catch (e) { + logger.warn("removal recovery failed", { + ...serializeError(e), + repoId: doc.repoId, + }); + await markErrorIfRemoving(doc.repoId, "removal_interrupted"); + } + } + } catch (e) { + logger.error("recoverStuckRemoving failed", serializeError(e)); + } +} + // avoid to load the queue outside the main server export function startWorker() { const connection = { @@ -103,10 +191,7 @@ export function startWorker() { host: config.REDIS_HOSTNAME, port: config.REDIS_PORT, }, - defaultJobOptions: { - removeOnComplete: true, - removeOnFail: true, - }, + defaultJobOptions: REMOVAL_JOB_OPTIONS, }); downloadQueue = new Queue("repository download", { connection, @@ -144,8 +229,18 @@ export function startWorker() { recordMetric("remove", "completed", (job.finishedOn || Date.now()) - (job.processedOn || job.timestamp)); await job.remove(); }); - removeWorker.on("failed", async (job) => { + removeWorker.on("failed", async (job, err) => { if (job) recordMetric("remove", "failed", Date.now() - (job.processedOn || job.timestamp)); + const repoId = job?.data?.repoId; + logger.error("removal failed", { + ...serializeError(err), + repoId, + }); + if (!repoId) return; + if (job && typeof job.attemptsMade === "number" && job.opts?.attempts) { + if (job.attemptsMade < job.opts.attempts) return; + } + await markErrorIfRemoving(repoId, err?.message || "removal_failed"); }); const downloadWorker = new Worker( diff --git a/src/queue/processes/removeRepository.ts b/src/queue/processes/removeRepository.ts index 60853db..07ae1e8 100644 --- a/src/queue/processes/removeRepository.ts +++ b/src/queue/processes/removeRepository.ts @@ -17,23 +17,31 @@ export async function processRemoveRepository( ) { const { connect, getRepository }: Database = database || require("../../server/database"); + let repo: Awaited> | undefined; try { await connect(); logger.info("removing repository", { repoId: job.data.repoId }); - const repo = await getRepository(job.data.repoId); + repo = await getRepository(job.data.repoId); await repo.updateStatus(RepositoryStatus.REMOVING, ""); - try { - await repo.remove(); - } catch (error) { - if (error instanceof Error) { - await repo.updateStatus(RepositoryStatus.ERROR, error.message); - } else if (typeof error === "string") { - await repo.updateStatus(RepositoryStatus.ERROR, error); - } - throw error; - } + await repo.remove(); logger.info("repository removed", { repoId: job.data.repoId }); } catch (error) { + if (repo) { + const message = + error instanceof Error + ? error.message + : typeof error === "string" + ? error + : "removal_failed"; + try { + await repo.updateStatus(RepositoryStatus.ERROR, message); + } catch (statusError) { + logger.error("failed to record repository removal error", { + ...serializeError(statusError), + repoId: job.data.repoId, + }); + } + } logger.error("repository removal failed", { ...serializeError(error), repoId: job.data.repoId, diff --git a/src/server/index.ts b/src/server/index.ts index f3b58c3..4fe323f 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -17,9 +17,14 @@ import router from "./routes"; import { conferenceStatusCheck, repositoryStatusCheck, + runRepositoryStatusCheck, dailyStatsSnapshot, } from "./schedule"; -import { startWorker, recoverStuckPreparing } from "../queue"; +import { + startWorker, + recoverStuckPreparing, + recoverStuckRemoving, +} from "../queue"; import { computeStats, ensureTodaySnapshot, @@ -363,6 +368,12 @@ export default async function start() { recoverStuckPreparing().catch((err) => logger.error("recoverStuckPreparing failed", serializeError(err)) ); + recoverStuckRemoving().catch((err) => + logger.error("recoverStuckRemoving failed", serializeError(err)) + ); + runRepositoryStatusCheck().catch((err) => + logger.error("initial repository status check failed", serializeError(err)) + ); } start(); diff --git a/src/server/routes/repository-private.ts b/src/server/routes/repository-private.ts index 7640fe2..9258583 100644 --- a/src/server/routes/repository-private.ts +++ b/src/server/routes/repository-private.ts @@ -17,7 +17,7 @@ import { IAnonymizedRepositoryDocument } from "../../core/model/anonymizedReposi import UserModel from "../../core/model/users/users.model"; import ConferenceModel from "../../core/model/conference/conferences.model"; import AnonymousError from "../../core/AnonymousError"; -import { downloadQueue, removeQueue } from "../../queue"; +import { addRemovalJob, downloadQueue } from "../../queue"; import RepositoryModel from "../../core/model/repositories/repositories.model"; import User from "../../core/User"; import { RepositoryStatus } from "../../core/types"; @@ -241,7 +241,14 @@ router.delete( const user = await getUser(req); isOwnerOrAdmin([repo.owner.id], user); await repo.updateStatus(RepositoryStatus.REMOVING); - await removeQueue.add(repo.repoId, { repoId: repo.repoId }, { jobId: `repo-${repo.repoId}` }); + try { + await addRemovalJob(repo.repoId); + } catch (error) { + const message = + error instanceof Error ? error.message : "removal_enqueue_failed"; + await repo.updateStatus(RepositoryStatus.ERROR, message); + throw error; + } return res.json({ status: repo.status }); } catch (error) { handleError(error, res, req); diff --git a/src/server/schedule.ts b/src/server/schedule.ts index c612c14..1941966 100644 --- a/src/server/schedule.ts +++ b/src/server/schedule.ts @@ -30,55 +30,120 @@ export function conferenceStatusCheck() { export function repositoryStatusCheck() { // check every 6 hours the status of the repositories schedule.scheduleJob("0 */6 * * *", async () => { - logger.info("checking repository status and unused repositories"); - const now = new Date(); - const fourMonthAgo = new Date(now); - fourMonthAgo.setMonth(fourMonthAgo.getMonth() - 4); - const cursor = AnonymizedRepositoryModel.find({ - status: RepositoryStatus.READY, - isReseted: false, - $or: [ - { - "options.expirationMode": { $in: ["redirect", "remove"] }, - "options.expirationDate": { $lte: now }, - }, - { lastView: { $lt: fourMonthAgo } }, - ], - }).cursor(); - const batch: Promise[] = []; - for await (const data of cursor) { - batch.push( - (async () => { - const repo = new Repository(data); - try { - await repo.check(); - } catch { - logger.info("repository expired", { repoId: repo.repoId }); - } + await runRepositoryStatusCheck(); + }); +} - if (repo.model.lastView < fourMonthAgo) { - try { - await repo.removeCache(); - } catch (error) { - logger.error("repository cache removal failed", { - ...serializeError(error), - repoId: repo.repoId, - }); - return; - } - logger.info("removed cache for unused repository", { +export function repositoryMaintenanceQuery(now: Date) { + const fourMonthAgo = new Date(now); + fourMonthAgo.setMonth(fourMonthAgo.getMonth() - 4); + return { + status: RepositoryStatus.READY, + $or: [ + { + "options.expirationMode": { $in: ["redirect", "remove"] }, + "options.expirationDate": { $lte: now }, + }, + { + isReseted: { $ne: true }, + lastView: { $lt: fourMonthAgo }, + }, + ], + }; +} + +export async function runRepositoryStatusCheck(now = new Date()) { + logger.info("checking repository status and unused repositories"); + const fourMonthAgo = new Date(now); + fourMonthAgo.setMonth(fourMonthAgo.getMonth() - 4); + const batch: Promise[] = []; + const flushBatch = async () => { + await Promise.all(batch); + batch.length = 0; + }; + + const cursor = AnonymizedRepositoryModel.find( + repositoryMaintenanceQuery(now) + ).cursor(); + for await (const data of cursor) { + batch.push( + (async () => { + const repo = new Repository(data); + const shouldExpire = + repo.options.expirationMode !== "never" && + repo.options.expirationDate != null && + repo.options.expirationDate <= now; + if (shouldExpire) { + try { + await repo.expire(); + logger.info("repository expired", { repoId: repo.repoId }); + } catch (error) { + logger.error("repository expiration failed", { + ...serializeError(error), repoId: repo.repoId, }); } - })() - ); - if (batch.length >= 10) { - await Promise.all(batch); - batch.length = 0; - } + return; + } + + if ( + repo.model.isReseted !== true && + repo.model.lastView < fourMonthAgo + ) { + try { + await repo.removeCache(); + logger.info("removed cache for unused repository", { + repoId: repo.repoId, + }); + } catch (error) { + logger.error("repository cache removal failed", { + ...serializeError(error), + repoId: repo.repoId, + }); + } + } + })() + ); + if (batch.length >= 10) { + await flushBatch(); } - await Promise.all(batch); - }); + } + await flushBatch(); + + // Repair terminal records left with data by an older or interrupted + // expiration. This makes the cleanup idempotent across deployments. + const dirtyTerminalCursor = AnonymizedRepositoryModel.find({ + $or: [ + { status: RepositoryStatus.EXPIRING }, + { + status: RepositoryStatus.EXPIRED, + isReseted: { $ne: true }, + }, + ], + }).cursor(); + for await (const data of dirtyTerminalCursor) { + batch.push( + (async () => { + const repo = new Repository(data); + try { + await repo.resetSate(); + await repo.updateStatus(RepositoryStatus.EXPIRED); + logger.info("recovered expired repository cleanup", { + repoId: repo.repoId, + }); + } catch (error) { + logger.error("expired repository cleanup failed", { + ...serializeError(error), + repoId: repo.repoId, + }); + } + })() + ); + if (batch.length >= 10) { + await flushBatch(); + } + } + await flushBatch(); } export function dailyStatsSnapshot() { diff --git a/test/backend-reliability.test.js b/test/backend-reliability.test.js index eb26d09..2978e09 100644 --- a/test/backend-reliability.test.js +++ b/test/backend-reliability.test.js @@ -16,6 +16,12 @@ const { const { processRemoveCache, } = require("../src/queue/processes/removeCache"); +const { addRemovalJob } = require("../src/queue"); +const { + repositoryMaintenanceQuery, +} = require("../src/server/schedule"); +const Repository = require("../src/core/Repository").default; +const AnonymizedRepositoryModel = require("../src/core/model/anonymizedRepositories/anonymizedRepositories.model").default; describe("conference edits", function () { const form = { @@ -149,6 +155,72 @@ describe("removal workers", function () { expect(statuses[statuses.length - 1][1]).to.equal(failure.message); }); + it("records errors that happen before repository removal starts", async function () { + const statuses = []; + const failure = new Error("status write failed"); + let statusCalls = 0; + const repo = { + updateStatus: async (status, message) => { + statusCalls++; + statuses.push([status, message]); + if (statusCalls === 1) throw failure; + }, + remove: async () => undefined, + }; + + let caught; + try { + await processRemoveRepository(job, { + connect: async () => undefined, + getRepository: async () => repo, + }); + } catch (error) { + caught = error; + } + expect(caught).to.equal(failure); + expect(statuses[statuses.length - 1][1]).to.equal(failure.message); + }); + + it("does not add a duplicate when a removal job is live", async function () { + let additions = 0; + const queue = { + getJob: async () => ({ + getState: async () => "active", + remove: async () => undefined, + }), + add: async () => { + additions++; + }, + }; + + expect(await addRemovalJob("repo-1", queue)).to.equal(false); + expect(additions).to.equal(0); + }); + + it("replaces a terminal removal job with a retryable job", async function () { + let removed = false; + let added; + const queue = { + getJob: async () => ({ + getState: async () => "failed", + remove: async () => { + removed = true; + }, + }), + add: async (...args) => { + added = args; + }, + }; + + expect(await addRemovalJob("repo-1", queue)).to.equal(true); + expect(removed).to.equal(true); + expect(added[0]).to.equal("repo-1"); + expect(added[1]).to.deep.equal({ repoId: "repo-1" }); + expect(added[2].jobId).to.equal("repo-repo-1"); + expect(added[2].attempts).to.equal(3); + expect(added[2].removeOnFail).to.deep.equal({ count: 1000 }); + }); + it("rejects cache jobs when cache removal fails", async function () { const failure = new Error("storage unavailable"); let caught; @@ -167,3 +239,32 @@ describe("removal workers", function () { expect(caught).to.equal(failure); }); }); + +describe("repository expiration maintenance", function () { + it("selects due repositories regardless of their cache reset flag", function () { + const now = new Date("2026-08-20T00:00:00.000Z"); + const query = repositoryMaintenanceQuery(now); + + expect(query).not.to.have.property("isReseted"); + expect(query.$or[0]).to.deep.equal({ + "options.expirationMode": { $in: ["redirect", "remove"] }, + "options.expirationDate": { $lte: now }, + }); + }); + + it("marks a cache as present again after content is restored", async function () { + const model = new AnonymizedRepositoryModel({ + repoId: "repo-cache-state", + owner: "507f1f77bcf86cd799439011", + isReseted: true, + status: "ready", + options: { terms: [], expirationMode: "never" }, + source: { type: "GitHubStream", repositoryName: "owner/repo" }, + }); + const repo = new Repository(model); + + await repo.markCachePresent(); + + expect(repo.model.isReseted).to.equal(false); + }); +});