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 d28363a..5184e74 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,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 INTERMEDIATE_STATES = Arrays.asList( + MsgStatusConstant.SENDING, + MsgStatusConstant.PENDING, + MsgStatusConstant.RETRYABLE_FAILED); + private static final List CHANNELS = Arrays.asList( ChannelTypeEnum.SMS.getCode(), ChannelTypeEnum.EMAIL.getCode(), @@ -63,33 +77,51 @@ public class MessageRetryQueueServiceImpl extends ServiceImpl() + .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= 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 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 new file mode 100644 index 0000000..78018a2 --- /dev/null +++ b/msgpush-module-push/msgpush-module-push-server/src/test/java/com/njcn/msgpush/module/push/service/retry/RetryDecisionLogicTest.java @@ -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); + } +}