From 12152fc893c6af67625b626ec056b11c157b1bc4 Mon Sep 17 00:00:00 2001 From: Dream Hunter Date: Sun, 9 Aug 2026 23:32:05 +0800 Subject: [PATCH] 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 | 3 + CHANGELOG_EN.md | 3 + e2e/fixtures/wrangler.toml.e2e | 1 + e2e/tests/api/cleanup.spec.ts | 104 ++++++++++++++++++++ vitepress-docs/docs/en/guide/worker-vars.md | 1 + vitepress-docs/docs/zh/guide/worker-vars.md | 1 + worker/src/common.ts | 30 ++++-- worker/src/types.d.ts | 1 + worker/wrangler.toml.template | 2 + 9 files changed, 140 insertions(+), 6 deletions(-) create mode 100644 e2e/tests/api/cleanup.spec.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index bce6e63..7bc7242 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/CHANGELOG_EN.md b/CHANGELOG_EN.md index 0f0b723..1c91ec0 100644 --- a/CHANGELOG_EN.md +++ b/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 diff --git a/e2e/fixtures/wrangler.toml.e2e b/e2e/fixtures/wrangler.toml.e2e index 5751441..b98b0c9 100644 --- a/e2e/fixtures/wrangler.toml.e2e +++ b/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}} """ diff --git a/e2e/tests/api/cleanup.spec.ts b/e2e/tests/api/cleanup.spec.ts new file mode 100644 index 0000000..59c82db --- /dev/null +++ b/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: ``, + }, + }) + )); + 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: ``, + }, + }); + 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: ``, + }, + }); + 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); + } + }); +}); diff --git a/vitepress-docs/docs/en/guide/worker-vars.md b/vitepress-docs/docs/en/guide/worker-vars.md index c1f85ff..9a9ff24 100644 --- a/vitepress-docs/docs/en/guide/worker-vars.md +++ b/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` diff --git a/vitepress-docs/docs/zh/guide/worker-vars.md b/vitepress-docs/docs/zh/guide/worker-vars.md index 9f96bd9..d1fe34e 100644 --- a/vitepress-docs/docs/zh/guide/worker-vars.md +++ b/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` diff --git a/worker/src/common.ts b/worker/src/common.ts index 820c3b4..25675f7 100644 --- a/worker/src/common.ts +++ b/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')` - ).run(); + DELETE FROM raw_mails WHERE id IN ( + 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')` - ).run(); + DELETE FROM sendbox WHERE id IN ( + 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 diff --git a/worker/src/types.d.ts b/worker/src/types.d.ts index f0a6116..ce194e8 100644 --- a/worker/src/types.d.ts +++ b/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 diff --git a/worker/wrangler.toml.template b/worker/wrangler.toml.template index 0109651..f57e44d 100644 --- a/worker/wrangler.toml.template +++ b/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" },