feat: ack gb32960 after kafka boundary

modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/concurrency/DisruptorEventBus.java

@@ -14,6 +14,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.atomic.AtomicLong;
 
@@ -28,10 +29,12 @@ public final class DisruptorEventBus implements AutoCloseable {
     private static final Logger log = LoggerFactory.getLogger(DisruptorEventBus.class);
 
     private final Disruptor<VehicleEventSlot> disruptor;
+    private final List<EventSink> sinks;
     private final AtomicLong published = new AtomicLong();
     private final AtomicLong failed = new AtomicLong();
 
     public DisruptorEventBus(int ringBufferSize, String waitStrategyName, List<EventSink> sinks) {
+        this.sinks = List.copyOf(sinks);
         ThreadFactory tf = Thread.ofVirtual().name("ingest-bus-", 0).factory();
         this.disruptor = new Disruptor<>(
                 VehicleEventSlot::new,
@@ -40,14 +43,14 @@ public final class DisruptorEventBus implements AutoCloseable {
                 ProducerType.MULTI,
                 waitStrategy(waitStrategyName));
 
-        EventHandler<VehicleEventSlot>[] handlers = sinks.stream()
+        EventHandler<VehicleEventSlot>[] handlers = this.sinks.stream()
                 .map(this::toHandler)
                 .toArray(EventHandler[]::new);
         // 每个 sink 一个独立 handler，事件按扇出模式同时写 Kafka、event-file-store 等目标。
         disruptor.handleEventsWith(handlers);
         disruptor.start();
         log.info("DisruptorEventBus started ringBuffer={} wait={} sinks={}",
-                ringBufferSize, waitStrategyName, sinks.size());
+                ringBufferSize, waitStrategyName, this.sinks.size());
     }
 
     public void publish(VehicleEvent event) {
@@ -55,6 +58,18 @@ public final class DisruptorEventBus implements AutoCloseable {
         published.incrementAndGet();
     }
 
+    public CompletableFuture<Void> publishAndAwait(VehicleEvent event) {
+        published.incrementAndGet();
+        List<CompletableFuture<Void>> futures = sinks.stream()
+                .filter(sink -> sink.accepts(event))
+                .map(sink -> publishToSinkAndTrack(sink, event))
+                .toList();
+        if (futures.isEmpty()) {
+            return CompletableFuture.completedFuture(null);
+        }
+        return CompletableFuture.allOf(futures.toArray(CompletableFuture[]::new));
+    }
+
     @Override
     public void close() {
         disruptor.shutdown();
@@ -66,18 +81,28 @@ public final class DisruptorEventBus implements AutoCloseable {
             VehicleEvent e = slot.event;
             if (e == null || !sink.accepts(e)) return;
             try {
-                sink.publish(e).exceptionally(ex -> {
+                publishToSinkAndTrack(sink, e);
-                    // sink 异步失败只计数和打日志，不反向阻塞入站连接；生产侧靠 sink 自身重试/DLQ 保证可观测。
-                    failed.incrementAndGet();
-                    log.warn("sink {} publish failed", sink.name(), ex);
-                    return null;
-                });
             } finally {
                 if (endOfBatch) slot.clear();
             }
         };
     }
 
+    private CompletableFuture<Void> publishToSinkAndTrack(EventSink sink, VehicleEvent event) {
+        try {
+            return sink.publish(event).whenComplete((ignored, ex) -> {
+                if (ex == null) return;
+                // sink 异步失败只计数和打日志；await 调用方仍会收到异常并决定是否 ACK。
+                failed.incrementAndGet();
+                log.warn("sink {} publish failed", sink.name(), ex);
+            });
+        } catch (Throwable t) {
+            failed.incrementAndGet();
+            log.warn("sink {} publish failed", sink.name(), t);
+            return CompletableFuture.failedFuture(t);
+        }
+    }
+
     private static WaitStrategy waitStrategy(String name) {
         return switch (name == null ? "yielding" : name.toLowerCase()) {
             case "blocking" -> new BlockingWaitStrategy();

modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/dispatcher/Dispatcher.java

@@ -10,11 +10,16 @@ import com.lingniu.ingest.core.pipeline.InterceptorChain;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.time.Duration;
 import java.time.Instant;
+import java.util.ArrayList;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 
 /**
@@ -52,13 +57,61 @@ public final class Dispatcher {
     public void dispatch(RawFrame frame) {
         IngestContext ctx = new IngestContext(UUID.randomUUID().toString());
         try {
-            // 在 interceptor 之前发 RawArchive，保证原始字节被无条件落盘（dedup/rate-limit
+            DispatchPlan plan = mapFrameToEvents(frame, ctx);
-            // 不会过滤它），满足"原始可回放"目标。archive 的写盘靠下游 ArchiveEventSink 消费。
+            for (VehicleEvent e : plan.events()) {
-            RawArchiveLookup rawArchive = emitRawArchive(frame, ctx);
+                eventBus.publish(e);
+            }
+            if (plan.failure() != null) {
+                throw plan.failure();
+            }
+        } catch (Throwable t) {
+            log.error("dispatch failure traceId={}", ctx.traceId(), t);
+            interceptors.onError(t, ctx);
+        }
+    }
 
+    public CompletableFuture<Void> dispatchAndAwait(RawFrame frame) {
+        IngestContext ctx = new IngestContext(UUID.randomUUID().toString());
+        try {
+            DispatchPlan plan = mapFrameToEvents(frame, ctx);
+            CompletableFuture<?>[] futures = plan.events().stream()
+                    .map(eventBus::publishAndAwait)
+                    .toArray(CompletableFuture[]::new);
+            CompletableFuture<Void> boundary = CompletableFuture.allOf(futures);
+            if (plan.failure() == null) {
+                return boundary;
+            }
+            return boundary.handle((ignored, publishFailure) -> {
+                if (publishFailure != null) {
+                    throw new CompletionException(publishFailure);
+                }
+                throw new CompletionException(plan.failure());
+            });
+        } catch (Throwable t) {
+            log.error("dispatch failure traceId={}", ctx.traceId(), t);
+            interceptors.onError(t, ctx);
+            return CompletableFuture.failedFuture(t);
+        }
+    }
+
+    public CompletableFuture<Void> dispatchAndAwait(RawFrame frame, Duration timeout) {
+        CompletableFuture<Void> future = dispatchAndAwait(frame);
+        if (timeout == null || timeout.isZero() || timeout.isNegative()) {
+            return future;
+        }
+        return future.orTimeout(timeout.toMillis(), TimeUnit.MILLISECONDS);
+    }
+
+    private DispatchPlan mapFrameToEvents(RawFrame frame, IngestContext ctx) {
+        List<VehicleEvent> out = new ArrayList<>();
+        // 在 interceptor 之前发 RawArchive，保证原始字节被无条件落盘（dedup/rate-limit
+        // 不会过滤它），满足"原始可回放"目标。archive 的写盘靠下游 Sink 消费。
+        RawArchiveLookup rawArchive = appendRawArchive(frame, ctx, out);
+
+        try {
             if (!interceptors.before(frame, ctx)) {
                 log.debug("frame aborted: {}", ctx.abortReason());
-                return;
+                return new DispatchPlan(out, null);
             }
 
             List<HandlerDefinition> handlers = registry.resolve(
@@ -67,7 +120,7 @@ public final class Dispatcher {
                 log.debug("no handler for {} cmd=0x{} info=0x{}",
                         frame.protocolId(), Integer.toHexString(frame.command()),
                         Integer.toHexString(frame.infoType()));
-                return;
+                return new DispatchPlan(out, null);
             }
 
             for (HandlerDefinition def : handlers) {
@@ -79,12 +132,12 @@ public final class Dispatcher {
                 for (VehicleEvent e : events) {
                     e = enrichWithRawArchive(e, rawArchive);
                     interceptors.after(e, ctx);
-                    eventBus.publish(e);
+                    out.add(e);
                 }
             }
+            return new DispatchPlan(out, null);
         } catch (Throwable t) {
-            log.error("dispatch failure traceId={}", ctx.traceId(), t);
+            return new DispatchPlan(out, t);
-            interceptors.onError(t, ctx);
         }
     }
 
@@ -95,7 +148,7 @@ public final class Dispatcher {
      * <p>VIN 取自 sourceMeta 里的 {@code vin} key（由各入站适配器负责填充）；缺失时
      * 留空字符串，archive sink 会用 "unknown-vin" 占位保证 key 可解析。
      */
-    private RawArchiveLookup emitRawArchive(RawFrame frame, IngestContext ctx) {
+    private RawArchiveLookup appendRawArchive(RawFrame frame, IngestContext ctx, List<VehicleEvent> out) {
         byte[] bytes = frame.rawBytes();
         if (bytes == null || bytes.length == 0) return RawArchiveLookup.empty();
 
@@ -117,7 +170,7 @@ public final class Dispatcher {
                 frame.command(),
                 frame.infoType(),
                 bytes);
-        eventBus.publish(raw);
+        out.add(raw);
         return new RawArchiveLookup(eventId, key, RawArchiveKeys.logicalUri(key));
     }
 
@@ -182,4 +235,7 @@ public final class Dispatcher {
             return key == null || key.isBlank();
         }
     }
+
+    private record DispatchPlan(List<VehicleEvent> events, Throwable failure) {
+    }
 }

modules/protocols/protocol-gb32960/src/main/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandler.java

@@ -122,41 +122,91 @@ public class Gb32960ChannelHandler extends SimpleChannelInboundHandler<byte[]> {
             return;
         }
 
-        // 鉴权通过：按命令类型回 ACK，写成功后统一 dispatch 到通用管线。
+        // 鉴权通过：先完成命令侧状态变更/日志，再统一 dispatch 到通用管线。
+        // 需要成功 ACK 的已接收帧，等待 dispatch 的持久化边界完成后再回 ACK。
         // 平台登入鉴权失败会短路 return，不进 dispatch。
         switch (cmd) {
-            case VEHICLE_LOGIN -> handleVehicleLogin(ctx, msg, rawVin);
+            case VEHICLE_LOGIN -> logVehicleLogin(ctx, msg);
-            case VEHICLE_LOGOUT -> handleVehicleLogout(ctx, msg, rawVin);
+            case VEHICLE_LOGOUT -> logVehicleLogout(ctx, msg);
             case PLATFORM_LOGIN -> {
                 if (!handlePlatformLogin(ctx, msg, rawVin)) return;
             }
-            case PLATFORM_LOGOUT -> handlePlatformLogout(ctx, msg, rawVin);
+            case PLATFORM_LOGOUT -> logPlatformLogout(ctx);
-            case REALTIME_REPORT, RESEND_REPORT, HEARTBEAT -> handleReportOrHeartbeat(ctx, msg, rawVin);
+            case REALTIME_REPORT, RESEND_REPORT, HEARTBEAT -> logReportOrHeartbeat(ctx, msg);
-            case TIME_CALIBRATION -> handleTimeCalibration(ctx, msg, rawVin);
+            case TIME_CALIBRATION -> { }
             default -> logOtherFrame(ctx, msg);
         }
 
+        RawFrame rf = rawFrame(ctx, msg, frame);
+        if (!requiresDurableAck(cmd)) {
+            dispatcher.dispatch(rf);
+            return;
+        }
+
+        dispatcher.dispatchAndAwait(rf).whenComplete((ignored, ex) ->
+                ctx.executor().execute(() -> {
+                    if (ex != null) {
+                        log.warn("[gb32960] dispatch failed before ack peer={} vin={} cmd=0x{}",
+                                addr(ctx), vin, Integer.toHexString(cmd.code()), ex);
+                        ctx.close();
+                        return;
+                    }
+                    writeAckForCommand(ctx, msg, rawVin);
+                }));
+    }
+
+    private RawFrame rawFrame(ChannelHandlerContext ctx, Gb32960Message msg, byte[] frame) {
         Map<String, String> sourceMeta = new HashMap<>(4);
-        sourceMeta.put("vin", vin);
+        sourceMeta.put("vin", msg.header().vin());
         sourceMeta.put("peer", addr(ctx));
         String platformAccount = accessService.platformAccount(ctx.channel());
         if (platformAccount != null && !platformAccount.isBlank()) {
             sourceMeta.put("platformAccount", platformAccount);
         }
         // RawFrame.rawBytes 是后续 raw archive 和按 rawArchiveUri 回查的源头；不要在这里裁剪或重编码。
-        RawFrame rf = new RawFrame(
+        return new RawFrame(
                 ProtocolId.GB32960,
-                cmd.code(),
+                msg.header().command().code(),
                 0,
                 msg,
                 frame,
                 sourceMeta,
                 Instant.now());
-        dispatcher.dispatch(rf);
+    }
+
+    private static boolean requiresDurableAck(CommandType cmd) {
+        return switch (cmd) {
+            case VEHICLE_LOGIN,
+                 VEHICLE_LOGOUT,
+                 PLATFORM_LOGIN,
+                 PLATFORM_LOGOUT,
+                 REALTIME_REPORT,
+                 RESEND_REPORT,
+                 HEARTBEAT,
+                 TIME_CALIBRATION -> true;
+            default -> false;
+        };
+    }
+
+    private void writeAckForCommand(ChannelHandlerContext ctx, Gb32960Message msg, byte[] rawVin) {
+        switch (msg.header().command()) {
+            case VEHICLE_LOGIN -> writeVehicleLoginAck(ctx, msg, rawVin);
+            case VEHICLE_LOGOUT -> writeVehicleLogoutAck(ctx, msg, rawVin);
+            case PLATFORM_LOGIN -> writePlatformLoginAck(ctx, msg, rawVin);
+            case PLATFORM_LOGOUT -> writePlatformLogoutAck(ctx, msg, rawVin);
+            case REALTIME_REPORT, RESEND_REPORT, HEARTBEAT -> writeReportOrHeartbeatAck(ctx, msg, rawVin);
+            case TIME_CALIBRATION -> writeTimeCalibrationAck(ctx, msg, rawVin);
+            default -> { }
+        }
     }
 
     /** 0x01 车辆登入：按 §6.3.2 带原采集时间回成功 ACK + INFO 日志。 */
-    private void handleVehicleLogin(ChannelHandlerContext ctx, Gb32960Message msg, byte[] rawVin) {
+    private void logVehicleLogin(ChannelHandlerContext ctx, Gb32960Message msg) {
+        log.info("[gb32960] vehicle login peer={} vin={} protocolVersion={}",
+                addr(ctx), msg.header().vin(), msg.header().protocolVersion());
+    }
+
+    private void writeVehicleLoginAck(ChannelHandlerContext ctx, Gb32960Message msg, byte[] rawVin) {
         ackService.writeResponse(
                 ctx,
                 msg.header().protocolVersion(),
@@ -166,13 +216,14 @@ public class Gb32960ChannelHandler extends SimpleChannelInboundHandler<byte[]> {
                 eventOrNow(msg),
                 null,
                 "vehicle-login-ack");
-        log.info("[gb32960] vehicle login peer={} vin={} protocolVersion={}",
-                addr(ctx), msg.header().vin(), msg.header().protocolVersion());
     }
 
     /** 0x04 车辆登出：INFO 日志 + 带原采集时间回成功 ACK。 */
-    private void handleVehicleLogout(ChannelHandlerContext ctx, Gb32960Message msg, byte[] rawVin) {
+    private void logVehicleLogout(ChannelHandlerContext ctx, Gb32960Message msg) {
         log.info("[gb32960] vehicle logout peer={} vin={}", addr(ctx), msg.header().vin());
+    }
+
+    private void writeVehicleLogoutAck(ChannelHandlerContext ctx, Gb32960Message msg, byte[] rawVin) {
         ackService.writeResponse(ctx,
                 msg.header().protocolVersion(),
                 CommandType.VEHICLE_LOGOUT,
@@ -189,8 +240,8 @@ public class Gb32960ChannelHandler extends SimpleChannelInboundHandler<byte[]> {
      * <ul>
      *   <li>消息体解析缺失：关闭连接，返回 false 短路
      *   <li>鉴权失败：写 {@code 0x02 OTHER_ERROR} NACK + 关闭连接，返回 false 短路
-     *   <li>鉴权通过：把 username 钉到 channel attribute（供后续 vendor profile 路由），
+     *   <li>鉴权通过：把 username 钉到 channel attribute（供后续 vendor profile 路由），INFO 日志，
-     *       写成功 ACK + INFO 日志，返回 true 继续 dispatch
+     *       返回 true 继续 dispatch；成功 ACK 等 dispatch 持久化边界完成后再写
      * </ul>
      *
      * @return true 表示鉴权通过应继续 dispatch；false 表示调用方应 short-circuit return
@@ -221,6 +272,10 @@ public class Gb32960ChannelHandler extends SimpleChannelInboundHandler<byte[]> {
                 addr(ctx), pl.username(), pl.encryptRule(), pl.serialNo(), pl.loginTime(), result);
         // 把 username 钉到 channel attribute，供后续 0x02/0x03 帧的 vendor profile 路由
         accessService.bindPlatformAccount(ctx.channel(), pl.username());
+        return true;
+    }
+
+    private void writePlatformLoginAck(ChannelHandlerContext ctx, Gb32960Message msg, byte[] rawVin) {
         ackService.writeResponse(
                 ctx,
                 msg.header().protocolVersion(),
@@ -230,12 +285,14 @@ public class Gb32960ChannelHandler extends SimpleChannelInboundHandler<byte[]> {
                 eventOrNow(msg),
                 null,
                 "platform-login-ack");
-        return true;
     }
 
     /** 0x06 平台登出：INFO 日志 + 带原采集时间回成功 ACK。 */
-    private void handlePlatformLogout(ChannelHandlerContext ctx, Gb32960Message msg, byte[] rawVin) {
+    private void logPlatformLogout(ChannelHandlerContext ctx) {
         log.info("[gb32960] platform logout peer={}", addr(ctx));
+    }
+
+    private void writePlatformLogoutAck(ChannelHandlerContext ctx, Gb32960Message msg, byte[] rawVin) {
         ackService.writeResponse(ctx,
                 msg.header().protocolVersion(),
                 CommandType.PLATFORM_LOGOUT,
@@ -247,7 +304,7 @@ public class Gb32960ChannelHandler extends SimpleChannelInboundHandler<byte[]> {
     }
 
     /**
-     * 0x02/0x03/0x07 实时 / 补发 / 心跳：按 §6.3.2 回应答帧，保留原帧采集时间。
+     * 0x02/0x03/0x07 实时 / 补发 / 心跳：记录帧诊断；成功 ACK 在 dispatch 持久化边界后写回。
      * 部分车端实现严格按规范，没收到应答会停止后续上报或重连。ack-tag 按 cmd.code 区分便于日志筛选。
      *
      * <p>注意：应答成功不代表下游已完成持久化，只表示平台已接收并通过 Dispatcher 投递。
@@ -264,17 +321,9 @@ public class Gb32960ChannelHandler extends SimpleChannelInboundHandler<byte[]> {
      * </ul>
      * 帧内字段全量 JSON 仍保留 DEBUG。
      */
-    private void handleReportOrHeartbeat(ChannelHandlerContext ctx, Gb32960Message msg, byte[] rawVin) {
+    private void logReportOrHeartbeat(ChannelHandlerContext ctx, Gb32960Message msg) {
         CommandType cmd = msg.header().command();
         String platformAccount = accessService.platformAccount(ctx.channel());
-        ackService.writeResponse(ctx,
-                msg.header().protocolVersion(),
-                cmd,
-                ResponseFlag.SUCCESS,
-                rawVin,
-                eventOrNow(msg),
-                null,
-                "report-ack-0x" + Integer.toHexString(cmd.code()));
         if (cmd == CommandType.HEARTBEAT) {
             if (log.isDebugEnabled()) {
                 log.debug("[gb32960] HEARTBEAT peer={} vin={} platformAccount={}",
@@ -304,8 +353,20 @@ public class Gb32960ChannelHandler extends SimpleChannelInboundHandler<byte[]> {
         }
     }
 
+    private void writeReportOrHeartbeatAck(ChannelHandlerContext ctx, Gb32960Message msg, byte[] rawVin) {
+        CommandType cmd = msg.header().command();
+        ackService.writeResponse(ctx,
+                msg.header().protocolVersion(),
+                cmd,
+                ResponseFlag.SUCCESS,
+                rawVin,
+                eventOrNow(msg),
+                null,
+                "report-ack-0x" + Integer.toHexString(cmd.code()));
+    }
+
     /** 0x08 终端校时：平台应答 data 段为平台当前时间 6B（用 Instant.now()）。 */
-    private void handleTimeCalibration(ChannelHandlerContext ctx, Gb32960Message msg, byte[] rawVin) {
+    private void writeTimeCalibrationAck(ChannelHandlerContext ctx, Gb32960Message msg, byte[] rawVin) {
         ackService.writeResponse(ctx,
                 msg.header().protocolVersion(),
                 CommandType.TIME_CALIBRATION,

modules/protocols/protocol-gb32960/src/test/java/com/lingniu/ingest/protocol/gb32960/inbound/Gb32960ChannelHandlerAckBoundaryTest.java

@@ -0,0 +1,133 @@
+package com.lingniu.ingest.protocol.gb32960.inbound;
+
+import com.lingniu.ingest.api.event.VehicleEvent;
+import com.lingniu.ingest.api.sink.EventSink;
+import com.lingniu.ingest.core.concurrency.AsyncBatchExecutor;
+import com.lingniu.ingest.core.concurrency.DisruptorEventBus;
+import com.lingniu.ingest.core.dispatcher.Dispatcher;
+import com.lingniu.ingest.core.dispatcher.HandlerInvoker;
+import com.lingniu.ingest.core.dispatcher.HandlerRegistry;
+import com.lingniu.ingest.core.pipeline.InterceptorChain;
+import com.lingniu.ingest.protocol.gb32960.auth.Gb32960PlatformAuthorizer;
+import com.lingniu.ingest.protocol.gb32960.auth.Gb32960VinAuthorizer;
+import com.lingniu.ingest.protocol.gb32960.codec.Gb32960BodyParser;
+import com.lingniu.ingest.protocol.gb32960.codec.Gb32960DecoderTest;
+import com.lingniu.ingest.protocol.gb32960.codec.Gb32960MessageDecoder;
+import com.lingniu.ingest.protocol.gb32960.codec.InfoBlockParserRegistry;
+import com.lingniu.ingest.protocol.gb32960.codec.parser.v2016.PositionV2016BlockParser;
+import com.lingniu.ingest.protocol.gb32960.codec.parser.v2016.VehicleV2016BlockParser;
+import com.lingniu.ingest.protocol.gb32960.config.Gb32960Properties;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.embedded.EmbeddedChannel;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class Gb32960ChannelHandlerAckBoundaryTest {
+
+    @Test
+    void realtimeReportAckWaitsForDispatchFutureCompletion() {
+        try (DispatchHarness dispatch = new DispatchHarness()) {
+            EmbeddedChannel channel = new EmbeddedChannel(handlerWith(dispatch.dispatcher()));
+
+            assertThat(channel.writeInbound(Gb32960DecoderTest.buildRealtimeFrame("LTEST000000000001"))).isFalse();
+            assertThat((Object) channel.readOutbound()).isNull();
+
+            dispatch.completeDurability();
+            channel.runPendingTasks();
+
+            ByteBuf ack = channel.readOutbound();
+            assertThat(ack).isNotNull();
+            assertThat(ack.getUnsignedByte(2)).isEqualTo((short) 0x02);
+            assertThat(ack.getUnsignedByte(3)).isEqualTo((short) 0x01);
+            ack.release();
+            channel.finishAndReleaseAll();
+        }
+    }
+
+    @Test
+    void realtimeReportAckIsNotWrittenWhenDispatchFutureFails() {
+        try (DispatchHarness dispatch = new DispatchHarness()) {
+            EmbeddedChannel channel = new EmbeddedChannel(handlerWith(dispatch.dispatcher()));
+
+            assertThat(channel.writeInbound(Gb32960DecoderTest.buildRealtimeFrame("LTEST000000000001"))).isFalse();
+
+            dispatch.failDurability();
+            channel.runPendingTasks();
+
+            assertThat((Object) channel.readOutbound()).isNull();
+            assertThat(channel.isActive()).isFalse();
+            channel.finishAndReleaseAll();
+        }
+    }
+
+    private static Gb32960ChannelHandler handlerWith(Dispatcher dispatcher) {
+        return new Gb32960ChannelHandler(
+                decoder(),
+                dispatcher,
+                accessService(),
+                new Gb32960AckService(),
+                new Gb32960FrameDiagnostics(16));
+    }
+
+    private static Gb32960MessageDecoder decoder() {
+        return new Gb32960MessageDecoder(new Gb32960BodyParser(new InfoBlockParserRegistry(List.of(
+                new VehicleV2016BlockParser(),
+                new PositionV2016BlockParser()))));
+    }
+
+    private static Gb32960AccessService accessService() {
+        Gb32960Properties.Auth auth = new Gb32960Properties.Auth();
+        auth.setEnabled(false);
+        return new Gb32960AccessService(
+                new Gb32960VinAuthorizer(auth),
+                new Gb32960PlatformAuthorizer(auth.getPlatforms()));
+    }
+
+    private static final class DispatchHarness implements AutoCloseable {
+        private final ControlledSink sink = new ControlledSink();
+        private final DisruptorEventBus eventBus = new DisruptorEventBus(1024, "blocking", List.of(sink));
+        private final AsyncBatchExecutor batchExecutor = new AsyncBatchExecutor(eventBus::publish);
+        private final Dispatcher dispatcher = new Dispatcher(
+                new HandlerRegistry(),
+                new InterceptorChain(List.of()),
+                new HandlerInvoker(),
+                eventBus,
+                batchExecutor);
+
+        Dispatcher dispatcher() {
+            return dispatcher;
+        }
+
+        void completeDurability() {
+            sink.future.complete(null);
+        }
+
+        void failDurability() {
+            sink.future.completeExceptionally(new IllegalStateException("durability failed"));
+        }
+
+        @Override
+        public void close() {
+            batchExecutor.close();
+            eventBus.close();
+        }
+    }
+
+    private static final class ControlledSink implements EventSink {
+        private final CompletableFuture<Void> future = new CompletableFuture<>();
+
+        @Override
+        public String name() {
+            return "controlled";
+        }
+
+        @Override
+        public CompletableFuture<Void> publish(VehicleEvent event) {
+            return future;
+        }
+    }
+}