From 81edd16652a5472e25fa8c1627dc9b1fdbda57df Mon Sep 17 00:00:00 2001 From: Tirthraj Chavan Date: Sat, 7 Nov 2026 11:52:46 +0530 Subject: [PATCH 1/9] =?UTF-8?q?test(relay):=20non-sequential=20socket=20ca?= =?UTF-8?q?ses=20=E2=80=94=20commands=20that=20race=20the=20connection=20(?= =?UTF-8?q?#1052)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit (cherry picked from commit 1a633921a01bc5e5ab73c9e251068a8ed9fe5cff) --- .../test-non-sequential-socket-cases.md | 13 ++ .../web-socket-server-adapter-real.spec.ts | 198 +++++++++++++++++- 2 files changed, 203 insertions(+), 8 deletions(-) create mode 100644 .changeset/test-non-sequential-socket-cases.md diff --git a/.changeset/test-non-sequential-socket-cases.md b/.changeset/test-non-sequential-socket-cases.md new file mode 100644 index 000000000..0b68981d9 --- /dev/null +++ b/.changeset/test-non-sequential-socket-cases.md @@ -0,0 +1,13 @@ +--- +"nostream": patch +--- + +test(relay): add non-sequential socket test suite for connection and command races + +Adds deterministic real-socket test cases covering racing connection states: +- Reconnect storm with immediate REQ on the open tick across multiple concurrent sockets +- Pipelined commands in a single write buffer preserving arrival order +- REQ immediately followed by CLOSE of the same subscription ID +- EVENT sent on open tick +- Binary-framed JSON commands +- Socket resilience and valid command processing following malformed JSON diff --git a/test/unit/adapters/web-socket-server-adapter-real.spec.ts b/test/unit/adapters/web-socket-server-adapter-real.spec.ts index 4af64cd91..4a757d3ea 100644 --- a/test/unit/adapters/web-socket-server-adapter-real.spec.ts +++ b/test/unit/adapters/web-socket-server-adapter-real.spec.ts @@ -20,13 +20,17 @@ const waitFor = async (predicate: () => boolean, timeoutMs = 3_000): Promise { +describe('WebSocketServerAdapter real socket non-sequential & race cases (#1052)', function () { + this.timeout(10_000) + let sandbox: Sinon.SinonSandbox let server: http.Server let webSocketServer: WebSocketServer let adapter: WebSocketServerAdapter let isRateLimitedStub: Sinon.SinonStub - let receivedFrames: Buffer[] | undefined + let receivedFrames: Buffer[] + let receivedFramesByClient: Map + let allReceivedFrames: { client: WebSocket; data: Buffer }[] const clients: WebSocket[] = [] const settings = () => @@ -37,7 +41,9 @@ describe('WebSocketServerAdapter real socket admission', () => { beforeEach(async () => { sandbox = Sinon.createSandbox() - receivedFrames = undefined + receivedFrames = [] + receivedFramesByClient = new Map() + allReceivedFrames = [] isRateLimitedStub = sandbox.stub(rateLimiterMiddleware, 'isRateLimited').resolves(false) server = http.createServer() @@ -49,7 +55,12 @@ describe('WebSocketServerAdapter real socket admission', () => { ([client]: [WebSocket, unknown, unknown]) => { const received: Buffer[] = [] receivedFrames = received - client.on('message', (data: Buffer) => received.push(data)) + receivedFramesByClient.set(client, received) + client.on('message', (data: Buffer) => { + const buf = Buffer.from(data) + received.push(buf) + allReceivedFrames.push({ client, data: buf }) + }) return { emit: () => undefined, @@ -75,6 +86,9 @@ describe('WebSocketServerAdapter real socket admission', () => { sandbox.restore() }) + // -------------------------------------------------------------------------- + // Baseline check: delayed admission buffering (from #1045) + // -------------------------------------------------------------------------- it('delivers frames sent immediately on open across a delayed admission check', async () => { let releaseAdmission: ((limited: boolean) => void) | undefined isRateLimitedStub.returns( @@ -88,7 +102,7 @@ describe('WebSocketServerAdapter real socket admission', () => { clients.push(client) await new Promise((resolve, reject) => { - client.once('open', () => resolve()) + client.once('open', resolve) client.once('error', reject) }) @@ -101,14 +115,182 @@ describe('WebSocketServerAdapter real socket admission', () => { client.send(frame) } - // Give the frames time to reach the relay while admission is still pending. + // Allow frames to hit the server socket while admission is still pending await new Promise((resolve) => setTimeout(resolve, 50)) releaseAdmission?.(false) await waitFor(() => (receivedFrames?.length ?? 0) >= frames.length) - expect((receivedFrames ?? []).map((frame) => frame.toString())).to.deep.equal( - frames.map((frame) => frame.toString()), + expect((receivedFrames ?? []).map((f) => f.toString())).to.deep.equal( + frames.map((f) => f.toString()), ) }) + + // -------------------------------------------------------------------------- + // Shape 1: Subscribe at open tick / reconnect storm across N concurrent sockets + // -------------------------------------------------------------------------- + it('Shape 1: handles concurrent reconnect storm with REQs sent at open tick across N sockets', async () => { + const address = server.address() as AddressInfo + const N = 10 + + const connectPromises = Array.from({ length: N }).map((_, i) => { + return new Promise((resolve, reject) => { + const ws = new WebSocket(`ws://127.0.0.1:${address.port}`) + clients.push(ws) + + ws.on('open', () => { + ws.send(Buffer.from(JSON.stringify(['REQ', `sub-storm-${i}`, { kinds: [1] }]))) + resolve() + }) + ws.on('error', reject) + }) + }) + + await Promise.all(connectPromises) + + await waitFor(() => { + let total = 0 + for (const frames of receivedFramesByClient.values()) { + total += frames.length + } + return total >= N + }) + + expect(receivedFramesByClient.size).to.equal(N) + for (const frames of receivedFramesByClient.values()) { + expect(frames.length).to.be.at.least(1) + const parsed = JSON.parse(frames[0].toString()) + expect(parsed[0]).to.equal('REQ') + } + }) + + // -------------------------------------------------------------------------- + // Shape 2: Pipelined commands in one write + // -------------------------------------------------------------------------- + it('Shape 2: delivers pipelined commands in the exact order sent', async () => { + const address = server.address() as AddressInfo + const client = new WebSocket(`ws://127.0.0.1:${address.port}`) + clients.push(client) + + await new Promise((resolve, reject) => { + client.once('open', resolve) + client.once('error', reject) + }) + + const cmd1 = Buffer.from('["REQ","pipelined-1",{"kinds":[1]}]') + const cmd2 = Buffer.from('["REQ","pipelined-2",{"kinds":[0]}]') + + // Pipelined writes into the socket stream back-to-back + client.send(cmd1) + client.send(cmd2) + + await waitFor(() => (receivedFrames?.length ?? 0) >= 2) + + expect(receivedFrames.map((f) => f.toString())).to.deep.equal([cmd1.toString(), cmd2.toString()]) + }) + + // -------------------------------------------------------------------------- + // Shape 3: REQ followed immediately by CLOSE of the same subscription id + // -------------------------------------------------------------------------- + it('Shape 3: processes REQ followed immediately by CLOSE without error or socket termination', async () => { + const address = server.address() as AddressInfo + const client = new WebSocket(`ws://127.0.0.1:${address.port}`) + clients.push(client) + + await new Promise((resolve, reject) => { + client.once('open', resolve) + client.once('error', reject) + }) + + client.send(Buffer.from('["REQ","race-cancel",{"kinds":[1]}]')) + client.send(Buffer.from('["CLOSE","race-cancel"]')) + + await waitFor(() => (receivedFrames?.length ?? 0) >= 2) + + const parsed = receivedFrames.map((f) => JSON.parse(f.toString())) + expect(parsed[0]).to.deep.equal(['REQ', 'race-cancel', { kinds: [1] }]) + expect(parsed[1]).to.deep.equal(['CLOSE', 'race-cancel']) + expect(client.readyState).to.equal(WebSocket.OPEN) + }) + + // -------------------------------------------------------------------------- + // Shape 4: EVENT sent in the same turn as open + // -------------------------------------------------------------------------- + it('Shape 4: delivers EVENT sent in the same turn as open', async () => { + const address = server.address() as AddressInfo + const client = new WebSocket(`ws://127.0.0.1:${address.port}`) + clients.push(client) + + const testEvent = ['EVENT', { id: 'evt-race-1', kind: 1, content: 'race event' }] + + await new Promise((resolve, reject) => { + client.on('open', () => { + client.send(Buffer.from(JSON.stringify(testEvent))) + resolve() + }) + client.on('error', reject) + }) + + await waitFor(() => (receivedFrames?.length ?? 0) >= 1) + + expect(JSON.parse(receivedFrames[0].toString())).to.deep.equal(testEvent) + }) + + // -------------------------------------------------------------------------- + // Shape 5: Binary and fragmented frames + // -------------------------------------------------------------------------- + it('Shape 5: correctly receives fragmented frames and binary frames and keeps socket usable', async () => { + const address = server.address() as AddressInfo + const client = new WebSocket(`ws://127.0.0.1:${address.port}`) + clients.push(client) + + await new Promise((resolve, reject) => { + client.once('open', resolve) + client.once('error', reject) + }) + + // Fragmented frame: split command across two frames + client.send('["REQ","frag', { fin: false }) + client.send('-sub",{"kinds":[1]}]', { fin: true }) + + // Binary frame carrying JSON command + const binaryCommand = Buffer.from('["REQ","binary-frame",{"kinds":[1]}]') + client.send(binaryCommand, { binary: true }) + + // Next command on the same socket + const nextCommand = Buffer.from('["REQ","next-cmd",{"kinds":[1]}]') + client.send(nextCommand) + + await waitFor(() => (receivedFrames?.length ?? 0) >= 3) + + expect(receivedFrames[0].toString()).to.equal('["REQ","frag-sub",{"kinds":[1]}]') + expect(receivedFrames[1].toString()).to.equal(binaryCommand.toString()) + expect(receivedFrames[2].toString()).to.equal(nextCommand.toString()) + }) + + // -------------------------------------------------------------------------- + // Shape 6: Invalid JSON mid-stream maintains socket health for next valid command + // -------------------------------------------------------------------------- + it('Shape 6: maintains socket usability for next valid command after invalid JSON', async () => { + const address = server.address() as AddressInfo + const client = new WebSocket(`ws://127.0.0.1:${address.port}`) + clients.push(client) + + await new Promise((resolve, reject) => { + client.once('open', resolve) + client.once('error', reject) + }) + + const badPayload = Buffer.from('{"broken_json_without_closing_bracket') + const nextValidCommand = Buffer.from('["REQ","valid-after-bad",{"kinds":[1]}]') + + client.send(badPayload) + client.send(nextValidCommand) + + await waitFor(() => (receivedFrames?.length ?? 0) >= 2) + + expect(receivedFrames[0].toString()).to.equal(badPayload.toString()) + expect(receivedFrames[1].toString()).to.equal(nextValidCommand.toString()) + expect(client.readyState).to.equal(WebSocket.OPEN) + }) }) From 49f027bfc6a84179f1b5ca8e5dc32e67a73bc2b0 Mon Sep 17 00:00:00 2001 From: chappie-daemon Date: Sat, 7 Nov 2026 17:41:04 -0500 Subject: [PATCH 2/9] test(relay): hold admission open in the #1052 race cases The cases all send at the open tick, but beforeEach stubs the admission check to resolve immediately, so the adapter installs its normal listener before any frame can arrive and the buffering path is never exercised. holdAdmission() leaves the check pending until the frames are in the server's queue, then admits; every case now goes through the replay path. Shape 1 also asserted only that each socket saw some REQ. It now requires exactly one frame per socket, the full command shape, and the multiset of subscription ids received to equal the ids sent. --- .../web-socket-server-adapter-real.spec.ts | 49 ++++++++++++++++++- 1 file changed, 47 insertions(+), 2 deletions(-) diff --git a/test/unit/adapters/web-socket-server-adapter-real.spec.ts b/test/unit/adapters/web-socket-server-adapter-real.spec.ts index 4a757d3ea..c4a1fdebd 100644 --- a/test/unit/adapters/web-socket-server-adapter-real.spec.ts +++ b/test/unit/adapters/web-socket-server-adapter-real.spec.ts @@ -20,6 +20,25 @@ const waitFor = async (predicate: () => boolean, timeoutMs = 3_000): Promise { + let release: (() => void) | undefined + stub.returns( + new Promise((resolve) => { + release = () => resolve(false) + }), + ) + + return async (settleMs = 50) => { + await new Promise((resolve) => setTimeout(resolve, settleMs)) + release?.() + } +} + describe('WebSocketServerAdapter real socket non-sequential & race cases (#1052)', function () { this.timeout(10_000) @@ -132,14 +151,19 @@ describe('WebSocketServerAdapter real socket non-sequential & race cases (#1052) it('Shape 1: handles concurrent reconnect storm with REQs sent at open tick across N sockets', async () => { const address = server.address() as AddressInfo const N = 10 + const admit = holdAdmission(isRateLimitedStub) + const sentIds = new Set() const connectPromises = Array.from({ length: N }).map((_, i) => { return new Promise((resolve, reject) => { const ws = new WebSocket(`ws://127.0.0.1:${address.port}`) clients.push(ws) + const id = `sub-storm-${i}` + sentIds.add(id) + ws.on('open', () => { - ws.send(Buffer.from(JSON.stringify(['REQ', `sub-storm-${i}`, { kinds: [1] }]))) + ws.send(Buffer.from(JSON.stringify(['REQ', id, { kinds: [1] }]))) resolve() }) ws.on('error', reject) @@ -147,6 +171,7 @@ describe('WebSocketServerAdapter real socket non-sequential & race cases (#1052) }) await Promise.all(connectPromises) + await admit() await waitFor(() => { let total = 0 @@ -157,11 +182,21 @@ describe('WebSocketServerAdapter real socket non-sequential & race cases (#1052) }) expect(receivedFramesByClient.size).to.equal(N) + + const receivedIds: string[] = [] for (const frames of receivedFramesByClient.values()) { - expect(frames.length).to.be.at.least(1) + expect(frames).to.have.length(1) const parsed = JSON.parse(frames[0].toString()) + expect(parsed).to.have.length(3) expect(parsed[0]).to.equal('REQ') + expect(parsed[1]).to.match(/^sub-storm-\d+$/) + expect(parsed[2]).to.deep.equal({ kinds: [1] }) + receivedIds.push(parsed[1]) } + + // One complete command per socket, and the N ids received are exactly the + // N ids sent — a duplicated or cross-wired command would break the set. + expect(receivedIds.sort()).to.deep.equal([...sentIds].sort()) }) // -------------------------------------------------------------------------- @@ -169,6 +204,7 @@ describe('WebSocketServerAdapter real socket non-sequential & race cases (#1052) // -------------------------------------------------------------------------- it('Shape 2: delivers pipelined commands in the exact order sent', async () => { const address = server.address() as AddressInfo + const admit = holdAdmission(isRateLimitedStub) const client = new WebSocket(`ws://127.0.0.1:${address.port}`) clients.push(client) @@ -184,6 +220,7 @@ describe('WebSocketServerAdapter real socket non-sequential & race cases (#1052) client.send(cmd1) client.send(cmd2) + await admit() await waitFor(() => (receivedFrames?.length ?? 0) >= 2) expect(receivedFrames.map((f) => f.toString())).to.deep.equal([cmd1.toString(), cmd2.toString()]) @@ -194,6 +231,7 @@ describe('WebSocketServerAdapter real socket non-sequential & race cases (#1052) // -------------------------------------------------------------------------- it('Shape 3: processes REQ followed immediately by CLOSE without error or socket termination', async () => { const address = server.address() as AddressInfo + const admit = holdAdmission(isRateLimitedStub) const client = new WebSocket(`ws://127.0.0.1:${address.port}`) clients.push(client) @@ -205,6 +243,7 @@ describe('WebSocketServerAdapter real socket non-sequential & race cases (#1052) client.send(Buffer.from('["REQ","race-cancel",{"kinds":[1]}]')) client.send(Buffer.from('["CLOSE","race-cancel"]')) + await admit() await waitFor(() => (receivedFrames?.length ?? 0) >= 2) const parsed = receivedFrames.map((f) => JSON.parse(f.toString())) @@ -218,6 +257,7 @@ describe('WebSocketServerAdapter real socket non-sequential & race cases (#1052) // -------------------------------------------------------------------------- it('Shape 4: delivers EVENT sent in the same turn as open', async () => { const address = server.address() as AddressInfo + const admit = holdAdmission(isRateLimitedStub) const client = new WebSocket(`ws://127.0.0.1:${address.port}`) clients.push(client) @@ -231,6 +271,7 @@ describe('WebSocketServerAdapter real socket non-sequential & race cases (#1052) client.on('error', reject) }) + await admit() await waitFor(() => (receivedFrames?.length ?? 0) >= 1) expect(JSON.parse(receivedFrames[0].toString())).to.deep.equal(testEvent) @@ -241,6 +282,7 @@ describe('WebSocketServerAdapter real socket non-sequential & race cases (#1052) // -------------------------------------------------------------------------- it('Shape 5: correctly receives fragmented frames and binary frames and keeps socket usable', async () => { const address = server.address() as AddressInfo + const admit = holdAdmission(isRateLimitedStub) const client = new WebSocket(`ws://127.0.0.1:${address.port}`) clients.push(client) @@ -261,6 +303,7 @@ describe('WebSocketServerAdapter real socket non-sequential & race cases (#1052) const nextCommand = Buffer.from('["REQ","next-cmd",{"kinds":[1]}]') client.send(nextCommand) + await admit() await waitFor(() => (receivedFrames?.length ?? 0) >= 3) expect(receivedFrames[0].toString()).to.equal('["REQ","frag-sub",{"kinds":[1]}]') @@ -273,6 +316,7 @@ describe('WebSocketServerAdapter real socket non-sequential & race cases (#1052) // -------------------------------------------------------------------------- it('Shape 6: maintains socket usability for next valid command after invalid JSON', async () => { const address = server.address() as AddressInfo + const admit = holdAdmission(isRateLimitedStub) const client = new WebSocket(`ws://127.0.0.1:${address.port}`) clients.push(client) @@ -287,6 +331,7 @@ describe('WebSocketServerAdapter real socket non-sequential & race cases (#1052) client.send(badPayload) client.send(nextValidCommand) + await admit() await waitFor(() => (receivedFrames?.length ?? 0) >= 2) expect(receivedFrames[0].toString()).to.equal(badPayload.toString()) From d0b4d78854596136ff55f2f6f8b9a1f13bbfa3b8 Mon Sep 17 00:00:00 2001 From: chappie-daemon Date: Sat, 7 Nov 2026 17:55:05 -0500 Subject: [PATCH 3/9] test(relay): assert the answers for commands that race the connection Issue #1052 asks for the reply a client is entitled to, not survival. The adapter spec stops at frame delivery, so it cannot see a dropped REQ or a subscription left live by a racing CLOSE. These six features drive a real app over a real socket and assert EOSE, OK and NOTICE: - 10 sockets subscribing at the open tick, 20 rounds, exactly one EOSE each - two REQs in one turn answered in order - REQ and CLOSE in one turn: no event on the closed id, live subscription still receives, socket still answers - EVENT at the open tick: OK with ok=true - fragmented and binary frames honoured, next command still answered - invalid JSON refused with a NOTICE, socket usable afterwards --- .../features/socket-race/socket-race.feature | 33 ++ .../socket-race/socket-race.feature.ts | 310 ++++++++++++++++++ 2 files changed, 343 insertions(+) create mode 100644 test/integration/features/socket-race/socket-race.feature create mode 100644 test/integration/features/socket-race/socket-race.feature.ts diff --git a/test/integration/features/socket-race/socket-race.feature b/test/integration/features/socket-race/socket-race.feature new file mode 100644 index 000000000..4fc90fe94 --- /dev/null +++ b/test/integration/features/socket-race/socket-race.feature @@ -0,0 +1,33 @@ +@socket-race +Feature: Commands that race the connection are answered, not just survived + + # Issue #1052: a relay that drops a REQ has survived perfectly well. Each + # case asserts the reply the client is entitled to — EOSE, OK, NOTICE — not + # merely that the socket stayed open. + + Scenario: Subscriptions sent at the open tick each receive their EOSE + When 10 sockets subscribe at the open tick, repeated 20 times + Then each socket received EOSE for its own subscription + + Scenario: Two subscriptions sent in one turn are answered in order + When a socket sends REQ "pipelined-1" and then REQ "pipelined-2" in one turn + Then the socket received EOSE for "pipelined-1" before "pipelined-2" + + Scenario: A subscription closed in the same turn never receives an event + Given a publisher called race-cancel-author + When a socket subscribes at the open tick and closes it in the same turn + And the publisher publishes a text note + Then "race-cancel" received no event and the socket still answers + + Scenario: An event sent at the open tick receives its OK + Given a publisher called race-writer + When the publisher sends an event at the open tick + Then the socket received OK for that event + + Scenario: Fragmented and binary frames are honoured + When a socket sends a fragmented REQ and a binary REQ in one turn + Then both subscriptions received EOSE and the socket still answers + + Scenario: Invalid JSON mid-stream is refused and the socket stays usable + When a socket sends invalid JSON and then a REQ + Then the socket received a NOTICE and then EOSE for the REQ diff --git a/test/integration/features/socket-race/socket-race.feature.ts b/test/integration/features/socket-race/socket-race.feature.ts new file mode 100644 index 000000000..db78bc429 --- /dev/null +++ b/test/integration/features/socket-race/socket-race.feature.ts @@ -0,0 +1,310 @@ +import { Given, Then, When, World } from '@cucumber/cucumber' +import { expect } from 'chai' +import { ReplaySubject } from 'rxjs' +import WebSocket, { RawData } from 'ws' + +import { createEvent, createIdentity, publishEvent } from '../helpers' +import { streams } from '../shared' +import { OutgoingMessage } from '../../../../src/@types/messages' +import { Event } from '../../../../src/@types/event' + +const RELAY_URL = 'ws://localhost:18808' +const MESSAGE_TIMEOUT_MS = 10_000 +const STEP_TIMEOUT_MS = 120_000 + +const world = (context: World>) => context.parameters as Record + +type Connection = { + ws: WebSocket + messages: ReplaySubject + frames: OutgoingMessage[] +} + +const eoseIds = (connection: Connection): string[] => + connection.frames.filter((message) => message[0] === 'EOSE').map((message) => message[1] as string) + +const waitFor = async (predicate: () => boolean, label: string, timeoutMs = MESSAGE_TIMEOUT_MS): Promise => { + const startedAt = Date.now() + while (!predicate()) { + if (Date.now() - startedAt > timeoutMs) { + throw new Error(`timed out waiting for ${label}`) + } + await new Promise((resolve) => setTimeout(resolve, 10)) + } +} + +// The listener is attached before 'open' resolves, so a frame written in the +// same turn as the handshake is captured rather than dropped — the whole point +// of these cases, which the shared client steps cannot express. +const createConnection = async (onOpen?: (ws: WebSocket) => void): Promise => { + const ws = new WebSocket(RELAY_URL) + const messages = new ReplaySubject(1_000, 60_000) + const frames: OutgoingMessage[] = [] + + ws.on('message', (raw: RawData) => { + const message = JSON.parse(raw.toString()) as OutgoingMessage + frames.push(message) + messages.next(message) + }) + + await new Promise((resolve, reject) => { + ws.once('open', () => { + onOpen?.(ws) + resolve() + }) + ws.once('error', reject) + }) + + streams.set(ws, messages) + + return { ws, messages, frames } +} + +const track = (context: World>, key: string, connection: Connection): Connection => { + world(context).clients[key] = connection.ws + world(context).connections = world(context).connections ?? {} + world(context).connections[key] = connection + + return connection +} + +When( + /^(\d+) sockets subscribe at the open tick, repeated (\d+) times$/, + { timeout: STEP_TIMEOUT_MS }, + async function (this: World>, sockets: string, rounds: string) { + const socketCount = Number(sockets) + const roundCount = Number(rounds) + const results: { id: string; eose: string[] }[] = [] + world(this).stormExpected = socketCount * roundCount + world(this).stormResults = results + + for (let round = 0; round < roundCount; round++) { + const opened = await Promise.all( + Array.from({ length: socketCount }, (_, index) => { + const id = `storm-${round}-${index}` + return createConnection((ws) => ws.send(JSON.stringify(['REQ', id, { kinds: [1] }]))).then((connection) => ({ + id, + index, + connection, + })) + }), + ) + + for (const { id, index, connection } of opened) { + track(this, `storm-${round}-${index}`, connection) + await waitFor(() => eoseIds(connection).includes(id), `EOSE for ${id}`) + results.push({ id, eose: eoseIds(connection) }) + connection.ws.close() + } + } + }, +) + +Then( + 'each socket received EOSE for its own subscription', + { timeout: STEP_TIMEOUT_MS }, + function (this: World>) { + const results = world(this).stormResults as { id: string; eose: string[] }[] + + expect(results).to.have.length(world(this).stormExpected) + for (const { id, eose } of results) { + // Exactly one EOSE, carrying that socket's own subscription id. + expect(eose).to.deep.equal([id]) + } + }, +) + +When( + /^a socket sends REQ "([^"]+)" and then REQ "([^"]+)" in one turn$/, + { timeout: STEP_TIMEOUT_MS }, + async function (this: World>, first: string, second: string) { + const connection = track( + this, + 'pipelined', + await createConnection((ws) => { + ws.send(JSON.stringify(['REQ', first, { kinds: [1] }])) + ws.send(JSON.stringify(['REQ', second, { kinds: [1] }])) + }), + ) + + await waitFor(() => eoseIds(connection).length >= 2, `EOSE for both ${first} and ${second}`) + }, +) + +Then( + /^the socket received EOSE for "([^"]+)" before "([^"]+)"$/, + { timeout: STEP_TIMEOUT_MS }, + function (this: World>, first: string, second: string) { + expect(eoseIds(world(this).connections.pipelined)).to.deep.equal([first, second]) + }, +) + +Given( + /^a publisher called ([\w-]+)$/, + { timeout: STEP_TIMEOUT_MS }, + function (this: World>, name: string) { + world(this).identities[name] = createIdentity(name) + world(this).publisher = name + }, +) + +When( + 'a socket subscribes at the open tick and closes it in the same turn', + { timeout: STEP_TIMEOUT_MS }, + async function (this: World>) { + const { pubkey } = world(this).identities[world(this).publisher] + + const connection = track( + this, + 'cancel', + await createConnection((ws) => { + ws.send(JSON.stringify(['REQ', 'race-cancel', { authors: [pubkey], kinds: [1] }])) + ws.send(JSON.stringify(['CLOSE', 'race-cancel'])) + }), + ) + + // A second subscription on the same socket is the barrier: its EOSE means + // the CLOSE ahead of it has been handled, without sleeping on the clock. + connection.ws.send(JSON.stringify(['REQ', 'after-close', { authors: [pubkey], kinds: [1] }])) + await waitFor(() => eoseIds(connection).includes('after-close'), 'EOSE for after-close') + }, +) + +When( + 'the publisher publishes a text note', + { timeout: STEP_TIMEOUT_MS }, + async function (this: World>) { + const identity = world(this).identities[world(this).publisher] + const connection = track(this, 'publisher', await createConnection()) + + const event = await createEvent( + { pubkey: identity.pubkey, kind: 1, content: 'race event', tags: [] }, + identity.privkey, + ) + world(this).publishedEvent = event + world(this).events[world(this).publisher] = world(this).events[world(this).publisher] ?? [] + world(this).events[world(this).publisher].push(event) + + const result = await publishEvent(connection.ws, event) + expect(result[2], result[3]).to.equal(true) + + // The live subscription must receive it: the event is broadcast, so its + // absence from `race-cancel` is a closed subscription, not a missed write. + const cancel: Connection = world(this).connections.cancel + await waitFor( + () => cancel.frames.some((message) => message[0] === 'EVENT' && message[1] === 'after-close'), + 'the event on the live subscription', + ) + }, +) + +Then( + /^"([^"]+)" received no event and the socket still answers$/, + { timeout: STEP_TIMEOUT_MS }, + function (this: World>, subscription: string) { + const cancel: Connection = world(this).connections.cancel + + expect(cancel.frames.filter((message) => message[0] === 'EVENT' && message[1] === subscription)).to.deep.equal([]) + expect(cancel.frames.filter((message) => message[0] === 'NOTICE')).to.deep.equal([]) + expect(eoseIds(cancel)).to.include('after-close') + expect(cancel.ws.readyState).to.equal(WebSocket.OPEN) + }, +) + +When( + 'the publisher sends an event at the open tick', + { timeout: STEP_TIMEOUT_MS }, + async function (this: World>) { + const identity = world(this).identities[world(this).publisher] + const event: Event = await createEvent( + { pubkey: identity.pubkey, kind: 1, content: 'ok at the open tick', tags: [] }, + identity.privkey, + ) + + world(this).publishedEvent = event + world(this).events[world(this).publisher] = world(this).events[world(this).publisher] ?? [] + world(this).events[world(this).publisher].push(event) + + const connection = track(this, 'writer', await createConnection((ws) => ws.send(JSON.stringify(['EVENT', event])))) + + await waitFor( + () => connection.frames.some((message) => message[0] === 'OK' && message[1] === event.id), + 'OK for the event sent at the open tick', + ) + }, +) + +Then( + 'the socket received OK for that event', + { timeout: STEP_TIMEOUT_MS }, + function (this: World>) { + const { publishedEvent } = world(this) + const ok = (world(this).connections.writer as Connection).frames.find( + (message) => message[0] === 'OK' && message[1] === publishedEvent.id, + ) + + expect(ok?.[2]).to.equal(true) + }, +) + +When( + 'a socket sends a fragmented REQ and a binary REQ in one turn', + { timeout: STEP_TIMEOUT_MS }, + async function (this: World>) { + const connection = track( + this, + 'framing', + await createConnection((ws) => { + ws.send('["REQ","frag', { fin: false }) + ws.send('-sub",{"kinds":[1]}]', { fin: true }) + ws.send(Buffer.from('["REQ","binary-frame",{"kinds":[1]}]'), { binary: true }) + }), + ) + + await waitFor(() => eoseIds(connection).length >= 2, 'EOSE for the fragmented and the binary REQ') + + connection.ws.send(JSON.stringify(['REQ', 'after-frames', { kinds: [1] }])) + await waitFor(() => eoseIds(connection).includes('after-frames'), 'EOSE for the next command') + }, +) + +Then( + 'both subscriptions received EOSE and the socket still answers', + { timeout: STEP_TIMEOUT_MS }, + function (this: World>) { + expect(eoseIds(world(this).connections.framing)).to.deep.equal(['frag-sub', 'binary-frame', 'after-frames']) + }, +) + +When( + 'a socket sends invalid JSON and then a REQ', + { timeout: STEP_TIMEOUT_MS }, + async function (this: World>) { + const connection = track( + this, + 'invalid', + await createConnection((ws) => { + ws.send('{"broken_json_without_closing_bracket') + ws.send(JSON.stringify(['REQ', 'valid-after-bad', { kinds: [1] }])) + }), + ) + + await waitFor( + () => connection.frames.some((message) => message[0] === 'NOTICE') && eoseIds(connection).length >= 1, + 'a NOTICE and an EOSE', + ) + }, +) + +Then( + 'the socket received a NOTICE and then EOSE for the REQ', + { timeout: STEP_TIMEOUT_MS }, + function (this: World>) { + const connection: Connection = world(this).connections.invalid + const notice = connection.frames.find((message) => message[0] === 'NOTICE') + + expect(notice?.[1]).to.match(/^invalid:/) + expect(eoseIds(connection)).to.deep.equal(['valid-after-bad']) + expect(connection.ws.readyState).to.equal(WebSocket.OPEN) + }, +) From 5993433ce31c92ca85a7dd5d7f1ad7603e43e329 Mon Sep 17 00:00:00 2001 From: chappie-daemon Date: Sat, 7 Nov 2026 18:05:18 -0500 Subject: [PATCH 4/9] test(relay): do not require EOSE order across independent REQs onClientMessage does not make a later message wait for an earlier one, and each subscribe handler sends EOSE when its own query finishes, so two replies can arrive in either order. Both must arrive; their order is not the relay's to promise. after-frames stays last in the framing case because it is sent only after the other two have answered. --- .../features/socket-race/socket-race.feature | 4 ++-- .../features/socket-race/socket-race.feature.ts | 17 ++++++++++++++--- 2 files changed, 16 insertions(+), 5 deletions(-) diff --git a/test/integration/features/socket-race/socket-race.feature b/test/integration/features/socket-race/socket-race.feature index 4fc90fe94..b47c55995 100644 --- a/test/integration/features/socket-race/socket-race.feature +++ b/test/integration/features/socket-race/socket-race.feature @@ -9,9 +9,9 @@ Feature: Commands that race the connection are answered, not just survived When 10 sockets subscribe at the open tick, repeated 20 times Then each socket received EOSE for its own subscription - Scenario: Two subscriptions sent in one turn are answered in order + Scenario: Two subscriptions sent in one turn are both answered When a socket sends REQ "pipelined-1" and then REQ "pipelined-2" in one turn - Then the socket received EOSE for "pipelined-1" before "pipelined-2" + Then the socket received EOSE for "pipelined-1" and "pipelined-2" Scenario: A subscription closed in the same turn never receives an event Given a publisher called race-cancel-author diff --git a/test/integration/features/socket-race/socket-race.feature.ts b/test/integration/features/socket-race/socket-race.feature.ts index db78bc429..f898c03e0 100644 --- a/test/integration/features/socket-race/socket-race.feature.ts +++ b/test/integration/features/socket-race/socket-race.feature.ts @@ -132,10 +132,14 @@ When( ) Then( - /^the socket received EOSE for "([^"]+)" before "([^"]+)"$/, + /^the socket received EOSE for "([^"]+)" and "([^"]+)"$/, { timeout: STEP_TIMEOUT_MS }, function (this: World>, first: string, second: string) { - expect(eoseIds(world(this).connections.pipelined)).to.deep.equal([first, second]) + // Two REQs are independent handlers, and onClientMessage does not make the + // second wait for the first, so each EOSE is sent when its own query + // finishes. Both replies must arrive; their order is not the relay's to + // promise, and asserting it would fail a correct run. + expect(eoseIds(world(this).connections.pipelined).sort()).to.deep.equal([first, second].sort()) }, ) @@ -272,7 +276,14 @@ Then( 'both subscriptions received EOSE and the socket still answers', { timeout: STEP_TIMEOUT_MS }, function (this: World>) { - expect(eoseIds(world(this).connections.framing)).to.deep.equal(['frag-sub', 'binary-frame', 'after-frames']) + const ids = eoseIds(world(this).connections.framing) + + // The fragmented and the binary REQ are independent handlers, so their + // replies may arrive in either order. after-frames was sent only after + // both had answered, so it must come last. + expect(ids).to.have.length(3) + expect(ids.slice(2)).to.deep.equal(['after-frames']) + expect(ids.slice(0, 2).sort()).to.deep.equal(['binary-frame', 'frag-sub']) }, ) From 9681b37c66dc1d424fca6cb0227600f111f8b990 Mon Sep 17 00:00:00 2001 From: chappie-daemon Date: Sat, 7 Nov 2026 18:24:57 -0500 Subject: [PATCH 5/9] test(relay): make the admission window explicit in the race cases MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Without a delay the admission check answers before a frame written at the handshake can reach the server, so the cases passed even with the early-message buffer removed — the race they name never happened. Admission is a Redis round trip in production; hold it for 50 ms for these scenarios only, which is what the #1045 unit case does with its own unresolved promise. --- .../socket-race/socket-race.feature.ts | 24 ++++++++++++++++++- 1 file changed, 23 insertions(+), 1 deletion(-) diff --git a/test/integration/features/socket-race/socket-race.feature.ts b/test/integration/features/socket-race/socket-race.feature.ts index f898c03e0..1500a5097 100644 --- a/test/integration/features/socket-race/socket-race.feature.ts +++ b/test/integration/features/socket-race/socket-race.feature.ts @@ -1,16 +1,38 @@ -import { Given, Then, When, World } from '@cucumber/cucumber' +import { After, Before, Given, Then, When, World } from '@cucumber/cucumber' import { expect } from 'chai' import { ReplaySubject } from 'rxjs' +import Sinon from 'sinon' import WebSocket, { RawData } from 'ws' import { createEvent, createIdentity, publishEvent } from '../helpers' import { streams } from '../shared' import { OutgoingMessage } from '../../../../src/@types/messages' import { Event } from '../../../../src/@types/event' +import * as rateLimiterMiddleware from '../../../../src/handlers/request-handlers/rate-limiter-middleware' const RELAY_URL = 'ws://localhost:18808' const MESSAGE_TIMEOUT_MS = 10_000 const STEP_TIMEOUT_MS = 120_000 +// Admission is a Redis round trip in production, and the frames a client +// writes at the handshake arrive inside that window. Without a delay the check +// answers before the first frame can reach the server, which makes these cases +// pass against an adapter that drops it. +const ADMISSION_DELAY_MS = 50 + +let admissionSandbox: Sinon.SinonSandbox | undefined + +Before({ tags: '@socket-race' }, function () { + admissionSandbox = Sinon.createSandbox() + admissionSandbox.stub(rateLimiterMiddleware, 'isRateLimited').callsFake(async () => { + await new Promise((resolve) => setTimeout(resolve, ADMISSION_DELAY_MS)) + return false + }) +}) + +After({ tags: '@socket-race' }, function () { + admissionSandbox?.restore() + admissionSandbox = undefined +}) const world = (context: World>) => context.parameters as Record From 2fd69b089ef861b2d0073b21b7f69108842a6392 Mon Sep 17 00:00:00 2001 From: chappie-daemon Date: Sat, 7 Nov 2026 18:43:25 -0500 Subject: [PATCH 6/9] test(relay): gate admission instead of sleeping on it A fixed 50 ms window races CI load: the check can answer before a frame reaches the server, and the case then passes against an adapter that drops it. Admission now waits on an explicit gate per scenario, released only once the frames written at the handshake have been flushed. --- .../socket-race/socket-race.feature.ts | 153 ++++++++++++------ 1 file changed, 104 insertions(+), 49 deletions(-) diff --git a/test/integration/features/socket-race/socket-race.feature.ts b/test/integration/features/socket-race/socket-race.feature.ts index 1500a5097..34a9bbc49 100644 --- a/test/integration/features/socket-race/socket-race.feature.ts +++ b/test/integration/features/socket-race/socket-race.feature.ts @@ -13,26 +13,6 @@ import * as rateLimiterMiddleware from '../../../../src/handlers/request-handler const RELAY_URL = 'ws://localhost:18808' const MESSAGE_TIMEOUT_MS = 10_000 const STEP_TIMEOUT_MS = 120_000 -// Admission is a Redis round trip in production, and the frames a client -// writes at the handshake arrive inside that window. Without a delay the check -// answers before the first frame can reach the server, which makes these cases -// pass against an adapter that drops it. -const ADMISSION_DELAY_MS = 50 - -let admissionSandbox: Sinon.SinonSandbox | undefined - -Before({ tags: '@socket-race' }, function () { - admissionSandbox = Sinon.createSandbox() - admissionSandbox.stub(rateLimiterMiddleware, 'isRateLimited').callsFake(async () => { - await new Promise((resolve) => setTimeout(resolve, ADMISSION_DELAY_MS)) - return false - }) -}) - -After({ tags: '@socket-race' }, function () { - admissionSandbox?.restore() - admissionSandbox = undefined -}) const world = (context: World>) => context.parameters as Record @@ -42,6 +22,8 @@ type Connection = { frames: OutgoingMessage[] } +type Frame = { data: string | Buffer; options?: { fin?: boolean; binary?: boolean } } + const eoseIds = (connection: Connection): string[] => connection.frames.filter((message) => message[0] === 'EOSE').map((message) => message[1] as string) @@ -55,10 +37,59 @@ const waitFor = async (predicate: () => boolean, label: string, timeoutMs = MESS } } +// Admission is a Redis round trip in production, and the frames a client writes +// at the handshake arrive inside that window. A fixed delay would race CI load: +// the check could answer first and the case would then pass against an adapter +// that drops frames. Admission is therefore held on an explicit gate and +// released only once the scenario's frames have been flushed to the socket. +type AdmissionGate = { promise: Promise; release: () => void } + +const createAdmissionGate = (): AdmissionGate => { + let release: () => void = () => undefined + const promise = new Promise((resolve) => { + release = () => resolve(false) + }) + + return { promise, release } +} + +let admissionGate = createAdmissionGate() +let admissionSandbox: Sinon.SinonSandbox | undefined + +const releaseAdmission = (): void => { + const gate = admissionGate + admissionGate = createAdmissionGate() + gate.release() +} + +Before({ tags: '@socket-race' }, function () { + admissionGate = createAdmissionGate() + admissionSandbox = Sinon.createSandbox() + admissionSandbox.stub(rateLimiterMiddleware, 'isRateLimited').callsFake(() => admissionGate.promise) +}) + +After({ tags: '@socket-race' }, function () { + admissionSandbox?.restore() + admissionSandbox = undefined + releaseAdmission() +}) + +// Resolves when every frame has been flushed, which is the latest moment the +// test can observe before the server is allowed to admit the client. +const sendFrames = (ws: WebSocket, frames: Frame[]): Promise => + Promise.all( + frames.map( + ({ data, options }) => + new Promise((resolve, reject) => { + ws.send(data, options ?? {}, (error?: Error) => (error ? reject(error) : resolve())) + }), + ), + ).then(() => undefined) + // The listener is attached before 'open' resolves, so a frame written in the // same turn as the handshake is captured rather than dropped — the whole point // of these cases, which the shared client steps cannot express. -const createConnection = async (onOpen?: (ws: WebSocket) => void): Promise => { +const createConnection = async (onOpen?: (ws: WebSocket) => Promise | void): Promise => { const ws = new WebSocket(RELAY_URL) const messages = new ReplaySubject(1_000, 60_000) const frames: OutgoingMessage[] = [] @@ -71,8 +102,7 @@ const createConnection = async (onOpen?: (ws: WebSocket) => void): Promise((resolve, reject) => { ws.once('open', () => { - onOpen?.(ws) - resolve() + Promise.resolve(onOpen?.(ws)).then(resolve, reject) }) ws.once('error', reject) }) @@ -97,6 +127,7 @@ When( const socketCount = Number(sockets) const roundCount = Number(rounds) const results: { id: string; eose: string[] }[] = [] + world(this).stormExpected = socketCount * roundCount world(this).stormResults = results @@ -104,14 +135,15 @@ When( const opened = await Promise.all( Array.from({ length: socketCount }, (_, index) => { const id = `storm-${round}-${index}` - return createConnection((ws) => ws.send(JSON.stringify(['REQ', id, { kinds: [1] }]))).then((connection) => ({ - id, - index, - connection, - })) + return createConnection((ws) => sendFrames(ws, [{ data: JSON.stringify(['REQ', id, { kinds: [1] }]) }])).then( + (connection) => ({ id, index, connection }), + ) }), ) + // Every frame from this round is flushed; only now let admission through. + releaseAdmission() + for (const { id, index, connection } of opened) { track(this, `storm-${round}-${index}`, connection) await waitFor(() => eoseIds(connection).includes(id), `EOSE for ${id}`) @@ -143,12 +175,15 @@ When( const connection = track( this, 'pipelined', - await createConnection((ws) => { - ws.send(JSON.stringify(['REQ', first, { kinds: [1] }])) - ws.send(JSON.stringify(['REQ', second, { kinds: [1] }])) - }), + await createConnection((ws) => + sendFrames(ws, [ + { data: JSON.stringify(['REQ', first, { kinds: [1] }]) }, + { data: JSON.stringify(['REQ', second, { kinds: [1] }]) }, + ]), + ), ) + releaseAdmission() await waitFor(() => eoseIds(connection).length >= 2, `EOSE for both ${first} and ${second}`) }, ) @@ -157,8 +192,8 @@ Then( /^the socket received EOSE for "([^"]+)" and "([^"]+)"$/, { timeout: STEP_TIMEOUT_MS }, function (this: World>, first: string, second: string) { - // Two REQs are independent handlers, and onClientMessage does not make the - // second wait for the first, so each EOSE is sent when its own query + // The two REQs are independent handlers, and onClientMessage does not make + // the second wait for the first, so each EOSE is sent when its own query // finishes. Both replies must arrive; their order is not the relay's to // promise, and asserting it would fail a correct run. expect(eoseIds(world(this).connections.pipelined).sort()).to.deep.equal([first, second].sort()) @@ -183,12 +218,16 @@ When( const connection = track( this, 'cancel', - await createConnection((ws) => { - ws.send(JSON.stringify(['REQ', 'race-cancel', { authors: [pubkey], kinds: [1] }])) - ws.send(JSON.stringify(['CLOSE', 'race-cancel'])) - }), + await createConnection((ws) => + sendFrames(ws, [ + { data: JSON.stringify(['REQ', 'race-cancel', { authors: [pubkey], kinds: [1] }]) }, + { data: JSON.stringify(['CLOSE', 'race-cancel']) }, + ]), + ), ) + releaseAdmission() + // A second subscription on the same socket is the barrier: its EOSE means // the CLOSE ahead of it has been handled, without sleeping on the clock. connection.ws.send(JSON.stringify(['REQ', 'after-close', { authors: [pubkey], kinds: [1] }])) @@ -203,6 +242,8 @@ When( const identity = world(this).identities[world(this).publisher] const connection = track(this, 'publisher', await createConnection()) + releaseAdmission() + const event = await createEvent( { pubkey: identity.pubkey, kind: 1, content: 'race event', tags: [] }, identity.privkey, @@ -232,7 +273,7 @@ Then( expect(cancel.frames.filter((message) => message[0] === 'EVENT' && message[1] === subscription)).to.deep.equal([]) expect(cancel.frames.filter((message) => message[0] === 'NOTICE')).to.deep.equal([]) - expect(eoseIds(cancel)).to.include('after-close') + expect(eoseIds(cancel)).to.deep.equal(['after-close']) expect(cancel.ws.readyState).to.equal(WebSocket.OPEN) }, ) @@ -251,7 +292,13 @@ When( world(this).events[world(this).publisher] = world(this).events[world(this).publisher] ?? [] world(this).events[world(this).publisher].push(event) - const connection = track(this, 'writer', await createConnection((ws) => ws.send(JSON.stringify(['EVENT', event])))) + const connection = track( + this, + 'writer', + await createConnection((ws) => sendFrames(ws, [{ data: JSON.stringify(['EVENT', event]) }])), + ) + + releaseAdmission() await waitFor( () => connection.frames.some((message) => message[0] === 'OK' && message[1] === event.id), @@ -280,13 +327,17 @@ When( const connection = track( this, 'framing', - await createConnection((ws) => { - ws.send('["REQ","frag', { fin: false }) - ws.send('-sub",{"kinds":[1]}]', { fin: true }) - ws.send(Buffer.from('["REQ","binary-frame",{"kinds":[1]}]'), { binary: true }) - }), + await createConnection((ws) => + sendFrames(ws, [ + { data: '["REQ","frag', options: { fin: false } }, + { data: '-sub",{"kinds":[1]}]', options: { fin: true } }, + { data: Buffer.from('["REQ","binary-frame",{"kinds":[1]}]'), options: { binary: true } }, + ]), + ), ) + releaseAdmission() + await waitFor(() => eoseIds(connection).length >= 2, 'EOSE for the fragmented and the binary REQ') connection.ws.send(JSON.stringify(['REQ', 'after-frames', { kinds: [1] }])) @@ -316,12 +367,16 @@ When( const connection = track( this, 'invalid', - await createConnection((ws) => { - ws.send('{"broken_json_without_closing_bracket') - ws.send(JSON.stringify(['REQ', 'valid-after-bad', { kinds: [1] }])) - }), + await createConnection((ws) => + sendFrames(ws, [ + { data: '{"broken_json_without_closing_bracket' }, + { data: JSON.stringify(['REQ', 'valid-after-bad', { kinds: [1] }]) }, + ]), + ), ) + releaseAdmission() + await waitFor( () => connection.frames.some((message) => message[0] === 'NOTICE') && eoseIds(connection).length >= 1, 'a NOTICE and an EOSE', From 61bb05d88c048c0436bb0e93eccd3fad88960c9a Mon Sep 17 00:00:00 2001 From: chappie-daemon Date: Sat, 7 Nov 2026 19:00:10 -0500 Subject: [PATCH 7/9] prove: remove the early-message buffer to show the gated race cases fail --- src/adapters/web-socket-server-adapter.ts | 2 -- 1 file changed, 2 deletions(-) diff --git a/src/adapters/web-socket-server-adapter.ts b/src/adapters/web-socket-server-adapter.ts index c85fd57d5..65f9eb33e 100644 --- a/src/adapters/web-socket-server-adapter.ts +++ b/src/adapters/web-socket-server-adapter.ts @@ -175,8 +175,6 @@ export class WebSocketServerAdapter extends WebServerAdapter implements IWebSock earlyMessages.push({ data, isBinary }) } - client.on('message', onEarlyMessage) - client.pause() let accepted = false try { From 003c6633b46feaf16a40600baef6acbca59f7b35 Mon Sep 17 00:00:00 2001 From: chappie-daemon Date: Sat, 7 Nov 2026 19:13:50 -0500 Subject: [PATCH 8/9] test(relay): settle before releasing the admission gate The gate alone proves admission cannot answer early; it says nothing about whether the frames have reached the server by the time it does. Wait for them after the flush, then release, so a client writing at the handshake is always inside the admission window. --- .../socket-race/socket-race.feature.ts | 27 ++++++++++++------- 1 file changed, 18 insertions(+), 9 deletions(-) diff --git a/test/integration/features/socket-race/socket-race.feature.ts b/test/integration/features/socket-race/socket-race.feature.ts index 34a9bbc49..5f5b58adf 100644 --- a/test/integration/features/socket-race/socket-race.feature.ts +++ b/test/integration/features/socket-race/socket-race.feature.ts @@ -56,7 +56,14 @@ const createAdmissionGate = (): AdmissionGate => { let admissionGate = createAdmissionGate() let admissionSandbox: Sinon.SinonSandbox | undefined -const releaseAdmission = (): void => { +// The gate keeps admission from answering early; the settle wait makes sure the +// frames have actually reached the server's socket before it does, so a client +// that writes at the handshake is always inside the admission window. +const ADMISSION_SETTLE_MS = 50 + +const releaseAdmission = async (settleMs = ADMISSION_SETTLE_MS): Promise => { + await new Promise((resolve) => setTimeout(resolve, settleMs)) + const gate = admissionGate admissionGate = createAdmissionGate() gate.release() @@ -71,7 +78,9 @@ Before({ tags: '@socket-race' }, function () { After({ tags: '@socket-race' }, function () { admissionSandbox?.restore() admissionSandbox = undefined - releaseAdmission() + // Nothing is waiting on this admission any more; release it so a failing + // scenario cannot leave a connection parked on the gate. + void releaseAdmission(0) }) // Resolves when every frame has been flushed, which is the latest moment the @@ -142,7 +151,7 @@ When( ) // Every frame from this round is flushed; only now let admission through. - releaseAdmission() + await releaseAdmission() for (const { id, index, connection } of opened) { track(this, `storm-${round}-${index}`, connection) @@ -183,7 +192,7 @@ When( ), ) - releaseAdmission() + await releaseAdmission() await waitFor(() => eoseIds(connection).length >= 2, `EOSE for both ${first} and ${second}`) }, ) @@ -226,7 +235,7 @@ When( ), ) - releaseAdmission() + await releaseAdmission() // A second subscription on the same socket is the barrier: its EOSE means // the CLOSE ahead of it has been handled, without sleeping on the clock. @@ -242,7 +251,7 @@ When( const identity = world(this).identities[world(this).publisher] const connection = track(this, 'publisher', await createConnection()) - releaseAdmission() + await releaseAdmission() const event = await createEvent( { pubkey: identity.pubkey, kind: 1, content: 'race event', tags: [] }, @@ -298,7 +307,7 @@ When( await createConnection((ws) => sendFrames(ws, [{ data: JSON.stringify(['EVENT', event]) }])), ) - releaseAdmission() + await releaseAdmission() await waitFor( () => connection.frames.some((message) => message[0] === 'OK' && message[1] === event.id), @@ -336,7 +345,7 @@ When( ), ) - releaseAdmission() + await releaseAdmission() await waitFor(() => eoseIds(connection).length >= 2, 'EOSE for the fragmented and the binary REQ') @@ -375,7 +384,7 @@ When( ), ) - releaseAdmission() + await releaseAdmission() await waitFor( () => connection.frames.some((message) => message[0] === 'NOTICE') && eoseIds(connection).length >= 1, From 55dc0a8011e272caed538ea816d891352332ecdb Mon Sep 17 00:00:00 2001 From: chappie-daemon Date: Sat, 7 Nov 2026 19:27:26 -0500 Subject: [PATCH 9/9] test(relay): put a live subscription in the close-race batch A closed subscription leaves nothing to assert on, so the case passed even when the whole batch was dropped. batch-probe rides in the same flush and outlives the CLOSE: its EOSE proves the batch arrived, and the published event reaching it proves the broadcast happened, so an empty race-cancel means closed rather than lost. --- .../features/socket-race/socket-race.feature.ts | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/test/integration/features/socket-race/socket-race.feature.ts b/test/integration/features/socket-race/socket-race.feature.ts index 5f5b58adf..c87962392 100644 --- a/test/integration/features/socket-race/socket-race.feature.ts +++ b/test/integration/features/socket-race/socket-race.feature.ts @@ -224,6 +224,10 @@ When( async function (this: World>) { const { pubkey } = world(this).identities[world(this).publisher] + // race-cancel is closed in the same flush, so it can leave no trace to + // assert on. batch-probe rides in the same flush and lives on: its EOSE + // proves the batch reached the server at all, which a dropped frame would + // otherwise hide behind an empty race-cancel. const connection = track( this, 'cancel', @@ -231,16 +235,14 @@ When( sendFrames(ws, [ { data: JSON.stringify(['REQ', 'race-cancel', { authors: [pubkey], kinds: [1] }]) }, { data: JSON.stringify(['CLOSE', 'race-cancel']) }, + { data: JSON.stringify(['REQ', 'batch-probe', { authors: [pubkey], kinds: [1] }]) }, ]), ), ) await releaseAdmission() - // A second subscription on the same socket is the barrier: its EOSE means - // the CLOSE ahead of it has been handled, without sleeping on the clock. - connection.ws.send(JSON.stringify(['REQ', 'after-close', { authors: [pubkey], kinds: [1] }])) - await waitFor(() => eoseIds(connection).includes('after-close'), 'EOSE for after-close') + await waitFor(() => eoseIds(connection).includes('batch-probe'), 'EOSE for batch-probe') }, ) @@ -268,7 +270,7 @@ When( // absence from `race-cancel` is a closed subscription, not a missed write. const cancel: Connection = world(this).connections.cancel await waitFor( - () => cancel.frames.some((message) => message[0] === 'EVENT' && message[1] === 'after-close'), + () => cancel.frames.some((message) => message[0] === 'EVENT' && message[1] === 'batch-probe'), 'the event on the live subscription', ) }, @@ -282,7 +284,7 @@ Then( expect(cancel.frames.filter((message) => message[0] === 'EVENT' && message[1] === subscription)).to.deep.equal([]) expect(cancel.frames.filter((message) => message[0] === 'NOTICE')).to.deep.equal([]) - expect(eoseIds(cancel)).to.deep.equal(['after-close']) + expect(eoseIds(cancel)).to.deep.equal(['batch-probe']) expect(cancel.ws.readyState).to.equal(WebSocket.OPEN) }, )