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

fix(retry): 修复 I3 限流/普通共用 retryCount 跨污染

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

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

@@ -75,13 +75,15 @@ public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueM
retryRecord.setMessageId(message.getId());
retryRecord.setChannel(message.getChannel());
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());
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));
LocalDateTime nextRetryTime = calculateNextRetryTime(message.getChannel(), retryRecord.getRetryCount(), rateLimit);
retryRecord.setNextRetryTime(nextRetryTime);
retryRecord.setProcessStatus(PROCESS_STATUS_WAITING);
retryRecord.setLockUntil(null);
@@ -97,11 +99,12 @@ public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueM
}
private void applyRetryUpdate(MessageRecordDO message, MessageRetryQueueDO existing) {
int newRetryCount = existing.getRetryCount() + 1;
boolean rateLimit = isRateLimit(message);
// I3:限流重试冻结 retryCount(不消耗普通失败的次数预算),普通失败才累加
int newRetryCount = nextRetryCount(existing.getRetryCount(), rateLimit);
existing.setLastRetryTime(message.getSendTime());
existing.setLastErrorMsg(message.getErrorMsg());
boolean rateLimit = isRateLimit(message);
boolean exhausted = rateLimit
? isRateLimitExhausted(existing.getFirstFailTime(), LocalDateTime.now())
: newRetryCount >= existing.getMaxRetry();
@@ -127,6 +130,16 @@ public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueM
}
}
/**
* 计算本次失败后队列记录的新 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());

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

@@ -6,6 +6,7 @@ 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;
@@ -55,4 +56,24 @@ class RetryDecisionLogicTest {
// 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, "普通失败才累加,且计数从未被限流污染");
}
}