Support latest server correlation contract

.github/workflows/ci.yml

@@ -62,7 +62,7 @@ jobs:
     runs-on: ubuntu-latest
     timeout-minutes: 20
     env:
-      FITZ_BROKER_IMAGE: ${{ github.event_name == 'schedule' && 'ghcr.io/cntryl/fitz:latest' || 'ghcr.io/cntryl/fitz@sha256:987770b7039873313a1c4d0102f06ea8eed0e5435550ea28d5d8107521f0b287' }}
+      FITZ_BROKER_IMAGE: ${{ github.event_name == 'schedule' && 'ghcr.io/cntryl/fitz:latest' || 'ghcr.io/cntryl/fitz@sha256:976b4baa57b91841021e43563112cf0b686a3e79af89c1242a4f5cbf362e7afa' }}
       FITZ_PATTERNED_LEASE_INVENTORY: "1"
     steps:
       - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7

compose.yml

@@ -16,7 +16,7 @@
 # - Anon TCP:     tcp://127.0.0.1:${FITZ_ANON_HOST_TCP_PORT:-4191}
 
 x-broker-common: &broker-common
-  image: "${FITZ_BROKER_IMAGE:-ghcr.io/cntryl/fitz@sha256:987770b7039873313a1c4d0102f06ea8eed0e5435550ea28d5d8107521f0b287}"
+  image: "${FITZ_BROKER_IMAGE:-ghcr.io/cntryl/fitz@sha256:976b4baa57b91841021e43563112cf0b686a3e79af89c1242a4f5cbf362e7afa}"
   restart: unless-stopped
   stop_grace_period: 15s
   healthcheck:

src/client/client-core.ts

@@ -181,6 +181,12 @@ export type Client<TConfig extends ClientConfig = ClientConfig> = {
   getUrl: () => string;
   /** Returns the current connection state. */
   getState: () => ConnectionState;
+  /** Returns capabilities advertised by the current broker session. */
+  getServerCapabilities: () => {
+    protocolVersion: number;
+    capabilities: number;
+    correlationEnabled: boolean;
+  };
 };
 
 export type ClientTransportFactory = (
@@ -513,6 +519,13 @@ export function createClientWithTransport<TConfig extends ClientConfig>(
     return state === ConnectionState.Closed ? ConnectionState.Disconnected : state;
   };
 
+  const getServerCapabilities = () =>
+    connection?.getServerCapabilities() ?? {
+      protocolVersion: 0,
+      capabilities: 0,
+      correlationEnabled: false,
+    };
+
   return {
     config: resolvedConfig,
     connect,
@@ -542,5 +555,6 @@ export function createClientWithTransport<TConfig extends ClientConfig>(
     },
     getUrl,
     getState,
+    getServerCapabilities,
   } satisfies Client<TConfig>;
 }

src/client/connection.ts

