From f01bd728eb25d021250de01a225e6cdbfa908962 Mon Sep 17 00:00:00 2001 From: Jeff Repanich Date: Sun, 13 Sep 2026 16:48:03 -0400 Subject: [PATCH] Support latest server correlation contract --- .github/workflows/ci.yml | 2 +- compose.yml | 2 +- src/client/client-core.ts | 14 +++++++++++ src/client/connection.ts | 10 ++++++++ src/client/multiplexer.ts | 35 ++++++++++++--------------- tests/integration/queue.test.ts | 35 +++++++++++++++++++++++++++ tests/unit/client/multiplexer.test.ts | 12 +++++++++ 7 files changed, 89 insertions(+), 21 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7516b90..b696a40 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/compose.yml b/compose.yml index 183ef82..75b4ccf 100644 --- a/compose.yml +++ b/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: diff --git a/src/client/client-core.ts b/src/client/client-core.ts index cc8a5cf..aefbfd5 100644 --- a/src/client/client-core.ts +++ b/src/client/client-core.ts @@ -181,6 +181,12 @@ export type Client = { 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( 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( }, getUrl, getState, + getServerCapabilities, } satisfies Client; } diff --git a/src/client/connection.ts b/src/client/connection.ts index 9bbce1d..df88fb0 100644 --- a/src/client/connection.ts +++ b/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,6 +852,9 @@ 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; } @@ -856,6 +862,9 @@ export function createConnection( 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, }; } diff --git a/src/client/multiplexer.ts b/src/client/multiplexer.ts index f8fa27e..fbabcd2 100644 --- a/src/client/multiplexer.ts +++ b/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); diff --git a/tests/integration/queue.test.ts b/tests/integration/queue.test.ts index d5440e9..b83ee77 100644 --- a/tests/integration/queue.test.ts +++ b/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(); diff --git a/tests/unit/client/multiplexer.test.ts b/tests/unit/client/multiplexer.test.ts index 1521bb4..2332436 100644 --- a/tests/unit/client/multiplexer.test.ts +++ b/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();