fix(secondary-sync): clean abandoned staging objects

Prevent failed or interrupted secondary sync workflows from exhausting the IHEP S3 quota by cleaning the current staging prefix on failure and pruning abandoned staging objects and multipart uploads.

Author: Wangnov

cloudflare/secondary-sync/README.md

@@ -8,12 +8,19 @@ hosted runner.
 
 1. GitHub publishes the Release and finishes the R2 `latest/*` sync.
 2. GitHub calls `POST /sync/start` with the release tag.
-3. The Workflow reads R2 in ranges and uploads objects to an IHEP staging prefix.
+3. The Workflow removes abandoned staging objects and multipart uploads older
+   than `SECONDARY_SYNC_STAGE_GRACE_HOURS`, then reads R2 in ranges and uploads
+   objects to an IHEP staging prefix.
 4. It copies staging objects to `latest/*` on IHEP, committing `latest/manifest`
    after installers, Sparkle archives, checksums, and appcasts.
 5. It removes stale latest aliases that are absent from the current manifest,
    prunes stale `latest/mac/*` Sparkle archives outside the grace window, and
-   removes staging objects.
+   removes staging objects. If upload or commit fails, it also attempts to
+   remove the current instance's staging objects before preserving the original
+   Workflow error.
+
+The default staging grace window is two hours. It protects recent concurrent
+instances while bounding storage leaked by interrupted or terminated Workflows.
 
 ## Required secrets
 

cloudflare/secondary-sync/src/core.js

@@ -5,6 +5,7 @@ const SHORT_CACHE_CONTROL = "public, max-age=600";
 const DEFAULT_STAGE_PREFIX = "staging/secondary-sync";
 const ALIAS_ROLLBACK_SUFFIX = ".rollback";
 const DEFAULT_PRUNE_GRACE_DAYS = 1;
