diff --git a/.agents/notes/implemented/architecture/2026-09-19-remote-duplex-stream.i18n.yaml b/.agents/notes/implemented/architecture/2026-09-19-remote-duplex-stream.i18n.yaml index d5903d0fc2..9a6d0df300 100644 --- a/.agents/notes/implemented/architecture/2026-09-19-remote-duplex-stream.i18n.yaml +++ b/.agents/notes/implemented/architecture/2026-09-19-remote-duplex-stream.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 .agents/notes/implemented/architecture/2026-09-19-remote-duplex-stream.md -2026-09-19-remote-duplex-stream.md: 0f1f6a575f6871d37822c5829fe4531261b8e95b -2026-09-19-remote-duplex-stream.zh.md: d3982aa433aa3ce84b945bb5831cef8064472853 +2026-09-19-remote-duplex-stream.md: ea8e8cebcc8285c97ebdb70f4e37805d3ce8bfe9 +2026-09-19-remote-duplex-stream.zh.md: 2e119bf054e20d076e22758178b5e0168092be18 diff --git a/.agents/notes/implemented/architecture/2026-09-19-remote-duplex-stream.md b/.agents/notes/implemented/architecture/2026-09-19-remote-duplex-stream.md index 0f1f6a575f..ea8e8cebcc 100644 --- a/.agents/notes/implemented/architecture/2026-09-19-remote-duplex-stream.md +++ b/.agents/notes/implemented/architecture/2026-09-19-remote-duplex-stream.md @@ -276,6 +276,12 @@ Parsing rules: `item` has exactly the keys `type` and `streamId` plus an optiona | `item` received after `end` | Any | The stream fails with a `gateway/protocol` error frame and aborts; the socket stays open | | `item` / `end` with an unknown `streamId` | Any | Ignored, like `cancel`: after the Host ends a stream and deletes its id, uplink frames still in flight from the Client are normal and must not take down the other streams on the same socket | +### Cancellation wait lifetime + +`cancellableStream`, `UplinkDecoder.next()`, and the Client uplink pump each create a cancellation or stop promise per read. `Promise.race` does not remove the losing input's reaction when a read completes: reusing a stream-wide pending promise retains completed races and their item results until that promise settles or becomes unreachable. Read-local promises keep retained cancellation state independent of the number of delivered items. + +The Host downlink clears its current reject callback after each read and checks cancellation before reading again. The uplink decoder retains only pending reads: abort rejects them, while normal close resolves all of them with `done`, including overlapping reads. The Client pump drops its stop promise after each read and does not start another read after stopping. Source iterators retain their existing release ownership; closing the Host uplink still precedes returning the method iterator. + ### Ordering guarantees Frames on one socket arrive in total order: a logical stream's `open`, `item`…, `end` reach the Host in send order, and uplink items enter `uplink()` in the order of `send`. There is no cross-direction ordering guarantee between uplink and downlink; a protocol that needs request-response pairing carries its own sequence numbers. @@ -538,6 +544,10 @@ for await (const reply of stream) replies.push(reply) // ['> a', '> b'] ## Alternatives considered +**A shared cancellation controller for all three reading loops.** Each loop already has an owner with distinct close behavior. Read-local promises bound retained state without moving those responsibilities into a new abstraction. + +**Requiring source iterators to cancel their own `next()`.** Gateway accepts arbitrary iterables, whose pending reads need not respond to cancellation or `return()`. Gateway must still wake its own waiters independently of source cleanup. + **A separate `mode: 'duplex'` plus a reserved parameter `uplink: AsyncIterable` alongside `signal`.** Special logic everywhere: the analyzer must recognize a second reserved parameter and its position rules, the generated signature gains a parameter, load time must validate `duplex ⟺ uplink`, and because the mux does not know the method mode, the Gateway must reject uplink frames on behalf of `'stream'` methods. Duplex is an inherent property of the transport-layer stream, not a mode of the method: the carrier already has `streamId`, `open`, and `cancel`; the uplink adds just two more frames, which any stream can receive and which are simply not sent when unused. **Declaring the uplink type in a decorator option, `@Remote({ mode: 'stream', uplink: 'JobInputFrame' })`.** A string reference to a type name is not checked by the compiler and breaks on rename. The return-type alias puts both types in one compiler-visible place. @@ -572,6 +582,8 @@ for await (const reply of stream) replies.push(reply) // ['> a', '> b'] ## Testing +Retention cases in `gateway-stream.host.spec.ts` and `gateway.client.spec.ts` count live iterator results after a full GC while the streams remain open. Lifecycle cases cover normal close waking multiple pending uplink reads, cancellation between reads, synchronous abort followed by a source exception, and termination during a Client send. + - `packages/api/gateway/tests/echo-stream.host.spec.ts`: the same echo method passes through the WebSocket carrier and through the in-process carrier. - Gateway Host: exact-key validation in frame parsing; `UplinkInbox` buffering during opening, `item` after `end`, overflow, ignored unknown `streamId`, and cleanup on socket closure; `UplinkDecoder` delivery with and without a codec and abort on decode failure; `uplink()` taken only once; the `extend({ invocation })` view and the origin of `peer`; `cancellableStream` closing the uplink before `return()`. - Gateway Client: the handle's `send` (JSON validation, queued before open, throws after termination), idempotent `end`, `dispose` driving the `cancel` frame and ending later reads; the pump's three exits; the pump stopping as soon as a terminal frame arrives. diff --git a/.agents/notes/implemented/architecture/2026-09-19-remote-duplex-stream.zh.md b/.agents/notes/implemented/architecture/2026-09-19-remote-duplex-stream.zh.md index d3982aa433..2e119bf054 100644 --- a/.agents/notes/implemented/architecture/2026-09-19-remote-duplex-stream.zh.md +++ b/.agents/notes/implemented/architecture/2026-09-19-remote-duplex-stream.zh.md @@ -276,6 +276,12 @@ export type RemoteStreamServerMessage = | `end` 之后又收到 `item` | 任意 | 该流以 `gateway/protocol` 错误帧失败并中止；socket 不关闭 | | 未知 `streamId` 的 `item` / `end` | 任意 | 忽略，与 `cancel` 相同：Host 结束流并删除 id 后，客户端在途的上行帧属正常现象，不能连累同一 socket 上的其他流 | +### 取消等待的生命周期 + +`cancellableStream`、`UplinkDecoder.next()` 和 Client 上行泵各自为单次读取创建取消或停止 Promise。读取完成后，`Promise.race` 不会移除另一输入上的 reaction：复用流级未完成 Promise 会保留已完成的竞争及其读取结果，直到该 Promise 完成或不可达。单次读取的 Promise 使取消状态的保留量不随已交付项数增长。 + +Host 下行在每次读取后清除当前拒绝回调，并在再次读取前检查取消状态。上行解码器只保留尚未完成的读取：取消时拒绝这些读取，正常关闭则以 `done` 唤醒全部读取，包括重叠的读取。Client 泵在每次读取后移除停止 Promise，停止后不再发起读取。源迭代器的释放仍由原有所有者负责；Host 仍先关闭上行，再返回方法迭代器。 + ### 顺序保证 同一条 socket 上的帧全序到达：一条逻辑流的 `open`、`item`…、`end` 按发送顺序到达 Host；上行项进入 `uplink()` 的顺序就是 `send` 的顺序。上行与下行之间没有跨方向顺序保证；需要请求应答配对的协议自带序号。 @@ -538,6 +544,10 @@ for await (const reply of stream) replies.push(reply) // ['> a', '> b'] ## 考虑过的替代方案 +**三处读取循环共用一个取消控制器。** 每处循环已有自己的所有者和不同的关闭行为。单次读取的 Promise 能限制保留状态，不需要把这些职责迁入新抽象。 + +**要求源迭代器自行取消 `next()`。** Gateway 接受任意 iterable，其待处理读取不一定响应取消或 `return()`。Gateway 仍需独立于源清理来唤醒自己的等待者。 + **独立的 `mode: 'duplex'` 加一个与 `signal` 同级的保留参数 `uplink: AsyncIterable`。** 每一处都是特殊逻辑：analyzer 要认第二个保留参数及其位置规则，生成签名多一个参数，加载期要校验 `duplex ⟺ uplink`，mux 不知道方法模式所以网关要替 `'stream'` 方法拒绝上行帧。双工是传输层流的固有性质，不是方法的一种模式：载体本来就有 `streamId`、`open`、`cancel`，上行只是再加两种帧，任何流都能收，不用就不发。 **用装饰器选项声明上行类型，`@Remote({ mode: 'stream', uplink: 'JobInputFrame' })`。** 字符串引用类型名不受编译器约束，重命名即断。返回类型别名把两个类型写在同一个编译器可见的位置。 @@ -572,6 +582,8 @@ for await (const reply of stream) replies.push(reply) // ['> a', '> b'] ## 测试 +`gateway-stream.host.spec.ts` 和 `gateway.client.spec.ts` 中的保留对象用例在流保持打开时进行全量 GC，再统计存活的迭代结果。生命周期用例覆盖正常关闭唤醒多个待处理上行读取、两次读取之间的取消、源同步取消后抛错，以及 Client 发送期间流结束。 + - `packages/api/gateway/tests/echo-stream.host.spec.ts`：同一个 echo 方法分别穿过 WebSocket 载体与进程内载体。 - 网关 Host：帧解析的精确键校验；`UplinkInbox` 的 opening 缓冲、`end` 后 `item`、超限、未知 `streamId` 忽略、socket 关闭时的收尾；`UplinkDecoder` 有无 codec 两种交付与解码失败中止；`uplink()` 只能取一次；`extend({ invocation })` 视图与 `peer` 的来源；`cancellableStream` 先关 uplink 再 `return()`。 - 网关 Client：句柄的 `send`（JSON 校验、open 前入队、终止后抛错）、`end` 幂等、`dispose` 驱动 `cancel` 帧并结束后续读取；泵的三种退出；终止帧到达即停泵。 diff --git a/packages/api/gateway/README.i18n.yaml b/packages/api/gateway/README.i18n.yaml index 7f47a6c9eb..b7daafd1e3 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: 1a746a83eb966fba6ae40b61061d42bf478e1c84 -README.zh.md: 7b37548f6ad894b761457df80278eeed7aa6ca7b +README.md: e0c986e290b1744017fca99e3a2553a68d754ef6 +README.zh.md: 8be11a7c5aea1e299c4ca45cde4b1c9dfa08ec4b diff --git a/packages/api/gateway/README.md b/packages/api/gateway/README.md index 1a746a83eb..e0c986e290 100644 --- a/packages/api/gateway/README.md +++ b/packages/api/gateway/README.md @@ -38,6 +38,8 @@ When the launcher supplies `ctx.appReady`, Gateway registers the WebSocket upgra 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. +Stream cancellation and stopping keep read-local state, not a history of delivered items, on the Host downlink, Host uplink, and Client uplink pump. Completing the downlink also wakes pending uplink reads. + 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 7b37548f6a..8be11a7c5a 100644 --- a/packages/api/gateway/README.zh.md +++ b/packages/api/gateway/README.zh.md @@ -38,6 +38,8 @@ Connection 可用时，Host 入口会在 Connection 共享的 `/api` FetchHandle 每条流还在同一条逻辑流上携带从 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 下行、Host 上行和 Client 上行泵的取消与停止状态归属单次读取，不保留已交付项的历史。下行结束也会唤醒尚未完成的上行读取。 + 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/src/client/stream-client.ts b/packages/api/gateway/src/client/stream-client.ts index 7a580c5fac..92255a10ee 100644 --- a/packages/api/gateway/src/client/stream-client.ts +++ b/packages/api/gateway/src/client/stream-client.ts @@ -158,7 +158,7 @@ export class RemoteStreamMuxClient { signal: AbortSignal, inbox: StreamInbox, ): UplinkPump { - const stopped = Promise.withResolvers>() + let stopped: PromiseWithResolvers> | undefined const interruption: IteratorReturnResult = { value: undefined, done: true } const uplinkIterator = uplink[Symbol.asyncIterator]() const state = { stopping: false, exhausted: false, released: false } @@ -171,7 +171,12 @@ export class RemoteStreamMuxClient { const done = (async (): Promise => { try { while (true) { + const stoppingBeforeRead = state.stopping + if (stoppingBeforeRead) return + // A stop promise belongs to one read, not the whole uplink history. + stopped = Promise.withResolvers>() const next = await Promise.race([uplinkIterator.next(), stopped.promise]) + stopped = undefined if (state.stopping || next === interruption || signal.aborted || this.socket !== socket) return if (next.done === true) { state.exhausted = true @@ -183,6 +188,7 @@ export class RemoteStreamMuxClient { } catch (error) { inbox.fail(error) } finally { + stopped = undefined release() } })() @@ -191,7 +197,7 @@ export class RemoteStreamMuxClient { stop: () => { if (state.stopping) return state.stopping = true - stopped.resolve(interruption) + stopped?.resolve(interruption) release() }, } diff --git a/packages/api/gateway/src/index.ts b/packages/api/gateway/src/index.ts index 777da35065..7c7a29b66a 100644 --- a/packages/api/gateway/src/index.ts +++ b/packages/api/gateway/src/index.ts @@ -1164,19 +1164,24 @@ async function *cancellableStream( ? Reflect.apply(asyncFactory, source, []) as AsyncIterator : Reflect.apply(syncFactory as (...args: never[]) => Iterator, source, []) let rejectAbort: ((error: unknown) => void) | undefined - const aborted = new Promise((_resolve, reject) => { rejectAbort = reject }) const onAbort = (): void => { rejectAbort?.(streamAbortFailure(endpoint, signal.reason)) } signal.addEventListener('abort', onAbort, { once: true }) try { - if (signal.aborted) throw streamAbortFailure(endpoint, signal.reason) while (true) { + if (signal.aborted) throw streamAbortFailure(endpoint, signal.reason) + // Reusing a pending cancellation promise retains every completed race. + const aborted = new Promise((_resolve, reject) => { rejectAbort = reject }) + // next() can abort and throw synchronously before the race subscribes. + void aborted.catch(() => undefined) const next = await Promise.race([Promise.resolve(iterator.next()), aborted]) + rejectAbort = undefined if (next.done === true) return yield next.value } } finally { + rejectAbort = undefined signal.removeEventListener('abort', onAbort) // The uplink closes first so a method blocked on `uplink.next()` unwinds // before its iterator is asked to return. @@ -1209,9 +1214,10 @@ function remoteCancelled(endpoint: string, cause: unknown): RemoteError<'gateway */ class UplinkDecoder implements AsyncIterable, AsyncIterator { private readonly source: AsyncIterator - private readonly interrupted = Promise.withResolvers>() + private readonly interrupted = new Set>>() private readonly onAbort = (): void => { - this.interrupted.reject(streamAbortFailure(this.endpoint, this.signal.reason)) + const failure = streamAbortFailure(this.endpoint, this.signal.reason) + for (const read of this.interrupted) read.reject(failure) } private closed = false @@ -1223,9 +1229,6 @@ class UplinkDecoder implements AsyncIterable, AsyncIterator { private readonly abort: (reason: unknown) => void, ) { this.source = uplink[Symbol.asyncIterator]() - // The rejection settles the race inside next(); an abort with no read in - // flight must not surface as an unhandled rejection. - void this.interrupted.promise.catch(() => undefined) signal.addEventListener('abort', this.onAbort, { once: true }) } @@ -1237,7 +1240,17 @@ class UplinkDecoder implements AsyncIterable, AsyncIterator { // A cancelled stream reports the cancellation on every read, closed or not. if (this.signal.aborted) throw streamAbortFailure(this.endpoint, this.signal.reason) if (this.closed) return UPLINK_DONE - const next = await Promise.race([this.source.next(), this.interrupted.promise]) + // Only pending reads belong to the decoder; finish() must wake all of them. + const interrupted = Promise.withResolvers>() + this.interrupted.add(interrupted) + // The source can abort and throw synchronously before the race subscribes. + void interrupted.promise.catch(() => undefined) + let next: IteratorResult + try { + next = await Promise.race([this.source.next(), interrupted.promise]) + } finally { + this.interrupted.delete(interrupted) + } if (next.done === true) { this.finish() return UPLINK_DONE @@ -1270,7 +1283,7 @@ class UplinkDecoder implements AsyncIterable, AsyncIterator { private finish(): void { this.closed = true this.signal.removeEventListener('abort', this.onAbort) - this.interrupted.resolve(UPLINK_DONE) + for (const read of this.interrupted) read.resolve(UPLINK_DONE) } } diff --git a/packages/api/gateway/tests/gateway-stream.host.spec.ts b/packages/api/gateway/tests/gateway-stream.host.spec.ts index 4c2f0e27c6..87aaaa9f75 100644 --- a/packages/api/gateway/tests/gateway-stream.host.spec.ts +++ b/packages/api/gateway/tests/gateway-stream.host.spec.ts @@ -1,5 +1,6 @@ import { randomUUID } from 'node:crypto' import { once } from 'node:events' +import { queryObjects } from 'node:v8' import { afterEach, describe, expect, it, vi } from 'vitest' import WebSocket, { type RawData } from 'ws' import { Context, symbols } from '@deepseek-ai/cordis' @@ -56,6 +57,7 @@ class FeedService extends TypertRemoteService { readonly peeked: unknown[] = [] readonly peers: PeerScope[] = [] leftover: AsyncIterator | undefined + source: Iterable = [] returns = 0 constructor(ctx: Context) { @@ -82,6 +84,11 @@ class FeedService extends TypertRemoteService { yield `${label}:two` } + @Remote({ mode: 'stream' }) + items(): Iterable { + return this.source + } + @Remote({ mode: 'stream' }) *invalid(): Iterable { yield 42 as unknown as string @@ -541,6 +548,110 @@ describe('Typert Remote streams', () => { await expect(iterator.return?.()).resolves.toEqual({ done: true, value: undefined }) }) + it('finishes every pending uplink read when the downlink returns', async () => { + const { ctx, service } = await setup(false) + const reads = [ + Promise.withResolvers>(), + Promise.withResolvers>(), + ] + let index = 0 + const returned = vi.fn(async (): Promise> => ({ done: true, value: undefined })) + const source = await ctx.typertGateway.stream({ + namespace: 'feed', method: 'hold', args: {}, + uplink: { [Symbol.asyncIterator]: () => ({ next: () => reads[index++]!.promise, return: returned }) }, + }) + const iterator = source[Symbol.asyncIterator]() + const completed: IteratorResult[] = [] + const pending: Promise[] = [] + try { + await expect(iterator.next()).resolves.toEqual({ done: false, value: 'held' }) + const uplink = service.leftover! + pending.push(...reads.map(() => uplink.next().then((result) => { completed.push(result) }))) + await iterator.return?.() + expect(completed).toEqual([ + { done: true, value: undefined }, + { done: true, value: undefined }, + ]) + expect(returned).toHaveBeenCalledOnce() + } finally { + for (const read of reads) read.resolve({ done: true, value: undefined }) + await Promise.all(pending) + await iterator.return?.() + } + }) + + it.each(['downlink', 'uplink'] as const)('releases consumed %s read results while the stream stays open', async (direction) => { + class ReadResult implements IteratorYieldResult { + readonly done = false + readonly value = 'item' + } + const { ctx, service } = await setup(false) + service.source = { [Symbol.iterator]: () => ({ next: () => new ReadResult() }) } + const source = await ctx.typertGateway.stream(direction === 'downlink' + ? { namespace: 'feed', method: 'items', args: {} } + : { + namespace: 'feed', method: 'echo', args: { prefix: '' }, + uplink: { [Symbol.asyncIterator]: () => ({ next: async () => new ReadResult() }) }, + }) + const iterator = source[Symbol.asyncIterator]() + try { + for (const count of [64, 256]) { + for (let index = 0; index < count; index++) { + await expect(iterator.next()).resolves.toEqual({ done: false, value: 'item' }) + } + // queryObjects collects garbage; the suspended read can retain its last result, not the history. + expect(queryObjects(ReadResult, { format: 'count' })).toBeLessThanOrEqual(2) + } + } finally { + await iterator.return?.() + } + }) + + it('does not read another downlink item after cancellation between reads', async () => { + const { ctx, service } = await setup(false) + const abort = new AbortController() + const next = vi.fn((): IteratorResult => ({ done: false, value: 'item' })) + service.source = { [Symbol.iterator]: () => ({ next }) } + const source = await ctx.typertGateway.stream({ + namespace: 'feed', method: 'items', args: {}, signal: abort.signal, + }) + const iterator = source[Symbol.asyncIterator]() + try { + await expect(iterator.next()).resolves.toEqual({ done: false, value: 'item' }) + abort.abort(new Error('caller cancelled between reads')) + await expect(iterator.next()).rejects.toMatchObject({ code: 'gateway/cancelled' }) + expect(next).toHaveBeenCalledOnce() + } finally { + await iterator.return?.() + } + }) + + it.each(['downlink', 'uplink'] as const)('handles a synchronous %s read that aborts and throws', async (direction) => { + const { ctx, service } = await setup(false) + const abort = new AbortController() + const failure = new Error('source aborted and threw') + const next = (): never => { + abort.abort(failure) + throw failure + } + service.source = { [Symbol.iterator]: () => ({ next }) } + const source = await ctx.typertGateway.stream(direction === 'downlink' + ? { namespace: 'feed', method: 'items', args: {}, signal: abort.signal } + : { + namespace: 'feed', method: 'hold', args: {}, signal: abort.signal, + uplink: { [Symbol.asyncIterator]: () => ({ next }) }, + }) + const iterator = source[Symbol.asyncIterator]() + try { + if (direction === 'uplink') { + await expect(iterator.next()).resolves.toEqual({ done: false, value: 'held' }) + } + await expect((direction === 'downlink' ? iterator : service.leftover!).next()).rejects.toBe(failure) + } finally { + await iterator.return?.() + } + }) + it('reports done on every read after the uplink ended', async () => { const { ctx } = await setup(false) await expect(collect(await ctx.typertGateway.stream({ @@ -1461,6 +1572,7 @@ function descriptors(): InvocationDescriptor[] { stream('context', [label], z.string()), { ...stream('follow', [label], z.string()), cancellation: { parameter: 'signal' } }, stream('sync', [label], z.string()), + stream('items', [], z.string()), stream('invalid', [], z.string()), stream('nonJson', [], z.unknown()), stream('missing', [], z.string()), diff --git a/packages/api/gateway/tests/gateway.client.spec.ts b/packages/api/gateway/tests/gateway.client.spec.ts index aeec14ecf1..b0c7d4f390 100644 --- a/packages/api/gateway/tests/gateway.client.spec.ts +++ b/packages/api/gateway/tests/gateway.client.spec.ts @@ -1,3 +1,4 @@ +import { queryObjects } from 'node:v8' import { RemoteError, typertOwnedValue } from '@deepseek-ai/dsh-typert-protocol' import { Context, Service } from '@deepseek-ai/cordis' import type { Fiber } from '@deepseek-ai/cordis' @@ -2804,6 +2805,81 @@ describe('Client Typert API', () => { }) describe('Remote stream client carrier lifecycle', () => { + it('does not pull another uplink item when sending ends the stream', async () => { + await withFakeWebSocket('https://harness.example', async () => { + const client = new RemoteStreamMuxClient() + client.start() + const socket = FakeWebSocket.sockets[0]! + const send = socket.send.bind(socket) + const sent = vi.spyOn(socket, 'send').mockImplementation((text) => { + send(text) + const frame = parseRemoteStreamClientMessage(text) + if (frame.type === 'item') socket.receive({ type: 'end', streamId: frame.streamId }) + }) + const next = vi.fn(async (): Promise> => ({ done: false, value: 'item' })) + const returned = vi.fn(async (): Promise> => ({ done: true, value: undefined })) + try { + const iterator = client.open('probe/attach', {}, new AbortController().signal, { + [Symbol.asyncIterator]: () => ({ next, return: returned }), + })[Symbol.asyncIterator]() + await expect(iterator.next()).resolves.toEqual({ done: true, value: undefined }) + expect(next).toHaveBeenCalledOnce() + expect(returned).toHaveBeenCalledOnce() + } finally { + await client.close() + sent.mockRestore() + } + }) + }) + + it('releases sent uplink read results while the stream stays open', async () => { + await withFakeWebSocket('https://harness.example', async () => { + class ReadResult implements IteratorYieldResult { + readonly done = false + readonly value = 'item' + } + const blocked = Promise.withResolvers() + const lastRead = Promise.withResolvers>() + let reads = 0 + const uplink: AsyncIterable = { + [Symbol.asyncIterator]: () => ({ + next: () => { + if (reads++ < 256) return Promise.resolve(new ReadResult()) + blocked.resolve(undefined) + return lastRead.promise + }, + return: async () => { + lastRead.resolve({ done: true, value: undefined }) + return { done: true, value: undefined } + }, + }), + } + const client = new RemoteStreamMuxClient() + client.start() + const iterator = client.open('probe/attach', {}, new AbortController().signal, uplink)[Symbol.asyncIterator]() + const first = iterator.next() + // Closing the carrier during failed assertions can reject this pending read. + void first.catch(() => undefined) + try { + await blocked.promise + const socket = FakeWebSocket.sockets[0]! + expect(socket.sent).toHaveLength(257) + const opened = parseRemoteStreamClientMessage(socket.sent[0]!) + if (opened.type !== 'open') throw new Error('fixture expected an open frame') + socket.receive({ type: 'item', streamId: opened.streamId, value: 'ready' }) + await expect(first).resolves.toEqual({ done: false, value: 'ready' }) + // A full GC runs while the uplink pump is still waiting for its next item. + expect(queryObjects(ReadResult, { format: 'count' })).toBeLessThanOrEqual(2) + socket.receive({ type: 'end', streamId: opened.streamId }) + await expect(iterator.next()).resolves.toEqual({ done: true, value: undefined }) + } finally { + await client.close() + await first.catch(() => undefined) + await iterator.return?.(undefined) + } + }) + }) + it('connects to the shell-owned Host while the document uses a local asset origin', async () => { await withFakeWebSocket('dsh-app://app', async () => { vi.stubGlobal('__DSH_TRANSPORT__', { streamBaseUrl: 'http://127.0.0.1:43210' })