fix(retry): 修复批次1+2 review 查出的两个 blocker(I1 撤唯一约束 / I2 SENDING 泄漏)

接 99c8bc3 多代理深度 code review,治本两个 blocker: I1 软删除 × 唯一约束 → 静默丢消息(撤批次2): - MessageRetryQueueDO 走 @TableLogic 逻辑删除,deleteByMessageIds 实为 UPDATE deleted=1,物理行残留;批次2 的 uk_pmrq_message_id 只盖 message_id, 与残留行冲突 → 同一 message_id 再入队抛 DuplicateKeyException → catch 再 selectOne 仍 null(软删行不可见)→ concurrent==null 静默 return → 消息永久丢失。 - 撤回:删迁移 SQL、移除 DuplicateKeyException catch 与 import,回到纯 selectOne→insert/update;applyRetryUpdate(含批次1 限流逻辑)保留为单点调用。 I2 SENDING 行泄漏 + 周期空转(最小治本): - processSingleRetry 的 else 把"保留中间态"白名单误用在重认领路径;能被重认领说明 120s 租约已过期、20s 回执窗口早过,SENDING/PENDING 属卡住却被保留 → 永不删除/推进、到不了 12h 兜底、周期空转。改前的无条件 delete 会自愈。 - 改回 else { delete }(重认领即清理);删死代码 isTerminalState / INTERMEDIATE_STATES; 保留 finalizeRetryLease 的 SENDING→return(发完即时窗口保住行,批次1 修 R1 关键)。测试:RetryDecisionLogicTest 删 isTerminalState 两例、清死 import,余 4 例全绿; 全模块重编译通过。设计见 docs/superpowers/specs/2026-06-19-sms-retry-blockers-fix-design.md。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-19 17:04:47 +08:00
parent 99c8bc3636
commit cbfee6b156
3 changed files with 20 additions and 74 deletions
--- a/msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/service/retry/MessageRetryQueueServiceImpl.java
+++ b/msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/service/retry/MessageRetryQueueServiceImpl.java
@@ -19,7 +19,6 @@ import com.njcn.msgpush.module.push.service.message.MessageRecordService;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.CommandLineRunner;
-import org.springframework.dao.DuplicateKeyException;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
@@ -49,11 +48,6 @@ public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueM
    private static final int RATE_LIMIT_INTERVAL_SECONDS = 120;
    /** 限流类时间兜底:首次失败起满 12 小时未成功才放弃 */
    private static final long RATE_LIMIT_MAX_DURATION_SECONDS = 43200;
