chore: smaller limit on rm stale (#375)

Author: Aslemammad

packages/app/server/routes/rm/stale.post.ts

@@ -1,9 +1,6 @@
 import type { H3Event } from "h3";
 
 export default eventHandler(async (event) => {
-  setResponseHeader(event, "Cache-Control", "no-cache");
-  setResponseHeader(event, "Content-Type", "application/json");
-
   const rmStaleKeyHeader = getHeader(event, "sb-rm-stale-key");
   const signal = toWebRequest(event).signal;
   const { rmStaleKey } = useRuntimeConfig(event);
@@ -18,40 +15,65 @@ export default eventHandler(async (event) => {
 
   const result = await iterateAndDelete(event, signal, {
     prefix: bucket === 'packages' ? usePackagesBucket.base : useTemplatesBucket.base,
-    limit: 100,
+    limit: 1000,
     cursor: cursor || undefined,
   }, remove);
 
+  setResponseHeader(event, "Content-Type", "application/json");
+
   return {
     result,
   };
 });
 
+// Helper for concurrency limiting
+async function mapWithConcurrency<T, R>(items: T[], concurrency: number, fn: (item: T) => Promise<R>): Promise<R[]> {
+  const results: R[] = [];
+  let i = 0;
+  const executing: Promise<void>[] = [];
+
+  async function enqueue(item: T) {
+    const result = await fn(item);
+    results.push(result);
+  }
+
+  while (i < items.length) {
+    const item = items[i++];
+    const p = enqueue(item);
+    executing.push(p.then(() => {
+      executing.splice(executing.indexOf(p), 1);
+    }));
+    if (executing.length >= concurrency) {
+      await Promise.race(executing);
+    }
+  }
+  await Promise.all(executing);
+  return results;
+}
+
 async function iterateAndDelete(event: H3Event, signal: AbortSignal, opts: R2ListOptions, remove: boolean) {
   const binding = useBinding(event);
   let truncated = true;
-  let cursor: string | undefined;
+  let cursor: string | undefined = opts.cursor;
+  let processed = 0;
   const removedItems: Array<{ key: string; uploaded: Date; downloadedAt?: Date }> = [];
   const downloadedAtBucket = useDownloadedAtBucket(event);
   const today = Date.parse(new Date().toString());
+  const CONCURRENCY = 10;
 
   while (truncated && !signal.aborted) {
-    if (removedItems.length >= 1000) {
-      break
+    if (removedItems.length >= 100 || processed >= 10000) {
+      break;
     }
     const next = await binding.list({
       ...opts,
       cursor,
     });
-    for (const object of next.objects) {
-      if (signal.aborted) {
-        break;
-      }
+    processed += next.objects.length;
+
+    await mapWithConcurrency(next.objects, CONCURRENCY, async (object) => {
+      if (signal.aborted) return;
       const uploaded = Date.parse(object.uploaded.toString());
-      // removedItems.push({
-      //   key: object.key,
-      //   uploaded: new Date(object.uploaded),
-      // });
       const uploadedDate = new Date(uploaded);
       const sixMonthsAgo = new Date(today);
       sixMonthsAgo.setMonth(sixMonthsAgo.getMonth() - 6);
@@ -64,12 +86,11 @@ async function iterateAndDelete(event: H3Event, signal: AbortSignal, opts: R2Lis
           await binding.delete(object.key);
           await downloadedAtBucket.removeItem(object.key);
         }
-        continue;
+        return;
       }
       const downloadedAt = await downloadedAtBucket.getItem(object.key);
-
       if (!downloadedAt) {
-        continue;
+        return;
       }
       const downloadedAtDate = new Date(downloadedAt);
       const oneMonthAgo = new Date(today);
@@ -88,9 +109,9 @@ async function iterateAndDelete(event: H3Event, signal: AbortSignal, opts: R2Lis
           await binding.delete(object.key);
           await downloadedAtBucket.removeItem(object.key);
         }
-
       }
-    }
+    });
+
     truncated = next.truncated;
     if (next.truncated) {
       cursor = next.cursor;
@@ -103,3 +124,4 @@ async function iterateAndDelete(event: H3Event, signal: AbortSignal, opts: R2Lis
   };
 }
 
+

script/remove-stale.js

@@ -18,6 +18,7 @@ async function processBucket(bucket) {
       remove,
     };
     try {
+      console.log(`[${bucket}] Batch ${batch} - Fetching...`);
       const res = await fetch(endpoint, {
         method: 'POST',
         headers: {
@@ -26,26 +27,24 @@ async function processBucket(bucket) {
         },
         body: JSON.stringify(body),
       });
-      const text = await res.text();
-      let json;
-      try {
-        json = JSON.parse(text);
-      } catch (e) {
-        console.error(`[${bucket}] Batch ${batch} - Failed to parse response:`, text);
-        process.exit(1);
-      }
       if (!res.ok) {
         console.error(`[${bucket}] Batch ${batch} - Request failed:`, json);
         process.exit(1);
       }
+
+      const json = await res.json();
+
       console.log(`[${bucket}] Batch ${batch} - Removed items:`, json.result.removedItems.length);
       if (json.result.removedItems.length > 0) {
         for (const item of json.result.removedItems) {
           console.log(`  -`, item);
         }
       }
+      console.log('just fetched', json.result.removedItems.length)
       cursor = json.result.cursor;
       truncated = json.result.truncated;
+      console.log('cursor', cursor)
+      console.log('truncated', truncated)
       batch++;
       if (!truncated) {
         console.log(`[${bucket}] Completed. Total batches: ${batch}`);
@@ -58,8 +57,10 @@ async function processBucket(bucket) {
 }
 
 (async () => {
-  for (const bucket of ['packages', 'templates']) {
+  // for (const bucket of ['packages', 'templates']) {
+  for (const bucket of ['packages']) {
     console.log(`Processing bucket: ${bucket}`);
     await processBucket(bucket);
   }
+  process.exit(0);
 })();