perf: limit indexed cleanup task batches (#1107)

Limit mail, sent-mail, and indexed address cleanup to configurable batches. Includes E2E coverage and documentation.

CHANGELOG.md

@@ -18,9 +18,12 @@
 
 - feat: |用户系统| 用户绑定地址列表改用服务端分页，并仅在第一页查询总数；用户邮件列表改用 JOIN、删除改用 `EXISTS` 在数据库侧校验地址归属，避免为大用户加载全部绑定地址（issue #1103）
 
+- feat: |Worker| 邮件、发件箱及按创建/活跃时间清理地址时改为分批处理，默认每次最多 3000 条并支持通过 `CLEANUP_BATCH_SIZE` 调整（上限 5000），减少单次扫描和删除量（issue #1103）
+
 ### Testing
 
 - fix: |E2E| 新增近期地址活跃时间不会被用户设置接口重复写入的回归测试
+- fix: |E2E| 新增清理批次上限、后续批次继续执行、保留未过期数据及地址关联数据清理测试
 
 ## v1.10.0
 

CHANGELOG_EN.md

@@ -18,9 +18,12 @@
 
 - feat: |User| Add server-side pagination for bound addresses, with totals queried only on the first page; validate user-mail list ownership with a JOIN and delete ownership with `EXISTS` instead of loading every bound address for large users (issue #1103)
 
+- feat: |Worker| Process mail, sent-mail, and creation/activity-based address cleanup in batches of 3000 by default, configurable through `CLEANUP_BATCH_SIZE` up to 5000, reducing per-run scans and deletes (issue #1103)
+
 ### Testing
 
 - fix: |E2E| Add regression coverage ensuring user settings do not rewrite recent address activity timestamps
+- fix: |E2E| Cover cleanup batch limits, continuation on later runs, preservation of recent data, and address-related data cleanup
 
 ## v1.10.0
 

e2e/fixtures/wrangler.toml.e2e

@@ -23,6 +23,7 @@ DISABLE_ADMIN_PASSWORD_CHECK = true
 ADMIN_PASSWORDS = '["e2e-admin-pass"]'
 ENABLE_WEBHOOK = true
 E2E_TEST_MODE = true
+CLEANUP_BATCH_SIZE = 10
 SMTP_CONFIG = """
 {"TEST.EXAMPLE.COM":{"host":"mailpit","port":1025,"secure":false}}
 """

e2e/tests/api/cleanup.spec.ts

@@ -0,0 +1,104 @@
+import { test, expect, type APIRequestContext } from '@playwright/test';
+import { WORKER_URL, createTestAddress, deleteAddress } from '../../fixtures/test-helpers';
+
+const listMails = async (request: APIRequestContext, address: string) => {
+  const response = await request.get(`${WORKER_URL}/admin/mails`, {
+    params: { address, limit: '100', offset: '0' },
+  });
+  expect(response.ok()).toBe(true);
+  return response.json();
+};
+
+test.describe('Bounded cleanup', () => {
+  test('cleans at most one batch and continues on the next run', async ({ request }) => {
+    const address = `cleanup-batch-${Date.now()}@test.example.com`;
+    const seedResponses = await Promise.all(Array.from({ length: 11 }, (_, index) =>
+      request.post(`${WORKER_URL}/admin/test/seed_mail`, {
+        data: {
+          address,
+          raw: 'old cleanup mail',
+          message_id: `<cleanup-old-${Date.now()}-${index}@test>`,
+        },
+      })
+    ));
+    expect(seedResponses.every((response) => response.ok())).toBe(true);
+    await new Promise((resolve) => setTimeout(resolve, 1100));
+
+    const firstCleanup = await request.post(`${WORKER_URL}/admin/cleanup`, {
+      data: { cleanType: 'mails', cleanDays: 0 },
+    });
+    expect(firstCleanup.ok()).toBe(true);
+
+    const afterFirstCleanup = await listMails(request, address);
+    expect(afterFirstCleanup.count).toBe(1);
+
+    const secondCleanup = await request.post(`${WORKER_URL}/admin/cleanup`, {
+      data: { cleanType: 'mails', cleanDays: 0 },
+    });
+    expect(secondCleanup.ok()).toBe(true);
+
+    const afterSecondCleanup = await listMails(request, address);
+    expect(afterSecondCleanup.count).toBe(0);
+
+    const recentMailResponse = await request.post(`${WORKER_URL}/admin/test/seed_mail`, {
+      data: {
+        address,
+        raw: 'recent cleanup mail',
+        message_id: `<cleanup-recent-${Date.now()}@test>`,
+      },
+    });
+    expect(recentMailResponse.ok()).toBe(true);
+
+    const recentCleanup = await request.post(`${WORKER_URL}/admin/cleanup`, {
+      data: { cleanType: 'mails', cleanDays: 1 },
+    });
+    expect(recentCleanup.ok()).toBe(true);
+
+    const recentMails = await listMails(request, address);
+    expect(recentMails.count).toBe(1);
+    expect(recentMails.results[0].raw).toBe('recent cleanup mail');
+    await request.delete(`${WORKER_URL}/admin/mails/${recentMails.results[0].id}`);
+  });
+
+  test('deletes one address batch and its related data', async ({ request }) => {
+    const oldAddress = await createTestAddress(request, 'cleanup-old');
+
+    const seedResponse = await request.post(`${WORKER_URL}/admin/test/seed_mail`, {
+      data: {
+        address: oldAddress.address,
+        raw: 'address cleanup mail',
+        message_id: `<cleanup-address-${Date.now()}@test>`,
+      },
+    });
+    expect(seedResponse.ok()).toBe(true);
+    await new Promise((resolve) => setTimeout(resolve, 1100));
+
+    const cleanupResponse = await request.post(`${WORKER_URL}/admin/cleanup`, {
+      data: { cleanType: 'addressCreated', cleanDays: 0 },
+    });
+    expect(cleanupResponse.ok()).toBe(true);
+
+    const oldAddressResponse = await request.get(`${WORKER_URL}/admin/address`, {
+      params: { query: oldAddress.address, limit: '20', offset: '0' },
+    });
+    expect(oldAddressResponse.ok()).toBe(true);
+    expect((await oldAddressResponse.json()).count).toBe(0);
+    expect((await listMails(request, oldAddress.address)).count).toBe(0);
+
+    const recentAddress = await createTestAddress(request, 'cleanup-recent');
+    try {
+      const recentCleanup = await request.post(`${WORKER_URL}/admin/cleanup`, {
+        data: { cleanType: 'addressCreated', cleanDays: 1 },
+      });
+      expect(recentCleanup.ok()).toBe(true);
+
+      const recentAddressResponse = await request.get(`${WORKER_URL}/admin/address`, {
+        params: { query: recentAddress.address, limit: '20', offset: '0' },
+      });
+      expect(recentAddressResponse.ok()).toBe(true);
+      expect((await recentAddressResponse.json()).count).toBe(1);
+    } finally {
+      await deleteAddress(request, recentAddress.jwt);
+    }
+  });
+});

vitepress-docs/docs/en/guide/worker-vars.md

@@ -103,6 +103,7 @@
 | `REMOVE_EXCEED_SIZE_ATTACHMENT` | Text/JSON | If attachment exceeds 2MB, remove it, email may lose some information due to parsing                                   | `true`                     |
 | `REMOVE_ALL_ATTACHMENT`         | Text/JSON | Remove all attachments, email may lose some information due to parsing                                                 | `true`                     |
 | `ENABLE_MAIL_GZIP`             | Text/JSON | When enabled, new emails are gzip-compressed and stored in `raw_blob` column to save D1 database space. Existing plaintext `raw` data is automatically compatible for reading. **Run database migration first (`Admin -> Quick Setup -> Database -> Migrate Database` or `POST /admin/db_migration`) to ensure the `raw_blob` column exists before enabling. This feature adds compression/decompression CPU overhead, so enabling it on a paid Cloudflare Worker plan is recommended.** | `true`                     |
+| `CLEANUP_BATCH_SIZE`           | Number    | Per-run limit for mail, sent-mail, and creation/activity-based address cleanup. Defaults to `3000`, valid range `1-5000`. Smaller values reduce per-run D1 pressure; larger values clear backlogs faster | `3000` |
 
 > [!NOTE]
 > Authentication results follow their standards: SPF `none` means no usable domain or SPF record was found, and SPF `neutral` must be treated like `none`; DKIM `none` means the message was unsigned, and DKIM `neutral` is also treated as unsigned; DMARC `none` means no applicable DMARC policy was found. Unregistered results and unsupported method versions are ignored. `JUNK_MAIL_CHECK_LIST` treats these results as absent, while `JUNK_MAIL_FORCE_PASS_LIST` still requires an explicit supported `pass`

vitepress-docs/docs/zh/guide/worker-vars.md

@@ -97,6 +97,7 @@
 | `REMOVE_EXCEED_SIZE_ATTACHMENT` | 文本/JSON | 如果附件大小超过 2MB，则删除附件，邮件可能由于解析而丢失一些信息           | `true`                     |
 | `REMOVE_ALL_ATTACHMENT`         | 文本/JSON | 移除所有附件，邮件可能由于解析而丢失一些信息                               | `true`                     |
 | `ENABLE_MAIL_GZIP`             | 文本/JSON | 启用后新邮件将 Gzip 压缩存储到 `raw_blob` 字段，可节省 D1 数据库空间。已有明文 `raw` 数据自动兼容读取。**启用前请先执行数据库迁移（`Admin -> 快速设置 -> 数据库 -> 升级数据库 Schema` 或 `POST /admin/db_migration`），确保 `raw_blob` 列已创建。该功能会增加压缩/解压 CPU 开销，建议使用 Cloudflare Worker 付费 Plan 再开启。** | `true`                     |
+| `CLEANUP_BATCH_SIZE`           | 数字      | 邮件、发件箱及按创建/活跃时间清理地址时的单次处理上限，默认 `3000`，有效范围 `1-5000`。较小值可降低单次 D1 压力，较大值可加快积压数据清理 | `3000` |
 
 > [!NOTE]
 > 认证结果遵循各自规范：SPF `none` 表示没有可检查的域名或 SPF 记录，SPF `neutral` 必须与 `none` 相同处理；DKIM `none` 表示邮件未签名，DKIM `neutral` 同样按未签名处理；DMARC `none` 表示没有适用的 DMARC 策略。未注册结果和不支持的方法版本也会被忽略。`JUNK_MAIL_CHECK_LIST` 将这些结果视为认证方法不存在，`JUNK_MAIL_FORCE_PASS_LIST` 仍只接受明确且受支持的 `pass`

worker/src/common.ts

@@ -494,18 +494,26 @@ export const cleanup = async (
     if (!cleanType || typeof cleanDays !== 'number' || cleanDays < 0 || cleanDays > 1000) {
         throw new Error(msgs.InvalidCleanupConfigMsg)
     }
+    let cleanupBatchSize = getIntValue(c.env.CLEANUP_BATCH_SIZE, 3000);
+    if (!Number.isInteger(cleanupBatchSize) || cleanupBatchSize < 1 || cleanupBatchSize > 5000) {
+        cleanupBatchSize = 3000;
+    }
     console.log(`Cleanup ${cleanType} before ${cleanDays} days`);
     switch (cleanType) {
         case "inactiveAddress":
             await batchDeleteAddressWithData(
                 c,
-                `updated_at < datetime('now', '-${cleanDays} day')`
+                `id IN (`
+                + `SELECT id FROM address WHERE updated_at < datetime('now', '-${cleanDays} day') `
+                + `ORDER BY updated_at, id LIMIT ${cleanupBatchSize})`
             )
             break;
         case "addressCreated":
             await batchDeleteAddressWithData(
                 c,
-                `created_at < datetime('now', '-${cleanDays} day')`
+                `id IN (`
+                + `SELECT id FROM address WHERE created_at < datetime('now', '-${cleanDays} day') `
+                + `ORDER BY created_at, id LIMIT ${cleanupBatchSize})`
             )
             break;
         case "unboundAddress":
@@ -516,8 +524,13 @@ export const cleanup = async (
             break;
         case "mails":
             await c.env.DB.prepare(`
-                DELETE FROM raw_mails WHERE created_at < datetime('now', '-${cleanDays} day')`
+                DELETE FROM raw_mails WHERE id IN (
-            ).run();
+                    SELECT id FROM raw_mails
+                    WHERE created_at < datetime('now', ?)
+                    ORDER BY created_at, id
+                    LIMIT ?
+                )`
+            ).bind(`-${cleanDays} day`, cleanupBatchSize).run();
             break;
         case "mails_unknow":
             await c.env.DB.prepare(`
@@ -527,8 +540,13 @@ export const cleanup = async (
             break;
         case "sendbox":
             await c.env.DB.prepare(`
-                DELETE FROM sendbox WHERE created_at < datetime('now', '-${cleanDays} day')`
+                DELETE FROM sendbox WHERE id IN (
-            ).run();
+                    SELECT id FROM sendbox
+                    WHERE created_at < datetime('now', ?)
+                    ORDER BY created_at, id
+                    LIMIT ?
+                )`
+            ).bind(`-${cleanDays} day`, cleanupBatchSize).run();
             break;
         case "emptyAddress":
             // Delete addresses that have no emails and were created more than N days ago

worker/src/types.d.ts

@@ -117,6 +117,7 @@ type Bindings = {
 
     // gzip compression for raw_mails
     ENABLE_MAIL_GZIP: string | boolean | undefined
+    CLEANUP_BATCH_SIZE: string | number | undefined
 
     // E2E testing
     E2E_TEST_MODE: string | boolean | undefined

worker/wrangler.toml.template

@@ -17,6 +17,8 @@ keep_vars = true
 # enable cron if you want set auto clean up
 # [triggers]
 # crons = [ "0 0 * * *" ]
+# Per-run limit for mail, sent-mail, and creation/activity-based address cleanup. Defaults to 3000, maximum 5000.
+# CLEANUP_BATCH_SIZE = 3000
 
 # send_email = [
 #    { name = "SEND_MAIL" },