From 8231b1e0e1486e137e878d8cc9bdfc0d812d43b6 Mon Sep 17 00:00:00 2001
From: hongawen <83944980@qq.com>
Date: Fri, 19 Jun 2026 19:33:00 +0800
Subject: [PATCH] =?UTF-8?q?fix(retry):=20=E4=BF=AE=E5=A4=8D=E6=89=B9?=
 =?UTF-8?q?=E9=87=8F=E5=85=A5=E9=98=9F=20NPE(next=5Fretry=5Ftime=20?=
 =?UTF-8?q?=E5=8F=96=E9=94=99=E6=95=B0=E6=8D=AE=E6=BA=90)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

根因: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>
---
 .../push/dal/redis/MessageRetryRedisDAO.java  |  5 +++++
 .../retry/MessageRetryQueueServiceImpl.java   | 22 +++++++++++++++++++
 2 files changed, 27 insertions(+)

diff --git a/msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/dal/redis/MessageRetryRedisDAO.java b/msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/dal/redis/MessageRetryRedisDAO.java
index b5578d0..c24cc3d 100644
--- a/msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/dal/redis/MessageRetryRedisDAO.java
+++ b/msgpush-module-push/msgpush-module-push-server/src/main/java/com/njcn/msgpush/module/push/dal/redis/MessageRetryRedisDAO.java
@@ -88,6 +88,11 @@ public class MessageRetryRedisDAO {
      * @param 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());
         double score = message.getNextRetryTime().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
 
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 05bc800..fdbae57 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
@@ -28,6 +28,7 @@ import java.time.LocalDateTime;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
 
@@ -310,6 +311,8 @@ public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueM
                     log.warn("渠道 [{}] 的消息记录不存在", channel);
                     continue;
                 }
+                // push_message_record.next_retry_time 不保证落库,用队列记录(权威调度时间)回填后再入队,避免 NPE
+                fillNextRetryTimeFromQueue(messages, dbRecords);
 
                 int successCount = messageRetryRedisDAO.batchAddToRetryQueue(messages);
                 log.info("渠道 [{}] 重建完成：数据库记录数={}, 成功入队 Redis 数={}", channel, dbRecords.size(), successCount);
@@ -378,12 +381,31 @@ public class MessageRetryQueueServiceImpl extends ServiceImpl<MessageRetryQueueM
         if (CollUtil.isEmpty(messages)) {
             return;
         }
+        // push_message_record.next_retry_time 不保证落库,用队列记录(权威调度时间)回填后再入队,避免 NPE
+        fillNextRetryTimeFromQueue(messages, dbRecords);
 
         if (messageRetryRedisDAO.isRedisAvailable()) {
             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来从数据库层抢占该条重试消息。
      */