Merge pull request #919 from tdurieux/fix/download-progress-fences

fix: preserve publication fences during download progress

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;

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,

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;

src/queue/processes/downloadRepository.ts

@@ -75,7 +75,7 @@ export default async function (job: SandboxedJob<RepoJobData, void>) {
               repoId: job.data.repoId,
               status: progress.status,
             });
-            await repo.updateStatus(repo.status, progress?.status || "");
+            await repo.updateProgressMessage(progress?.status || "");
           }
         } catch {
           // ignore error

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
       ) {

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 }, {

test/download-progress-fences.test.js

@@ -0,0 +1,78 @@
+const { expect } = require("chai");
+const { setImmediate } = require("timers");
+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");
+  });
+});