Wait for an upload's URL to expire before deleting its bytes

convex/imageAssetLifecycle.test.ts

@@ -29,6 +29,9 @@ const sweepDeletedImageAssets = makeFunctionReference<"action">(
   "images:sweepDeletedImageAssets",
 );
 const completeUpload = makeFunctionReference<"action">("images:completeUpload");
+const generateUploadUrl = makeFunctionReference<"action">(
+  "images:generateUploadUrl",
+);
 const getAssetUrl = makeFunctionReference<"query">("images:getAssetUrl");
 
 type Harness = TestConvexForDataModel<DataModel>;
@@ -551,6 +554,51 @@ describe("image asset lifecycle", () => {
     expect(fetchMock).toHaveBeenCalledTimes(1);
   });
 
+  test("an upload deleted while its URL still works keeps its row until the URL dies", async () => {
+    vi.useFakeTimers();
+    const fetchMock = mockR2Deletes();
+    const { t, owner } = await createHarness();
+    await seedStrategy(owner);
+    const upload = (await owner.action(generateUploadUrl, {
+      clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION,
+      strategyPublicId,
+      assetPublicId: "in-flight",
+      mimeType: "image/png",
+      fileExtension: "png",
+    })) as { objectKey: string; expiresAt: number };
+    expect(await allAssets(t)).toMatchObject([
+      { uploadStatus: "pending", uploadUrlExpiresAt: upload.expiresAt },
+    ]);
+
+    // The strategy goes while the client's PUT may still be sending bytes.
+    await owner.mutation(deleteStrategy, {
+      clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION,
+      strategyPublicId,
+      expectedRevision: 0,
+    });
+    await t.finishAllScheduledFunctions(vi.runAllTimers);
+    expect(fetchMock).not.toHaveBeenCalled();
+    expect(await allAssets(t)).toMatchObject([
+      { uploadStatus: "deleted", objectKey: upload.objectKey },
+    ]);
+
+    // Well after expiry, a PUT that started in time may still be sending
+    // over a slow connection.
+    vi.setSystemTime(upload.expiresAt + 23 * 60 * 60 * 1000);
+    await expect(
+      t.action(sweepDeletedImageAssets, {}),
+    ).resolves.toMatchObject({ deleted: 0, failed: 0 });
+
+    vi.setSystemTime(upload.expiresAt + 24 * 60 * 60 * 1000);
+    await expect(
+      t.action(sweepDeletedImageAssets, {}),
+    ).resolves.toMatchObject({ deleted: 1, failed: 0 });
+    expect(await allAssets(t)).toEqual([]);
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+    expect(String(fetchMock.mock.calls[0]?.[0])).toContain(upload.objectKey);
+    await t.finishAllScheduledFunctions(vi.runAllTimers);
+  });
+
   test("legacy reads survive while completion inserts an exact-owned replacement", async () => {
     const { t, owner } = await createHarness();
     await seedStrategy(owner);

convex/images.ts

@@ -35,6 +35,7 @@ import {
   deleteR2Object,
   expectedMimeTypeForExtension,
   getR2Config,
+  getUploadUrlExpiresSeconds,
   headR2Object,
   normalizeImageExtension,
   presignR2PutUrl,
@@ -74,6 +75,11 @@ const physicalDeletionBatch = 25;
 // size, so a batch stays far below Convex's per-transaction limits.
 const reclaimCandidateBatch = 25;
 const staleDeletionClaimAgeMs = 15 * 60 * 1000;
+// R2 checks a presigned URL when the PUT starts, so a PUT that started just
+// before expiry can still be sending bytes after it, for as long as a slow
+// connection takes. A day covers any real upload, and waiting costs little:
+// a deleted element's image already stays for its 30-day tombstone.
+const uploadInFlightGraceMs = 24 * 60 * 60 * 1000;
 const deletionRetryDelayMs = 60 * 1000;
 
 export const markDeletedStrategyImageAssetsRef =
@@ -269,6 +275,7 @@ export const generateUploadUrl = action({
       uploadId: Id<"imageAssets">;
       uploadAttemptPublicId: string;
       objectKey: string;
+      issuedAt: number;
     } =
       await ctx.runMutation(internal.images.createR2UploadIntent, {
         strategyPublicId: args.strategyPublicId,
@@ -281,10 +288,13 @@ export const generateUploadUrl = action({
         width: args.width,
         height: args.height,
       });
+    // Signed at the intent's time, so the URL expires exactly when the
+    // intent's uploadUrlExpiresAt says it does.
     const signed = await presignR2PutUrl({
       config,
       objectKey: intent.objectKey,
       mimeType: validated.mimeType,
+      now: new Date(intent.issuedAt),
     });
 
     return {
@@ -337,6 +347,7 @@ export const createR2UploadIntent = internalMutation({
       width: args.width,
       height: args.height,
       byteSize: args.byteSize,
+      uploadUrlExpiresAt: now + getUploadUrlExpiresSeconds() * 1000,
       updatedAt: now,
     };
     // Content that shows this image may have landed first and left a
@@ -367,6 +378,7 @@ export const createR2UploadIntent = internalMutation({
       uploadId,
       uploadAttemptPublicId: args.uploadAttemptPublicId,
       objectKey: args.objectKey,
+      issuedAt: now,
     };
   },
 });
@@ -1052,6 +1064,10 @@ export const claimDeletedImageAssets = internalMutation({
       1,
       Math.min(args.limit ?? physicalDeletionBatch, physicalDeletionBatch),
     );
+    const now = Date.now();
+    // An upload deleted while its PUT may still be in flight waits for the
+    // URL to die. Deleting first would let the bytes land afterwards with no
+    // row left to reclaim them; the hourly sweep picks the asset up later.
     const assets = await ctx.db
       .query("imageAssets")
       .withIndex("by_uploadStatus_and_updatedAt", (q) =>
@@ -1061,10 +1077,16 @@ export const claimDeletedImageAssets = internalMutation({
         q.and(
           q.neq(q.field("strategyId"), undefined),
           q.eq(q.field("cleanupClaimedAt"), undefined),
+          q.or(
+            q.eq(q.field("uploadUrlExpiresAt"), undefined),
+            q.lte(
+              q.field("uploadUrlExpiresAt"),
+              now - uploadInFlightGraceMs,
+            ),
+          ),
         ),
       )
       .take(limit);
-    const now = Date.now();
     for (const asset of assets) {
       await ctx.db.patch(asset._id, { cleanupClaimedAt: now });
     }

convex/lib/r2.ts

@@ -68,6 +68,14 @@ function optionalPositiveIntEnv(
   return parsed;
 }
 
+export function getUploadUrlExpiresSeconds(): number {
+  return optionalPositiveIntEnv(
+    "R2_UPLOAD_URL_EXPIRES_SECONDS",
+    defaultUploadUrlExpiresSeconds,
+    { min: 1, max: 604800 },
+  );
+}
+
 export function getR2Config(): R2Config {
   const accountId = requiredEnv("R2_ACCOUNT_ID");
   const bucket = requiredEnv("R2_BUCKET");
@@ -82,11 +90,7 @@ export function getR2Config(): R2Config {
     accessKeyId: requiredEnv("R2_ACCESS_KEY_ID"),
     secretAccessKey: requiredEnv("R2_SECRET_ACCESS_KEY"),
     publicBaseUrl: getR2PublicBaseUrl(),
-    uploadUrlExpiresSeconds: optionalPositiveIntEnv(
-      "R2_UPLOAD_URL_EXPIRES_SECONDS",
-      defaultUploadUrlExpiresSeconds,
-      { min: 1, max: 604800 },
-    ),
+    uploadUrlExpiresSeconds: getUploadUrlExpiresSeconds(),
     maxImageBytes: optionalPositiveIntEnv(
       "R2_MAX_IMAGE_BYTES",
       defaultMaxImageBytes,

convex/schema.ts

@@ -230,6 +230,9 @@ export default defineSchema({
     byteSize: v.optional(v.number()),
     etag: v.optional(v.string()),
     uploadedAt: v.optional(v.number()),
+    // When the presigned PUT URL for this upload attempt stops working. Until
+    // then the bytes can still land, so they are not deleted before it.
+    uploadUrlExpiresAt: v.optional(v.number()),
     deletedAt: v.optional(v.number()),
     cleanupClaimedAt: v.optional(v.number()),
     createdAt: v.optional(v.number()),