-    /** 白名单中间态:这三个状态保留队列记录,其余一律视为终态删除 */
-    private static final List<String> INTERMEDIATE_STATES = Arrays.asList(
-            MsgStatusConstant.SENDING,
-            MsgStatusConstant.PENDING,
-            MsgStatusConstant.RETRYABLE_FAILED);

    private static final List<String> CHANNELS = Arrays.asList(
            ChannelTypeEnum.SMS.getCode(),
@@ -77,34 +71,24 @@ public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueM
        message.setLastRetryTime(message.getSendTime());

        if (ObjectUtil.isNull(existing)) {
-            try {
-                MessageRetryQueueDO retryRecord = new MessageRetryQueueDO();
-                retryRecord.setMessageId(message.getId());
-                retryRecord.setChannel(message.getChannel());
-                retryRecord.setReceiver(message.getReceiver());
-                retryRecord.setRetryCount(1);
-                RetryStrategyConfigDO strategyConfig = retryStrategyConfigService.getStrategyConfig(message.getChannel());
-                retryRecord.setMaxRetry(ObjectUtil.isNull(strategyConfig) ? DEFAULT_MAX_RETRY_COUNT : strategyConfig.getMaxRetryCount());
-                retryRecord.setFirstFailTime(message.getSendTime());
-                retryRecord.setLastRetryTime(message.getSendTime());
-                retryRecord.setLastErrorMsg(message.getErrorMsg());
-                LocalDateTime nextRetryTime = calculateNextRetryTime(message.getChannel(), retryRecord.getRetryCount(), isRateLimit(message));
-                retryRecord.setNextRetryTime(nextRetryTime);
-                retryRecord.setProcessStatus(PROCESS_STATUS_WAITING);
-                retryRecord.setLockUntil(null);
-                message.setNextRetryTime(nextRetryTime);
-                save(retryRecord);
-                if (!messageRetryRedisDAO.addToRetryQueue(message)) {
-                    log.warn("消息已写入数据库重试队列，但 Redis 入队失败: messageId={}", message.getId());
-                }
-            } catch (DuplicateKeyException e) {
-                // 并发下另一线程已为同一 message_id 插入 → 退化为更新(依赖 uk_pmrq_message_id 唯一约束)
-                log.warn("并发插入冲突，退化为更新: messageId={}", message.getId());
-                MessageRetryQueueDO concurrent = baseMapper.selectOne(new LambdaQueryWrapper<MessageRetryQueueDO>()
-                        .eq(MessageRetryQueueDO::getMessageId, message.getId()));
-                if (concurrent != null) {
-                    applyRetryUpdate(message, concurrent);
-                }
+            MessageRetryQueueDO retryRecord = new MessageRetryQueueDO();
+            retryRecord.setMessageId(message.getId());
+            retryRecord.setChannel(message.getChannel());
+            retryRecord.setReceiver(message.getReceiver());
+            retryRecord.setRetryCount(1);
+            RetryStrategyConfigDO strategyConfig = retryStrategyConfigService.getStrategyConfig(message.getChannel());
+            retryRecord.setMaxRetry(ObjectUtil.isNull(strategyConfig) ? DEFAULT_MAX_RETRY_COUNT : strategyConfig.getMaxRetryCount());
+            retryRecord.setFirstFailTime(message.getSendTime());
+            retryRecord.setLastRetryTime(message.getSendTime());
+            retryRecord.setLastErrorMsg(message.getErrorMsg());
+            LocalDateTime nextRetryTime = calculateNextRetryTime(message.getChannel(), retryRecord.getRetryCount(), isRateLimit(message));
+            retryRecord.setNextRetryTime(nextRetryTime);
+            retryRecord.setProcessStatus(PROCESS_STATUS_WAITING);
+            retryRecord.setLockUntil(null);
+            message.setNextRetryTime(nextRetryTime);
+            save(retryRecord);
+            if (!messageRetryRedisDAO.addToRetryQueue(message)) {
+                log.warn("消息已写入数据库重试队列，但 Redis 入队失败: messageId={}", message.getId());
            }
            return;
        }
@@ -159,11 +143,6 @@ public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueM
        return Duration.between(firstFailTime, now).getSeconds() >= RATE_LIMIT_MAX_DURATION_SECONDS;
    }

-    /** 是否终态(默认删):不在白名单中间态里即为终态 */
-    static boolean isTerminalState(String status) {
-        return !INTERMEDIATE_STATES.contains(status);
-    }
-
    LocalDateTime calculateNextRetryTime(String channel, int retryCount, boolean rateLimit) {
        if (rateLimit) {
            return LocalDateTime.now().plusSeconds(RATE_LIMIT_INTERVAL_SECONDS);
@@ -251,8 +230,8 @@ public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueM
            messageRecordService.processSendMsg(Collections.singletonList(messageRecordDO), retrySource);
            messageRecordService.updateRetryCount(messageRecordDO.getId());
            finalizeRetryLease(messageRecordDO);
-        } else if (isTerminalState(messageRecordDO.getStatus())) {
-            // 仅终态才清理;sending / pending 等中间态保留,等其各自链路推进
+        } else {
+            // 能被重新认领说明 120s 租约已过期、20s 回执窗口早过 —— SENDING/PENDING 即卡住,与终态一并清理
            deleteByMessageIds(Collections.singletonList(messageRecordDO.getId()));
        }
    }
--- a/msgpush-module-push/msgpush-module-push-server/src/main/resources/sql/push_message_retry_queue_uk_message_id.sql
+++ b/msgpush-module-push/msgpush-module-push-server/src/main/resources/sql/push_message_retry_queue_uk_message_id.sql
@@ -1,13 +0,0 @@
-- push_message_retry_queue.message_id 唯一约束迁移
-- 背景:saveOrUpdateRetryMessage 走 selectOne→insert/update,并发下可能对同一 message_id 产生重复行。
-- 治本核心(批次1)后,同一 message_id 在 sending→回执 周期内只应有一行;唯一约束 + 应用层兜底做廉价保险。
-
-- 1) 前置去重检查:执行前确认无重复(应返回 0 行;若有,需先人工清理重复,保留 id 最大的一行)
-SELECT message_id, COUNT(*) AS cnt
-FROM push_message_retry_queue
-GROUP BY message_id
-HAVING cnt > 1;
-
-- 2) 加唯一约束(确认上面无重复后再执行)
-ALTER TABLE push_message_retry_queue
-    ADD CONSTRAINT uk_pmrq_message_id UNIQUE (message_id);
--- a/msgpush-module-push/msgpush-module-push-server/src/test/java/com/njcn/msgpush/module/push/service/retry/RetryDecisionLogicTest.java
+++ b/msgpush-module-push/msgpush-module-push-server/src/test/java/com/njcn/msgpush/module/push/service/retry/RetryDecisionLogicTest.java
@@ -1,13 +1,11 @@
 package com.njcn.msgpush.module.push.service.retry;

-import com.njcn.msgpush.module.push.constant.MsgStatusConstant;
 import com.njcn.msgpush.module.push.dal.dataobject.message.MessageRecordDO;
 import org.junit.jupiter.api.Test;

 import java.time.Duration;
 import java.time.LocalDateTime;

-import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertTrue;

@@ -48,24 +46,6 @@ class RetryDecisionLogicTest {
        assertTrue(MessageRetryQueueServiceImpl.isRateLimitExhausted(now.minusSeconds(43201), now));
    }

-    @Test
-    void isTerminalState_intermediatesAreNotTerminal() {
-        assertFalse(MessageRetryQueueServiceImpl.isTerminalState(MsgStatusConstant.SENDING));
-        assertFalse(MessageRetryQueueServiceImpl.isTerminalState(MsgStatusConstant.PENDING));
-        assertFalse(MessageRetryQueueServiceImpl.isTerminalState(MsgStatusConstant.RETRYABLE_FAILED));
-    }
-
-    @Test
-    void isTerminalState_everythingElseIsTerminal() {
-        assertTrue(MessageRetryQueueServiceImpl.isTerminalState(MsgStatusConstant.SUCCESS));
-        assertTrue(MessageRetryQueueServiceImpl.isTerminalState(MsgStatusConstant.FAILED));
-        assertTrue(MessageRetryQueueServiceImpl.isTerminalState(MsgStatusConstant.FINALFAILED));
-        assertTrue(MessageRetryQueueServiceImpl.isTerminalState(MsgStatusConstant.QUOTAEXCEEDED));
-        assertTrue(MessageRetryQueueServiceImpl.isTerminalState(MsgStatusConstant.RATE_LIMITED));
-        // 白名单默认删：未知/ null 状态视为终态（与 spec「默认删，只保留三个」一致）
-        assertTrue(MessageRetryQueueServiceImpl.isTerminalState(null));
-    }
-
    @Test
    void calculateNextRetryTime_rateLimitUsesFixed120s() {
        MessageRetryQueueServiceImpl service = new MessageRetryQueueServiceImpl();