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/integration/features/socket-race/socket-race.feature b/test/integration/features/socket-race/socket-race.feature new file mode 100644 index 000000000..b47c55995 --- /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 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" and "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..f898c03e0 --- /dev/null +++ b/test/integration/features/socket-race/socket-race.feature.ts @@ -0,0 +1,321 @@ +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 "([^"]+)" 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 + // 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()) + }, +) + +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>) { + 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']) + }, +) + +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) + }, +) 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..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,13 +20,36 @@ const waitFor = async (predicate: () => boolean, timeoutMs = 3_000): Promise { +// Every case below sends at the open tick, but an immediately resolved +// admission check answers before the client's frames can reach the server, so +// the adapter installs its normal listener first and the buffering path never +// runs. Hold admission open until the frames are known to be in the server's +// queue, then admit every client at once. +const holdAdmission = (stub: Sinon.SinonStub) => { + 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) + 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 +60,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 +74,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 +105,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 +121,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 +134,208 @@ 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 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', id, { kinds: [1] }]))) + resolve() + }) + ws.on('error', reject) + }) + }) + + await Promise.all(connectPromises) + await admit() + + await waitFor(() => { + let total = 0 + for (const frames of receivedFramesByClient.values()) { + total += frames.length + } + return total >= N + }) + + expect(receivedFramesByClient.size).to.equal(N) + + const receivedIds: string[] = [] + for (const frames of receivedFramesByClient.values()) { + 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()) + }) + + // -------------------------------------------------------------------------- + // 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 admit = holdAdmission(isRateLimitedStub) + 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 admit() + 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 admit = holdAdmission(isRateLimitedStub) + 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 admit() + 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 admit = holdAdmission(isRateLimitedStub) + 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 admit() + 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 admit = holdAdmission(isRateLimitedStub) + 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 admit() + 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 admit = holdAdmission(isRateLimitedStub) + 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 admit() + 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) + }) })