Split an indexer across processes when the connection budget affords it

.claude/hooks/check-format.sh

@@ -0,0 +1,44 @@
+#!/usr/bin/env bash
+# Blocks `git push` while a file this branch touches is unformatted. Edits made
+# through Bash bypass the PostToolUse formatter, so without this the first
+# anyone hears of the formatting is a CI failure.
+set -uo pipefail
+
+cmd=$(jq -r '.tool_input.command // ""')
+case "$cmd" in
+*"git push"*) ;;
+*) exit 0 ;;
+esac
+
+cd "${CLAUDE_PROJECT_DIR:-.}" || exit 0
+
+base=$(git merge-base HEAD origin/main 2>/dev/null || true)
+changed=$(
+  {
+    git diff --name-only HEAD
+    [ -n "$base" ] && git diff --name-only "$base"...HEAD
+    git ls-files -o --exclude-standard
+  } | sort -u | while read -r f; do [ -f "$f" ] && echo "$f"; done
+)
+
+problems=""
+
+res=$(printf '%s\n' "$changed" | grep -E '\.resi?$' || true)
+if [ -n "$res" ]; then
+  out=$(printf '%s\n' "$res" | xargs pnpx rescript@12.2.0 format --check 2>&1 | grep '^\[format check\]' || true)
+  if [ -n "$out" ]; then
+    problems="$problems"$'\n'"Unformatted ReScript (fix: pnpx rescript@12.2.0 format <files>):"$'\n'"$out"
+  fi
+fi
+
+if printf '%s\n' "$changed" | grep -q '^packages/cli/'; then
+  if ! out=$(cd packages/cli && cargo fmt --check 2>&1); then
+    problems="$problems"$'\n'"Unformatted Rust (fix: cd packages/cli && cargo fmt):"$'\n'"$out"
+  fi
+fi
+
+if [ -n "$problems" ]; then
+  jq -n --arg r "Formatting check failed, so the push was not run.$problems" \
+    '{hookSpecificOutput:{hookEventName:"PreToolUse",permissionDecision:"deny",permissionDecisionReason:$r}}'
+fi
+exit 0

.claude/settings.json

@@ -34,6 +34,19 @@
           }
         ]
       }
+    ],
+    "PreToolUse": [
+      {
+        "matcher": "Bash",
+        "hooks": [
+          {
+            "type": "command",
+            "command": "$CLAUDE_PROJECT_DIR/.claude/hooks/check-format.sh",
+            "timeout": 120,
+            "statusMessage": "Checking formatting before push..."
+          }
+        ]
+      }
     ]
   }
 }

CLAUDE.md

@@ -1,4 +1,5 @@
 - Use `pnpm` over `npm`/`npx`.
+- Edit `.res`/`.resi` and Rust files with Write/Edit, never a Bash heredoc or `sed`. The formatter runs on what those tools write; a push carrying an unformatted file is refused.
 - Always use single assert to check the whole value instead of multiple asserts for every field.
 
 ## Comments

CONTRIBUTING.md

@@ -118,7 +118,6 @@ Entry point:
   - For `start`: primes the config JSON (`Config.prime`), sets `cwd` + env vars, then calls `Main.start(~migrate?)`.
   - For `migrate` / `drop-schema`: primes config and calls `Main.migrate` / `Main.dropSchema`.
 - `Main.start` (in `packages/envio`) is the indexer entry proper. Responsibilities:
-  - Parses CLI flags (`--tui-off`, etc.).
   - Loads runtime configuration (`Config.res`).
   - Starts an Express server that serves `/metrics`, `/health`, and the Development Console endpoints.
   - Initializes the Persistence layer (Postgres + Hasura) — a single `init()` call that also handles `~reset` + `upsertPersistedState` when `~migrate` is provided.

packages/cli/CommandLineHelp.md

@@ -378,7 +378,9 @@ Start the indexer. Runs codegen automatically before launching so the on-disk ty
 ###### **Options:**
 
 * `-r`, `--restart` — Clear your database and restart indexing from scratch
-* `--chain <CHAIN_ID>` — Index only this chain, leaving the others to their own `envio start --chain` processes. Repeat the flag for several chains. Requires a schema whose entities are all per-chain, created for every chain by `envio local db-migrate up` before any process starts. Assign each configured chain to exactly one process, and give each its own `ENVIO_INDEXER_PORT`. Each process builds the indexes for its own chains and reports them ready as they catch up, independently of the others
+* `--chain <CHAIN_ID>` — Index only this chain, leaving the others to their own `envio start --chain` processes. Repeat the flag for several chains.
+
+   Only needed to place chains yourself. Requires a per-chain schema, migrated for every chain before any process starts, and a separate `ENVIO_INDEXER_PORT` per process.
 
 
 

