From bbec4cb413da4b0f78f84a0d823f3e97d794c6fc Mon Sep 17 00:00:00 2001 From: kyscott18 <43524469+kyscott18@users.noreply.github.com> Date: Mon, 26 Oct 2026 19:37:25 +0000 Subject: [PATCH] perf(runtime): commit accepted mutations outside the speculative-state lock Batches and force-included mutations committed to Postgres while holding the speculative-state lock, so submit, watch, and the next batch waited on the commit. Both now commit after the lock is released. Commits still run one at a time in execution order: each waits for the previous one. Restart resumes after the highest committed execution index, so a later commit must not land before an earlier one. A batch enters the submit queue when it executes, so simulate always sees the newest journals. Submit waits for the commits it covers after simulating, outside the lock, and before it broadcasts. Co-Authored-By: Claude Opus 5.5 --- packages/typewriter/src/typewriter.ts | 145 +++++++++++++++++--------- 1 file changed, 93 insertions(+), 52 deletions(-) diff --git a/packages/typewriter/src/typewriter.ts b/packages/typewriter/src/typewriter.ts index 21ef693..92dcada 100644 --- a/packages/typewriter/src/typewriter.ts +++ b/packages/typewriter/src/typewriter.ts @@ -797,6 +797,53 @@ export function createTypewriterEffect< }); } + const insertAcceptedMutations = ( + acceptedMutations: readonly { + mutation: AcceptedMutation; + executeResult: ExecuteResult; + preimages: readonly Hex.Hex[]; + }[], + ) => + db.transaction((tx) => + Effect.gen(function* () { + yield* insertMutations( + tx, + app.mutationSchema, + acceptedMutations.map((entry) => entry.mutation), + ); + yield* insertStorageSlots( + tx, + app.typewriterSchema, + acceptedMutations, + ); + yield* insertKeccakPreimages( + tx, + app.typewriterSchema, + acceptedMutations.flatMap(({ preimages }) => preimages), + ); + }), + ); + + // Accepted mutations commit outside the speculative-state lock, so + // execution never waits on Postgres. Commits still run one at a time in + // execution order: restart resumes after the highest committed execution + // index, so a later commit must not land before an earlier one. + let lastCommit: Deferred.Deferred | undefined; + + // Call under the speculative-state lock, so commits are ordered like + // executions. Returns the commit, to run after the lock is released. + const sequenceCommit = (commit: Effect.Effect) => + Effect.gen(function* () { + const previous = lastCommit; + const done = yield* Deferred.make(); + lastCommit = done; + return Effect.gen(function* () { + if (previous !== undefined) yield* Deferred.await(previous); + yield* commit; + yield* Deferred.succeed(done, undefined); + }); + }); + const acceptBatch = ( submittedMutations: { mutation: ReceivedMutation; @@ -858,40 +905,36 @@ export function createTypewriterEffect< emitBatch(batchToEvent(batch)); - const persistStartedAtMs = startTimer(); - yield* db.transaction((tx) => - Effect.gen(function* () { - yield* insertMutations( - tx, - app.mutationSchema, - acceptedMutations.map((entry) => entry.mutation), - ); - yield* insertStorageSlots( - tx, - app.typewriterSchema, - acceptedMutations, - ); - yield* insertKeccakPreimages( - tx, - app.typewriterSchema, - acceptedMutations.flatMap(({ preimages }) => preimages), - ); - }), - ); + yield* Queue.offer(acceptedBatchesQueue, batch.id); - for (const { mutation, deferred } of acceptedMutations) { - yield* Deferred.succeed(deferred, mutation); - } + return { + acceptedMutations, + commit: yield* sequenceCommit( + insertAcceptedMutations(acceptedMutations), + ), + }; + }), + ).pipe( + Effect.flatMap((accepted) => + Effect.gen(function* () { + if (accepted === undefined) return; + const { acceptedMutations, commit } = accepted; - yield* Effect.logDebug("persisted accepted mutations").pipe( - Effect.annotateLogs({ - mutationCount: acceptedMutations.length, - duration: durationMs(persistStartedAtMs), - }), - ); + const persistStartedAtMs = startTimer(); + yield* commit; - 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), + }), + ); + }), + ), ); const submit = Effect.gen(function* () { @@ -934,27 +977,15 @@ export function createTypewriterEffect< ({ mutation }) => mutation, ); - if (acceptedEnqueuedMutations.length > 0) { - yield* db.transaction((tx) => - Effect.gen(function* () { - yield* insertMutations( - tx, - app.mutationSchema, - acceptedEnqueuedMutations.map((entry) => entry.mutation), - ); - yield* insertStorageSlots( - tx, - app.typewriterSchema, - acceptedEnqueuedMutations, - ); - yield* insertKeccakPreimages( - tx, - app.typewriterSchema, - acceptedEnqueuedMutations.flatMap((entry) => entry.preimages), - ); - }), - ); - } + const commitEnqueuedMutations = + acceptedEnqueuedMutations.length > 0 + ? yield* sequenceCommit( + insertAcceptedMutations(acceptedEnqueuedMutations), + ) + : undefined; + // Commits are ordered, so this covers every mutation in the + // submission. + const committed = lastCommit; const speculativeJournalIds = batches .flatMap((batch) => batch.mutations) @@ -1003,6 +1034,8 @@ export function createTypewriterEffect< acceptedForceIncludedMutations, calldata, simulateResult, + commitEnqueuedMutations, + committed, }; }), ); @@ -1014,8 +1047,16 @@ export function createTypewriterEffect< acceptedForceIncludedMutations, calldata, simulateResult, + commitEnqueuedMutations, + committed, } = simulation; + // Settle only committed mutations. + if (commitEnqueuedMutations !== undefined) { + yield* commitEnqueuedMutations; + } + if (committed !== undefined) yield* Deferred.await(committed); + const transactionStartedAtMs = startTimer(); const filled = yield* rpc.request({