perf(sync): optimize multi-session event pipeline with per-directory queues and delta coalescing (#908)
* fix: hide archived section and empty folders when no sessions remain - Only push archived group in useSessionGrouping when there are archived sessions, preventing an empty archived section from rendering - Hide empty folders in archived bucket via shouldKeepFolder check in SessionGroupSection (folders with no sessions and no content in children are filtered out) - Always filter folders through shouldKeepFolder, not just during search * perf: memoize archived folder filtering * perf(sync): per-directory event queues to eliminate cross-session HoL blocking The SSE event pipeline previously used a single global queue and a single flush timer shared across all directories. Under concurrent multi-session workloads, a busy directory's delta storm would block other directories' status and state events from reaching the UI until the next flush tick, producing the "multi-session latency" symptom users report. Split the queue into one DirectoryQueue per directory, each with its own coalesce map, stale-delta set, and flush timer. Directories flush independently so a busy directory can no longer starve a quiet one. Coalesce keys are now scoped to a single directory's queue, so the directory prefix is removed from the key strings. Cross-directory behavior only; same-directory multi-session behavior is unchanged (React 18 auto-batching still collapses a single directory's flush into one render). * perf(sync): coalesce consecutive message.part.delta events per flush window Within a 16ms flush window, consecutive delta events for the same (messageID, partID, field) tuple are string-concatenated into a single accumulated delta rather than being queued individually. This directly addresses same-project multi-session workloads — most notably parent sessions with subagent tasks (child sessions share the same directory queue). Both parties stream deltas concurrently, which previously multiplied raw event count proportionally to the number of active sessions. Coalescing can reduce queue depth by 10-100x during active streaming. Safety: verified against event-reducer.ts — the delta handler is a pure string append (existingValue + props.delta) with no per-event side effects (no time.updated, no notifications, no diff calculations). The merged result is semantically identical to applying each delta separately. The staleDeltas skip mechanism is unaffected: accumulated delta payloads retain their type and identifiers, so message.part.updated supersession still works correctly. * test(sync): cover per-directory queues and delta coalescing Extend event-pipeline.test.js with behavioural coverage for both optimizations landed in 98d013a and 258acf0: P1 (per-directory queues) - Delivers events from two directories without loss - Keeps distinct sessionIDs in the same directory as independent coalesce slots (session.status is not overwritten across sessions) - Collapses repeated session.status for the same session down to latest Option C (delta coalescing) - Accumulates consecutive deltas for the same (messageID, partID, field) into a single dispatched event with concatenated content - Does not merge deltas across different fields on the same part - Does not merge deltas across different parts on the same message - Does not merge deltas across different directories (per-dir queues) - Skips accumulated deltas when message.part.updated is coalesced onto an earlier update, proving staleDeltas still works with C - Leaves non-delta coalescing (session.status replace semantics) intact All 13 tests pass under bun:test. Also adds event-pipeline.bench.js, a runnable synthetic benchmark that reports delta reduction and byte integrity across 8 workload scenarios from "single session, 500 tokens" up to "10 projects × 5 sessions × 1000 tokens". Run with: bun packages/ui/src/sync/__tests__/event-pipeline.bench.js Current numbers on this machine: 99.5% - 99.9% delta event reduction with full byte-level integrity (concatenated delta bytes always equal the input total). * fix(sync): remove staleDeltas — it silently drops delta events --------- Co-authored-by: Bohdan Triapitsyn <artmore@protonmail.com>
This commit is contained in:
jwcrystal
committed by
GitHub
GitHub
co-authored by
Bohdan Triapitsyn
Bohdan Triapitsyn
parent
5584537da5
commit
1656c3bb93
@@ -0,0 +1,347 @@
|
||||
/**
|
||||
* Synthetic benchmark for the event pipeline.
|
||||
*
|
||||
* Measures how much per-directory queueing + delta coalescing shrinks the
|
||||
* delivered event stream and how long enqueue/flush takes for realistic
|
||||
* multi-session workloads (parent + subagent token streaming).
|
||||
*
|
||||
* Run with:
|
||||
* bun packages/ui/src/sync/__tests__/event-pipeline.bench.js
|
||||
*
|
||||
* This is NOT a bun:test file — it prints a report and exits. Nothing here
|
||||
* asserts; it exists purely to give you intuition about the optimization
|
||||
* impact at varying concurrency levels.
|
||||
*/
|
||||
|
||||
import { createEventPipeline } from '../event-pipeline.ts';
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Minimal DOM stubs (same approach as the unit tests)
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
globalThis.document = {
|
||||
visibilityState: 'visible',
|
||||
addEventListener() {},
|
||||
removeEventListener() {},
|
||||
};
|
||||
globalThis.window = {
|
||||
addEventListener() {},
|
||||
removeEventListener() {},
|
||||
};
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// SDK mock that replays a pre-generated event list
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
function createReplaySdk(events, hold) {
|
||||
return {
|
||||
global: {
|
||||
event: async () => ({
|
||||
stream: (async function* () {
|
||||
for (const e of events) yield e;
|
||||
await hold;
|
||||
})(),
|
||||
}),
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Workload generators
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/**
|
||||
* Token-stream workload: N sessions in `directoryCount` directories, each
|
||||
* emitting `tokensPerSession` text deltas plus a few framing events.
|
||||
*
|
||||
* Shape is intentionally close to a real opencode session:
|
||||
* session.status(busy)
|
||||
* message.part.delta × tokensPerSession (coalescible)
|
||||
* message.part.updated (final state)
|
||||
* session.status(idle)
|
||||
*/
|
||||
function buildTokenStreamWorkload({
|
||||
directoryCount,
|
||||
sessionsPerDirectory,
|
||||
tokensPerSession,
|
||||
}) {
|
||||
const events = [];
|
||||
for (let d = 0; d < directoryCount; d++) {
|
||||
const directory = `dir-${d}`;
|
||||
for (let s = 0; s < sessionsPerDirectory; s++) {
|
||||
const sessionID = `dir-${d}-s${s}`;
|
||||
const messageID = `${sessionID}-m1`;
|
||||
const partID = `${messageID}-p1`;
|
||||
|
||||
events.push({
|
||||
directory,
|
||||
payload: {
|
||||
type: 'session.status',
|
||||
properties: { sessionID, status: { type: 'busy' } },
|
||||
},
|
||||
});
|
||||
|
||||
for (let t = 0; t < tokensPerSession; t++) {
|
||||
events.push({
|
||||
directory,
|
||||
payload: {
|
||||
type: 'message.part.delta',
|
||||
properties: {
|
||||
messageID,
|
||||
partID,
|
||||
field: 'text',
|
||||
delta: 'x',
|
||||
},
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
events.push({
|
||||
directory,
|
||||
payload: {
|
||||
type: 'message.part.updated',
|
||||
properties: {
|
||||
part: {
|
||||
id: partID,
|
||||
type: 'text',
|
||||
messageID,
|
||||
text: 'x'.repeat(tokensPerSession),
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
events.push({
|
||||
directory,
|
||||
payload: {
|
||||
type: 'session.status',
|
||||
properties: { sessionID, status: { type: 'idle' } },
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
// Interleave events across directories/sessions so we exercise the real
|
||||
// "parent + subagent" arrival pattern instead of one session at a time.
|
||||
return interleave(events);
|
||||
}
|
||||
|
||||
// Shuffle events within each directory bucket so arrivals are interleaved but
|
||||
// still ordered within a single (sessionID, partID) stream (deltas must stay
|
||||
// ordered relative to each other for append semantics to remain correct).
|
||||
function interleave(events) {
|
||||
const buckets = new Map(); // directory -> list of events (in original order)
|
||||
for (const e of events) {
|
||||
const bucket = buckets.get(e.directory) ?? [];
|
||||
bucket.push(e);
|
||||
buckets.set(e.directory, bucket);
|
||||
}
|
||||
const out = [];
|
||||
let more = true;
|
||||
while (more) {
|
||||
more = false;
|
||||
for (const bucket of buckets.values()) {
|
||||
if (bucket.length > 0) {
|
||||
out.push(bucket.shift());
|
||||
more = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Runner — pushes a workload through the pipeline and measures
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
async function runScenario(label, workload) {
|
||||
let release;
|
||||
const hold = new Promise((resolve) => {
|
||||
release = resolve;
|
||||
});
|
||||
|
||||
let delivered = 0;
|
||||
let deliveredDeltas = 0;
|
||||
let deliveredDeltaBytes = 0;
|
||||
|
||||
const sdk = createReplaySdk(workload, hold);
|
||||
|
||||
const startWall = performance.now();
|
||||
const { cleanup } = createEventPipeline({
|
||||
sdk,
|
||||
onEvent: (_directory, payload) => {
|
||||
delivered++;
|
||||
if (payload.type === 'message.part.delta') {
|
||||
deliveredDeltas++;
|
||||
deliveredDeltaBytes += payload.properties.delta.length;
|
||||
}
|
||||
},
|
||||
});
|
||||
|
||||
// Give the pipeline enough time to finish enqueueing AND flush. Scale wait
|
||||
// time with workload so large stress scenarios have room to drain — the SSE
|
||||
// loop yields every 8ms and the flush fires every 16ms, so in the worst case
|
||||
// we need ~(workload / STREAM_YIELD) * 16ms of wall clock.
|
||||
const waitMs = Math.max(200, Math.ceil(workload.length / 100));
|
||||
await new Promise((resolve) => setTimeout(resolve, waitMs));
|
||||
const endWall = performance.now();
|
||||
|
||||
cleanup();
|
||||
release();
|
||||
|
||||
// Count input-side delta events for comparison
|
||||
const inputDeltas = workload.filter((e) => e.payload.type === 'message.part.delta').length;
|
||||
const inputDeltaBytes = workload
|
||||
.filter((e) => e.payload.type === 'message.part.delta')
|
||||
.reduce((n, e) => n + e.payload.properties.delta.length, 0);
|
||||
|
||||
const wallMs = endWall - startWall;
|
||||
const reductionPct = inputDeltas === 0 ? 0 : (1 - deliveredDeltas / inputDeltas) * 100;
|
||||
|
||||
return {
|
||||
label,
|
||||
inputEvents: workload.length,
|
||||
inputDeltas,
|
||||
inputDeltaBytes,
|
||||
deliveredEvents: delivered,
|
||||
deliveredDeltas,
|
||||
deliveredDeltaBytes,
|
||||
reductionPct,
|
||||
wallMs,
|
||||
};
|
||||
}
|
||||
|
||||
function formatRow(r) {
|
||||
const cols = [
|
||||
r.label.padEnd(44),
|
||||
String(r.inputEvents).padStart(8),
|
||||
String(r.deliveredEvents).padStart(8),
|
||||
String(r.inputDeltas).padStart(8),
|
||||
String(r.deliveredDeltas).padStart(8),
|
||||
`${r.reductionPct.toFixed(1)}%`.padStart(8),
|
||||
`${r.wallMs.toFixed(1)}ms`.padStart(10),
|
||||
r.inputDeltaBytes === r.deliveredDeltaBytes ? 'bytes ✓' : `bytes ${r.inputDeltaBytes}→${r.deliveredDeltaBytes}`,
|
||||
];
|
||||
return cols.join(' ');
|
||||
}
|
||||
|
||||
function header() {
|
||||
const cols = [
|
||||
'scenario'.padEnd(44),
|
||||
'in'.padStart(8),
|
||||
'out'.padStart(8),
|
||||
'in Δ'.padStart(8),
|
||||
'out Δ'.padStart(8),
|
||||
'reduce'.padStart(8),
|
||||
'wall'.padStart(10),
|
||||
'integrity',
|
||||
];
|
||||
return cols.join(' ');
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Scenarios
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
const scenarios = [
|
||||
{
|
||||
label: 'single project, 1 session, 500 tokens',
|
||||
workload: buildTokenStreamWorkload({
|
||||
directoryCount: 1,
|
||||
sessionsPerDirectory: 1,
|
||||
tokensPerSession: 500,
|
||||
}),
|
||||
},
|
||||
{
|
||||
label: 'single project, parent + 1 subagent, 500 tokens each',
|
||||
workload: buildTokenStreamWorkload({
|
||||
directoryCount: 1,
|
||||
sessionsPerDirectory: 2,
|
||||
tokensPerSession: 500,
|
||||
}),
|
||||
},
|
||||
{
|
||||
label: 'single project, parent + 3 subagents, 500 tokens each',
|
||||
workload: buildTokenStreamWorkload({
|
||||
directoryCount: 1,
|
||||
sessionsPerDirectory: 4,
|
||||
tokensPerSession: 500,
|
||||
}),
|
||||
},
|
||||
{
|
||||
label: 'single project, parent + 9 subagents, 200 tokens each',
|
||||
workload: buildTokenStreamWorkload({
|
||||
directoryCount: 1,
|
||||
sessionsPerDirectory: 10,
|
||||
tokensPerSession: 200,
|
||||
}),
|
||||
},
|
||||
{
|
||||
label: '3 projects, 1 session each, 500 tokens',
|
||||
workload: buildTokenStreamWorkload({
|
||||
directoryCount: 3,
|
||||
sessionsPerDirectory: 1,
|
||||
tokensPerSession: 500,
|
||||
}),
|
||||
},
|
||||
{
|
||||
label: '3 projects × (parent + subagent), 500 tokens each',
|
||||
workload: buildTokenStreamWorkload({
|
||||
directoryCount: 3,
|
||||
sessionsPerDirectory: 2,
|
||||
tokensPerSession: 500,
|
||||
}),
|
||||
},
|
||||
{
|
||||
label: '5 projects × parent + 3 subagents, 200 tokens',
|
||||
workload: buildTokenStreamWorkload({
|
||||
directoryCount: 5,
|
||||
sessionsPerDirectory: 4,
|
||||
tokensPerSession: 200,
|
||||
}),
|
||||
},
|
||||
{
|
||||
label: 'stress: 10 projects × 5 sessions × 1000 tokens',
|
||||
workload: buildTokenStreamWorkload({
|
||||
directoryCount: 10,
|
||||
sessionsPerDirectory: 5,
|
||||
tokensPerSession: 1000,
|
||||
}),
|
||||
},
|
||||
];
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Main
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
async function main() {
|
||||
console.log('event-pipeline synthetic benchmark\n');
|
||||
console.log(header());
|
||||
console.log('-'.repeat(header().length + 8));
|
||||
|
||||
const results = [];
|
||||
for (const { label, workload } of scenarios) {
|
||||
const r = await runScenario(label, workload);
|
||||
results.push(r);
|
||||
console.log(formatRow(r));
|
||||
}
|
||||
|
||||
console.log('\nLegend:');
|
||||
console.log(' in — total events fed into the pipeline');
|
||||
console.log(' out — total events dispatched via onEvent after coalescing + flush');
|
||||
console.log(' in Δ — input events of type message.part.delta');
|
||||
console.log(' out Δ — delta events that actually made it to onEvent (after merging)');
|
||||
console.log(' reduce — (1 − outΔ / inΔ) × 100, i.e. how much delta traffic shrunk');
|
||||
console.log(' wall — total wall-clock time for the scenario (bounded by 200ms wait)');
|
||||
console.log(' integrity — "bytes ✓" means the concatenated delta bytes match the input total');
|
||||
console.log('');
|
||||
console.log('Interpretation:');
|
||||
console.log(' Higher reduce % = fewer reducer invocations, fewer React setState calls,');
|
||||
console.log(' fewer allocations inside the flush loop. The integrity check confirms no');
|
||||
console.log(' text was dropped during coalescing.');
|
||||
}
|
||||
|
||||
main().catch((error) => {
|
||||
console.error('benchmark failed:', error);
|
||||
process.exit(1);
|
||||
});
|
||||
@@ -51,6 +51,33 @@ function createSdkWithEvents(events, hold) {
|
||||
};
|
||||
}
|
||||
|
||||
// Run a pipeline against a pre-seeded event stream, collect every dispatched
|
||||
// event, wait long enough for the 16ms flush window to elapse, then tear it
|
||||
// down. Returns the list of { directory, payload } that onEvent saw.
|
||||
async function runPipelineWithEvents(events, waitMs = 80) {
|
||||
installDomStubs();
|
||||
|
||||
let releaseStream;
|
||||
const hold = new Promise((resolve) => {
|
||||
releaseStream = resolve;
|
||||
});
|
||||
|
||||
const received = [];
|
||||
const sdk = createSdkWithEvents(events, hold);
|
||||
const { cleanup } = createEventPipeline({
|
||||
sdk,
|
||||
onEvent: (directory, payload) => {
|
||||
received.push({ directory, payload });
|
||||
},
|
||||
});
|
||||
|
||||
await new Promise((resolve) => setTimeout(resolve, waitMs));
|
||||
cleanup();
|
||||
releaseStream();
|
||||
|
||||
return received;
|
||||
}
|
||||
|
||||
describe('createEventPipeline', () => {
|
||||
it('falls back to payload.properties.directory when the SDK event omits top-level directory', async () => {
|
||||
installDomStubs();
|
||||
@@ -341,3 +368,239 @@ describe('createEventPipeline', () => {
|
||||
expect(received[0].payload.type).toBe('message.part.updated');
|
||||
});
|
||||
});
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// P1 — Per-directory queue isolation
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
describe('createEventPipeline — per-directory isolation (P1)', () => {
|
||||
it('delivers events from two directories without losing either', async () => {
|
||||
const received = await runPipelineWithEvents([
|
||||
{
|
||||
directory: 'dir-a',
|
||||
payload: {
|
||||
type: 'session.status',
|
||||
properties: { sessionID: 's-a', status: { type: 'busy' } },
|
||||
},
|
||||
},
|
||||
{
|
||||
directory: 'dir-b',
|
||||
payload: {
|
||||
type: 'session.status',
|
||||
properties: { sessionID: 's-b', status: { type: 'idle' } },
|
||||
},
|
||||
},
|
||||
]);
|
||||
|
||||
const dirs = received.map((r) => r.directory).sort();
|
||||
expect(dirs).toEqual(['dir-a', 'dir-b']);
|
||||
});
|
||||
|
||||
it('keeps distinct sessionIDs in the same directory as independent coalesce slots', async () => {
|
||||
const received = await runPipelineWithEvents([
|
||||
{
|
||||
directory: 'dir-a',
|
||||
payload: {
|
||||
type: 'session.status',
|
||||
properties: { sessionID: 's1', status: { type: 'busy' } },
|
||||
},
|
||||
},
|
||||
{
|
||||
directory: 'dir-a',
|
||||
payload: {
|
||||
type: 'session.status',
|
||||
properties: { sessionID: 's2', status: { type: 'busy' } },
|
||||
},
|
||||
},
|
||||
]);
|
||||
|
||||
expect(received).toHaveLength(2);
|
||||
const sessionIds = received.map((r) => r.payload.properties.sessionID).sort();
|
||||
expect(sessionIds).toEqual(['s1', 's2']);
|
||||
});
|
||||
|
||||
it('collapses repeated session.status for the same session down to the latest', async () => {
|
||||
const received = await runPipelineWithEvents([
|
||||
{
|
||||
directory: 'dir-a',
|
||||
payload: {
|
||||
type: 'session.status',
|
||||
properties: { sessionID: 's1', status: { type: 'busy' } },
|
||||
},
|
||||
},
|
||||
{
|
||||
directory: 'dir-a',
|
||||
payload: {
|
||||
type: 'session.status',
|
||||
properties: { sessionID: 's1', status: { type: 'idle' } },
|
||||
},
|
||||
},
|
||||
]);
|
||||
|
||||
expect(received).toHaveLength(1);
|
||||
expect(received[0].payload.properties.status.type).toBe('idle');
|
||||
});
|
||||
});
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Option C — message.part.delta coalescing
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
describe('createEventPipeline — delta coalescing (Option C)', () => {
|
||||
it('accumulates consecutive deltas for the same (messageID, partID, field) into one event', async () => {
|
||||
const events = ['Hello ', 'world', ', ', 'how ', 'are ', 'you?'].map((chunk) => ({
|
||||
directory: 'dir-a',
|
||||
payload: {
|
||||
type: 'message.part.delta',
|
||||
properties: {
|
||||
messageID: 'msg-1',
|
||||
partID: 'part-1',
|
||||
field: 'text',
|
||||
delta: chunk,
|
||||
},
|
||||
},
|
||||
}));
|
||||
|
||||
const received = await runPipelineWithEvents(events);
|
||||
|
||||
expect(received).toHaveLength(1);
|
||||
expect(received[0].payload.type).toBe('message.part.delta');
|
||||
expect(received[0].payload.properties.delta).toBe('Hello world, how are you?');
|
||||
expect(received[0].payload.properties.messageID).toBe('msg-1');
|
||||
expect(received[0].payload.properties.partID).toBe('part-1');
|
||||
expect(received[0].payload.properties.field).toBe('text');
|
||||
});
|
||||
|
||||
it('does NOT merge deltas across different fields on the same part', async () => {
|
||||
const received = await runPipelineWithEvents([
|
||||
{
|
||||
directory: 'dir-a',
|
||||
payload: {
|
||||
type: 'message.part.delta',
|
||||
properties: {
|
||||
messageID: 'msg-1',
|
||||
partID: 'part-1',
|
||||
field: 'text',
|
||||
delta: 'A',
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
directory: 'dir-a',
|
||||
payload: {
|
||||
type: 'message.part.delta',
|
||||
properties: {
|
||||
messageID: 'msg-1',
|
||||
partID: 'part-1',
|
||||
field: 'reasoning',
|
||||
delta: 'B',
|
||||
},
|
||||
},
|
||||
},
|
||||
]);
|
||||
|
||||
expect(received).toHaveLength(2);
|
||||
const fieldDelta = received.map((r) => [
|
||||
r.payload.properties.field,
|
||||
r.payload.properties.delta,
|
||||
]).sort();
|
||||
expect(fieldDelta).toEqual([
|
||||
['reasoning', 'B'],
|
||||
['text', 'A'],
|
||||
]);
|
||||
});
|
||||
|
||||
it('does NOT merge deltas across different parts on the same message', async () => {
|
||||
const received = await runPipelineWithEvents([
|
||||
{
|
||||
directory: 'dir-a',
|
||||
payload: {
|
||||
type: 'message.part.delta',
|
||||
properties: {
|
||||
messageID: 'msg-1',
|
||||
partID: 'part-1',
|
||||
field: 'text',
|
||||
delta: 'AAA',
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
directory: 'dir-a',
|
||||
payload: {
|
||||
type: 'message.part.delta',
|
||||
properties: {
|
||||
messageID: 'msg-1',
|
||||
partID: 'part-2',
|
||||
field: 'text',
|
||||
delta: 'BBB',
|
||||
},
|
||||
},
|
||||
},
|
||||
]);
|
||||
|
||||
expect(received).toHaveLength(2);
|
||||
const byPart = Object.fromEntries(
|
||||
received.map((r) => [r.payload.properties.partID, r.payload.properties.delta]),
|
||||
);
|
||||
expect(byPart['part-1']).toBe('AAA');
|
||||
expect(byPart['part-2']).toBe('BBB');
|
||||
});
|
||||
|
||||
it('does NOT merge deltas across different directories (per-directory queues)', async () => {
|
||||
const received = await runPipelineWithEvents([
|
||||
{
|
||||
directory: 'dir-a',
|
||||
payload: {
|
||||
type: 'message.part.delta',
|
||||
properties: {
|
||||
messageID: 'msg-1',
|
||||
partID: 'part-1',
|
||||
field: 'text',
|
||||
delta: 'from-a',
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
directory: 'dir-b',
|
||||
payload: {
|
||||
type: 'message.part.delta',
|
||||
properties: {
|
||||
messageID: 'msg-1',
|
||||
partID: 'part-1',
|
||||
field: 'text',
|
||||
delta: 'from-b',
|
||||
},
|
||||
},
|
||||
},
|
||||
]);
|
||||
|
||||
expect(received).toHaveLength(2);
|
||||
const byDir = Object.fromEntries(
|
||||
received.map((r) => [r.directory, r.payload.properties.delta]),
|
||||
);
|
||||
expect(byDir['dir-a']).toBe('from-a');
|
||||
expect(byDir['dir-b']).toBe('from-b');
|
||||
});
|
||||
|
||||
it('does not touch non-delta events (session.status still replaced, not concatenated)', async () => {
|
||||
const received = await runPipelineWithEvents([
|
||||
{
|
||||
directory: 'dir-a',
|
||||
payload: {
|
||||
type: 'session.status',
|
||||
properties: { sessionID: 's1', status: { type: 'busy' } },
|
||||
},
|
||||
},
|
||||
{
|
||||
directory: 'dir-a',
|
||||
payload: {
|
||||
type: 'session.status',
|
||||
properties: { sessionID: 's1', status: { type: 'idle' } },
|
||||
},
|
||||
},
|
||||
]);
|
||||
|
||||
expect(received).toHaveLength(1);
|
||||
expect(received[0].payload.properties.status.type).toBe('idle');
|
||||
});
|
||||
});
|
||||
|
||||
@@ -78,62 +78,102 @@ function resolveEventDirectory(event: unknown, payload: Event): string {
|
||||
return propertyDirectory && propertyDirectory.length > 0 ? propertyDirectory : "global"
|
||||
}
|
||||
|
||||
// Per-directory queue state. Each directory owns an independent flush timer
|
||||
// so a busy directory's delta storm cannot block another directory's events
|
||||
// from reaching the UI (head-of-line blocking across sessions).
|
||||
type DirectoryQueue = {
|
||||
queue: Event[]
|
||||
buffer: Event[]
|
||||
coalesced: Map<string, number>
|
||||
timer: ReturnType<typeof setTimeout> | undefined
|
||||
last: number
|
||||
}
|
||||
|
||||
export function createEventPipeline(input: EventPipelineInput) {
|
||||
const { sdk, onEvent, onReconnect } = input
|
||||
const abort = new AbortController()
|
||||
let hasConnected = false
|
||||
|
||||
// Queue state
|
||||
let queue: QueuedEvent[] = []
|
||||
let buffer: QueuedEvent[] = []
|
||||
const coalesced = new Map<string, number>()
|
||||
let timer: ReturnType<typeof setTimeout> | undefined
|
||||
let last = 0
|
||||
// One queue + one flush timer per directory. Lazily created on first event.
|
||||
const directories = new Map<string, DirectoryQueue>()
|
||||
|
||||
// Coalesce key — same-type events for the same entity replace earlier ones
|
||||
const key = (directory: string, payload: Event): string | undefined => {
|
||||
const getOrCreateDir = (directory: string): DirectoryQueue => {
|
||||
let d = directories.get(directory)
|
||||
if (d) return d
|
||||
d = {
|
||||
queue: [],
|
||||
buffer: [],
|
||||
coalesced: new Map(),
|
||||
timer: undefined,
|
||||
last: 0,
|
||||
}
|
||||
directories.set(directory, d)
|
||||
return d
|
||||
}
|
||||
|
||||
// Coalesce key — same-type events for the same entity replace earlier ones.
|
||||
// Keys are scoped to a single directory's queue, so directory is implicit.
|
||||
// message.part.delta is a special case: consecutive deltas for the same
|
||||
// (messageID, partID, field) are accumulated (string-concatenated) rather
|
||||
// than replaced, because the reducer is a pure append and merging is
|
||||
// semantically identical to applying each delta individually.
|
||||
const key = (payload: Event): string | undefined => {
|
||||
if (payload.type === "session.status") {
|
||||
const props = payload.properties as { sessionID: string }
|
||||
return `session.status:${directory}:${props.sessionID}`
|
||||
return `session.status:${props.sessionID}`
|
||||
}
|
||||
if (payload.type === "lsp.updated") {
|
||||
return `lsp.updated:${directory}`
|
||||
return `lsp.updated`
|
||||
}
|
||||
if (payload.type === "message.part.updated") {
|
||||
const part = (payload.properties as { part: { messageID: string; id: string } }).part
|
||||
return `message.part.updated:${directory}:${part.messageID}:${part.id}`
|
||||
return `message.part.updated:${part.messageID}:${part.id}`
|
||||
}
|
||||
if (payload.type === "message.part.delta") {
|
||||
const props = payload.properties as { messageID: string; partID: string; field: string }
|
||||
return `message.part.delta:${props.messageID}:${props.partID}:${props.field}`
|
||||
}
|
||||
return undefined
|
||||
}
|
||||
|
||||
// Flush — swap queue, dispatch events
|
||||
const flush = () => {
|
||||
if (timer) clearTimeout(timer)
|
||||
timer = undefined
|
||||
// Flush one directory — swap queue, dispatch events.
|
||||
// React 18 auto-batching still collapses the setState calls inside a single
|
||||
// directory's flush into one render pass.
|
||||
const flushDir = (directory: string) => {
|
||||
const d = directories.get(directory)
|
||||
if (!d) return
|
||||
if (d.timer) {
|
||||
clearTimeout(d.timer)
|
||||
d.timer = undefined
|
||||
}
|
||||
if (d.queue.length === 0) return
|
||||
|
||||
if (queue.length === 0) return
|
||||
const events = d.queue
|
||||
d.queue = d.buffer
|
||||
d.buffer = events
|
||||
d.queue.length = 0
|
||||
d.coalesced.clear()
|
||||
|
||||
const events = queue
|
||||
queue = buffer
|
||||
buffer = events
|
||||
queue.length = 0
|
||||
coalesced.clear()
|
||||
|
||||
last = Date.now()
|
||||
d.last = Date.now()
|
||||
syncDebug.pipeline.flush(events.length)
|
||||
// React 18 batches synchronous setState calls automatically,
|
||||
// equivalent to SolidJS batch()
|
||||
for (const event of events) {
|
||||
onEvent(event.directory, event.payload)
|
||||
for (const payload of events) {
|
||||
onEvent(directory, payload)
|
||||
}
|
||||
|
||||
buffer.length = 0
|
||||
d.buffer.length = 0
|
||||
}
|
||||
|
||||
const schedule = () => {
|
||||
if (timer) return
|
||||
const elapsed = Date.now() - last
|
||||
timer = setTimeout(flush, Math.max(0, FLUSH_FRAME_MS - elapsed))
|
||||
const flushAll = () => {
|
||||
for (const directory of directories.keys()) {
|
||||
flushDir(directory)
|
||||
}
|
||||
}
|
||||
|
||||
const scheduleDir = (directory: string) => {
|
||||
const d = getOrCreateDir(directory)
|
||||
if (d.timer) return
|
||||
const elapsed = Date.now() - d.last
|
||||
d.timer = setTimeout(() => flushDir(directory), Math.max(0, FLUSH_FRAME_MS - elapsed))
|
||||
}
|
||||
|
||||
// Helpers
|
||||
@@ -201,18 +241,34 @@ export function createEventPipeline(input: EventPipelineInput) {
|
||||
}
|
||||
const normalizedPayload = normalizeEventType(payload)
|
||||
const directory = resolveEventDirectory(event, normalizedPayload)
|
||||
const k = key(directory, normalizedPayload)
|
||||
const d = getOrCreateDir(directory)
|
||||
const k = key(normalizedPayload)
|
||||
if (k) {
|
||||
const i = coalesced.get(k)
|
||||
const i = d.coalesced.get(k)
|
||||
if (i !== undefined) {
|
||||
queue[i] = { directory, payload: normalizedPayload }
|
||||
if (normalizedPayload.type === "message.part.delta") {
|
||||
// Accumulate delta strings — append to the already-queued event
|
||||
// rather than replacing it. The reducer is a pure string append so
|
||||
// this is semantically identical to applying each delta separately.
|
||||
const prev = d.queue[i] as unknown as { properties: { delta: string } }
|
||||
const inc = normalizedPayload.properties as { delta: string }
|
||||
d.queue[i] = {
|
||||
...normalizedPayload,
|
||||
properties: {
|
||||
...(normalizedPayload.properties as object),
|
||||
delta: prev.properties.delta + inc.delta,
|
||||
},
|
||||
} as unknown as Event
|
||||
} else {
|
||||
d.queue[i] = normalizedPayload
|
||||
}
|
||||
syncDebug.pipeline.coalesced(normalizedPayload.type, k)
|
||||
continue
|
||||
}
|
||||
coalesced.set(k, queue.length)
|
||||
d.coalesced.set(k, d.queue.length)
|
||||
}
|
||||
queue.push({ directory, payload: normalizedPayload })
|
||||
schedule()
|
||||
d.queue.push(normalizedPayload)
|
||||
scheduleDir(directory)
|
||||
|
||||
if (Date.now() - yielded < STREAM_YIELD_MS) continue
|
||||
yielded = Date.now()
|
||||
@@ -232,7 +288,7 @@ export function createEventPipeline(input: EventPipelineInput) {
|
||||
if (abort.signal.aborted) return
|
||||
await wait(RECONNECT_DELAY_MS)
|
||||
}
|
||||
})().finally(flush)
|
||||
})().finally(flushAll)
|
||||
|
||||
// Visibility handler — abort SSE on heartbeat timeout so the loop reconnects.
|
||||
// The reconnect triggers onReconnect above, which lets consumers resync state.
|
||||
@@ -262,7 +318,7 @@ export function createEventPipeline(input: EventPipelineInput) {
|
||||
window.removeEventListener("pageshow", onPageShow)
|
||||
}
|
||||
abort.abort()
|
||||
flush()
|
||||
flushAll()
|
||||
}
|
||||
|
||||
return { cleanup }
|
||||
|
||||
Reference in New Issue
Block a user