From cbfee6b15655e698a46dfbb7a0ab0595771f1c13 Mon Sep 17 00:00:00 2001 From: hongawen <83944980@qq.com> Date: Fri, 19 Jun 2026 17:04:47 +0800 Subject: [PATCH] =?UTF-8?q?fix(retry):=20=E4=BF=AE=E5=A4=8D=E6=89=B9?= =?UTF-8?q?=E6=AC=A11+2=20review=20=E6=9F=A5=E5=87=BA=E7=9A=84=E4=B8=A4?= =?UTF-8?q?=E4=B8=AA=20blocker(I1=20=E6=92=A4=E5=94=AF=E4=B8=80=E7=BA=A6?= =?UTF-8?q?=E6=9D=9F=20/=20I2=20SENDING=20=E6=B3=84=E6=BC=8F)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 接 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) --- .../retry/MessageRetryQueueServiceImpl.java | 61 ++++++------------- ...push_message_retry_queue_uk_message_id.sql | 13 ---- .../service/retry/RetryDecisionLogicTest.java | 20 ------ 3 files changed, 20 insertions(+), 74 deletions(-) delete mode 100644 msgpush-module-push/msgpush-module-push-server/src/main/resources/sql/push_message_retry_queue_uk_message_id.sql diff --git 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 index 5184e74..9e371cf 100644 --- 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 INTERMEDIATE_STATES = Arrays.asList( - MsgStatusConstant.SENDING, - MsgStatusConstant.PENDING, - MsgStatusConstant.RETRYABLE_FAILED); private static final List CHANNELS = Arrays.asList( ChannelTypeEnum.SMS.getCode(), @@ -77,34 +71,24 @@ public class MessageRetryQueueServiceImpl extends ServiceImpl() - .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= 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 1; - --- 2) 加唯一约束(确认上面无重复后再执行) -ALTER TABLE push_message_retry_queue - ADD CONSTRAINT uk_pmrq_message_id UNIQUE (message_id); diff --git 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 index 78018a2..0dff716 100644 --- 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();