fix(mq-starter-core): 打烊二次检查拒收时复位 processing 标记

review 发现 #6(终审残余观察①的 follow-up):批量分支锁内的 closed 二次
检查发生在 tryAcquire 之后,拒收抛 InProgress 时去重标记已置 processing
却无人复位。重投每次都撞 stale 标记再抛 InProgress,白烧重试计数直到 TTL
过期;redis-stream 端若 deliveryCount 先超 maxRetry 且未配 errorHandler,
业务一次没跑就被判耗尽、静默丢弃。

修复:锁内只判定拒收,锁外 markFail 复位标记后再抛(markFail 是 Redis 往返,
不进 buffer 锁)。dispatch 入口的打烊快速检查在 tryAcquire 之前,不受影响。

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

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

@@ -83,18 +83,29 @@ public class MqConsumeDispatcher {
         }
         // 攒批：满 batchSize 才 flush。锁内只攒/取快照，消费放锁外，避免持锁做 IO。
         List<MqMessage> flush = null;
+        boolean rejected = false;
         synchronized (buffer) {
             if (closed) {
-                throw new MqMessageInProgressException(msg.getKey());
+                rejected = true;
+            } else {
+                if (buffer.isEmpty()) {
+                    batchStartAtMs = clock.getAsLong();
+                }
+                buffer.add(msg);
+                if (buffer.size() >= batchSize) {
+                    flush = new ArrayList<>(buffer);
+                    buffer.clear();
+                }
             }
-            if (buffer.isEmpty()) {
-                batchStartAtMs = clock.getAsLong();
-            }
-            buffer.add(msg);
-            if (buffer.size() >= batchSize) {
-                flush = new ArrayList<>(buffer);
-                buffer.clear();
+        }
+        if (rejected) {
+            // 打烊二次检查发生在 tryAcquire 之后：拒收前必须把 processing 标记复位（markFail 是
+            // Redis 往返，放 buffer 锁外做），否则重投每次都撞 stale 标记再抛 InProgress，白烧重试
+            // 计数——redis-stream 端可能业务一次没跑就被判耗尽。
+            if (idempotent && store != null) {
+                store.markFail(msg.getKey());
             }
+            throw new MqMessageInProgressException(msg.getKey());
         }
         if (flush != null) {
             doFlush(flush);

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

@@ -416,6 +416,37 @@ class MqConsumeDispatcherTest {
         assertEquals(1, consumed.size(), "拒收的消息不得进入缓冲(再次强刷无动作)");
     }
 
+    @Test
+    void closedDuringBatchDispatch_afterAcquire_resetsProcessingMark() {
+        // 竞态窗口:dispatch 已过打烊快速检查、tryAcquire 已把标记置 processing,
+        // 此刻停机 forceFlush 打烊 → 批量分支锁内二次检查拒收抛 InProgress。
+        // 拒收前必须 markFail 复位标记:否则重投每次都撞 stale processing 再抛 InProgress,
+        // 白烧重试计数(redis-stream 端 deliveryCount 超 maxRetry 且未配 errorHandler 时,
+        // 业务一次没跑就被判耗尽、静默丢弃)。
+        List<String> failed = new ArrayList<>();
+        MqConsumeDispatcher[] ref = new MqConsumeDispatcher[1];
+        MqIdempotentStore store = new MqIdempotentStore() {
+            @Override public boolean tryAcquire(String key) {
+                ref[0].forceFlush(); // 模拟 tryAcquire 成功瞬间停机打烊
+                return true;
+            }
+            @Override public void markSuccess(String key) { }
+            @Override public void markFail(String key) { failed.add(key); }
+        };
+        MqConsumeDispatcher d = MqConsumeDispatcher.builder()
+                .batchSize(3).batchTimeoutMs(60_000)
+                .idempotent(true).store(store)
+                .consumer(batch -> { })
+                .build();
+        ref[0] = d;
+
+        assertThrows(MqMessageInProgressException.class,
+                () -> d.dispatch(MqMessage.of("k1", "", "a")),
+                "打烊后批量路径锁内二次检查应拒收");
+        assertEquals(Arrays.asList("k1"), failed,
+                "已 tryAcquire 的拒收必须 markFail 复位,否则重投一直撞 processing 标记");
+    }
+
     @Test
     void closed_rejectsSingleModeToo_beforeTouchingDedupStore() {
         List<String> acquired = new ArrayList<>();