From 4260eec306c1d4c38a3f3837bdc2ad2e6ca991fd Mon Sep 17 00:00:00 2001 From: Menci Date: Thu, 1 Oct 2026 09:51:12 +0800 Subject: [PATCH 1/3] feat: publish streamed file writes atomically Accept streamed file bodies with storage backpressure. Node stages on the target filesystem and publishes by rename; reserved staging paths support exact-key cleanup after killed writers. R2 uses bounded multipart parts for generated unknown-length bodies and retains the prior object on failure. Restores the reviewed streaming foundation from its preserved archive. Verified 20 file-store tests, platform and deployment typechecks, and scoped lint; checked current Cloudflare API references. Preserve original errors and cleanup chains. --- .../__tests__/r2-file-store_test.ts | 72 ++++++++++++++- apps/platform-cloudflare/src/r2-file-store.ts | 57 +++++++++++- .../fixtures/concurrent-file-writers.ts | 26 ++++++ .../__tests__/fixtures/killed-file-writer.ts | 13 +++ .../__tests__/fs-file-store_test.ts | 87 ++++++++++++++++++- apps/platform-node/src/fs-file-store.ts | 87 +++++++++++++------ packages/platform/README.md | 32 +++++-- .../platform/__tests__/file-store_test.ts | 15 +++- packages/platform/src/file-store.ts | 10 ++- 9 files changed, 354 insertions(+), 45 deletions(-) create mode 100644 apps/platform-node/__tests__/fixtures/concurrent-file-writers.ts create mode 100644 apps/platform-node/__tests__/fixtures/killed-file-writer.ts diff --git a/apps/platform-cloudflare/__tests__/r2-file-store_test.ts b/apps/platform-cloudflare/__tests__/r2-file-store_test.ts index 9b5e800536..99ce9d388c 100644 --- a/apps/platform-cloudflare/__tests__/r2-file-store_test.ts +++ b/apps/platform-cloudflare/__tests__/r2-file-store_test.ts @@ -1,11 +1,29 @@ -import { test } from 'vitest'; +import { expect, test } from 'vitest'; -import { R2FileStore, type R2BucketLike } from '../src/r2-file-store.ts'; +import { R2FileStore, type R2BucketLike, type R2MultipartUploadLike, type R2UploadedPartLike } from '../src/r2-file-store.ts'; import { assertEquals } from '@floway-dev/test-utils'; class FakeR2Bucket implements R2BucketLike { store = new Map(); deleteCalls: string[][] = []; + parts: Uint8Array[] = []; + aborted = false; + + async createMultipartUpload(key: string): Promise { + return { + uploadPart: async (partNumber, value) => { + this.parts.push(value.slice()); + return { partNumber, etag: String(partNumber) }; + }, + complete: async (_parts: R2UploadedPartLike[]) => { + const bytes = new Uint8Array(this.parts.reduce((size, part) => size + part.byteLength, 0)); + let offset = 0; + for (const part of this.parts) { bytes.set(part, offset); offset += part.byteLength; } + this.store.set(key, bytes); + }, + abort: async () => { this.aborted = true; }, + }; + } async put(key: string, value: ReadableStream | ArrayBuffer | ArrayBufferView | string | null): Promise { if (!(value instanceof Uint8Array)) throw new Error('FakeR2Bucket only supports Uint8Array'); @@ -37,3 +55,53 @@ test('R2FileStore deletes exact keys in one R2 batch', async () => { assertEquals([...bucket.store.keys()], ['drop/ab']); assertEquals(bucket.deleteCalls, [['drop/a', 'missing']]); }); + +test('unknown-length streaming writes keep parts bounded and publish exact bytes at EOF', async () => { + const bucket = new FakeR2Bucket(); + const store = new R2FileStore(bucket); + const size = 5 * 1024 * 1024; + let finish!: () => void; + const gate = new Promise(resolve => { finish = resolve; }); + let chunk = 0; + const source = new ReadableStream({ + async pull(controller) { + if (chunk++ === 0) controller.enqueue(new Uint8Array(size + 3).fill(7)); + else { await gate; controller.close(); } + }, + }, { highWaterMark: 0 }); + const writing = store.put('stream', source); + await expect.poll(() => bucket.parts.length).toBe(1); + expect(bucket.parts[0]?.byteLength).toBe(size); + expect(await store.get('stream')).toBeNull(); + finish(); + await writing; + expect(bucket.parts.map(part => part.byteLength)).toEqual([size, 3]); + const stored = (await store.get('stream'))!; + expect(stored.byteLength).toBe(size + 3); + expect(stored.every(byte => byte === 7)).toBe(true); + expect(bucket.aborted).toBe(false); +}); + +test.each([0, 3])('a short stream of %i bytes uses a complete single put', async size => { + const bucket = new FakeR2Bucket(); + const store = new R2FileStore(bucket); + await store.put('short', new ReadableStream({ start(controller) { controller.enqueue(new Uint8Array(size)); controller.close(); } })); + expect(await store.get('short')).toEqual(new Uint8Array(size)); + expect(bucket.parts).toEqual([]); +}); + +test('a failed streamed upload leaves the previous object and preserves the original error', async () => { + const bucket = new FakeR2Bucket(); + const store = new R2FileStore(bucket); + await store.put('failed', new Uint8Array([9])); + const error = new Error('source interrupted'); + let chunk = 0; + await expect(store.put('failed', new ReadableStream({ + pull(controller) { + if (chunk++ === 0) controller.enqueue(new Uint8Array(5 * 1024 * 1024)); + else controller.error(error); + }, + }, { highWaterMark: 0 }))).rejects.toBe(error); + expect(bucket.aborted).toBe(true); + expect(await store.get('failed')).toEqual(new Uint8Array([9])); +}); diff --git a/apps/platform-cloudflare/src/r2-file-store.ts b/apps/platform-cloudflare/src/r2-file-store.ts index ab1c018e16..721fa22123 100644 --- a/apps/platform-cloudflare/src/r2-file-store.ts +++ b/apps/platform-cloudflare/src/r2-file-store.ts @@ -1,9 +1,17 @@ -import type { FileStore } from '@floway-dev/platform'; +import type { FileBody, FileStore } from '@floway-dev/platform'; + +export interface R2UploadedPartLike { partNumber: number; etag: string } +export interface R2MultipartUploadLike { + uploadPart(partNumber: number, value: Uint8Array): Promise; + complete(parts: R2UploadedPartLike[]): Promise; + abort(): Promise; +} export interface R2BucketLike { put(key: string, value: ReadableStream | ArrayBuffer | ArrayBufferView | string | null): Promise; get(key: string): Promise<{ arrayBuffer(): Promise } | null>; delete(keys: string | string[]): Promise; + createMultipartUpload(key: string): Promise; } // R2 caps both `list` and `delete` at 1000 keys per call. @@ -11,11 +19,54 @@ export interface R2BucketLike { // https://developers.cloudflare.com/r2/api/workers/workers-api-reference/#delete const R2_BATCH_LIMIT = 1000; +// Generated streams have no known content length, so R2's single put cannot consume +// them. Multipart parts must be at least 5 MiB except for the last part. +// https://developers.cloudflare.com/r2/objects/upload-objects/#multipart-upload +const R2_PART_BYTES = 5 * 1024 * 1024; + export class R2FileStore implements FileStore { constructor(private readonly bucket: R2BucketLike) {} - async put(key: string, body: Uint8Array): Promise { - await this.bucket.put(key, body); + async put(key: string, body: FileBody): Promise { + if (body instanceof Uint8Array) { await this.bucket.put(key, body); return; } + const reader = body.getReader(); + let upload: R2MultipartUploadLike | undefined; + const parts: R2UploadedPartLike[] = []; + let buffer = new Uint8Array(R2_PART_BYTES); + let length = 0; + const flush = async () => { + upload ??= await this.bucket.createMultipartUpload(key); + parts.push(await upload.uploadPart(parts.length + 1, length === buffer.byteLength ? buffer : buffer.subarray(0, length))); + buffer = new Uint8Array(R2_PART_BYTES); + length = 0; + }; + try { + for (;;) { + const { done, value } = await reader.read(); + if (done) break; + let offset = 0; + while (offset < value.byteLength) { + const count = Math.min(buffer.byteLength - length, value.byteLength - offset); + buffer.set(value.subarray(offset, offset + count), length); + length += count; + offset += count; + if (length === buffer.byteLength) await flush(); + } + } + if (upload === undefined) await this.bucket.put(key, buffer.subarray(0, length)); + else { + if (length > 0) await flush(); + await upload.complete(parts); + } + } catch (error) { + const errors: unknown[] = [error]; + try { await reader.cancel(error); } catch (cleanupError) { if (cleanupError !== error) errors.push(cleanupError); } + if (upload !== undefined) { + try { await upload.abort(); } catch (cleanupError) { errors.push(cleanupError); } + } + if (errors.length > 1) throw new AggregateError(errors, 'R2 upload failed and cleanup also failed', { cause: error }); + throw error; + } finally { reader.releaseLock(); } } async get(key: string): Promise { diff --git a/apps/platform-node/__tests__/fixtures/concurrent-file-writers.ts b/apps/platform-node/__tests__/fixtures/concurrent-file-writers.ts new file mode 100644 index 0000000000..ad7541ceb6 --- /dev/null +++ b/apps/platform-node/__tests__/fixtures/concurrent-file-writers.ts @@ -0,0 +1,26 @@ +import fsPromises from 'node:fs/promises'; +import { syncBuiltinESMExports } from 'node:module'; + +const originalOpen = fsPromises.open; +let pause = true; +let staged!: () => void; +const started = new Promise(resolve => { staged = resolve; }); +let resume!: () => void; +const gate = new Promise(resolve => { resume = resolve; }); +fsPromises.open = async (...args: Parameters) => { + if (pause) { pause = false; staged(); await gate; } + return await originalOpen(...args); +}; +syncBuiltinESMExports(); +const { FsFileStore } = await import('../../src/fs-file-store.ts'); +const store = new FsFileStore(process.argv[2]!); +try { + const first = store.put('file', new Uint8Array([1])); + await started; + await store.put('file', new Uint8Array([2])); + resume(); + await first; +} finally { + fsPromises.open = originalOpen; + syncBuiltinESMExports(); +} diff --git a/apps/platform-node/__tests__/fixtures/killed-file-writer.ts b/apps/platform-node/__tests__/fixtures/killed-file-writer.ts new file mode 100644 index 0000000000..88b584ca84 --- /dev/null +++ b/apps/platform-node/__tests__/fixtures/killed-file-writer.ts @@ -0,0 +1,13 @@ +import { FsFileStore } from '../../src/fs-file-store.ts'; + +const store = new FsFileStore(process.argv[2]!); +let chunk = 0; +await store.put('file', new ReadableStream({ + async pull(controller) { + if (chunk++ === 0) controller.enqueue(new Uint8Array([1, 2])); + else { + process.send!('staged'); + await new Promise(() => {}); + } + }, +}, { highWaterMark: 0 })); diff --git a/apps/platform-node/__tests__/fs-file-store_test.ts b/apps/platform-node/__tests__/fs-file-store_test.ts index 34d1b2e114..060390c4f7 100644 --- a/apps/platform-node/__tests__/fs-file-store_test.ts +++ b/apps/platform-node/__tests__/fs-file-store_test.ts @@ -1,8 +1,11 @@ -import { mkdtemp, rm } from 'node:fs/promises'; +import { spawn } from 'node:child_process'; +import { once } from 'node:events'; +import { mkdtemp, readdir, rm, stat } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; +import { fileURLToPath } from 'node:url'; -import { test } from 'vitest'; +import { expect, test } from 'vitest'; import { FsFileStore } from '../src/fs-file-store.ts'; import { assertEquals } from '@floway-dev/test-utils'; @@ -47,3 +50,83 @@ test('put creates intermediate directories', () => withTempRoot(async root => { const read = await store.get('deeply/nested/path/file.bin'); assertEquals(read, new Uint8Array([42])); })); + +test('a streamed replacement stays private until EOF and publishes complete binary bytes', () => withTempRoot(async root => { + const store = new FsFileStore(root); + await store.put('file', new Uint8Array([9])); + let finish!: () => void; + const gate = new Promise(resolve => { finish = resolve; }); + let chunk = 0; + const writing = store.put('file', new ReadableStream({ + async pull(controller) { + if (chunk++ === 0) controller.enqueue(new Uint8Array([1, 2])); + else { await gate; controller.enqueue(new Uint8Array([3, 4])); controller.close(); } + }, + }, { highWaterMark: 0 })); + await expect.poll(() => chunk).toBe(2); + expect(await store.get('file')).toEqual(new Uint8Array([9])); + finish(); + await writing; + expect(await store.get('file')).toEqual(new Uint8Array([1, 2, 3, 4])); + expect(await readdir(root)).toEqual(['.floway-staging', 'file']); + const directories = await readdir(join(root, '.floway-staging')); + expect(directories).toHaveLength(1); + expect(await readdir(join(root, '.floway-staging', directories[0]!))).toEqual([]); +})); + +test('source failure removes staging data without replacing the prior file', () => withTempRoot(async root => { + const store = new FsFileStore(root); + await store.put('file', new Uint8Array([9])); + const error = new Error('source interrupted'); + let chunk = 0; + await expect(store.put('file', new ReadableStream({ + pull(controller) { + if (chunk++ === 0) controller.enqueue(new Uint8Array([1, 2])); + else controller.error(error); + }, + }, { highWaterMark: 0 }))).rejects.toBe(error); + expect(await store.get('file')).toEqual(new Uint8Array([9])); + expect(await readdir(root)).toEqual(['.floway-staging', 'file']); + const directories = await readdir(join(root, '.floway-staging')); + expect(directories).toHaveLength(1); + expect(await readdir(join(root, '.floway-staging', directories[0]!))).toEqual([]); +})); + +test('deleteKeys collects an upload whose writer was killed without matching another key by name', () => withTempRoot(async root => { + const store = new FsFileStore(root); + const namedLikeStaging = 'file.00000000-0000-4000-8000-000000000000.tmp'; + await store.put(namedLikeStaging, new Uint8Array([9])); + const child = spawn(process.execPath, ['--import', import.meta.resolve('tsx'), fileURLToPath(new URL('./fixtures/killed-file-writer.ts', import.meta.url)), root], { stdio: ['ignore', 'pipe', 'pipe', 'ipc'] }); + const exited = once(child, 'exit'); + try { + await Promise.race([once(child, 'message'), exited.then(([code]) => { throw new Error(`Writer exited before staging: ${String(code)}`); })]); + expect((await stat(join(root, '.floway-staging'))).dev).toBe((await stat(root)).dev); + child.kill('SIGKILL'); + await exited; + await store.deleteKeys(['file']); + expect(await store.get('file')).toBeNull(); + expect(await store.get(namedLikeStaging)).toEqual(new Uint8Array([9])); + expect(await readdir(root)).toEqual(['.floway-staging', namedLikeStaging]); + expect(await readdir(join(root, '.floway-staging'))).toHaveLength(1); + } finally { + if (child.exitCode === null && child.signalCode === null) { child.kill('SIGKILL'); await exited; } + } +})); + +test.each(['.floway-staging', '.FLOWAY-STAGING', '.Floway-Staging'])('Node reserves private segment %s while retaining published key paths', segment => withTempRoot(async root => { + const store = new FsFileStore(root); + await expect(store.put(`nested/${segment}/key`, new Uint8Array([1]))).rejects.toThrow('reserved staging segment'); + await expect(store.get(`${segment}/key`)).rejects.toThrow('reserved staging segment'); + await expect(store.deleteKeys([`${segment}/key`])).rejects.toThrow('reserved staging segment'); + await store.put('nested/key', new Uint8Array([7])); + expect(await readdir(join(root, 'nested'))).toEqual(['.floway-staging', 'key']); +})); + +test('a finishing writer leaves the shared staging directory available to a writer opening its file', () => withTempRoot(async root => { + const child = spawn(process.execPath, ['--import', import.meta.resolve('tsx'), fileURLToPath(new URL('./fixtures/concurrent-file-writers.ts', import.meta.url)), root], { stdio: ['ignore', 'pipe', 'pipe'] }); + let stderr = ''; + child.stderr.on('data', chunk => { stderr += String(chunk); }); + const [code] = await once(child, 'exit'); + expect({ code, stderr }).toEqual({ code: 0, stderr: '' }); + expect(await new FsFileStore(root).get('file')).toEqual(new Uint8Array([1])); +})); diff --git a/apps/platform-node/src/fs-file-store.ts b/apps/platform-node/src/fs-file-store.ts index 7560d16202..42ec8f3fa6 100644 --- a/apps/platform-node/src/fs-file-store.ts +++ b/apps/platform-node/src/fs-file-store.ts @@ -1,38 +1,58 @@ +import { createHash, randomUUID } from 'node:crypto'; import { mkdirSync } from 'node:fs'; -import { mkdir, readFile, rm, writeFile } from 'node:fs/promises'; -import { dirname, isAbsolute, resolve, sep } from 'node:path'; - -import type { FileStore } from '@floway-dev/platform'; - -// Filesystem-backed FileStore. Every key resolves to a path under `root`. -// Keys use forward-slash POSIX separators (matching R2's surface) and are -// translated to native path segments on the way in/out so the same key reads -// identically on Windows and POSIX hosts. -// -// Threat model: `root` (`FLOWAY_FILES_DIR`) is gateway-trusted. Everything -// dumped here is data the gateway already holds in its database (API keys, -// upstream credentials, request payloads); fs-level access to this directory -// is already equivalent to gateway compromise. We deliberately do not mode -// 0o600 / 0o700 the writes — bodies are stored verbatim and the OS-level -// confidentiality boundary belongs to the operator (umask, mount perms, -// dedicated user). The dashboard redacts sensitive headers at render time -// for human display, but the on-disk record stays untouched so an operator -// can replay or diff against upstream byte-for-byte. +import { mkdir, open, readFile, rename, rm } from 'node:fs/promises'; +import { basename, dirname, isAbsolute, join, relative, resolve, sep } from 'node:path'; + +import type { FileBody, FileStore } from '@floway-dev/platform'; + +const STAGING_SEGMENT = '.floway-staging'; +const isStagingSegment = (segment: string): boolean => segment.toLowerCase() === STAGING_SEGMENT; + +// Keys use POSIX separators so filesystem and object-storage deployments agree. +// The operator's umask, mount permissions and service account own confidentiality; +// a file contains the writer's already-encoded bytes, including pipeline redaction. +// Access to this directory exposes recorded request contents and is equivalent to +// access to the gateway's stored configuration. We do not add a second permission policy. export class FsFileStore implements FileStore { private readonly root: string; constructor(root: string) { - // Resolve once so `pathFor` can verify resolved paths still live under it. this.root = resolve(root); - // Ensure the root exists so the first put() doesn't race against a missing - // directory and so tests / fresh deploys see a consistent structure. mkdirSync(this.root, { recursive: true }); } - async put(key: string, body: Uint8Array): Promise { - const path = this.pathFor(key); - await mkdir(dirname(path), { recursive: true }); - await writeFile(path, body); + async put(key: string, body: FileBody): Promise { + const reader = body instanceof Uint8Array ? null : body.getReader(); + let staging: string | undefined; + try { + const path = this.pathFor(key); + await mkdir(dirname(path), { recursive: true }); + const stagingDirectory = this.stagingFor(path); + await mkdir(stagingDirectory, { recursive: true }); + staging = join(stagingDirectory, randomUUID()); + { + await using file = await open(staging, 'wx'); + if (body instanceof Uint8Array) await file.writeFile(body); + else { + for (;;) { + const { done, value } = await reader!.read(); + if (done) break; + await file.writeFile(value); + } + } + } + await rename(staging, path); + } catch (error) { + const errors: unknown[] = [error]; + if (reader !== null) { + try { await reader.cancel(error); } catch (cleanupError) { if (cleanupError !== error) errors.push(cleanupError); } + } + if (staging !== undefined) { + try { await rm(staging, { force: true }); } catch (cleanupError) { errors.push(cleanupError); } + } + if (errors.length > 1) throw new AggregateError(errors, 'File write failed and cleanup also failed', { cause: error }); + throw error; + } finally { reader?.releaseLock(); } } async get(key: string): Promise { @@ -45,7 +65,18 @@ export class FsFileStore implements FileStore { } async deleteKeys(keys: readonly string[]): Promise { - await Promise.all(keys.map(async key => await rm(this.pathFor(key), { force: true }))); + await Promise.all(keys.map(async key => { + const path = this.pathFor(key); + const staging = this.stagingFor(path); + await Promise.all([rm(path, { force: true }), rm(staging, { recursive: true, force: true })]); + })); + } + + // Staging lives on the target's filesystem, including separately mounted key + // prefixes. Its reserved namespace lets key-based orphan cleanup collect a + // killed writer without interpreting another public key as a temporary file. + private stagingFor(path: string): string { + return join(dirname(path), STAGING_SEGMENT, createHash('sha256').update(basename(path)).digest('hex')); } // Resolve a key against `root` and reject paths that escape it. Even though @@ -53,11 +84,13 @@ export class FsFileStore implements FileStore { // to scrub user-controlled segments and a `..`-laden key would otherwise // walk to arbitrary host paths under R2 it would simply be a strange key. private pathFor(key: string): string { + if (key.split('/').some(isStagingSegment)) throw new Error(`FsFileStore: reserved staging segment (${key})`); if (isAbsolute(key)) throw new Error(`FsFileStore: absolute keys are not supported (${key})`); const path = resolve(this.root, ...key.split('/')); if (path !== this.root && !path.startsWith(this.root + sep)) { throw new Error(`FsFileStore: key escapes root (${key})`); } + if (relative(this.root, path).split(sep).some(isStagingSegment)) throw new Error(`FsFileStore: reserved staging segment (${key})`); return path; } } diff --git a/packages/platform/README.md b/packages/platform/README.md index 829472636e..509298c14a 100644 --- a/packages/platform/README.md +++ b/packages/platform/README.md @@ -4,12 +4,30 @@ This package defines portable runtime services. Implementations live in the platform applications; runtime composition installs them through the exported initializers. -## Stored files +## Durable file writes -`FileStore.put(key, bytes)` stores a complete `Uint8Array`. `get` returns the -stored bytes or null, and `deleteKeys` removes exact keys. Node uses the -configured filesystem directory, Cloudflare uses R2, and the in-memory -implementation stores byte copies. +`FileStore.put(key, body)` accepts bytes or a `ReadableStream`. +It consumes the stream with storage backpressure and publishes the complete +object at EOF. Until then, readers see the previous value or a missing key. +Failures preserve the prior object and expose the original error; cleanup +failures retain it in the error chain. `get` still returns complete bytes. + +Node stages each write inside the target parent’s reserved `.floway-staging` +namespace and atomically renames it after closing. This preserves the published +key paths and the target filesystem when a key prefix is a separate mount. +Node keys cannot contain that private segment in any ASCII case spelling, +including components of the normalized filesystem path. `deleteKeys` removes the exact +key’s unfinished writes as well as its published file, so retention cleanup +also collects a writer killed before EOF. Public keys resembling UUID temporary +filenames remain ordinary keys. Empty staging directories stay available to +concurrent writers until key deletion; the common private directory stays in +place like other parent directories. + +Cloudflare uses bounded 5 MiB multipart parts for unknown-length +streams, since R2's single put requires a known length. Short and empty streams +use a single byte put. A storage/source failure cancels the source and cleans +up unfinished multipart state; live-view degradation does not affect this path. +The in-memory implementation collects bytes as part of its storage semantics. ## Transient run streams @@ -27,8 +45,8 @@ A `LogStream` contains bytes, independent of any event format: the stream ended; cancellation or expiry interrupts the read. `LOG_STREAM_IDLE_MS` is the shared idle lease. Append activity, including an -empty heartbeat, and read progress renew it. To keep a quiet active stream open, its writer sends empty -appends while work is pending; a completed or abandoned stream is +empty heartbeat, and read progress renew it. A quiet active writer sends empty +appends while upstream work is pending; a completed or abandoned stream is reclaimed after idle activity stops. Existing handles then raise `LogStreamExpiredError`. Expiration affects transient live viewing; the durable artifact is a separate writer-owned path. diff --git a/packages/platform/__tests__/file-store_test.ts b/packages/platform/__tests__/file-store_test.ts index f94fd61073..9d93c8011c 100644 --- a/packages/platform/__tests__/file-store_test.ts +++ b/packages/platform/__tests__/file-store_test.ts @@ -1,4 +1,4 @@ -import { test } from 'vitest'; +import { expect, test } from 'vitest'; import { getFileStore, initFileStore, MemoryFileStore } from '../src/file-store.ts'; import { assertEquals } from '@floway-dev/test-utils'; @@ -35,3 +35,16 @@ test('MemoryFileStore deletes exact keys without treating them as prefixes', asy assertEquals(await store.get('drop/a'), null); assertEquals(await store.get('drop/ab'), new Uint8Array([2])); }); + +test('MemoryFileStore accepts streamed bytes and publishes only at completion', async () => { + const store = new MemoryFileStore(); + const source = new TransformStream(); + const writing = store.put('stream', source.readable); + const writer = source.writable.getWriter(); + await writer.write(new Uint8Array([1, 2])); + expect(await store.get('stream')).toBeNull(); + await writer.write(new Uint8Array([3])); + await writer.close(); + await writing; + expect(await store.get('stream')).toEqual(new Uint8Array([1, 2, 3])); +}); diff --git a/packages/platform/src/file-store.ts b/packages/platform/src/file-store.ts index 13bb33c2ce..ff4a45d995 100644 --- a/packages/platform/src/file-store.ts +++ b/packages/platform/src/file-store.ts @@ -1,5 +1,8 @@ +export type FileBody = Uint8Array | ReadableStream; + export interface FileStore { - put(key: string, body: Uint8Array): Promise; + /** Consumes the body and publishes the complete object atomically at EOF. */ + put(key: string, body: FileBody): Promise; get(key: string): Promise; deleteKeys(keys: readonly string[]): Promise; } @@ -18,8 +21,9 @@ export const getFileStore = (): FileStore => { export class MemoryFileStore implements FileStore { private readonly files = new Map(); - async put(key: string, body: Uint8Array): Promise { - this.files.set(key, body.slice()); + async put(key: string, body: FileBody): Promise { + const bytes = body instanceof Uint8Array ? body.slice() : new Uint8Array(await new Response(body).arrayBuffer()); + this.files.set(key, bytes); } async get(key: string): Promise { From 2a21641069e58316166a5ccd3ffcd566e0a8ca13 Mon Sep 17 00:00:00 2001 From: Menci Date: Thu, 1 Oct 2026 10:02:39 +0800 Subject: [PATCH 2/3] feat: record pipeline runs as live durable byte streams Stream one ordered NDJSON artifact to durable storage and transient position-addressed readers. Keep metadata publication behind complete file writes, renew active leases, and preserve durable recording when live appends fail. Expose authenticated attach-only byte reads without active-run UI changes. Abort failed durable readers, stop heartbeat races, and wait for artifact publication before clean live EOF; cover lifecycle, backpressure, ownership, retries and retention with focused tests. --- README.md | 21 +- .../__tests__/control-plane/dump_test.ts | 46 ++- .../gateway/__tests__/dump/run-sink_test.ts | 266 +++++++++++++++++- .../gateway/__tests__/dump/test-fixtures.ts | 10 +- packages/gateway/__tests__/dump/write-run.ts | 9 + .../gateway/__tests__/repo/dump-store_test.ts | 100 +++++-- .../scheduled/expiration-sweeps_test.ts | 9 +- packages/gateway/__tests__/scheduled_test.ts | 3 +- .../__tests__/test-utils/log-stream.ts | 43 +++ packages/gateway/__tests__/vitest.setup.ts | 6 +- packages/gateway/src/control-plane/dump.ts | 13 +- packages/gateway/src/dump/run-sink.ts | 222 +++++++++------ packages/gateway/src/dump/store-contract.ts | 15 +- packages/gateway/src/dump/types.ts | 7 +- packages/gateway/src/repo/dump-store.ts | 52 ++-- packages/gateway/src/shared/gzip.ts | 3 + 16 files changed, 674 insertions(+), 151 deletions(-) create mode 100644 packages/gateway/__tests__/dump/write-run.ts create mode 100644 packages/gateway/__tests__/test-utils/log-stream.ts diff --git a/README.md b/README.md index cd807e6a89..3f27a9ed20 100644 --- a/README.md +++ b/README.md @@ -206,11 +206,22 @@ stage throws. Client JSON serialization runs inside settlement and publishes UTF while retaining the canonical content for inspection. Streaming quantities settle at the deferred exit. Explicit protocol and transport errors take precedence over the settlement's generic failure marker. -Run dumps encode stage facts, logs and protocol frames into one object space. Closing waits -for owned readers and deferred outcomes before storing the completed NDJSON record and -publishing its metadata. The Collected view selects the recorded client stream and uses its -`stream.end` event to establish recording completion. LogStream infrastructure is available -on both platforms; live recording and its business readers remain outside these three PRs. +Run dumps encode stage facts, logs and protocol frames into one object space. The live +recording layer streams those encoded NDJSON bytes into durable storage with backpressure +and appends the same bytes to a temporary LogStream. Live appends retry at the same byte +offset; a persistently unavailable live stream leaves durable recording active. Quiet runs +renew their temporary-stream and staged-file leases. Closing waits for owned readers and +deferred outcomes before publishing completed metadata. The live stream ends only after +the durable artifact is published; an encoding or durable-write failure aborts its reader +and stops its leases while retaining the original error. Run IDs are assigned at entry; +listing remains ordered by completion time and ID. + +The authenticated `GET /api/dump/keys/:keyId/records/:recordId/live?offset=0` endpoint attaches +to an existing per-key stream and uses the platform's framed byte protocol. The Collected +view selects the recorded client stream and uses its `stream.end` event to establish +recording completion. Active-run discovery, reconnect/list semantics and live dashboard +interactions remain separate design work; completed records and their stage viewer retain +the same run-only format. Pipeline endpoints are assembled from individual stage modules. Each stage owns its request and response contract; a large stage keeps local helpers in the same diff --git a/packages/gateway/__tests__/control-plane/dump_test.ts b/packages/gateway/__tests__/control-plane/dump_test.ts index fd8e6ca65a..830113342b 100644 --- a/packages/gateway/__tests__/control-plane/dump_test.ts +++ b/packages/gateway/__tests__/control-plane/dump_test.ts @@ -1,10 +1,13 @@ -import { test } from 'vitest'; +import { expect, test, vi } from 'vitest'; import { initDumpBroker, initDumpStore } from '../../src/dump/registry.ts'; +import { runStreamId } from '../../src/dump/run-sink.ts'; import type { DumpStore } from '../../src/dump/store-contract.ts'; import type { DumpMetadata, DumpRecord, StoredDumpRecord } from '../../src/dump/types.ts'; import { fakeMeta as baseFakeMeta, fakeRunRecord, installDumpStubs } from '../dump/test-fixtures.ts'; import { requestApp, setupAppTest } from '../test-utils/app.ts'; +import { testLogStreamStore } from '../test-utils/log-stream.ts'; +import { initLogStreamStore, readLogStream } from '@floway-dev/platform'; import { assertEquals, assertExists } from '@floway-dev/test-utils'; const fakeMeta = (id: string, completedAt: number): DumpMetadata => @@ -255,3 +258,44 @@ test('GET /api/dump/keys/:keyId/stream delivers a frame appended between snapsho // it. assertEquals((parseSseFrame(appended).data as { id: string }).id, '01HZZ0000000000000000000A2'); }); + +test('live record reads are framed, seekable, key-scoped and never create an unknown stream', async () => { + const { repo, apiKey } = await setupAppTest(); + await repo.apiKeys.save({ ...apiKey, dumpRetentionSeconds: 3600 }); + const store = testLogStreamStore(); + initLogStreamStore(store); + const first = new TextEncoder().encode('{"before":"中"}\n'); + const second = new TextEncoder().encode('{"after":"β"}\n'); + const stream = await store.open(runStreamId(apiKey.id, 'active')); + await stream.append(0, first); await stream.append(first.byteLength, second); await stream.end(); + const open = vi.spyOn(store, 'open'); + const response = await requestApp(`/api/dump/keys/${apiKey.id}/records/active/live?offset=${first.byteLength}`, { headers: { 'x-api-key': apiKey.key } }); + expect(response.status).toBe(200); + const chunks: Uint8Array[] = []; + for await (const bytes of readLogStream(response)) chunks.push(bytes); + expect(chunks.map(bytes => new TextDecoder().decode(bytes)).join('')).toBe(new TextDecoder().decode(second)); + const missing = await requestApp(`/api/dump/keys/${apiKey.id}/records/absent/live`, { headers: { 'x-api-key': apiKey.key } }); + expect(missing.status).toBe(404); + expect(open).not.toHaveBeenCalled(); + const invalid = await requestApp(`/api/dump/keys/${apiKey.id}/records/active/live?offset=-1`, { headers: { 'x-api-key': apiKey.key } }); + expect(invalid.status).toBe(400); + await store.open(runStreamId('foreign-key', 'foreign')); + const foreign = await requestApp(`/api/dump/keys/${apiKey.id}/records/foreign/live`, { headers: { 'x-api-key': apiKey.key } }); + expect(foreign.status).toBe(404); +}); + +test('live readers require authentication, key ownership and enabled retention before attaching', async () => { + const { repo, apiKey, adminSession } = await setupAppTest(); + await repo.apiKeys.save({ ...apiKey, dumpRetentionSeconds: 3600 }); + const store = testLogStreamStore(); + const stream = await store.open(runStreamId(apiKey.id, 'private')); + await stream.end(); + initLogStreamStore(store); + const get = vi.spyOn(store, 'get'); + const path = `/api/dump/keys/${apiKey.id}/records/private/live`; + expect((await requestApp(path, {})).status).toBe(401); + expect((await requestApp(path, { headers: { 'x-floway-session': adminSession } })).status).toBe(404); + await repo.apiKeys.save({ ...apiKey, dumpRetentionSeconds: null }); + expect((await requestApp(path, { headers: { 'x-api-key': apiKey.key } })).status).toBe(404); + expect(get).not.toHaveBeenCalled(); +}); diff --git a/packages/gateway/__tests__/dump/run-sink_test.ts b/packages/gateway/__tests__/dump/run-sink_test.ts index 5a11c738cb..7fbf239dd5 100644 --- a/packages/gateway/__tests__/dump/run-sink_test.ts +++ b/packages/gateway/__tests__/dump/run-sink_test.ts @@ -1,12 +1,14 @@ -import { expect, test } from 'vitest'; +import { expect, test, vi } from 'vitest'; import { installDumpStubs } from './test-fixtures.ts'; import { initDumpBroker, initDumpStore } from '../../src/dump/registry.ts'; -import { openRunDump } from '../../src/dump/run-sink.ts'; +import { openRunDump, runStreamId } from '../../src/dump/run-sink.ts'; import type { StoredDumpRecord } from '../../src/dump/types.ts'; import type { ApiKey } from '../../src/repo/types.ts'; import { flushBackground, trackBackground } from '../test-utils/background-tracker.ts'; +import { testLogStreamStore } from '../test-utils/log-stream.ts'; import { compose, defineStage, defer, move, run, type DumpEvent, type Event } from '@floway-dev/pipeline'; +import { getLogStreamStore, initLogStreamStore, LOG_STREAM_IDLE_MS } from '@floway-dev/platform'; import { assertEquals } from '@floway-dev/test-utils'; const apiKey = (dumpRetentionSeconds: number | null): ApiKey => ({ @@ -184,7 +186,120 @@ test('a run reads the stable timing state at completion alongside its stage hist assertEquals(lines(record).some(event => event.type === 'stage.entered'), true); }); +test('live bytes are visible before completion and equal the durable NDJSON artifact', async () => { + initLogStreamStore(testLogStreamStore()); + const stubs = installDumpStubs(initDumpStore, initDumpBroker); + const dump = openRunDump(apiKey(3600), turn, trackBackground, false, { upstreamCallStartedAt: null, firstOutputTokenAt: null })!; + await dump.sink({ type: 'stage.entered', stageId: 1, name: 'open', parentStageId: null, facts: move({ input: '中' }) }); + const live = await getLogStreamStore().get(runStreamId(apiKey(3600).id, dump.id)); + const reader = live!.read(0, new AbortController().signal)[Symbol.asyncIterator](); + const first = await reader.next(); + expect(new TextDecoder().decode(first.value)).toContain('stage.entered'); + expect(stubs.stored).toHaveLength(0); + dump.finalize(200, 0); + await flushBackground(); + const rest: Uint8Array[] = [first.value!]; + for (;;) { const item = await reader.next(); if (item.done) break; rest.push(item.value); } + expect(rest.map(bytes => new TextDecoder().decode(bytes)).join('')).toBe(ndjson(runRecordOf(stubs.stored[0]))); + expect(stubs.stored[0]!.record.meta.id).toBe(dump.id); +}); + +test('retries a lost live acknowledgement at the same offset without duplicating bytes', async () => { + const underlying = testLogStreamStore(); + const positions: number[] = []; + initLogStreamStore({ + get: id => underlying.get(id), + open: async id => { + const stream = await underlying.open(id); + return { + ...stream, append: async (offset, bytes) => { + positions.push(offset); + await stream.append(offset, bytes); + if (positions.length === 1) throw new Error('lost acknowledgement'); + }, + }; + }, + }); + const stubs = installDumpStubs(initDumpStore, initDumpBroker); + const dump = openRunDump(apiKey(3600), turn, trackBackground, false, { upstreamCallStartedAt: null, firstOutputTokenAt: null })!; + await dump.sink({ type: 'stage.entered', stageId: 1, name: 'retry', parentStageId: null, facts: move({ input: 'x' }) }); + dump.finalize(200, 0); + await flushBackground(); + expect(positions).toEqual([0, 0]); + const stream = await underlying.get(runStreamId(apiKey(3600).id, dump.id)); + const text: string[] = []; + for await (const bytes of stream!.read(0, new AbortController().signal)) text.push(new TextDecoder().decode(bytes)); + expect(text.join('')).toBe(ndjson(runRecordOf(stubs.stored[0]))); +}); + +test('a repeatedly failed live stream still stores the complete durable artifact', async () => { + const original = new Error('live transport unavailable'); + const append = vi.fn(async () => { throw original; }); + initLogStreamStore({ + open: async () => ({ append, end: async () => {}, read: () => (async function* () {})() }), + get: async () => null, + }); + const reported = vi.spyOn(console, 'error').mockImplementation(() => {}); + const stubs = installDumpStubs(initDumpStore, initDumpBroker); + try { + const dump = openRunDump(apiKey(3600), turn, trackBackground, false, { upstreamCallStartedAt: null, firstOutputTokenAt: null })!; + const { drain } = await run(pipeline, move({ 'in.text': 'hello' }), { dump: dump.sink }); + dump.afterRun(drain); + dump.finalize(200, 0); + await flushBackground(); + expect(append).toHaveBeenCalledTimes(3); + expect(lines(runRecordOf(stubs.stored[0])).filter(event => event.type === 'stage.entered')).toHaveLength(2); + expect(reported.mock.calls.some(call => call[1] === original)).toBe(true); + } finally { reported.mockRestore(); initLogStreamStore(testLogStreamStore()); } +}); + +test('durable backpressure blocks a producer behind an acquired native reader', async () => { + const underlying = testLogStreamStore(); + const append = vi.fn(); + initLogStreamStore({ + get: id => underlying.get(id), open: async id => { + const stream = await underlying.open(id); + return { ...stream, append: async (offset, bytes) => { append(); await stream.append(offset, bytes); } }; + }, + }); + const stubs = installDumpStubs(initDumpStore, initDumpBroker); + const gate = Promise.withResolvers(); + const acquired = Promise.withResolvers(); + initDumpStore({ + ...stubs.store, putRun: async (keyId, writing) => { + const reader = writing.events.getReader(); + acquired.resolve(); + await gate.promise; + try { + await stubs.store.putRun(keyId, { + ...writing, events: new ReadableStream({ + async pull(controller) { + const { value, done } = await reader.read(); + if (done) controller.close(); else controller.enqueue(value); + }, + }), + }); + } finally { reader.releaseLock(); } + }, + }); + const dump = openRunDump(apiKey(3600), turn, trackBackground, false, { upstreamCallStartedAt: null, firstOutputTokenAt: null })!; + await acquired.promise; + let wrote = false; + const write = dump.sink({ type: 'stage.entered', stageId: 1, name: 'blocked', parentStageId: null, facts: move({}) }).then(() => { wrote = true; }); + await new Promise(resolve => setImmediate(resolve)); + expect(wrote).toBe(false); + expect(append).not.toHaveBeenCalled(); + gate.resolve(); + await write; + expect(wrote).toBe(true); + expect(append).toHaveBeenCalledOnce(); + dump.finalize(200, 0); + await flushBackground(); + expect(stubs.stored).toHaveLength(1); +}); + test('closing waits for deferred outcomes and persists their settlement event', async () => { + initLogStreamStore(testLogStreamStore()); const stubs = installDumpStubs(initDumpStore, initDumpBroker); const result = Promise.withResolvers(); const deferred = defer(result.promise); @@ -202,3 +317,150 @@ test('closing waits for deferred outcomes and persists their settlement event', expect(lines(runRecordOf(stubs.stored[0])).some(event => event.type === 'deferred.settled')).toBe(true); expect(ndjson(runRecordOf(stubs.stored[0]))).toContain('late value'); }); + +test('heartbeat append waits for the in-flight event and uses its resulting byte offset', async () => { + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] }); + const underlying = testLogStreamStore(); + const entered = Promise.withResolvers(); + const release = Promise.withResolvers(); + const positions: { offset: number; length: number }[] = []; + initLogStreamStore({ + get: id => underlying.get(id), + open: async id => { + const stream = await underlying.open(id); + return { + ...stream, append: async (offset, bytes) => { + positions.push({ offset, length: bytes.byteLength }); + if (positions.length === 1) { entered.resolve(); await release.promise; } + await stream.append(offset, bytes); + }, + }; + }, + }); + const stubs = installDumpStubs(initDumpStore, initDumpBroker); + try { + const dump = openRunDump(apiKey(3600), turn, trackBackground, false, { upstreamCallStartedAt: null, firstOutputTokenAt: null })!; + const event = dump.sink({ type: 'stage.entered', stageId: 1, name: 'slow append', parentStageId: null, facts: move({}) }); + await entered.promise; + await vi.advanceTimersByTimeAsync(LOG_STREAM_IDLE_MS / 2); + expect(positions).toHaveLength(1); + release.resolve(); + await event; + dump.finalize(200, 0); + await flushBackground(); + expect(positions).toEqual([{ offset: 0, length: positions[0]!.length }, { offset: positions[0]!.length, length: 0 }]); + expect(stubs.stored).toHaveLength(1); + } finally { release.resolve(); vi.useRealTimers(); initLogStreamStore(testLogStreamStore()); } +}); + +test('an encoding failure aborts the durable reader instead of leaving it waiting for EOF', async () => { + initLogStreamStore(testLogStreamStore()); + const stubs = installDumpStubs(initDumpStore, initDumpBroker); + const rejected = Promise.withResolvers(); + initDumpStore({ + ...stubs.store, putRun: async (_keyId, writing) => { + try { await new Response(writing.events).arrayBuffer(); } catch (error) { rejected.resolve(error); throw error; } + await writing.metadata; + }, + }); + const work: Promise[] = []; + const dump = openRunDump(apiKey(3600), turn, pending => { work.push(pending); }, false, { upstreamCallStartedAt: null, firstOutputTokenAt: null })!; + const error = await dump.sink({ type: 'stage.entered', stageId: 1, name: 'invalid', parentStageId: null, facts: move({ upload: new Blob(['data']) }) }).catch(error => error as Error); + expect(error).toBeInstanceOf(TypeError); + expect((error as Error).message).toContain('Blob'); + dump.finalize(500, 0); + expect(await rejected.promise).toBe(error); + const outcomes = await Promise.allSettled(work); + expect(outcomes).toHaveLength(2); + expect(outcomes.every(outcome => outcome.status === 'rejected' && outcome.reason === error)).toBe(true); + expect(stubs.stored).toEqual([]); +}); + +test('an early durable failure stops heartbeats even when the live stream opens later', async () => { + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] }); + const original = new Error('durable store unavailable'); + const opening = Promise.withResolvers(); + const underlying = testLogStreamStore(); + initLogStreamStore({ get: id => underlying.get(id), open: async id => { await opening.promise; return await underlying.open(id); } }); + const stubs = installDumpStubs(initDumpStore, initDumpBroker); + initDumpStore({ ...stubs.store, putRun: async () => { throw original; } }); + const work: Promise[] = []; + try { + const dump = openRunDump(apiKey(3600), turn, pending => { work.push(pending); }, false, { upstreamCallStartedAt: null, firstOutputTokenAt: null })!; + expect(await Promise.allSettled(work)).toEqual([{ status: 'rejected', reason: original }]); + opening.resolve(); + await getLogStreamStore().open('test-open-barrier'); + expect(vi.getTimerCount()).toBe(0); + await expect(dump.sink({ type: 'stage.entered', stageId: 1, name: 'failed storage', parentStageId: null, facts: move({}) })).rejects.toBe(original); + } finally { opening.resolve(); vi.clearAllTimers(); vi.useRealTimers(); initLogStreamStore(testLogStreamStore()); } +}); + +test('live EOF waits until the durable artifact has been published', async () => { + const underlying = testLogStreamStore(); + const end = vi.fn(); + initLogStreamStore({ + get: id => underlying.get(id), open: async id => { + const stream = await underlying.open(id); + return { ...stream, end: async () => { end(); await stream.end(); } }; + }, + }); + const stubs = installDumpStubs(initDumpStore, initDumpBroker); + const consumed = Promise.withResolvers(); + const publish = Promise.withResolvers(); + initDumpStore({ + ...stubs.store, putRun: async (keyId, writing) => { + const bytes = new Uint8Array(await new Response(writing.events).arrayBuffer()); + await writing.metadata; + consumed.resolve(); + await publish.promise; + await stubs.store.putRun(keyId, { + ...writing, events: new ReadableStream({ start(controller) { controller.enqueue(bytes); controller.close(); } }), + }); + }, + }); + const dump = openRunDump(apiKey(3600), turn, trackBackground, false, { upstreamCallStartedAt: null, firstOutputTokenAt: null })!; + await dump.sink({ type: 'stage.entered', stageId: 1, name: 'publication', parentStageId: null, facts: move({}) }); + dump.finalize(200, 0); + await consumed.promise; + expect(stubs.stored).toHaveLength(0); + expect(end).not.toHaveBeenCalled(); + publish.resolve(); + await flushBackground(); + expect(stubs.stored).toHaveLength(1); + expect(end).toHaveBeenCalledOnce(); +}); + +test('a durable publication failure never ends the live stream and preserves its original error', async () => { + const underlying = testLogStreamStore(); + const end = vi.fn(); + initLogStreamStore({ + get: id => underlying.get(id), open: async id => { + const stream = await underlying.open(id); + return { ...stream, end: async () => { end(); await stream.end(); } }; + }, + }); + const stubs = installDumpStubs(initDumpStore, initDumpBroker); + const consumed = Promise.withResolvers(); + const publish = Promise.withResolvers(); + const original = new Error('durable publication failed'); + initDumpStore({ + ...stubs.store, putRun: async (_keyId, writing) => { + await new Response(writing.events).arrayBuffer(); + consumed.resolve(); + await publish.promise; + await writing.metadata; + }, + }); + const work: Promise[] = []; + const dump = openRunDump(apiKey(3600), turn, pending => { work.push(pending); }, false, { upstreamCallStartedAt: null, firstOutputTokenAt: null })!; + await dump.sink({ type: 'stage.entered', stageId: 1, name: 'publication failure', parentStageId: null, facts: move({}) }); + dump.finalize(200, 0); + await consumed.promise; + publish.reject(original); + const outcomes = await Promise.allSettled(work); + expect(outcomes).toHaveLength(2); + expect(outcomes.every(outcome => outcome.status === 'rejected' && outcome.reason === original)).toBe(true); + expect(stubs.stored).toHaveLength(0); + expect(stubs.published).toHaveLength(0); + expect(end).not.toHaveBeenCalled(); +}); diff --git a/packages/gateway/__tests__/dump/test-fixtures.ts b/packages/gateway/__tests__/dump/test-fixtures.ts index e89e9ec859..10de46673d 100644 --- a/packages/gateway/__tests__/dump/test-fixtures.ts +++ b/packages/gateway/__tests__/dump/test-fixtures.ts @@ -40,7 +40,7 @@ export const eventsOf = (record: StoredDumpRecord): readonly Record JSON.parse(line) as Record); type DumpStubFailMethod = - | 'put' + | 'putRun' | 'list' | 'get' | 'publish' @@ -68,9 +68,11 @@ export const installDumpStubs = ( const throws: Partial> = {}; const store: DumpStore = { - async put(keyId, record) { - if (throws.put) throw throws.put; - const storedRecord: StoredDumpRecord = record; + async putRun(keyId, writing) { + const events = new Uint8Array(await new Response(writing.events).arrayBuffer()); + const meta = await writing.metadata; + if (throws.putRun) throw throws.putRun; + const storedRecord: StoredDumpRecord = { meta, events }; stored.push({ keyId, record: storedRecord }); const list = records.get(keyId) ?? []; list.unshift(storedRecord); diff --git a/packages/gateway/__tests__/dump/write-run.ts b/packages/gateway/__tests__/dump/write-run.ts new file mode 100644 index 0000000000..3e73326d8c --- /dev/null +++ b/packages/gateway/__tests__/dump/write-run.ts @@ -0,0 +1,9 @@ +import type { DumpStore } from '../../src/dump/store-contract.ts'; +import type { StoredDumpRecord } from '../../src/dump/types.ts'; + +export const writeRun = (store: DumpStore, keyId: string, record: StoredDumpRecord): Promise => store.putRun(keyId, { + id: record.meta.id, + startedAt: record.meta.startedAt, + events: new ReadableStream({ start(controller) { controller.enqueue(record.events); controller.close(); } }), + metadata: Promise.resolve(record.meta), +}); diff --git a/packages/gateway/__tests__/repo/dump-store_test.ts b/packages/gateway/__tests__/repo/dump-store_test.ts index b1751fb401..3ad23e6813 100644 --- a/packages/gateway/__tests__/repo/dump-store_test.ts +++ b/packages/gateway/__tests__/repo/dump-store_test.ts @@ -2,15 +2,17 @@ import { mkdir, mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { dirname, join, resolve } from 'node:path'; -import { expect, test } from 'vitest'; +import { expect, test, vi } from 'vitest'; import { createSqliteTestDb, mapRunChangeCount } from './test-sqlite.ts'; import { decodeDumpBodyDescriptor } from '../../src/dump/storage-codec.ts'; import type { StoredDumpRecord } from '../../src/dump/types.ts'; import { FileDumpStore } from '../../src/repo/dump-store.ts'; import { initRepo } from '../../src/repo/index.ts'; +import { SPILLED_FILE_STAGE_GRACE_MS } from '../../src/repo/spilled-files-policy.ts'; import { SqlRepo } from '../../src/repo/sql.ts'; import { collectSpilledFiles } from '../../src/scheduled/spilled-files.ts'; +import { writeRun } from '../dump/write-run.ts'; import { encodeRun, toNdjson, type Facts } from '@floway-dev/pipeline'; import { initFileStore, MemoryFileStore } from '@floway-dev/platform'; import type { FileStore, SqlDatabase } from '@floway-dev/platform'; @@ -59,7 +61,7 @@ test('FileDumpStore round-trips a run record as its NDJSON event stream', async const store = new FileDumpStore(db, files); const record = runRecord('01HZZ00000000000000000RUN1', Date.UTC(2026, 5, 1, 12, 0, 0)); - await store.put('key_x', record); + await writeRun(store, 'key_x', record); const fetched = await store.get('key_x', record.meta.id); assertExists(fetched); assertEquals(new TextDecoder().decode(fetched.events), new TextDecoder().decode(record.events)); @@ -80,8 +82,8 @@ test('FileDumpStore lists what a row renders without opening the stream', async const db = await openDb(); const store = new FileDumpStore(db, new MemoryFileStore()); const base = Date.UTC(2026, 5, 1, 12, 0, 0); - await store.put('key_x', runRecord('01HZZ00000000000000000RUN1', base)); - await store.put('key_x', runRecord('01HZZ00000000000000000RUN2', base + 1)); + await writeRun(store, 'key_x', runRecord('01HZZ00000000000000000RUN1', base)); + await writeRun(store, 'key_x', runRecord('01HZZ00000000000000000RUN2', base + 1)); const listed = await store.list('key_x', { limit: 10 }); assertEquals(listed.map(meta => meta.id), ['01HZZ00000000000000000RUN2', '01HZZ00000000000000000RUN1']); @@ -98,8 +100,8 @@ test('FileDumpStore retires an expired run record and collects its stream file', initFileStore(files); const store = new FileDumpStore(db, files); const now = Date.UTC(2026, 5, 1, 12, 0, 0); - await store.put('key_x', runRecord('01HZZ00000000000000000RUN3', Date.UTC(2026, 5, 1, 9, 0, 0))); - await store.put('key_x', runRecord('01HZZ00000000000000000RUN4', now)); + await writeRun(store, 'key_x', runRecord('01HZZ00000000000000000RUN3', Date.UTC(2026, 5, 1, 9, 0, 0))); + await writeRun(store, 'key_x', runRecord('01HZZ00000000000000000RUN4', now)); await repo.apiKeys.update('key_x', { dumpRetentionSeconds: 2 * 3600 }); const originalNow = Date.now; @@ -124,7 +126,7 @@ test('FileDumpStore rejects malformed metadata with its row identity', async () const db = await openDb(); const store = new FileDumpStore(db, new MemoryFileStore()); const record = runRecord('01HZZ000000000000000000BADM', Date.UTC(2026, 5, 1, 12)); - await store.put('key_x', record); + await writeRun(store, 'key_x', record); await db.prepare('UPDATE dump_records SET meta_json = ? WHERE key_id = ? AND id = ?') .bind(JSON.stringify({ ...record.meta, upstream: undefined, status: '200' }), 'key_x', record.meta.id) .run(); @@ -137,7 +139,7 @@ test('FileDumpStore rejects a malformed body descriptor before file access', asy const db = await openDb(); const store = new FileDumpStore(db, new MemoryFileStore()); const record = runRecord('01HZZ000000000000000000BADD', Date.UTC(2026, 5, 1, 12)); - await store.put('key_x', record); + await writeRun(store, 'key_x', record); await db.prepare('UPDATE dump_records SET response_body_descriptor = ? WHERE key_id = ? AND id = ?') .bind(JSON.stringify({ key: 'dumps/v1/key_x/run.gz', type: 'chunks' }), 'key_x', record.meta.id) .run(); @@ -153,7 +155,7 @@ test('FileDumpStore refuses a row that points at no run stream', async () => { const db = await openDb(); const store = new FileDumpStore(db, new MemoryFileStore()); const record = runRecord('01HZZ000000000000000000NORUN', Date.UTC(2026, 5, 1, 12)); - await store.put('key_x', record); + await writeRun(store, 'key_x', record); await db.prepare('UPDATE dump_records SET response_body_descriptor = ? WHERE key_id = ? AND id = ?') .bind(JSON.stringify({ key: 'dumps/v1/key_x/run.gz', type: 'bytes' }), 'key_x', record.meta.id) .run(); @@ -168,7 +170,7 @@ test('FileDumpStore.list paginates newest-first with the (createdAt, id) cursor' const store = new FileDumpStore(db, files); const base = Date.UTC(2026, 5, 1, 12, 0, 0); for (let i = 0; i < 5; i++) { - await store.put('key_x', runRecord(`01HZZ000000000000000000A0${i}`, base + i)); + await writeRun(store, 'key_x', runRecord(`01HZZ000000000000000000A0${i}`, base + i)); } const first = await store.list('key_x', { limit: 2 }); assertEquals(first.map(m => m.id), ['01HZZ000000000000000000A04', '01HZZ000000000000000000A03']); @@ -185,8 +187,8 @@ test('FileDumpStore applies retention immediately and retires exact expired file const store = new FileDumpStore(db, files); const now = Date.UTC(2026, 5, 1, 12, 0, 0); // Old bucket 9:xx, current bucket 12:xx. - await store.put('key_x', runRecord('01HZZ0000000000000000000A1', Date.UTC(2026, 5, 1, 9, 0, 0))); - await store.put('key_x', runRecord('01HZZ0000000000000000000A2', now)); + await writeRun(store, 'key_x', runRecord('01HZZ0000000000000000000A1', Date.UTC(2026, 5, 1, 9, 0, 0))); + await writeRun(store, 'key_x', runRecord('01HZZ0000000000000000000A2', now)); await repo.apiKeys.update('key_x', { dumpRetentionSeconds: 2 * 3600 }); const originalNow = Date.now; Date.now = () => now + 1; @@ -215,7 +217,7 @@ test('growing dump retention can reveal a row not yet physically deleted', async const files = new MemoryFileStore(); const store = new FileDumpStore(db, files); const now = Date.UTC(2026, 5, 1, 12); - await store.put('key_x', runRecord('01HZZ0000000000000000000A4', now - 3 * 3600_000)); + await writeRun(store, 'key_x', runRecord('01HZZ0000000000000000000A4', now - 3 * 3600_000)); const originalNow = Date.now; Date.now = () => now; try { @@ -235,8 +237,8 @@ test('FileDumpStore retires every dump record when retention is disabled and col const files = new MemoryFileStore(); initFileStore(files); const store = new FileDumpStore(db, files); - await store.put('key_x', runRecord('01HZZ0000000000000000000A1', Date.UTC(2026, 5, 1, 9, 0, 0))); - await store.put('key_x', runRecord('01HZZ0000000000000000000A2', Date.UTC(2026, 5, 1, 12, 0, 0))); + await writeRun(store, 'key_x', runRecord('01HZZ0000000000000000000A1', Date.UTC(2026, 5, 1, 9, 0, 0))); + await writeRun(store, 'key_x', runRecord('01HZZ0000000000000000000A2', Date.UTC(2026, 5, 1, 12, 0, 0))); await repo.apiKeys.update('key_x', { dumpRetentionSeconds: null }); assertEquals((await store.list('key_x', { limit: 10 })).length, 0); assertEquals(await store.deleteExpiredBatch('key_x', Date.now(), 100), 2); @@ -250,8 +252,8 @@ test('FileDumpStore counts returned dump rows instead of trigger-amplified chang const repo = new SqlRepo(db); const files = new MemoryFileStore(); const store = new FileDumpStore(db, files); - await store.put('key_x', runRecord('01HZZ0000000000000000000C1', Date.UTC(2026, 5, 1, 9))); - await store.put('key_x', runRecord('01HZZ0000000000000000000C2', Date.UTC(2026, 5, 1, 10))); + await writeRun(store, 'key_x', runRecord('01HZZ0000000000000000000C1', Date.UTC(2026, 5, 1, 9))); + await writeRun(store, 'key_x', runRecord('01HZZ0000000000000000000C2', Date.UTC(2026, 5, 1, 10))); await repo.apiKeys.update('key_x', { dumpRetentionSeconds: null }); const d1LikeStore = new FileDumpStore(mapRunChangeCount(db, changes => changes * 3), files); @@ -265,8 +267,8 @@ test('FileDumpStore counts returned active dump rows instead of trigger-amplifie const files = new MemoryFileStore(); const store = new FileDumpStore(db, files); const now = Date.UTC(2026, 5, 1, 12); - await store.put('key_x', runRecord('01HZZ0000000000000000000C3', Date.UTC(2026, 5, 1, 9))); - await store.put('key_x', runRecord('01HZZ0000000000000000000C4', Date.UTC(2026, 5, 1, 10))); + await writeRun(store, 'key_x', runRecord('01HZZ0000000000000000000C3', Date.UTC(2026, 5, 1, 9))); + await writeRun(store, 'key_x', runRecord('01HZZ0000000000000000000C4', Date.UTC(2026, 5, 1, 10))); await repo.apiKeys.update('key_x', { dumpRetentionSeconds: 3600 }); const d1LikeStore = new FileDumpStore(mapRunChangeCount(db, changes => changes * 3), files); @@ -282,11 +284,10 @@ test('a record-ID race leaves only the losing write\'s uniquely keyed files coll initFileStore(files); const store = new FileDumpStore(db, files); const record = runRecord('01HZZ0000000000000000000A3', Date.now()); - await store.put('key_x', record); + await writeRun(store, 'key_x', record); - // One file per record now — the run's own stream — so the winner owns one and the loser - // staged one under a key of its own. - await expect(store.put('key_x', record)).rejects.toThrow(); + // The winner owns its artifact; the losing write leaves a distinct staged file. + await expect(writeRun(store, 'key_x', record)).rejects.toThrow(); expect((await db.prepare("SELECT COUNT(*) AS count FROM spilled_files WHERE state = 'owned'").first<{ count: number }>())?.count).toBe(1); expect((await db.prepare("SELECT COUNT(*) AS count FROM spilled_files WHERE state = 'staged'").first<{ count: number }>())?.count).toBe(1); @@ -334,14 +335,14 @@ class TmpDirFileStore implements FileStore { } } -test('FileDumpStore: put + get round-trips through real-filesystem IO', async () => { +test('FileDumpStore: streaming write + get round-trips through real-filesystem IO', async () => { const root = await mkdtemp(join(tmpdir(), 'dump-store-')); try { const db = await openDb(); const store = new FileDumpStore(db, new TmpDirFileStore(join(root, 'files'))); const record = runRecord('01HZZ0000000000000000000A1', Date.UTC(2026, 5, 1, 12, 0, 0)); - await store.put('key_x', record); + await writeRun(store, 'key_x', record); const fetched = await store.get('key_x', '01HZZ0000000000000000000A1'); assertExists(fetched); assertEquals(new TextDecoder().decode(fetched.events), new TextDecoder().decode(record.events)); @@ -358,10 +359,57 @@ test('FileDumpStore filters all retained history before applying the page limit' const record = runRecord(`filter-${i}`, now - i); record.meta.model = i === 4 ? 'needle-model' : 'other'; record.meta.error = i === 4 ? { kind: 'failed', reason: 'socket reset' } : null; - await store.put('key_x', record); + await writeRun(store, 'key_x', record); } expect((await store.list('key_x', { q: 'needle', limit: 1 })).map(meta => meta.id)).toEqual(['filter-4']); expect((await store.list('key_x', { q: 'SOCKET', failures: true, limit: 1 })).map(meta => meta.id)).toEqual(['filter-4']); expect(await store.list('key_x', { q: "' OR 1=1 --", limit: 1 })).toEqual([]); expect(await store.list('other-key', { q: 'needle', limit: 1 })).toEqual([]); }); + +test('FileDumpStore writes streaming run bytes before publishing its metadata row', async () => { + const db = await openDb(); + const files = new MemoryFileStore(); + const store = new FileDumpStore(db, files); + const metadata = Promise.withResolvers(); + const pipe = new TransformStream(); + const writer = pipe.writable.getWriter(); + const id = 'streamed-run'; + const startedAt = Date.now(); + const writing = store.putRun('key_x', { id, startedAt, events: pipe.readable, metadata: metadata.promise }); + const bytes = new TextEncoder().encode('{"type":"stage.entered","facts":{"text":"中"}}\n'); + await writer.write(bytes); + expect(await store.get('key_x', id)).toBeNull(); + await writer.close(); + metadata.resolve({ ...runRecord(id, Date.now()).meta, startedAt }); + await writing; + const stored = await store.get('key_x', id); + expect(stored!.meta.id).toBe(id); + expect(stored!.events).toEqual(bytes); +}); + +test('a long active run keeps its staged file out of orphan collection', async () => { + const db = await openDb(); + const files = new MemoryFileStore(); + initRepo(new SqlRepo(db)); + initFileStore(files); + const store = new FileDumpStore(db, files); + const metadata = Promise.withResolvers(); + const pipe = new TransformStream(); + const writer = pipe.writable.getWriter(); + const id = 'long-active-run'; + const startedAt = Date.now(); + vi.useFakeTimers({ toFake: ['Date', 'setTimeout', 'clearTimeout'] }); + try { + const writing = store.putRun('key_x', { id, startedAt, events: pipe.readable, metadata: metadata.promise }); + await writer.write(new TextEncoder().encode('{"type":"stage.entered"}\n')); + await vi.advanceTimersByTimeAsync(SPILLED_FILE_STAGE_GRACE_MS * 2 + 1); + await collectSpilledFiles(Date.now()); + await writer.close(); + metadata.resolve({ ...runRecord(id, Date.now()).meta, startedAt }); + await writing; + const stored = await store.get('key_x', id); + expect(stored!.meta.id).toBe(id); + expect(new TextDecoder().decode(stored!.events)).toBe('{"type":"stage.entered"}\n'); + } finally { vi.useRealTimers(); } +}); diff --git a/packages/gateway/__tests__/scheduled/expiration-sweeps_test.ts b/packages/gateway/__tests__/scheduled/expiration-sweeps_test.ts index 4e5df87543..fb40c98a24 100644 --- a/packages/gateway/__tests__/scheduled/expiration-sweeps_test.ts +++ b/packages/gateway/__tests__/scheduled/expiration-sweeps_test.ts @@ -8,6 +8,7 @@ import { quantizeOpenAIResponsesRefreshedAt, OPENAI_RESPONSES_REFRESH_GRANULARIT import { SqlRepo } from '../../src/repo/sql.ts'; import type { ApiKey, StoredOpenAIResponsesItem } from '../../src/repo/types.ts'; import { sweepExpirations } from '../../src/scheduled/expiration-sweeps.ts'; +import { writeRun } from '../dump/write-run.ts'; import { InMemoryRepo } from '../repo/memory.ts'; import { createSqliteTestDb, createSqlJsDatabase, migrationSqlByFilename, wrapSqlJsDatabase } from '../repo/test-sqlite.ts'; import { initFileStore, MemoryFileStore } from '@floway-dev/platform'; @@ -80,9 +81,9 @@ test('one fair driver drains bounded OpenAI Responses and dump backlogs', async ); await repo.openaiResponsesItems.insertMany([responseItem('msg-current', now)], 0); for (let index = 0; index < 150; index += 1) { - await dumps.put('key-a', dumpRecord(`01K00000000000000000${String(index).padStart(4, '0')}`, dumpExpiredAt)); + await writeRun(dumps, 'key-a', dumpRecord(`01K00000000000000000${String(index).padStart(4, '0')}`, dumpExpiredAt)); } - await dumps.put('key-a', dumpRecord('01K00000000000000000LIVE', now)); + await writeRun(dumps, 'key-a', dumpRecord('01K00000000000000000LIVE', now)); await repo.apiKeys.update('key-a', { dumpRetentionSeconds: 3600, openaiResponsesRetentionSeconds: OPENAI_RESPONSES_RETENTION_SECONDS }); await sweepExpirations(now); @@ -114,14 +115,14 @@ test('minute ticks catch up with a growing hot dump key without widening a batch await repo.apiKeys.save(key(start)); for (let index = 0; index < 150; index += 1) { - await dumps.put('key-a', dumpRecord(`01K00000000000000001${String(index).padStart(4, '0')}`, start - 3600_001)); + await writeRun(dumps, 'key-a', dumpRecord(`01K00000000000000001${String(index).padStart(4, '0')}`, start - 3600_001)); } for (let tick = 0; tick < 4; tick += 1) { const now = start + tick * 60_000; vi.setSystemTime(now); for (let index = 0; index < 11; index += 1) { - await dumps.put('key-a', dumpRecord(`01K00000000000000002${tick}${String(index).padStart(3, '0')}`, now - 3600_001)); + await writeRun(dumps, 'key-a', dumpRecord(`01K00000000000000002${tick}${String(index).padStart(3, '0')}`, now - 3600_001)); } await sweepExpirations(now); const expected = Math.max(0, 150 + 11 * (tick + 1) - 50 * (tick + 1)); diff --git a/packages/gateway/__tests__/scheduled_test.ts b/packages/gateway/__tests__/scheduled_test.ts index bc4354ce17..738adfee13 100644 --- a/packages/gateway/__tests__/scheduled_test.ts +++ b/packages/gateway/__tests__/scheduled_test.ts @@ -1,5 +1,6 @@ import { afterEach, expect, test, vi } from 'vitest'; +import { writeRun } from './dump/write-run.ts'; import { initDumpStore } from '../src/dump/registry.ts'; import type { StoredDumpRecord } from '../src/dump/types.ts'; import { FileDumpStore } from '../src/repo/dump-store.ts'; @@ -127,7 +128,7 @@ test('one maintenance tick collects every file retired by its four dump units', const keyId = `key-${keyIndex}`; await repo.apiKeys.save(apiKey(keyId, now, keyIndex + 1)); for (let rowIndex = 0; rowIndex < 50; rowIndex += 1) { - await dumps.put(keyId, fileBackedDumpRecord(`dump-${keyIndex}-${rowIndex}`, now - 3600_001)); + await writeRun(dumps, keyId, fileBackedDumpRecord(`dump-${keyIndex}-${rowIndex}`, now - 3600_001)); } } const { results: ownedFiles } = await db.prepare('SELECT file_key FROM spilled_files ORDER BY file_key') diff --git a/packages/gateway/__tests__/test-utils/log-stream.ts b/packages/gateway/__tests__/test-utils/log-stream.ts new file mode 100644 index 0000000000..f48629313e --- /dev/null +++ b/packages/gateway/__tests__/test-utils/log-stream.ts @@ -0,0 +1,43 @@ +import type { LogStream, LogStreamStore } from '@floway-dev/platform'; + +export const testLogStreamStore = (): LogStreamStore => { + const streams = new Map(); + return { + async open(id) { + const chunks: { offset: number; bytes: Uint8Array }[] = []; + let length = 0; + let ended = false; + const listeners = new Set<() => void>(); + const wake = () => { for (const listener of listeners) listener(); }; + const stream: LogStream = { + async append(offset, bytes) { + if (ended) throw new Error('Test LogStream is ended'); + if (offset > length) throw new Error('Test LogStream append would leave a hole'); + const tail = bytes.slice(length - offset); + if (tail.byteLength > 0) { chunks.push({ offset: length, bytes: tail }); length += tail.byteLength; } + wake(); + }, + async end() { ended = true; wake(); }, + async *read(offset, signal) { + while (!signal.aborted) { + const chunk = chunks.find(chunk => chunk.offset + chunk.bytes.byteLength > offset); + if (chunk) { + const bytes = chunk.bytes.slice(Math.max(0, offset - chunk.offset)); + offset += bytes.byteLength; + yield bytes; + } else if (ended) return; + else await new Promise(resolve => { + const finish = () => { listeners.delete(finish); signal.removeEventListener('abort', finish); resolve(); }; + listeners.add(finish); + signal.addEventListener('abort', finish, { once: true }); + }); + } + signal.throwIfAborted(); + }, + }; + streams.set(id, stream); + return stream; + }, + async get(id) { return streams.get(id) ?? null; }, + }; +}; diff --git a/packages/gateway/__tests__/vitest.setup.ts b/packages/gateway/__tests__/vitest.setup.ts index cf3f0b8144..ac0233291d 100644 --- a/packages/gateway/__tests__/vitest.setup.ts +++ b/packages/gateway/__tests__/vitest.setup.ts @@ -1,4 +1,5 @@ import { trackBackground } from './test-utils/background-tracker.ts'; +import { testLogStreamStore } from './test-utils/log-stream.ts'; import type { DumpBroker } from '../src/dump/broker.ts'; import { initDumpBroker, initDumpStore } from '../src/dump/registry.ts'; import type { DumpStore } from '../src/dump/store-contract.ts'; @@ -7,7 +8,7 @@ import { handleExecutionRequest } from '../src/execution/handler.ts'; import { initBackgroundSchedulerResolver } from '../src/runtime/background.ts'; import { initExecutionCellNamespace } from '../src/runtime/execution.ts'; import { isReplayableBody } from '@floway-dev/http'; -import { initEnv, initFetch, initRuntimeKind, initTimingSafeEqual, InProcessExecutionCellNamespace } from '@floway-dev/platform'; +import { initEnv, initFetch, initRuntimeKind, initTimingSafeEqual, InProcessExecutionCellNamespace, initLogStreamStore } from '@floway-dev/platform'; // Production always initializes the environment getter at boot. Mirror that // here with a neutral default; tests needing real values (RUNTIME_LOCATION, @@ -16,6 +17,7 @@ initEnv(() => ''); // Tests run as 'node' by default. The few tests that exercise CF-specific // runtime behaviour re-init this with 'cloudflare'. initRuntimeKind('node'); +initLogStreamStore(testLogStreamStore()); initTimingSafeEqual((a, b) => a.every((byte, index) => byte === b[index])); initFetch((url, init) => { const body = init.body; @@ -33,7 +35,7 @@ initExecutionCellNamespace(new InProcessExecutionCellNamespace(handleExecutionRe // independent of that subsystem. Dump-specific tests install real or recording // implementations. const noopStore: DumpStore = { - async put(): Promise {}, + async putRun(_keyId, run) { await new Response(run.events).arrayBuffer(); await run.metadata; }, async list(): Promise { return []; }, async get(_keyId: string, _id: DumpRecordId): Promise { return null; }, async deleteExpiredBatch(): Promise { return 0; }, diff --git a/packages/gateway/src/control-plane/dump.ts b/packages/gateway/src/control-plane/dump.ts index 50463abc94..85dd096065 100644 --- a/packages/gateway/src/control-plane/dump.ts +++ b/packages/gateway/src/control-plane/dump.ts @@ -5,8 +5,10 @@ import { z } from 'zod'; import { ownedKeyForUser } from './shared/owned-key.ts'; import { getDumpBroker, getDumpStore } from '../dump/registry.ts'; +import { runStreamId } from '../dump/run-sink.ts'; import { dumpRecordToWire } from '../dump/wire.ts'; import { zValidator } from '../middleware/zod-validator.ts'; +import { getLogStreamStore, serveLogStream } from '@floway-dev/platform'; const LIST_LIMIT_DEFAULT = 100; const LIST_LIMIT_MAX = 200; @@ -23,7 +25,7 @@ const ownedKey = async (c: Context): Promise<{ id: string; error: null } | { id: const owned = await ownedKeyForUser(c, keyId); if (!owned) return { id: null, error: 'Key not found' }; if (owned.dumpRetentionSeconds === null) { - return { id: null, error: 'Dump capture is not enabled for this key.' }; + return { id: null, error: 'Run recording is not enabled for this key.' }; } return { id: owned.id, error: null }; }; @@ -40,6 +42,15 @@ export const dumpRoutes = new Hono() }); return c.json({ records }); }) + .get('/keys/:keyId/records/:recordId/live', zValidator('query', z.object({ + offset: z.coerce.number().int().nonnegative().max(Number.MAX_SAFE_INTEGER).optional(), + })), async c => { + const owned = await ownedKey(c); + if (owned.id === null) return c.json({ error: owned.error }, 404); + const stream = await getLogStreamStore().get(runStreamId(owned.id, c.req.param('recordId')!)); + if (stream === null) return c.json({ error: 'Live recording is unavailable' }, 404); + return serveLogStream(stream, c.req.valid('query').offset ?? 0, c.req.raw.signal); + }) .get('/keys/:keyId/records/:recordId', async c => { const owned = await ownedKey(c); if (owned.id === null) return c.json({ error: owned.error }, 404); diff --git a/packages/gateway/src/dump/run-sink.ts b/packages/gateway/src/dump/run-sink.ts index 30c9845c60..5da2dc72ad 100644 --- a/packages/gateway/src/dump/run-sink.ts +++ b/packages/gateway/src/dump/run-sink.ts @@ -5,12 +5,11 @@ import { attemptTtftMs, type AttemptTiming } from '../data-plane/shared/attempt- import type { RequestBody } from '../data-plane/shared/request-body.ts'; import type { ApiKey, TokenUsage } from '../repo/types.ts'; import { ulid } from '../shared/ulid.ts'; -import { createRunEncoder, isStreamFact, streamFact, toNdjson, type DumpEvent, type Event, type StreamFact } from '@floway-dev/pipeline'; -import type { BackgroundScheduler } from '@floway-dev/platform'; +import { createRunEncoder, isStreamFact, streamFact, toNdjson, type Event, type StreamFact } from '@floway-dev/pipeline'; +import { getLogStreamStore, LOG_STREAM_IDLE_MS, type BackgroundScheduler, type LogStream } from '@floway-dev/platform'; import type { ProtocolFrame } from '@floway-dev/protocols/common'; import type { TelemetryModelIdentity } from '@floway-dev/provider'; -// Only metadata is snapshotted here; request contents are facts recorded by the run. interface RequestSnapshot { readonly method: string; readonly path: string; @@ -18,10 +17,25 @@ interface RequestSnapshot { readonly streamError: string | null; } +const LIVE_APPEND_ATTEMPTS = 3; +const CHUNK_BYTES = 64 * 1024; + +export const runStreamId = (keyId: string, runId: string): string => `${keyId}/${runId}`; + export class RunDump { + readonly id: string; private readonly attribution = new DumpAttribution(); private readonly encode = createRunEncoder(); - private readonly events: DumpEvent[] = []; + private readonly metadata = Promise.withResolvers(); + private readonly writer: WritableStreamDefaultWriter; + private readonly stored: Promise; + private readonly liveReady: Promise; + private live: LogStream | null = null; + private offset = 0; + private heartbeat: ReturnType | undefined; + private heartbeatWrite: Promise = Promise.resolve(); + private closing = false; + private tail: Promise = Promise.resolve(); private runDrain: (() => Promise) | null = null; private sentPayloadBytes = 0; private streams = 0; @@ -34,36 +48,92 @@ export class RunDump { private readonly backgroundScheduler: BackgroundScheduler, private readonly wantsStream: boolean, private readonly timing: AttemptTiming, - ) {} + ) { + const id = ulid(startedAt); + this.id = id; + const bytes = new TransformStream(); + this.writer = bytes.writable.getWriter(); + // The metadata wait can reject before the storage reader reaches it; storage owns reporting. + void this.metadata.promise.catch(() => {}); + const stored = getDumpStore().putRun(apiKey.id, { + id, startedAt, events: bytes.readable, metadata: this.metadata.promise, + }).catch(error => this.abort(error)); + this.stored = stored; + this.backgroundScheduler((async () => { + await stored; + await getDumpBroker().publish(apiKey.id, await this.metadata.promise); + })()); + this.liveReady = getLogStreamStore().open(runStreamId(apiKey.id, id)).then(stream => { + this.live = stream; + this.scheduleHeartbeat(); + return stream; + }).catch(error => { + console.error('[dump] live stream open failed', error); + return null; + }); + } + + afterRun(drain: () => Promise): void { + this.runDrain = drain; + } - afterRun(drain: () => Promise): void { this.runDrain = drain; } - readonly sink = (event: Event): void => { - for (const encoded of this.encode(event)) this.events.push(encoded); + // One event owns its entire object batch; concurrent deferred outcomes cannot overtake it. + readonly sink = (event: Event): Promise => { + const write = this.tail.then(async () => { + const bytes = new TextEncoder().encode(toNdjson(this.encode(event))); + await this.liveReady; + for (let offset = 0; offset < bytes.byteLength; offset += CHUNK_BYTES) { + const chunk = bytes.subarray(offset, offset + CHUNK_BYTES); + await this.writer.write(chunk); + await this.appendLive(chunk); + } + }); + this.tail = write; + return write; }; - requestedModel(model: string): void { - this.attribution.requestedModel(model); + private async appendLive(bytes: Uint8Array): Promise { + const stream = this.live; + if (stream === null) return; + const atOffset = this.offset; + for (let attempt = 0; attempt < LIVE_APPEND_ATTEMPTS; attempt += 1) { + try { + await stream.append(atOffset, bytes); + this.offset += bytes.byteLength; + return; + } catch (error) { + if (attempt + 1 === LIVE_APPEND_ATTEMPTS) { + this.live = null; + console.error('[dump] live stream unavailable; durable recording continues', error); + } + } + } } - error(kind: 'upstream' | 'gateway', upstream?: string): void { - this.attribution.error(kind, upstream); + private scheduleHeartbeat(): void { + if (this.closing || this.live === null) return; + this.heartbeat = setTimeout(() => { + const heartbeat = this.tail.then(() => this.appendLive(new Uint8Array())); + this.tail = heartbeat; + this.heartbeatWrite = heartbeat.then(() => { this.scheduleHeartbeat(); }); + }, LOG_STREAM_IDLE_MS / 2); } - failed(reason: unknown, options?: { readonly fallback: boolean }): void { - this.attribution.failed(reason, options); - } + requestedModel(model: string): void { this.attribution.requestedModel(model); } + error(kind: 'upstream' | 'gateway', upstream?: string): void { this.attribution.error(kind, upstream); } + failed(reason: unknown, options?: { readonly fallback: boolean }): void { this.attribution.failed(reason, options); } + success(identity: TelemetryModelIdentity, usage: TokenUsage | null): void { this.attribution.success(identity, usage); } - frame(frame: ProtocolFrame): void | Promise { + async frame(frame: ProtocolFrame): Promise { this.answerStream ??= this.openStream(); - return this.answerStream.frame(frame); + await this.answerStream.frame(frame); } - // Stream references and frame events share one run-wide identity; end marks complete recording. openStream(): StreamRecording { const streamId = ++this.streams; return { - frame: frame => { this.sink({ type: 'stream.frame', streamId, frames: [frame] }); }, - end: () => { this.sink({ type: 'stream.end', streamId }); }, + frame: frame => this.sink({ type: 'stream.frame', streamId, frames: [frame] }), + end: () => this.sink({ type: 'stream.end', streamId }), fact: streamFact(streamId), }; } @@ -79,29 +149,17 @@ export class RunDump { ); } - success(identity: TelemetryModelIdentity, usage: TokenUsage | null): void { - this.attribution.success(identity, usage); - } - - recordSentPayloadBytes(byteLength: number): void { - this.sentPayloadBytes += byteLength; - } + recordSentPayloadBytes(byteLength: number): void { this.sentPayloadBytes += byteLength; } finalize(status: number | null, responseBytes: number): void; finalize(response: Response): Response; finalize(...args: [number | null, number] | [Response]): void | Response { if (args.length === 2) { - const [status, responseBytes] = args; - this.backgroundScheduler(this.write(status, responseBytes + this.sentPayloadBytes, null)); + this.backgroundScheduler(this.write(args[0], args[1] + this.sentPayloadBytes, null)); return; } - const [response] = args; - if (response.body === null) { - this.finalize(response.status, 0); - return response; - } - + if (response.body === null) { this.finalize(response.status, 0); return response; } const [forClient, forMeasure] = response.body.tee(); this.backgroundScheduler((async () => { const reader = forMeasure.getReader(); @@ -113,45 +171,56 @@ export class RunDump { if (done) break; payloadBytes += value.byteLength; } - } catch (err) { - streamError = oneLineError(err); + } catch (error) { + streamError = oneLineError(error); } finally { reader.releaseLock(); } await this.write(response.status, payloadBytes, streamError); })()); + return new Response(forClient, { status: response.status, statusText: response.statusText, headers: response.headers }); + } - return new Response(forClient, { - status: response.status, - statusText: response.statusText, - headers: response.headers, - }); + private async abort(error: unknown): Promise { + this.closing = true; + clearTimeout(this.heartbeat); + this.metadata.reject(error); + try { await this.writer.abort(error); } catch (cleanupError) { + if (cleanupError !== error) throw new AggregateError([error, cleanupError], 'Run recording failed and writer abort also failed', { cause: error }); + } + throw error; } private async write(status: number | null, responseBytes: number, responseStreamError: string | null): Promise { - if (this.runDrain !== null) { - try { await this.runDrain(); } catch (error) { this.failed(error); } + try { + if (this.runDrain !== null) { + try { await this.runDrain(); } catch (error) { this.failed(error); } + } + await this.tail; + const completedAt = Date.now(); + const meta = await this.attribution.metadata({ + id: this.id, startedAt: this.startedAt, completedAt, + method: this.requestSnapshot.method, path: this.requestSnapshot.path, status, + requestBytes: this.requestSnapshot.bodyByteLength, responseBytes, + ttftMs: this.wantsStream ? attemptTtftMs(this.timing) : null, + fallbackError: streamReadError(this.requestSnapshot.streamError, responseStreamError), + }); + this.metadata.resolve(meta); + await this.writer.close(); + await this.stored; + await this.liveReady; + this.closing = true; + clearTimeout(this.heartbeat); + await this.heartbeatWrite; + if (this.live !== null) { + try { await this.live.end(); } catch (error) { console.error('[dump] live stream end failed', error); } + } + } catch (error) { + await this.abort(error); + } finally { + this.closing = true; + clearTimeout(this.heartbeat); } - const completedAt = Date.now(); - const recordId = ulid(completedAt); - const meta: DumpMetadata = await this.attribution.metadata({ - id: recordId, - startedAt: this.startedAt, - completedAt, - method: this.requestSnapshot.method, - path: this.requestSnapshot.path, - status, - requestBytes: this.requestSnapshot.bodyByteLength, - responseBytes, - ttftMs: this.wantsStream ? attemptTtftMs(this.timing) : null, - fallbackError: streamReadError(this.requestSnapshot.streamError, responseStreamError), - }); - - // Publishing follows durable storage so metadata subscribers can immediately fetch detail. - await getDumpStore().put(this.apiKey.id, { - meta, events: new TextEncoder().encode(toNdjson(this.events)), - }); - await getDumpBroker().publish(this.apiKey.id, meta); } } @@ -161,26 +230,15 @@ export const openRunDump = ( backgroundScheduler: BackgroundScheduler, wantsStream: boolean, timing: AttemptTiming, -): RunDump | null => { - if (apiKey.dumpRetentionSeconds === null) return null; - return new RunDump( - apiKey, - { - method: turn.method, - path: turn.path, - bodyByteLength: turn.body.bytes.byteLength, - streamError: turn.body.streamError, - }, - Date.now(), - backgroundScheduler, - wantsStream, - timing, - ); -}; +): RunDump | null => apiKey.dumpRetentionSeconds === null ? null : new RunDump( + apiKey, + { method: turn.method, path: turn.path, bodyByteLength: turn.body.bytes.byteLength, streamError: turn.body.streamError }, + Date.now(), backgroundScheduler, wantsStream, timing, +); export interface StreamRecording { - frame(frame: ProtocolFrame): void | Promise; - end(): void | Promise; + frame(frame: ProtocolFrame): Promise; + end(): Promise; readonly fact: StreamFact; } diff --git a/packages/gateway/src/dump/store-contract.ts b/packages/gateway/src/dump/store-contract.ts index 8132db953d..5c33553008 100644 --- a/packages/gateway/src/dump/store-contract.ts +++ b/packages/gateway/src/dump/store-contract.ts @@ -1,7 +1,7 @@ import type { DumpMetadata, DumpRecordId, StoredDumpRecord } from './types.ts'; // Per-API-key run storage: metadata in SQL and compressed NDJSON in the -// FileStore. Completed writes stage their file before publishing the row; +// FileStore. Streaming writes stage their file before publishing the row; // detail reads rehydrate the event bytes for the control plane. export interface DumpListOptions { @@ -11,10 +11,17 @@ export interface DumpListOptions { limit: number; } +export interface DumpRunWrite { + readonly id: string; + readonly startedAt: number; + readonly events: ReadableStream; + readonly metadata: Promise; +} + export interface DumpStore { - // Store the completed event artifact before publishing metadata. Failed writes - // leave staged files collectible by the retention sweep. - put(keyId: string, record: StoredDumpRecord): Promise; + // Publish metadata only after the complete run artifact exists. Failed + // writes leave staged files collectible by the retention sweep. + putRun(keyId: string, run: DumpRunWrite): Promise; // Newest-first, paginated by ULID cursor. Reads enforce the API key's // current rolling retention even before queued physical deletion runs. diff --git a/packages/gateway/src/dump/types.ts b/packages/gateway/src/dump/types.ts index 91a2fd572b..ba5414d938 100644 --- a/packages/gateway/src/dump/types.ts +++ b/packages/gateway/src/dump/types.ts @@ -34,14 +34,15 @@ export type DumpMetadata = z.infer; // the event shape; client streams already contain their final wire projection. export type DumpStreamEvent = z.infer; -// The completed NDJSON artifact is stored after owned readers and deferred work -// settle. Cold detail reads rehydrate the same event bytes. +// Detail reads rehydrate the completed NDJSON artifact. Streaming writes carry +// the same encoded bytes through DumpStore.putRun while metadata settles. export type StoredDumpRecord = { meta: DumpMetadata; events: Uint8Array; }; -// The control plane exposes the completed UTF-8 NDJSON verbatim. +// The control plane exposes the completed UTF-8 NDJSON verbatim; the live +// byte reader receives the same encoded event stream. export type DumpRecord = { meta: DumpMetadata; events: string; diff --git a/packages/gateway/src/repo/dump-store.ts b/packages/gateway/src/repo/dump-store.ts index ddecbb410c..fff8e3e3c8 100644 --- a/packages/gateway/src/repo/dump-store.ts +++ b/packages/gateway/src/repo/dump-store.ts @@ -7,14 +7,14 @@ import { encodePersistedDumpMetadata, } from '../dump/storage-codec.ts'; import type { DumpBodyDescriptor } from '../dump/storage-codec.ts'; -import type { DumpListOptions, DumpStore } from '../dump/store-contract.ts'; +import type { DumpListOptions, DumpStore, DumpRunWrite } from '../dump/store-contract.ts'; import type { DumpMetadata, DumpRecordId, DumpUpstreamRef, StoredDumpRecord, } from '../dump/types.ts'; -import { gunzipBytes, gzipBytes } from '../shared/gzip.ts'; +import { gunzipBytes, gzipStream } from '../shared/gzip.ts'; import type { FileStore, SqlDatabase } from '@floway-dev/platform'; // Bodies live at `dumps/v1/{keyId}/{YYYYMMDDHH}/{recordId}-{uniqueSuffix}.run.gz`. @@ -75,23 +75,43 @@ const EMPTY_HEADERS_JSON = '[]'; export class FileDumpStore implements DumpStore { constructor(private readonly db: SqlDatabase, private readonly files: FileStore) {} - async put(keyId: string, record: StoredDumpRecord): Promise { - const { meta, events } = record; - const fileKey = bodyPath(keyId, hourBucket(meta.completedAt), meta.id); + async putRun(keyId: string, run: DumpRunWrite): Promise { + const fileKey = bodyPath(keyId, hourBucket(run.startedAt), run.id); await this.db.prepare( `INSERT INTO spilled_files (file_key, owner_kind, owner_key, state, collect_after) VALUES (?, 'dump-response', ?, 'staged', ?)`, - ).bind(fileKey, JSON.stringify([keyId, meta.id]), Date.now() + SPILLED_FILE_STAGE_GRACE_MS).run(); - await this.files.put(fileKey, await gzipBytes(events)); - await this.db.prepare( - `INSERT INTO dump_records - (key_id, id, created_at, upstream_id, meta_json, request_headers_json, response_headers_json, request_body_descriptor, response_body_descriptor) - VALUES (?, ?, ?, ?, ?, ?, NULL, NULL, ?)`, - ).bind( - keyId, meta.id, meta.completedAt, meta.upstream?.id ?? null, - encodePersistedDumpMetadata(meta, `dump record ${meta.id} metadata`), EMPTY_HEADERS_JSON, - encodeDumpBodyDescriptor({ key: fileKey, type: 'run' }, `dump record ${meta.id} run descriptor`), - ).run(); + ).bind(fileKey, JSON.stringify([keyId, run.id]), Date.now() + SPILLED_FILE_STAGE_GRACE_MS).run(); + let timer: ReturnType | undefined; + let finished = false; + let renewal: Promise = Promise.resolve(); + const stopRenewing = () => { finished = true; clearTimeout(timer); }; + const scheduleRenewal = () => { + timer = setTimeout(() => { + renewal = this.db.prepare( + "UPDATE spilled_files SET collect_after = ? WHERE file_key = ? AND state = 'staged'", + ).bind(Date.now() + SPILLED_FILE_STAGE_GRACE_MS, fileKey).run().then(() => { + if (!finished) scheduleRenewal(); + }); + // Awaited before publication even if the lease write failed while upstream was quiet. + void renewal.catch(() => {}); + }, SPILLED_FILE_STAGE_GRACE_MS / 2); + }; + scheduleRenewal(); + try { + await this.files.put(fileKey, gzipStream(run.events)); + const meta = await run.metadata; + stopRenewing(); + await renewal; + await this.db.prepare( + `INSERT INTO dump_records + (key_id, id, created_at, upstream_id, meta_json, request_headers_json, response_headers_json, request_body_descriptor, response_body_descriptor, response_upstream_body_descriptor) + VALUES (?, ?, ?, ?, ?, ?, NULL, NULL, ?, NULL)`, + ).bind( + keyId, meta.id, meta.completedAt, meta.upstream?.id ?? null, + encodePersistedDumpMetadata(meta, `dump record ${meta.id} metadata`), EMPTY_HEADERS_JSON, + encodeDumpBodyDescriptor({ key: fileKey, type: 'run' }, `dump record ${meta.id} run descriptor`), + ).run(); + } finally { stopRenewing(); } } async list(keyId: string, opts: DumpListOptions): Promise { diff --git a/packages/gateway/src/shared/gzip.ts b/packages/gateway/src/shared/gzip.ts index 74bda3d0e6..77a1f8241b 100644 --- a/packages/gateway/src/shared/gzip.ts +++ b/packages/gateway/src/shared/gzip.ts @@ -34,3 +34,6 @@ export const gzipBytes = async (bytes: Uint8Array): Promise => export const gunzipBytes = async (bytes: Uint8Array): Promise => await collect(bytesStream(bytes).pipeThrough(new DecompressionStream('gzip'))); + +export const gzipStream = (stream: ReadableStream): ReadableStream => + (stream as ReadableStream).pipeThrough(new CompressionStream('gzip')); From ba5d0cdf116af82aeff5be263bbc0e15c74e4e5d Mon Sep 17 00:00:00 2001 From: Menci Date: Thu, 1 Oct 2026 10:05:53 +0800 Subject: [PATCH 3/3] fix: renew live stream leases through durable backpressure Empty heartbeats use the last acknowledged offset independently of the ordered data writer. This keeps a live run from expiring while durable reads or append acknowledgements are pending, without changing bytes or data offsets. Verify long native-reader gates and concurrent heartbeat positions; stream end still awaits the in-flight heartbeat. --- README.md | 3 +- .../gateway/__tests__/dump/run-sink_test.ts | 51 +++++++++++++++++-- packages/gateway/src/dump/run-sink.ts | 5 +- 3 files changed, 53 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index 3f27a9ed20..2b9d57f7ce 100644 --- a/README.md +++ b/README.md @@ -210,7 +210,8 @@ Run dumps encode stage facts, logs and protocol frames into one object space. Th recording layer streams those encoded NDJSON bytes into durable storage with backpressure and appends the same bytes to a temporary LogStream. Live appends retry at the same byte offset; a persistently unavailable live stream leaves durable recording active. Quiet runs -renew their temporary-stream and staged-file leases. Closing waits for owned readers and +renew their temporary-stream and staged-file leases even while durable writes apply +backpressure. Closing waits for owned readers and deferred outcomes before publishing completed metadata. The live stream ends only after the durable artifact is published; an encoding or durable-write failure aborts its reader and stops its leases while retaining the original error. Run IDs are assigned at entry; diff --git a/packages/gateway/__tests__/dump/run-sink_test.ts b/packages/gateway/__tests__/dump/run-sink_test.ts index 7fbf239dd5..7b94bc7bdc 100644 --- a/packages/gateway/__tests__/dump/run-sink_test.ts +++ b/packages/gateway/__tests__/dump/run-sink_test.ts @@ -318,7 +318,7 @@ test('closing waits for deferred outcomes and persists their settlement event', expect(ndjson(runRecordOf(stubs.stored[0]))).toContain('late value'); }); -test('heartbeat append waits for the in-flight event and uses its resulting byte offset', async () => { +test('heartbeats renew the acknowledged offset while an event append awaits acknowledgement', async () => { vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] }); const underlying = testLogStreamStore(); const entered = Promise.withResolvers(); @@ -343,12 +343,16 @@ test('heartbeat append waits for the in-flight event and uses its resulting byte const event = dump.sink({ type: 'stage.entered', stageId: 1, name: 'slow append', parentStageId: null, facts: move({}) }); await entered.promise; await vi.advanceTimersByTimeAsync(LOG_STREAM_IDLE_MS / 2); - expect(positions).toHaveLength(1); + expect(positions).toEqual([{ offset: 0, length: positions[0]!.length }, { offset: 0, length: 0 }]); release.resolve(); await event; + await vi.advanceTimersByTimeAsync(LOG_STREAM_IDLE_MS / 2); dump.finalize(200, 0); await flushBackground(); - expect(positions).toEqual([{ offset: 0, length: positions[0]!.length }, { offset: positions[0]!.length, length: 0 }]); + expect(positions).toEqual([ + { offset: 0, length: positions[0]!.length }, { offset: 0, length: 0 }, + { offset: positions[0]!.length, length: 0 }, + ]); expect(stubs.stored).toHaveLength(1); } finally { release.resolve(); vi.useRealTimers(); initLogStreamStore(testLogStreamStore()); } }); @@ -464,3 +468,44 @@ test('a durable publication failure never ends the live stream and preserves its expect(stubs.published).toHaveLength(0); expect(end).not.toHaveBeenCalled(); }); + +test('heartbeats remain active through a long durable reader backpressure gate', async () => { + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] }); + const underlying = testLogStreamStore(); + const positions: { offset: number; length: number }[] = []; + initLogStreamStore({ + get: id => underlying.get(id), open: async id => { + const stream = await underlying.open(id); + return { ...stream, append: async (offset, bytes) => { positions.push({ offset, length: bytes.byteLength }); await stream.append(offset, bytes); } }; + }, + }); + const stubs = installDumpStubs(initDumpStore, initDumpBroker); + const acquired = Promise.withResolvers(); + const release = Promise.withResolvers(); + initDumpStore({ + ...stubs.store, putRun: async (_keyId, writing) => { + const reader = writing.events.getReader(); + acquired.resolve(); + await release.promise; + try { while (!(await reader.read()).done) {} } finally { reader.releaseLock(); } + await writing.metadata; + }, + }); + try { + const dump = openRunDump(apiKey(3600), turn, trackBackground, false, { upstreamCallStartedAt: null, firstOutputTokenAt: null })!; + await acquired.promise; + let wrote = false; + const writing = dump.sink({ type: 'stage.entered', stageId: 1, name: 'long backpressure', parentStageId: null, facts: move({}) }).then(() => { wrote = true; }); + await new Promise(resolve => setImmediate(resolve)); + expect(wrote).toBe(false); + await vi.advanceTimersByTimeAsync(LOG_STREAM_IDLE_MS * 2); + expect(wrote).toBe(false); + expect(positions).toEqual(Array.from({ length: 4 }, () => ({ offset: 0, length: 0 }))); + release.resolve(); + await writing; + expect(positions[4]!.length).toBeGreaterThan(0); + dump.finalize(200, 0); + await flushBackground(); + expect(await underlying.get(runStreamId(apiKey(3600).id, dump.id))).not.toBeNull(); + } finally { release.resolve(); vi.useRealTimers(); initLogStreamStore(testLogStreamStore()); } +}); diff --git a/packages/gateway/src/dump/run-sink.ts b/packages/gateway/src/dump/run-sink.ts index 5da2dc72ad..4ade53b743 100644 --- a/packages/gateway/src/dump/run-sink.ts +++ b/packages/gateway/src/dump/run-sink.ts @@ -113,8 +113,9 @@ export class RunDump { private scheduleHeartbeat(): void { if (this.closing || this.live === null) return; this.heartbeat = setTimeout(() => { - const heartbeat = this.tail.then(() => this.appendLive(new Uint8Array())); - this.tail = heartbeat; + // The acknowledged offset is safe during an in-flight data append: an empty write + // adds no bytes, and durable backpressure must not block the temporary stream lease. + const heartbeat = this.appendLive(new Uint8Array()); this.heartbeatWrite = heartbeat.then(() => { this.scheduleHeartbeat(); }); }, LOG_STREAM_IDLE_MS / 2); }