Skip to content

Fix race condition in streamer when multiple writes share a promise runId#704

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
VaguelySerious merged 1 commit into from
Dec 30, 2025
Merged

Fix race condition in streamer when multiple writes share a promise runId #704

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions .changeset/quiet-streams-order.md
View file
Open in desktop
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
---
"@workflow/world-local": patch
---

Fix race condition in streamer when multiple writes share a promise runId.
6 changes: 4 additions & 2 deletions packages/world-local/src/streamer.test.ts
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -102,8 +102,10 @@ describe('streamer', () => {
const chunk = deserializeChunk(
await fs.readFile(`${testDir}/streams/chunks/${file}`)
);
const stream_id = String(file.split('-').at(-1)).split('.')[0];
const time = decodeTime(stream_id);
// Extract ULID from filename: "streamName-chnk_ULID.json"
const chunkIdPart = String(file.split('-').at(-1)).split('.')[0]; // "chnk_ULID"
const ulid = chunkIdPart.replace('chnk_', ''); // Just the ULID
const time = decodeTime(ulid);
const timeDiff = time - lastTime;
lastTime = time;

Expand Down
12 changes: 8 additions & 4 deletions packages/world-local/src/streamer.ts
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -93,14 +93,17 @@ export function createStreamer(basedir: string): Streamer {
_runId: string | Promise<string>,
chunk: string | Uint8Array
) {
// Generate ULID synchronously BEFORE any await to preserve call order.
// This ensures that chunks written in sequence maintain their order even
// when runId is a promise that multiple writes are waiting on.
const chunkId = `chnk_${monotonicUlid()}`;

// Await runId if it's a promise to ensure proper flushing
const runId = await _runId;

// Register this stream for the run
await registerStreamForRun(runId, name);

const chunkId = `chnk_${monotonicUlid()}`;

// Convert chunk to buffer for serialization
let chunkBuffer: Buffer;
if (typeof chunk === 'string') {
Expand DownExpand Up@@ -136,13 +139,14 @@ export function createStreamer(basedir: string): Streamer {
},

async closeStream(name: string, _runId: string | Promise<string>) {
// Generate ULID synchronously BEFORE any await to preserve call order.
const chunkId = `chnk_${monotonicUlid()}`;

// Await runId if it's a promise to ensure proper flushing
const runId = await _runId;

// Register this stream for the run (in case writeToStream wasn't called)
await registerStreamForRun(runId, name);

const chunkId = `chnk_${monotonicUlid()}`;
const chunkPath = path.join(
basedir,
'streams',
Expand Down
Loading