Compare commits

..

7 Commits

Author SHA1 Message Date
9da61452a9 chore(retry): 新增 SMS 限流重试的库侧脚本(MA:0038 映射 + 普通类阶梯)
resources/sql 下两份手工执行迁移脚本(供 DBA 在目标库运行,均幂等):
- provider-error-code-mapping/telecom_sms_MA0038_retryable.sql:
  给 push_provider_error_code_mapping 配 (telecom,sms,MA:0038 ->
  unified_error_code=RATE_LIMIT, should_retry=1)。与 TelecomSmsSender
  的 status=6 改动(c892f71)缺一不可——MA:0038 才能进重试链并被识别为限流类。
- sms_retry_strategy_ladder.sql:
  普通类 sms 失败重试阶梯 retry_intervals='60,120,300,600'、max_retry_count=6
  (TD-3 决策),取代旧 quickfix 的 60s 平铺;限流类由代码 120s/12h 保证,不读此表。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-20 23:11:00 +08:00
c892f71711 fix(sms): 电信回执 status=6(MA:0038 限流)纳入失败映射,不再静默丢弃
电信下行回执查询对"短期频繁发送"类返回 status=6 / stat=MA:0038,
原代码仅处理 status=5,status=6 无分支被直接丢弃 —— 该类短信永远
进不了失败/重试链路。改为 status==5 || status==6 一并走 buildFailureResult,
由 push_provider_error_code_mapping.should_retry 决定是否重试。

这是限流类无限重试治本(批次1-3)在回执侧的触发点,与库内
MA:0038 -> RATE_LIMIT 映射种子(TD-2)缺一不可。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-20 15:40:43 +08:00
8231b1e0e1 fix(retry): 修复批量入队 NPE(next_retry_time 取错数据源)
根因:fallbackToDatabaseDirectly / rebuildRetryQueueOnStartup 批量入队时,
取的是 push_message_record.next_retry_time(只在内存设置、不保证落库,常为 null),
而调度权威其实是 push_message_retry_queue.next_retry_time(已落库、查询已保证非空)。
定时任务回落 DB 时拿到 null 的 message,addToRetryQueue 调 atZone 即 NPE,
整个 SMS 重试定时任务被带崩。

治本:新增 fillNextRetryTimeFromQueue,用队列记录的权威 nextRetryTime 回填
message 再入队,两条批量路径各调用一次 —— 不丢消息、照常重试。
兜底:addToRetryQueue 对 nextRetryTime==null 打 warn 并跳过单条,
保证任何坏数据都不会再把整批/整个定时任务带崩。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-19 19:33:00 +08:00
f076dcd9b2 feat(retry): 批次3 限流类自动重试守卫豁免(让无限重试在生产真正生效)
接 16974d0(I3)。Task8 核实:限流类自动重发途经 MsgPushGuardChain 的
Quota/RateLimitChecker,per-receiver 日限远小于 ~360 次/12h,会被守卫置
rate_limited/quota_exceeded 终态 → 队列记录被删 → 无限重试静默失效。
批次1/2/I3 修对了重试编排,但不豁免守卫则功能在生产仍发不出去。

方案 A+B(用户选:完全豁免 AUTO_RETRY):
- A:checkAll 加 RetrySourceEnum 参,AUTO_RETRY 跳过 quota+ratelimit
  (blacklist 仍执行);processSendMsg 透传。重发节奏由 120s 间隔 + 12h
  兜底天然封顶,无需配额墙再约束。
- B:applySendResult 加 retrySource 参,仅首发(null)累加配额/频率计数;
  重发(AUTO/MANUAL)是同一条消息再投,不重复扣额。
- MANUAL_RETRY 仍受守卫检查(只豁免 AUTO)。

