fix: delay jt808 ack until durable dispatch

modules/protocols/protocol-jt808/src/main/java/com/lingniu/ingest/protocol/jt808/inbound/Jt808ChannelHandler.java

@@ -25,6 +25,7 @@ import com.lingniu.ingest.session.SessionStore;
 import io.netty.buffer.Unpooled;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.util.AttributeKey;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -34,6 +35,7 @@ import java.time.Instant;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
 
 /**
  * JT808 Netty 入站处理器。职责：
@@ -49,6 +51,8 @@ public class Jt808ChannelHandler extends SimpleChannelInboundHandler<Object> {
 
     private static final Logger log = LoggerFactory.getLogger(Jt808ChannelHandler.class);
     private static final String PROCESSING_ERROR_PREFIX = "processing failed: ";
+    private static final AttributeKey<CompletableFuture<Void>> DURABLE_ACK_CHAIN_ATTR =
+            AttributeKey.valueOf("jt808.durableAckChain");
 
     private final Jt808MessageDecoder decoder;
     private final Dispatcher dispatcher;
@@ -119,25 +123,21 @@ public class Jt808ChannelHandler extends SimpleChannelInboundHandler<Object> {
         // ====== 会话维护 & 同步应答匹配 ======
         String phone = msg.header().phone();
         IdentityResolution identity = resolveIdentity(msg);
+        byte[] durableAck = null;
         switch (msg.header().messageId()) {
             case Jt808MessageId.TERMINAL_REGISTER -> {
                 DeviceSession session = upsertSession(ctx, msg, identity.identity());
                 channelRegistry.bind(phone, ctx.channel());
-                byte[] registerAck = Jt808FrameEncoder.encode(
-                        Jt808Commands.registerAck(msg.header().serialNo(), 0, session.token()).messageId(),
-                        phone,
-                        channelRegistry.nextSerial(phone),
-                        Jt808Commands.registerAck(msg.header().serialNo(), 0, session.token()).body());
-                ctx.writeAndFlush(Unpooled.wrappedBuffer(registerAck));
+                durableAck = registerAckFrame(phone, msg.header().serialNo(), session.token());
             }
             case Jt808MessageId.TERMINAL_AUTH -> {
                 upsertSession(ctx, msg, identity.identity());
                 channelRegistry.bind(phone, ctx.channel());
-                ackTerminal(ctx, phone, msg.header().serialNo(), msg.header().messageId(), 0);
+                durableAck = terminalAckFrame(phone, msg.header().serialNo(), msg.header().messageId(), 0);
             }
             case Jt808MessageId.TERMINAL_HEARTBEAT -> {
                 channelRegistry.bind(phone, ctx.channel());
-                ackTerminal(ctx, phone, msg.header().serialNo(), msg.header().messageId(), 0);
+                durableAck = terminalAckFrame(phone, msg.header().serialNo(), msg.header().messageId(), 0);
             }
             case Jt808MessageId.TERMINAL_GENERAL_RESPONSE -> {
                 // body: ackSerial(2) + ackMsgId(2) + result(1)
@@ -173,7 +173,63 @@ public class Jt808ChannelHandler extends SimpleChannelInboundHandler<Object> {
                 frame,
                 meta,
                 Instant.now());
-        dispatcher.dispatch(rf);
+        if (durableAck == null) {
+            dispatcher.dispatch(rf);
+            return;
+        }
+        enqueueDurableAck(ctx, phone, msg.header().messageId(), dispatcher.dispatchAndAwait(rf), durableAck);
+    }
+
+    private void enqueueDurableAck(ChannelHandlerContext ctx,
+                                   String phone,
+                                   int messageId,
+                                   CompletableFuture<Void> dispatchFuture,
+                                   byte[] ackFrame) {
+        CompletableFuture<Void> previous = ctx.channel().attr(DURABLE_ACK_CHAIN_ATTR).get();
+        if (previous == null) {
+            previous = CompletableFuture.completedFuture(null);
+        }
+        CompletableFuture<Void> next = previous
+                .handle((ignored, previousFailure) -> null)
+                .thenCompose(ignored -> dispatchFuture.handle((ok, failure) -> failure))
+                .thenCompose(failure -> runOnEventLoop(ctx,
+                        () -> handleDurableAckResult(ctx, phone, messageId, failure, ackFrame)));
+        ctx.channel().attr(DURABLE_ACK_CHAIN_ATTR).set(next);
+    }
+
+    private CompletableFuture<Void> runOnEventLoop(ChannelHandlerContext ctx, Runnable task) {
+        CompletableFuture<Void> done = new CompletableFuture<>();
+        Runnable wrapped = () -> {
+            try {
+                task.run();
+                done.complete(null);
+            } catch (Throwable t) {
+                done.completeExceptionally(t);
+            }
+        };
+        if (ctx.executor().inEventLoop()) {
+            wrapped.run();
+        } else {
+            ctx.executor().execute(wrapped);
+        }
+        return done;
+    }
+
+    private void handleDurableAckResult(ChannelHandlerContext ctx,
+                                        String phone,
+                                        int messageId,
+                                        Throwable failure,
+                                        byte[] ackFrame) {
+        if (!ctx.channel().isActive()) {
+            return;
+        }
+        if (failure != null) {
+            log.warn("[jt808] dispatch failed before ack peer={} phone={} msgId=0x{}",
+                    addr(ctx), phone, Integer.toHexString(messageId), failure);
+            ctx.close();
+            return;
+        }
+        ctx.writeAndFlush(Unpooled.wrappedBuffer(ackFrame));
     }
 
     private IdentityResolution resolveIdentity(Jt808Message msg) {
@@ -341,10 +397,14 @@ public class Jt808ChannelHandler extends SimpleChannelInboundHandler<Object> {
         return value == null ? "" : value;
     }
 
-    private void ackTerminal(ChannelHandlerContext ctx, String phone, int ackSerial, int ackMsgId, int result) {
+    private byte[] registerAckFrame(String phone, int ackSerial, String token) {
+        var cmd = Jt808Commands.registerAck(ackSerial, 0, token);
+        return Jt808FrameEncoder.encode(cmd.messageId(), phone, channelRegistry.nextSerial(phone), cmd.body());
+    }
+
+    private byte[] terminalAckFrame(String phone, int ackSerial, int ackMsgId, int result) {
         var cmd = Jt808Commands.platformAck(ackSerial, ackMsgId, result);
-        byte[] frame = Jt808FrameEncoder.encode(cmd.messageId(), phone, channelRegistry.nextSerial(phone), cmd.body());
-        ctx.writeAndFlush(Unpooled.wrappedBuffer(frame));
+        return Jt808FrameEncoder.encode(cmd.messageId(), phone, channelRegistry.nextSerial(phone), cmd.body());
     }
 
     private static String generateToken(String phone) {

modules/protocols/protocol-jt808/src/test/java/com/lingniu/ingest/protocol/jt808/inbound/Jt808ChannelHandlerTest.java

@@ -29,6 +29,7 @@ import com.lingniu.ingest.protocol.jt808.model.Jt808MessageId;
 import com.lingniu.ingest.protocol.jt808.session.Jt808ChannelRegistry;
 import com.lingniu.ingest.protocol.jt808.session.Jt808PendingRequests;
 import com.lingniu.ingest.session.InMemorySessionStore;
+import io.netty.buffer.ByteBuf;
 import io.netty.channel.embedded.EmbeddedChannel;
 import org.junit.jupiter.api.Test;
 
@@ -49,7 +50,7 @@ class Jt808ChannelHandlerTest {
         InMemoryVehicleIdentityService identity = new InMemoryVehicleIdentityService();
         identity.bind(new VehicleIdentityBinding(
                 ProtocolId.JT808, "LNVIN000000000808", "123456789012", "DEV808", "B80808"));
-        DisruptorEventBus eventBus = new DisruptorEventBus(1024, "blocking", List.of());
+        DisruptorEventBus eventBus = new DisruptorEventBus(1024, "blocking", List.of(new ImmediateKafkaSink()));
         AsyncBatchExecutor batchExecutor = new AsyncBatchExecutor(eventBus::publish);
         Dispatcher dispatcher = new Dispatcher(
                 new HandlerRegistry(),
@@ -88,7 +89,7 @@ class Jt808ChannelHandlerTest {
     void unresolvedRegisterDoesNotBindUnknownVinToExternalIdentifiers() {
         InMemorySessionStore sessions = new InMemorySessionStore();
         InMemoryVehicleIdentityService identity = new InMemoryVehicleIdentityService();
-        DisruptorEventBus eventBus = new DisruptorEventBus(1024, "blocking", List.of());
+        DisruptorEventBus eventBus = new DisruptorEventBus(1024, "blocking", List.of(new ImmediateKafkaSink()));
         AsyncBatchExecutor batchExecutor = new AsyncBatchExecutor(eventBus::publish);
         Dispatcher dispatcher = new Dispatcher(
                 new HandlerRegistry(),
@@ -129,7 +130,7 @@ class Jt808ChannelHandlerTest {
         InMemoryVehicleIdentityService identity = new InMemoryVehicleIdentityService();
         identity.bind(new VehicleIdentityBinding(
                 ProtocolId.JT808, "LNVIN000000AUTH01", "123456789012", "123456789012345", ""));
-        DisruptorEventBus eventBus = new DisruptorEventBus(1024, "blocking", List.of());
+        DisruptorEventBus eventBus = new DisruptorEventBus(1024, "blocking", List.of(new ImmediateKafkaSink()));
         AsyncBatchExecutor batchExecutor = new AsyncBatchExecutor(eventBus::publish);
         Dispatcher dispatcher = new Dispatcher(
                 new HandlerRegistry(),
@@ -161,10 +162,48 @@ class Jt808ChannelHandlerTest {
         eventBus.close();
     }
 
+    @Test
+    void registerAckWaitsForDispatchDurability() {
+        ControlledSink sink = new ControlledSink();
+        DisruptorEventBus eventBus = new DisruptorEventBus(1024, "blocking", List.of(sink));
+        AsyncBatchExecutor batchExecutor = new AsyncBatchExecutor(eventBus::publish);
+        Dispatcher dispatcher = new Dispatcher(
+                new HandlerRegistry(),
+                new InterceptorChain(List.of()),
+                new HandlerInvoker(),
+                eventBus,
+                batchExecutor);
+        Jt808ChannelHandler handler = new Jt808ChannelHandler(
+                new Jt808MessageDecoder(new BodyParserRegistry(List.of(new RegisterBodyParser()))),
+                dispatcher,
+                new InMemorySessionStore(),
+                new InMemoryVehicleIdentityService(),
+                new Jt808ChannelRegistry(),
+                new Jt808PendingRequests());
+        EmbeddedChannel channel = new EmbeddedChannel(handler);
+
+        channel.writeInbound(buildFrame(
+                Jt808MessageId.TERMINAL_REGISTER,
+                "123456789012",
+                1,
+                buildRegisterBody("DEV808", "B80808")));
+
+        assertThat((Object) channel.readOutbound()).isNull();
+        sink.awaitPublishCount(1);
+        sink.future(0).complete(null);
+
+        ByteBuf ack = readOutboundAfterRunningTasks(channel);
+        assertThat(ack).isNotNull();
+        ack.release();
+
+        batchExecutor.close();
+        eventBus.close();
+    }
+
     @Test
     void inactiveChannelRemovesSessionStoreEntry() {
         InMemorySessionStore sessions = new InMemorySessionStore();
-        DisruptorEventBus eventBus = new DisruptorEventBus(1024, "blocking", List.of());
+        DisruptorEventBus eventBus = new DisruptorEventBus(1024, "blocking", List.of(new ImmediateKafkaSink()));
         AsyncBatchExecutor batchExecutor = new AsyncBatchExecutor(eventBus::publish);
         Dispatcher dispatcher = new Dispatcher(
                 new HandlerRegistry(),
@@ -596,6 +635,62 @@ class Jt808ChannelHandlerTest {
         os.write((int) (v & 0xFF));
     }
 
+    private static ByteBuf readOutboundAfterRunningTasks(EmbeddedChannel channel) {
+        long deadline = System.currentTimeMillis() + 3000;
+        while (System.currentTimeMillis() < deadline) {
+            channel.runPendingTasks();
+            ByteBuf outbound = channel.readOutbound();
+            if (outbound != null) {
+                return outbound;
+            }
+            Thread.onSpinWait();
+        }
+        return null;
+    }
+
+    private static final class ControlledSink implements EventSink {
+        private final List<CompletableFuture<Void>> futures = new CopyOnWriteArrayList<>();
+
+        @Override
+        public String name() {
+            return "kafka";
+        }
+
+        @Override
+        public CompletableFuture<Void> publish(VehicleEvent event) {
+            CompletableFuture<Void> future = new CompletableFuture<>();
+            futures.add(future);
+            return future;
+        }
+
+        private CompletableFuture<Void> future(int index) {
+            return futures.get(index);
+        }
+
+        private void awaitPublishCount(int expected) {
+            long deadline = System.currentTimeMillis() + 3000;
+            while (System.currentTimeMillis() < deadline) {
+                if (futures.size() >= expected) {
+                    return;
+                }
+                Thread.onSpinWait();
+            }
+            throw new AssertionError("expected " + expected + " sink publishes, actual=" + futures.size());
+        }
+    }
+
+    private static final class ImmediateKafkaSink implements EventSink {
+        @Override
+        public String name() {
+            return "kafka";
+        }
+
+        @Override
+        public CompletableFuture<Void> publish(VehicleEvent event) {
+            return CompletableFuture.completedFuture(null);
+        }
+    }
+
     private static final class RecordingSink implements EventSink {
         private final List<VehicleEvent> events = new CopyOnWriteArrayList<>();
         private final CountDownLatch latch;
@@ -606,7 +701,7 @@ class Jt808ChannelHandlerTest {
 
         @Override
         public String name() {
-            return "recording";
+            return "kafka";
         }
 
         @Override