From 1b97207150ca8dc36cf6814aba8edeaec4f9f7ca Mon Sep 17 00:00:00 2001 From: Yichen Jiang Date: Thu, 3 Sep 2026 15:53:59 +0800 Subject: [PATCH] refactor(jobs-local): extract the output ring and the registry-owned pump --- packages/jobs/jobs-local/src/pump.ts | 85 +++++++++++++++ packages/jobs/jobs-local/src/ring.ts | 113 ++++++++++++++++++++ packages/jobs/jobs-local/tests/pump.spec.ts | 111 +++++++++++++++++++ packages/jobs/jobs-local/tests/ring.spec.ts | 65 +++++++++++ 4 files changed, 374 insertions(+) create mode 100644 packages/jobs/jobs-local/src/pump.ts create mode 100644 packages/jobs/jobs-local/src/ring.ts create mode 100644 packages/jobs/jobs-local/tests/pump.spec.ts create mode 100644 packages/jobs/jobs-local/tests/ring.spec.ts diff --git a/packages/jobs/jobs-local/src/pump.ts b/packages/jobs/jobs-local/src/pump.ts new file mode 100644 index 0000000000..e57cb02a09 --- /dev/null +++ b/packages/jobs/jobs-local/src/pump.ts @@ -0,0 +1,85 @@ +/** + * The registry-owned pull pump: copies a job's {@link JobOutputSource}s into + * its ring at a bounded cadence and drains them once more after the + * producer settles, so the ring holds every byte before settlement trims and + * closes it. Pure utility — no cordis, no timers retained past settlement. + * @module @deepseek-ai/dsh-jobs-local/pump + */ + +import type { JobAppendOptions, JobOutputSource } from '@deepseek-ai/dsh-jobs' + +/** One pump run; `done` resolves after the final post-settlement drain. */ +export interface PumpHandle { + done: Promise +} + +/** + * Drain every source in array order, sleep `pollMs` or until `until` + * settles, repeat, then drain one final time. A lossy source read appends + * its surviving tail with `gapBefore`, so the discontinuity stays visible to + * observers. Sources drain in array order each round, so bytes two sources + * produced inside one poll window land in that order, not in the order they + * were written: the ring is a best-effort live view whose cross-source + * reordering is bounded by `pollMs`. + * + * The wait holds constant resources however long the job runs: one + * subscription on `until` for the whole run and one pending timer at a time. + * @param sources - producer streams, each pumped at its own offset. + * @param append - the ring append; receives each copied chunk. + * @param pollMs - poll interval in milliseconds; a positive finite number. + * @param until - settles (or rejects, which counts as settlement) when the producer finished. + * @returns the handle whose `done` resolves after the final drain. + */ +export function startPump( + sources: readonly JobOutputSource[], + append: (text: string, options?: JobAppendOptions) => void, + pollMs: number, + until: Promise, +): PumpHandle { + if (!Number.isFinite(pollMs) || pollMs <= 0) { + throw new Error(`invalid pump pollMs: expected a positive finite number of milliseconds, got ${JSON.stringify(pollMs)}`) + } + const states = sources.map(source => ({ source, cursor: 0 })) + const drain = (): void => { + for (const state of states) { + const { source } = state + const read = source.read(state.cursor) + state.cursor = read.nextOffset + if (read.text.length === 0) continue + if (source.channel === undefined && !read.lossy) { + append(read.text) + } else { + append(read.text, { + ...source.channel !== undefined ? { channel: source.channel } : {}, + ...read.lossy ? { gapBefore: true as const } : {}, + }) + } + } + } + const done = (async (): Promise => { + let settled = false + let timer: ReturnType | undefined + let wake: (() => void) | undefined + const finish = (): void => { + settled = true + // Clear the pending poll so a finished pump holds no timer for up to one + // interval, and release the current wait so the final drain runs at once. + // `finish` always runs from a microtask, after the loop parked on its + // first timer, so the timer is never absent here. + clearTimeout(timer) + wake?.() + } + void until.then(finish, finish) + while (!settled) { + drain() + await new Promise((resolve) => { + wake = resolve + timer = setTimeout(resolve, pollMs) + }) + wake = undefined + timer = undefined + } + drain() + })() + return { done } +} diff --git a/packages/jobs/jobs-local/src/ring.ts b/packages/jobs/jobs-local/src/ring.ts new file mode 100644 index 0000000000..fb4bcb3c5b --- /dev/null +++ b/packages/jobs/jobs-local/src/ring.ts @@ -0,0 +1,113 @@ +/** + * The bounded output ring behind one job: chunks at absolute byte offsets, + * head eviction that never moves an assigned offset, and non-consuming reads + * from any offset. + * @module @deepseek-ai/dsh-jobs-local/ring + */ + +import type { JobAppendOptions, JobChannel, JobOutputRead } from '@deepseek-ai/dsh-jobs' + +/** One retained ring entry; `bytes` caches the chunk's UTF-8 length. */ +interface RingChunk { + at: number + text: string + bytes: number + channel?: JobChannel + gapBefore?: true +} + +/** + * The UTF-8-safe tail of `text` no longer than `maxBytes`: the byte cut + * advances past continuation bytes so the surviving text never starts inside + * a code point. + * @param text - the oversized chunk text. + * @param maxBytes - positive byte budget for the surviving tail. + * @returns the surviving tail and its exact byte length. + */ +function utf8Tail(text: string, maxBytes: number): { text: string; bytes: number } { + const raw = Buffer.from(text, 'utf8') + let start = raw.length - maxBytes + // The loop bound proves the index is in range; the assertion only + // discharges noUncheckedIndexedAccess. + while (start < raw.length && ((raw[start] as number) & 0xC0) === 0x80) start += 1 + const tail = raw.subarray(start) + return { text: tail.toString('utf8'), bytes: tail.length } +} + +/** Offsets stay absolute across eviction: `earliest` only ever advances. */ +export class OutputRing { + /** Retained chunks in offset order. */ + private readonly chunks: RingChunk[] = [] + /** Sum of the retained chunks' byte lengths. */ + retainedBytes = 0 + /** Total UTF-8 bytes ever appended — the offset the next chunk starts at. */ + total = 0 + /** Offset of the oldest retained byte (equals {@link total} when nothing is retained). */ + earliest = 0 + + /** + * Append one chunk and trim the head to `cap`. + * @param text - the chunk text, exactly as produced. + * @param options - stream label and gap marker. + * @param cap - retention cap in UTF-8 bytes after this append. + * @returns false when the chunk was empty and nothing changed. + */ + append(text: string, options: JobAppendOptions | undefined, cap: number): boolean { + if (text.length === 0) return false + const bytes = Buffer.byteLength(text, 'utf8') + this.chunks.push({ + at: this.total, + text, + bytes, + ...options?.channel !== undefined ? { channel: options.channel } : {}, + ...options?.gapBefore !== undefined ? { gapBefore: options.gapBefore } : {}, + }) + this.total += bytes + this.retainedBytes += bytes + this.trim(cap) + return true + } + + /** + * Drop retained head chunks until the ring fits `cap`; a single oversized + * chunk keeps only its UTF-8-safe tail with a `gapBefore` marker. + * @param cap - retention cap in UTF-8 bytes. + */ + trim(cap: number): void { + while (this.retainedBytes > cap && this.chunks.length > 1) { + const dropped = this.chunks.shift() + /* v8 ignore next -- the length guard proves shift() returns a chunk; the check only satisfies noUncheckedIndexedAccess. */ + if (dropped === undefined) break + this.retainedBytes -= dropped.bytes + } + const single = this.chunks.length === 1 ? this.chunks[0] : undefined + if (single !== undefined && single.bytes > cap) { + const tail = utf8Tail(single.text, cap) + single.at += single.bytes - tail.bytes + single.text = tail.text + single.bytes = tail.bytes + single.gapBefore = true + this.retainedBytes = tail.bytes + } + this.earliest = this.chunks[0]?.at ?? this.total + } + + /** + * Retained chunks overlapping `[from, total)` as fresh wire chunks. + * @param from - absolute byte offset to read from. + * @returns the chunks, the resume offset, and whether bytes before them were evicted. + */ + readFrom(from: number): JobOutputRead { + const chunks = [] + for (const chunk of this.chunks) { + if (chunk.at + chunk.bytes <= from) continue + chunks.push({ + at: chunk.at, + text: chunk.text, + ...chunk.channel !== undefined ? { channel: chunk.channel } : {}, + ...chunk.gapBefore !== undefined ? { gapBefore: chunk.gapBefore } : {}, + }) + } + return { chunks, next: this.total, lossy: from < this.earliest } + } +} diff --git a/packages/jobs/jobs-local/tests/pump.spec.ts b/packages/jobs/jobs-local/tests/pump.spec.ts new file mode 100644 index 0000000000..31ee5d4d3f --- /dev/null +++ b/packages/jobs/jobs-local/tests/pump.spec.ts @@ -0,0 +1,111 @@ +import { describe, expect, it, vi } from 'vitest' +import type { JobAppendOptions, JobOutputSource } from '@deepseek-ai/dsh-jobs' +import { startPump } from '../src/pump.ts' + +/** Record every append so pump behavior is observable without a registry. */ +function sink() { + const appends: { text: string; options?: JobAppendOptions }[] = [] + return { + appends, + append: (text: string, options?: JobAppendOptions) => { appends.push({ text, ...options !== undefined ? { options } : {} }) }, + } +} + +/** A scripted source: each read() shifts the next scripted result. */ +function scriptedSource( + reads: { text: string; lossy?: boolean }[], + channel?: 'stdout' | 'stderr', +): { source: JobOutputSource; offsets: number[] } { + const offsets: number[] = [] + let offset = 0 + const source: JobOutputSource = { + ...channel !== undefined ? { channel } : {}, + read(fromByte) { + offsets.push(fromByte) + const next = reads.shift() ?? { text: '' } + offset += Buffer.byteLength(next.text, 'utf8') + return { text: next.text, nextOffset: offset, lossy: next.lossy ?? false } + }, + } + return { source, offsets } +} + +describe('startPump', () => { + it('copies labeled deltas at the poll cadence and resumes each source at its own offset', async () => { + vi.useFakeTimers() + try { + const { appends, append } = sink() + let settle!: () => void + const until = new Promise((resolve) => { settle = resolve }) + const out = scriptedSource([{ text: 'a' }, { text: 'bc' }], 'stdout') + const err = scriptedSource([{ text: '' }, { text: 'E' }], 'stderr') + const pump = startPump([out.source, err.source], append, 50, until) + + await vi.advanceTimersByTimeAsync(50) + settle() + await pump.done + expect(appends).toEqual([ + { text: 'a', options: { channel: 'stdout' } }, + { text: 'bc', options: { channel: 'stdout' } }, + { text: 'E', options: { channel: 'stderr' } }, + ]) + // Every read resumed from the previous nextOffset, never from a shared cursor. + expect(out.offsets).toEqual([0, 1, 3]) + expect(err.offsets).toEqual([0, 0, 1]) + // Settlement cleared the pending poll timer instead of leaving it pending. + expect(vi.getTimerCount()).toBe(0) + } finally { + vi.useRealTimers() + } + }) + + it('marks a lossy source read as a gap so observers see the discontinuity', async () => { + const { appends, append } = sink() + const { source } = scriptedSource([{ text: 'tail', lossy: true }]) + await startPump([source], append, 1, Promise.resolve()).done + expect(appends).toEqual([{ text: 'tail', options: { gapBefore: true } }]) + }) + + it('treats a rejected settlement as settlement and still drains the final bytes', async () => { + const { appends, append } = sink() + const { source } = scriptedSource([{ text: 'last' }]) + await startPump([source], append, 1, Promise.reject(new Error('producer broke'))).done + expect(appends).toEqual([{ text: 'last' }]) + }) + + it('subscribes to the settlement once and holds one poll timer however long the job runs', async () => { + vi.useFakeTimers() + try { + const { append } = sink() + let subscriptions = 0 + let settle!: () => void + const settled = new Promise((resolve) => { settle = resolve }) + const counting = (promise: Promise): Promise => ({ + then(onFulfilled?: (value: unknown) => unknown, onRejected?: (reason: unknown) => unknown) { + subscriptions += 1 + return counting(promise.then(onFulfilled, onRejected)) + }, + }) as unknown as Promise + const { source, offsets } = scriptedSource([]) + const pump = startPump([source], append, 50, counting(settled)) + + const rounds = 10_000 + await vi.advanceTimersByTimeAsync(50 * rounds) + expect(offsets).toHaveLength(rounds + 1) + expect(subscriptions).toBeLessThanOrEqual(2) + expect(vi.getTimerCount()).toBe(1) + + settle() + await pump.done + expect(offsets).toHaveLength(rounds + 2) + expect(vi.getTimerCount()).toBe(0) + } finally { + vi.useRealTimers() + } + }) + + it('rejects a non-positive poll interval before touching any source', () => { + const { append } = sink() + expect(() => startPump([], append, 0, Promise.resolve())).toThrow(/invalid pump pollMs/) + }) +}) diff --git a/packages/jobs/jobs-local/tests/ring.spec.ts b/packages/jobs/jobs-local/tests/ring.spec.ts new file mode 100644 index 0000000000..670e121d0b --- /dev/null +++ b/packages/jobs/jobs-local/tests/ring.spec.ts @@ -0,0 +1,65 @@ +import { describe, expect, it } from 'vitest' +import { OutputRing } from '../src/ring.ts' + +describe('OutputRing', () => { + it('assigns absolute offsets by UTF-8 byte length and drops empty chunks silently', () => { + const ring = new OutputRing() + expect(ring.append('', undefined, 1024)).toBe(false) + expect(ring.append('héllo', { channel: 'stdout' }, 1024)).toBe(true) + expect(ring.append('wörld', { channel: 'stderr', gapBefore: true }, 1024)).toBe(true) + expect(ring.total).toBe(12) + expect(ring.earliest).toBe(0) + expect(ring.retainedBytes).toBe(12) + expect(ring.readFrom(0)).toEqual({ + chunks: [ + { at: 0, text: 'héllo', channel: 'stdout' }, + { at: 6, text: 'wörld', channel: 'stderr', gapBefore: true }, + ], + next: 12, + lossy: false, + }) + }) + + it('reads from a foreign offset by returning the whole overlapping chunk', () => { + const ring = new OutputRing() + ring.append('abc', undefined, 1024) + ring.append('def', undefined, 1024) + expect(ring.readFrom(4).chunks).toEqual([{ at: 3, text: 'def' }]) + expect(ring.readFrom(6)).toEqual({ chunks: [], next: 6, lossy: false }) + }) + + it('trims whole head chunks past the cap and reports the evicted span as lossy', () => { + const ring = new OutputRing() + ring.append('aaaa', undefined, 8) + ring.append('bbbb', undefined, 8) + ring.append('cc', undefined, 8) + // 10 bytes retained > cap 8: the oldest whole chunk goes. + expect(ring.earliest).toBe(4) + expect(ring.retainedBytes).toBe(6) + expect(ring.readFrom(0)).toEqual({ chunks: [{ at: 4, text: 'bbbb' }, { at: 8, text: 'cc' }], next: 10, lossy: true }) + expect(ring.readFrom(4).lossy).toBe(false) + }) + + it('keeps a lone oversized chunk as a UTF-8-safe tail marked gapBefore', () => { + const ring = new OutputRing() + // 3 two-byte code points = 6 bytes; a 3-byte cap must not start inside a code point. + ring.append('ééé', undefined, 3) + expect(ring.retainedBytes).toBe(2) + expect(ring.earliest).toBe(4) + expect(ring.readFrom(0)).toEqual({ chunks: [{ at: 4, text: 'é', gapBefore: true }], next: 6, lossy: true }) + }) + + it('re-trims to a smaller cap on demand and leaves total untouched', () => { + const ring = new OutputRing() + ring.append('12345678', undefined, 64) + ring.append('abcdefgh', undefined, 64) + ring.trim(8) + expect(ring.total).toBe(16) + expect(ring.earliest).toBe(8) + expect(ring.readFrom(8)).toEqual({ chunks: [{ at: 8, text: 'abcdefgh' }], next: 16, lossy: false }) + // Trimming an empty ring is a no-op that keeps earliest at total. + const empty = new OutputRing() + empty.trim(1) + expect(empty.earliest).toBe(0) + }) +})