From 81edd16652a5472e25fa8c1627dc9b1fdbda57df Mon Sep 17 00:00:00 2001 From: Tirthraj Chavan Date: Sat, 7 Nov 2026 11:52:46 +0530 Subject: [PATCH 1/4] =?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/4] 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/4] 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/4] 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']) }, )