Skip to content

perf(runtime): commit accepted batches while the next batch executes #61

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

Open
kyscott18 wants to merge 2 commits into
base: main
Choose a base branch
from
Open

perf(runtime): commit accepted batches while the next batch executes #61

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
b459a59
perf(runtime): wake the FIFO batch loop when a mutation arrives
kyscott18 Oct 13, 2026
aca1f79
perf(runtime): commit accepted batches while the next batch executes
kyscott18 Oct 13, 2026
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
252 changes: 166 additions & 86 deletions packages/typewriter/src/runtime.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 = {
Expand Down Expand Up @@ -636,103 +635,170 @@ 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<AcceptedMutation, unknown>;
};

if (submittedMutations.length === 0) {
return;
}
type PersistJob = {
batch: RuntimeBatch;
acceptedMutations: {
mutation: AcceptedMutation;
executeResult: ExecuteResult;
preimages: readonly Hex.Hex[];
deferred: Deferred.Deferred<AcceptedMutation, unknown>;
}[];
committed: Deferred.Deferred<void, unknown>;
};

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)
);
});
}
// 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<PersistJob>();
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<void, unknown> | undefined;

const acceptBatch = (submittedMutations: ReceivedMutationEntry[]) =>
withSpeculativeStateLock(
Effect.gen(function* () {
const batchStartedAtMs = startTimer();

const acceptedMutations: {
mutation: AcceptedMutation;
executeResult: ExecuteResult;
preimages: readonly Hex.Hex[];
deferred: Deferred.Deferred<AcceptedMutation, unknown>;
}[] = [];

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 (submittedMutations.length === 0) {
return;
}
}

if (acceptedMutations.length === 0) {
return;
}
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)
);
});
}

const batch: RuntimeBatch = {
status: "accepted",
id: batchId++,
position: batchPosition++,
mutations: acceptedMutations.map(({ mutation }) => mutation),
};
const acceptedMutations: {
mutation: AcceptedMutation;
executeResult: ExecuteResult;
preimages: readonly Hex.Hex[];
deferred: Deferred.Deferred<AcceptedMutation, unknown>;
}[] = [];

yield* Effect.logDebug("batched mutations").pipe(
Effect.annotateLogs({
id: batch.id,
mutationCount: batch.mutations.length,
duration: durationMs(batchStartedAtMs),
}),
);
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);
}
}

batchesById.set(batch.id, batch);
if (acceptedMutations.length === 0) {
return;
}

emitBatch(batchToEvent(batch));
const batch: RuntimeBatch = {
status: "accepted",
id: batchId++,
position: batchPosition++,
mutations: acceptedMutations.map(({ mutation }) => mutation),
};

yield* Effect.logDebug("batched mutations").pipe(
Effect.annotateLogs({
id: batch.id,
mutationCount: batch.mutations.length,
duration: durationMs(batchStartedAtMs),
}),
);

batchesById.set(batch.id, batch);

emitBatch(batchToEvent(batch));

const committed = yield* Deferred.make<void, unknown>();
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();
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),
);
}),
);
const acceptedMutations = jobs.flatMap((job) => job.acceptedMutations);

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),
);
}),
)
.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);
}

yield* Effect.logDebug("persisted accepted mutations").pipe(
Effect.annotateLogs({
batchCount: jobs.length,
mutationCount: acceptedMutations.length,
duration: durationMs(persistStartedAtMs),
}),
);

yield* Queue.offer(acceptedBatchesQueue, batch.id);
}),
);
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);
}
Comment on lines +799 to +801

@greptile-apps greptile-apps Bot Oct 13, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Commit ordering lacks a test

The new wait in submit has no focused regression test. Existing tests do not hold a database commit open while another batch executes, so they would not catch a broken wait that makes settlement use the wrong local state.

Add a test that delays batch N's commit, lets N+1 execute, and checks that submit waits for both commits before simulating them. Also check that a failed commit rejects waiting public callers.

Prompt To Fix With AI 
This is a comment left during a code review.
Path: packages/typewriter/src/runtime.ts
Line: 799-801

Comment:
**Commit ordering lacks a test**

The new wait in `submit` has no focused regression test. Existing tests do not hold a database commit open while another batch executes, so they would not catch a broken wait that makes settlement use the wrong local state.

Add a test that delays batch N's commit, lets N+1 execute, and checks that `submit` waits for both commits before simulating them. Also check that a failed commit rejects waiting public callers.

---

For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.

Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!

Fix in Claude Code

const batchIds = yield* Queue.clear(acceptedBatchesQueue);
const enqueuedMutationIds = yield* Queue.clear(
enqueuedMutationsQueue,
Expand Down Expand Up @@ -1086,20 +1152,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(
Expand All @@ -1109,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,
Expand Down
Loading