From a44ced5b96899f5bfa81e11719e58db4c401dc29 Mon Sep 17 00:00:00 2001 From: imccyu Date: Tue, 22 Sep 2026 19:06:54 +0800 Subject: [PATCH] fix(web): gate Remote WebSockets on application readiness --- apps/web/tests/fixtures/restart-startup.js | 30 +++++++++ .../fixtures/restart-startup.overlay.yml | 5 ++ apps/web/tests/server-restart.e2e.ts | 51 ++++++++++++--- docs/config-catalog.i18n.yaml | 4 +- docs/config-catalog.md | 2 +- docs/config-catalog.zh.md | 2 +- packages/api/gateway/README.i18n.yaml | 4 +- packages/api/gateway/README.md | 2 + packages/api/gateway/README.zh.md | 2 + packages/api/gateway/package.json | 1 + packages/api/gateway/src/index.ts | 62 ++++++++++++------- .../gateway/tests/gateway-stream.host.spec.ts | 62 +++++++++++++++++++ packages/api/gateway/tsconfig.host.json | 3 + pnpm-lock.yaml | 3 + 14 files changed, 196 insertions(+), 37 deletions(-) create mode 100644 apps/web/tests/fixtures/restart-startup.js create mode 100644 apps/web/tests/fixtures/restart-startup.overlay.yml diff --git a/apps/web/tests/fixtures/restart-startup.js b/apps/web/tests/fixtures/restart-startup.js new file mode 100644 index 0000000000..2967fcee10 --- /dev/null +++ b/apps/web/tests/fixtures/restart-startup.js @@ -0,0 +1,30 @@ +/** Hold one Host service until the restart test releases application startup. */ +export const name = 'web-restart-startup' + +/** Keep the carrier services available while Session Controller startup is held. */ +export const inject = ['typertGateway', 'connection', 'webServer'] + +/** + * Provide the test-only dependency after startup has been released. + * @param {import('@deepseek-ai/cordis').Context} ctx - Host plugin context. + * @returns {Promise} Once the barrier is released or the plugin is disposed. + */ +export async function apply(ctx) { + if (process.env.DSH_WEB_RESTART_HOLD_STARTUP === '1') { + if (typeof process.send !== 'function') throw new Error('web restart startup fixture requires an IPC channel') + const released = Promise.withResolvers() + const resume = (message) => { + if (message === 'resume-startup') released.resolve(true) + } + ctx.effect(() => { + process.on('message', resume) + return () => { + process.off('message', resume) + released.resolve(false) + } + }, 'web-restart: startup barrier') + process.send('startup-blocked') + if (!await released.promise) return + } + ctx.reflect.provide('webRestartReady', {}) +} diff --git a/apps/web/tests/fixtures/restart-startup.overlay.yml b/apps/web/tests/fixtures/restart-startup.overlay.yml new file mode 100644 index 0000000000..3f7b9bc45c --- /dev/null +++ b/apps/web/tests/fixtures/restart-startup.overlay.yml @@ -0,0 +1,5 @@ +- id: session-controller + inject: [webRestartReady] +- insert: + - id: web-restart-startup + name: ./restart-startup.js diff --git a/apps/web/tests/server-restart.e2e.ts b/apps/web/tests/server-restart.e2e.ts index c3d5565e40..736165c2dc 100644 --- a/apps/web/tests/server-restart.e2e.ts +++ b/apps/web/tests/server-restart.e2e.ts @@ -14,16 +14,18 @@ class RestartableServer { private child: ChildProcess | undefined private closed: Promise | undefined private output = '' + readonly startupBlocked = Promise.withResolvers() constructor(private readonly world: string, private readonly modelUrl: string) {} - async start(port: number): Promise { + async start(port: number, holdStartup = false): Promise { if (this.child !== undefined) throw new Error('Server is already running') this.output = '' const ready = Promise.withResolvers() const child = spawn(process.execPath, [ join(REPO_ROOT, 'apps/cli/lib/bin.js'), '--profile', 'web', '--patch', fileURLToPath(new URL('./pin-browse-picker.overlay.yml', import.meta.url)), + '--patch', fileURLToPath(new URL('./fixtures/restart-startup.overlay.yml', import.meta.url)), '--no-open', '--port', String(port), ], { cwd: this.world, @@ -32,8 +34,9 @@ class RestartableServer { DSH_HOME: join(this.world, 'home'), DSH_AGENTS_HOME: join(this.world, 'agents'), DSH_TELEMETRY_DISABLED: '1', DEEPSEEK_API_KEY: 'keyless-server-restart-fixture', DEEPSEEK_BASE_URL: this.modelUrl, + DSH_WEB_RESTART_HOLD_STARTUP: holdStartup ? '1' : '0', }, - stdio: ['ignore', 'pipe', 'pipe'], + stdio: ['ignore', 'pipe', 'pipe', 'ipc'], }) this.child = child const receive = (data: Buffer): void => { @@ -43,6 +46,9 @@ class RestartableServer { } child.stdout?.on('data', receive) child.stderr?.on('data', receive) + child.on('message', (message) => { + if (message === 'startup-blocked') this.startupBlocked.resolve(undefined) + }) child.once('error', (error) => { ready.reject(error) }) this.closed = new Promise((resolve) => { child.once('close', (code, signal) => { @@ -58,6 +64,7 @@ class RestartableServer { async stop(): Promise { const child = this.child if (child === undefined) return + this.resumeStartup() let forced = false const timer = setTimeout(() => { forced = true; child.kill('SIGKILL') }, 15_000) try { @@ -70,10 +77,14 @@ class RestartableServer { if (forced) throw new Error('Server did not stop after SIGTERM\n' + this.logs()) } + resumeStartup(): void { + if (this.child?.connected) this.child.send('resume-startup') + } + logs(): string { return this.output.replace(/([?&]token=)[A-Za-z0-9_-]+/gu, '$1') } } -it('keeps an opened Session and draft when the server restarts on the same port', async () => { +it.each([false, true])('keeps the same revision, Session and page across a server restart (delayed startup: %s)', async (holdStartup) => { const world = await mkdtemp(join(tmpdir(), 'dsh-server-restart-')) onTestFinished(() => rm(world, { recursive: true, force: true })) const model = createServer((request, response) => { @@ -126,6 +137,7 @@ it('keeps an opened Session and draft when the server restarts on the same port' onTestFinished(() => browser.close()) const page = await newEnglishPage(browser) const errors: string[] = [] + const streamErrors: unknown[] = [] const graphs: { rev: string; entries: { id: string; rev: string }[] }[] = [] let readyFrames = 0 let sessionBaselines = 0 @@ -140,7 +152,8 @@ it('keeps an opened Session and draft when the server restarts on the same port' }) socket.on('framereceived', ({ payload }) => { if (typeof payload !== 'string') return - const message = JSON.parse(payload) as { type?: string; streamId: string; value?: { type?: string } } + const message = JSON.parse(payload) as { type?: string; streamId: string; value?: { type?: string }; error?: unknown } + if (message.type === 'error') streamErrors.push(message.error) if (message.type === 'item' && message.value?.type === 'ready') readyFrames++ if (message.type === 'item' && endpoints.get(message.streamId) === 'session/follow' && message.value?.type === 'snapshot') sessionBaselines++ @@ -154,7 +167,7 @@ it('keeps an opened Session and draft when the server restarts on the same port' }) onTestFailed(async () => { await saveFailureShot(page, 'web-server-restart-' + scenarioId) - await writeFile(join(REPO_ROOT, '.artifacts', 'server-restart-' + scenarioId + '.json'), JSON.stringify({ errors, graphs, server: server.logs() }, null, 2)) + await writeFile(join(REPO_ROOT, '.artifacts', 'server-restart-' + scenarioId + '.json'), JSON.stringify({ errors, streamErrors, graphs, server: server.logs() }, null, 2)) }) await page.goto(url, { waitUntil: 'load' }) await page.locator('[data-slot="root"]').waitFor({ state: 'attached', timeout: 20_000 }) @@ -177,7 +190,29 @@ it('keeps an opened Session and draft when the server restarts on the same port' page.on('framenavigated', () => { navigations++ }) await server.stop() - const restarted = await server.start(port) + let restarted: string + const restarting = server.start(port, holdStartup) + try { + if (holdStartup) { + await Promise.race([ + server.startupBlocked.promise, + restarting.then(() => { throw new Error('the restarted Host did not hold controller startup') }), + ]) + const opened = await page.evaluate(() => new Promise((resolve) => { + const endpoint = new URL('api/remote.mux', location.href) + endpoint.protocol = 'ws:' + const socket = new WebSocket(endpoint) + let accepted = false + socket.addEventListener('open', () => { accepted = true; socket.close() }) + socket.addEventListener('close', () => { resolve(accepted) }) + })) + expect(opened).toBe(false) + expect(readyFrames).toBe(beforeReady) + } + } finally { + server.resumeStartup() + restarted = await restarting + } expect(new URL(restarted).origin).toBe(new URL(url).origin) await expect.poll(() => readyFrames, { timeout: 30_000 }).toBeGreaterThan(beforeReady) await expect.poll(() => sessionBaselines, { timeout: 30_000 }).toBeGreaterThan(beforeBaseline) @@ -187,7 +222,8 @@ it('keeps an opened Session and draft when the server restarts on the same port' })) expect(errors).toEqual([]) - expect(graphs.at(-1)).toEqual(graphs[beforeGraphs - 1]) + expect(streamErrors).toEqual([]) + for (const graph of graphs.slice(beforeGraphs)) expect(graph).toEqual(graphs[beforeGraphs - 1]) expect(await originalRoot!.evaluate(element => element.isConnected)).toBe(true) expect(await composer.textContent()).toBe(draft) expect(await response.isVisible()).toBe(true) @@ -195,5 +231,6 @@ it('keeps an opened Session and draft when the server restarts on the same port' await composer.press('Enter') await expect.poll(() => response.count(), { timeout: 20_000 }).toBe(2) expect(errors).toEqual([]) + expect(streamErrors).toEqual([]) expect(navigations).toBe(0) }) diff --git a/docs/config-catalog.i18n.yaml b/docs/config-catalog.i18n.yaml index 16c39be505..106db59c55 100644 --- a/docs/config-catalog.i18n.yaml +++ b/docs/config-catalog.i18n.yaml @@ -2,5 +2,5 @@ # side as of the last confirmed-consistent state. Both languages carry equal authority; # after editing either side, bring the other along and re-record with: # pnpm run verify-translation-pairing --write docs/config-catalog.md -config-catalog.md: 563c8d21896e93a76394937b55f841036ee74c42 -config-catalog.zh.md: f3505a577660a534e6c5a3d255d4c46a4cfaea05 +config-catalog.md: b9e75f6679d43aef41808a30ed370e26070c1331 +config-catalog.zh.md: d76c88fb806998abee414347822a3d11c3b2b7de diff --git a/docs/config-catalog.md b/docs/config-catalog.md index 563c8d2189..b9e75f6679 100644 --- a/docs/config-catalog.md +++ b/docs/config-catalog.md @@ -194,7 +194,7 @@ export interface Config { } ``` -Source: [`packages/api/gateway/src/index.ts:144`](../packages/api/gateway/src/index.ts) +Source: [`packages/api/gateway/src/index.ts:145`](../packages/api/gateway/src/index.ts) diff --git a/docs/config-catalog.zh.md b/docs/config-catalog.zh.md index f3505a5776..d76c88fb80 100644 --- a/docs/config-catalog.zh.md +++ b/docs/config-catalog.zh.md @@ -196,7 +196,7 @@ export interface Config { } ``` -来源：[`packages/api/gateway/src/index.ts:144`](../packages/api/gateway/src/index.ts) +来源：[`packages/api/gateway/src/index.ts:145`](../packages/api/gateway/src/index.ts) diff --git a/packages/api/gateway/README.i18n.yaml b/packages/api/gateway/README.i18n.yaml index e08b3475e6..7f47a6c9eb 100644 --- a/packages/api/gateway/README.i18n.yaml +++ b/packages/api/gateway/README.i18n.yaml @@ -2,5 +2,5 @@ # side as of the last confirmed-consistent state. Both languages carry equal authority; # after editing either side, bring the other along and re-record with: # pnpm run verify-translation-pairing --write packages/api/gateway/README.md -README.md: b5b64d3fa073c4f992529b63bc2e02117a28cecf -README.zh.md: 87919703fdb2a1f9e32603bf6a1d152d4cc5404e +README.md: 1a746a83eb966fba6ae40b61061d42bf478e1c84 +README.zh.md: 7b37548f6ad894b761457df80278eeed7aa6ca7b diff --git a/packages/api/gateway/README.md b/packages/api/gateway/README.md index b5b64d3fa0..1a746a83eb 100644 --- a/packages/api/gateway/README.md +++ b/packages/api/gateway/README.md @@ -34,6 +34,8 @@ A cancellation-aware Remote method declares `signal: AbortSignal` as its final H A stream Remote uses `@Remote({ mode: 'stream' })` and returns an `Iterable` or `AsyncIterable`. `ctx.typertGateway.stream()` applies the same endpoint, argument, lookup, and cancellation checks as unary invocation, then returns a cancellation-aware iterable over the business items. The Client opens the Gateway-owned `/api/remote.mux` WebSocket when its plugin activates and keeps it connected while idle. Connection owns the retry schedule; before each retry it asks the mux to cancel any candidate or active socket and make exactly one fresh physical attempt. The Host sends Ping control frames at the configured `websocketHeartbeatIntervalMs` interval (two seconds by default), and the browser answers Pong at the WebSocket protocol layer, so idle network intermediaries see traffic without any Remote stream frame. A socket that has not answered the previous Ping is terminated at the next interval. Independently cancellable logical streams share that socket; an in-process Connection carrier provides equivalent streams directly without opening it. +When the launcher supplies `ctx.appReady`, Gateway registers the WebSocket upgrade route only after successful application startup. Earlier connection attempts remain carrier failures handled by Connection's retry policy, so a restarting Host cannot accept streams while its controllers are still initializing. Unloading cancels a pending readiness subscription and closes any registered carrier. Embedded Hosts without `appReady` register immediately and own their startup ordering; in-process invocation is unchanged. + Every stream also carries a Client-to-Host uplink on the same logical stream. A method declares the uplink item type as the second type argument of its return type, `RemoteStream` from [`dsh-typert-protocol`](../../typert/protocol/README.md); `In` defaults to `never`, and a method that returns `Iterable`, `AsyncIterable`, or `RemoteStream` declares that it reads none. The running method reads its call as `this.ctx.invocation`: the Gateway derives the receiver from a Context carrying that `RemoteInvocation` — `request`, `service`, `peer`, `signal`, and `uplink()` — so nothing enters the parameter list, and `ctx.invocation` is `undefined` on a Context no Remote call derived. `uplink()` is available once per call. With a descriptor codec it decodes every item before delivery; without one — an SRC method, or a method whose `In` is `never` — it delivers JSON-safe `unknown` values. The Client sends each item as an `item` frame on the open logical stream, a top-level `undefined` item as an `item` frame without `value`, and its half-close as an `end` frame; the method observes the half-close as the end of the `uplink()` iteration and cancellation through `signal`. A rejected item fails the whole stream with `gateway/input-invalid` (field `uplink`). Uplink items are validated one by one at the Host because they arrive from the browser; downlink items are values the Host method produced and pass through unchecked. When the method finishes its downlink the Gateway returns the uplink iterator and drops unread items; items sent to a method that never takes its uplink wait in the inbox until then. Each logical stream buffers at most `streamInboxBytes` (262144 by default) of uplink frame bytes; overflow fails the stream with `gateway/uplink-overflow`, and an item after `end` fails it with `gateway/protocol`, in both cases without closing the socket. `item`, `end`, and `cancel` frames for a stream id the Host has already finished are dropped; only a duplicate `open` closes the socket. Gateway-owned streams such as `$events` return their uplink as soon as they open, so their `item` frames are dropped instead of buffered. A unary method may also call `uplink()`; its items are readable only while the method runs. Every stream speaks for the operator Peer: a WebSocket's streams for the Peer Connection admitted at upgrade, whose scope disposal closes the socket, and an in-process carrier's for the operator directly. Host composition can register one application event source through `registerRemoteEvents()`. Gateway reserves the internal `$events` logical endpoint for that source, accepts only empty `args`, and aborts streams opened by the registration when the source is withdrawn. API Remotes owns the event selection, argument validation, per-Client queues, and the Host home sent in the opening `{ type: 'ready', clientId, host: { home } }` frame. Its source factory attaches incremental listeners synchronously, so the Client publishes the generation and starts baseline reads only after incremental delivery is ready. diff --git a/packages/api/gateway/README.zh.md b/packages/api/gateway/README.zh.md index 87919703fd..7b37548f6a 100644 --- a/packages/api/gateway/README.zh.md +++ b/packages/api/gateway/README.zh.md @@ -34,6 +34,8 @@ Connection 可用时，Host 入口会在 Connection 共享的 `/api` FetchHandle 流式 Remote 使用 `@Remote({ mode: 'stream' })` 并返回 `Iterable` 或 `AsyncIterable`。`ctx.typertGateway.stream()` 执行与一元调用相同的 endpoint、参数、lookup 和取消校验，再返回可取消的业务项 iterable。Client 插件激活时打开 Gateway 自有的 `/api/remote.mux` WebSocket，并让它在空闲时保持连接。Connection 拥有重试调度；每次 retry 前，它要求 mux 取消候选或活动 socket，并且只做一次全新的物理连接尝试。Host 按配置的 `websocketHeartbeatIntervalMs` 间隔（默认 2 秒）发送 Ping 控制帧，浏览器在 WebSocket 协议层自动回复 Pong，使空闲网络中间层持续看到流量，而不新增 Remote 流帧。若 socket 尚未回复上一次 Ping，Host 会在下一间隔终止它。可独立取消的逻辑流共享这条连接；进程内 Connection 载体直接提供等价的流，不打开该 WebSocket。 +启动器提供 `ctx.appReady` 时，Gateway 仅在应用成功启动后注册 WebSocket 升级路由。此前的连接尝试仍属于载体故障，由 Connection 的重试策略处理，因此重启中的 Host 不会在控制器仍在初始化时接受流。卸载会取消尚未触发的就绪订阅，并关闭已注册的载体。不提供 `appReady` 的嵌入式 Host 会立即注册，并自行负责启动顺序；进程内调用不变。 + 每条流还在同一条逻辑流上携带从 Client 到 Host 的上行。方法把上行项类型声明为返回类型的第二个类型参数——来自 [`dsh-typert-protocol`](../../typert/protocol/README.zh.md) 的 `RemoteStream`；`In` 缺省为 `never`，返回 `Iterable`、`AsyncIterable` 或 `RemoteStream` 的方法即声明自己不读上行。运行中的方法通过 `this.ctx.invocation` 读取本次调用：Gateway 从携带该 `RemoteInvocation`（`request`、`service`、`peer`、`signal` 与 `uplink()`）的 Context 派生接收者，因此没有任何东西进入参数列表，而非 Remote 调用派生的 Context 上 `ctx.invocation` 为 `undefined`。`uplink()` 每次调用只能取一次。描述符带 codec 时它在交付前逐项解码；不带时——SRC 方法，或 `In` 为 `never` 的方法——它交付经 JSON 安全校验的 `unknown` 值。Client 在已打开的逻辑流上把每一项作为 `item` 帧发送（顶层 `undefined` 项是不带 `value` 的 `item` 帧），以 `end` 帧表示半关闭；方法通过 `uplink()` 迭代结束感知半关闭，通过 `signal` 感知取消。被拒绝的项使整条流以 `gateway/input-invalid`（字段 `uplink`）失败。上行项在 Host 逐项校验，因为它们来自浏览器；下行项是 Host 方法产出的值，不经校验原样透传。方法结束下行时 Gateway 调用 uplink 迭代器的 return 并丢弃未读的项；发给从未取用 uplink 的方法的项在 inbox 里等到那时为止。每条逻辑流最多缓冲 `streamInboxBytes`（默认 262144）字节的上行帧；超限使流以 `gateway/uplink-overflow` 失败，`end` 之后的项使流以 `gateway/protocol` 失败，两种情况都不关闭 socket。发给 Host 已结束的流 id 的 `item`、`end` 与 `cancel` 帧被丢弃；只有重复的 `open` 会关闭 socket。`$events` 这类 Gateway 自有的流在打开时立即 return 其 uplink，因此它们的 `item` 帧被丢弃而不缓冲。一元方法也可以调用 `uplink()`；其项只在方法运行期间可读。每条流都代表操作者 Peer：WebSocket 上的流代表 Connection 在升级时接纳的 Peer，该 scope 释放时 socket 关闭；进程内载体的流直接代表操作者。 Host 组合可通过 `registerRemoteEvents()` 注册唯一的应用事件 source。Gateway 为它保留内部 `$events` logical endpoint，只接受空 `args`，并在 source 撤回时中止该注册打开的流。事件名单、参数校验、每个 Client 的队列及 opening `{ type: 'ready', clientId, host: { home } }` frame 中的 Host home 由 API Remotes 拥有。source factory 在返回 iterable 前同步挂好增量 listener，因此 Client 只在增量投递就绪后发布 generation 并开始 baseline 读取。 diff --git a/packages/api/gateway/package.json b/packages/api/gateway/package.json index 9985966d8b..823fd1a6b5 100644 --- a/packages/api/gateway/package.json +++ b/packages/api/gateway/package.json @@ -67,6 +67,7 @@ }, "devDependencies": { "@deepseek-ai/dsh-brand": "workspace:^", + "@deepseek-ai/dsh-cmdline": "workspace:^", "@deepseek-ai/dsh-client-connection": "workspace:^", "@deepseek-ai/dsh-host-webserver": "workspace:^", "@deepseek-ai/dsh-remote-mock": "workspace:^", diff --git a/packages/api/gateway/src/index.ts b/packages/api/gateway/src/index.ts index d9bc91d52a..777da35065 100644 --- a/packages/api/gateway/src/index.ts +++ b/packages/api/gateway/src/index.ts @@ -15,6 +15,7 @@ import { import { Deque } from '@deepseek-ai/dsh-deque' import type { WebUpgradeRoute } from '@deepseek-ai/dsh-host-webserver' import { MAX_TIMER_DELAY_MS } from '@deepseek-ai/dsh-timeout' +import type {} from '@deepseek-ai/dsh-cmdline' import z from '@deepseek-ai/schemastery' export type { TypertGatewayFaultDetails } from './remote-error-codes.ts' import { @@ -217,6 +218,8 @@ export class TypertGatewayService extends Service implements TypertGateway { /** * Register the Gateway against the active Typert registry. + * WebSocket admission waits for launcher-owned application readiness when supplied; + * direct invocation and in-process streams remain available independently. * @param ctx - owning Host Context with Typert registry access. * @param config - validated Gateway transport configuration. */ @@ -234,31 +237,42 @@ export class TypertGatewayService extends Service implements TypertGateway { ) }) ctx.inject(['connection', 'webServer'], (webCtx) => { - const mux = new RemoteStreamMuxServer( - (endpoint, payload, uplink, peer, control) => - this.openWireStream(endpoint, payload, uplink, peer, control.signal, control), - this.wireStream.failure, - resolved.websocketHeartbeatIntervalMs, - resolved.streamInboxBytes, - ) - webCtx.effect(() => { - const route: WebUpgradeRoute = { - path: REMOTE_STREAM_MUX_PATH, - handler: (req, socket, head) => { - const admission = webCtx.connection.admit(req) - if ('rejection' in admission) { - rejectRemoteStreamUpgrade(socket, admission.rejection) - return - } - mux.handleUpgrade(req, socket, head, admission.peer) - }, + const listen = (): void => { + const mux = new RemoteStreamMuxServer( + (endpoint, payload, uplink, peer, control) => + this.openWireStream(endpoint, payload, uplink, peer, control.signal, control), + this.wireStream.failure, + resolved.websocketHeartbeatIntervalMs, + resolved.streamInboxBytes, + ) + webCtx.effect(function* () { + yield () => mux.close() + const route: WebUpgradeRoute = { + path: REMOTE_STREAM_MUX_PATH, + handler: (req, socket, head) => { + const admission = webCtx.connection.admit(req) + if ('rejection' in admission) { + rejectRemoteStreamUpgrade(socket, admission.rejection) + return + } + mux.handleUpgrade(req, socket, head, admission.peer) + }, + } + yield webCtx.webServer.registerUpgrade(route) + }, `api-gateway: ${REMOTE_STREAM_MUX_PATH} WebSocket`) + } + // Existing pages reconnect before the new Host prints its URL. No stream + // may enter until the launcher has activated and audited its controllers. + const ready = webCtx.get('appReady') + if (ready === undefined) listen() + else webCtx.effect(() => { + let closed = false + const cancel = ready.onReady(() => { if (!closed) listen() }) + return () => { + closed = true + cancel() } - const unregister = webCtx.webServer.registerUpgrade(route) - return async () => { - unregister() - await mux.close() - } - }, `api-gateway: ${REMOTE_STREAM_MUX_PATH} WebSocket`) + }, 'api-gateway: application readiness') }) } diff --git a/packages/api/gateway/tests/gateway-stream.host.spec.ts b/packages/api/gateway/tests/gateway-stream.host.spec.ts index 7ed908f620..4c2f0e27c6 100644 --- a/packages/api/gateway/tests/gateway-stream.host.spec.ts +++ b/packages/api/gateway/tests/gateway-stream.host.spec.ts @@ -6,6 +6,7 @@ import { Context, symbols } from '@deepseek-ai/cordis' import { apply as applyConnection, inject as connectionInject } from '@deepseek-ai/dsh-client-connection' import WebServer from '@deepseek-ai/dsh-host-webserver' import { MAX_TIMER_DELAY_MS } from '@deepseek-ai/dsh-timeout' +import type { AppReady } from '@deepseek-ai/dsh-cmdline' import { Remote, remoteErrorOf, @@ -229,6 +230,30 @@ class FeedService extends TypertRemoteService { const roots: Context[] = [] +class StartupProbe implements AppReady { + private ready = false + private readonly listeners = new Set<() => void>() + lastListener: (() => void) | undefined + + get pending(): number { return this.listeners.size } + + onReady(listener: () => void): () => void { + this.lastListener = listener + if (this.ready) { + listener() + return () => {} + } + this.listeners.add(listener) + return () => { this.listeners.delete(listener) } + } + + commit(): void { + this.ready = true + for (const listener of this.listeners) listener() + this.listeners.clear() + } +} + class RemoteEventSourceProbe { readonly source = (signal: AbortSignal): AsyncIterable => { this.signal = signal @@ -307,6 +332,30 @@ afterEach(async () => { }) describe('Typert Remote streams', () => { + it.each([false, true])('accepts WebSockets only after application readiness (already ready: %s)', async (alreadyReady) => { + const startup = new StartupProbe() + if (alreadyReady) startup.commit() + const { ctx } = await setup(true, {}, startup) + if (!alreadyReady) { + expect(startup.pending).toBe(1) + expect(await acceptsSocket(ctx)).toBe(false) + startup.commit() + } + expect(await acceptsSocket(ctx)).toBe(true) + expect(startup.pending).toBe(0) + }) + + it('withdraws a pending WebSocket startup subscription when the Gateway unloads', async () => { + const startup = new StartupProbe() + const { ctx } = await setup(true, {}, startup) + expect(startup.pending).toBe(1) + await ctx.fiber.dispose() + expect(startup.pending).toBe(0) + // A launcher commit can already hold a copy of the cancelled listener. + expect(() => { startup.lastListener?.() }).not.toThrow() + expect(() => { startup.commit() }).not.toThrow() + }) + it('validates the WebSocket heartbeat timer range and the stream inbox bound', () => { expect(TypertGatewayService.Config({})).toEqual({ websocketHeartbeatIntervalMs: 2_000, streamInboxBytes: 262_144 }) expect(TypertGatewayService.Config({ websocketHeartbeatIntervalMs: MAX_TIMER_DELAY_MS, streamInboxBytes: 1 })) @@ -1341,9 +1390,11 @@ describe('Typert Remote streams', () => { async function setup( transport: boolean, gatewayConfig: GatewayConfig = {}, + ready?: AppReady, ): Promise<{ readonly ctx: Context; readonly service: FeedService }> { const ctx = new Context() roots.push(ctx) + if (ready !== undefined) ctx.provide('appReady', ready) if (transport) { await ctx.plugin(WebServer, { host: '127.0.0.1', port: 0 }) provideBrowserCredentials(ctx) @@ -1365,6 +1416,17 @@ async function setup( return { ctx, service: receiver[symbols.original] ?? receiver } } +async function acceptsSocket(ctx: Context): Promise { + const socket = new WebSocket(`ws://127.0.0.1:${String(ctx.webServer.port)}/api/remote.mux`, { + headers: { cookie: browserCookie(ctx) }, + }) + const closed = new Promise((resolve) => { socket.once('close', () => { resolve() }) }) + const opened = await once(socket, 'open').then(() => true, () => false) + if (opened) socket.close() + await closed + return opened +} + function descriptors(): InvocationDescriptor[] { const label = { name: 'label', diff --git a/packages/api/gateway/tsconfig.host.json b/packages/api/gateway/tsconfig.host.json index 0b3353158f..5cb68a7183 100644 --- a/packages/api/gateway/tsconfig.host.json +++ b/packages/api/gateway/tsconfig.host.json @@ -13,6 +13,9 @@ "src/types.ts" ], "references": [ + { + "path": "../../boot/cmdline" + }, { "path": "../../../vendor/cosmokit" }, diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 688e808575..c1afb1ed9a 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1034,6 +1034,9 @@ importers: '@deepseek-ai/dsh-client-connection': specifier: workspace:^ version: link:../../client/connection + '@deepseek-ai/dsh-cmdline': + specifier: workspace:^ + version: link:../../boot/cmdline '@deepseek-ai/dsh-host-webserver': specifier: workspace:^ version: link:../../host/webserver