From b459a598aed124a981d0f3f04d97ff5aaaa38921 Mon Sep 17 00:00:00 2001 From: kyscott18 <43524469+kyscott18@users.noreply.github.com> Date: Tue, 13 Oct 2026 18:27:15 +0000 Subject: [PATCH 1/2] perf(runtime): wake the FIFO batch loop when a mutation arrives FIFO mode polled the received-mutation queue every 4ms, so each mutation waited for the next tick. That wait was most of its latency, and the loop woke 250 times per second when idle. FIFO now blocks on Queue.takeAll and accepts everything that queued during the previous batch. Batch mode keeps its fixed window for batchOrder. Both modes take mutations outside the speculative-state lock. Counter benchmark (sequential FIFO): 4.04ms -> 1.56ms average, 247 -> 641 mutations/s. --- packages/typewriter/src/runtime.ts | 201 ++++++++++++++++------------- 1 file changed, 108 insertions(+), 93 deletions(-) diff --git a/packages/typewriter/src/runtime.ts b/packages/typewriter/src/runtime.ts index c468739..bc3e614 100644 --- a/packages/typewriter/src/runtime.ts +++ b/packages/typewriter/src/runtime.ts @@ -76,7 +76,6 @@ import type { import type { LocalLog } from "./watch"; import { Watch } from "./watch"; -const FIFO_BATCH_INTERVAL_MS = 4; const SUBMIT_RETRY_TIMES = 5; type RuntimeExecuteInput = { @@ -636,99 +635,106 @@ export function createRuntimeEffect( }); } - const acceptBatch = withSpeculativeStateLock( - Effect.gen(function* () { - const batchStartedAtMs = startTimer(); - const submittedMutations = yield* Queue.clear(receivedMutationQueue); + type ReceivedMutationEntry = { + mutationId: number; + deferred: Deferred.Deferred; + }; - if (submittedMutations.length === 0) { - return; - } + const acceptBatch = (submittedMutations: ReceivedMutationEntry[]) => + withSpeculativeStateLock( + Effect.gen(function* () { + const batchStartedAtMs = startTimer(); - if (batchOrder.length > 0) { - submittedMutations.sort((a, b) => { - const aName = (mutationsById.get(a.mutationId)! as ReceivedMutation) - .name; - const bName = (mutationsById.get(b.mutationId)! as ReceivedMutation) - .name; - return ( - (batchOrderByName.get(aName) ?? batchOrder.length) - - (batchOrderByName.get(bName) ?? batchOrder.length) - ); - }); - } + if (submittedMutations.length === 0) { + return; + } - const acceptedMutations: { - mutation: AcceptedMutation; - executeResult: ExecuteResult; - preimages: readonly Hex.Hex[]; - deferred: Deferred.Deferred; - }[] = []; - - for (const { mutationId, deferred } of submittedMutations) { - const mutation = mutationsById.get(mutationId)! as ReceivedMutation; - const result = yield* Effect.result(acceptMutation(mutation)); - if (Result.isSuccess(result)) { - acceptedMutations.push({ ...result.success, deferred }); - } else { - yield* Deferred.fail(deferred, result.failure); + if (batchOrder.length > 0) { + submittedMutations.sort((a, b) => { + const aName = ( + mutationsById.get(a.mutationId)! as ReceivedMutation + ).name; + const bName = ( + mutationsById.get(b.mutationId)! as ReceivedMutation + ).name; + return ( + (batchOrderByName.get(aName) ?? batchOrder.length) - + (batchOrderByName.get(bName) ?? batchOrder.length) + ); + }); } - } - if (acceptedMutations.length === 0) { - return; - } + const acceptedMutations: { + mutation: AcceptedMutation; + executeResult: ExecuteResult; + preimages: readonly Hex.Hex[]; + deferred: Deferred.Deferred; + }[] = []; - const batch: RuntimeBatch = { - status: "accepted", - id: batchId++, - position: batchPosition++, - mutations: acceptedMutations.map(({ mutation }) => mutation), - }; + for (const { mutationId, deferred } of submittedMutations) { + const mutation = mutationsById.get(mutationId)! as ReceivedMutation; + const result = yield* Effect.result(acceptMutation(mutation)); + if (Result.isSuccess(result)) { + acceptedMutations.push({ ...result.success, deferred }); + } else { + yield* Deferred.fail(deferred, result.failure); + } + } - yield* Effect.logDebug("batched mutations").pipe( - Effect.annotateLogs({ - id: batch.id, - mutationCount: batch.mutations.length, - duration: durationMs(batchStartedAtMs), - }), - ); + if (acceptedMutations.length === 0) { + return; + } - batchesById.set(batch.id, batch); + const batch: RuntimeBatch = { + status: "accepted", + id: batchId++, + position: batchPosition++, + mutations: acceptedMutations.map(({ mutation }) => mutation), + }; - emitBatch(batchToEvent(batch)); + yield* Effect.logDebug("batched mutations").pipe( + Effect.annotateLogs({ + id: batch.id, + mutationCount: batch.mutations.length, + duration: durationMs(batchStartedAtMs), + }), + ); - const persistStartedAtMs = startTimer(); - yield* db.transaction((tx) => - Effect.gen(function* () { - yield* insertMutations( - tx, - schema, - acceptedMutations.map((entry) => entry.mutation), - ); - yield* insertSlotWritesMany(tx, schema, acceptedMutations); - yield* insertKeccakPreimages( - tx, - schema, - acceptedMutations.flatMap(({ preimages }) => preimages), - ); - }), - ); + batchesById.set(batch.id, batch); - for (const { mutation, deferred } of acceptedMutations) { - yield* Deferred.succeed(deferred, mutation); - } + emitBatch(batchToEvent(batch)); - yield* Effect.logDebug("persisted accepted mutations").pipe( - Effect.annotateLogs({ - mutationCount: acceptedMutations.length, - duration: durationMs(persistStartedAtMs), - }), - ); + const persistStartedAtMs = startTimer(); + yield* db.transaction((tx) => + Effect.gen(function* () { + yield* insertMutations( + tx, + schema, + acceptedMutations.map((entry) => entry.mutation), + ); + yield* insertSlotWritesMany(tx, schema, acceptedMutations); + yield* insertKeccakPreimages( + tx, + schema, + acceptedMutations.flatMap(({ preimages }) => preimages), + ); + }), + ); - yield* Queue.offer(acceptedBatchesQueue, batch.id); - }), - ); + for (const { mutation, deferred } of acceptedMutations) { + yield* Deferred.succeed(deferred, mutation); + } + + yield* Effect.logDebug("persisted accepted mutations").pipe( + Effect.annotateLogs({ + mutationCount: acceptedMutations.length, + duration: durationMs(persistStartedAtMs), + }), + ); + + yield* Queue.offer(acceptedBatchesQueue, batch.id); + }), + ); const submit = Effect.gen(function* () { const simulation = yield* withSpeculativeStateLock( @@ -1086,20 +1092,29 @@ export function createRuntimeEffect( ); }).pipe(Effect.withLogSpan("watch")); - const batchIntervalMs = - app.sequencing.order === "batch" - ? app.sequencing.batchIntervalMs - : FIFO_BATCH_INTERVAL_MS; const submitIntervalMs = app.sequencing.submitIntervalMs; - const batchProgram = Effect.sleep(Duration.millis(batchIntervalMs)).pipe( - Effect.andThen( - Effect.repeat( - acceptBatch, - Schedule.fixed(Duration.millis(batchIntervalMs)), - ), - ), - ); + // FIFO wakes as soon as a mutation arrives and accepts everything that + // queued while the previous batch ran. Batch mode collects for a fixed + // window so it can reorder by `batchOrder`. Both take mutations outside + // the state lock, so a wait never blocks submit. + const batchProgram = + app.sequencing.order === "batch" + ? Effect.sleep(Duration.millis(app.sequencing.batchIntervalMs)).pipe( + Effect.andThen( + Effect.repeat( + Queue.clear(receivedMutationQueue).pipe( + Effect.flatMap(acceptBatch), + ), + Schedule.fixed(Duration.millis(app.sequencing.batchIntervalMs)), + ), + ), + ) + : Effect.forever( + Queue.takeAll(receivedMutationQueue).pipe( + Effect.flatMap(acceptBatch), + ), + ); const submitProgram = Effect.sleep(Duration.millis(submitIntervalMs)).pipe( Effect.andThen( Effect.repeat( From aca1f7942d4323c7f7623afdc94cc6d54cd8b371 Mon Sep 17 00:00:00 2001 From: kyscott18 <43524469+kyscott18@users.noreply.github.com> Date: Tue, 13 Oct 2026 18:33:49 +0000 Subject: [PATCH 2/2] perf(runtime): commit accepted batches while the next batch executes Persisting a batch (~1.2ms) ran inside the speculative-state lock, so execution waited on Postgres. A single persist fiber now commits accepted batches in order, outside the lock. It groups all batches that queued during the previous commit into one transaction. Callers still get a response only after their batch commits, and a batch reaches submit only after it commits. Simulate is correct only for the newest journals, so submit waits under the lock for the in-flight commit before it reads the accepted batches. A failed commit now fails the waiting callers instead of leaving them pending. Counter benchmark, open loop at 4,000/s (2 runs each): 3,180-3,250 -> 3,577-3,580 mutations/s; p50 516-571ms -> 272-285ms. Closed-loop results do not change: those clients send the next mutation only after the commit. --- packages/typewriter/src/runtime.ts | 97 +++++++++++++++++++++++++----- 1 file changed, 81 insertions(+), 16 deletions(-) diff --git a/packages/typewriter/src/runtime.ts b/packages/typewriter/src/runtime.ts index bc3e614..e3841ee 100644 --- a/packages/typewriter/src/runtime.ts +++ b/packages/typewriter/src/runtime.ts @@ -640,6 +640,27 @@ export function createRuntimeEffect( deferred: Deferred.Deferred; }; + type PersistJob = { + batch: RuntimeBatch; + acceptedMutations: { + mutation: AcceptedMutation; + executeResult: ExecuteResult; + preimages: readonly Hex.Hex[]; + deferred: Deferred.Deferred; + }[]; + committed: Deferred.Deferred; + }; + + // Accepted batches are persisted by one fiber, in order, outside the + // speculative-state lock, so batch N+1 executes while batch N commits. + // Callers get a response only after their batch commits. A batch reaches + // `submit` only after it commits. + const persistQueue = yield* Queue.unbounded(); + yield* Scope.addFinalizer(scope, Queue.shutdown(persistQueue)); + // Resolves when the most recently accepted batch has committed. Because + // commits are ordered, all earlier batches have committed too. + let lastBatchCommitted: Deferred.Deferred | undefined; + const acceptBatch = (submittedMutations: ReceivedMutationEntry[]) => withSpeculativeStateLock( Effect.gen(function* () { @@ -704,8 +725,25 @@ export function createRuntimeEffect( emitBatch(batchToEvent(batch)); - const persistStartedAtMs = startTimer(); - yield* db.transaction((tx) => + const committed = yield* Deferred.make(); + lastBatchCommitted = committed; + yield* Queue.offer(persistQueue, { + batch, + acceptedMutations, + committed, + }); + }), + ); + + // Commits every batch that queued during the previous commit in one + // transaction. + const persistBatches = (jobs: PersistJob[]) => + Effect.gen(function* () { + const persistStartedAtMs = startTimer(); + const acceptedMutations = jobs.flatMap((job) => job.acceptedMutations); + + yield* db + .transaction((tx) => Effect.gen(function* () { yield* insertMutations( tx, @@ -719,26 +757,48 @@ export function createRuntimeEffect( acceptedMutations.flatMap(({ preimages }) => preimages), ); }), + ) + .pipe( + Effect.tapError((error) => + Effect.gen(function* () { + for (const { deferred } of acceptedMutations) { + yield* Deferred.fail(deferred, error); + } + for (const { committed } of jobs) { + yield* Deferred.fail(committed, error); + } + }), + ), ); - for (const { mutation, deferred } of acceptedMutations) { - yield* Deferred.succeed(deferred, mutation); - } + for (const { mutation, deferred } of acceptedMutations) { + yield* Deferred.succeed(deferred, mutation); + } - yield* Effect.logDebug("persisted accepted mutations").pipe( - Effect.annotateLogs({ - mutationCount: acceptedMutations.length, - duration: durationMs(persistStartedAtMs), - }), - ); + yield* Effect.logDebug("persisted accepted mutations").pipe( + Effect.annotateLogs({ + batchCount: jobs.length, + mutationCount: acceptedMutations.length, + duration: durationMs(persistStartedAtMs), + }), + ); + for (const { batch, committed } of jobs) { yield* Queue.offer(acceptedBatchesQueue, batch.id); - }), - ); + yield* Deferred.succeed(committed, undefined); + } + }); const submit = Effect.gen(function* () { const simulation = yield* withSpeculativeStateLock( Effect.gen(function* () { + // Simulate rewinds and replays the journals it is given, which is + // only correct when they are the newest journals. So every executed + // batch must be in `acceptedBatchesQueue`: wait for the in-flight + // commit. The lock keeps new batches from executing meanwhile. + if (lastBatchCommitted !== undefined) { + yield* Deferred.await(lastBatchCommitted); + } const batchIds = yield* Queue.clear(acceptedBatchesQueue); const enqueuedMutationIds = yield* Queue.clear( enqueuedMutationsQueue, @@ -1124,9 +1184,14 @@ export function createRuntimeEffect( ), ); - const program = Effect.all([batchProgram, submitProgram, watchProgram], { - concurrency: "unbounded", - }); + const persistProgram = Effect.forever( + Queue.takeAll(persistQueue).pipe(Effect.flatMap(persistBatches)), + ); + + const program = Effect.all( + [batchProgram, persistProgram, submitProgram, watchProgram], + { concurrency: "unbounded" }, + ); function execute( mutation: RuntimeExecuteInput,