diff --git a/.changeset/bright-dolphins-laugh.md b/.changeset/bright-dolphins-laugh.md new file mode 100644 index 0000000000..2ff5980138 --- /dev/null +++ b/.changeset/bright-dolphins-laugh.md @@ -0,0 +1,7 @@ +--- +'@workflow/core': patch +'@workflow/world': patch +'@workflow/world-vercel': patch +--- + +Add an opt-in replay-safe stream lease append prototype. diff --git a/docs/content/docs/v5/configuration/runtime-tuning.mdx b/docs/content/docs/v5/configuration/runtime-tuning.mdx index 0d72640318..44ed85c08d 100644 --- a/docs/content/docs/v5/configuration/runtime-tuning.mdx +++ b/docs/content/docs/v5/configuration/runtime-tuning.mdx @@ -228,6 +228,11 @@ For example, a workflow can run a 10-minute inline step even with `WORKFLOW_REPL These variables are primarily for tests, debugging, or unusual deployments. +### `WORKFLOW_STREAM_LEASE_FAST_PATH` + +- Default: unset (off) +- Experimental deploy-only stream-write prototype. Set to `1` only when targeting a workflow-server preview implementing the matching lease protocol. It gives each writable sink a stable writer identity and sequence so lease appends can safely retry 5xx responses; streams that cannot retain a sole-writer lease de-opt to the existing allocator. This does not enable connection persistence or write pipelining. + ### `WORKFLOW_STREAM_FLUSH_INTERVAL_MS` - Default: `0` (dispatch the first chunk of an idle stream immediately) diff --git a/docs/content/worlds/v5/vercel.mdx b/docs/content/worlds/v5/vercel.mdx index ac96611222..ca7c6b20da 100644 --- a/docs/content/worlds/v5/vercel.mdx +++ b/docs/content/worlds/v5/vercel.mdx @@ -196,6 +196,10 @@ Per-request timeout, in milliseconds, for Vercel World HTTP calls to workflow-se Maximum stream chunks written in one Vercel World request. Larger batches are split across multiple requests. Default: `1000`. Minimum: `1`. +### `WORKFLOW_STREAM_LEASE_FAST_PATH` + +Experimental deploy-only stream-write prototype. Set to `1` only with a matching workflow-server preview. It enables replay-safe lease appends and 5xx retry while retaining the legacy allocator as a per-stream fallback. Default: off. It does not enable persistent connections or pipelining. + ### `WORKFLOW_EVENTS_TRANSPORT` Experimental. Set `WORKFLOW_EVENTS_TRANSPORT=ws` to ship workflow run events to the Vercel World over a WebSocket instead of one HTTP request each. Default: `http`. diff --git a/packages/core/src/serialization.ts b/packages/core/src/serialization.ts index e626cd3f91..8fbbcee4a5 100644 --- a/packages/core/src/serialization.ts +++ b/packages/core/src/serialization.ts @@ -149,6 +149,10 @@ export type SerializationFormatType = * (e.g., when starting a workflow or handling step return values). */ const defaultUlid = monotonicFactory(); +// Lease epochs fence writers across sink instances. Date.now() alone can repeat +// for sinks constructed in the same millisecond, so retain a process-local +// monotonic floor while keeping the token an ordinary number on the wire. +let latestStreamLeaseEpoch = Date.now(); /** * Detect if a readable stream is a byte stream. @@ -1279,6 +1283,23 @@ export class WorkflowServerWritableStream extends WritableStream { * early-ack sink. */ let sinkError: unknown; + // The lease protocol is deliberately opt-in: worlds without this optional + // transport (and every default deployment) retain the existing writes. + // Identity and ordering remain sink-local; this is not a connection. + const leaseFastPathEnabled = + process.env.WORKFLOW_STREAM_LEASE_FAST_PATH === '1'; + const writerId = defaultUlid(); + // The server fences this client-supplied token. A new sink always starts a + // new writer session, so this process-local monotonic token needs no + // lease-acquisition round trip. + const epoch = ++latestStreamLeaseEpoch; + let nextLeaseSeq = 0; + let leaseFastPathActive = true; + // A seq-gap names the first writer sequence the server lacks. Because it + // can be from an earlier flush, retain the per-sink sequence until this + // experimental path de-opts; a gap can then be repaired in FIFO order. + // This is intentionally scoped to the flag-gated measurement prototype. + const leaseHistory: Uint8Array[] = []; // Group-commit window. The env var, when set, overrides the World // option; otherwise `world.streamFlushIntervalMs` governs (default 0) — // including the very first chunk. When it must come from the world, @@ -1354,13 +1375,102 @@ export class WorkflowServerWritableStream extends WritableStream { await ensureRunReady(); const world = await worldPromise; const dispatchAt = Date.now(); - if (typeof world.streams.writeMulti === 'function' && group.length > 1) { - await world.streams.writeMulti(runId, name, group); - } else { - // Fall back to sequential writes - for (const chunk of group) { - await world.streams.write(runId, name, chunk); + const writeLegacy = async (): Promise => { + if ( + typeof world.streams.writeMulti === 'function' && + group.length > 1 + ) { + await world.streams.writeMulti(runId, name, group); + } else { + for (const chunk of group) { + await world.streams.write(runId, name, chunk); + } } + }; + + if ( + leaseFastPathEnabled && + leaseFastPathActive && + typeof world.streams.writeLease === 'function' + ) { + const seqStart = nextLeaseSeq; + leaseHistory.push(...group); + const result = await world.streams.writeLease(runId, name, group, { + writerId, + epoch, + seqStart, + }); + if (result.status === 'need-reserve') { + // De-opt is sticky: a multi-writer/expired lease stream remains on + // the known-safe allocator for this sink's remaining lifetime. A + // paged transport can have already acknowledged a leading prefix. + leaseFastPathActive = false; + const unacknowledged = group.slice(result.acknowledged ?? 0); + if (unacknowledged.length === group.length) { + await writeLegacy(); + } else if ( + typeof world.streams.writeMulti === 'function' && + unacknowledged.length > 1 + ) { + await world.streams.writeMulti(runId, name, unacknowledged); + } else { + for (const chunk of unacknowledged) { + await world.streams.write(runId, name, chunk); + } + } + } else if (result.status === 'seq-gap') { + // `expected` is the server's next sequence, so it precedes this + // request's seqStart. Replay every retained missing chunk in FIFO + // order, including this group, before accepting later groups. + if (result.expected === undefined) { + throw new Error('Stream lease seq-gap response omitted expected'); + } + // `writeLease` can page one core group. In that case a later page + // reports its own expected sequence, which may lie inside the group + // even though no later core group has begun. + if ( + result.expected < 0 || + result.expected >= seqStart + group.length + ) { + throw new Error( + `Stream lease seq-gap expected ${result.expected} outside retained history` + ); + } + const recovery = leaseHistory.slice(result.expected); + const recovered = await world.streams.writeLease( + runId, + name, + recovery, + { + writerId, + epoch, + seqStart: result.expected, + } + ); + if (recovered.status === 'need-reserve') { + // A paged recovery may have acknowledged a leading prefix before + // it lost the lease. Resume legacy allocation from the remaining + // server-reported gap, never re-writing that acknowledged range. + leaseFastPathActive = false; + const unacknowledged = recovery.slice(recovered.acknowledged ?? 0); + if ( + typeof world.streams.writeMulti === 'function' && + unacknowledged.length > 1 + ) { + await world.streams.writeMulti(runId, name, unacknowledged); + } else { + for (const chunk of unacknowledged) { + await world.streams.write(runId, name, chunk); + } + } + } else if (recovered.status === 'seq-gap') { + throw new Error('Stream lease seq-gap recovery did not advance'); + } + } + // `ok` and `replay` both durably account for this entire group. + nextLeaseSeq += group.length; + } else { + await writeLegacy(); } if (groupT0 !== undefined) { recordStreamWriteFlush( diff --git a/packages/core/src/writable-stream.test.ts b/packages/core/src/writable-stream.test.ts index 8dcbe253bd..bc8b12122d 100644 --- a/packages/core/src/writable-stream.test.ts +++ b/packages/core/src/writable-stream.test.ts @@ -33,6 +33,7 @@ describe('WorkflowServerWritableStream', () => { let mockStreams: { write: ReturnType; writeMulti: ReturnType; + writeLease: ReturnType; close: ReturnType; }; let mockWorld: { @@ -45,6 +46,11 @@ describe('WorkflowServerWritableStream', () => { mockStreams = { write: vi.fn().mockResolvedValue(undefined), writeMulti: vi.fn().mockResolvedValue(undefined), + writeLease: vi.fn().mockResolvedValue({ + base: 0, + committed: 0, + status: 'ok', + }), close: vi.fn().mockResolvedValue(undefined), }; @@ -54,6 +60,7 @@ describe('WorkflowServerWritableStream', () => { }); afterEach(() => { + delete process.env.WORKFLOW_STREAM_LEASE_FAST_PATH; setWorld(undefined); vi.clearAllMocks(); }); @@ -78,6 +85,113 @@ describe('WorkflowServerWritableStream', () => { }); }); + describe('stream lease fast path', () => { + it('is off by default and leaves legacy writes unchanged', async () => { + const stream = new WorkflowServerWritableStream('run-123', 'test-stream'); + const writer = stream.getWriter(); + await writer.write(new Uint8Array([1])); + await writer.close(); + + expect(mockStreams.writeLease).not.toHaveBeenCalled(); + expect(mockStreams.write).toHaveBeenCalledTimes(1); + }); + + it('keeps writer identity and increments sequence across flushes', async () => { + process.env.WORKFLOW_STREAM_LEASE_FAST_PATH = '1'; + const stream = new WorkflowServerWritableStream('run-123', 'test-stream'); + const writer = stream.getWriter(); + await writer.write(new Uint8Array([1])); + await writer.write(new Uint8Array([2])); + await writer.close(); + + expect(mockStreams.writeLease).toHaveBeenCalledTimes(2); + const first = mockStreams.writeLease.mock.calls[0][3]; + const second = mockStreams.writeLease.mock.calls[1][3]; + expect(first.writerId).toBe(second.writerId); + expect(first.epoch).toBe(second.epoch); + expect([first.seqStart, second.seqStart]).toEqual([0, 1]); + }); + + it('treats replay as success and permanently de-opts after need-reserve', async () => { + process.env.WORKFLOW_STREAM_LEASE_FAST_PATH = '1'; + mockStreams.writeLease + .mockResolvedValueOnce({ base: 0, committed: 0, status: 'replay' }) + .mockResolvedValueOnce({ + base: 1, + committed: 1, + status: 'need-reserve', + }); + const stream = new WorkflowServerWritableStream('run-123', 'test-stream'); + const writer = stream.getWriter(); + await writer.write(new Uint8Array([1])); + await writer.write(new Uint8Array([2])); + await writer.write(new Uint8Array([3])); + await writer.close(); + + expect(mockStreams.writeLease).toHaveBeenCalledTimes(2); + const legacyChunks = [ + ...mockStreams.write.mock.calls.map((call) => call[2]), + ...mockStreams.writeMulti.mock.calls.flatMap((call) => call[2]), + ]; + expect(legacyChunks.map((chunk) => chunk[0])).toEqual([2, 3]); + }); + + it('repairs a seq-gap from retained history in FIFO order', async () => { + process.env.WORKFLOW_STREAM_LEASE_FAST_PATH = '1'; + mockStreams.writeLease + .mockResolvedValueOnce({ base: 0, committed: 0, status: 'ok' }) + .mockResolvedValueOnce({ + base: 0, + committed: 0, + status: 'seq-gap', + expected: 0, + }) + .mockResolvedValueOnce({ base: 0, committed: 0, status: 'ok' }); + const stream = new WorkflowServerWritableStream('run-123', 'test-stream'); + const writer = stream.getWriter(); + await writer.write(new Uint8Array([1])); + await writer.write(new Uint8Array([2])); + await writer.close(); + + expect(mockStreams.writeLease).toHaveBeenCalledTimes(3); + expect( + mockStreams.writeLease.mock.calls.map((call) => call[3].seqStart) + ).toEqual([0, 1, 0]); + expect(mockStreams.writeLease.mock.calls[2][2]).toEqual([ + new Uint8Array([1]), + new Uint8Array([2]), + ]); + }); + + it('falls back from the entire missing suffix after seq-gap recovery loses its lease', async () => { + process.env.WORKFLOW_STREAM_LEASE_FAST_PATH = '1'; + mockStreams.writeLease + .mockResolvedValueOnce({ base: 0, committed: 0, status: 'ok' }) + .mockResolvedValueOnce({ + base: 0, + committed: 0, + status: 'seq-gap', + expected: 0, + }) + .mockResolvedValueOnce({ + base: 0, + committed: 0, + status: 'need-reserve', + }); + const stream = new WorkflowServerWritableStream('run-123', 'test-stream'); + const writer = stream.getWriter(); + await writer.write(new Uint8Array([1])); + await writer.write(new Uint8Array([2])); + await writer.close(); + + expect(mockStreams.writeMulti).toHaveBeenCalledTimes(1); + expect(mockStreams.writeMulti.mock.calls[0][2]).toEqual([ + new Uint8Array([1]), + new Uint8Array([2]), + ]); + }); + }); + describe('group-commit write behavior', () => { it('write() resolves on buffer entry; the leading chunk dispatches immediately (no window tax)', async () => { const stream = new WorkflowServerWritableStream('run-123', 'test-stream'); diff --git a/packages/world-vercel/src/http-client.test.ts b/packages/world-vercel/src/http-client.test.ts index 989653f314..1d2909c19b 100644 --- a/packages/world-vercel/src/http-client.test.ts +++ b/packages/world-vercel/src/http-client.test.ts @@ -19,6 +19,7 @@ import { isRecyclableTransportError, STREAM_AGENT_OPTIONS, STREAM_CLOSE_RETRY_OPTIONS, + STREAM_LEASE_RETRY_OPTIONS, STREAM_RETRY_OPTIONS, } from './http-client.js'; @@ -78,6 +79,13 @@ describe('getStreamDispatcher', () => { // unsafe close shapes awaiting in-flight backups) surface as retriable // 503s with the stream left durably closing. Without 5xx here, that 503 // rejects writer.close() and the stream stays fenced until run expiry. + it('retries idempotent lease appends on 5xx', () => { + expect(STREAM_LEASE_RETRY_OPTIONS.methods).toEqual(['PUT']); + for (const code of [429, 500, 502, 503, 504]) { + expect(STREAM_LEASE_RETRY_OPTIONS.statusCodes).toContain(code); + } + }); + it('retries stream close on 5xx (idempotent, and the close barrier depends on it)', () => { expect(STREAM_CLOSE_RETRY_OPTIONS.methods).toEqual(['PUT']); for (const code of [429, 500, 502, 503, 504]) { diff --git a/packages/world-vercel/src/http-client.ts b/packages/world-vercel/src/http-client.ts index a07ea7a652..4b42c15e0a 100644 --- a/packages/world-vercel/src/http-client.ts +++ b/packages/world-vercel/src/http-client.ts @@ -3,6 +3,7 @@ import type { APIConfig } from './utils.js'; let _dispatcher: RetryAgent | undefined; let _streamDispatcher: RetryAgent | undefined; +let _streamLeaseDispatcher: RetryAgent | undefined; let _streamCloseDispatcher: RetryAgent | undefined; /** @@ -218,6 +219,16 @@ export const STREAM_RETRY_OPTIONS: RetryHandler.RetryOptions = { statusCodes: [429], }; +/** + * Lease appends carry a writer sequence and are replay-safe, so unlike legacy + * stream appends they can retry a 5xx response without duplicating chunks. + */ +export const STREAM_LEASE_RETRY_OPTIONS: RetryHandler.RetryOptions = { + retryAfter: true, + methods: ['PUT'], + statusCodes: [429, 500, 502, 503, 504], +}; + /** * Retry options for stream CLOSE (the `X-Stream-Done` PUT). Unlike chunk * appends, close is idempotent on the server: a duplicate close of a @@ -556,6 +567,10 @@ export function getStreamDispatcher(config?: APIConfig): unknown { * shared close agent whose retry policy includes 5xx — close is idempotent * (see STREAM_CLOSE_RETRY_OPTIONS), unlike chunk appends. */ +export function getStreamLeaseDispatcher(config?: APIConfig): unknown { + return config?.dispatcher ?? getDefaultStreamLeaseDispatcher(); +} + export function getStreamCloseDispatcher(config?: APIConfig): unknown { return config?.dispatcher ?? getDefaultStreamCloseDispatcher(); } @@ -676,6 +691,11 @@ function getDefaultStreamDispatcher(): RetryAgent { } /** Shared agent for the idempotent stream close (5xx retriable). */ +function getDefaultStreamLeaseDispatcher(): RetryAgent { + _streamLeaseDispatcher ??= createStreamDispatcher(STREAM_LEASE_RETRY_OPTIONS); + return _streamLeaseDispatcher; +} + function getDefaultStreamCloseDispatcher(): RetryAgent { _streamCloseDispatcher ??= createStreamDispatcher(STREAM_CLOSE_RETRY_OPTIONS); return _streamCloseDispatcher; diff --git a/packages/world-vercel/src/streamer.test.ts b/packages/world-vercel/src/streamer.test.ts index f7ddac8d77..1c1e9b5e3e 100644 --- a/packages/world-vercel/src/streamer.test.ts +++ b/packages/world-vercel/src/streamer.test.ts @@ -181,6 +181,94 @@ vi.mock('./utils.js', () => ({ }), })); +describe('streams.writeLease', () => { + async function getStreamer() { + const { createStreamer } = await import('./streamer.js'); + return createStreamer(); + } + + afterEach(() => { + vi.restoreAllMocks(); + }); + + it('sends lease metadata query parameters and parses a successful response', async () => { + const fetchSpy = vi + .spyOn(globalThis, 'fetch') + .mockImplementation(async () => + Response.json({ base: 10, committed: 9, status: 'ok' }) + ); + const streamer = await getStreamer(); + const result = await streamer.streams.writeLease?.( + 'run-1', + 'stream-1', + [new Uint8Array([1]), new Uint8Array([2])], + { writerId: 'writer-1', epoch: 42, seqStart: 7 } + ); + + expect(result).toEqual({ base: 10, committed: 9, status: 'ok' }); + const url = new URL(fetchSpy.mock.calls[0][0] as string); + expect(url.searchParams.get('writerId')).toBe('writer-1'); + expect(url.searchParams.get('epoch')).toBe('42'); + expect(url.searchParams.get('seqStart')).toBe('7'); + expect(url.searchParams.get('count')).toBe('2'); + const headers = (fetchSpy.mock.calls[0][1] as RequestInit) + .headers as Headers; + expect(headers.get('X-Stream-Multi')).toBe('true'); + }); + + it('advances the lease sequence for paged writes', async () => { + const fetchSpy = vi + .spyOn(globalThis, 'fetch') + .mockImplementation(async () => + Response.json({ base: 0, committed: 0, status: 'ok' }) + ); + const streamer = await getStreamer(); + const chunks = Array.from( + { length: MAX_CHUNKS_PER_REQUEST + 1 }, + () => new Uint8Array([1]) + ); + await streamer.streams.writeLease?.('run-1', 'stream-1', chunks, { + writerId: 'writer-1', + epoch: 42, + seqStart: 7, + }); + + expect(fetchSpy).toHaveBeenCalledTimes(2); + const first = new URL(fetchSpy.mock.calls[0][0] as string); + const second = new URL(fetchSpy.mock.calls[1][0] as string); + expect(first.searchParams.get('seqStart')).toBe('7'); + expect(second.searchParams.get('seqStart')).toBe( + String(7 + MAX_CHUNKS_PER_REQUEST) + ); + }); + + it('returns a seq-gap result without appending later pages', async () => { + const fetchSpy = vi + .spyOn(globalThis, 'fetch') + .mockImplementation(async () => + Response.json({ base: 0, committed: 0, status: 'seq-gap', expected: 3 }) + ); + const streamer = await getStreamer(); + const chunks = Array.from( + { length: MAX_CHUNKS_PER_REQUEST + 1 }, + () => new Uint8Array([1]) + ); + const result = await streamer.streams.writeLease?.( + 'run-1', + 'stream-1', + chunks, + { + writerId: 'writer-1', + epoch: 42, + seqStart: 7, + } + ); + + expect(result).toMatchObject({ status: 'seq-gap', expected: 3 }); + expect(fetchSpy).toHaveBeenCalledTimes(1); + }); +}); + describe('streams.get', () => { async function getStreamer() { const { createStreamer } = await import('./streamer.js'); diff --git a/packages/world-vercel/src/streamer.ts b/packages/world-vercel/src/streamer.ts index 59538ea6d2..c10c9503c5 100644 --- a/packages/world-vercel/src/streamer.ts +++ b/packages/world-vercel/src/streamer.ts @@ -4,11 +4,14 @@ import { type StreamChunksResponse, type Streamer, type StreamInfoResponse, + type StreamWriteLease, + type StreamWriteLeaseResult, } from '@workflow/world'; import { z } from 'zod'; import { getStreamCloseDispatcher, getStreamDispatcher, + getStreamLeaseDispatcher, } from './http-client.js'; import { errorForResponse, @@ -183,6 +186,13 @@ const StreamInfoResponseSchema = z.object({ done: z.boolean(), }); +const StreamWriteLeaseResultSchema = z.object({ + base: z.number(), + committed: z.number(), + status: z.enum(['ok', 'replay', 'seq-gap', 'need-reserve']), + expected: z.number().optional(), +}); + /** * Zod schema for the paginated stream chunks response from the server. * When using CBOR (the default for makeRequest), chunk data arrives as @@ -287,6 +297,69 @@ export function createStreamer(config?: APIConfig): Streamer { } }, + async writeLease( + runId: string | Promise, + name: string, + chunks: (string | Uint8Array)[], + lease: StreamWriteLease + ): Promise { + if (chunks.length === 0) { + throw new Error('Stream lease append requires at least one chunk'); + } + const resolvedRunId = await runId; + const maxChunksPerRequest = getMaxChunksPerRequest(); + let lastResult: StreamWriteLeaseResult | undefined; + + for (let i = 0; i < chunks.length; i += maxChunksPerRequest) { + const batch = chunks.slice(i, i + maxChunksPerRequest); + const httpConfig = await getHttpConfig(config); + const headers = new Headers(httpConfig.headers); + if (batch.length > 1) headers.set('X-Stream-Multi', 'true'); + const url = getStreamUrl(name, resolvedRunId, httpConfig); + // The stream body is opaque chunk data (or the existing multi-frame + // format), so lease negotiation travels as additive query fields. + // All fields are present together: their presence opts one PUT into + // the server's lease path without changing the stream endpoint. + url.searchParams.set('writerId', lease.writerId); + url.searchParams.set('epoch', String(lease.epoch)); + url.searchParams.set('seqStart', String(lease.seqStart + i)); + url.searchParams.set('count', String(batch.length)); + const response = await instrumentedFetch({ + method: 'PUT', + url: url.toString(), + body: batch.length === 1 ? batch[0] : encodeMultiChunks(batch), + headers, + dispatcher: getStreamLeaseDispatcher(config), + timeoutMs: null, + logLabel: url.pathname, + spanName: 'workflow.stream.write', + durationAttribute: 'workflow.stream.write.chunk_rtt', + attributes: streamSpanAttributes({ + runId: resolvedRunId, + name, + operation: 'write_multi', + }), + buildError: async (res) => + createStreamRequestError('write', url, res, await res.text()), + }); + const result = StreamWriteLeaseResultSchema.parse( + await response.json() + ); + if (result.status === 'seq-gap' || result.status === 'need-reserve') { + // Earlier pages in this call have an acknowledged contiguous + // prefix. Core needs that count to avoid legacy-writing them if a + // later page de-opts. + return { ...result, acknowledged: i }; + } + lastResult = result; + } + + if (!lastResult) { + throw new Error('Stream lease append produced no result'); + } + return lastResult; + }, + async close(runId: string | Promise, name: string) { // Await runId if it's a promise to ensure proper flushing const resolvedRunId = await runId; diff --git a/packages/world-vercel/src/utils.ts b/packages/world-vercel/src/utils.ts index b9d46fae48..d43f3fafb0 100644 --- a/packages/world-vercel/src/utils.ts +++ b/packages/world-vercel/src/utils.ts @@ -154,11 +154,10 @@ export interface APIConfig { * version's dispatcher, or any object implementing the dispatcher contract. * * Note: when provided, this dispatcher replaces *every* default — including - * the one used for stream writes (the `PUT` write/close path). Stream appends - * are not idempotent, and undici's `RetryAgent` retries `PUT` on 5xx by - * default, which can duplicate a chunk the server already persisted. A custom - * dispatcher used with stream writes should therefore not retry `PUT` on 5xx - * (the built-in stream dispatcher retries only on transient errors and 429). + * the one used for stream writes (the `PUT` write/close path). Legacy stream + * appends are not idempotent, while lease appends are explicitly replay-safe. + * A custom dispatcher replaces either built-in policy, so callers choosing + * one for legacy writes should not retry `PUT` on 5xx. */ dispatcher?: unknown; projectConfig?: { diff --git a/packages/world/src/interfaces.ts b/packages/world/src/interfaces.ts index 33542825f5..643c24ba35 100644 --- a/packages/world/src/interfaces.ts +++ b/packages/world/src/interfaces.ts @@ -79,6 +79,18 @@ export interface Streamer { chunks: (string | Uint8Array)[] ): Promise; + /** + * Optional idempotent stream-append protocol. The core writable sink owns + * the writer identity and sequence; worlds only transport the lease + * metadata and report the server's protocol result. + */ + writeLease?( + runId: string, + name: string, + chunks: (string | Uint8Array)[], + lease: StreamWriteLease + ): Promise; + close(runId: string, name: string): Promise; /** @@ -482,6 +494,24 @@ export interface WorldCapabilities { /** * The "World" interface represents how Workflows are able to communicate with the outside world. */ +export interface StreamWriteLease { + writerId: string; + epoch: number; + seqStart: number; +} + +export interface StreamWriteLeaseResult { + status: 'ok' | 'replay' | 'seq-gap' | 'need-reserve'; + base: number; + committed: number; + expected?: number; + /** + * Chunks from this local write call already acknowledged by the fast path. + * This is transport-local bookkeeping for a paged call, not a server field. + */ + acknowledged?: number; +} + export interface World extends Queue, Streamer, Storage { /** * Optional analytics read namespace for observability surfaces.