Hold relay reads until startup barrier

proxy-core/src/main/java/dev/strataproxy/core/relay/RawRelay.java

@@ -155,7 +155,7 @@ public final class RawRelay {
         private int writesInFlight;
         private int pausedBytes;
         private volatile boolean closed;
-        private boolean paused;
+        private boolean paused = true;
         private volatile boolean detached;
         private boolean readsEnabled;
         private CompletableFuture<Void> pauseComplete;
@@ -179,7 +179,15 @@ public final class RawRelay {
                 return;
             }
             readsEnabled = true;
-            requestRead();
+            // An in-flight read may arrive after attach but before start removes the
+            // login codecs. Flush those owned buffers before requesting new data.
+            if (paused) {
+                resume().whenComplete((ignored, failure) -> {
+                    if (failure != null) closePair(context);
+                });
+            } else {
+                requestRead();
+            }
         }
 
         private void requestRead() {
@@ -383,15 +391,23 @@ public final class RawRelay {
                 return;
             }
             writesInFlight++;
-            peer.writeAndFlush(next).addListener((ChannelFutureListener) future ->
-                    context.executor().execute(() -> {
-                        writesInFlight--;
-                        if (!future.isSuccess()) {
-                            failResume(future.cause());
-                            return;
-                        }
-                        drainPausedMessages();
-                    }));
+            try {
+                Consumer<ByteBuf> activeObserver = observer;
+                if (activeObserver != null) activeObserver.accept(next.duplicate());
+                peer.writeAndFlush(next).addListener((ChannelFutureListener) future ->
+                        context.executor().execute(() -> {
+                            writesInFlight--;
+                            if (!future.isSuccess()) {
+                                failResume(future.cause());
+                                return;
+                            }
+                            drainPausedMessages();
+                        }));
+            } catch (RuntimeException failure) {
+                writesInFlight--;
+                next.release();
+                failResume(failure);
+            }
         }
 
         private void failResume(Throwable failure) {

proxy-core/src/test/java/dev/strataproxy/core/relay/RawRelayTest.java

@@ -15,6 +15,35 @@ import static org.junit.jupiter.api.Assertions.assertSame;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 final class RawRelayTest {
+    @Test
+    void holdsAnInFlightReadUntilTheRelayStarts() {
+        var client = new EmbeddedChannel();
+        var backend = new EmbeddedChannel();
+        try {
+            var observed = new AtomicInteger();
+            var link = RawRelay.attach(client, backend, bytes -> observed.incrementAndGet(), null);
+            pump(client, backend);
+
+            var inFlight = Unpooled.wrappedBuffer(new byte[] {42});
+            client.writeInbound(inFlight);
+            pump(client, backend);
+            assertNull(backend.readOutbound());
+            assertEquals(1, inFlight.refCnt());
+            assertEquals(0, observed.get());
+
+            link.start();
+            pump(client, backend);
+            var forwarded = (io.netty.buffer.ByteBuf) backend.readOutbound();
+            assertSame(inFlight, forwarded);
+            assertEquals(1, observed.get());
+            forwarded.release();
+            assertEquals(0, inFlight.refCnt());
+        } finally {
+            client.finishAndReleaseAll();
+            backend.finishAndReleaseAll();
+        }
+    }
+
     @Test
     void forwardsTheOriginalBufferAndClosesBothSides() {
         var client = new EmbeddedChannel();