fix(mq-starter): 去重跳过区分 processing/success，processing 残留不 ACK 避免静默丢消息

tryAcquire 返回 false 的两类跳过此前一律静默 return，导致 driver 无差别 ACK：
消费者在 markSuccess/markFail 前崩溃残留的 processing 标记，会被当成"已处理"
而 ACK，该消息从未成功消费却被丢弃（Critical 静默丢消息）。

- MqIdempotentStore 新增 isCompleted(key)，默认 true 保持旧行为，状态机实现须覆写
- MqConsumeDispatcher：跳过时按 isCompleted 分流——success 正常 return 让 driver ACK；
  仍 processing 则抛 MqMessageInProgressException（新，纯控制流信号、不带栈）让 driver 不 ACK
- RedisMqIdempotentStore.isCompleted 仅 success 视为完成；RedisStreamMqDriver 捕获信号异常时留 PEL 待重投
- 补 dispatcher 两类跳过 + store isCompleted 语义单测

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

mq-starter-core/src/main/java/com/njcn/mq/container/MqConsumeDispatcher.java

@@ -4,6 +4,7 @@ import com.njcn.mq.core.MqMessage;
 import com.njcn.mq.spi.MqConsumeErrorHandler;
 import com.njcn.mq.spi.MqErrorIdentity;
 import com.njcn.mq.spi.MqIdempotentStore;
+import com.njcn.mq.spi.MqMessageInProgressException;
 import lombok.extern.slf4j.Slf4j;
 
 import java.util.ArrayList;
