228 lines
8.2 KiB
TypeScript
228 lines
8.2 KiB
TypeScript
/** Reconnecting lifecycle for one single-consumer Remote stream. */
|
|
|
|
import { RemoteError, remoteErrorOf } from '@deepseek-ai/dsh-typert-protocol'
|
|
import type { ConnectionHandle } from '@deepseek-ai/dsh-client-connection/client'
|
|
import { RemoteStreamCarrierError } from './stream-client.ts'
|
|
|
|
/** One item annotated with the physical Remote-stream generation that delivered it. */
|
|
export interface RemoteStreamItem<Item> {
|
|
/** Monotone physical generation number within this logical stream. */
|
|
readonly generation: number
|
|
/** Decoded item yielded by the generated Remote method. */
|
|
readonly value: Item
|
|
/** Cancellation lifetime of the generation that delivered this item. */
|
|
readonly signal: AbortSignal
|
|
/** Mark this generation's opening baseline or cursor as accepted. */
|
|
accept(): void
|
|
}
|
|
|
|
/** Domain-owned operations used by {@link RemoteStream}. */
|
|
export interface RemoteStreamOptions<Item> {
|
|
/** Diagnostic owner name used for cancellation failures. */
|
|
readonly name: string
|
|
/** Open one physical generation of the logical stream. */
|
|
readonly open: (signal: AbortSignal) => AsyncIterable<Item>
|
|
/** Classify a normal generation end after or before its opening item was accepted. */
|
|
readonly ended: (accepted: boolean) => Error
|
|
/** Observe a retryable carrier loss before the supervisor waits or reopens. */
|
|
readonly carrierFailed?: (error: RemoteStreamCarrierError) => void
|
|
}
|
|
|
|
/**
|
|
* Reopens one logical Remote stream across carrier generations.
|
|
*
|
|
* Connection owns physical retry timing; Gateway performs each requested
|
|
* replacement. The domain consumer owns its opening item and every later
|
|
* item, and calls {@link RemoteStreamItem.accept} only after validating the
|
|
* opening baseline or cursor.
|
|
*/
|
|
export class RemoteStream<Item> implements AsyncIterable<RemoteStreamItem<Item>> {
|
|
private readonly lifetime = new AbortController()
|
|
private generationAbort: AbortController | undefined
|
|
private iterator: AsyncGenerator<RemoteStreamItem<Item>> | undefined
|
|
private closing: Promise<void> | undefined
|
|
private revision = 0
|
|
private taken = false
|
|
|
|
/**
|
|
* @param connection - observable Host generation source used to pace retries.
|
|
* @param options - domain stream opener, end classification, and diagnostics.
|
|
*/
|
|
constructor(
|
|
private readonly connection: Pick<ConnectionHandle, 'generation'>,
|
|
private readonly options: RemoteStreamOptions<Item>,
|
|
) {}
|
|
|
|
/** Cancellation lifetime shared by the stream and sibling page requests. */
|
|
get signal(): AbortSignal {
|
|
return this.lifetime.signal
|
|
}
|
|
|
|
/** Interrupt the current generation and immediately request a replacement. */
|
|
restart(): void {
|
|
if (this.lifetime.signal.aborted) return
|
|
this.revision++
|
|
this.generationAbort?.abort(new Error(`${this.options.name} generation restarted`))
|
|
}
|
|
|
|
/**
|
|
* Permanently stop this stream and wait for its iterator to close.
|
|
* @returns when the active generation and consumer iterator are quiescent.
|
|
*/
|
|
dispose(): Promise<void> {
|
|
if (this.closing !== undefined) return this.closing
|
|
if (!this.lifetime.signal.aborted) {
|
|
const reason = new Error(`${this.options.name} disposed`)
|
|
this.lifetime.abort(reason)
|
|
this.generationAbort?.abort(reason)
|
|
}
|
|
const iterator = this.iterator
|
|
if (iterator === undefined) return Promise.resolve()
|
|
const closing = closeRemoteStreamIterator(iterator)
|
|
this.closing = closing
|
|
return closing
|
|
}
|
|
|
|
/** @inheritdoc */
|
|
[Symbol.asyncIterator](): AsyncIterator<RemoteStreamItem<Item>> {
|
|
if (this.taken) throw new Error(`${this.options.name} already has a consumer`)
|
|
this.taken = true
|
|
const iterator = this.read()
|
|
this.iterator = iterator
|
|
return iterator
|
|
}
|
|
|
|
private async * read(): AsyncGenerator<RemoteStreamItem<Item>> {
|
|
let attempt = 0
|
|
let generation = 0
|
|
let observedRevision = this.revision
|
|
try {
|
|
while (!isAborted(this.lifetime.signal)) {
|
|
if (observedRevision !== this.revision) {
|
|
observedRevision = this.revision
|
|
attempt = 0
|
|
}
|
|
const revision = this.revision
|
|
const generationAbort = new AbortController()
|
|
this.generationAbort = generationAbort
|
|
const signal = AbortSignal.any([this.lifetime.signal, generationAbort.signal])
|
|
const generationId = ++generation
|
|
let accepted = false
|
|
try {
|
|
for await (const value of this.options.open(signal)) {
|
|
if (isAborted(this.lifetime.signal)) return
|
|
if (revision !== this.revision) break
|
|
yield {
|
|
generation: generationId,
|
|
value,
|
|
signal,
|
|
accept: () => {
|
|
if (this.generationAbort !== generationAbort || revision !== this.revision) return
|
|
accepted = true
|
|
attempt = 0
|
|
},
|
|
}
|
|
}
|
|
if (isAborted(this.lifetime.signal)) return
|
|
if (revision !== this.revision) continue
|
|
throw this.options.ended(accepted)
|
|
} catch (error) {
|
|
if (isAborted(this.lifetime.signal)) return
|
|
if (revision !== this.revision) continue
|
|
if (!(error instanceof RemoteStreamCarrierError)) throw terminalStreamFailure(error)
|
|
this.options.carrierFailed?.(error)
|
|
if (revision !== this.revision) continue
|
|
attempt++
|
|
try {
|
|
await waitForRemoteStreamRetry(this.connection, error, attempt, signal)
|
|
} catch (retryError) {
|
|
if (isAborted(this.lifetime.signal)) return
|
|
if (revision !== this.revision) continue
|
|
throw terminalStreamFailure(retryError)
|
|
}
|
|
} finally {
|
|
this.generationAbort = undefined
|
|
if (!generationAbort.signal.aborted) {
|
|
generationAbort.abort(new Error(`${this.options.name} generation ended`))
|
|
}
|
|
}
|
|
}
|
|
} finally {
|
|
if (!this.lifetime.signal.aborted) {
|
|
this.lifetime.abort(new Error(`${this.options.name} consumer closed`))
|
|
}
|
|
this.generationAbort?.abort(this.lifetime.signal.reason)
|
|
this.generationAbort = undefined
|
|
}
|
|
}
|
|
}
|
|
|
|
async function waitForRemoteStreamRetry(
|
|
connection: Pick<ConnectionHandle, 'generation'>,
|
|
error: RemoteStreamCarrierError,
|
|
attempt: number,
|
|
signal: AbortSignal,
|
|
): Promise<void> {
|
|
signal.throwIfAborted()
|
|
if (connection.generation.getSnapshot() !== undefined) {
|
|
if (attempt === 1) return
|
|
throw error
|
|
}
|
|
await new Promise<void>((resolve, reject) => {
|
|
const subscription: {
|
|
dispose?: () => void
|
|
finished: boolean
|
|
} = { finished: false }
|
|
const finish = (failure?: Error): void => {
|
|
if (subscription.finished) return
|
|
subscription.finished = true
|
|
subscription.dispose?.()
|
|
signal.removeEventListener('abort', aborted)
|
|
if (failure === undefined) resolve()
|
|
else reject(failure)
|
|
}
|
|
const inspect = (): void => {
|
|
if (connection.generation.getSnapshot() !== undefined) finish()
|
|
}
|
|
const aborted = (): void => {
|
|
finish(new Error('Remote stream retry aborted', { cause: signal.reason }))
|
|
}
|
|
const dispose = connection.generation.subscribe(inspect)
|
|
subscription.dispose = dispose
|
|
if (subscription.finished) dispose()
|
|
signal.addEventListener('abort', aborted, { once: true })
|
|
if (signal.aborted) aborted()
|
|
else inspect()
|
|
})
|
|
}
|
|
|
|
/**
|
|
* Mark a terminal escape before it crosses the stream boundary: consumers
|
|
* discriminate failures by code, so an unmarked throw reads as a local bug.
|
|
* Marked failures pass through verbatim. The carrier class never escapes as a
|
|
* terminal outcome — it stays the retry-internal signal fed to `carrierFailed`
|
|
* and the `ended(true)` retry trigger.
|
|
*/
|
|
function terminalStreamFailure(error: unknown): Error {
|
|
return remoteErrorOf(error) ?? new RemoteError(
|
|
'gateway/internal',
|
|
error instanceof Error ? error.message : String(error),
|
|
{},
|
|
{ cause: error },
|
|
)
|
|
}
|
|
|
|
function isAborted(signal: AbortSignal): boolean {
|
|
return signal.aborted
|
|
}
|
|
|
|
async function closeRemoteStreamIterator<Item>(
|
|
iterator: AsyncIterator<RemoteStreamItem<Item>>,
|
|
): Promise<void> {
|
|
try {
|
|
await iterator.return?.()
|
|
} catch {
|
|
// The disposed logical stream has no remaining consumer for cancellation failures.
|
|
}
|
|
}
|