feat: validate workspace protocol frames in browser
This commit is contained in:
@@ -6,7 +6,7 @@
|
||||
"dev": "deno run -A npm:vite@7.2.7 dev",
|
||||
"dev:backend": "cd ../.. && cargo run -p yoi-workspace-server --bin yoi-server -- serve --listen 127.0.0.1:8787",
|
||||
"check": "deno run -A npm:@sveltejs/kit@2.49.4 sync && deno run -A npm:svelte-check@4.3.4 --tsconfig ./tsconfig.json",
|
||||
"test": "deno test --allow-read=src,test,tests,../../docs --allow-env=LOG,VSCODE_TEXTMATE_DEBUG,NODE_ENV tests/workspace-model.test.ts tests/repository-api.test.ts tests/workspace-catalog.test.ts tests/profile-api.test.ts tests/skill-api.test.ts tests/tooltip-contract.test.ts tests/bevel-contract.test.ts tests/bevel-line-contract.test.ts tests/typography-contract.test.ts src/lib/workspace/auth/model.test.ts tests/auth-api.test.ts src/lib/workspace/api/http.test.ts src/lib/workspace/api/workers.test.ts src/lib/workspace/api/ticket-browser.test.ts src/lib/workspace/header/breadcrumb-model.test.ts src/lib/workspace/console/chat-submit.test.ts src/lib/workspace/console/composer-delivery.test.ts test/composer-history.test.ts tests/composer-paste.test.ts src/lib/workspace/console/composer-command.test.ts src/lib/workspace/console/composer-draft.test.ts src/lib/workspace/console/composer-completion.test.ts test/console/ansi.test.ts src/lib/workspace/console/model.test.ts src/lib/workspace/console/run-status.test.ts src/lib/workspace/companion/api.test.ts tests/workdir-api.test.ts src/lib/workspace/console/tasks.test.ts test/ticket-detail-route-reuse.test.ts test/spa-shell.test.ts test/objective-pages.test.ts test/repositories/ui.test.ts src/lib/workspace/console/worker-console.ui.test.ts src/lib/workspace/settings/model.test.ts tests/runtime-connection.test.ts tests/runtime-management.test.ts tests/runtime-management-source.test.ts src/lib/workspace/sidebar/override-stack.test.ts src/lib/workspace/sidebar/route-ownership.test.ts src/lib/workspace/sidebar/workers.test.ts src/lib/workspace/sidebar/workspace-switcher.test.ts src/lib/workspace/sidebar/worker-subscription.test.ts src/lib/workspace/sidebar/worker-workdir-meta.test.ts src/lib/workspace/sidebar/worker-launch.test.ts test/sidebar/worker-actions.test.ts src/lib/workspace/tickets/merge-request-resources.test.ts src/lib/workspace/tickets/ticket-panel.test.ts test/merge-request-status.test.ts test/config-source/decodal-grammar.test.ts test/config-source/editor-state.test.ts test/config-source/fixed-schema-wrapper.test.ts test/config-source/toolchain.test.ts test/config-source/wasm-parity.test.ts test/config-source/api.test.ts tests/memory-api.test.ts test/repository-access/api.test.ts test/repository-access/loader.test.ts test/repository-access/model.test.ts test/repository-access/structure.test.ts test/repository-ssh-connection-ui.test.ts",
|
||||
"test": "deno test --allow-read=src,test,tests,../../docs --allow-env=LOG,VSCODE_TEXTMATE_DEBUG,NODE_ENV tests/workspace-model.test.ts tests/repository-api.test.ts tests/workspace-catalog.test.ts tests/profile-api.test.ts tests/skill-api.test.ts tests/tooltip-contract.test.ts tests/bevel-contract.test.ts tests/bevel-line-contract.test.ts tests/typography-contract.test.ts src/lib/workspace/auth/model.test.ts src/lib/workspace/protocol-validator.test.ts tests/auth-api.test.ts src/lib/workspace/api/http.test.ts src/lib/workspace/api/workers.test.ts src/lib/workspace/api/ticket-browser.test.ts src/lib/workspace/header/breadcrumb-model.test.ts src/lib/workspace/console/chat-submit.test.ts src/lib/workspace/console/composer-delivery.test.ts test/composer-history.test.ts tests/composer-paste.test.ts src/lib/workspace/console/composer-command.test.ts src/lib/workspace/console/composer-draft.test.ts src/lib/workspace/console/composer-completion.test.ts test/console/ansi.test.ts src/lib/workspace/console/model.test.ts src/lib/workspace/console/run-status.test.ts src/lib/workspace/companion/api.test.ts tests/workdir-api.test.ts src/lib/workspace/console/tasks.test.ts test/ticket-detail-route-reuse.test.ts test/spa-shell.test.ts test/objective-pages.test.ts test/repositories/ui.test.ts src/lib/workspace/console/worker-console.ui.test.ts src/lib/workspace/settings/model.test.ts tests/runtime-connection.test.ts tests/runtime-management.test.ts tests/runtime-management-source.test.ts src/lib/workspace/sidebar/override-stack.test.ts src/lib/workspace/sidebar/route-ownership.test.ts src/lib/workspace/sidebar/workers.test.ts src/lib/workspace/sidebar/workspace-switcher.test.ts src/lib/workspace/sidebar/worker-subscription.test.ts src/lib/workspace/sidebar/worker-workdir-meta.test.ts src/lib/workspace/sidebar/worker-launch.test.ts test/sidebar/worker-actions.test.ts src/lib/workspace/tickets/merge-request-resources.test.ts src/lib/workspace/tickets/ticket-panel.test.ts test/merge-request-status.test.ts test/config-source/decodal-grammar.test.ts test/config-source/editor-state.test.ts test/config-source/fixed-schema-wrapper.test.ts test/config-source/toolchain.test.ts test/config-source/wasm-parity.test.ts test/config-source/api.test.ts tests/memory-api.test.ts test/repository-access/api.test.ts test/repository-access/loader.test.ts test/repository-access/model.test.ts test/repository-access/structure.test.ts test/repository-ssh-connection-ui.test.ts",
|
||||
"test:component": "deno run -A npm:vitest@3.2.4 run --config vitest.config.ts",
|
||||
"build": "deno run -A npm:vite@7.2.7 build",
|
||||
"preview": "deno run -A npm:vite@7.2.7 preview",
|
||||
|
||||
@@ -15,6 +15,21 @@ cargo run -q -p server-api --example generate_repository_openapi_types -- --chec
|
||||
|
||||
The generator fails closed on unsupported or lossy OpenAPI constructs, including ambiguous nullable unions, external references, semantic `$ref` siblings, unimplemented ordinary schema keywords, unsupported enum literals/formats, and integer ranges that cannot be represented safely by JavaScript numbers. Its one explicit format mapping is canonical `uint64` to a nonnegative JavaScript safe `number`; the runtime parser enforces the same range. The legacy TypeScript generator imports the two shared Repository source projection types it still needs and no longer declares the migrated Repository list/detail types.
|
||||
|
||||
## Workspace subscription / Worker protocol
|
||||
|
||||
`protocol.ts` and `protocol-validator.ts` are generated together from the Serde/`schemars` closure of `protocol::subscription::SubscriptionFrame`. Rust DTOs and validation constants in `crates/protocol` remain the sole wire authority. The validator closes every generated object shape, rejects unknown tagged variants and unsafe integers, applies Rust-owned aggregate bounds, and mirrors the Browser-relevant `SubscriptionFrame::validate()` selector/payload checks. The Workspace multiplexer additionally checks live event payloads against its locally retained selector.
|
||||
|
||||
Inbound Browser frames only are validated: Browser-to-Server methods remain compile-time typed and are decoded plus semantically validated by the Rust server. Invalid inbound frames close the shared socket with a fixed, non-reflective diagnostic; the existing reconnect path clears routing state and requests fresh snapshots. A valid `subscription_closed` frame continues to isolate and resubscribe only that subscription.
|
||||
|
||||
Regenerate or verify both artifacts from the repository root:
|
||||
|
||||
```sh
|
||||
cargo run -q -p protocol --example generate_typescript --features typescript
|
||||
cargo run -q -p protocol --example generate_typescript --features typescript -- --check
|
||||
```
|
||||
|
||||
The `protocol` crate tests also compare both checked-in files byte-for-byte with fresh generator output.
|
||||
|
||||
## Other generated contracts
|
||||
|
||||
The remaining files retain their generator command in their header. `legacy-server-api.ts` is transitional: new bounded domain slices should use a focused contract artifact instead of adding another API surface to it.
|
||||
|
||||
File diff suppressed because one or more lines are too long
@@ -0,0 +1,185 @@
|
||||
// @vitest-environment happy-dom
|
||||
|
||||
import { afterEach, beforeEach, expect, test, vi } from 'vitest';
|
||||
import { WorkspaceMultiplexer } from './multiplexer';
|
||||
|
||||
type SocketEvent = 'open' | 'message' | 'error' | 'close';
|
||||
type SocketListener = (event: { data?: unknown }) => void;
|
||||
|
||||
class FakeWebSocket {
|
||||
static readonly CONNECTING = 0;
|
||||
static readonly OPEN = 1;
|
||||
static readonly CLOSED = 3;
|
||||
static instances: FakeWebSocket[] = [];
|
||||
|
||||
readyState = FakeWebSocket.CONNECTING;
|
||||
readonly sent: string[] = [];
|
||||
readonly closes: Array<[number | undefined, string | undefined]> = [];
|
||||
readonly #listeners = new Map<SocketEvent, SocketListener[]>();
|
||||
|
||||
constructor(readonly url: URL) {
|
||||
FakeWebSocket.instances.push(this);
|
||||
}
|
||||
|
||||
addEventListener(kind: SocketEvent, listener: SocketListener): void {
|
||||
const listeners = this.#listeners.get(kind) ?? [];
|
||||
listeners.push(listener);
|
||||
this.#listeners.set(kind, listeners);
|
||||
}
|
||||
|
||||
send(value: string): void {
|
||||
this.sent.push(value);
|
||||
}
|
||||
|
||||
close(code?: number, reason?: string): void {
|
||||
this.closes.push([code, reason]);
|
||||
if (this.readyState === FakeWebSocket.CLOSED) return;
|
||||
this.readyState = FakeWebSocket.CLOSED;
|
||||
this.#emit('close');
|
||||
}
|
||||
|
||||
open(): void {
|
||||
this.readyState = FakeWebSocket.OPEN;
|
||||
this.#emit('open');
|
||||
}
|
||||
|
||||
message(value: unknown): void {
|
||||
this.#emit('message', { data: value });
|
||||
}
|
||||
|
||||
#emit(kind: SocketEvent, event: { data?: unknown } = {}): void {
|
||||
for (const listener of this.#listeners.get(kind) ?? []) listener(event);
|
||||
}
|
||||
}
|
||||
|
||||
beforeEach(() => {
|
||||
FakeWebSocket.instances = [];
|
||||
vi.stubGlobal('WebSocket', FakeWebSocket);
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
vi.useRealTimers();
|
||||
vi.unstubAllGlobals();
|
||||
});
|
||||
|
||||
test('invalid inbound frame closes before dispatch and reconnects for a fresh snapshot', () => {
|
||||
vi.useFakeTimers();
|
||||
const multiplexer = new WorkspaceMultiplexer('workspace-validation-test');
|
||||
const onFrame = vi.fn();
|
||||
const onStatus = vi.fn();
|
||||
multiplexer.subscribe({ topic: 'workspace_workers' }, { onFrame, onStatus });
|
||||
|
||||
const first = FakeWebSocket.instances[0];
|
||||
expect(first).toBeDefined();
|
||||
first.open();
|
||||
expect(first.sent).toHaveLength(1);
|
||||
|
||||
first.message(
|
||||
JSON.stringify({
|
||||
protocol_version: 1,
|
||||
frame: 'event',
|
||||
message: {
|
||||
event: 'subscription_closed',
|
||||
data: {
|
||||
subscription_id: 'subscription-1',
|
||||
code: 'lagged',
|
||||
message: 'untrusted detail',
|
||||
newer_field: true,
|
||||
},
|
||||
},
|
||||
}),
|
||||
);
|
||||
|
||||
expect(onFrame).not.toHaveBeenCalled();
|
||||
expect(first.closes).toEqual([[1002, 'Invalid workspace protocol frame']]);
|
||||
expect(onStatus).toHaveBeenLastCalledWith('closed', 'Workspace protocol frame rejected');
|
||||
expect(JSON.stringify(onStatus.mock.calls)).not.toContain('untrusted detail');
|
||||
|
||||
vi.advanceTimersByTime(500);
|
||||
const second = FakeWebSocket.instances[1];
|
||||
expect(second).toBeDefined();
|
||||
second.open();
|
||||
expect(second.sent).toHaveLength(1);
|
||||
expect(JSON.parse(second.sent[0]).message.method).toBe('subscribe_events');
|
||||
multiplexer.dispose();
|
||||
});
|
||||
|
||||
test('valid subscription closure remains isolated and resubscribes only its selector', () => {
|
||||
const multiplexer = new WorkspaceMultiplexer('workspace-isolation-test');
|
||||
const workersFrame = vi.fn();
|
||||
const workersStatus = vi.fn();
|
||||
const workdirsFrame = vi.fn();
|
||||
const workdirsStatus = vi.fn();
|
||||
multiplexer.subscribe(
|
||||
{ topic: 'workspace_workers' },
|
||||
{ onFrame: workersFrame, onStatus: workersStatus },
|
||||
);
|
||||
multiplexer.subscribe(
|
||||
{ topic: 'workspace_workdirs' },
|
||||
{ onFrame: workdirsFrame, onStatus: workdirsStatus },
|
||||
);
|
||||
|
||||
const socket = FakeWebSocket.instances[0];
|
||||
socket.open();
|
||||
const requests = socket.sent.map((value) => JSON.parse(value));
|
||||
expect(requests).toHaveLength(2);
|
||||
for (const [index, request] of requests.entries()) {
|
||||
if (request.frame !== 'request' || request.message.method !== 'subscribe_events') {
|
||||
throw new Error('expected subscribe request');
|
||||
}
|
||||
const selector = request.message.params.selector;
|
||||
const snapshot =
|
||||
selector.topic === 'workspace_workers'
|
||||
? { topic: 'workers', data: { workers: [] } }
|
||||
: { topic: 'workspace_workdirs', data: { workdirs: [] } };
|
||||
socket.message(
|
||||
JSON.stringify({
|
||||
protocol_version: 1,
|
||||
frame: 'response',
|
||||
message: {
|
||||
result: 'subscribed',
|
||||
payload: {
|
||||
request_id: request.message.params.request_id,
|
||||
subscription_id: `subscription-${index + 1}`,
|
||||
selector,
|
||||
snapshot_revision: 1,
|
||||
snapshot,
|
||||
},
|
||||
},
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
const sentBeforeClosure = socket.sent.length;
|
||||
socket.message(
|
||||
JSON.stringify({
|
||||
protocol_version: 1,
|
||||
frame: 'event',
|
||||
message: {
|
||||
event: 'subscription_closed',
|
||||
data: {
|
||||
subscription_id: 'subscription-1',
|
||||
code: 'lagged',
|
||||
message: 'resubscribe for a fresh snapshot',
|
||||
},
|
||||
},
|
||||
}),
|
||||
);
|
||||
|
||||
expect(workersFrame).toHaveBeenCalledTimes(2);
|
||||
expect(workersStatus).toHaveBeenLastCalledWith(
|
||||
'connecting',
|
||||
'resubscribe for a fresh snapshot',
|
||||
);
|
||||
expect(workdirsFrame).toHaveBeenCalledTimes(1);
|
||||
expect(workdirsStatus).not.toHaveBeenCalledWith(
|
||||
'connecting',
|
||||
'resubscribe for a fresh snapshot',
|
||||
);
|
||||
expect(socket.sent).toHaveLength(sentBeforeClosure + 1);
|
||||
expect(JSON.parse(socket.sent.at(-1)!).message.params.selector).toEqual({
|
||||
topic: 'workspace_workers',
|
||||
});
|
||||
expect(socket.closes).toEqual([]);
|
||||
multiplexer.dispose();
|
||||
});
|
||||
@@ -5,6 +5,10 @@ import type {
|
||||
SubscriptionFrame,
|
||||
SubscriptionId,
|
||||
} from '$lib/generated/protocol';
|
||||
import {
|
||||
decodeSubscriptionFrame,
|
||||
subscriptionFrameMatchesSelector,
|
||||
} from '$lib/generated/protocol-validator';
|
||||
import { workspaceApiPath } from '$lib/workspace/api/http';
|
||||
|
||||
type Listener = {
|
||||
@@ -56,6 +60,7 @@ export class WorkspaceMultiplexer {
|
||||
readonly #runtimeSubscriptions = new Map<string, string>();
|
||||
#socket: WebSocket | null = null;
|
||||
#reconnectTimer: ReturnType<typeof setTimeout> | null = null;
|
||||
#disconnectMessage: string | null = null;
|
||||
#closed = false;
|
||||
|
||||
constructor(workspaceId: string) {
|
||||
@@ -106,17 +111,20 @@ export class WorkspaceMultiplexer {
|
||||
this.#sendSubscribe(subscription);
|
||||
}
|
||||
});
|
||||
socket.addEventListener('message', (event) => this.#receive(String(event.data)));
|
||||
socket.addEventListener('message', (event) => this.#receive(event.data));
|
||||
socket.addEventListener('error', () => socket.close());
|
||||
socket.addEventListener('close', () => {
|
||||
if (this.#socket !== socket) return;
|
||||
this.#socket = null;
|
||||
const disconnectMessage =
|
||||
this.#disconnectMessage ?? 'Workspace subscription disconnected';
|
||||
this.#disconnectMessage = null;
|
||||
this.#requests.clear();
|
||||
this.#runtimeSubscriptions.clear();
|
||||
for (const subscription of this.#subscriptions.values()) {
|
||||
subscription.requestId = null;
|
||||
subscription.subscriptionId = null;
|
||||
subscription.listener.onStatus?.('closed', 'Workspace subscription disconnected');
|
||||
subscription.listener.onStatus?.('closed', disconnectMessage);
|
||||
}
|
||||
if (!this.#closed && this.#subscriptions.size > 0) {
|
||||
this.#reconnectTimer = setTimeout(() => this.#ensureConnected(), 500);
|
||||
@@ -138,25 +146,23 @@ export class WorkspaceMultiplexer {
|
||||
});
|
||||
}
|
||||
|
||||
#receive(text: string): void {
|
||||
let frame: SubscriptionFrame;
|
||||
try {
|
||||
frame = JSON.parse(text) as SubscriptionFrame;
|
||||
} catch {
|
||||
this.#socket?.close();
|
||||
return;
|
||||
}
|
||||
if (frame.protocol_version !== 1) {
|
||||
this.#socket?.close();
|
||||
#receive(input: unknown): void {
|
||||
const decoded = decodeSubscriptionFrame(input);
|
||||
if (!decoded.ok) {
|
||||
this.#rejectInboundFrame();
|
||||
return;
|
||||
}
|
||||
const frame = decoded.value;
|
||||
if (frame.frame === 'response' && frame.message.result === 'subscribed') {
|
||||
const clientId = this.#requests.get(frame.message.payload.request_id);
|
||||
const subscription = clientId ? this.#subscriptions.get(clientId) : undefined;
|
||||
if (!clientId || !subscription) return;
|
||||
if (!subscriptionFrameMatchesSelector(frame, subscription.selector)) {
|
||||
this.#rejectInboundFrame();
|
||||
return;
|
||||
}
|
||||
this.#requests.delete(frame.message.payload.request_id);
|
||||
const subscriptionId = frame.message.payload.subscription_id;
|
||||
if (!subscriptionId) return;
|
||||
subscription.subscriptionId = subscriptionId;
|
||||
this.#runtimeSubscriptions.set(subscriptionId, clientId);
|
||||
subscription.listener.onFrame(frame);
|
||||
@@ -175,6 +181,13 @@ export class WorkspaceMultiplexer {
|
||||
if (frame.frame === 'event') {
|
||||
const clientId = this.#runtimeSubscriptions.get(frame.message.data.subscription_id);
|
||||
const subscription = clientId ? this.#subscriptions.get(clientId) : undefined;
|
||||
if (
|
||||
subscription &&
|
||||
!subscriptionFrameMatchesSelector(frame, subscription.selector)
|
||||
) {
|
||||
this.#rejectInboundFrame();
|
||||
return;
|
||||
}
|
||||
subscription?.listener.onFrame(frame);
|
||||
if (
|
||||
frame.message.event === 'subscription_closed' &&
|
||||
@@ -190,6 +203,11 @@ export class WorkspaceMultiplexer {
|
||||
}
|
||||
}
|
||||
|
||||
#rejectInboundFrame(): void {
|
||||
this.#disconnectMessage = 'Workspace protocol frame rejected';
|
||||
this.#socket?.close(1002, 'Invalid workspace protocol frame');
|
||||
}
|
||||
|
||||
#sendWorkerMethod(clientId: string, method: Method): void {
|
||||
const subscription = this.#subscriptions.get(clientId);
|
||||
if (!subscription?.subscriptionId) throw new Error('Worker protocol subscription is not open');
|
||||
|
||||
@@ -0,0 +1,190 @@
|
||||
import type { EventSubscriptionSelector, SubscriptionFrame } from '$lib/generated/protocol';
|
||||
import {
|
||||
decodeSubscriptionFrame,
|
||||
subscriptionFrameMatchesSelector,
|
||||
} from '$lib/generated/protocol-validator';
|
||||
|
||||
declare const Deno: {
|
||||
test(name: string, fn: () => void | Promise<void>): void;
|
||||
};
|
||||
|
||||
function assert(condition: unknown, message: string): asserts condition {
|
||||
if (!condition) throw new Error(message);
|
||||
}
|
||||
|
||||
function subscribedFrame(overrides: Record<string, unknown> = {}): Record<string, unknown> {
|
||||
return {
|
||||
protocol_version: 1,
|
||||
frame: 'response',
|
||||
message: {
|
||||
result: 'subscribed',
|
||||
payload: {
|
||||
request_id: 'request-1',
|
||||
subscription_id: 'subscription-1',
|
||||
selector: { topic: 'workspace_workers' },
|
||||
snapshot_revision: 1,
|
||||
snapshot: { topic: 'workers', data: { workers: [] } },
|
||||
...overrides,
|
||||
},
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function decode(value: unknown) {
|
||||
return decodeSubscriptionFrame(JSON.stringify(value));
|
||||
}
|
||||
|
||||
Deno.test('generated subscription validator accepts a current exact frame', () => {
|
||||
const result = decode(subscribedFrame());
|
||||
assert(result.ok, `expected valid frame, got ${JSON.stringify(result)}`);
|
||||
assert(result.value.frame === 'response', 'expected response frame');
|
||||
});
|
||||
|
||||
Deno.test('generated subscription validator rejects malformed and newer shapes', () => {
|
||||
const cases: Array<[string, unknown]> = [
|
||||
['malformed json', '{'],
|
||||
['invalid discriminator', { ...subscribedFrame(), frame: 'future' }],
|
||||
['missing required field', { frame: 'response', message: subscribedFrame().message }],
|
||||
['unknown root field', { ...subscribedFrame(), secret: 'must-not-reflect' }],
|
||||
[
|
||||
'unknown nested field',
|
||||
subscribedFrame({ snapshot: { topic: 'workers', data: { workers: [], future: true } } }),
|
||||
],
|
||||
['wrong field type', subscribedFrame({ snapshot_revision: '1' })],
|
||||
['unsafe integer', subscribedFrame({ snapshot_revision: Number.MAX_SAFE_INTEGER + 1 })],
|
||||
['unsupported version', { ...subscribedFrame(), protocol_version: 2 }],
|
||||
[
|
||||
'newer Worker event',
|
||||
{
|
||||
protocol_version: 1,
|
||||
frame: 'event',
|
||||
message: {
|
||||
event: 'event',
|
||||
data: {
|
||||
subscription_id: 'subscription-1',
|
||||
subject_revision: 2,
|
||||
payload: {
|
||||
event: 'worker_protocol',
|
||||
data: { worker_id: 'worker-1', event: { event: 'future_event' } },
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
],
|
||||
];
|
||||
|
||||
for (const [name, value] of cases) {
|
||||
const result = typeof value === 'string' ? decodeSubscriptionFrame(value) : decode(value);
|
||||
assert(!result.ok, `${name} unexpectedly passed`);
|
||||
assert(!JSON.stringify(result).includes('must-not-reflect'), `${name} reflected frame content`);
|
||||
}
|
||||
});
|
||||
|
||||
Deno.test('generated subscription validator enforces Rust semantic bounds and selector pairing', () => {
|
||||
const oversizedIds = Array.from({ length: 257 }, (_, index) => `worker-${index}`);
|
||||
const cases: Array<[string, unknown]> = [
|
||||
[
|
||||
'oversized correlation id',
|
||||
subscribedFrame({ subscription_id: 'x'.repeat(129) }),
|
||||
],
|
||||
[
|
||||
'oversized rejection message',
|
||||
{
|
||||
protocol_version: 1,
|
||||
frame: 'response',
|
||||
message: {
|
||||
result: 'subscription_rejected',
|
||||
payload: {
|
||||
request_id: 'request-1',
|
||||
code: 'invalid_request',
|
||||
message: 'x'.repeat(1025),
|
||||
},
|
||||
},
|
||||
},
|
||||
],
|
||||
[
|
||||
'oversized selector collection',
|
||||
subscribedFrame({
|
||||
selector: { topic: 'worker_lifecycle', worker_ids: oversizedIds },
|
||||
}),
|
||||
],
|
||||
[
|
||||
'selector snapshot mismatch',
|
||||
subscribedFrame({
|
||||
selector: { topic: 'worker_protocol', worker_id: 'worker-1' },
|
||||
}),
|
||||
],
|
||||
[
|
||||
'duplicate snapshot worker',
|
||||
subscribedFrame({
|
||||
snapshot: {
|
||||
topic: 'workers',
|
||||
data: {
|
||||
workers: [
|
||||
{
|
||||
worker_id: 'worker-1',
|
||||
runtime_id: 'runtime-1',
|
||||
availability: 'observed',
|
||||
subject_revision: 1,
|
||||
state: 'idle',
|
||||
has_running_internal_workers: false,
|
||||
},
|
||||
{
|
||||
worker_id: 'worker-1',
|
||||
runtime_id: 'runtime-1',
|
||||
availability: 'observed',
|
||||
subject_revision: 2,
|
||||
state: 'running',
|
||||
has_running_internal_workers: false,
|
||||
},
|
||||
],
|
||||
},
|
||||
},
|
||||
}),
|
||||
],
|
||||
[
|
||||
'server-to-Browser direction mismatch',
|
||||
{
|
||||
protocol_version: 1,
|
||||
frame: 'request',
|
||||
message: {
|
||||
method: 'unsubscribe_events',
|
||||
params: { request_id: 'request-1', subscription_id: 'subscription-1' },
|
||||
},
|
||||
},
|
||||
],
|
||||
];
|
||||
|
||||
for (const [name, value] of cases) {
|
||||
const result = decode(value);
|
||||
assert(!result.ok, `${name} unexpectedly passed`);
|
||||
}
|
||||
});
|
||||
|
||||
Deno.test('stateful selector validation fences a routed event before projection', () => {
|
||||
const frame: SubscriptionFrame = {
|
||||
protocol_version: 1,
|
||||
frame: 'event',
|
||||
message: {
|
||||
event: 'event',
|
||||
data: {
|
||||
subscription_id: 'subscription-1',
|
||||
subject_revision: 2,
|
||||
payload: {
|
||||
event: 'worker_protocol',
|
||||
data: { worker_id: 'worker-2', event: { event: 'thinking_start' } },
|
||||
},
|
||||
},
|
||||
},
|
||||
};
|
||||
const decoded = decode(frame);
|
||||
assert(decoded.ok, `expected structurally valid event, got ${JSON.stringify(decoded)}`);
|
||||
const selector: EventSubscriptionSelector = {
|
||||
topic: 'worker_protocol',
|
||||
worker_id: 'worker-1',
|
||||
};
|
||||
assert(
|
||||
!subscriptionFrameMatchesSelector(decoded.value, selector),
|
||||
'mismatched Worker event crossed its subscription selector',
|
||||
);
|
||||
});
|
||||
Reference in New Issue
Block a user