diff --git a/docs/notification-email.md b/docs/notification-email.md index 5bfabbc3a..0f81c4178 100644 --- a/docs/notification-email.md +++ b/docs/notification-email.md @@ -151,4 +151,6 @@ Chart 配置与安装方式见 [Helm 手册](../charts/skillhub/README.md)。渲 新入箱仅接受发生时间距今七天内的事件；旧事件重放不能在去重记录清理后重新发信。正常投递重试和人工重试复用原记录，不重新入箱。已发送及抑制记录默认保留 30 天，失败及未知默认 90 天；未完成队列不按终态记录清理，清理后不能恢复或重试该批次。 +到期历史按有限批次逐步清理。恢复组开始清理后即视为过期，用户明细和管理员投递列表不再显示，也不能人工重试；物理记录可能在后续多轮才全部删除。清理中断后继续原进度，不会因为重启而恢复部分已删除的邮件内容。分页限制约束单轮写入规模，不代表任意数据量下都有固定清理耗时。 + 回退应用先关闭邮件、等待 worker 结束并保留表。旧应用不维护邮箱修订号，因此回退前须由受控维护操作失效全部通知验邮证明/挑战、抑制未发项；重新升级保持关闭，用户重新验邮和订阅。不得仅回退镜像后恢复旧验邮证明，也不要删除业务或站内通知数据来“回滚”。具体维护命令应由部署者针对已验证的版本与数据库备份执行。 diff --git a/openspec/changes/add-notification-email-delivery/design.md b/openspec/changes/add-notification-email-delivery/design.md index ab225cb45..537496cba 100644 --- a/openspec/changes/add-notification-email-delivery/design.md +++ b/openspec/changes/add-notification-email-delivery/design.md @@ -134,6 +134,8 @@ flowchart TD **取消与保留：** 平台关闭、类别 OFF、账号停用等取消必须覆盖 PREPARING 父与 MERGED 源的未授权项，恢复开关不能使源项复活。人工重试 UNKNOWN 仍须重复风险确认；恢复合并不能借此自动重试 UNKNOWN。清理按整个恢复组的完成状态处理，未完成父及其来源和去重事实不得先删；失败与未知的保留限制仍适用。 +**历史清理的有界事务：** 仅选取整组已终结且满足保留期限的 root，持有 root 行锁后写入内部 `purge_started_at` 标记；该结构字段不新增业务状态或用户配置。标记后的 root 及其来源对用户明细、管理列表和人工重试按过期不可用处理，避免重试复活已部分删除的内容。每轮最多选择 10 个 root，每个最多删除 100 条 item、100 条已无 item 的来源；只有两者均为空才删除 root，不能依赖无界级联。尚未清完的组轮转到后续轮次，进程重启继续读取持久标记，不要求单事务清空整个组。清理轮转复用被标记终态行的 `next_attempt_at` 排序，不改变未清理任务的投递时间或重试截止时间；真实 PostgreSQL 用例须验证单轮上限、多个组轮转、部分清理后重试拒绝及中断恢复。 + **未来尝试取消屏障：** 在现有策略→账号→contact→偏好锁序内，平台关闭、类别 OFF、账号停用的同一短事务写持久取消边界。边界记录覆盖当时已入箱的 item，包括父批次处于 SENDING、来源已 MERGED 的项；重新开启不清除边界。已授权在途尝试保持 SENDING、租约、payload 与最终 ACCEPTED/UNKNOWN/TEMP 记账，只有后续准备与授权按有限候选 item 检查边界；TEMP 回 RETRY 也不得让旧项再获授权。关闭不能在全局独占锁内全量载入批次或一次更新所有 item；独立扫尾按有限页逐步将命中项持久抑制，扫尾重启可续，期间由边界保证安全。item 的数据库 ID 边界必须与入箱和关闭事务锁序一致，并用真实 PostgreSQL 竞争测试验证未提交入箱不会落到屏障之外。普通 SMTP 故障恢复不产生关闭屏障；接纳水位只处理新入箱迟到事件，不代替此取消屏障。 恢复父批次因无合格项或关闭而终结时，只更新有条件租约的批次状态；不得随之一次更新所有来源项。有限页扫尾负责物理项状态，终态父和取消屏障在此期间作为详情与再次授权的逻辑过滤条件；已 SENT 的历史按原样保留。人工重试不清除取消边界，只能处理未被边界取消的新合法项。 diff --git a/openspec/changes/add-notification-email-delivery/specs/notification-email-delivery/spec.md b/openspec/changes/add-notification-email-delivery/specs/notification-email-delivery/spec.md index 606d6b6fd..ae2717b3b 100644 --- a/openspec/changes/add-notification-email-delivery/specs/notification-email-delivery/spec.md +++ b/openspec/changes/add-notification-email-delivery/specs/notification-email-delivery/spec.md @@ -130,9 +130,11 @@ IMMEDIATE SHALL 表示不等待摘要窗口，按独立通知实例尽快异步 平台关闭、类别关闭与账号停用的取消记录 SHALL 在既有授权锁顺序下短事务提交；待发积压可随后按有限页扫尾。快速关闭再开启及扫尾中断均不得绕过持久取消边界。 +到期历史清理 SHALL 同时限制每轮处理的恢复组、邮件项和来源批次数量；不得仅限制 root 数量后通过级联删除一次清空巨型恢复组。开始分批清理的组 SHALL 持久标记为不可读取及不可人工重试，后续轮次可在中断后继续；仅在项及来源均清空后删除 root。未完成组和未达到保留期限的记录不得进入清理，多个到期组应轮转推进。 + #### Scenario: SCN-030 慢供应商、巨批次及公平性 -- **WHEN** SMTP 阻塞、同用户大量重复触发，或存在海量摘要源/项及其他用户的即时通知 -- **THEN** 公共定时任务继续运行，执行器及领取数有界，无无界内存队列；准备分页提交且其他用户可推进，即时/摘要都有处理机会，明细不全量加载后假分页；邮件原始数量不冒称精确可见总数 +- **WHEN** SMTP 阻塞、同用户大量重复触发，或存在海量摘要源/项及其他用户的即时通知，或多个历史恢复组达到清理期限 +- **THEN** 公共定时任务继续运行，执行器及领取数有界，无无界内存队列；准备及历史清理均分页提交且其他用户或恢复组可推进，即时/摘要都有处理机会，明细不全量加载后假分页；邮件原始数量不冒称精确可见总数；清理中断后可续跑，部分清理历史不能被读取或重试复活 ### Requirement: REQ-016 临时故障恢复期限 diff --git a/openspec/changes/add-notification-email-delivery/tasks.md b/openspec/changes/add-notification-email-delivery/tasks.md index faae90fb3..7ab42e844 100644 --- a/openspec/changes/add-notification-email-delivery/tasks.md +++ b/openspec/changes/add-notification-email-delivery/tasks.md @@ -1,36 +1,39 @@ +本任务保留实现项与最终交付项的区别：已勾选项有对应实现和已执行检查；历史清理修复、最新提交部署及最终验收仍未放行。真实企业邮箱收件、旧应用镜像回退及生产部署不因本地替身检查而记为完成。 + ## 1. 方案批准与契约冻结 - [x] 1.1 评审 proposal/design/两个规格及 verification，确认类别共用、摘要周期、默认关闭、入口 best-effort 与 UNKNOWN 取舍；完成标准为批准记录及所有影响实现的分歧已回写 OpenSpec。 -- [ ] 1.2 枚举每个现有通知事件的收件人、类别、发生实例去重键和权限规则；以同事件重放/同技能不同操作的对照测试证明不会漏通知。 -- [ ] 1.3 确定新增 API/设置/表结构与错误契约；用 OpenAPI 校验、并发迁移编号检查及未新增 SQL CHECK 的审查通过证明完成。 +- [x] 1.2 枚举每个现有通知事件的收件人、类别、发生实例去重键和权限规则；以同事件重放/同技能不同操作的对照测试证明不会漏通知。 +- [x] 1.3 确定新增 API/设置/表结构与错误契约；用 OpenAPI 校验、并发迁移编号检查及未新增 SQL CHECK 的审查通过证明完成。 ## 2. 收件资格与设置 -- [ ] 2.1 实现邮箱修订号，覆盖外部登录、合并和初始化写入，验证 A→B→A、同地址重存及并发更新不接受旧证明。 -- [ ] 2.2 实现通知专用验邮及限流，验证过期、错误上限、重发竞争、消费一次、SMTP 故障和跨用户/用途拒绝。 -- [ ] 2.3 实现独立邮件偏好和系统配置组，验证四类×三模式、缺省 OFF、普通用户越权及现有 IN_APP API 完全兼容。 +- [x] 2.1 实现邮箱修订号，覆盖外部登录、合并和初始化写入，验证 A→B→A、同地址重存及并发更新不接受旧证明。 +- [x] 2.2 实现通知专用验邮及限流，验证过期、错误上限、重发竞争、消费一次、SMTP 故障和跨用户/用途拒绝。 +- [x] 2.3 实现独立邮件偏好和系统配置组，验证四类×三模式、缺省 OFF、普通用户越权及现有 IN_APP API 完全兼容。 ## 3. 入箱、摘要与投递 -- [ ] 3.1 抽取共享通知命令/收件解析并拆开渠道事务，原站内监听流程回归通过；故障注入证明邮件分支失败不阻断站内及已提交业务。 -- [ ] 3.2 实现稳定事件身份/发生时间、有限历史接纳与结构唯一键，真实 PostgreSQL 测试验证窗口内重复只一项、不同版本/操作不丢失，过期重放被拒且保留期不会提前移除去重事实。 -- [ ] 3.3 实现窗口附加与封口锁，验证多技能跨类别一封、空窗口、上限不拆信、迟到及封口并发无丢失/重开。 -- [ ] 3.4 实现 CLAIMED 与 SENDING 授权线性点、统一锁顺序、事务外 SMTP、有限重试/UNKNOWN 与受限人工重试，双 worker、资格变更两侧屏障、过期 token 和 SMTP 捕获器故障测试通过。 -- [ ] 3.5 实现发送前邮箱/权限/订阅复核与安全模板，验证换址、停用、退出、退订、删除、XSS/链接注入和同邮箱不同账号隔离。 +- [x] 3.1 抽取共享通知命令/收件解析并拆开渠道事务，原站内监听流程回归通过；故障注入证明邮件分支失败不阻断站内及已提交业务。 +- [x] 3.2 实现稳定事件身份/发生时间、有限历史接纳与结构唯一键，真实 PostgreSQL 测试验证窗口内重复只一项、不同版本/操作不丢失，过期重放被拒且保留期不会提前移除去重事实。 +- [x] 3.3 实现窗口附加与封口锁，验证多技能跨类别一封、空窗口、上限不拆信、迟到及封口并发无丢失/重开。 +- [x] 3.4 实现 CLAIMED 与 SENDING 授权线性点、统一锁顺序、事务外 SMTP、有限重试/UNKNOWN 与受限人工重试，双 worker、资格变更两侧屏障、过期 token 和 SMTP 捕获器故障测试通过。 +- [x] 3.5 实现发送前邮箱/权限/订阅复核与安全模板，验证换址、停用、退出、退订、删除、XSS/链接注入和同邮箱不同账号隔离。 - [ ] 3.6 实现独立清理、分页脱敏投递记录和失败指标；证明未完成任务不被清理，日志无验证码/密钥/正文。 ## 4. 用户交互与部署文档 -- [ ] 4.1 在通知设置加入验邮状态和四类模式选择，现有站内列保持不变；增加独立邮件批次明细视图及分页，验证站内OFF/已删仍能查看有权明细、过期和跨用户拒绝；组件及浏览器验证同时覆盖不可用、保存失败和类别说明。 -- [ ] 4.2 在系统配置加入普通邮件设置，不显示 SMTP 密钥；生成 API 类型及中英文资源检查通过。 +- [x] 4.1 在通知设置加入验邮状态和四类模式选择，现有站内列保持不变；增加独立邮件批次明细视图及分页，验证站内OFF/已删仍能查看有权明细、过期和跨用户拒绝；组件及浏览器验证同时覆盖不可用、保存失败和类别说明。 +- [x] 4.2 在系统配置加入普通邮件设置，不显示 SMTP 密钥；生成 API 类型及中英文资源检查通过。 - [ ] 4.3 补充 Compose/Helm/环境参数、SMTP 超时/发件人、摘要示例、UNKNOWN/重试和故障排查说明；按实际镜像和本地捕获器复测文档命令。 ## 4a. 积压恢复补充（用户指出后新增） -- [ ] 4a.1 邮件调度与公共定时任务隔离；以阻塞 SMTP、重复触发和关闭屏障证明其他任务不中断、没有无界任务堆积。 -- [ ] 4a.2 完成到期摘要按用户恢复合并的状态、来源追溯与并发方案；实现后验证多实例只发送一份恢复汇总，未知与已授权批次不自动重发。 +- [x] 4a.1 邮件调度与公共定时任务隔离；以阻塞 SMTP、重复触发和关闭屏障证明其他任务不中断、没有无界任务堆积。 +- [x] 4a.2 完成到期摘要按用户恢复合并的状态、来源追溯与并发方案；实现后验证多实例只发送一份恢复汇总，未知与已授权批次不自动重发。 - [ ] 4a.3 明确并实现有界准备、数据库分页、即时/摘要调度与容量可观测性；大批次、数小时故障及恢复场景通过前不放行。 -- [ ] 4a.4 关闭事务仅写持久取消边界，后台有限页扫尾；真实 PostgreSQL 验证入箱 ID 已分配未提交的竞争、三类关闭范围及 SENDING→TEMP 后快速恢复不会给旧项第二次授权，新合法项仍可发送。 +- [x] 4a.4 关闭事务仅写持久取消边界，后台有限页扫尾；真实 PostgreSQL 验证入箱 ID 已分配未提交的竞争、三类关闭范围及 SENDING→TEMP 后快速恢复不会给旧项第二次授权，新合法项仍可发送。 +- [ ] 4a.5 严格复核修复：历史清理采用持久内部标记及 root/item/source 有限分页，验证单轮上限、超过单轮 root 数的轮转、部分清理后读取及重试拒绝、中断续跑和保留期限；补真实准备中进程重启，不能以新 dispatcher 实例代替服务重启。 ## 5. 验收与交付 diff --git a/openspec/changes/add-notification-email-delivery/verification.md b/openspec/changes/add-notification-email-delivery/verification.md index bba361a7a..3774652a5 100644 --- a/openspec/changes/add-notification-email-delivery/verification.md +++ b/openspec/changes/add-notification-email-delivery/verification.md @@ -1,6 +1,6 @@ # #978 验证设计与追踪矩阵 -**状态：实施中，正式验收矩阵尚未冻结执行。** 基础实现及开发反馈测试已有结果；积压恢复需求补充后仍在修改。本表是正式验收依据，不把开发中测试、静态审查或旧 SHA 结果冒称为全部验收通过。精确 SHA 的部署、SMTP 捕获器和浏览器验收须在实现与独立审查收敛后记录。 +**状态：严格验收复核中（2026-10-22），暂不可合并。** 已完成一轮固定实现提交的源测试、真实 PostgreSQL、Compose、SMTP 捕获器及浏览器执行；独立复核发现历史清理事务无界、部分租约边界及准备中真实重启证据不足，因此撤回整体验收通过结论。租约边界已补真实数据库验证，其余修复和补验继续进行。详细原始证据保留在本地验收记录，不将私人环境或凭据写入公开仓库；新提交的最终结果须重新绑定实际版本。 ## 1. 源码基线与事实入口 @@ -26,7 +26,7 @@ ## 2. 正常流程与变更流程 -下表每一行均须在实施后保存请求/操作、数据、原始结果及最终状态；同方法的数据变体可合并执行，但不能把正常回归借作新功能证据。所有状态当前均为 **计划/未执行**。 +下表定义验收判据；逐项执行状态由固定提交的验收记录维护。同方法的数据变体可合并执行，但不能把正常回归借作新功能证据，也不能以“过期之后”一次断言代替“之前／恰好到期／之后”三个边界。受控状态或旧写入方式替身不等同于真实进程崩溃、重启或旧应用镜像回退。 | 用例 | 轨道/场景 | 需求 | 数据和动作 | 独立判据/最终状态 | 验证层 | |---|---|---|---|---|---| @@ -114,7 +114,7 @@ | TC-023 | REQ-014 | 同用户 4/1000 个到期窗口、多实例同时 CLAIMED、恢复中继续来新事件、两个用户同邮箱、不同 emailRevision | 固定范围只一封恢复汇总；原事件时间和来源不变；用户/修订隔离；旧 CLAIMED 不能独立发；即时不混入 | | TC-024 | REQ-014/REQ-004/REQ-013；SCN-033 | 源封一半后崩溃；PREPARING 租约到期；SENDING 阻塞时平台/类别 OFF→ON 或账号停用→恢复，随后 SMTP 返回明确 TEMP；UNKNOWN/FAILED 后关闭重开并人工 retry；人工 retry 遇另一 root；保留清理 | 从持久游标续同一 root；在途一次仍按结果记账，但取消前旧项不获得第二次自动或人工授权；新合法项可发；UNKNOWN 不自动重发；活跃恢复组与去重事实保留；人工竞争明确冲突 | | TC-025 | REQ-015/REQ-008 | 海量项/全部失权/首有效项在后页、资源在分页与授权之间变化、详情翻页 | 每次最多有限行/有限事务；不全量装内存；无有效项不发空信；最终预览复核；展示原始数而非精确可见数，详情按当前权限过滤 | -| TC-026 | REQ-015/REQ-007 | SMTP 和清理同时阻塞、重复触发千次、其他用户即时与摘要、实例关闭、供应商临时拒绝与 DATA 不确定；巨型恢复组大量待发项下关闭并快速重开，随后批次终结、详情访问及扫尾 | 公共 timer 继续工作；独立执行容量有界且无 caller-runs；公平轮转推进；CLAIMED 与 SENDING 恢复区分；FAILED/UNKNOWN/SENDING/SENT 不被当成普通待发重合并；明确 TEMP 的 RETRY 沿原 root 恢复；关闭事务只记录有界持久屏障，批次终结无全组UPDATE，详情按逻辑状态过滤，有限页扫尾，不因重开或扫尾中断复活旧项 | +| TC-026 | REQ-015/REQ-007 | SMTP 和清理同时阻塞、重复触发千次、其他用户即时与摘要、实例关闭、供应商临时拒绝与 DATA 不确定；巨型恢复组大量待发项下关闭并快速重开，随后批次终结、详情访问及扫尾；到期巨型恢复组及超过单轮上限的多个组，部分清理后中断及人工重试 | 公共 timer 继续工作；独立执行容量有界且无 caller-runs；公平轮转推进；CLAIMED 与 SENDING 恢复区分；FAILED/UNKNOWN/SENDING/SENT 不被当成普通待发重合并；明确 TEMP 的 RETRY 沿原 root 恢复；关闭事务只记录有界持久屏障，批次终结无全组UPDATE，详情按逻辑状态过滤，有限页扫尾，不因重开或扫尾中断复活旧项；真实数据库证明历史清理每轮 root/item/source 上限及轮转，部分清理内容不可读、不可重试复活，未完成及未到保留期组保持不变 | | TC-027 | REQ-016/REQ-014/REQ-007 | 明确临时故障4h持续入箱→恢复；24h截止前/等于/后、重启、补源、人工retry、永久拒绝、UNKNOWN | 同root有界补源并恢复一封；自动路径不延长deadline；截止边界失败且不自动复活；永久/UNKNOWN分类不被恢复窗口覆盖；人工新窗口有审计 | | TC-028 | REQ-017/REQ-004/REQ-013 | 开关水位前/等于/后、关闭期间延迟事件、类别恢复/首次验邮/账号恢复、仅改周期/URL/模式、同事件重放 | 新入箱严格比较原始发生时间，水位不被普通配置改动刷新；旧合法队列不误杀，站内与发生时间不变；多实例持锁初始化只补一次 | diff --git a/server/skillhub-app/src/main/resources/db/migration/V121__notification_email_delivery.sql b/server/skillhub-app/src/main/resources/db/migration/V121__notification_email_delivery.sql index ffc003ada..272342115 100644 --- a/server/skillhub-app/src/main/resources/db/migration/V121__notification_email_delivery.sql +++ b/server/skillhub-app/src/main/resources/db/migration/V121__notification_email_delivery.sql @@ -18,6 +18,7 @@ CREATE TABLE notification_email_delivery ( item_cursor BIGINT NOT NULL DEFAULT 0, items_complete BOOLEAN NOT NULL DEFAULT FALSE, preview_item_ids VARCHAR(1200) NOT NULL DEFAULT '', item_count BIGINT NOT NULL DEFAULT 0, retry_started_at TIMESTAMPTZ, retry_deadline TIMESTAMPTZ, terminal_at TIMESTAMPTZ, + purge_started_at TIMESTAMPTZ, UNIQUE(recipient_id, window_key) ); CREATE INDEX idx_notification_email_due ON notification_email_delivery(status,next_attempt_at,due_at); diff --git a/server/skillhub-app/src/test/java/com/iflytek/skillhub/service/notificationemail/NotificationEmailDeliveryPostgresTest.java b/server/skillhub-app/src/test/java/com/iflytek/skillhub/service/notificationemail/NotificationEmailDeliveryPostgresTest.java index d444da09c..2cc2d8c4c 100644 --- a/server/skillhub-app/src/test/java/com/iflytek/skillhub/service/notificationemail/NotificationEmailDeliveryPostgresTest.java +++ b/server/skillhub-app/src/test/java/com/iflytek/skillhub/service/notificationemail/NotificationEmailDeliveryPostgresTest.java @@ -1076,4 +1076,191 @@ class NotificationEmailDeliveryPostgresTest { assertThat(root.retryStartedAt()).isEqualTo(clock.instant().plusSeconds(901)); assertThat(root.retryDeadline()).isEqualTo(clock.instant().plusSeconds(901 + 86400)); } + private long cleanupFixture(String user, String status, int items, int sources) { + jdbc.update("INSERT INTO notification_email_contact(user_id) VALUES(?) ON CONFLICT DO NOTHING", user); + var row = tx.execute(ignored -> repository.openBatch( + user, "DIGEST", "cleanup", 0, clock.instant(), clock.instant())); + jdbc.update("UPDATE notification_email_delivery SET status=?,terminal_at=? WHERE id=?", status, + java.sql.Timestamp.from(clock.instant()), row.id()); + jdbc.update(""" + INSERT INTO notification_email_delivery(recipient_id,email_revision,kind,window_key,status, + due_at,next_attempt_at,created_at,merged_into_id) + SELECT ?,0,'DIGEST','source-'||n,'MERGED',?,?,?,? FROM generate_series(1,?) n + """, user, java.sql.Timestamp.from(clock.instant()), java.sql.Timestamp.from(clock.instant()), + java.sql.Timestamp.from(clock.instant()), row.id(), sources); + jdbc.update(""" + INSERT INTO notification_email_item(delivery_id,recipient_id,event_key,category,event_type, + occurred_at,enqueued_at) + SELECT ?,?,'item-'||n,'PUBLISH','PUBLISHED',?,? FROM generate_series(1,?) n + """, row.id(), user, java.sql.Timestamp.from(clock.instant()), + java.sql.Timestamp.from(clock.instant()), items); + return row.id(); + } + + private int cleanupExpired() { + return tx.execute(ignored -> repository.cleanup(clock.instant().plusSeconds(1), clock.instant().plusSeconds(1))); + } + + private long rawRows(String table) { + return jdbc.queryForObject("SELECT count(*) FROM " + table, Long.class); + } + + @Test + void cleanupLargeRootIsBoundedResumableAndNeverExposesPartiallyDeletedHistory() { + long root = cleanupFixture("large", "FAILED", 350, 250); + long source = jdbc.queryForObject("SELECT min(id) FROM notification_email_delivery WHERE merged_into_id=?", + Long.class, root); + jdbc.update("UPDATE notification_email_item SET delivery_id=? WHERE delivery_id=?", source, root); + assertThat(cleanupExpired()).isEqualTo(100); + assertThat(rawRows("notification_email_item")).isEqualTo(250); + assertThat(rawRows("notification_email_delivery")).isEqualTo(151); + assertThat(repository.find(root)).isEmpty(); + assertThat(repository.find(source)).isEmpty(); + assertThat(repository.list(null, 0, 100)).isEmpty(); + assertThat(repository.count(null)).isZero(); + var query = new NotificationEmailDeliveryQueryService( + repository, eligibility, mock(com.iflytek.skillhub.domain.audit.AuditLogService.class), clock, policy, recipients); + assertThatThrownBy(() -> query.detail(root, "large", null, 50)) + .isInstanceOf(com.iflytek.skillhub.domain.shared.exception.DomainNotFoundException.class); + assertThatThrownBy(() -> tx.executeWithoutResult(ignored -> query.retry(root, "admin", false))) + .isInstanceOf(com.iflytek.skillhub.domain.shared.exception.DomainNotFoundException.class); + tx.executeWithoutResult(ignored -> repository.retry(root, clock.instant())); + assertThat(jdbc.queryForObject("SELECT status FROM notification_email_delivery WHERE id=?", String.class, root)) + .isEqualTo("FAILED"); + // A new repository instance resumes from the durable marker after an interrupted process. + repository = new JdbcNotificationEmailDeliveryRepository(jdbc); + for (int turn = 0; turn < 4 && rawRows("notification_email_delivery") > 0; turn++) { + long beforeItems = rawRows("notification_email_item"); + long beforeRows = rawRows("notification_email_delivery"); + cleanupExpired(); + assertThat(beforeItems - rawRows("notification_email_item")).isBetween(0L, 100L); + assertThat(beforeRows - rawRows("notification_email_delivery")).isBetween(0L, 101L); + } + assertThat(rawRows("notification_email_item")).isZero(); + assertThat(rawRows("notification_email_delivery")).isZero(); + } + + @Test + void cleanupRotatesBeyondFirstTenLargeRootsAndBoundsEveryTurn() { + for (int n = 0; n < 11; n++) cleanupFixture("fair-" + n, "SENT", 350, 0); + cleanupExpired(); + assertThat(rawRows("notification_email_item")).isEqualTo(2850); + assertThat(jdbc.queryForObject("SELECT count(*) FROM notification_email_delivery WHERE purge_started_at IS NOT NULL", + Long.class)).isEqualTo(10); + cleanupExpired(); + assertThat(rawRows("notification_email_item")).isEqualTo(1850); + assertThat(jdbc.queryForObject("SELECT count(*) FROM notification_email_delivery WHERE purge_started_at IS NOT NULL", + Long.class)).isEqualTo(11); + assertThat(jdbc.queryForObject("SELECT count(*) FROM notification_email_item WHERE recipient_id='fair-10'", + Long.class)).isEqualTo(250); + } + + @Test + void cleanupRetentionBoundaryAndRecentMergedItemProtectWholeGroup() { + for (String status : List.of("SENT", "SUPPRESSED", "FAILED", "UNKNOWN")) { + cleanupFixture("boundary-" + status, status, 1, 1); + } + tx.executeWithoutResult(ignored -> repository.cleanup(clock.instant(), clock.instant())); + assertThat(rawRows("notification_email_delivery")).isEqualTo(8); + assertThat(rawRows("notification_email_item")).isEqualTo(4); + long source = jdbc.queryForObject("SELECT min(id) FROM notification_email_delivery WHERE status='MERGED'", Long.class); + jdbc.update(""" + INSERT INTO notification_email_item(delivery_id,recipient_id,event_key,category,event_type,occurred_at,enqueued_at) + VALUES(?,'recent-source','recent','PUBLISH','PUBLISHED',?,?) + """, source, java.sql.Timestamp.from(clock.instant()), java.sql.Timestamp.from(clock.instant().plusSeconds(1))); + cleanupExpired(); + assertThat(rawRows("notification_email_delivery")).isEqualTo(2); + assertThat(rawRows("notification_email_item")).isEqualTo(2); + assertThat(jdbc.queryForObject("SELECT count(*) FROM notification_email_delivery WHERE purge_started_at IS NOT NULL", + Long.class)).isZero(); + tx.executeWithoutResult(ignored -> repository.cleanup(clock.instant().plusSeconds(2), clock.instant().plusSeconds(2))); + assertThat(rawRows("notification_email_delivery")).isZero(); + } + + @Test + void cleanupRollbackKeepsRetryableHistoryWhole() { + long root = cleanupFixture("rollback", "UNKNOWN", 350, 150); + assertThatThrownBy(() -> tx.executeWithoutResult(ignored -> { + cleanupExpired(); + throw new IllegalStateException("simulated interruption before commit"); + })).isInstanceOf(IllegalStateException.class); + assertThat(repository.find(root)).isPresent(); + assertThat(rawRows("notification_email_item")).isEqualTo(350); + assertThat(rawRows("notification_email_delivery")).isEqualTo(151); + cleanupExpired(); + assertThat(rawRows("notification_email_item")).isEqualTo(250); + } + + @Test + void cleanupWinsRaceAfterRetryReadAndPreventsRevivingPartiallyRemovedItems() throws Exception { + long root = cleanupFixture("race", "FAILED", 350, 0); + CountDownLatch retryRead = new CountDownLatch(1), continueRetry = new CountDownLatch(1); + when(recipients.lockCurrent("race")).thenAnswer(inv -> { + retryRead.countDown(); + if (!continueRetry.await(10, TimeUnit.SECONDS)) throw new AssertionError("retry latch timed out"); + return Optional.of(new NotificationEmailRecipient("race", "same@example.test", 0, "en")); + }); + var query = new NotificationEmailDeliveryQueryService( + repository, eligibility, mock(com.iflytek.skillhub.domain.audit.AuditLogService.class), clock, policy, recipients); + try (var pool = Executors.newSingleThreadExecutor()) { + Future pending = pool.submit(() -> tx.executeWithoutResult(ignored -> query.retry(root, "admin", false))); + try { + assertThat(retryRead.await(10, TimeUnit.SECONDS)).isTrue(); + cleanupExpired(); + } finally { + continueRetry.countDown(); + } + assertThatThrownBy(() -> pending.get(10, TimeUnit.SECONDS)) + .hasCauseInstanceOf(com.iflytek.skillhub.domain.shared.exception.DomainNotFoundException.class); + } + assertThat(rawRows("notification_email_item")).isEqualTo(250); + assertThat(jdbc.queryForObject("SELECT status FROM notification_email_delivery WHERE id=?", String.class, root)) + .isEqualTo("FAILED"); + } + + @Test + void cleanupSkipsLockedRetryRootThenRechecksItsNonterminalState() throws Exception { + long root = cleanupFixture("retry-first", "FAILED", 350, 0); + long other = cleanupFixture("unrelated", "SENT", 1, 0); + CountDownLatch locked = new CountDownLatch(1), release = new CountDownLatch(1); + try (var pool = Executors.newSingleThreadExecutor()) { + Future pending = pool.submit(() -> tx.executeWithoutResult(ignored -> { + repository.lock(root); + locked.countDown(); + try { + if (!release.await(10, TimeUnit.SECONDS)) throw new AssertionError("cleanup latch timed out"); + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); + throw new IllegalStateException(interrupted); + } + repository.retry(root, clock.instant()); + })); + try { + assertThat(locked.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(cleanupExpired()).isEqualTo(1); + assertThat(repository.find(other)).isEmpty(); + } finally { + release.countDown(); + } + pending.get(10, TimeUnit.SECONDS); + } + assertThat(cleanupExpired()).isZero(); + assertThat(repository.find(root).orElseThrow().status()).isEqualTo("RETRY"); + assertThat(rawRows("notification_email_item")).isEqualTo(350); + } + + @Test + void cleanupNeverMarksActiveStatesOrTerminalRootWithUnfinishedSource() { + for (String status : List.of("PENDING", "CLAIMED", "PREPARING", "SENDING", "RETRY")) { + cleanupFixture("unfinished-" + status, status, 1, 0); + } + long root = cleanupFixture("unfinished-source", "SENT", 1, 1); + jdbc.update("UPDATE notification_email_delivery SET status='RETRY' WHERE merged_into_id=?", root); + assertThat(cleanupExpired()).isZero(); + assertThat(rawRows("notification_email_item")).isEqualTo(6); + assertThat(rawRows("notification_email_delivery")).isEqualTo(7); + assertThat(jdbc.queryForObject("SELECT count(*) FROM notification_email_delivery WHERE purge_started_at IS NOT NULL", + Long.class)).isZero(); + } + } diff --git a/server/skillhub-infra/src/main/java/com/iflytek/skillhub/infra/jpa/JdbcNotificationEmailDeliveryRepository.java b/server/skillhub-infra/src/main/java/com/iflytek/skillhub/infra/jpa/JdbcNotificationEmailDeliveryRepository.java index 1224d0345..c55deb49b 100644 --- a/server/skillhub-infra/src/main/java/com/iflytek/skillhub/infra/jpa/JdbcNotificationEmailDeliveryRepository.java +++ b/server/skillhub-infra/src/main/java/com/iflytek/skillhub/infra/jpa/JdbcNotificationEmailDeliveryRepository.java @@ -40,14 +40,20 @@ public class JdbcNotificationEmailDeliveryRepository implements NotificationEmai @Override public Optional find(long id) { - return jdbc.query("SELECT * FROM notification_email_delivery WHERE id=?", this::delivery, id) + return jdbc.query(""" + SELECT * FROM notification_email_delivery d WHERE id=? AND purge_started_at IS NULL + AND NOT EXISTS (SELECT 1 FROM notification_email_delivery p + WHERE p.id=d.merged_into_id AND p.purge_started_at IS NOT NULL) + """, this::delivery, id) .stream() .findFirst(); } @Override public NotificationEmailDelivery lock(long id) { - return jdbc.query("SELECT * FROM notification_email_delivery WHERE id=? FOR UPDATE", this::delivery, id) + return jdbc.query( + "SELECT * FROM notification_email_delivery WHERE id=? AND purge_started_at IS NULL FOR UPDATE", + this::delivery, id) .stream() .findFirst() .orElse(null); @@ -413,7 +419,10 @@ public class JdbcNotificationEmailDeliveryRepository implements NotificationEmai @Override public List list(String status, int offset, int limit) { - return jdbc.query("SELECT * FROM notification_email_delivery WHERE (CAST(? AS VARCHAR) IS NULL OR status=?) " + return jdbc.query("SELECT * FROM notification_email_delivery d WHERE purge_started_at IS NULL " + + "AND NOT EXISTS (SELECT 1 FROM notification_email_delivery p " + + "WHERE p.id=d.merged_into_id AND p.purge_started_at IS NOT NULL) " + + "AND (CAST(? AS VARCHAR) IS NULL OR status=?) " + "ORDER BY id DESC LIMIT ? OFFSET ?", this::delivery, status, status, limit, offset); } @@ -421,7 +430,10 @@ public class JdbcNotificationEmailDeliveryRepository implements NotificationEmai @Override public long count(String status) { return Objects.requireNonNull(jdbc.queryForObject( - "SELECT count(*) FROM notification_email_delivery WHERE (CAST(? AS VARCHAR) IS NULL OR status=?)", + "SELECT count(*) FROM notification_email_delivery d WHERE purge_started_at IS NULL " + + "AND NOT EXISTS (SELECT 1 FROM notification_email_delivery p " + + "WHERE p.id=d.merged_into_id AND p.purge_started_at IS NOT NULL) " + + "AND (CAST(? AS VARCHAR) IS NULL OR status=?)", Long.class, status, status)); } @@ -430,21 +442,51 @@ public class JdbcNotificationEmailDeliveryRepository implements NotificationEmai jdbc.update(""" UPDATE notification_email_delivery SET status='RETRY',attempts=0,next_attempt_at=?,lease_token=NULL,lease_until=NULL, result_code='ADMIN_RETRY',retry_started_at=NULL,retry_deadline=NULL,terminal_at=NULL, - recovery_cutoff_at=NULL,sources_complete=FALSE,source_cursor=0,item_cursor=0,items_complete=FALSE,preview_item_ids='' WHERE id=? AND status IN ('FAILED','UNKNOWN') + recovery_cutoff_at=NULL,sources_complete=FALSE,source_cursor=0,item_cursor=0,items_complete=FALSE,preview_item_ids='' WHERE id=? AND purge_started_at IS NULL AND status IN ('FAILED','UNKNOWN') """, ts(now), id); } @Override public int cleanup(Instant sentBefore, Instant failedBefore) { - return jdbc.update(""" - DELETE FROM notification_email_delivery d WHERE d.merged_into_id IS NULL - AND ((d.status IN ('SENT','SUPPRESSED') AND coalesce(d.terminal_at,d.created_at)'MERGED') - AND NOT EXISTS (SELECT 1 FROM notification_email_item i JOIN notification_email_delivery s ON s.id=i.delivery_id - WHERE (s.id=d.id OR s.merged_into_id=d.id) AND i.enqueued_at>= - CASE WHEN d.status IN ('SENT','SUPPRESSED') THEN CAST(? AS TIMESTAMPTZ) ELSE CAST(? AS TIMESTAMPTZ) END) - """, ts(sentBefore), ts(failedBefore), ts(sentBefore), ts(failedBefore)); + // Lock roots first, as manual retry does. Once purging starts it cannot be retried or read partially. + var roots = jdbc.queryForList(""" + SELECT d.id FROM notification_email_delivery d WHERE d.merged_into_id IS NULL + AND (d.purge_started_at IS NOT NULL OR ( + ((d.status IN ('SENT','SUPPRESSED') AND coalesce(d.terminal_at,d.created_at)'MERGED') + AND NOT EXISTS (SELECT 1 FROM notification_email_item i JOIN notification_email_delivery s ON s.id=i.delivery_id + WHERE (s.id=d.id OR s.merged_into_id=d.id) AND i.enqueued_at>= + CASE WHEN d.status IN ('SENT','SUPPRESSED') THEN CAST(? AS TIMESTAMPTZ) ELSE CAST(? AS TIMESTAMPTZ) END))) + ORDER BY d.next_attempt_at,d.id LIMIT 10 FOR UPDATE OF d SKIP LOCKED + """, Long.class, ts(sentBefore), ts(failedBefore), ts(sentBefore), ts(failedBefore)); + int removed = 0; + for (long root : roots) { + // Only expired, marked roots reuse next_attempt_at for fair cleanup rotation, never sendable batches. + jdbc.update(""" + UPDATE notification_email_delivery SET purge_started_at=coalesce(purge_started_at,clock_timestamp()), + next_attempt_at=clock_timestamp() WHERE id=? + """, root); + jdbc.update(""" + DELETE FROM notification_email_item WHERE id IN ( + SELECT i.id FROM notification_email_item i JOIN notification_email_delivery d ON d.id=i.delivery_id + WHERE d.id=? OR d.merged_into_id=? ORDER BY i.id LIMIT 100) + """, root, root); + removed += jdbc.update(""" + DELETE FROM notification_email_delivery WHERE id IN ( + SELECT s.id FROM notification_email_delivery s WHERE s.merged_into_id=? AND s.status='MERGED' + AND NOT EXISTS (SELECT 1 FROM notification_email_item i WHERE i.delivery_id=s.id) + AND NOT EXISTS (SELECT 1 FROM notification_email_delivery c WHERE c.merged_into_id=s.id) + ORDER BY s.id LIMIT 100) + """, root); + // Explicit emptiness checks ensure FK cascades have no unbounded work left to do. + removed += jdbc.update(""" + DELETE FROM notification_email_delivery d WHERE d.id=? AND d.purge_started_at IS NOT NULL + AND NOT EXISTS (SELECT 1 FROM notification_email_item i WHERE i.delivery_id=d.id) + AND NOT EXISTS (SELECT 1 FROM notification_email_delivery s WHERE s.merged_into_id=d.id) + """, root); + } + return removed; } } diff --git a/server/skillhub-notification/src/main/java/com/iflytek/skillhub/notification/email/delivery/NotificationEmailDeliveryRepository.java b/server/skillhub-notification/src/main/java/com/iflytek/skillhub/notification/email/delivery/NotificationEmailDeliveryRepository.java index 2383be4c3..3d4820e12 100644 --- a/server/skillhub-notification/src/main/java/com/iflytek/skillhub/notification/email/delivery/NotificationEmailDeliveryRepository.java +++ b/server/skillhub-notification/src/main/java/com/iflytek/skillhub/notification/email/delivery/NotificationEmailDeliveryRepository.java @@ -35,5 +35,9 @@ public interface NotificationEmailDeliveryRepository { List list(String status, int offset, int limit); long count(String status); void retry(long id, Instant now); + /** + * One transaction: at most 10 roots, each with 100 items, 100 empty sources, one empty root and one marker update. + * Returns the number of delivery rows removed; purging groups are no longer readable or retryable. + */ int cleanup(Instant sentBefore, Instant failedBefore); }