Middleware/msgpush
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(retry): SMS限流类无限重试治本(批次1核心 + 批次2唯一约束)

Browse Source 
批次1 治本核心(MessageRetryQueueServiceImpl):
- 新增常量与纯函数 helper:isRateLimit / isRateLimitExhausted / isTerminalState,
  及 RetryDecisionLogicTest 纯JUnit5单测(6例)
- A1/A2 删除时机:sending 中间态不再误删,仅终态才删 —— 根治 firstFailTime/retryCount
  被重置的 R1 根因(回执回来命中既有记录而非走首次入队)
- B1/B2 限流分流:限流类(errorCode=RATE_LIMIT)按"首次失败起满12h"时间兜底、
  固定120s间隔;普通类维持次数上限与阶梯间隔,行为不变

批次2 唯一约束:
- 新增 push_message_retry_queue.message_id 唯一约束迁移SQL(uk_pmrq_message_id,含去重前置查)
- saveOrUpdateRetryMessage 抽 applyRetryUpdate + 捕获 DuplicateKeyException 退化更新(MySQL安全)

说明:批次3(守卫链对限流类重发的豁免)经核实为"功能真正生效的必要条件",
findings 见 docs/push模块/配额墙核实-findings-2026-06-19.md,留待下轮。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
hongawen
2026-06-19 16:04:34 +08:00
parent c2bbdd865d
commit 99c8bc3636
3 changed files with 183 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

83
msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/service/retry/MessageRetryQueueServiceImpl.java
View File