packages/cli/src/cli_args/clap_definitions.rs

@@ -145,11 +145,10 @@ pub struct StartArgs {
     pub restart: bool,
 
     ///Index only this chain, leaving the others to their own `envio start --chain` processes.
-    ///Repeat the flag for several chains. Requires a schema whose entities are all per-chain,
-    ///created for every chain by `envio local db-migrate up` before any process starts.
-    ///Assign each configured chain to exactly one process, and give each its own
-    ///`ENVIO_INDEXER_PORT`. Each process builds the indexes for its own chains and reports them
-    ///ready as they catch up, independently of the others.
+    ///Repeat the flag for several chains.
+    ///
+    ///Only needed to place chains yourself. Requires a per-chain schema, migrated for every
+    ///chain before any process starts, and a separate `ENVIO_INDEXER_PORT` per process.
     #[arg(long = "chain", value_name = "CHAIN_ID")]
     pub chains: Vec<u64>,
 }

packages/e2e-tests/src/dependency-tests/install.test.ts

@@ -183,7 +183,7 @@ describe("Isolated dependency e2e", () => {
 
       await waitForOutput(
         indexerProcess,
-        "All chains are caught up to end blocks",
+        "Indexed to the end block",
         120_000
       );
 

packages/e2e-tests/src/e2e/e2e.test.ts

@@ -4,7 +4,7 @@
  * Tests the full indexer flow with database and ClickHouse sink:
  * 1. Ensure ClickHouse is running (CI service or local container)
  * 2. Start `envio dev` in background with ClickHouse sink enabled
- * 3. Wait for "All chains are caught up to end blocks" in stdout
+ * 3. Wait for "Indexed to the end block" in stdout
  * 4. Verify GraphQL queries return expected data
  * 5. Verify ClickHouse sink received the indexed data
  */
@@ -96,7 +96,7 @@ describe.skipIf(!dockerAvailable)("E2E: Indexer with GraphQL and ClickHouse sink
 
     await waitForOutput(
       indexerProcess,
-      "All chains are caught up to end blocks",
+      "Indexed to the end block",
       120_000
     );
 
@@ -842,7 +842,7 @@ describe.skipIf(!dockerAvailable)("E2E: Indexer with GraphQL and ClickHouse sink
       // waitForOutput rejects. Success means DB state was used.
       await waitForOutput(
         secondProcess,
-        "All chains are caught up to end blocks",
+        "Indexed to the end block",
         120_000
       );
     } finally {

packages/e2e-tests/src/e2e/split-run.test.ts

@@ -0,0 +1,161 @@
+/**
+ * A plain `envio start` over a per-chain schema, with a connection budget that
+ * affords two processes, splits the chains across forked workers and stays
+ * one indexer to the operator: one metrics endpoint over every process, one
+ * exit once every chain reaches its end block, one signal to stop it all.
+ *
+ * Needs Postgres but no Docker: it drives `envio start` with Hasura disabled.
+ */
+
+import { describe, it, expect, beforeAll, afterAll } from "vitest";
+import { ChildProcess } from "child_process";
+import path from "path";
+import { config } from "../config.js";
+import { runCommand, startBackground, waitForOutput } from "../utils/process.js";
+import { pgRows, closePg, isPgReachable } from "../utils/pg-direct.js";
+
+const PROJECT_DIR = path.join(config.scenariosDir, "split_test");
+const PG_SCHEMA = "e2e_split_run";
+const PORT = 9897;
+
+const indexerEnv = {
+  ENVIO_PG_SCHEMA: PG_SCHEMA,
+  // Two processes' worth: the split's own condition.
+  ENVIO_PG_MAX_CONNECTIONS: "4",
+  ENVIO_HASURA: "false",
+  ENVIO_TUI: "false",
+  ENVIO_INDEXER_PORT: String(PORT),
+  ENVIO_API_TOKEN: process.env.ENVIO_API_TOKEN ?? "",
+};
+
+const reachable = await isPgReachable();
+
+if (!reachable && process.env.CI) {
+  throw new Error(
+    "Postgres is unreachable, so the split-run suite cannot run. Refusing to skip it in CI."
+  );
+}
+
+const exitCode = (child: ChildProcess) =>
+  new Promise<number | null>((resolve) => child.on("close", resolve));
+
+/**
+ * Leaves no indexer behind when a test fails before its own shutdown: one that
+ * survived would hold the port and the schema against everything after it.
+ */
+const stopIfRunning = async (indexer: ChildProcess) => {
+  if (indexer.exitCode !== null || indexer.signalCode !== null) return;
+  const exited = exitCode(indexer);
+  indexer.kill("SIGINT");
+  const abandon = setTimeout(() => indexer.kill("SIGKILL"), 10_000);
+  await exited;
+  clearTimeout(abandon);
+};
+
+/** Polls an endpoint of the supervisor until its body satisfies `ready`. */
+const scrapeUntil = async (route: string, ready: (body: string) => boolean) => {
+  const deadline = Date.now() + config.timeouts.indexerStartup;
+  let body = "";
+  while (Date.now() < deadline) {
+    try {
+      body = await (await fetch(`http://localhost:${PORT}${route}`)).text();
+      if (ready(body)) return body;
+    } catch {}
+    await new Promise((r) => setTimeout(r, 500));
+  }
+  throw new Error(`Timed out scraping ${route}\n--- last body ---\n${body}`);
+};
+
+const start = (args: string[]) =>
+  startBackground(config.envioCommand, [...config.envioArgs, "start", ...args], {
+    cwd: PROJECT_DIR,
+    env: indexerEnv,
+  });
+
+describe.skipIf(!reachable)("E2E: a split run is one indexer", () => {
+  beforeAll(async () => {
+    const codegen = await runCommand(
+      config.envioCommand,
+      [...config.envioArgs, "codegen"],
+      { cwd: PROJECT_DIR, env: indexerEnv, timeout: config.timeouts.codegen }
+    );
+    expect(codegen.exitCode, `codegen failed: ${codegen.stderr}`).toBe(0);
+  }, config.timeouts.codegen);
+
+  afterAll(async () => {
+    await closePg();
+  });
+
+  it("Exits once every chain is done, with both chains' rows written", async () => {
+    const indexer = start(["-r"]);
+    try {
+      const exit = exitCode(indexer);
+      await waitForOutput(indexer, "Indexing will be split across multiple processes", config.timeouts.indexerStartup);
+
+      expect({
+        exitCode: await exit,
+        rowsPerChain: await pgRows(
+          `SELECT "chain_id", COUNT(*) > 0 FROM "${PG_SCHEMA}"."Transfer" GROUP BY "chain_id" ORDER BY "chain_id"`
+        ),
+        // Reaching an end block doesn't make a worker done: it owes the schema
+        // the indexes its chains deferred, and until the run goes realtime it
+        // has no leave to commit them.
+        readyPerChain: await pgRows(
+          `SELECT "id"::text, "ready_at" IS NOT NULL FROM "${PG_SCHEMA}"."envio_chains" ORDER BY "id"`
+        ),
+      }).toEqual({
+        exitCode: 0,
+        rowsPerChain: [
+          [1, true],
+          [8453, true],
+        ],
+        readyPerChain: [
+          ["1", true],
+          ["8453", true],
+        ],
+      });
+    } finally {
+      await stopIfRunning(indexer);
+    }
+  });
+
+  // The same chains with no end block: a run that nothing but a stop ends, so
+  // there is time to read what it serves.
+  it("Serves every process's metrics, and stops them all on one interrupt", async () => {
+    const indexer = start(["-r", "--config", "config.head.yaml"]);
+    try {
+      const exit = exitCode(indexer);
+      await waitForOutput(indexer, "Indexing will be split across multiple processes", config.timeouts.indexerStartup);
+
+      const [runtime, metrics] = await Promise.all([
+        // Each worker's readings, told apart by label.
+        scrapeUntil("/metrics/runtime", (body) => body.includes('worker="8453"')),
+        // Both chains on one endpoint, whichever process drives each.
+        scrapeUntil(
+          "/metrics",
+          (body) => body.includes('chainId="1"') && body.includes('chainId="8453"')
+        ),
+      ]);
+
+      // Only the supervisor is signalled, the way a process manager would.
+      indexer.kill("SIGINT");
+
+      expect({
+        exitCode: await exit,
+        // Workers are named by the chains they drive.
+        runtimeWorkers: ["1", "8453"].map((worker) =>
+          runtime.includes(`nodejs_heap_size_used_bytes{worker="${worker}"}`)
+        ),
+        metricsChains: [1, 8453].map((chainId) =>
+          metrics.includes(`envio_progress_block{chainId="${chainId}"}`)
+        ),
+      }).toEqual({
+        exitCode: 0,
+        runtimeWorkers: [true, true],
+        metricsChains: [true, true],
+      });
+    } finally {
+      await stopIfRunning(indexer);
+    }
+  });
+});

packages/envio-tests/test/BelowHeadPollingPin_test.res

@@ -1,6 +1,7 @@
 open Vitest
 
 let scenario = Scenario.make(
+  ~supervised=false,
   ~configYaml=`
 name: below-head-polling
 contracts:
@@ -84,9 +85,7 @@ describe("PIN: chains keep indexing after entering the reorg threshold", () => {
 
       await MockSource.waitItemsQuery(chainWithThresholdWork)
       t.expect(
-        chainWithThresholdWork.getItemsOrThrowCalls->Array.map(
-          call => call.payload["fromBlock"],
-        ),
+        chainWithThresholdWork.getItemsOrThrowCalls->Array.map(call => call.payload["fromBlock"]),
         ~message="the zero-lag chain first fetches to its pre-threshold head",
       ).toEqual([1])
       chainWithThresholdWork.resolveGetItemsOrThrow(
@@ -101,9 +100,7 @@ describe("PIN: chains keep indexing after entering the reorg threshold", () => {
       // progress and lets it lead, which sidesteps the production ordering.
       await MockSource.waitItemsQuery(chainWithThresholdWork)
       t.expect(
-        chainWithThresholdWork.getItemsOrThrowCalls->Array.map(
-          call => call.payload["fromBlock"],
-        ),
+        chainWithThresholdWork.getItemsOrThrowCalls->Array.map(call => call.payload["fromBlock"]),
         ~message="the second response reaches the zero-lag chain's pre-threshold head",
       ).toEqual([401])
       chainWithThresholdWork.resolveGetItemsOrThrow(
@@ -154,9 +151,7 @@ describe("PIN: chains keep indexing after entering the reorg threshold", () => {
       // lets chain 100 claim the progress-alignment line before discovering that
       // it is WaitingForNewBlock, which clamps chain 1337 behind block 800.
       t.expect(
-        chainWithThresholdWork.getItemsOrThrowCalls->Array.map(
-          call => call.payload["fromBlock"],
-        ),
+        chainWithThresholdWork.getItemsOrThrowCalls->Array.map(call => call.payload["fromBlock"]),
         ~message="the below-head chain is not blocked by an unchanged source",
       ).toEqual([801])
 

packages/envio-tests/test/E2E_test.res

@@ -29,6 +29,7 @@ let chainYaml = (chainId, ~startBlock=1) =>
 
 let makeScenario = (~name, ~rollback=true, ~chains) =>
   Scenario.make(
+    ~supervised=false,
     ~configYaml=`
 name: ${name}
 rollback_on_reorg: ${rollback ? "true" : "false"}${contractsYaml}chains:${chains}`,
@@ -45,6 +46,7 @@ let scenario = makeScenario(~name="e2e", ~chains=chainYaml(1337))
 // Partition ids and the chain's range-cost budget follow the contract set, so
 // this scenario keeps the address-less contracts alongside the addressed ones.
 let partitionScenario = Scenario.make(
+  ~supervised=false,
   ~configYaml=`
 name: e2e-partitions
 rollback_on_reorg: true${contractsYaml}  - name: SimpleNft

packages/envio-tests/test/EnterReorgThreshold_test.res

@@ -21,6 +21,7 @@ type Gravatar {
 // Two chains, each lagging maxReorgDepth (200) below head before the
 // threshold. Head starts at 1000, so the pre-threshold head is 800.
 let multichain = Scenario.make(
+  ~supervised=false,
   ~configYaml=`
 name: enter-reorg-threshold-multichain
 contracts:
@@ -53,6 +54,7 @@ chains:
 )
 
 let singleChain = Scenario.make(
+  ~supervised=false,
   ~configYaml=`
 name: enter-reorg-threshold-single
 chains:
@@ -119,11 +121,7 @@ describe("PIN: multichain indexer enters the reorg threshold", () => {
           logIndex: 0,
         },
       )
-      chainA.resolveGetItemsOrThrow(
-        densitySeed,
-        ~latestFetchedBlockNumber=800,
-        ~knownHeight=1000,
-      )
+      chainA.resolveGetItemsOrThrow(densitySeed, ~latestFetchedBlockNumber=800, ~knownHeight=1000)
       await indexer.getBatchWritePromise()
 
       // Chain A is now at its lagged head with an empty buffer — momentarily