@@ -841,6 +841,9 @@ export function createConnection(
         let pendingCorrelation: bigint | undefined;
         for (const frame of frames) {
           if (frame.messageType === MSG_SERVER_HELLO) {
+            if (pendingCorrelation !== undefined) {
+              throw new ProtocolError("A CORRELATED record cannot label SERVER_HELLO");
+            }
             const hello = decodeServerHello(frame.payload);
             if (hello) {
               multiplexer.setCapabilities(hello.protocolVersion, hello.capabilities);
@@ -849,13 +852,19 @@ export function createConnection(
           }
 
           if (frame.messageType === MSG_CORRELATED) {
+            if (pendingCorrelation !== undefined) {
+              throw new ProtocolError("A CORRELATED record cannot label another CORRELATED record");
+            }
             pendingCorrelation = decodeCorrelation(frame.payload);
             continue;
           }
 
           multiplexer.dispatch(frame.messageType, frame.payload, pendingCorrelation);
           pendingCorrelation = undefined;
         }
+        if (pendingCorrelation !== undefined) {
+          throw new ProtocolError("Transport frame ended after a CORRELATED record");
+        }
       } catch (error) {
         if (receiveLoopAbort || closeRequested) {
           return;
@@ -1127,6 +1136,7 @@ export function createConnection(
     getState,
     isConnected,
     getUrl,
+    getServerCapabilities: multiplexer.getCapabilities,
     reportBackgroundError,
   };
 }

src/client/multiplexer.ts

@@ -665,29 +665,26 @@ export function createMultiplexer(observability: MultiplexerObservability = {})
   };
 
   const dispatch = (messageType: number, payload: Uint8Array, correlationId?: bigint): void => {
-    // A correlated frame answers exactly one request and nothing else. Resolve
-    // it before any type-based routing: a response and a notification can share
-    // a message type, and only the label distinguishes them reliably.
+    // Resolve a live correlation before type-based routing. A correlation may
+    // produce more than one domain-defined phase, so an unknown identifier must
+    // fall through to the normal message-type path rather than be discarded.
     if (correlationId !== undefined) {
       const correlated = pendingByCorrelation.get(correlationId);
-      pendingByCorrelation.delete(correlationId);
-      if (!correlated) {
-        responsesDropped++;
-        meter?.counter("fitz.response.dropped", 1, { messageType });
-        return;
-      }
-      if (correlated.discardResponse) {
-        responsesIgnored++;
-        meter?.counter("fitz.response.ignored", 1, { messageType });
+      if (correlated) {
+        pendingByCorrelation.delete(correlationId);
+        if (correlated.discardResponse) {
+          responsesIgnored++;
+          meter?.counter("fitz.response.ignored", 1, { messageType });
+          return;
+        }
+        correlated.deadline = -1;
+        recordRequestFinished(messageType);
+        responsesTotal++;
+        meter?.counter("fitz.response.received", 1, { messageType });
+        correlated.deferred.resolve(payload);
+        correlated.onComplete?.();
         return;
       }
-      correlated.deadline = -1;
-      recordRequestFinished(messageType);
-      responsesTotal++;
-      meter?.counter("fitz.response.received", 1, { messageType });
-      correlated.deferred.resolve(payload);
-      correlated.onComplete?.();
-      return;
     }
 
     const handler = notificationHandlers.get(messageType);

tests/integration/queue.test.ts

@@ -24,6 +24,41 @@ describe("Queue integration", () => {
       await expect(items[0].complete()).resolves.toBeUndefined();
     });
 
+    it("should correlate same-type reserves when the broker responds out of order", async () => {
+      const f = new TestFixture(transport, authMode);
+      await f.connectOrFail();
+      await waitFor(() => f.client().getServerCapabilities().correlationEnabled, {
+        timeoutMs: 1000,
+        intervalMs: 10,
+        timeoutMessage: "broker did not advertise correlation capability",
+      });
+      expect(f.client().getServerCapabilities()).toEqual({
+        protocolVersion: 1,
+        capabilities: 1,
+        correlationEnabled: true,
+      });
+      const parkedRoute = f.uniqueRoute("queue");
+      const readyRoute = f.uniqueRoute("queue");
+      await f.client().queue.enqueue(readyRoute, { body: b("second") });
+
+      const parked = f.client().queue.reserve(parkedRoute, {
+        leaseSeconds: 30,
+        batchSize: 1,
+        waitSeconds: 5,
+      });
+      await sleep(100);
+      const second = await f.client().queue.reserve(readyRoute, {
+        leaseSeconds: 30,
+        batchSize: 1,
+        waitSeconds: 0,
+      });
+      expect(Buffer.from(second[0].body).toString()).toBe("second");
+
+      await f.client().queue.enqueue(parkedRoute, { body: b("first") });
+      const first = await parked;
+      expect(Buffer.from(first[0].body).toString()).toBe("first");
+    });
+
     it("should return message to queue after lease expiry", async () => {
       const f = new TestFixture(transport, authMode);
       await f.connectOrFail();

tests/unit/client/multiplexer.test.ts

@@ -123,6 +123,18 @@ describe("Multiplexer", () => {
     expect(meter.histograms).toContain("fitz.request.duration");
   });
 
+  it("routes a later phase normally after its correlation already completed", () => {
+    const multiplexer = createMultiplexer();
+    multiplexer.setConnected();
+    const handler = vi.fn();
+    multiplexer.registerNotificationHandler(400, handler);
+
+    multiplexer.dispatch(400, new Uint8Array([7]), 999n);
+
+    expect(handler).toHaveBeenCalledWith(new Uint8Array([7]));
+    expect(multiplexer.getMetrics().responsesDropped).toBe(0);
+  });
+
   it("records timeout failures once and closes the span", async () => {
     vi.useFakeTimers();
     const tracer = new FakeTracer();