From 57ffbe368d1394ac3f1de0e27e99a7a1249daacd Mon Sep 17 00:00:00 2001 From: Dinh Le Date: Fri, 3 Jul 2026 14:01:53 +0700 Subject: [PATCH 1/3] fix(client, server): preserve message ordering for blob messages in peer adapters Blob messages require an async step to load bytes before sending, while text messages are sent synchronously. This meant that when a blob message was queued before a text message, the text message could reach the peer first, breaking expected ordering. --- .../adapters/message-port/rpc-link.test.ts | 20 ++++ .../src/adapters/message-port/transport.ts | 36 ++++-- .../src/adapters/websocket/transport.ts | 15 ++- .../src/adapters/message-port/handler.ts | 59 +++++----- .../adapters/message-port/rpc-handler.test.ts | 9 ++ .../server/src/adapters/websocket/handler.ts | 50 +++++--- .../adapters/websocket/rpc-handler.test.ts | 8 +- packages/shared/src/buffer.test.ts | 110 ++++++++++++++---- packages/shared/src/buffer.ts | 42 +++++-- packages/shared/src/index.ts | 2 + tests/rpc/data-transfer.test.ts | 3 +- 11 files changed, 255 insertions(+), 99 deletions(-) diff --git a/packages/client/src/adapters/message-port/rpc-link.test.ts b/packages/client/src/adapters/message-port/rpc-link.test.ts index 1b63ccb23..8b64a325c 100644 --- a/packages/client/src/adapters/message-port/rpc-link.test.ts +++ b/packages/client/src/adapters/message-port/rpc-link.test.ts @@ -161,4 +161,24 @@ describe('rpcLink', () => { await promise }) + + it('ignore invalid messages', async () => { + const port = createPort() + const orpc = createORPCClient(new RPCLink({ port })) as any + + const promise = expect(orpc.ping('input')).resolves.toEqual('pong') + + await vi.waitFor(() => expect(port.postMessage).toHaveBeenCalledTimes(1)) + + const decoded = decodeRequest(port.postMessage.mock.calls[0]![0]) + const id = decoded.message.id + + // Invalid message — should be ignored + onMessage({ data: { invalid: true } }) + + // Correct message — should be processed + onMessage({ data: await createResponseMessage({ id }) }) + + await promise + }) }) diff --git a/packages/client/src/adapters/message-port/transport.ts b/packages/client/src/adapters/message-port/transport.ts index 1cff85fcc..33ac8fe73 100644 --- a/packages/client/src/adapters/message-port/transport.ts +++ b/packages/client/src/adapters/message-port/transport.ts @@ -1,11 +1,11 @@ import type { Promisable, Value } from '@orpc/shared' import type { StandardLazyResponse, StandardRequest } from '@standardserver/core' -import type { DecodePeerMessageOptions, EncodePeerMessageOptions } from '@standardserver/peer' +import type { DecodePeerMessageOptions, EncodePeerMessageOptions, ServerPeerSendMessage } from '@standardserver/peer' import type { ClientContext, ClientOptions } from '../../types' import type { StandardLinkTransport } from '../standard' import type { SupportedMessagePort } from './message-port' -import { isPlainObject, toStringOrBytes, value } from '@orpc/shared' -import { ClientPeer, decodePeerMessage, encodePeerMessage, isServerPeerSendMessage } from '@standardserver/peer' +import { value } from '@orpc/shared' +import { ClientPeer, decodePeerMessage, encodePeerMessage, isPeerMessage, isServerPeerSendMessage } from '@standardserver/peer' import { onMessagePortClose, onMessagePortMessage, postMessagePortMessage } from './message-port' type DecodedRequestMessage = ConstructorParameters[0] extends (message: infer TMessage) => unknown @@ -53,19 +53,31 @@ export class MessagePortLinkTransport implements Standa } }) - onMessagePortMessage(port, async (message) => { - if (isPlainObject(message)) { - await this.peer.message(message as any) - return - } + onMessagePortMessage(port, async (data) => { + let peerMessage: ServerPeerSendMessage | undefined - const encodedMessage = await toStringOrBytes(message) + if (typeof data === 'string' || data instanceof Uint8Array) { + // MessagePort receives the exact payload sent, and `encodePeerMessage` only returns string or Uint8Array. + const result = decodePeerMessage(data as string | Uint8Array, decodePeerMessageOptions) + if (result.matched && isServerPeerSendMessage(result.message)) { + peerMessage = result.message + } + } - const result = decodePeerMessage(encodedMessage, decodePeerMessageOptions) + else if (isPeerMessage(data) && isServerPeerSendMessage(data)) { + peerMessage = data + } - if (result.matched && isServerPeerSendMessage(result.message)) { - await this.peer.message(result.message) + if (peerMessage === undefined) { + return { matched: false } } + + /** + * Message order is important: loading -> decode -> .message. + * This flow must stay synchronous, or we need to use `sequential` helper + */ + await this.peer.message(peerMessage) + return { matched: true } }) onMessagePortClose(port, () => { diff --git a/packages/client/src/adapters/websocket/transport.ts b/packages/client/src/adapters/websocket/transport.ts index 7023728dd..0f09f4b34 100644 --- a/packages/client/src/adapters/websocket/transport.ts +++ b/packages/client/src/adapters/websocket/transport.ts @@ -3,7 +3,7 @@ import type { StandardLazyResponse, StandardRequest } from '@standardserver/core import type { DecodePeerMessageOptions, EncodePeerMessageOptions } from '@standardserver/peer' import type { ClientContext, ClientOptions } from '../../types' import type { StandardLinkTransport } from '../standard' -import { AbortError, promiseWithResolvers, runWithSignal, sleep, toStringOrBytes } from '@orpc/shared' +import { AbortError, loadBytes, promiseWithResolvers, runWithSignal, sequential, sleep, toStringOrBytes } from '@orpc/shared' import { ClientPeer, decodePeerMessage, encodePeerMessage, isServerPeerSendMessage } from '@standardserver/peer' /** @@ -196,13 +196,18 @@ export class WebSocketLinkTransport implements Standard }) } - websocket.addEventListener('message', async (event: MessageEvent) => { - const message = await toStringOrBytes(event.data) + /** + * Message order is important: loading -> decode -> .message. + * This flow must stay synchronous, or we need to use `sequential` helper + */ + websocket.addEventListener('message', sequential(async (event: MessageEvent) => { + // For better compatibility avoid control or depend on websocket.binaryType + const message = event.data instanceof Blob ? await loadBytes(event.data) : toStringOrBytes(event.data) const result = decodePeerMessage(message, this.decodePeerMessageOptions) if (result.matched && isServerPeerSendMessage(result.message)) { - await peer.message(result.message) + peer.message(result.message) } - }) + })) websocket.addEventListener('close', async (event) => { connectingResolvers?.resolve() diff --git a/packages/server/src/adapters/message-port/handler.ts b/packages/server/src/adapters/message-port/handler.ts index 9f87eeb28..98737cde8 100644 --- a/packages/server/src/adapters/message-port/handler.ts +++ b/packages/server/src/adapters/message-port/handler.ts @@ -1,12 +1,12 @@ import type { SupportedMessagePort } from '@orpc/client/message-port' import type { MaybeOptionalOptions, Promisable, Value } from '@orpc/shared' -import type { DecodePeerMessageOptions, EncodePeerMessageOptions } from '@standardserver/peer' +import type { ClientPeerSendMessage, DecodePeerMessageOptions, EncodePeerMessageOptions } from '@standardserver/peer' import type { Context } from '../../context' import type { StandardHandler } from '../standard' import type { StandardPeerRequestHandlerOptions } from '../standard-peer' import { onMessagePortClose, onMessagePortMessage, postMessagePortMessage } from '@orpc/client/message-port' -import { isPlainObject, resolveMaybeOptionalOptions, toStringOrBytes, value } from '@orpc/shared' -import { decodePeerMessage, encodePeerMessage, isClientPeerSendMessage, ServerPeer } from '@standardserver/peer' +import { resolveMaybeOptionalOptions, value } from '@orpc/shared' +import { decodePeerMessage, encodePeerMessage, isClientPeerSendMessage, isPeerMessage, ServerPeer } from '@standardserver/peer' import { createStandardPeerRequestHandler } from '../standard-peer' type DecodedResponseMessage = ConstructorParameters[0] extends (message: infer TMessage) => unknown @@ -54,12 +54,18 @@ export class MessagePortHandler { } /** - * Attaches necessary event listeners to a message port to handle incoming messages and peer management. + * Attaches message and close listeners to a message port. + * + * Prefer this over calling `.message()` and `.close()` manually. */ upgrade( port: SupportedMessagePort, ...rest: MaybeOptionalOptions> ): void { + /** + * Message order is important: loading -> decode -> .message. + * This flow must stay synchronous, or we need to use `sequential` helper + */ onMessagePortMessage(port, message => this.message(port, message, ...rest)) onMessagePortClose(port, () => this.close(port)) } @@ -67,13 +73,11 @@ export class MessagePortHandler { /** * Handles a single message received from a message port. * - * @warning AVOID calling this method directly if `.upgrade()` is used, as `.upgrade()` already sets up necessary event listeners to call this method for incoming messages and manage peer lifecycle. - * - * @param port The message port instance, require consistent instance across messages for proper peer management + * @param port The message port instance. Use the same instance for all messages. */ async message( port: SupportedMessagePort, - data: any, + data: unknown, ...rest: MaybeOptionalOptions> ): Promise<{ matched: boolean }> { let peer = this.peers.get(port) @@ -91,35 +95,36 @@ export class MessagePortHandler { })) } - if (isPlainObject(data)) { - await peer.message( - data as any, - createStandardPeerRequestHandler(this.handler, resolveMaybeOptionalOptions(rest)), - ) + let peerMessage: ClientPeerSendMessage | undefined - return { matched: true } + if (typeof data === 'string' || data instanceof Uint8Array) { + // MessagePort receives the exact payload sent, and `encodePeerMessage` only returns string or Uint8Array. + const result = decodePeerMessage(data as string | Uint8Array, this.decodePeerMessageOptions) + if (result.matched && isClientPeerSendMessage(result.message)) { + peerMessage = result.message + } } - const message = await toStringOrBytes(data) - - const result = decodePeerMessage(message, this.decodePeerMessageOptions) + else if (isPeerMessage(data) && isClientPeerSendMessage(data)) { + peerMessage = data + } - if (result.matched && isClientPeerSendMessage(result.message)) { - await peer.message( - result.message, - createStandardPeerRequestHandler(this.handler, resolveMaybeOptionalOptions(rest)), - ) + if (peerMessage === undefined) { + return { matched: false } } - return result + /** + * Message order is important: loading -> decode -> .message. + * This flow must stay synchronous, or we need to use `sequential` helper + */ + await peer.message(peerMessage, createStandardPeerRequestHandler(this.handler, resolveMaybeOptionalOptions(rest))) + return { matched: true } } /** - * Called when a message port is closed, to clean up any associated peer state. - * - * @warning AVOID calling this method directly if `.upgrade()` is used, as `.upgrade()` already sets up necessary event listeners to call this method for incoming messages and manage peer lifecycle. + * Cleans up peer state for a closed message port. * - * @param port The message port instance to clean up, must be the same instance used in `.message()` calls to properly clean up + * @param port The same message port instance passed to `.message()`. */ async close(port: SupportedMessagePort): Promise { const peer = this.peers.get(port) diff --git a/packages/server/src/adapters/message-port/rpc-handler.test.ts b/packages/server/src/adapters/message-port/rpc-handler.test.ts index 05110524f..9bc283b79 100644 --- a/packages/server/src/adapters/message-port/rpc-handler.test.ts +++ b/packages/server/src/adapters/message-port/rpc-handler.test.ts @@ -196,6 +196,8 @@ describe('rpcHandler', () => { }) expect(postMessage).not.toHaveBeenCalled() + + await handler.close(serverPort as any) // safe to invoke multiple times }) it('can receive and send un-encoded messages with transfer option (structured clone)', async () => { @@ -285,4 +287,11 @@ describe('rpcHandler', () => { expect(ws.send).not.toHaveBeenCalled() }) + + it('ignore invalid message format', async () => { + const handler = createHandler() + const { serverPort } = createPort() + const result = await handler.message(serverPort as any, { invalid: true }) + expect(result.matched).toBe(false) + }) }) diff --git a/packages/server/src/adapters/websocket/handler.ts b/packages/server/src/adapters/websocket/handler.ts index 4071eb051..8cbd23ba7 100644 --- a/packages/server/src/adapters/websocket/handler.ts +++ b/packages/server/src/adapters/websocket/handler.ts @@ -1,9 +1,9 @@ -import type { MaybeOptionalOptions } from '@orpc/shared' +import type { Arrayable, MaybeOptionalOptions } from '@orpc/shared' import type { DecodePeerMessageOptions, EncodePeerMessageOptions } from '@standardserver/peer' import type { Context } from '../../context' import type { StandardHandler } from '../standard' import type { StandardPeerRequestHandlerOptions } from '../standard-peer' -import { resolveMaybeOptionalOptions, toStringOrBytes } from '@orpc/shared' +import { loadBytes, resolveMaybeOptionalOptions, sequential, toStringOrBytes } from '@orpc/shared' import { decodePeerMessage, encodePeerMessage, isClientPeerSendMessage, ServerPeer } from '@standardserver/peer' import { createStandardPeerRequestHandler } from '../standard-peer' @@ -41,16 +41,20 @@ export class WebSocketHandler { } /** - * Handles a single message received from a WebSocket. + * Handles a single WebSocket message. * - * @warning AVOID calling this method, if `.upgrade()` is used, as `.upgrade()` already sets up necessary event listeners to call this method for incoming messages and manage peer lifecycle. + * Message order matters. Call this immediately after receiving a message, + * before any other async work, to preserve ordering. * - * @param ws The WebSocket instance, require consistent instance across messages for proper peer management - * @param data The message data received from the WebSocket, can be string, ArrayBuffer, Blob, ... + * To avoid async Blob reads, configure `ws.binaryType` to return binary data + * directly (for example, `arraybuffer`). + * + * @param ws The WebSocket instance. Use the same instance for all messages. + * @param data The received message. Binary data should already be loaded (not a `Blob`). */ async message( ws: WebSocketLike, - data: string | ArrayBuffer | Blob | Exclude[0], undefined>[0][] | Pick, 'buffer' | 'byteOffset' | 'byteLength'>, + data: Arrayable, 'buffer' | 'byteOffset' | 'byteLength'>>, ...rest: MaybeOptionalOptions> ): Promise<{ matched: boolean }> { let peer = this.peers.get(ws) @@ -62,10 +66,12 @@ export class WebSocketHandler { })) } - const encoded = await toStringOrBytes(data) - + /** + * Message order is important: loading -> decode -> .message. + * This flow must stay synchronous, or we need to use `sequential` helper + */ + const encoded = toStringOrBytes(data) const result = decodePeerMessage(encoded, this.decodePeerMessageOptions) - if (result.matched && isClientPeerSendMessage(result.message)) { await peer.message( result.message, @@ -77,11 +83,9 @@ export class WebSocketHandler { } /** - * Called when a websocket is closed, to clean up any associated peer state. + * Cleans up peer state for a closed WebSocket. * - * @warning AVOID calling this method, if `.upgrade()` is used, as `.upgrade()` already sets up necessary event listeners to call this method for incoming messages and manage peer lifecycle. - * - * @param ws The WebSocket instance to clean up, must be the same instance used in `.message()` calls to properly clean up + * @param ws The same WebSocket instance passed to `.message()`. */ async close(ws: WebSocketLike): Promise { const peer = this.peers.get(ws) @@ -93,16 +97,26 @@ export class WebSocketHandler { } /** - * Attaches websocket event listeners for message and close handling. + * Attaches message and close event listeners to a WebSocket. * - * Use this instead of calling `.message()` and `.close()` manually. - * Requires a websocket-like object that supports `addEventListener` and `removeEventListener`. + * Prefer this over calling `.message()` and `.close()` manually. */ upgrade( ws: Pick, ...rest: MaybeOptionalOptions> ): void { - ws.addEventListener('message', event => this.message(ws, event.data, ...rest)) + /** + * Message order is important: loading -> decode -> .message. + * This flow must stay synchronous, or we need to use `sequential` helper + */ + ws.addEventListener('message', sequential(async (event) => { + // For better compatibility avoid control or depend on websocket.binaryType + const data = event.data instanceof Blob ? loadBytes(event.data) : event.data + + // Not awaited: `this.message` runs business logic that may be slow, + // and awaiting it would block decoding of subsequent messages. + this.message(ws, data, ...rest) + })) ws.addEventListener('close', () => this.close(ws)) } } diff --git a/packages/server/src/adapters/websocket/rpc-handler.test.ts b/packages/server/src/adapters/websocket/rpc-handler.test.ts index b6d7c8c19..a481858cb 100644 --- a/packages/server/src/adapters/websocket/rpc-handler.test.ts +++ b/packages/server/src/adapters/websocket/rpc-handler.test.ts @@ -75,13 +75,9 @@ describe('rpcHandler', () => { const message = await createRequestMessage() return new TextEncoder().encode(message as string).buffer }], - ['blob', async () => { + ['arrayBuffer parts', async () => { const message = await createRequestMessage() - return new Blob([message]) - }], - ['blob parts', async () => { - const message = await createRequestMessage() - return [message] + return [new TextEncoder().encode(message as string).buffer] }], ])('handles %s request', async (_type, createMessage) => { const handler = createHandler() diff --git a/packages/shared/src/buffer.test.ts b/packages/shared/src/buffer.test.ts index 3d726cbad..70de0f1b4 100644 --- a/packages/shared/src/buffer.test.ts +++ b/packages/shared/src/buffer.test.ts @@ -17,43 +17,109 @@ it('loadBytes', async () => { }) describe('toStringOrBytes', () => { - it('string passthrough', async () => { - expect(await toStringOrBytes('test')).toBe('test') + it('string passthrough', () => { + expect(toStringOrBytes('test')).toBe('test') }) - it('arrayBuffer', async () => { + it('arrayBuffer', () => { const input = new TextEncoder().encode('test').buffer - const result = await toStringOrBytes(input) + const result = toStringOrBytes(input) expect(result).toBeInstanceOf(Uint8Array) expect(new TextDecoder().decode(result as Uint8Array)).toBe('test') }) - it('blob', async () => { - const blob = new Blob(['test'], { type: 'text/plain' }) - const result = await toStringOrBytes(blob) - expect(result).toBeInstanceOf(Uint8Array) - expect(new TextDecoder().decode(result as Uint8Array)).toBe('test') - }) - - it('blobPart[]', async () => { - const result = await toStringOrBytes(['te', 'st']) - expect(result).toBeInstanceOf(Uint8Array) - expect(new TextDecoder().decode(result as Uint8Array)).toBe('test') - }) - - it('uint8Array passthrough', async () => { + it('uint8Array passthrough', () => { const input = new Uint8Array([0, 1, 2, 3]) - const result = await toStringOrBytes(input) + const result = toStringOrBytes(input) expect(result).toBe(input) // should be the same instance }) - it('uint8Array view-like input', async () => { + it('uint8Array view-like input', () => { const bytes = new Uint8Array([0, 1, 2, 3]).subarray(1, 3) - const input: Pick, 'buffer' | 'byteOffset' | 'byteLength'> = { buffer: bytes.buffer, byteOffset: bytes.byteOffset, byteLength: bytes.byteLength } + const input: Pick, 'buffer' | 'byteOffset' | 'byteLength'> = { + buffer: bytes.buffer, + byteOffset: bytes.byteOffset, + byteLength: bytes.byteLength, + } - const result = await toStringOrBytes(input) + const result = toStringOrBytes(input) expect(result).toBeInstanceOf(Uint8Array) expect(result).not.toBe(bytes) // should be a different view, not the original input expect(result).toEqual(bytes) }) + + describe('array input', () => { + it('all-string array concatenates to bytes (UTF-8 encoded)', () => { + const result = toStringOrBytes(['te', 'st']) + expect(result).toBeInstanceOf(Uint8Array) + expect(new TextDecoder().decode(result as Uint8Array)).toBe('test') + }) + + it('single-string array concatenates to bytes', () => { + const result = toStringOrBytes(['solo']) + expect(result).toBeInstanceOf(Uint8Array) + expect(new TextDecoder().decode(result as Uint8Array)).toBe('solo') + }) + + it('empty array returns empty Uint8Array', () => { + const result = toStringOrBytes([]) + expect(result).toBeInstanceOf(Uint8Array) + expect((result as Uint8Array).byteLength).toBe(0) + }) + + it('array of ArrayBuffers concatenates to bytes', () => { + const a = new TextEncoder().encode('ab').buffer + const b = new TextEncoder().encode('cd').buffer + const result = toStringOrBytes([a, b]) + expect(result).toBeInstanceOf(Uint8Array) + expect(new TextDecoder().decode(result as Uint8Array)).toBe('abcd') + }) + + it('array of Uint8Arrays concatenates to bytes', () => { + const a = new Uint8Array([1, 2]) + const b = new Uint8Array([3, 4]) + const result = toStringOrBytes([a, b]) + expect(result).toBeInstanceOf(Uint8Array) + expect(result).toEqual(new Uint8Array([1, 2, 3, 4])) + }) + + it('array of byte-view-like objects concatenates to bytes', () => { + const source = new Uint8Array([9, 8, 7, 6]) + const view: Pick, 'buffer' | 'byteOffset' | 'byteLength'> = { + buffer: source.buffer, + byteOffset: 1, + byteLength: 2, + } + const result = toStringOrBytes([view]) + expect(result).toBeInstanceOf(Uint8Array) + expect(result).toEqual(new Uint8Array([8, 7])) + }) + + it('mixed array (strings + bytes) concatenates to bytes, UTF-8 encoding strings', () => { + const result = toStringOrBytes(['te', new Uint8Array([0x73, 0x74])]) // 'st' + expect(result).toBeInstanceOf(Uint8Array) + expect(new TextDecoder().decode(result as Uint8Array)).toBe('test') + }) + + it('mixed array preserves item order across all byte-like variants', () => { + const strPart = 'a' // 1 byte + const bufPart = new TextEncoder().encode('b').buffer // 1 byte + const u8Part = new Uint8Array([0x63]) // 'c', 1 byte + const source = new Uint8Array([0x64, 0x65]) // 'd', 'e' + const viewPart: Pick, 'buffer' | 'byteOffset' | 'byteLength'> = { + buffer: source.buffer, + byteOffset: 0, + byteLength: 1, + } // 'd' + + const result = toStringOrBytes([strPart, bufPart, u8Part, viewPart]) + expect(new TextDecoder().decode(result as Uint8Array)).toBe('abcd') + }) + + it('mixed array with multi-byte UTF-8 string encodes correctly', () => { + const result = toStringOrBytes(['€', new Uint8Array([0x21])]) // '€' is 3 bytes in UTF-8, then '!' + const expectedBytes = new Uint8Array([...new TextEncoder().encode('€'), 0x21]) + expect(result).toEqual(expectedBytes) + }) + }) }) diff --git a/packages/shared/src/buffer.ts b/packages/shared/src/buffer.ts index 43c653e83..f8f6d1bfe 100644 --- a/packages/shared/src/buffer.ts +++ b/packages/shared/src/buffer.ts @@ -1,3 +1,5 @@ +import type { Arrayable } from 'type-fest' + /** * Load Request/Response/Blob/File/.. to a buffer (Uint8Array). * @@ -17,7 +19,7 @@ export async function loadBytes(source: Pick): Re * - the original string value, or * - a Uint8Array view over the source bytes. */ -export async function toStringOrBytes(source: string | ArrayBuffer | Blob | Exclude[0], undefined>[0][] | Pick, 'buffer' | 'byteOffset' | 'byteLength'>): Promise>> { +export function toStringOrBytes(source: Arrayable, 'buffer' | 'byteOffset' | 'byteLength'>>): string | Awaited> { if (typeof source === 'string') { return source } @@ -26,17 +28,43 @@ export async function toStringOrBytes(source: string | ArrayBuffer | Blob | Excl return new Uint8Array(source) } - if (source instanceof Blob) { - return loadBytes(source) - } - if (Array.isArray(source)) { - return loadBytes(new Blob(source)) + return concatBytes(source) } if (source instanceof Uint8Array) { - return source as Uint8Array + return source as Awaited> } return new Uint8Array(source.buffer, source.byteOffset, source.byteLength) } + +function toBytes(item: string | ArrayBuffer | Pick, 'buffer' | 'byteOffset' | 'byteLength'>): Awaited> { + if (typeof item === 'string') { + return new TextEncoder().encode(item) + } + + if (item instanceof ArrayBuffer) { + return new Uint8Array(item) + } + + if (item instanceof Uint8Array) { + return item as Awaited> + } + + return new Uint8Array(item.buffer, item.byteOffset, item.byteLength) +} + +function concatBytes(items: Array, 'buffer' | 'byteOffset' | 'byteLength'>>): Uint8Array { + const chunks = items.map(toBytes) + const totalLength = chunks.reduce((sum, chunk) => sum + chunk.byteLength, 0) + + const result = new Uint8Array(totalLength) + let offset = 0 + for (const chunk of chunks) { + result.set(chunk, offset) + offset += chunk.byteLength + } + + return result +} diff --git a/packages/shared/src/index.ts b/packages/shared/src/index.ts index 84052f80d..c8fac2d46 100644 --- a/packages/shared/src/index.ts +++ b/packages/shared/src/index.ts @@ -28,6 +28,7 @@ export { isAsyncIteratorObject, isTypescriptObject, parseEmptyableJSON, + sequential, SequentialIdGenerator, sleep, stringifyJSON, @@ -40,6 +41,7 @@ export type { } from '@standardserver/shared' export type { + Arrayable, IsEqual, PartialDeep, Promisable, diff --git a/tests/rpc/data-transfer.test.ts b/tests/rpc/data-transfer.test.ts index 320271209..4456c73db 100644 --- a/tests/rpc/data-transfer.test.ts +++ b/tests/rpc/data-transfer.test.ts @@ -38,8 +38,7 @@ describe.each([ return expect(client.lastEventId(null, { lastEventId })).resolves.toEqual(lastEventId) }) - // TODO: https://github.com/h3js/crossws/issues/198 - it.skipIf(adapter === 'crossws')('support octet stream and transfer octet in parallel', async () => { + it('support octet stream and transfer octet in parallel', async () => { const stream = new ReadableStream({ async start(controller) { controller.enqueue('order 1') From aa8f1e51435771391859b44ef5832519a2911832 Mon Sep 17 00:00:00 2001 From: Dinh Le Date: Fri, 3 Jul 2026 14:06:19 +0700 Subject: [PATCH 2/3] improve crossws jsdocs --- packages/server/src/adapters/crossws/handler.ts | 17 +++++++++++------ 1 file changed, 11 insertions(+), 6 deletions(-) diff --git a/packages/server/src/adapters/crossws/handler.ts b/packages/server/src/adapters/crossws/handler.ts index 0bd250853..828e42788 100644 --- a/packages/server/src/adapters/crossws/handler.ts +++ b/packages/server/src/adapters/crossws/handler.ts @@ -37,9 +37,12 @@ export class experimental_CrosswsHandler { } /** - * Handles a single message received from a crossws Peer. + * Handles a single WebSocket message. * - * @param ws The crossws Peer instance, require consistent instance across messages for proper peer management + * Message order matters. Call this immediately after receiving a message, + * before any other async work, to preserve ordering. + * + * @param ws The crossws peer instance. Use the same instance for all messages. */ async message( ws: CrosswsPeerLike, @@ -54,10 +57,12 @@ export class experimental_CrosswsHandler { })) } + /** + * Message order is important: loading -> decode -> .message. + * This flow must stay synchronous, or we need to use `sequential` helper + */ const encodedMessage = typeof message.rawData === 'string' ? message.rawData : message.uint8Array() as Uint8Array - const result = decodePeerMessage(encodedMessage, this.decodePeerMessageOptions) - if (result.matched && isClientPeerSendMessage(result.message)) { await peer.message( result.message, @@ -69,9 +74,9 @@ export class experimental_CrosswsHandler { } /** - * Closes the peer connection and cleans up associated resources. + * Cleans up peer state for a closed WebSocket. * - * @param ws The crossws Peer instance to close, require consistent instance for proper peer management + * @param ws The same crossws peer instance passed to `.message()`. */ async close(ws: CrosswsPeerLike): Promise { const server = this.peers.get(ws) From 9e689907d61e042114837caf5012043d72b61a89 Mon Sep 17 00:00:00 2001 From: Dinh Le Date: Fri, 3 Jul 2026 14:39:42 +0700 Subject: [PATCH 3/3] fix: ensure proper message handling and closure in transport adapters --- .../src/adapters/message-port/transport.ts | 4 +-- .../src/adapters/websocket/transport.ts | 2 ++ .../src/adapters/message-port/handler.ts | 3 +- .../server/src/adapters/websocket/handler.ts | 5 +-- .../adapters/websocket/rpc-handler.test.ts | 32 +++++++++++++++++++ 5 files changed, 41 insertions(+), 5 deletions(-) diff --git a/packages/client/src/adapters/message-port/transport.ts b/packages/client/src/adapters/message-port/transport.ts index 33ac8fe73..0558d8f51 100644 --- a/packages/client/src/adapters/message-port/transport.ts +++ b/packages/client/src/adapters/message-port/transport.ts @@ -80,8 +80,8 @@ export class MessagePortLinkTransport implements Standa return { matched: true } }) - onMessagePortClose(port, () => { - this.peer.close() + onMessagePortClose(port, async () => { + await this.peer.close() }) } diff --git a/packages/client/src/adapters/websocket/transport.ts b/packages/client/src/adapters/websocket/transport.ts index 0f09f4b34..14c037fd9 100644 --- a/packages/client/src/adapters/websocket/transport.ts +++ b/packages/client/src/adapters/websocket/transport.ts @@ -205,6 +205,8 @@ export class WebSocketLinkTransport implements Standard const message = event.data instanceof Blob ? await loadBytes(event.data) : toStringOrBytes(event.data) const result = decodePeerMessage(message, this.decodePeerMessageOptions) if (result.matched && isServerPeerSendMessage(result.message)) { + // Not awaited: `peer.message` runs handling message may be slow, + // and awaiting it would block decoding of subsequent messages. peer.message(result.message) } })) diff --git a/packages/server/src/adapters/message-port/handler.ts b/packages/server/src/adapters/message-port/handler.ts index 98737cde8..36139fe25 100644 --- a/packages/server/src/adapters/message-port/handler.ts +++ b/packages/server/src/adapters/message-port/handler.ts @@ -130,8 +130,9 @@ export class MessagePortHandler { const peer = this.peers.get(port) if (peer) { - await peer.close() + // delete before close to avoid potential race conditions this.peers.delete(port) + await peer.close() } } } diff --git a/packages/server/src/adapters/websocket/handler.ts b/packages/server/src/adapters/websocket/handler.ts index 8cbd23ba7..fa9759ce4 100644 --- a/packages/server/src/adapters/websocket/handler.ts +++ b/packages/server/src/adapters/websocket/handler.ts @@ -91,8 +91,9 @@ export class WebSocketHandler { const peer = this.peers.get(ws) if (peer) { - await peer.close() + // delete before close to avoid potential race conditions this.peers.delete(ws) + await peer.close() } } @@ -111,7 +112,7 @@ export class WebSocketHandler { */ ws.addEventListener('message', sequential(async (event) => { // For better compatibility avoid control or depend on websocket.binaryType - const data = event.data instanceof Blob ? loadBytes(event.data) : event.data + const data = event.data instanceof Blob ? await loadBytes(event.data) : event.data // Not awaited: `this.message` runs business logic that may be slow, // and awaiting it would block decoding of subsequent messages. diff --git a/packages/server/src/adapters/websocket/rpc-handler.test.ts b/packages/server/src/adapters/websocket/rpc-handler.test.ts index a481858cb..39df02cf7 100644 --- a/packages/server/src/adapters/websocket/rpc-handler.test.ts +++ b/packages/server/src/adapters/websocket/rpc-handler.test.ts @@ -189,5 +189,37 @@ describe('rpcHandler', () => { }) expect(ws.send).not.toHaveBeenCalled() + + onClose?.() // safely call again to ensure no error is thrown + }) + + it('handles Blob messages via upgrade', async () => { + let onMessage: ((event: { data: Blob }) => void) | undefined + + const handler = createHandler() + + const ws = { + addEventListener: vi.fn((event: string, callback: any) => { + if (event === 'message') { + onMessage = callback + } + }), + send: vi.fn(() => undefined), + } + + handler.upgrade(ws as any) + + const request = await createRequestMessage() + onMessage?.({ data: new Blob([request as string]) }) + + await vi.waitFor(() => { + expect(ws.send).toHaveBeenCalledTimes(1) + }) + + const decoded = decodePeerMessage((ws as any).send.mock.calls[0][0]) as any + + expect(decoded.matched).toBe(true) + expect(decoded.message.kind).toBe('response') + expect(decoded.message.json.status).toBe(200) }) })