diff --git a/packages/typewriter/simulation-test/chain.ts b/packages/typewriter/simulation-test/chain.ts index ce49e36..b60f167 100644 --- a/packages/typewriter/simulation-test/chain.ts +++ b/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); diff --git a/packages/typewriter/src/db-migrate.ts b/packages/typewriter/src/db-migrate.ts index 63ca709..49bef48 100644 --- a/packages/typewriter/src/db-migrate.ts +++ b/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. diff --git a/packages/typewriter/src/rpc-request.ts b/packages/typewriter/src/rpc-request.ts index 19c00bf..2df976c 100644 --- a/packages/typewriter/src/rpc-request.ts +++ b/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 = Block< bigint, includeTransactions, @@ -41,16 +41,19 @@ export function requestBlock( 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( export function requestExecutionIndex( address: Address.Address, + blockHash: Hash, ): Effect.Effect { return Effect.gen(function* () { const rpc = yield* Rpc; @@ -79,7 +83,7 @@ export function requestExecutionIndex( functionName: "executionIndex", }), }, - "latest", + { blockHash }, ], }); diff --git a/packages/typewriter/src/typewriter.ts b/packages/typewriter/src/typewriter.ts index 0342a3d..8aed1b3 100644 --- a/packages/typewriter/src/typewriter.ts +++ b/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", +) { + 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( diff --git a/packages/typewriter/src/watch.ts b/packages/typewriter/src/watch.ts index 45ea5bd..c154562 100644 --- a/packages/typewriter/src/watch.ts +++ b/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 { +): 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 =>