Cold-seed the pool-indexer from on-chain events

crates/e2e/tests/e2e/pool_indexer.rs

@@ -1,7 +1,5 @@
-//! End-to-end check that the driver consumes pool data from `pool-indexer`
-//! when `pool-indexer-url` is set. Pre-seeds the indexer checkpoint so the
-//! subgraph_seeder bootstrap is skipped (Anvil has no subgraph); only the
-//! live-indexing and HTTP-serving paths are exercised.
+//! End-to-end tests for `pool-indexer`: on-chain cold-seed and catch-up, the
+//! HTTP API, and the driver consuming pool data from it.
 
 use {
     alloy::{
@@ -139,9 +137,6 @@ const POOL_INDEXER_DB_URL: &str = "postgresql:///pool_indexer";
 // sqrt(1) * 2^96 — valid starting price
 const INITIAL_SQRT_PRICE: u128 = 1u128 << 96;
 
-// Never queried — the pre-seeded checkpoint short-circuits the seeder.
-const PLACEHOLDER_SUBGRAPH_URL: &str = "http://127.0.0.1:1/never-queried";
-
 /// Typed shape of `GET /api/v1/{network}/uniswap/v3/pools`.
 #[derive(Deserialize)]
 struct PoolsListResponse {
@@ -200,12 +195,13 @@ fn pool_indexer_config(factory: Address, metrics_port: u16) -> Configuration {
             name: NetworkName::new("mainnet"),
             chain_id: 1,
             rpc_url: "http://127.0.0.1:8545".parse().unwrap(),
-            factories: vec![FactoryConfig { address: factory }],
+            factories: vec![FactoryConfig {
+                address: factory,
+                deploy_block: 0,
+            }],
             chunk_size: 1000,
             poll_interval_secs: 1,
             use_latest: true,
-            subgraph_url: PLACEHOLDER_SUBGRAPH_URL.parse().unwrap(),
-            seed_block: None,
             fetch_concurrency: 8,
             prefetch_concurrency: 50,
         },
@@ -658,17 +654,16 @@ 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.
+/// `--bootstrap-only` on an already-caught-up DB must be a fast no-op: the
+/// catch-up loop sees the checkpoint is already at the head and returns without
+/// binding any ports, like a bootstrap initContainer re-run on a pod restart.
 async fn bootstrap_idempotent(web3: Web3) {
     let db = PgPool::connect(POOL_INDEXER_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.
+    // A checkpoint at the head means the DB is already caught up, so the
+    // catch-up loop finds nothing to do. No on-chain factory is needed; the RPC
+    // only serves the chain_id check and reading the finalized head.
     let factory = Address::repeat_byte(0x11);
     let head = web3.provider.get_block_number().await.unwrap();
     seed_checkpoint(&db, factory, head).await;
@@ -693,3 +688,44 @@ async fn bootstrap_idempotent(web3: Web3) {
         "bootstrap mutated an already-seeded checkpoint"
     );
 }
+
+#[tokio::test]
+#[ignore]
+async fn local_node_pool_indexer_onchain_cold_seed() {
+    run_test(onchain_cold_seed).await;
+}
+
+/// Cold bootstrap with no checkpoint: the indexer discovers the pool and
+/// reconstructs its state by replaying on-chain events from the factory's
+/// deploy block.
+async fn onchain_cold_seed(web3: Web3) {
+    let db = PgPool::connect(POOL_INDEXER_DB_URL).await.unwrap();
+    clear_pool_indexer_tables(&db).await;
+
+    let (factory, pool_addr) = deploy_univ3(&web3).await;
+    let factory_addr = *factory.address();
+    let head = web3.provider.get_block_number().await.unwrap();
+
+    // No seed_checkpoint: bootstrap must cold-seed on-chain (deploy_block = 0).
+    with_pool_indexer_at(factory_addr, POOL_INDEXER_METRICS_PORT, || async {
+        wait_for_indexer(head, 1).await;
+
+        // Pool discovered + state rebuilt from the Initialize (sqrt_price) and
+        // Mint (liquidity) events alone; no pre-seeded checkpoint.
+        let (count, sqrt_price, _tick, liquidity) = snapshot_pool_state(&db, pool_addr).await;
+        assert_eq!(
+            count, 1,
+            "the pool should be discovered by the on-chain scan"
+        );
+        assert_eq!(
+            sqrt_price,
+            INITIAL_SQRT_PRICE.to_string(),
+            "sqrt_price reconstructed from the Initialize event"
+        );
+        assert_eq!(
+            liquidity, "1000000",
+            "liquidity reconstructed from the Mint event"
+        );
+    })
+    .await;
+}

crates/pool-indexer/README.md

@@ -1,34 +1,33 @@
 # pool-indexer
 
 Indexes Uniswap V3 pools, ticks, and pool state into Postgres and serves
-the data over HTTP. The driver consumes this API in place of a
-third-party Uniswap V3 subgraph.
+the data over HTTP. The driver consumes this API for its Uniswap V3
+liquidity.
 
-For each (chain, factory) pair, the indexer seeds its DB from a subgraph
-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.
+For each (chain, factory) pair, the indexer cold-seeds its DB by replaying the
+factory's on-chain events from its deploy block, catches up to the finalized
+head, 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).
+- **Bootstrap** — on-chain cold-seed from the factory's deploy block, catching
+  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 true --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 is a fast, safe
-no-op.
+phase and then exits 0, binding no HTTP ports. A DB already caught up to the
+head returns at once, while one left partially seeded by an interrupted run
+resumes from its checkpoint to completion.
 
 Running bootstrap as a separate step ahead of serving keeps serve startup fast:
-the serve process finds the checkpoint already present and flips `/startup` ready
-almost immediately.
+the serve process finds the DB already caught up and flips `/startup` ready
+promptly.
 
 ## Running locally
 

crates/pool-indexer/src/config.rs

@@ -102,11 +102,6 @@ pub struct NetworkConfig {
     /// against Anvil, which doesn't simulate finality.
     #[serde(skip)]
     pub use_latest: bool,
-    /// Subgraph GraphQL endpoint for the initial seed.
-    #[serde(deserialize_with = "configs::deserialize_env::deserialize_url_from_env")]
-    pub subgraph_url: Url,
-    /// Block to seed at. Defaults to the subgraph's current block.
-    pub seed_block: Option<u64>,
 }
 
 impl NetworkConfig {
@@ -138,12 +133,14 @@ impl NetworkConfig {
     }
 }
 
-/// One factory address. The indexer runs a dedicated seed + live-indexing
-/// loop per entry in [`NetworkConfig::factories`].
+/// The factory and the block it was deployed at. The indexer cold-seeds by
+/// replaying on-chain events from `deploy_block`, then live-indexes.
 #[derive(Debug, Clone, Copy, Deserialize)]
 #[serde(rename_all = "kebab-case", deny_unknown_fields)]
 pub struct FactoryConfig {
     pub address: Address,
+    /// Block the factory was deployed at; on-chain cold-seed scans from here
+    pub deploy_block: u64,
 }
 
 /// Subset of [`NetworkConfig`] handed to [`UniswapV3Indexer`] at runtime.

crates/pool-indexer/src/db/uniswap_v3.rs

@@ -335,72 +335,6 @@ pub async fn batch_update_ticks(
     Ok(())
 }
 
-/// Set `liquidity_net` directly (no delta accumulation). Used by the seeder
-/// where the subgraph value is already the net.
-pub async fn batch_seed_ticks(
-    tx: &mut Transaction<'_, Postgres>,
-    factory: &Address,
-    ticks: &[TickDeltaData],
-) -> Result<()> {
-    if ticks.is_empty() {
-        return Ok(());
-    }
-    let addresses: Vec<&[u8]> = ticks
-        .iter()
-        .map(|tick| tick.pool_address.as_slice())
-        .collect();
-    let tick_idxs: Vec<i32> = ticks.iter().map(|tick| tick.tick_idx).collect();
-    let values: Vec<BigDecimal> = ticks
-        .iter()
-        .map(|tick| BigDecimal::from(tick.delta))
-        .collect();
-
-    sqlx::query(
-        "WITH input AS (
-             SELECT t.addr, t.tick_idx, SUM(t.val) AS net
-             FROM UNNEST($2::BYTEA[], $3::INT4[], $4::NUMERIC[]) AS t(addr, tick_idx, val)
-             GROUP BY t.addr, t.tick_idx
-         )
-         INSERT INTO uniswap_v3_ticks (pool_address, tick_idx, liquidity_net)
-         SELECT i.addr, i.tick_idx, i.net
-         FROM input i
-         WHERE EXISTS (
-             SELECT 1 FROM uniswap_v3_pools
-             WHERE address = i.addr AND factory = $1
-         )
-           AND i.net <> 0
-         ON CONFLICT (pool_address, tick_idx) DO UPDATE
-             SET liquidity_net = EXCLUDED.liquidity_net",
-    )
-    .bind(factory.as_slice())
-    .bind(addresses)
-    .bind(tick_idxs)
-    .bind(values)
-    .execute(&mut **tx)
-    .await
-    .context("batch_seed_ticks")?;
-    Ok(())
-}
-
-/// Deletes ticks for all pools of `factory`. Used by the seeder before a
-/// reseed. Scoped to one factory so other factories aren't affected.
-pub async fn delete_ticks_for_factory(
-    tx: &mut Transaction<'_, Postgres>,
-    factory: &Address,
-) -> Result<()> {
-    sqlx::query(
-        "DELETE FROM uniswap_v3_ticks t
-         USING uniswap_v3_pools p
-         WHERE p.address = t.pool_address
-           AND p.factory = $1",
-    )
-    .bind(factory.as_slice())
-    .execute(&mut **tx)
-    .await
-    .context("delete_ticks_for_factory")?;
-    Ok(())
-}
-
 /// A pool joined with its current state row.
 pub struct PoolRow {
     pub address: Address,

crates/pool-indexer/src/indexer/uniswap_v3.rs

@@ -18,7 +18,10 @@ use {
     futures::{StreamExt, TryStreamExt},
     itertools::Itertools,
     sqlx::PgPool,
-    std::collections::{HashMap, HashSet},
+    std::{
+        collections::{HashMap, HashSet},
+        time::Duration,
+    },
     tracing::instrument,
 };
 
@@ -121,7 +124,7 @@ impl UniswapV3Indexer {
 
     /// Per-factory live-indexing loop. Backfill tasks live at the network
     /// level (see `run_network_indexer`).
-    pub async fn run(self, poll_interval: std::time::Duration) -> ! {
+    pub async fn run(self, poll_interval: Duration) -> ! {
         let mut interval = tokio::time::interval(poll_interval);
         interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
         loop {
@@ -136,17 +139,16 @@ impl UniswapV3Indexer {
         }
     }
 
-    /// Brings a freshly-seeded factory up to the current finalized block
-    /// and returns. Loops in case more blocks finalize during a long
-    /// catch-up. Call exactly once, right after seeding.
+    /// Cold-seeds a factory by replaying on-chain history from `from_block` up
+    /// to the current finalized block, then returns. Loops in case more blocks
+    /// finalize during a long catch-up. Call exactly once, before live
+    /// indexing.
     ///
-    /// The subgraph's `block: { number: X }` returns state as of the end
-    /// of block `X` — events at `X` are included. The checkpoint stores
-    /// the last indexed block and `run_once` resumes at `checkpoint + 1`,
-    /// so we set the checkpoint to `from_block` itself to avoid replaying
-    /// (and double-applying) the seed block's Mint/Burn events.
+    /// `run_once` resumes at `checkpoint + 1`, so the checkpoint is set to
+    /// `from_block`; callers pass `deploy_block - 1` to include the deploy
+    /// block's `PoolCreated` events.
     ///
-    /// Bails if a checkpoint already exists — overwriting would silently
+    /// Bails if a checkpoint already exists; overwriting would silently
     /// regress and re-index history.
     pub async fn catch_up(&self, from_block: u64) -> Result<()> {
         if db::get_checkpoint(&self.db, &self.factory).await?.is_some() {
@@ -160,6 +162,13 @@ impl UniswapV3Indexer {
         db::set_checkpoint(&mut tx, &self.factory, from_block).await?;
         tx.commit().await.context("commit checkpoint tx")?;
 
+        self.catch_up_to_finalized().await
+    }
+
+    /// Indexes until the checkpoint reaches the finalized head. A checkpoint
+    /// already at the head returns at once; one left behind (e.g. by an
+    /// interrupted cold-seed) is driven the rest of the way.
+    pub(crate) async fn catch_up_to_finalized(&self) -> Result<()> {
         loop {
             let finalized_block = self.finalized_block().await?;
             let last_indexed_block = self.last_indexed_block().await?;
@@ -444,7 +453,7 @@ pub(crate) async fn backfill_symbols(
     db: sqlx::PgPool,
     network: NetworkName,
     prefetch_concurrency: usize,
-    poll_interval: std::time::Duration,
+    poll_interval: Duration,
 ) {
     let mut interval = tokio::time::interval(poll_interval);
     interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
@@ -532,7 +541,7 @@ pub(crate) async fn backfill_decimals(
     db: sqlx::PgPool,
     network: NetworkName,
     prefetch_concurrency: usize,
-    poll_interval: std::time::Duration,
+    poll_interval: Duration,
 ) {
     let mut interval = tokio::time::interval(poll_interval);
     interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
@@ -670,6 +679,13 @@ pub(crate) fn is_range_too_large(err: &alloy_transport::TransportError) -> bool
 /// be the cause.
 const MAX_BISECTION_DEPTH: u32 = 8;
 
+/// Retry transient `eth_getLogs` failures (timeout, reset, throttle) with
+/// backoff, capped by a per-call timeout, so one blip can't abort a long
+/// cold-seed scan. Range-size rejections are bisected, not retried.
+const GETLOGS_TIMEOUT: Duration = Duration::from_secs(45);
+const MAX_GETLOGS_RETRIES: u32 = 6;
+const GETLOGS_RETRY_BACKOFF: Duration = Duration::from_secs(2);
+
 /// Fetches logs for `[from, to]` filtered by the given contract addresses
 /// and `topic0` event signatures, sequentially bisecting the block range on
 /// "too large" rejections until each sub-range is tractable. An empty
@@ -700,12 +716,30 @@ fn bisecting_get_logs_with_depth(
             .from_block(from)
             .to_block(to);
 
-        let err = match provider.get_logs(&filter).await {
-            Ok(logs) => return Ok(logs),
-            Err(err) => err,
-        };
-        if !is_range_too_large(&err) || to <= from || depth >= MAX_BISECTION_DEPTH {
-            return Err(anyhow::Error::new(err).context(format!("get_logs({from}..={to})")));
+        let mut attempt = 0u32;
+        loop {
+            let err = match tokio::time::timeout(GETLOGS_TIMEOUT, provider.get_logs(&filter)).await
+            {
+                Ok(Ok(logs)) => return Ok(logs),
+                // Range-size rejection: bisect (below), not retried.
+                Ok(Err(err)) if is_range_too_large(&err) => {
+                    if to <= from || depth >= MAX_BISECTION_DEPTH {
+                        return Err(
+                            anyhow::Error::new(err).context(format!("get_logs({from}..={to})"))
+                        );
+                    }
+                    break;
+                }
+                Ok(Err(err)) => anyhow::Error::new(err).context(format!("get_logs({from}..={to})")),
+                Err(_elapsed) => anyhow::anyhow!("get_logs({from}..={to}) timed out"),
+            };
+            // Transient failure: retry with backoff, then give up.
+            if attempt >= MAX_GETLOGS_RETRIES {
+                return Err(err);
+            }
+            attempt += 1;
+            tracing::warn!(%err, attempt, from, to, "get_logs failed, retrying");
+            tokio::time::sleep(GETLOGS_RETRY_BACKOFF * attempt).await;
         }
 
         let mid = (from + to) / 2;

crates/pool-indexer/src/lib.rs

@@ -7,4 +7,3 @@ mod db;
 mod indexer;
 mod metrics;
 mod run;
-mod subgraph_seeder;