@@ -35,7 +36,15 @@ public class MqConsumeDispatcher {
 
     public void dispatch(MqMessage msg) throws Exception {
         if (idempotent && store != null && !store.tryAcquire(msg.getKey())) {
-            return; // processing 中或已 success，跳过
+            // 跳过分两类，决定 driver 能否 ACK：
+            //  - 已 success：正常返回 → driver ACK，挡掉重复投递（正常去重）；
+            //  - 仍 processing（并发处理中，或消费者在 mark 前崩溃残留的标记）：抛信号异常 →
+            //    driver 不 ACK、保留消息，待 processing 标记过期后真正重投。否则该消息可能从未被
+            //    成功消费却被去重挡回并 ACK → 静默丢消息。
+            if (store.isCompleted(msg.getKey())) {
+                return;
+            }
+            throw new MqMessageInProgressException(msg.getKey());
         }
         if (batchSize <= 1) {
             try {

mq-starter-core/src/main/java/com/njcn/mq/spi/MqIdempotentStore.java

@@ -21,4 +21,22 @@ public interface MqIdempotentStore {
 
     /** 消费失败,标记 fail(允许后续重新消费)。 */
     void markFail(String key);
+
+    /**
+     * {@link #tryAcquire(String)} 返回 false(跳过)后,判断该 key 是否已<b>消费成功</b>。
+     *
+     * <p>用于让 driver 区分两类"跳过",这决定了 driver 能否 ACK 该消息:
+     * <ul>
+     *   <li>已 success:可安全 ACK —— 正常去重,挡掉重复投递;</li>
+     *   <li>仍 processing(并发处理中,或消费者在 mark 前崩溃残留的标记):<b>不可 ACK</b> ——
+     *       该消息可能从未被成功消费,必须保留在 MQ,待 processing 标记过期后真正重投;
+     *       否则会被去重挡回并 ACK,造成<b>静默丢消息</b>。</li>
+     * </ul>
+     *
+     * @return true=已成功(driver 可 ACK);false=processing 中(driver 应保留待重投)。
+     *         默认返回 true 以保持旧行为(向后兼容);具备状态机的实现应覆写。
+     */
+    default boolean isCompleted(String key) {
+        return true;
+    }
 }

mq-starter-core/src/main/java/com/njcn/mq/spi/MqMessageInProgressException.java

@@ -0,0 +1,21 @@
+package com.njcn.mq.spi;
+
+/**
+ * 控制流信号:消费去重标记仍处于 {@code processing}(并发处理中,或消费者在 mark 前崩溃残留的标记)。
+ *
+ * <p>本次投递既不能视为成功(否则 driver 会 ACK,导致该消息若从未被成功消费即<b>静默丢失</b>),
+ * 也不属于业务处理失败。driver 捕获后应<b>保留消息(不 ACK)</b>,待 processing 标记过期后由 MQ 真正重投。
+ *
+ * <p>不携带堆栈(纯控制流,规避噪音与开销)。
+ */
+public class MqMessageInProgressException extends RuntimeException {
+
+    public MqMessageInProgressException(String key) {
+        super("message in-progress (dedup processing), leave for replay: key=" + key);
+    }
+
+    @Override
+    public synchronized Throwable fillInStackTrace() {
+        return this;
+    }
+}

mq-starter-core/src/test/java/com/njcn/mq/container/MqConsumeDispatcherTest.java

@@ -143,6 +143,50 @@ class MqConsumeDispatcherTest {
         assertEquals(2, succeeded.size(), "flush 成功后应对每条标记 success");
     }
 
+    @Test
+    void idempotent_skipInProgress_throwsSoDriverWontAckAndReplays() {
+        // 崩溃残留场景:去重标记停在 processing(tryAcquire=false 且 isCompleted=false)。
+        // 消费者在 tryAcquire 之后、markSuccess/markFail 之前崩溃,消息从未被成功消费。
+        // 若此时静默返回,driver 会把这条消息 ACK → 静默丢消息(Critical)。
+        // 正确行为:抛异常,使 driver 不 ACK、保留消息,待 processing 标记过期后真正重投。
+        List<List<MqMessage>> consumed = new ArrayList<>();
+        MqIdempotentStore store = new MqIdempotentStore() {
+            @Override public boolean tryAcquire(String key) { return false; }
+            @Override public void markSuccess(String key) { }
+            @Override public void markFail(String key) { }
+            @Override public boolean isCompleted(String key) { return false; }
+        };
+        MqConsumeDispatcher d = MqConsumeDispatcher.builder()
+                .batchSize(1).idempotent(true).store(store)
+                .consumer(consumed::add)
+                .build();
+
+        assertThrows(Exception.class,
+                () -> d.dispatch(MqMessage.of("k1", "", "a")),
+                "processing 残留(非 success)跳过时应抛异常,使 driver 不 ACK、留待过期后重投,避免静默丢消息");
+        assertEquals(0, consumed.size(), "不应消费");
+    }
+
+    @Test
+    void idempotent_skipCompleted_returnsNormallySoDriverAcks() {
+        // 已成功的重复投递:应被静默去重(driver 正常 ACK),不抛异常、不重复消费。
+        List<List<MqMessage>> consumed = new ArrayList<>();
+        MqIdempotentStore store = new MqIdempotentStore() {
+            @Override public boolean tryAcquire(String key) { return false; }
+            @Override public void markSuccess(String key) { }
+            @Override public void markFail(String key) { }
+            @Override public boolean isCompleted(String key) { return true; }
+        };
+        MqConsumeDispatcher d = MqConsumeDispatcher.builder()
+                .batchSize(1).idempotent(true).store(store)
+                .consumer(consumed::add)
+                .build();
+
+        assertDoesNotThrow(() -> d.dispatch(MqMessage.of("k1", "", "a")),
+                "已成功的重复投递应被静默去重(driver ACK),不抛异常");
+        assertEquals(0, consumed.size(), "不应重复消费");
+    }
+
     @Test
     void singleMode_whenErrorHandlerThrows_stillMarksFailAndRethrowsOriginal() {
         List<String> failed = new ArrayList<>();

mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisMqIdempotentStore.java

@@ -59,4 +59,11 @@ public class RedisMqIdempotentStore implements MqIdempotentStore {
     public void markFail(String key) {
         redis.opsForValue().set(prefix + key, FAIL, Duration.ofSeconds(failTtl));
     }
+
+    @Override
+    public boolean isCompleted(String key) {
+        // 仅 success 视为"已完成"(driver 可 ACK)。processing(并发处理中/崩溃残留)与
+        // 无记录(标记已过期)一律 false，使 driver 保留消息、待重投，避免静默丢消息。
+        return SUCCESS.equals(redis.opsForValue().get(prefix + key));
+    }
 }

mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriver.java

@@ -6,6 +6,7 @@ import com.njcn.middle.stream.constant.StreamMessageConstant;
 import com.njcn.middle.stream.support.StreamKeyBuilder;
 import com.njcn.mq.core.MqMessage;
 import com.njcn.mq.spi.MqDriver;
+import com.njcn.mq.spi.MqMessageInProgressException;
 import com.njcn.mq.spi.MqSubscription;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.data.redis.connection.stream.*;
@@ -98,6 +99,11 @@ public class RedisStreamMqDriver implements MqDriver {
         try {
             sub.getListener().onMessage(msg);
             return true;
+        } catch (MqMessageInProgressException inProgress) {
+            // 去重标记仍 processing（并发处理中/上次崩溃残留）：不 ACK，留 PEL，
+            // 待 processing 标记过期后由 reclaimPending 重新 acquire 并真正重投。
+            log.debug("[mq-redis] in-progress, leave PEL key={}", msg.getKey());
+            return false;
         } catch (Exception e) {
             log.warn("[mq-redis] handle failed key={}, leave PEL for retry (max={})", msg.getKey(), maxRetry, e);
             return false; // 留 PEL，由 reclaimPending 重投或落库

mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisMqIdempotentStoreTest.java

@@ -0,0 +1,34 @@
+package com.njcn.mq.driver.redis;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.data.redis.core.ValueOperations;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+/**
+ * {@link RedisMqIdempotentStore#isCompleted(String)} 纯单测(mock StringRedisTemplate，不依赖真实 Redis）。
+ *
+ * <p>isCompleted 必须仅在 key 值为 {@code success} 时返回 true：processing / 无记录 一律 false，
+ * 否则消费者崩溃后残留的 processing 标记会被 driver 当成"已完成"而 ACK，造成静默丢消息。
+ */
+class RedisMqIdempotentStoreTest {
+
+    @Test
+    void isCompleted_trueOnlyForSuccessValue() {
+        StringRedisTemplate redis = mock(StringRedisTemplate.class);
+        @SuppressWarnings("unchecked")
+        ValueOperations<String, String> ops = mock(ValueOperations.class);
+        when(redis.opsForValue()).thenReturn(ops);
+        RedisMqIdempotentStore store = new RedisMqIdempotentStore(redis, "mq:idem:", 60, 300, 300);
+
+        when(ops.get("mq:idem:kSucc")).thenReturn("success");
+        when(ops.get("mq:idem:kProc")).thenReturn("processing");
+        when(ops.get("mq:idem:kNone")).thenReturn(null);
+
+        assertTrue(store.isCompleted("kSucc"), "success 应视为已完成，driver 可 ACK");
+        assertFalse(store.isCompleted("kProc"), "processing 不可视为完成，否则崩溃残留会被 ACK 丢消息");
+        assertFalse(store.isCompleted("kNone"), "无记录(标记已过期)不可视为完成");
+    }
+}