checkAll 全仓仅 1 调用点;applySendResult 为 impl 内部方法、2 处 this 调用,
接口签名不变。Spring 编排+Redis 副作用无 H2 脚手架,靠编译+diff+人工冒烟
验证(spec §8):RetryDecisionLogicTest 仍 6/6 绿,全模块编译通过。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-19 17:39:12 +08:00
16974d04f9 fix(retry): 修复 I3 限流/普通共用 retryCount 跨污染
接 cbfee6b。retryCount 原被限流与普通失败共用累加,而终止判定按本次
errorCode 在 12h 兜底与次数上限间翻转:限流重试撑高 retryCount 后一旦
出现普通失败,会被 retryCount>=maxRetry 提前判死;反之普通累计接近上限
后转限流又借 12h 续命绕过次数上限。

最小治本(不动 DB):retryCount 语义收敛为普通失败次数,限流重试冻结
它(靠 firstFailTime+12h 兜底终止)。
- 新增 nextRetryCount(current, rateLimit):限流冻结、普通累加。
- applyRetryUpdate 改用 nextRetryCount;insert 路径首次限流 retryCount=0、
  普通=1。firstFailTime 维度本无污染(只设一次),不动。

测试:RetryDecisionLogicTest 新增 2 例锁回归,原 4 例不变,共 6/6 绿;
全模块编译通过。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-19 17:28:28 +08:00
cbfee6b156 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
99c8bc3636 feat(retry): SMS限流类无限重试治本(批次1核心 + 批次2唯一约束)
批次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>
2026-06-19 16:04:34 +08:00
8 changed files with 272 additions and 19 deletions

View File

@@ -4,6 +4,7 @@ import com.njcn.msgpush.module.push.checker.impl.BlacklistChecker;
import com.njcn.msgpush.module.push.checker.impl.QuotaChecker; import com.njcn.msgpush.module.push.checker.impl.QuotaChecker;
import com.njcn.msgpush.module.push.checker.impl.RateLimitChecker; import com.njcn.msgpush.module.push.checker.impl.RateLimitChecker;
import com.njcn.msgpush.module.push.dal.dataobject.message.MessageRecordDO; import com.njcn.msgpush.module.push.dal.dataobject.message.MessageRecordDO;
import com.njcn.msgpush.module.push.enums.RetrySourceEnum;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
@@ -24,9 +25,15 @@ public class MsgPushGuardChain {
private RateLimitChecker rateLimitChecker; private RateLimitChecker rateLimitChecker;
public void checkAll(List<MessageRecordDO> messageRecordList) { public void checkAll(List<MessageRecordDO> messageRecordList, RetrySourceEnum retrySource) {
// 黑名单:首发与重发(含限流自动重试)都必须执行
blacklistChecker.check(messageRecordList); blacklistChecker.check(messageRecordList);
quotaChecker.check(messageRecordList); // 配额/接收者频率:系统自动重试(限流类无限重试)完全豁免——否则会被守卫置终态、队列记录被删,
rateLimitChecker.check(messageRecordList); // 无限重试静默失效。重发节奏由 120s 固定间隔 + 12h 兜底天然封顶,无需配额墙再约束。
// 首发(retrySource==null)与人工重试(MANUAL_RETRY)仍正常受这两道守卫。
if (retrySource != RetrySourceEnum.AUTO_RETRY) {
quotaChecker.check(messageRecordList);
rateLimitChecker.check(messageRecordList);
}
} }
} }

View File

