Skip to content

perf(runtime): wake the FIFO batch loop when a mutation arrives #58

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
kyscott18 merged 2 commits into from
Oct 13, 2026
Merged

perf(runtime): wake the FIFO batch loop when a mutation arrives #58

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
5bbc65a
refactor(runtime): inline the accept batch entry type
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
197 changes: 104 additions & 93 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,99 +635,106 @@ export function createRuntimeEffect(
});
}

const acceptBatch = withSpeculativeStateLock(
Effect.gen(function* () {
const batchStartedAtMs = startTimer();
const submittedMutations = yield* Queue.clear(receivedMutationQueue);
const acceptBatch = (
submittedMutations: {
mutationId: number;
deferred: Deferred.Deferred<AcceptedMutation, unknown>;
}[],
) =>
withSpeculativeStateLock(
Effect.gen(function* () {
const batchStartedAtMs = startTimer();

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

const batch: RuntimeBatch = {
status: "accepted",
id: batchId++,
position: batchPosition++,
mutations: acceptedMutations.map(({ mutation }) => mutation),
};
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),
}),
);
yield* Effect.logDebug("batched mutations").pipe(
Effect.annotateLogs({
id: batch.id,
mutationCount: batch.mutations.length,
duration: durationMs(batchStartedAtMs),
}),
);

batchesById.set(batch.id, batch);
batchesById.set(batch.id, batch);

emitBatch(batchToEvent(batch));
emitBatch(batchToEvent(batch));

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

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({
mutationCount: acceptedMutations.length,
duration: durationMs(persistStartedAtMs),
}),
);

yield* Queue.offer(acceptedBatchesQueue, batch.id);
}),
);
yield* Queue.offer(acceptedBatchesQueue, batch.id);
}),
);

const submit = Effect.gen(function* () {
const simulation = yield* withSpeculativeStateLock(
Expand Down Expand Up @@ -1086,20 +1092,25 @@ 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)),
),
),
);
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),
),
Comment on lines +1109 to +1112

@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 FIFO changes escape tests

Both helpers in runtime.test.ts select batch mode, so the existing suite never exercises the new Queue.takeAll branch. The suite can pass even if FIFO acceptance or submission breaks. Add a FIFO test that waits while idle, accepts consecutive mutations in arrival order, and confirms that submit can still run while the queue is empty.

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

Comment:
**FIFO changes escape tests**

Both helpers in `runtime.test.ts` select batch mode, so the existing suite never exercises the new `Queue.takeAll` branch. The suite can pass even if FIFO acceptance or submission breaks. Add a FIFO test that waits while idle, accepts consecutive mutations in arrival order, and confirms that `submit` can still run while the queue is empty.

---

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 submitProgram = Effect.sleep(Duration.millis(submitIntervalMs)).pipe(
Effect.andThen(
Effect.repeat(
Expand Down
Loading