diff --git a/src/core/Repository.ts b/src/core/Repository.ts index 2511c97..464314d 100644 --- a/src/core/Repository.ts +++ b/src/core/Repository.ts @@ -524,7 +524,7 @@ export default class Repository { this.status == RepositoryStatus.PREPARING || this.status == RepositoryStatus.QUEUE || (this.status == RepositoryStatus.DOWNLOAD && - this._model.statusDate > fiveMinuteAgo) + this.statusActivityDate > fiveMinuteAgo) ) { const rlMatch = (this._model.statusMessage || "").match(/^rate_limited:(\d+)$/); if (rlMatch) { @@ -840,6 +840,35 @@ export default class Repository { */ public protectLifecycle = false; + get statusActivityDate(): Date { + const { statusDate, progressDate } = this.model; + return this.status === RepositoryStatus.DOWNLOAD && progressDate && + (!statusDate || progressDate > statusDate) ? progressDate : statusDate; + } + + /** Progress is activity, not a lifecycle transition or a new publication fence. */ + async updateProgressMessage(statusMessage: string) { + const status = this.status; + if (status !== RepositoryStatus.DOWNLOAD) return; + const statusDate = this.model.statusDate; + const progressDate = new Date(); + if (isConnected) { + const result = await AnonymizedRepositoryModel.updateOne({ + _id: this.model._id, ...this.refreshFilter(), status, + statusDate: statusDate || { $exists: false }, + anonymizeDate: this.model.anonymizeDate, + "source.commit": this.model.source.commit || { $exists: false }, + "githubAccess.revision": this.model.githubAccess?.revision || { $exists: false }, + }, { $set: { statusMessage, progressDate } }).exec(); + if (!result.matchedCount) return; + } + // An awaited write can finish after this instance has moved to READY. + if (this.status === status && this.model.statusDate?.getTime() === statusDate?.getTime()) { + this.model.statusMessage = statusMessage; + this.model.progressDate = progressDate; + } + } + async updateStatus(status: RepositoryStatus, statusMessage?: string) { if (status !== RepositoryStatus.ARCHIVED) this.assertNotArchived(); if (!status) return this.model; diff --git a/src/core/model/anonymizedRepositories/anonymizedRepositories.schema.ts b/src/core/model/anonymizedRepositories/anonymizedRepositories.schema.ts index 23f197b..d11bb6d 100644 --- a/src/core/model/anonymizedRepositories/anonymizedRepositories.schema.ts +++ b/src/core/model/anonymizedRepositories/anonymizedRepositories.schema.ts @@ -17,6 +17,7 @@ const AnonymizedRepositorySchema = new Schema({ default: "preparing", }, statusDate: Date, + progressDate: Date, refreshToken: { type: String, select: false }, refreshUntil: { type: Date, select: false }, archivedAt: Date, diff --git a/src/core/model/anonymizedRepositories/anonymizedRepositories.types.ts b/src/core/model/anonymizedRepositories/anonymizedRepositories.types.ts index 5c3cd88..b1c2d2d 100644 --- a/src/core/model/anonymizedRepositories/anonymizedRepositories.types.ts +++ b/src/core/model/anonymizedRepositories/anonymizedRepositories.types.ts @@ -7,6 +7,7 @@ export interface IAnonymizedRepository { status?: RepositoryStatus; statusMessage?: string; statusDate: Date; + progressDate?: Date; archivedAt?: Date; archiveReason?: string; archiveCachePending?: boolean; diff --git a/src/queue/processes/downloadRepository.ts b/src/queue/processes/downloadRepository.ts index 2b9a8e1..85e36ee 100644 --- a/src/queue/processes/downloadRepository.ts +++ b/src/queue/processes/downloadRepository.ts @@ -75,7 +75,7 @@ export default async function (job: SandboxedJob) { repoId: job.data.repoId, status: progress.status, }); - await repo.updateStatus(repo.status, progress?.status || ""); + await repo.updateProgressMessage(progress?.status || ""); } } catch { // ignore error diff --git a/src/server/routes/repository-private.ts b/src/server/routes/repository-private.ts index 18f1f7f..0c3c9bd 100644 --- a/src/server/routes/repository-private.ts +++ b/src/server/routes/repository-private.ts @@ -159,7 +159,7 @@ router.post( repo.status == RepositoryStatus.PREPARING || repo.status == RepositoryStatus.QUEUE || (repo.status == RepositoryStatus.DOWNLOAD && - repo.model.statusDate > new Date(Date.now() - 5 * 60_000)) || + repo.statusActivityDate > new Date(Date.now() - 5 * 60_000)) || repo.status == RepositoryStatus.REMOVING || repo.status == RepositoryStatus.EXPIRING ) { diff --git a/src/server/routes/repository-public.ts b/src/server/routes/repository-public.ts index e6388f1..c2839c7 100644 --- a/src/server/routes/repository-public.ts +++ b/src/server/routes/repository-public.ts @@ -323,7 +323,7 @@ router.get( if (repo.status != "ready") { if ( repo.status != RepositoryStatus.QUEUE && - repo.model.statusDate < fiveMinuteAgo + repo.statusActivityDate < fiveMinuteAgo ) { await repo.updateStatus(RepositoryStatus.PREPARING); await downloadQueue.add(repo.repoId, { repoId: repo.repoId }, { diff --git a/test/download-progress-fences.test.js b/test/download-progress-fences.test.js new file mode 100644 index 0000000..3d84c32 --- /dev/null +++ b/test/download-progress-fences.test.js @@ -0,0 +1,77 @@ +const { expect } = require("chai"); +require("ts-node/register/transpile-only"); +const db = require("../src/server/database"); +const Model = require("../src/core/model/anonymizedRepositories/anonymizedRepositories.model").default; +const Repository = require("../src/core/Repository").default; +const gh = require("../src/core/GitHubUtils"); +const sift = require("sift").default; + +describe("download progress publication fences", function () { + const restores = []; + function stub(object, key, value) { const before = object[key]; restores.push(() => { object[key] = before; }); object[key] = value; } + afterEach(() => { while (restores.length) restores.pop()(); }); + function setup() { + const model = new Model({ repoId: "progress-test", status: "download", statusDate: new Date(1000), + owner: "507f1f77bcf86cd799439011", anonymizeDate: new Date(500), + source: { repositoryName: "owner/repo", commit: "abc", branch: "main" }, options: { terms: [], expirationMode: "never" } }); + const repo = new Repository(model); repo.protectLifecycle = true; + const stored = model.toObject(); + stub(db, "isConnected", true); + stub(Model, "updateOne", (filter, update) => ({ exec: async () => { + if (!sift(filter)(stored)) return { matchedCount: 0 }; + Object.assign(stored, update.$set); return { matchedCount: 1 }; + } })); + return { repo, stored }; + } + + it("keeps a captured publication fence valid when a queue progress tick runs", async function () { + const { repo, stored } = setup(); let tick; + stub(global, "setInterval", callback => { tick = callback; return 1; }); + stub(global, "clearInterval", () => {}); + stub(db, "connect", async () => {}); + stub(db, "getRepository", async () => repo); + stub(gh, "getToken", async () => "test-token"); + stub(gh, "getRedisGateResetAt", async () => 0); + stub(repo, "resetSate", async () => {}); + stub(repo, "anonymize", async progress => { + const fence = { status: stored.status, statusDate: stored.statusDate }; + progress("Writing files"); tick(); + await new Promise(resolve => setImmediate(resolve)); + expect(sift(fence)(stored), "progress invalidated the publication fence").to.equal(true); + expect(stored.statusMessage).to.equal("Writing files"); + }); + await require("../src/queue/processes/downloadRepository").default({ data: { repoId: repo.repoId }, updateProgress: async () => {} }); + expect(stored.status).to.equal("ready"); + }); + + it("keeps fresh long-running downloads busy without changing statusDate", async function () { + const { repo, stored } = setup(); const before = repo.model.statusDate; + await repo.updateProgressMessage("Writing files"); + expect(stored.statusDate).to.deep.equal(before); + expect(repo.model.statusDate).to.deep.equal(before); + expect(repo.statusActivityDate.getTime()).to.be.greaterThan(Date.now() - 1000); + try { await repo.check(); throw new Error("expected not ready"); } + catch (error) { expect(error.message).to.equal("repository_not_ready"); } + }); + + for (const replacement of ["ready", "removed", "download"]) { + it(`does not overwrite progress for a newer ${replacement} state`, async function () { + const { repo, stored } = setup(); + stored.status = replacement; stored.statusDate = new Date(2000); stored.statusMessage = "new operation"; + await repo.updateProgressMessage("old progress"); + expect(stored.statusMessage).to.equal("new operation"); + expect(repo.model.statusMessage).not.to.equal("old progress"); + expect(stored).not.to.have.property("progressDate"); + }); + } + + it("does not apply a completed progress write to a locally advanced status", async function () { + const { repo } = setup(); + stub(Model, "updateOne", () => ({ exec: async () => { + repo.model.status = "ready"; repo.model.statusDate = new Date(2000); repo.model.statusMessage = "done"; + return { matchedCount: 1 }; + } })); + await repo.updateProgressMessage("old progress"); + expect(repo.model.statusMessage).to.equal("done"); + }); +});