mirror of
https://github.com/stablyai/orca.git
synced 2026-09-23 08:02:31 +00:00
fix(relay): key the abort index by the id's string form so a string id can still be cancelled
The flat map's template key folded a request id of 7 and "7" onto one entry; keying the raw value split them, so rpc.cancel (which coerces through Number) missed a string-id request. Restore the coercion at the index.
This commit is contained in:
@@ -11,7 +11,12 @@ export class ClientRequestAborts {
|
||||
// keys are scattered through the map, so any correct loop still visits every entry, which made a
|
||||
// full churn of N clients cost K*N*(N+1)/2 visits. Only an index makes a teardown proportional to
|
||||
// what that client actually owns.
|
||||
private readonly byClient = new Map<number, Map<number, AbortController>>()
|
||||
//
|
||||
// Why the inner key is a string: the codec only checks `jsonrpc === '2.0'`, so a request `id` can
|
||||
// arrive as `"7"` while `rpc.cancel` coerces its `id` through `Number(...)` and looks up `7`. The
|
||||
// flat map's template key folded both onto `"7"`; keying the raw value would file them in
|
||||
// different buckets and silently drop the cancel. `String(...)` is the template literal's coercion.
|
||||
private readonly byClient = new Map<number, Map<string, AbortController>>()
|
||||
|
||||
create(
|
||||
clientId: number,
|
||||
@@ -20,15 +25,15 @@ export class ClientRequestAborts {
|
||||
const controller = new AbortController()
|
||||
let requests = this.byClient.get(clientId)
|
||||
if (!requests) {
|
||||
requests = new Map<number, AbortController>()
|
||||
requests = new Map<string, AbortController>()
|
||||
this.byClient.set(clientId, requests)
|
||||
}
|
||||
requests.set(requestId, controller)
|
||||
requests.set(String(requestId), controller)
|
||||
return { key: { clientId, requestId }, controller }
|
||||
}
|
||||
|
||||
get(clientId: number, requestId: number): AbortController | undefined {
|
||||
return this.byClient.get(clientId)?.get(requestId)
|
||||
return this.byClient.get(clientId)?.get(String(requestId))
|
||||
}
|
||||
|
||||
delete(key: ClientRequestAbortHandle): void {
|
||||
@@ -36,7 +41,7 @@ export class ClientRequestAborts {
|
||||
if (!requests) {
|
||||
return
|
||||
}
|
||||
requests.delete(key.requestId)
|
||||
requests.delete(String(key.requestId))
|
||||
// Why drop the empty bucket: otherwise a churned client leaves an entry behind for the life of
|
||||
// the relay, which is the retention the index exists to avoid.
|
||||
if (requests.size === 0) {
|
||||
|
||||
@@ -13,7 +13,7 @@ type Probed = {
|
||||
requestHandlers: Map<string, unknown>
|
||||
notificationHandlers: Map<string, unknown>
|
||||
requestAborts: {
|
||||
byClient: Map<number, Map<number, AbortController>>
|
||||
byClient: Map<number, Map<string, AbortController>>
|
||||
create: (clientId: number, requestId: number) => unknown
|
||||
}
|
||||
publicationLedger: { clientBytes: Map<string, number>; aggregateBytes: number }
|
||||
|
||||
@@ -0,0 +1,43 @@
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import { RelayDispatcher } from './dispatcher'
|
||||
import { encodeFrame, MessageType } from './protocol'
|
||||
|
||||
// Why both id shapes: parseJsonRpcMessage only checks the version, so a request id may arrive as a
|
||||
// string, while rpc.cancel coerces its id through Number(...). The abort index must file both
|
||||
// under one key or the cancel for a string-id request is silently dropped.
|
||||
describe('rpc.cancel request-id coercion', () => {
|
||||
let dispatcher: RelayDispatcher
|
||||
|
||||
beforeEach(() => {
|
||||
vi.useFakeTimers()
|
||||
dispatcher = new RelayDispatcher(() => {})
|
||||
})
|
||||
|
||||
afterEach(() => {
|
||||
dispatcher.dispose()
|
||||
vi.useRealTimers()
|
||||
})
|
||||
|
||||
it.each([
|
||||
{ label: 'numeric', requestId: 7, cancelId: 7 },
|
||||
{ label: 'string', requestId: '7', cancelId: '7' },
|
||||
{ label: 'string request, numeric cancel', requestId: '7', cancelId: 7 }
|
||||
])('aborts an in-flight request with a $label id', async ({ requestId, cancelId }) => {
|
||||
let signal: AbortSignal | undefined
|
||||
dispatcher.onRequest('test.slow', (_params, ctx) => {
|
||||
signal = ctx.signal
|
||||
return new Promise(() => {})
|
||||
})
|
||||
// Raw frames: the typed encoder would not admit a string id, and that is the point.
|
||||
const rawFrame = (msg: Record<string, unknown>, seq: number): Buffer =>
|
||||
encodeFrame(MessageType.Regular, seq, 0, Buffer.from(JSON.stringify(msg), 'utf-8'))
|
||||
|
||||
dispatcher.feed(rawFrame({ jsonrpc: '2.0', id: requestId, method: 'test.slow' }, 1))
|
||||
await vi.advanceTimersByTimeAsync(0)
|
||||
expect(signal?.aborted).toBe(false)
|
||||
|
||||
dispatcher.feed(rawFrame({ jsonrpc: '2.0', method: 'rpc.cancel', params: { id: cancelId } }, 2))
|
||||
|
||||
expect(signal?.aborted).toBe(true)
|
||||
})
|
||||
})
|
||||
@@ -114,7 +114,7 @@ describe('relay hot-path operation counts', () => {
|
||||
}
|
||||
}
|
||||
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: byClient is the private index this test exists to measure; the shape mirrors its declaration in client-request-aborts.ts.
|
||||
const byClient = (aborts as unknown as { byClient: Map<number, Map<number, AbortController>> })
|
||||
const byClient = (aborts as unknown as { byClient: Map<number, Map<string, AbortController>> })
|
||||
.byClient
|
||||
|
||||
// The census must be able to find things: prove the maps really hold 160 controllers across 40
|
||||
@@ -126,11 +126,11 @@ describe('relay hot-path operation counts', () => {
|
||||
}
|
||||
expect(totalControllers).toBe(clientCount * inFlightPerClient)
|
||||
|
||||
const index = new CountingMap<number, Map<number, AbortController>>()
|
||||
const index = new CountingMap<number, Map<string, AbortController>>()
|
||||
for (const [k, v] of byClient) {
|
||||
index.set(k, v)
|
||||
}
|
||||
const targetBucket = new CountingMap<number, AbortController>()
|
||||
const targetBucket = new CountingMap<string, AbortController>()
|
||||
for (const [k, v] of byClient.get(1)!) {
|
||||
targetBucket.set(k, v)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user