Split pool-indexer bootstrap from serve via --bootstrap-only flag

Author: AryanGodara

crates/e2e/tests/e2e/pool_indexer.rs

@@ -225,10 +225,8 @@ async fn seed_checkpoint(db: &PgPool, factory: Address, block: u64) {
     .unwrap();
 }
 
-/// Spawns the pool-indexer task and waits for its `/health` endpoint to come
-/// up.
-async fn spawn_pool_indexer(factory: Address, metrics_port: u16) -> tokio::task::JoinHandle<()> {
-    let config = Configuration {
+fn pool_indexer_config(factory: Address, metrics_port: u16) -> Configuration {
+    Configuration {
         database: DatabaseConfig {
             url: LOCAL_DB_URL.parse().unwrap(),
             max_connections: NonZeroU32::new(5).unwrap(),
@@ -252,7 +250,13 @@ async fn spawn_pool_indexer(factory: Address, metrics_port: u16) -> tokio::task:
         metrics: MetricsConfig {
             bind_address: SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::LOCALHOST, metrics_port)),
         },
-    };
+    }
+}
+
+/// Spawns the pool-indexer task and waits for its `/health` endpoint to come
+/// up.
+async fn spawn_pool_indexer(factory: Address, metrics_port: u16) -> tokio::task::JoinHandle<()> {
+    let config = pool_indexer_config(factory, metrics_port);
     let handle = tokio::task::spawn(pool_indexer::run(config));
     wait_for_condition(TIMEOUT, || async {
         reqwest::get(format!("{POOL_INDEXER_HOST}/health"))
@@ -682,3 +686,45 @@ async fn pagination(web3: Web3) {
     })
     .await;
 }
+
+#[tokio::test]
+#[ignore]
+async fn local_node_pool_indexer_bootstrap_idempotent() {
+    run_test(bootstrap_idempotent).await;
+}
+
+/// `--bootstrap-only` on an already-seeded DB must be a fast no-op: detect the
+/// existing checkpoint, skip the (here unreachable) subgraph seeder, and return
+/// without binding any ports — mirroring a re-run of the bootstrap
+/// initContainer on a pod restart.
+async fn bootstrap_idempotent(web3: Web3) {
+    let db = PgPool::connect(LOCAL_DB_URL).await.unwrap();
+    clear_pool_indexer_tables(&db).await;
+
+    // A pre-seeded checkpoint marks the DB as already bootstrapped. No on-chain
+    // factory is needed: bootstrap reads the checkpoint and returns before any
+    // seeding or catch-up. The RPC is only used for the chain_id sanity check.
+    let factory = Address::repeat_byte(0x11);
+    let head = web3.provider.get_block_number().await.unwrap();
+    seed_checkpoint(&db, factory, head).await;
+
+    tokio::time::timeout(
+        TIMEOUT,
+        pool_indexer::bootstrap(pool_indexer_config(factory, POOL_INDEXER_METRICS_PORT)),
+    )
+    .await
+    .expect("bootstrap-only did not exit on an already-seeded DB");
+
+    let checkpoint: i64 = sqlx::query_scalar(
+        "SELECT block_number FROM pool_indexer_checkpoints WHERE contract_address = $1",
+    )
+    .bind(factory.as_slice())
+    .fetch_one(&db)
+    .await
+    .unwrap();
+    assert_eq!(
+        checkpoint,
+        head.cast_signed(),
+        "bootstrap mutated an already-seeded checkpoint"
+    );
+}

crates/pool-indexer/README.md

@@ -9,6 +9,38 @@ at a fixed block, catches up to the chain tip via RPC events, then stays
 live by polling new blocks. Drivers consume it via the `pool-indexer-url`
 field in their Uniswap V3 liquidity config.
 
+## Bootstrap and serve
+
+Startup has two phases:
+
+- **Bootstrap** — initial subgraph seed plus catch-up to the finalized head.
+  One-time and slow (minutes on a large chain).
+- **Serve** — live block polling and the HTTP API. No long startup cost.
+
+`pool-indexer --config <toml>` runs both in one process: it bootstraps when the
+DB has no checkpoint, then serves. This is the single-container deployment.
+
+`pool-indexer --bootstrap-only --config <toml>` runs only the bootstrap phase
+and then exits 0, binding no HTTP ports. It is **idempotent**: on a DB that
+already has a checkpoint it skips the seed and catch-up entirely (never touching
+the subgraph) and returns immediately, so re-running it — for example a
+restarted bootstrap initContainer — is a fast, safe no-op.
+
+This lets K8s run bootstrap as an initContainer and apply a tight startupProbe
+to the serve container, which finds the checkpoint already present and flips
+`/startup` ready almost immediately:
+
+```yaml
+initContainers:
+  - name: init-db      # flyway, schema from the indexer's own location
+    command: ["flyway", "-locations=filesystem:/flyway/sql-pool-indexer", "migrate"]
+  - name: bootstrap    # one-time seed + catch-up; idempotent on restart
+    command: ["pool-indexer", "--bootstrap-only", "--config", "/etc/config/pool-indexer.toml"]
+containers:
+  - name: pool-indexer # serve; DB already seeded, so startup is fast
+    command: ["pool-indexer", "--config", "/etc/config/pool-indexer.toml"]
+```
+
 ## Running locally
 
 Create `crates/pool-indexer/config.local.toml` first (schema = the
@@ -17,12 +49,19 @@ Create `crates/pool-indexer/config.local.toml` first (schema = the
 sections. String fields accept `%ENV_VAR` so secrets can come from the
 environment instead of being written into the file.
 
-Then, from the repository root, reset the local stack and start the indexer:
+The indexer uses its own database, migrated from `database/sql-pool-indexer`
+(separate from the shared autopilot/orderbook set in `database/sql`). From the
+repository root:
 
 ```bash
-# wipes the local DB — dev machines only
-docker compose down --volumes
 docker compose up -d db
-docker compose run --rm migrations
+# apply the indexer schema (database/sql-pool-indexer) to the indexer's DB
+psql "$POOL_INDEXER_DB_URL" -f database/sql-pool-indexer/V110__pool_indexer_uniswap_v3.sql
+
+# one process for both phases:
+cargo run --release -p pool-indexer -- --config crates/pool-indexer/config.local.toml
+
+# or split bootstrap from serve, as the K8s deployment does:
+cargo run --release -p pool-indexer -- --bootstrap-only --config crates/pool-indexer/config.local.toml
 cargo run --release -p pool-indexer -- --config crates/pool-indexer/config.local.toml
 ```

crates/pool-indexer/src/arguments.rs

@@ -7,6 +7,32 @@ pub struct Arguments {
     #[clap(long, env)]
     pub config: PathBuf,
 
+    /// Run only the bootstrap phase (initial seed + catch-up to the finalized
+    /// head), then exit; bind no HTTP ports. Idempotent: a fast no-op when the
+    /// DB is already seeded. Lets K8s run bootstrap as an initContainer and
+    /// apply tight startup probes to the serve container.
+    #[clap(long, env)]
+    pub bootstrap_only: bool,
+
     #[clap(flatten)]
     pub logging: LoggingArguments,
 }
+
+#[cfg(test)]
+mod tests {
+    use {super::*, clap::Parser};
+
+    #[test]
+    fn bootstrap_only_flag_parses() {
+        let serve = Arguments::parse_from(["pool-indexer", "--config", "/tmp/c.toml"]);
+        assert!(!serve.bootstrap_only);
+
+        let bootstrap = Arguments::parse_from([
+            "pool-indexer",
+            "--config",
+            "/tmp/c.toml",
+            "--bootstrap-only",
+        ]);
+        assert!(bootstrap.bootstrap_only);
+    }
+}

crates/pool-indexer/src/lib.rs

@@ -1,5 +1,5 @@
 pub mod config;
-pub use run::{run, start};
+pub use run::{bootstrap, run, start};
 
 mod api;
 mod arguments;

crates/pool-indexer/src/run.rs

@@ -2,7 +2,7 @@ use {
     crate::{
         api::AppState,
         arguments::Arguments,
-        config::{Configuration, NetworkConfig},
+        config::{Configuration, FactoryConfig, NetworkConfig},
         indexer::uniswap_v3::UniswapV3Indexer,
     },
     alloy_provider::Provider,
@@ -21,8 +21,49 @@ pub async fn start(args: impl Iterator<Item = String>) {
     initialize_observability(&args);
     observe::metrics::setup_registry(None, None);
     let config = Configuration::from_path(&args.config).expect("failed to load configuration");
-    tracing::info!("pool-indexer starting");
-    run(config).await;
+    if args.bootstrap_only {
+        tracing::info!("pool-indexer bootstrap-only starting");
+        bootstrap(config).await;
+        tracing::info!("pool-indexer bootstrap complete, exiting");
+    } else {
+        tracing::info!("pool-indexer starting");
+        run(config).await;
+    }
+}
+
+/// Runs the bootstrap phase (seed + catch-up to the finalized head) for every
+/// factory, then returns. Binds no HTTP ports — this is migration-style work
+/// meant to run as a K8s initContainer ahead of the serve container.
+///
+/// Idempotent: each factory with an existing checkpoint is skipped (see
+/// [`bootstrap_factory`]), so re-running on an already-seeded DB is a fast
+/// no-op that never touches the subgraph. On return, a subsequent `run` finds
+/// the checkpoints present and flips `/startup` ready almost immediately.
+pub async fn bootstrap(config: Configuration) {
+    let db = connect_db(&config).await;
+    let network = config.network;
+    let provider = build_provider_checked(&network).await;
+    let network = Arc::new(network);
+
+    // One factory per network (enforced in NetworkConfig::validate), so this
+    // sequential loop is equivalent to the serve path's per-factory JoinSet.
+    for factory in network.factories.iter().copied() {
+        let indexer = factory_indexer(&provider, &db, &network, factory);
+        bootstrap_factory(&db, &indexer, &network, &factory).await;
+    }
+}
+
+fn factory_indexer(
+    provider: &AlloyProvider,
+    db: &PgPool,
+    network: &NetworkConfig,
+    factory: FactoryConfig,
+) -> UniswapV3Indexer {
+    UniswapV3Indexer::new(
+        provider.clone(),
+        db.clone(),
+        &network.indexer_config(factory.address),
+    )
 }
 
 pub async fn run(config: Configuration) {
@@ -125,31 +166,14 @@ async fn run_network_indexer(db: PgPool, network: NetworkConfig, barrier: Arc<St
         "starting network indexer",
     );
 
-    let provider = build_provider(&network);
-
-    // Catch misconfigured RPC-vs-network pairings (e.g. chain_id=1 pointed
-    // at an Arbitrum node) before we index the wrong chain into the DB.
-    let actual_chain_id = provider
-        .get_chain_id()
-        .await
-        .expect("failed to fetch chain_id from RPC");
-    assert_eq!(
-        actual_chain_id, network.chain_id,
-        "chain_id mismatch for network {}: config says {}, RPC reports {}",
-        network.name, network.chain_id, actual_chain_id,
-    );
-
+    let provider = build_provider_checked(&network).await;
     let network = Arc::new(network);
 
     // One task per factory. Provider + DB pool are shared; checkpoints are
     // per-factory because they're keyed by `contract_address`.
     let mut factory_set = JoinSet::new();
     for factory in network.factories.iter().copied() {
-        let indexer = UniswapV3Indexer::new(
-            provider.clone(),
-            db.clone(),
-            &network.indexer_config(factory.address),
-        );
+        let indexer = factory_indexer(&provider, &db, &network, factory);
         factory_set.spawn(run_factory_indexer(
             db.clone(),
             indexer,
@@ -255,6 +279,23 @@ fn build_provider(network: &NetworkConfig) -> AlloyProvider {
     .clone()
 }
 
+/// Builds the RPC provider and asserts the node's chain_id matches config.
+/// Catches misconfigured RPC-vs-network pairings (e.g. chain_id=1 pointed at
+/// an Arbitrum node) before we index the wrong chain into the DB.
+async fn build_provider_checked(network: &NetworkConfig) -> AlloyProvider {
+    let provider = build_provider(network);
+    let actual_chain_id = provider
+        .get_chain_id()
+        .await
+        .expect("failed to fetch chain_id from RPC");
+    assert_eq!(
+        actual_chain_id, network.chain_id,
+        "chain_id mismatch for network {}: config says {}, RPC reports {}",
+        network.name, network.chain_id, actual_chain_id,
+    );
+    provider
+}
+
 async fn connect_db(config: &Configuration) -> sqlx::PgPool {
     PgPoolOptions::new()
         .max_connections(config.database.max_connections.get())