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

feat(retry): 批次3 限流类自动重试守卫豁免(让无限重试在生产真正生效)

Browse Source 
接 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>
This commit is contained in:
hongawen
2026-06-19 17:39:12 +08:00
parent 16974d04f9
commit f076dcd9b2
2 changed files with 20 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/checker/MsgPushGuardChain.java
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.RateLimitChecker;
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.stereotype.Component;
@@ -24,9 +25,15 @@ public class MsgPushGuardChain {
private RateLimitChecker rateLimitChecker;
public void checkAll(List<MessageRecordDO> messageRecordList) {
public void checkAll(List<MessageRecordDO> messageRecordList, RetrySourceEnum retrySource) {
// 黑名单:首发与重发(含限流自动重试)都必须执行
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);
}
}
}

16
msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/service/message/MessageRecordServiceImpl.java
View File

@@ -93,7 +93,7 @@ public class MessageRecordServiceImpl extends ServiceImpl<MessageRecordMapper, M
*/
@Override
public List<MessageSendResultVO> processSendMsg(List<MessageRecordDO> messageRecordDOList, RetrySourceEnum retrySource) {
msgPushGuardChain.checkAll(messageRecordDOList);
msgPushGuardChain.checkAll(messageRecordDOList, retrySource);
List<MessageSendResultVO> resultList = new ArrayList<>();
for (MessageRecordDO messageRecordDO : messageRecordDOList) {
try {
@@ -131,11 +131,11 @@ public class MessageRecordServiceImpl extends ServiceImpl<MessageRecordMapper, M
if (sendResult == null) {
sendResult = this.sendMessage(messageRecordDO, channelProviderConfigDO, messageProviderFactory);
}
this.applySendResult(messageRecordDO, sendResult);
this.applySendResult(messageRecordDO, sendResult, retrySource);
this.recordRetryHistory(messageRecordDO, retrySource);
} else {
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)
public void applySendResult(MessageRecordDO messageRecordDO, SendResult sendResult) {
public void applySendResult(MessageRecordDO messageRecordDO, SendResult sendResult, RetrySourceEnum retrySource) {
messageRecordDO.setStatus(this.mapOutcomeToStatus(sendResult.getOutcome()));
messageRecordDO.setErrorCode(sendResult.getErrorCode());
messageRecordDO.setErrorMsg(sendResult.getMessage());
@@ -194,8 +194,12 @@ public class MessageRecordServiceImpl extends ServiceImpl<MessageRecordMapper, M
if (SendOutcome.ACCEPTED.equals(sendResult.getOutcome())) {
messageConfirmRedisDAO.addToConfirmQueue(messageRecordDO);
systemQuotaRedisDAO.set(messageRecordDO.getChannel(), messageRecordDO.getAppName());
rateLimitRedisDAO.set(messageRecordDO.getChannel(), messageRecordDO.getAppName(), messageRecordDO.getReceiver());
// 仅首发(retrySource==null)消耗每日配额/接收者频率计数;重发(AUTO_RETRY/MANUAL_RETRY)
// 是同一条消息的再次投递,不重复扣额——与守卫豁免配套,避免重发把自己的配额耗尽再被拦。
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())) {
messageRetryQueueService.saveOrUpdateRetryMessage(messageRecordDO);
channelProviderConfigService.failureUpdate(messageRecordDO.getProviderType(), messageRecordDO.getChannel());