fix(watch): emit force-inclusion logs from the runtime's first block

packages/typewriter/simulation-test/chain.ts

@@ -40,6 +40,11 @@ export function makeChain(): Effect.Effect<
         "--no-mining",
         "--timestamp",
         String(SIM.chain.genesisTimestamp),
+        // TODO(kyle) remove once typewriter caps a settlement's gas. It
+        // settles a whole backlog in one transaction, which can go over the
+        // 30M block gas limit and fail with "intrinsic gas too high".
+        "--gas-limit",
+        "1000000000",
       ],
       { stderr: "inherit" },
     ).pipe(Effect.orDie);

packages/typewriter/src/db-migrate.ts

@@ -121,6 +121,11 @@ CREATE TABLE IF NOT EXISTS ${sql.identifier(schemaName)}.typewriter_metadata (
         const max = yield* selectMaxExecutionIndex(tx, mutationSchema);
         const nextExecutionIndex = max === undefined ? 0n : max + 1n;
 
+        // TODO(kyle) check the length of the force-inclusion queue in storage
+        // against the queue onchain at the startup block, as well as the
+        // execution index. A missing or extra entry shifts every later
+        // enqueued mutation, so `execute` runs a different queued mutation
+        // onchain than the one revm accepted.
         if (nextExecutionIndex < executionIndex) {
           // TODO(kyle) catch up from calldata and `ForceInclusionQueued` logs
           // instead of failing.

packages/typewriter/src/rpc-request.ts

@@ -10,7 +10,7 @@ import {
 import { Rpc, type RpcError } from "./rpc";
 import { TYPEWRITER_ABI } from "./typewriter-constants";
 
-export type BlockIdentifier = number | bigint | Hash;
+export type BlockIdentifier = number | bigint | Hash | "latest";
 export type RequestedBlock<includeTransactions extends boolean = false> = Block<
   bigint,
   includeTransactions,
@@ -41,16 +41,19 @@ export function requestBlock<includeTransactions extends boolean = false>(
   return Effect.gen(function* () {
     const rpc = yield* Rpc;
 
-    const id = typeof block === "string" ? block : BigInt(block);
+    const id = typeof block === "number" ? BigInt(block) : block;
     const rpcBlock =
-      typeof id === "string"
+      typeof id === "string" && id !== "latest"
         ? yield* rpc.request({
             method: "eth_getBlockByHash",
             params: [id, includeTransactions],
           })
         : yield* rpc.request({
             method: "eth_getBlockByNumber",
-            params: [Hex.fromNumber(id), includeTransactions],
+            params: [
+              id === "latest" ? id : Hex.fromNumber(id),
+              includeTransactions,
+            ],
           });
 
     if (rpcBlock === null) {
@@ -66,6 +69,7 @@ export function requestBlock<includeTransactions extends boolean = false>(
 
 export function requestExecutionIndex(
   address: Address.Address,
+  blockHash: Hash,
 ): Effect.Effect<bigint, RpcError | NoDataReturnedError, Rpc> {
   return Effect.gen(function* () {
     const rpc = yield* Rpc;
@@ -79,7 +83,7 @@ export function requestExecutionIndex(
             functionName: "executionIndex",
           }),
         },
-        "latest",
+        { blockHash },
       ],
     });
 

packages/typewriter/src/typewriter.ts

@@ -60,6 +60,7 @@ import {
 import { durationMs, loggerLayer, startTimer } from "./logger";
 import { Rpc, type RpcError } from "./rpc";
 import {
+  type BlockNotFoundError,
   type NoDataReturnedError,
   requestBlock,
   requestExecutionIndex,
@@ -86,7 +87,12 @@ import type {
   TypewriterMutationInput,
   TypewriterMutationResult,
 } from "./typewriter-types";
-import { type LocalLog, makeWatch, type WatchError } from "./watch";
+import {
+  type IncompleteBlockError,
+  type LocalLog,
+  makeWatch,
+  type WatchError,
+} from "./watch";
 
 export type StorageRootProperty<
   storageConfig extends StorageConfig,
@@ -177,6 +183,24 @@ export class SettlementSimulationHaltedError extends Data.TaggedError(
   }
 }
 
+/** A mutation that was enqueued onchain reverted when enqueued in revm. */
+export class EnqueueRevertedError extends Data.TaggedError(
+  "EnqueueRevertedError",
+)<Revert> {
+  override get message() {
+    return revertMessage("revm enqueue reverted", this.errorName);
+  }
+}
+
+/** A mutation that was enqueued onchain halted when enqueued in revm. */
+export class EnqueueHaltedError extends Data.TaggedError("EnqueueHaltedError")<{
+  reason: string;
+}> {
+  override get message() {
+    return `revm enqueue halted: ${this.reason}`;
+  }
+}
+
 /** A settlement transaction was mined but reverted onchain. */
 export class SettlementRevertedError extends Data.TaggedError(
   "SettlementRevertedError",
@@ -211,12 +235,16 @@ export type ProgramError =
   | SettlementSimulationRevertedError
   | SettlementSimulationHaltedError
   | SettlementRevertedError
+  | EnqueueRevertedError
+  | EnqueueHaltedError
   | SignTransactionError
   | ProgramStoppedError;
 
 export type CreateTypewriterError =
   | DbError
   | RpcError
+  | BlockNotFoundError
+  | IncompleteBlockError
   | NoDataReturnedError
   | MigrationError
   | EvmError;
@@ -640,11 +668,17 @@ export function createTypewriterEffect<
       DEPLOYMENT_LOCK_TIMEOUT_MS,
     );
 
+    // The runtime starts from this block, and watches the blocks after it.
+    // Read it by hash, so each read sees the same block on any provider.
+    const startBlock = yield* requestBlock("latest");
     const code = yield* rpc.request({
       method: "eth_getCode",
-      params: [app.address, "latest"],
+      params: [app.address, { blockHash: startBlock.hash }],
     });
-    let nextExecutionIndex = yield* requestExecutionIndex(app.address);
+    let nextExecutionIndex = yield* requestExecutionIndex(
+      app.address,
+      startBlock.hash,
+    );
     yield* migrate(
       db,
       app.schemaName,
@@ -662,6 +696,7 @@ export function createTypewriterEffect<
       yield* createRuntimeState(app, db, code);
 
     const watch = yield* makeWatch({
+      fromBlock: startBlock.hash,
       pollIntervalMs: app.blockPollingIntervalMs,
       maxChainDepth: app.confirmations.finalizedBlockDepth,
       logFilter: {
@@ -946,6 +981,12 @@ export function createTypewriterEffect<
               );
             } else {
               // TODO(kyle) how to handle failed enqueue.
+              // TODO(kyle) the discarded mutation stays in the queue onchain,
+              // and anyone can `forceExecute` it after FORCE_INCLUSION_DELAY.
+              // It can succeed then, for example once its nonce is next, and
+              // change state and `executionIndex` outside revm. The watch only
+              // sees `ForceInclusionQueued` logs and `forceExecute` emits no
+              // event, so revm would not see it.
             }
           }
 
@@ -1193,18 +1234,21 @@ export function createTypewriterEffect<
             });
 
             for (const log of message.block.logs) {
-              // TODO(kyle) check the log's queue index against the length of
-              // the queue in revm. A missed log shifts every later enqueued
-              // mutation, so `execute` runs a different queued mutation than
-              // the one revm accepted, and the settlement reverts onchain.
               const enqueuedMutation = decodeEnqueuedMutation({ app, log });
 
               const executeResult = yield* withSpeculativeStateLock(
                 enqueueMutation({ app, evm, mutation: enqueuedMutation }),
               );
 
-              if (executeResult.status !== "success") {
-                continue;
+              if (executeResult.status === "revert") {
+                return yield* new EnqueueRevertedError(
+                  decodeRevert(app.abi, executeResult.data),
+                );
+              }
+              if (executeResult.status === "halt") {
+                return yield* new EnqueueHaltedError({
+                  reason: executeResult.reason,
+                });
               }
 
               invalidateStorageCache(

packages/typewriter/src/watch.ts

@@ -50,6 +50,8 @@ export type WatchMessage =
     };
 
 export type WatchConfig = {
+  /** The last block the caller has seen. The watch emits each block after it. */
+  fromBlock: Hex.Hex;
   maxChainDepth: number;
   pollIntervalMs: number;
   logFilter?: WatchLogFilter;
@@ -105,7 +107,11 @@ const quantity = (value: bigint): Hex.Hex => `0x${value.toString(16)}`;
 
 export function makeWatch(
   config: WatchConfig,
-): Effect.Effect<Watch, never, Rpc | Scope.Scope> {
+): Effect.Effect<
+  Watch,
+  IncompleteBlockError | RpcError | BlockNotFoundError,
+  Rpc | Scope.Scope
+> {
   return Effect.gen(function* () {
     const rpc = yield* Rpc;
 
@@ -114,9 +120,6 @@ export function makeWatch(
       (q) => Queue.shutdown(q),
     );
 
-    // oldest first; tip is last. capped at maxChainDepth.
-    let localChain: readonly LocalBlock[] = [];
-
     const getLocalBlockWithLogs = (block: RpcBlock) =>
       Effect.gen(function* () {
         // TODO(kyle) write RPC response normalizers
@@ -213,6 +216,11 @@ export function makeWatch(
         return yield* getLocalBlockWithLogs(block);
       });
 
+    // oldest first; tip is last. capped at maxChainDepth.
+    let localChain: readonly LocalBlock[] = [
+      yield* getBlockByHash(config.fromBlock),
+    ];
+
     const reconcileReorg = (
       block: LocalBlock,
     ): Effect.Effect<void, WatchError> =>