@@ -200,7 +200,9 @@ public class TelecomSmsSender implements SmsSender {
return; return;
} }
TelecomSmsSelectDetailRes detailRes = selectResponse.getList().get(0); TelecomSmsSelectDetailRes detailRes = selectResponse.getList().get(0);
if (detailRes.getStatus() == 5) { // status=5 为明确失败status=6 为“短期频繁发送”等场景stat 如 MA:0038
// 同样走错误码映射,由 push_provider_error_code_mapping.should_retry 决定是否重试。
if (detailRes.getStatus() == 5 || detailRes.getStatus() == 6) {
SendResult failedResult = this.sender.buildFailureResult( SendResult failedResult = this.sender.buildFailureResult(
message, message,
detailRes.getStat(), detailRes.getStat(),

View File

@@ -88,6 +88,11 @@ public class MessageRetryRedisDAO {
* @param message 消息 * @param message 消息
*/ */
public boolean addToRetryQueue(MessageRecordDO message) { public boolean addToRetryQueue(MessageRecordDO message) {
// 兜底守卫:nextRetryTime 为空则跳过单条,绝不让一条坏数据把整批/整个定时任务带崩
if (message.getNextRetryTime() == null) {
log.warn("跳过入队:nextRetryTime 为空, messageId={}, channel={}", message.getId(), message.getChannel());
return false;
}
String key = getRetryQueueKey(message.getChannel()); String key = getRetryQueueKey(message.getChannel());
double score = message.getNextRetryTime().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(); double score = message.getNextRetryTime().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();

View File

@@ -93,7 +93,7 @@ public class MessageRecordServiceImpl extends ServiceImpl<MessageRecordMapper, M
*/ */
@Override @Override
public List<MessageSendResultVO> processSendMsg(List<MessageRecordDO> messageRecordDOList, RetrySourceEnum retrySource) { public List<MessageSendResultVO> processSendMsg(List<MessageRecordDO> messageRecordDOList, RetrySourceEnum retrySource) {
msgPushGuardChain.checkAll(messageRecordDOList); msgPushGuardChain.checkAll(messageRecordDOList, retrySource);
List<MessageSendResultVO> resultList = new ArrayList<>(); List<MessageSendResultVO> resultList = new ArrayList<>();
for (MessageRecordDO messageRecordDO : messageRecordDOList) { for (MessageRecordDO messageRecordDO : messageRecordDOList) {
try { try {
@@ -131,11 +131,11 @@ public class MessageRecordServiceImpl extends ServiceImpl<MessageRecordMapper, M
if (sendResult == null) { if (sendResult == null) {
sendResult = this.sendMessage(messageRecordDO, channelProviderConfigDO, messageProviderFactory); sendResult = this.sendMessage(messageRecordDO, channelProviderConfigDO, messageProviderFactory);
} }
this.applySendResult(messageRecordDO, sendResult); this.applySendResult(messageRecordDO, sendResult, retrySource);
this.recordRetryHistory(messageRecordDO, retrySource); this.recordRetryHistory(messageRecordDO, retrySource);
} else { } else {
sendResult = this.validateProviderAndChannel(messageRecordDO, null, null); sendResult = this.validateProviderAndChannel(messageRecordDO, null, null);
this.applySendResult(messageRecordDO, sendResult); this.applySendResult(messageRecordDO, sendResult, retrySource);
} }
@@ -176,7 +176,7 @@ public class MessageRecordServiceImpl extends ServiceImpl<MessageRecordMapper, M
* 主流程在这里统一做状态落库、重试编排和健康度更新。 * 主流程在这里统一做状态落库、重试编排和健康度更新。
*/ */
@Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = Exception.class) @Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = Exception.class)
public void applySendResult(MessageRecordDO messageRecordDO, SendResult sendResult) { public void applySendResult(MessageRecordDO messageRecordDO, SendResult sendResult, RetrySourceEnum retrySource) {
messageRecordDO.setStatus(this.mapOutcomeToStatus(sendResult.getOutcome())); messageRecordDO.setStatus(this.mapOutcomeToStatus(sendResult.getOutcome()));
messageRecordDO.setErrorCode(sendResult.getErrorCode()); messageRecordDO.setErrorCode(sendResult.getErrorCode());
messageRecordDO.setErrorMsg(sendResult.getMessage()); messageRecordDO.setErrorMsg(sendResult.getMessage());
@@ -194,8 +194,12 @@ public class MessageRecordServiceImpl extends ServiceImpl<MessageRecordMapper, M
if (SendOutcome.ACCEPTED.equals(sendResult.getOutcome())) { if (SendOutcome.ACCEPTED.equals(sendResult.getOutcome())) {
messageConfirmRedisDAO.addToConfirmQueue(messageRecordDO); messageConfirmRedisDAO.addToConfirmQueue(messageRecordDO);
systemQuotaRedisDAO.set(messageRecordDO.getChannel(), messageRecordDO.getAppName()); // 仅首发(retrySource==null)消耗每日配额/接收者频率计数;重发(AUTO_RETRY/MANUAL_RETRY)
rateLimitRedisDAO.set(messageRecordDO.getChannel(), messageRecordDO.getAppName(), messageRecordDO.getReceiver()); // 是同一条消息的再次投递,不重复扣额——与守卫豁免配套,避免重发把自己的配额耗尽再被拦。
if (retrySource == null) {
systemQuotaRedisDAO.set(messageRecordDO.getChannel(), messageRecordDO.getAppName());
rateLimitRedisDAO.set(messageRecordDO.getChannel(), messageRecordDO.getAppName(), messageRecordDO.getReceiver());
}
} else if (SendOutcome.RETRYABLE_FAILED.equals(sendResult.getOutcome())) { } else if (SendOutcome.RETRYABLE_FAILED.equals(sendResult.getOutcome())) {
messageRetryQueueService.saveOrUpdateRetryMessage(messageRecordDO); messageRetryQueueService.saveOrUpdateRetryMessage(messageRecordDO);
channelProviderConfigService.failureUpdate(messageRecordDO.getProviderType(), messageRecordDO.getChannel()); channelProviderConfigService.failureUpdate(messageRecordDO.getProviderType(), messageRecordDO.getChannel());

View File

@@ -23,10 +23,12 @@ import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional; import org.springframework.transaction.annotation.Transactional;
import java.time.Duration;
import java.time.LocalDateTime; import java.time.LocalDateTime;
import java.util.Arrays; import java.util.Arrays;
import java.util.Collections; import java.util.Collections;
import java.util.List; import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors; import java.util.stream.Collectors;
@@ -41,6 +43,13 @@ public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueM
private static final int PROCESS_STATUS_PROCESSING = 1; private static final int PROCESS_STATUS_PROCESSING = 1;
private static final int DEFAULT_LOCK_SECONDS = 120; 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> CHANNELS = Arrays.asList( private static final List<String> CHANNELS = Arrays.asList(
ChannelTypeEnum.SMS.getCode(), ChannelTypeEnum.SMS.getCode(),
ChannelTypeEnum.EMAIL.getCode(), ChannelTypeEnum.EMAIL.getCode(),
@@ -67,13 +76,15 @@ public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueM
retryRecord.setMessageId(message.getId()); retryRecord.setMessageId(message.getId());
retryRecord.setChannel(message.getChannel()); retryRecord.setChannel(message.getChannel());
retryRecord.setReceiver(message.getReceiver()); retryRecord.setReceiver(message.getReceiver());
retryRecord.setRetryCount(1); boolean rateLimit = isRateLimit(message);
// I3:首次失败若是限流,普通失败计数从 0 起(限流不占普通次数预算);普通失败则从 1 起
retryRecord.setRetryCount(rateLimit ? 0 : 1);
RetryStrategyConfigDO strategyConfig = retryStrategyConfigService.getStrategyConfig(message.getChannel()); RetryStrategyConfigDO strategyConfig = retryStrategyConfigService.getStrategyConfig(message.getChannel());
retryRecord.setMaxRetry(ObjectUtil.isNull(strategyConfig) ? DEFAULT_MAX_RETRY_COUNT : strategyConfig.getMaxRetryCount()); retryRecord.setMaxRetry(ObjectUtil.isNull(strategyConfig) ? DEFAULT_MAX_RETRY_COUNT : strategyConfig.getMaxRetryCount());
retryRecord.setFirstFailTime(message.getSendTime()); retryRecord.setFirstFailTime(message.getSendTime());
retryRecord.setLastRetryTime(message.getSendTime()); retryRecord.setLastRetryTime(message.getSendTime());
retryRecord.setLastErrorMsg(message.getErrorMsg()); retryRecord.setLastErrorMsg(message.getErrorMsg());
LocalDateTime nextRetryTime = calculateNextRetryTime(message.getChannel(), retryRecord.getRetryCount()); LocalDateTime nextRetryTime = calculateNextRetryTime(message.getChannel(), retryRecord.getRetryCount(), rateLimit);
retryRecord.setNextRetryTime(nextRetryTime); retryRecord.setNextRetryTime(nextRetryTime);
retryRecord.setProcessStatus(PROCESS_STATUS_WAITING); retryRecord.setProcessStatus(PROCESS_STATUS_WAITING);
retryRecord.setLockUntil(null); retryRecord.setLockUntil(null);
@@ -85,11 +96,20 @@ public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueM
return; return;
} }
int newRetryCount = existing.getRetryCount() + 1; applyRetryUpdate(message, existing);
}
private void applyRetryUpdate(MessageRecordDO message, MessageRetryQueueDO existing) {
boolean rateLimit = isRateLimit(message);
// I3:限流重试冻结 retryCount(不消耗普通失败的次数预算),普通失败才累加
int newRetryCount = nextRetryCount(existing.getRetryCount(), rateLimit);
existing.setLastRetryTime(message.getSendTime()); existing.setLastRetryTime(message.getSendTime());
existing.setLastErrorMsg(message.getErrorMsg()); existing.setLastErrorMsg(message.getErrorMsg());
if (newRetryCount >= existing.getMaxRetry()) { boolean exhausted = rateLimit
? isRateLimitExhausted(existing.getFirstFailTime(), LocalDateTime.now())
: newRetryCount >= existing.getMaxRetry();
if (exhausted) {
message.setStatus(MsgStatusConstant.FINALFAILED); message.setStatus(MsgStatusConstant.FINALFAILED);
message.setNextRetryTime(null); message.setNextRetryTime(null);
existing.setNextRetryTime(null); existing.setNextRetryTime(null);
@@ -99,7 +119,7 @@ public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueM
return; return;
} }
LocalDateTime nextRetryTime = calculateNextRetryTime(message.getChannel(), newRetryCount); LocalDateTime nextRetryTime = calculateNextRetryTime(message.getChannel(), newRetryCount, rateLimit);
existing.setRetryCount(newRetryCount); existing.setRetryCount(newRetryCount);
existing.setNextRetryTime(nextRetryTime); existing.setNextRetryTime(nextRetryTime);
existing.setProcessStatus(PROCESS_STATUS_WAITING); existing.setProcessStatus(PROCESS_STATUS_WAITING);
@@ -111,7 +131,36 @@ public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueM
} }
} }
private LocalDateTime calculateNextRetryTime(String channel, int retryCount) { /**
* 计算本次失败后队列记录的新 retryCount。
* I3:限流重试不计入次数预算 —— 限流时冻结(返回原值),普通失败时累加。
* 这样普通失败的 maxRetry 上限只统计普通失败,既不被限流重试撑爆提前 FINALFAILED,
* 也不会让普通失败借限流的 12h 兜底绕过次数上限。
*/
static int nextRetryCount(int currentRetryCount, boolean rateLimit) {
return rateLimit ? currentRetryCount : currentRetryCount + 1;
}
/** 限流类识别:仅当统一错误码为 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;
}
LocalDateTime calculateNextRetryTime(String channel, int retryCount, boolean rateLimit) {
if (rateLimit) {
return LocalDateTime.now().plusSeconds(RATE_LIMIT_INTERVAL_SECONDS);
}
RetryStrategyConfigDO strategyConfig = retryStrategyConfigService.getStrategyConfig(channel); RetryStrategyConfigDO strategyConfig = retryStrategyConfigService.getStrategyConfig(channel);
long plusSeconds = 0; long plusSeconds = 0;
if (ObjectUtil.isNull(strategyConfig)) { if (ObjectUtil.isNull(strategyConfig)) {
@@ -196,6 +245,7 @@ public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueM
messageRecordService.updateRetryCount(messageRecordDO.getId()); messageRecordService.updateRetryCount(messageRecordDO.getId());
finalizeRetryLease(messageRecordDO); finalizeRetryLease(messageRecordDO);
} else { } else {
// 能被重新认领说明 120s 租约已过期、20s 回执窗口早过 —— SENDING/PENDING 即卡住,与终态一并清理
deleteByMessageIds(Collections.singletonList(messageRecordDO.getId())); deleteByMessageIds(Collections.singletonList(messageRecordDO.getId()));
} }
} }
@@ -261,6 +311,8 @@ public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueM
log.warn("渠道 [{}] 的消息记录不存在", channel); log.warn("渠道 [{}] 的消息记录不存在", channel);
continue; continue;
} }
// push_message_record.next_retry_time 不保证落库,用队列记录(权威调度时间)回填后再入队,避免 NPE
fillNextRetryTimeFromQueue(messages, dbRecords);
int successCount = messageRetryRedisDAO.batchAddToRetryQueue(messages); int successCount = messageRetryRedisDAO.batchAddToRetryQueue(messages);
log.info("渠道 [{}] 重建完成:数据库记录数={}, 成功入队 Redis 数={}", channel, dbRecords.size(), successCount); log.info("渠道 [{}] 重建完成:数据库记录数={}, 成功入队 Redis 数={}", channel, dbRecords.size(), successCount);
@@ -329,12 +381,31 @@ public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueM
if (CollUtil.isEmpty(messages)) { if (CollUtil.isEmpty(messages)) {
return; return;
} }
// push_message_record.next_retry_time 不保证落库,用队列记录(权威调度时间)回填后再入队,避免 NPE
fillNextRetryTimeFromQueue(messages, dbRecords);
if (messageRetryRedisDAO.isRedisAvailable()) { if (messageRetryRedisDAO.isRedisAvailable()) {
messageRetryRedisDAO.batchAddToRetryQueue(messages); messageRetryRedisDAO.batchAddToRetryQueue(messages);
} }
} }
/**
* 用队列记录(push_message_retry_queue,调度权威)的 nextRetryTime 回填消息对象。
* push_message_record.next_retry_time 只在内存设置、不保证持久化,批量入队若直接取它会 NPE。
* 这些 message 仅用于 Redis 入队、不会回写库,故内存回填安全。
*/
private void fillNextRetryTimeFromQueue(List<MessageRecordDO> messages, List<MessageRetryQueueDO> queueRecords) {
Map<Long, LocalDateTime> nextRetryTimeMap = queueRecords.stream()
.filter(r -> r.getNextRetryTime() != null)
.collect(Collectors.toMap(MessageRetryQueueDO::getMessageId, MessageRetryQueueDO::getNextRetryTime, (a, b) -> a));
messages.forEach(m -> {
LocalDateTime queueTime = nextRetryTimeMap.get(m.getId());
if (queueTime != null) {
m.setNextRetryTime(queueTime);
}
});
}
/** /**
* 通过设置lockUntil时间、process_status=1来从数据库层抢占该条重试消息。 * 通过设置lockUntil时间、process_status=1来从数据库层抢占该条重试消息。
*/ */
@@ -345,18 +416,27 @@ public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueM
/** /**
* 完成重试租约处理,根据消息状态执行相应的清理或重置操作 * 完成重试租约处理,根据消息状态执行相应的清理或重置操作
* - 如果消息状态为可重试失败,直接返回,不进行任何处理 * - 如果消息状态为可重试失败,直接返回,不进行任何处理
* - SENDING(两段式中间态)直接返回,保留记录与租约,等回执推进
* - 如果消息状态为待处理重置队列记录的处理状态和锁时间并将消息重新加入Redis重试队列 * - 如果消息状态为待处理重置队列记录的处理状态和锁时间并将消息重新加入Redis重试队列
* - 其他状态成功、最终失败等从Redis重试队列移除并删除数据库中的重试队列记录 * - 其他状态成功、最终失败等从Redis重试队列移除并删除数据库中的重试队列记录
* *
* @param messageRecordDO 消息记录对象 * @param messageRecordDO 消息记录对象
*/ */
private void finalizeRetryLease(MessageRecordDO messageRecordDO) { private void finalizeRetryLease(MessageRecordDO messageRecordDO) {
if (MsgStatusConstant.RETRYABLE_FAILED.equals(messageRecordDO.getStatus())) { String status = messageRecordDO.getStatus();
// 可重试失败:保留记录,等下一轮认领重发 —— 不变
if (MsgStatusConstant.RETRYABLE_FAILED.equals(status)) {
return; return;
} }
// “处理过程中断,但消息仍停留在 PENDING 时,主动释放 lease 并回队列”的兜底补偿逻辑 // 两段式发送的中间态:重发已受理、消息在途,保留记录、保持租约锁住、不回 Redis,等 20s 回执推进
if (MsgStatusConstant.PENDING.equals(messageRecordDO.getStatus())) { if (MsgStatusConstant.SENDING.equals(status)) {
return;
}
// 处理过程中断但仍停留在 PENDING:主动释放 lease 并回队列(兜底补偿)—— 不变
if (MsgStatusConstant.PENDING.equals(status)) {
lambdaUpdate() lambdaUpdate()
.eq(MessageRetryQueueDO::getMessageId, messageRecordDO.getId()) .eq(MessageRetryQueueDO::getMessageId, messageRecordDO.getId())
.set(MessageRetryQueueDO::getProcessStatus, PROCESS_STATUS_WAITING) .set(MessageRetryQueueDO::getProcessStatus, PROCESS_STATUS_WAITING)
@@ -368,6 +448,7 @@ public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueM
return; return;
} }
// 其余即终态:从 Redis 移除并删除数据库重试队列记录
messageRetryRedisDAO.removeFromRetryQueue(messageRecordDO.getChannel(), messageRecordDO.getId()); messageRetryRedisDAO.removeFromRetryQueue(messageRecordDO.getChannel(), messageRecordDO.getId());
deleteByMessageIds(Collections.singletonList(messageRecordDO.getId())); deleteByMessageIds(Collections.singletonList(messageRecordDO.getId()));
} }

View File

@@ -0,0 +1,29 @@
-- ============================================================================
-- 电信短信「短期频繁发送」限流码 MA:0038 配置为可重试
-- ----------------------------------------------------------------------------
-- 背景:电信下行回执查询(action=select)对该类消息返回 status=6 / stat=MA:0038。
-- 1) 代码侧已在 TelecomSmsSender.getDownInfo 把 status=6 纳入失败映射分支;
-- 2) 本脚本在错误码映射表为 MA:0038 配置 should_retry=1
-- 使其经 Sender.buildFailureResult 判定为 RETRYABLE_FAILED进入重试队列。
-- 两者缺一不可。SMS 首次重试间隔默认 5 分钟,足以越过服务商 1 分钟限流窗口。
--
-- 该脚本幂等:已存在同 (provider_type, channel, original_code) 的有效记录时不重复插入。
-- ============================================================================
INSERT INTO push_provider_error_code_mapping
(provider_type, channel, original_code, original_message,
unified_error_code, unified_error_message, error_category,
should_retry, final_status, remark,
creator, updater, create_time, update_time, deleted)
SELECT
'telecom', 'sms', 'MA:0038', '短期频繁发送',
'RATE_LIMIT', '服务商短期频繁发送限流', 'PROVIDER_ERROR',
1, 'retryable_failed', '电信限流码,需自动重试',
'system', 'system', NOW(), NOW(), 0
FROM DUAL
WHERE NOT EXISTS (
SELECT 1 FROM push_provider_error_code_mapping
WHERE provider_type = 'telecom'
AND channel = 'sms'
AND original_code = 'MA:0038'
AND deleted = 0
);

View File

@@ -0,0 +1,46 @@
-- ============================================================================
-- 普通类短信失败的重试阶梯配置(rootfix 后)
-- ----------------------------------------------------------------------------
-- 背景:限流类(电信 MA:0038)已由代码硬编码 120s 间隔 + 12h 兜底,忽略本表。
-- 故 push_retry_strategy_config.retry_intervals 现仅作用于「普通类」sms 可
-- 重试失败(超时 / 通道故障 / 第三方调用失败等),与限流类无关。
--
-- 决策(TD-3,2026-06-20):普通类用收紧的阶梯退避,替代旧 quickfix 的 60s 平铺。
-- retry_intervals = '60,120,300,600' → 1→2→5→10 分钟,超出后按 10 分钟重复
-- max_retry_count = 6 → 第 6 次普通失败转 final_failed(约 28 分钟放弃)
--
-- 幂等:可重复执行。⚠ retry_intervals 不能含空格(代码 Long.parseLong 不容忍空格)。
-- 关系:本文件取代 sms_ratelimit_retry_quickfix.sql 的 ② 段;
-- ① 错误码映射见 provider-error-code-mapping/telecom_sms_MA0038_retryable.sql。
-- ============================================================================
-- 1) 已存在启用中的 sms 策略行 → 覆盖为新阶梯
-- (同时纠正 max_retry_count,覆盖旧 quickfix 写入的 '60' / max 10)
UPDATE push_retry_strategy_config
SET retry_intervals = '60,120,300,600',
max_retry_count = 6,
update_time = NOW()
WHERE channel = 'sms'
AND enabled = 1
AND deleted = 0;
-- 2) 没有启用中的 sms 策略行 → 插一条(否则代码回落硬编码 300/600/1800、max 5)
INSERT INTO push_retry_strategy_config
(channel, max_retry_count, retry_intervals, enabled,
creator, updater, create_time, update_time, deleted)
SELECT 'sms', 6, '60,120,300,600', 1,
'system', 'system', NOW(), NOW(), 0
FROM DUAL
WHERE NOT EXISTS (
SELECT 1 FROM push_retry_strategy_config
WHERE channel = 'sms'
AND enabled = 1
AND deleted = 0
);
-- ----------------------------------------------------------------------------
-- 执行后自检(可选):确认普通类阶梯已就位
-- ----------------------------------------------------------------------------
-- SELECT channel, max_retry_count, retry_intervals, enabled
-- FROM push_retry_strategy_config
-- WHERE channel = 'sms' AND enabled = 1 AND deleted = 0;

View File

@@ -0,0 +1,79 @@
package com.njcn.msgpush.module.push.service.retry;
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 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);
}
@Test
void nextRetryCount_rateLimitFreezes_normalIncrements() {
// 普通失败:累加,消耗次数预算
assertEquals(2, MessageRetryQueueServiceImpl.nextRetryCount(1, false));
// 限流失败:冻结,不消耗次数预算(靠 12h 兜底终止)
assertEquals(1, MessageRetryQueueServiceImpl.nextRetryCount(1, true));
}
@Test
void nextRetryCount_rateLimitRetriesDoNotConsumeNormalBudget() {
// I3 回归:首次普通失败后,连续多次限流重试不应吃掉普通失败的次数预算
int rc = 1; // 首次普通失败入队
rc = MessageRetryQueueServiceImpl.nextRetryCount(rc, true); // 限流重试
rc = MessageRetryQueueServiceImpl.nextRetryCount(rc, true); // 限流重试
rc = MessageRetryQueueServiceImpl.nextRetryCount(rc, true); // 限流重试
assertEquals(1, rc, "限流重试不应增加普通失败计数");
rc = MessageRetryQueueServiceImpl.nextRetryCount(rc, false); // 第 2 次普通失败
assertEquals(2, rc, "普通失败才累加,且计数从未被限流污染");
}
}