@@ -19,10 +19,12 @@ 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;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.Arrays;
import java.util.Collections;
@@ -41,6 +43,18 @@ public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueM
private static final int PROCESS_STATUS_PROCESSING = 1;
private static final int DEFAULT_LOCK_SECONDS = 120;
/** 限流类失败的统一错误码(回执后由 Sender/MessageRecordServiceImpl 写入 message.errorCode) */
private static final String RATE_LIMIT_ERROR_CODE = "RATE_LIMIT";
/** 限流类固定重试间隔:2 分钟(不走阶梯) */
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(),
ChannelTypeEnum.EMAIL.getCode(),
@@ -63,6 +77,7 @@ 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());
@@ -73,7 +88,7 @@ public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueM
retryRecord.setFirstFailTime(message.getSendTime());
retryRecord.setLastRetryTime(message.getSendTime());
retryRecord.setLastErrorMsg(message.getErrorMsg());
LocalDateTime nextRetryTime = calculateNextRetryTime(message.getChannel(), retryRecord.getRetryCount());
LocalDateTime nextRetryTime = calculateNextRetryTime(message.getChannel(), retryRecord.getRetryCount(), isRateLimit(message));
retryRecord.setNextRetryTime(nextRetryTime);
retryRecord.setProcessStatus(PROCESS_STATUS_WAITING);
retryRecord.setLockUntil(null);
@@ -82,14 +97,31 @@ public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueM
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);
}
}
return;
}
applyRetryUpdate(message, existing);
}
private void applyRetryUpdate(MessageRecordDO message, MessageRetryQueueDO existing) {
int newRetryCount = existing.getRetryCount() + 1;
existing.setLastRetryTime(message.getSendTime());
existing.setLastErrorMsg(message.getErrorMsg());
if (newRetryCount >= existing.getMaxRetry()) {
boolean rateLimit = isRateLimit(message);
boolean exhausted = rateLimit
? isRateLimitExhausted(existing.getFirstFailTime(), LocalDateTime.now())
: newRetryCount >= existing.getMaxRetry();
if (exhausted) {
message.setStatus(MsgStatusConstant.FINALFAILED);
message.setNextRetryTime(null);
existing.setNextRetryTime(null);
@@ -99,7 +131,7 @@ public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueM
return;
}
LocalDateTime nextRetryTime = calculateNextRetryTime(message.getChannel(), newRetryCount);
LocalDateTime nextRetryTime = calculateNextRetryTime(message.getChannel(), newRetryCount, rateLimit);
existing.setRetryCount(newRetryCount);
existing.setNextRetryTime(nextRetryTime);
existing.setProcessStatus(PROCESS_STATUS_WAITING);
@@ -111,7 +143,31 @@ public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueM
}
}
private LocalDateTime calculateNextRetryTime(String channel, int retryCount) {
/** 限流类识别:仅当统一错误码为 RATE_LIMIT */
static boolean isRateLimit(MessageRecordDO message) {
return message != null && RATE_LIMIT_ERROR_CODE.equals(message.getErrorCode());
}
/**
* 限流类时间兜底判定:首次失败起是否已满 12 小时。
* firstFailTime 为 null(历史数据)时视为未满,继续重试。
*/
static boolean isRateLimitExhausted(LocalDateTime firstFailTime, LocalDateTime now) {
if (firstFailTime == null) {
return false;
}
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);
}
RetryStrategyConfigDO strategyConfig = retryStrategyConfigService.getStrategyConfig(channel);
long plusSeconds = 0;
if (ObjectUtil.isNull(strategyConfig)) {
@@ -195,7 +251,8 @@ public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueM
messageRecordService.processSendMsg(Collections.singletonList(messageRecordDO), retrySource);
messageRecordService.updateRetryCount(messageRecordDO.getId());
finalizeRetryLease(messageRecordDO);
} else {
} else if (isTerminalState(messageRecordDO.getStatus())) {
// 仅终态才清理;sending / pending 等中间态保留,等其各自链路推进
deleteByMessageIds(Collections.singletonList(messageRecordDO.getId()));
}
}
@@ -345,18 +402,27 @@ public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueM
/**
* 完成重试租约处理，根据消息状态执行相应的清理或重置操作
* - 如果消息状态为可重试失败，直接返回，不进行任何处理
* - SENDING(两段式中间态)直接返回，保留记录与租约，等回执推进
* - 如果消息状态为待处理，重置队列记录的处理状态和锁时间，并将消息重新加入Redis重试队列
* - 其他状态（成功、最终失败等），从Redis重试队列移除并删除数据库中的重试队列记录
*
* @param messageRecordDO 消息记录对象
*/
private void finalizeRetryLease(MessageRecordDO messageRecordDO) {
if (MsgStatusConstant.RETRYABLE_FAILED.equals(messageRecordDO.getStatus())) {
String status = messageRecordDO.getStatus();
// 可重试失败:保留记录,等下一轮认领重发 —— 不变
if (MsgStatusConstant.RETRYABLE_FAILED.equals(status)) {
return;
}
// “处理过程中断，但消息仍停留在 PENDING 时，主动释放 lease 并回队列”的兜底补偿逻辑
if (MsgStatusConstant.PENDING.equals(messageRecordDO.getStatus())) {
// 两段式发送的中间态:重发已受理、消息在途,保留记录、保持租约锁住、不回 Redis,等 20s 回执推进
if (MsgStatusConstant.SENDING.equals(status)) {
return;
}
// 处理过程中断但仍停留在 PENDING:主动释放 lease 并回队列(兜底补偿)—— 不变
if (MsgStatusConstant.PENDING.equals(status)) {
lambdaUpdate()
.eq(MessageRetryQueueDO::getMessageId, messageRecordDO.getId())
.set(MessageRetryQueueDO::getProcessStatus, PROCESS_STATUS_WAITING)
@@ -368,6 +434,7 @@ public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueM
return;
}
// 其余即终态:从 Redis 移除并删除数据库重试队列记录
messageRetryRedisDAO.removeFromRetryQueue(messageRecordDO.getChannel(), messageRecordDO.getId());
deleteByMessageIds(Collections.singletonList(messageRecordDO.getId()));
}

13
msgpush-module-push/msgpush-module-push-server/src/main/resources/sql/push_message_retry_queue_uk_message_id.sql Normal file
View File

@@ -0,0 +1,13 @@
-- 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);

78
msgpush-module-push/msgpush-module-push-server/src/test/java/com/njcn/msgpush/module/push/service/retry/RetryDecisionLogicTest.java Normal file
View File

@@ -0,0 +1,78 @@
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;
/**
* 限流重试治本核心的纯决策逻辑单测（无 Spring / 无 DB / 无 Redis）。
*/
class RetryDecisionLogicTest {
@Test
void isRateLimit_trueOnlyForRateLimitErrorCode() {
MessageRecordDO rateLimited = new MessageRecordDO();
rateLimited.setErrorCode("RATE_LIMIT");
assertTrue(MessageRetryQueueServiceImpl.isRateLimit(rateLimited));
MessageRecordDO other = new MessageRecordDO();
other.setErrorCode("SOMETHING_ELSE");
assertFalse(MessageRetryQueueServiceImpl.isRateLimit(other));
MessageRecordDO noCode = new MessageRecordDO();
assertFalse(MessageRetryQueueServiceImpl.isRateLimit(noCode));
assertFalse(MessageRetryQueueServiceImpl.isRateLimit(null));
}
@Test
void isRateLimitExhausted_nullFirstFailTimeMeansNotExhausted() {
assertFalse(MessageRetryQueueServiceImpl.isRateLimitExhausted(null, LocalDateTime.now()));
}
@Test
void isRateLimitExhausted_falseBelow12h_trueAtOrAbove12h() {
LocalDateTime now = LocalDateTime.now();
// 43199s < 12h → 未满
assertFalse(MessageRetryQueueServiceImpl.isRateLimitExhausted(now.minusSeconds(43199), now));
// 43200s == 12h → 满（>=）
assertTrue(MessageRetryQueueServiceImpl.isRateLimitExhausted(now.minusSeconds(43200), now));
// 43201s > 12h → 满
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();
LocalDateTime before = LocalDateTime.now();
LocalDateTime next = service.calculateNextRetryTime("SMS", 1, true);
long seconds = Duration.between(before, next).getSeconds();
// rate-limit 分支在使用任何字段前就 return，故 new 裸实例可调；允许执行抖动 ±2s
assertTrue(seconds >= 118 && seconds <= 122, "expected ~120s, got " + seconds);
}
}