+const DEFAULT_STAGE_GRACE_HOURS = 2;
 
 export class NonRetryableMirrorError extends Error {
   constructor(message) {
@@ -606,6 +607,47 @@ export async function pruneStaleLatestMacObjects(env, keepKeys, options = {}) {
   return { pruned, graceDays };
 }
 
+export async function pruneAbandonedStageObjects(env, keepStagePrefix = "", options = {}) {
+  const s3 = createS3Client(env);
+  const stageRoot = `${cleanPrefix(env.SECONDARY_SYNC_STAGE_PREFIX || DEFAULT_STAGE_PREFIX)}/`;
+  const graceHours = parseNonNegativeInteger(
+    options.graceHours ?? env.SECONDARY_SYNC_STAGE_GRACE_HOURS,
+    DEFAULT_STAGE_GRACE_HOURS,
+  );
+  const cutoff = Date.now() - graceHours * 60 * 60 * 1000;
+  const deleted = [];
+  const aborted = [];
+
+  for (const object of await s3.listObjects(stageRoot)) {
+    if (!isAbandonedStageEntry(object.key, object.lastModified, keepStagePrefix, cutoff)) {
+      continue;
+    }
+    await s3.deleteObject(object.key);
+    deleted.push(object.key);
+  }
+
+  for (const upload of await s3.listMultipartUploads(stageRoot)) {
+    if (!isAbandonedStageEntry(upload.key, upload.initiated, keepStagePrefix, cutoff)) {
+      continue;
+    }
+    await s3.abortMultipartUpload(upload.key, upload.uploadId);
+    aborted.push({ key: upload.key, uploadId: upload.uploadId });
+  }
+
+  return { deleted, aborted, graceHours };
+}
+
+function isAbandonedStageEntry(key, timestamp, keepStagePrefix, cutoff) {
+  if (!key || !timestamp || Number.isNaN(timestamp.getTime())) {
+    return false;
+  }
+  const keepPrefix = cleanPrefix(keepStagePrefix);
+  if (keepPrefix && (key === keepPrefix || key.startsWith(`${keepPrefix}/`))) {
+    return false;
+  }
+  return timestamp.getTime() < cutoff;
+}
+
 export function decoratePlanWithStage(plan, instanceId, env = {}) {
   const safeTag = safeKeySegment(plan.releaseTag);
   const safeInstance = safeKeySegment(instanceId);
@@ -852,6 +894,35 @@ export class S3Client {
     return response;
   }
 
+  async listMultipartUploads(prefix) {
+    const uploads = [];
+    let keyMarker = "";
+    let uploadIdMarker = "";
+    do {
+      const query = [
+        ["uploads", ""],
+        ["prefix", prefix],
+      ];
+      if (keyMarker) {
+        query.push(["key-marker", keyMarker]);
+      }
+      if (uploadIdMarker) {
+        query.push(["upload-id-marker", uploadIdMarker]);
+      }
+      const response = await this.request({
+        method: "GET",
+        key: "",
+        query,
+      });
+      await assertOk(response, "LIST MULTIPART", prefix);
+      const xml = await response.text();
+      uploads.push(...xmlMultipartUploads(xml));
+      keyMarker = xmlText(xml, "NextKeyMarker");
+      uploadIdMarker = xmlText(xml, "NextUploadIdMarker");
+    } while (keyMarker);
+    return uploads;
+  }
+
   async copyObject(sourceKey, destinationKey) {
     const response = await this.request({
       method: "PUT",
@@ -1237,6 +1308,26 @@ function xmlContents(xml) {
   return entries;
 }
 
+function xmlMultipartUploads(xml) {
+  const entries = [];
+  const uploadRegex = /<Upload>([\s\S]*?)<\/Upload>/g;
+  let match;
+  while ((match = uploadRegex.exec(xml))) {
+    const block = match[1];
+    const key = xmlText(block, "Key");
+    const uploadId = xmlText(block, "UploadId");
+    if (!key || !uploadId) {
+      continue;
+    }
+    entries.push({
+      key,
+      uploadId,
+      initiated: parseHttpDate(xmlText(block, "Initiated")),
+    });
+  }
+  return entries;
+}
+
 function decodeXmlText(value) {
   return String(value)
     .replace(/&lt;/g, "<")

cloudflare/secondary-sync/src/index.js

@@ -9,6 +9,7 @@ import {
   decoratePlanWithStage,
   discoverSourceManifest,
   handleApiRequest,
+  pruneAbandonedStageObjects,
   pruneStaleLatestMacObjects,
   reconcileSync,
   uploadObjectToStage,
@@ -46,51 +47,76 @@ export class SecondaryMirrorWorkflow extends WorkflowEntrypoint {
       this.env,
     );
 
-    const uploadResults = [];
-    for (const item of plan.items) {
-      uploadResults.push(
-        await step.do(stepName("upload", item), UPLOAD_STEP, () =>
-          nonRetryableGuard(() =>
-            uploadObjectToStage(this.env, item, {
-              forceUpload: Boolean(payload.force),
-            }),
+    const abandonedStageResult = await step.do(
+      "prune abandoned secondary staging",
+      HOUSEKEEPING_STEP,
+      () => nonRetryableGuard(() => pruneAbandonedStageObjects(this.env, plan.stagePrefix)),
+    );
+
+    try {
+      const uploadResults = [];
+      for (const item of plan.items) {
+        uploadResults.push(
+          await step.do(stepName("upload", item), UPLOAD_STEP, () =>
+            nonRetryableGuard(() =>
+              uploadObjectToStage(this.env, item, {
+                forceUpload: Boolean(payload.force),
+              }),
+            ),
           ),
-        ),
-      );
-    }
+        );
+      }
 
-    const commitResults = [];
-    for (const item of commitOrder(plan.items)) {
-      commitResults.push(
-        await step.do(stepName("commit", item), COMMIT_STEP, () =>
-          nonRetryableGuard(() => commitObjectToAliases(this.env, item)),
-        ),
-      );
-    }
+      const commitResults = [];
+      for (const item of commitOrder(plan.items)) {
+        commitResults.push(
+          await step.do(stepName("commit", item), COMMIT_STEP, () =>
+            nonRetryableGuard(() => commitObjectToAliases(this.env, item)),
+          ),
+        );
+      }
 
-    const staleAliasResult = await step.do("delete stale latest aliases", HOUSEKEEPING_STEP, () =>
-      nonRetryableGuard(() => deleteStaleAliasObjects(this.env, plan.staleAliasKeys)),
-    );
+      const staleAliasResult = await step.do("delete stale latest aliases", HOUSEKEEPING_STEP, () =>
+        nonRetryableGuard(() => deleteStaleAliasObjects(this.env, plan.staleAliasKeys)),
+      );
 
-    const pruneResult = await step.do("prune stale Sparkle archives", HOUSEKEEPING_STEP, () =>
-      nonRetryableGuard(() => pruneStaleLatestMacObjects(this.env, plan.keepLatestMacKeys)),
-    );
+      const pruneResult = await step.do("prune stale Sparkle archives", HOUSEKEEPING_STEP, () =>
+        nonRetryableGuard(() => pruneStaleLatestMacObjects(this.env, plan.keepLatestMacKeys)),
+      );
 
-    const cleanupResult = await step.do("cleanup secondary staging objects", HOUSEKEEPING_STEP, () =>
-      nonRetryableGuard(() => cleanupStageObjects(this.env, plan.items)),
-    );
+      const cleanupResult = await step.do("cleanup secondary staging objects", HOUSEKEEPING_STEP, () =>
+        nonRetryableGuard(() => cleanupStageObjects(this.env, plan.items)),
+      );
 
-    return {
-      releaseTag: source.releaseTag,
-      manifestSha256: source.manifestSha256,
-      stagePrefix: plan.stagePrefix,
-      uploaded: uploadResults.length,
-      committed: commitResults.reduce((count, result) => count + result.aliases.length, 0),
-      staleAliasesDeleted: staleAliasResult.deleted.length,
-      pruned: pruneResult.pruned.length,
-      cleaned: cleanupResult.deleted.length,
-      completedAt: new Date().toISOString(),
-    };
+      return {
+        releaseTag: source.releaseTag,
+        manifestSha256: source.manifestSha256,
+        stagePrefix: plan.stagePrefix,
+        abandonedStageObjectsDeleted: abandonedStageResult.deleted.length,
+        abandonedMultipartUploadsAborted: abandonedStageResult.aborted.length,
+        uploaded: uploadResults.length,
+        committed: commitResults.reduce((count, result) => count + result.aliases.length, 0),
+        staleAliasesDeleted: staleAliasResult.deleted.length,
+        pruned: pruneResult.pruned.length,
+        cleaned: cleanupResult.deleted.length,
+        completedAt: new Date().toISOString(),
+      };
+    } catch (error) {
+      try {
+        await step.do("cleanup failed secondary staging objects", HOUSEKEEPING_STEP, () =>
+          nonRetryableGuard(() => cleanupStageObjects(this.env, plan.items)),
+        );
+      } catch (cleanupError) {
+        console.error(
+          JSON.stringify({
+            event: "secondary_sync_failure_cleanup_failed",
+            stagePrefix: plan.stagePrefix,
+            error: cleanupError.message,
+          }),
+        );
+      }
+      throw error;
+    }
   }
 }
 

cloudflare/secondary-sync/test/core.test.mjs

@@ -8,6 +8,7 @@ import {
   deleteStaleAliasObjects,
   decoratePlanWithStage,
   deriveReleaseTag,
+  pruneAbandonedStageObjects,
   pruneStaleLatestMacObjects,
   uploadObjectToStage,
 } from "../src/core.js";
@@ -188,6 +189,45 @@ test("prunes stale unreferenced Sparkle archives but keeps current and recent ob
   assert.equal(s3.objects.has("latest/mac/intel/recent.zip"), true);
 });
 
+test("prunes abandoned staging objects and multipart uploads without touching active work", async () => {
+  const s3 = createMockS3();
+  const old = new Date("2020-01-01T00:00:00Z");
+  const recent = new Date(Date.now() + 60_000);
+  const activePrefix = "staging/secondary-sync/release/active";
+  s3.objects.set("staging/secondary-sync/release/old/objects/archive", objectEntry("old", old));
+  s3.objects.set("staging/secondary-sync/release/recent/objects/archive", objectEntry("recent", recent));
+  s3.objects.set(`${activePrefix}/objects/archive`, objectEntry("active", old));
+  s3.objects.set("latest/win", objectEntry("published", old));
+  s3.uploads.set("upload-old", {
+    key: "staging/secondary-sync/release/old/objects/partial",
+    parts: new Map(),
+    initiated: old,
+  });
+  s3.uploads.set("upload-active", {
+    key: `${activePrefix}/objects/partial`,
+    parts: new Map(),
+    initiated: old,
+  });
+  globalThis.fetch = s3.fetch;
+
+  const result = await pruneAbandonedStageObjects(fixtureEnv(new Map()), activePrefix, {
+    graceHours: 1,
+  });
+
+  assert.deepEqual(result.deleted, ["staging/secondary-sync/release/old/objects/archive"]);
+  assert.deepEqual(result.aborted, [
+    {
+      key: "staging/secondary-sync/release/old/objects/partial",
+      uploadId: "upload-old",
+    },
+  ]);
+  assert.equal(s3.objects.has("staging/secondary-sync/release/recent/objects/archive"), true);
+  assert.equal(s3.objects.has(`${activePrefix}/objects/archive`), true);
+  assert.equal(s3.objects.has("latest/win"), true);
+  assert.equal(s3.uploads.has("upload-old"), false);
+  assert.equal(s3.uploads.has("upload-active"), true);
+});
+
 test("falls back to ListObjectsV2 when nested HEAD returns 403", async () => {
   const s3 = createMockS3();
   globalThis.fetch = s3.fetch;
@@ -354,6 +394,20 @@ function createMockS3(options = {}) {
       return new Response(`<ListBucketResult>${contents}</ListBucketResult>`, { status: 200 });
     }
 
+    if (method === "GET" && url.searchParams.has("uploads")) {
+      const prefix = url.searchParams.get("prefix") || "";
+      const entries = [...uploads.entries()]
+        .filter(([, upload]) => upload.key.startsWith(prefix))
+        .map(
+          ([uploadId, upload]) =>
+            `<Upload><Key>${xmlEscape(upload.key)}</Key><UploadId>${xmlEscape(uploadId)}</UploadId><Initiated>${upload.initiated.toISOString()}</Initiated></Upload>`,
+        )
+        .join("");
+      return new Response(`<ListMultipartUploadsResult>${entries}</ListMultipartUploadsResult>`, {
+        status: 200,
+      });
+    }
+
     if (method === "HEAD") {
       if (key.includes("head-403")) {
         return new Response("forbidden", { status: 403 });
@@ -385,7 +439,7 @@ function createMockS3(options = {}) {
 
     if (method === "POST" && url.searchParams.has("uploads")) {
       const uploadId = `upload-${++uploadCounter}`;
-      uploads.set(uploadId, { key, parts: new Map() });
+      uploads.set(uploadId, { key, parts: new Map(), initiated: new Date() });
       return new Response(`<InitiateMultipartUploadResult><UploadId>${uploadId}</UploadId></InitiateMultipartUploadResult>`);
     }
 

cloudflare/secondary-sync/wrangler.jsonc

@@ -32,6 +32,7 @@
     "SECONDARY_SYNC_PART_SIZE_BYTES": "16777216",
     "SECONDARY_SYNC_SINGLE_PUT_MAX_BYTES": "33554432",
     "SECONDARY_SYNC_PRUNE_GRACE_DAYS": "1",
+    "SECONDARY_SYNC_STAGE_GRACE_HOURS": "2",
     "WORKFLOW_SUCCESS_RETENTION": "1 day",
     "WORKFLOW_ERROR_RETENTION": "7 days"
   },