From 3335f06159de502b3449236be65ce98aaf48afb0 Mon Sep 17 00:00:00 2001 From: Aryan Godara <65490434+AryanGodara@users.noreply.github.com> Date: Thu, 25 Jun 2026 17:46:24 +0530 Subject: [PATCH 1/5] Isolate pool-indexer flyway migrations from the shared DB set --- .devcontainer/setup-e2e.sh | 13 +++ .github/workflows/pull-request.yaml | 1 + crates/database/src/lib.rs | 9 +- crates/e2e/tests/e2e/pool_indexer.rs | 17 ++-- database/README.md | 78 +-------------- database/sql-pool-indexer/README.md | 99 +++++++++++++++++++ .../V110__pool_indexer_uniswap_v3.sql | 59 +++++++++++ .../V111__drop_pool_indexer_uniswap_v3.sql | 13 +++ docker-compose.yaml | 37 +++++++ 9 files changed, 236 insertions(+), 90 deletions(-) create mode 100644 database/sql-pool-indexer/README.md create mode 100644 database/sql-pool-indexer/V110__pool_indexer_uniswap_v3.sql create mode 100644 database/sql/V111__drop_pool_indexer_uniswap_v3.sql diff --git a/.devcontainer/setup-e2e.sh b/.devcontainer/setup-e2e.sh index 044740f317..f72f020b4d 100755 --- a/.devcontainer/setup-e2e.sh +++ b/.devcontainer/setup-e2e.sh @@ -66,4 +66,17 @@ docker run --rm --network=host \ -url="jdbc:postgresql://127.0.0.1:${PGPORT}/${PGDATABASE}?user=${PGUSER}&password=" \ migrate +# The pool-indexer has its own database migrated from its own directory (mirrors +# the per-network prod DB), separate from the autopilot/orderbook set above. +psql -h 127.0.0.1 -U postgres -d postgres -tAc \ + "SELECT 1 FROM pg_database WHERE datname='pool_indexer'" | grep -q 1 \ + || psql -h 127.0.0.1 -U postgres -d postgres -c \ + "CREATE DATABASE pool_indexer OWNER \"${PGUSER}\";" +docker run --rm --network=host \ + -v "$REPO_ROOT/database/sql-pool-indexer:/flyway/sql:ro" \ + -v "$REPO_ROOT/database/conf:/flyway/conf:ro" \ + flyway/flyway:10.7.1 \ + -url="jdbc:postgresql://127.0.0.1:${PGPORT}/pool_indexer?user=${PGUSER}&password=" \ + migrate + echo "==> e2e environment ready (postgres)" diff --git a/.github/workflows/pull-request.yaml b/.github/workflows/pull-request.yaml index f0dd558329..c2c2c4bd61 100644 --- a/.github/workflows/pull-request.yaml +++ b/.github/workflows/pull-request.yaml @@ -167,6 +167,7 @@ jobs: tool: nextest - run: docker compose -f docker-compose.yaml up -d db - run: docker compose -f docker-compose.yaml up migrations --abort-on-container-failure + - run: docker compose -f docker-compose.yaml up migrations-pool-indexer --abort-on-container-failure - uses: extractions/setup-just@53165ef7e734c5c07cb06b3c8e7b647c5aa16db3 # v4.0.0 with: just-version: 1.39.0 diff --git a/crates/database/src/lib.rs b/crates/database/src/lib.rs index 16257f6849..b47d4f7775 100644 --- a/crates/database/src/lib.rs +++ b/crates/database/src/lib.rs @@ -62,7 +62,6 @@ pub const TABLES: &[&str] = &[ "last_indexed_blocks", "onchain_order_invalidations", "onchain_placed_orders", - "pool_indexer_checkpoints", "presignature_events", "proposed_jit_orders", "quotes", @@ -72,8 +71,6 @@ pub const TABLES: &[&str] = &[ "solver_competitions", "surplus_capturing_jit_order_owners", "trades", - "uniswap_v3_pool_states", - "uniswap_v3_pools", ]; /// The names of potentially big volume tables we use in the db. @@ -88,7 +85,6 @@ pub const LARGE_TABLES: &[&str] = &[ "order_quotes", "proposed_solutions", "proposed_trade_executions", - "uniswap_v3_ticks", ]; pub fn all_tables() -> impl Iterator { @@ -98,9 +94,8 @@ pub fn all_tables() -> impl Iterator { /// Delete all data in the database. Only used by tests. /// /// Truncates all tables in a single statement so Postgres accepts foreign-key -/// cycles between listed tables (e.g. `uniswap_v3_pool_states` → -/// `uniswap_v3_pools`). Individual per-table `TRUNCATE`s error out when any -/// other listed table references the one being truncated. +/// cycles between listed tables. Individual per-table `TRUNCATE`s error out +/// when any other listed table references the one being truncated. #[expect(non_snake_case)] pub async fn clear_DANGER_(ex: &mut PgTransaction<'_>) -> sqlx::Result<()> { let tables = all_tables().collect::>().join(", "); diff --git a/crates/e2e/tests/e2e/pool_indexer.rs b/crates/e2e/tests/e2e/pool_indexer.rs index 86a9c3b84c..6ee3b4c88c 100644 --- a/crates/e2e/tests/e2e/pool_indexer.rs +++ b/crates/e2e/tests/e2e/pool_indexer.rs @@ -131,7 +131,10 @@ sol! { const POOL_INDEXER_PORT: u16 = 7778; const POOL_INDEXER_HOST: &str = "http://127.0.0.1:7778"; const POOL_INDEXER_METRICS_PORT: u16 = 7779; -const LOCAL_DB_URL: &str = "postgresql://"; +// The indexer has its own database (mirrors the per-network prod DB), migrated +// from `database/sql-pool-indexer` by the `migrations-pool-indexer` flyway step +// (docker-compose / setup-e2e.sh), separate from the shared autopilot DB. +const POOL_INDEXER_DB_URL: &str = "postgresql:///pool_indexer"; // sqrt(1) * 2^96 — valid starting price const INITIAL_SQRT_PRICE: u128 = 1u128 << 96; @@ -162,6 +165,8 @@ struct TicksResponse { #[derive(Deserialize)] struct TickEntry {} +/// Truncates the indexer's tables between tests. The schema itself is +/// provisioned by flyway (`migrations-pool-indexer`), so this just clears rows. async fn clear_pool_indexer_tables(db: &PgPool) { sqlx::query( "TRUNCATE uniswap_v3_ticks, uniswap_v3_pool_states, uniswap_v3_pools, \ @@ -190,7 +195,7 @@ async fn seed_checkpoint(db: &PgPool, factory: Address, block: u64) { async fn spawn_pool_indexer(factory: Address, metrics_port: u16) -> tokio::task::JoinHandle<()> { let config = Configuration { database: DatabaseConfig { - url: LOCAL_DB_URL.parse().unwrap(), + url: POOL_INDEXER_DB_URL.parse().unwrap(), max_connections: NonZeroU32::new(5).unwrap(), }, network: NetworkConfig { @@ -390,7 +395,7 @@ async fn driver_integration(web3: Web3) { const POOLS_BY_IDS_ROUTE: &str = "/api/v1/{network}/uniswap/v3/pools/by-ids"; const TICKS_ROUTE: &str = "/api/v1/{network}/uniswap/v3/pools/ticks"; - let db = PgPool::connect(LOCAL_DB_URL).await.unwrap(); + let db = PgPool::connect(POOL_INDEXER_DB_URL).await.unwrap(); clear_pool_indexer_tables(&db).await; let mut onchain = OnchainComponents::deploy(web3.clone()).await; @@ -482,7 +487,7 @@ async fn local_node_pool_indexer_checkpoint_resume() { /// count, sqrt_price / tick / liquidity, and the checkpoint all survive a /// stop+start. async fn checkpoint_resume(web3: Web3) { - let db = PgPool::connect(LOCAL_DB_URL).await.unwrap(); + let db = PgPool::connect(POOL_INDEXER_DB_URL).await.unwrap(); clear_pool_indexer_tables(&db).await; let (factory, pool_addr) = deploy_univ3(&web3).await; @@ -533,7 +538,7 @@ async fn local_node_pool_indexer_api_errors() { /// 400, a valid-but-unknown address must come back as 200 with empty ticks. /// Lets callers distinguish "garbage input" from "no data yet". async fn api_errors(web3: Web3) { - let db = PgPool::connect(LOCAL_DB_URL).await.unwrap(); + let db = PgPool::connect(POOL_INDEXER_DB_URL).await.unwrap(); clear_pool_indexer_tables(&db).await; let (factory, _pool) = deploy_univ3(&web3).await; @@ -585,7 +590,7 @@ async fn local_node_pool_indexer_pagination() { /// every pool exactly once. Three pools is the smallest set that exercises /// a mid-stream cursor and the `next_cursor = null` terminator. async fn pagination(web3: Web3) { - let db = PgPool::connect(LOCAL_DB_URL).await.unwrap(); + let db = PgPool::connect(POOL_INDEXER_DB_URL).await.unwrap(); clear_pool_indexer_tables(&db).await; let (factory, _pool1) = deploy_univ3(&web3).await; diff --git a/database/README.md b/database/README.md index 4a4679f9e5..68e4b3a8fe 100644 --- a/database/README.md +++ b/database/README.md @@ -532,83 +532,7 @@ Indexes: - jit\_user\_order\_creation\_timestamp: btree(`owner`, `creation_timestamp` DESC) - jit\_event\_id: btree(`block_number`, `log_index`) -### pool\_indexer\_checkpoints - -Highest finalized block processed per `contract_address` by `pool-indexer`. `contract_address` is the factory address. The indexer runs one process per network against its own DB, so there's no `chain_id` column. - - Column | Type | Nullable | Details ---------------------|--------|----------|-------- - contract\_address | bytea | not null | Factory address (20 bytes) - block\_number | bigint | not null | - -Indexes: -- PRIMARY KEY: btree (`contract_address`) - -### uniswap\_v3\_pools - -One row per pool discovered from a `PoolCreated` event. `token{0,1}_{decimals,symbol}` are nullable and filled in by the backfill task. `factory` partitions the table when multiple V3-compatible factories run on the same network so each indexer touches only its own rows. - - Column | Type | Nullable | Details --------------------|----------|----------|-------- - address | bytea | not null | Pool address (20 bytes) - factory | bytea | not null | Address of the V3 factory that emitted `PoolCreated` - token0 | bytea | not null | - token1 | bytea | not null | - fee | int | not null | Hundredths of a basis point (500 = 0.05%, 3000 = 0.3%, 10000 = 1%). `CHECK (fee > 0)`. - token0\_decimals | smallint | nullable | `NULL` = not yet fetched. `-1` = sentinel for "fetched but call failed" - token1\_decimals | smallint | nullable | - token0\_symbol | text | nullable | `NULL` = not yet fetched. `""` = sentinel for "fetched but call failed" - token1\_symbol | text | nullable | - created\_block | bigint | not null | Block in which the pool was created on-chain - -Indexes: -- PRIMARY KEY: btree (`address`) -- Four partial indexes on `(token{0,1})` with predicate `token{0,1}_{symbol,decimals} IS NULL` to power the backfill scan. - -### uniswap\_v3\_pool\_states - -Current state per pool: `sqrt_price_x96` and `tick` come from the latest `Swap`/`Initialize`; `liquidity` and `block_number` also update on in-range `Mint`/`Burn`. FK → `uniswap_v3_pools`. - -**Uniswap V3 pool-state primer.** Three values capture a pool's instantaneous state: - -- `sqrt_price_x96` — `sqrt(price) * 2^96` where `price = token1/token0`, stored in Q64.96 fixed-point. The square-root form keeps swap math additive and bounds precision loss over the uint160 range. Mirrors on-chain `slot0.sqrtPriceX96`. -- `tick` — `floor(log_{1.0001}(price))`. Each tick is a ~0.01% price step; the current tick is the bucket the live price falls into. Routers use it to decide which positions are in-range. -- `liquidity` — sum of every position's liquidity whose `tickLower <= current_tick < tickUpper`. This is the `L` in V3's invariant `Δsqrt_price = Δamount / L`. Updates on `Swap` (the event carries the new value) and on `Mint`/`Burn` whose range spans the current tick. - -The per-tick deltas that move `liquidity` when the price crosses a tick boundary live in [`uniswap_v3_ticks`](#uniswap_v3_ticks). - - Column | Type | Nullable | Details --------------------|---------|----------|-------- - pool\_address | bytea | not null | FK → `uniswap_v3_pools(address)` - block\_number | bigint | not null | Block of the most recent state-changing event (`Swap`, `Initialize`, or in-range `Mint`/`Burn`). - sqrt\_price\_x96 | numeric | not null | uint160 — see primer above - liquidity | numeric | not null | uint128 — see primer above - tick | int | not null | signed int24 — see primer above - -Indexes: -- PRIMARY KEY: btree (`pool_address`) - -### uniswap\_v3\_ticks - -Per-tick liquidity deltas. Rows with `liquidity_net = 0` are pruned. FK → `uniswap_v3_pools`. - -**Why deltas instead of per-tick totals.** A V3 position covers `[tickLower, tickUpper)` and contributes to pool liquidity only when the current tick is in that range. We store the entering / exiting deltas at the bounds: - -- At `tickLower`: `liquidity_net += position.liquidity` (entering) -- At `tickUpper`: `liquidity_net -= position.liquidity` (exiting) - -When a swap crosses a tick boundary, the pool's `liquidity` shifts by `± tick.liquidity_net`. This encoding makes the per-tick aggregate O(1) at swap time — no per-position iteration. - -Quoters consult these to predict liquidity changes at tick crossings during swap simulation. Without them, large swaps would be priced as if the liquidity stayed flat, producing wildy wrong quotes - - Column | Type | Nullable | Details -----------------|---------|----------|-------- - pool\_address | bytea | not null | FK → `uniswap_v3_pools(address)` - tick\_idx | int | not null | Tick coordinate (signed int24); same domain as [`uniswap_v3_pool_states.tick`](#uniswap_v3_pool_states) - liquidity\_net | numeric | not null | int128, signed — net liquidity entering (+) / exiting (-) at this tick - -Indexes: -- PRIMARY KEY: btree (`pool_address`, `tick_idx`) +The `pool-indexer` service uses its own per-network database, not these shared DBs. Its tables (`pool_indexer_checkpoints`, `uniswap_v3_pools`, `uniswap_v3_pool_states`, `uniswap_v3_ticks`) and migrations live in [`sql-pool-indexer/`](sql-pool-indexer/). ### cow\_amms diff --git a/database/sql-pool-indexer/README.md b/database/sql-pool-indexer/README.md new file mode 100644 index 0000000000..fec9aa2f60 --- /dev/null +++ b/database/sql-pool-indexer/README.md @@ -0,0 +1,99 @@ +# Pool-indexer migrations + +Flyway migrations for the pool-indexer's own per-network database (e.g. +`ink_pool_indexer`), kept out of the shared `../sql/` set so they don't run +against the autopilot/orderbook main DBs. + +The migration image ships both dirs; init containers pick one via `-locations`: + +| DB | location | +|---------------------|-----------------------------------------------------| +| autopilot/orderbook | `/flyway/sql` (default) | +| pool-indexer | `-locations=filesystem:/flyway/sql-pool-indexer` | + +New pool-indexer migrations go here, never in `../sql/`. `V110` is duplicated +from `../sql/` on purpose: the shared copy can't be deleted (Flyway checksums +applied migrations) so it's cancelled there by `../sql/V111`. + +## Schema + +The tables below live in the indexer's own per-network database (e.g. +`ink_pool_indexer`), created by the migrations in this directory. + +### pool\_indexer\_checkpoints + +Highest finalized block processed per `contract_address` by `pool-indexer`. `contract_address` is the factory address. The indexer runs one process per network against its own DB, so there's no `chain_id` column. + + Column | Type | Nullable | Details +--------------------|--------|----------|-------- + contract\_address | bytea | not null | Factory address (20 bytes) + block\_number | bigint | not null | + +Indexes: +- PRIMARY KEY: btree (`contract_address`) + +### uniswap\_v3\_pools + +One row per pool discovered from a `PoolCreated` event. `token{0,1}_{decimals,symbol}` are nullable and filled in by the backfill task. `factory` partitions the table when multiple V3-compatible factories run on the same network so each indexer touches only its own rows. + + Column | Type | Nullable | Details +-------------------|----------|----------|-------- + address | bytea | not null | Pool address (20 bytes) + factory | bytea | not null | Address of the V3 factory that emitted `PoolCreated` + token0 | bytea | not null | + token1 | bytea | not null | + fee | int | not null | Hundredths of a basis point (500 = 0.05%, 3000 = 0.3%, 10000 = 1%). `CHECK (fee > 0)`. + token0\_decimals | smallint | nullable | `NULL` = not yet fetched. `-1` = sentinel for "fetched but call failed" + token1\_decimals | smallint | nullable | + token0\_symbol | text | nullable | `NULL` = not yet fetched. `""` = sentinel for "fetched but call failed" + token1\_symbol | text | nullable | + created\_block | bigint | not null | Block in which the pool was created on-chain + +Indexes: +- PRIMARY KEY: btree (`address`) +- Four partial indexes on `(token{0,1})` with predicate `token{0,1}_{symbol,decimals} IS NULL` to power the backfill scan. + +### uniswap\_v3\_pool\_states + +Current state per pool: `sqrt_price_x96` and `tick` come from the latest `Swap`/`Initialize`; `liquidity` and `block_number` also update on in-range `Mint`/`Burn`. FK → `uniswap_v3_pools`. + +**Uniswap V3 pool-state primer.** Three values capture a pool's instantaneous state: + +- `sqrt_price_x96` — `sqrt(price) * 2^96` where `price = token1/token0`, stored in Q64.96 fixed-point. The square-root form keeps swap math additive and bounds precision loss over the uint160 range. Mirrors on-chain `slot0.sqrtPriceX96`. +- `tick` — `floor(log_{1.0001}(price))`. Each tick is a ~0.01% price step; the current tick is the bucket the live price falls into. Routers use it to decide which positions are in-range. +- `liquidity` — sum of every position's liquidity whose `tickLower <= current_tick < tickUpper`. This is the `L` in V3's invariant `Δsqrt_price = Δamount / L`. Updates on `Swap` (the event carries the new value) and on `Mint`/`Burn` whose range spans the current tick. + +The per-tick deltas that move `liquidity` when the price crosses a tick boundary live in [`uniswap_v3_ticks`](#uniswap_v3_ticks). + + Column | Type | Nullable | Details +-------------------|---------|----------|-------- + pool\_address | bytea | not null | FK → `uniswap_v3_pools(address)` + block\_number | bigint | not null | Block of the most recent state-changing event (`Swap`, `Initialize`, or in-range `Mint`/`Burn`). + sqrt\_price\_x96 | numeric | not null | uint160 — see primer above + liquidity | numeric | not null | uint128 — see primer above + tick | int | not null | signed int24 — see primer above + +Indexes: +- PRIMARY KEY: btree (`pool_address`) + +### uniswap\_v3\_ticks + +Per-tick liquidity deltas. Rows with `liquidity_net = 0` are pruned. FK → `uniswap_v3_pools`. + +**Why deltas instead of per-tick totals.** A V3 position covers `[tickLower, tickUpper)` and contributes to pool liquidity only when the current tick is in that range. We store the entering / exiting deltas at the bounds: + +- At `tickLower`: `liquidity_net += position.liquidity` (entering) +- At `tickUpper`: `liquidity_net -= position.liquidity` (exiting) + +When a swap crosses a tick boundary, the pool's `liquidity` shifts by `± tick.liquidity_net`. This encoding makes the per-tick aggregate O(1) at swap time — no per-position iteration. + +Quoters consult these to predict liquidity changes at tick crossings during swap simulation. Without them, large swaps would be priced as if the liquidity stayed flat, producing wildy wrong quotes + + Column | Type | Nullable | Details +----------------|---------|----------|-------- + pool\_address | bytea | not null | FK → `uniswap_v3_pools(address)` + tick\_idx | int | not null | Tick coordinate (signed int24); same domain as [`uniswap_v3_pool_states.tick`](#uniswap_v3_pool_states) + liquidity\_net | numeric | not null | int128, signed — net liquidity entering (+) / exiting (-) at this tick + +Indexes: +- PRIMARY KEY: btree (`pool_address`, `tick_idx`) diff --git a/database/sql-pool-indexer/V110__pool_indexer_uniswap_v3.sql b/database/sql-pool-indexer/V110__pool_indexer_uniswap_v3.sql new file mode 100644 index 0000000000..0993d80e1d --- /dev/null +++ b/database/sql-pool-indexer/V110__pool_indexer_uniswap_v3.sql @@ -0,0 +1,59 @@ +-- Tracks the highest finalized block fully processed per factory contract. +-- A DB instance hosts a single network, so no `chain_id` column is needed. +CREATE TABLE pool_indexer_checkpoints ( + contract_address BYTEA NOT NULL, -- factory address + block_number BIGINT NOT NULL, + PRIMARY KEY (contract_address) +); + +-- One row per pool, discovered from `PoolCreated` events. `factory` +-- partitions the table so multiple V3-compatible factories on the same +-- network can coexist (logs are fetched chain-wide, then partitioned at +-- the write boundary). +CREATE TABLE uniswap_v3_pools ( + address BYTEA NOT NULL, -- pool address + factory BYTEA NOT NULL, + token0 BYTEA NOT NULL, + token1 BYTEA NOT NULL, + fee INT NOT NULL CHECK (fee > 0), -- hundredths of a basis point (500 = 0.05%, 3000 = 0.3%, 10000 = 1%) + token0_decimals SMALLINT, + token1_decimals SMALLINT, + token0_symbol TEXT, + token1_symbol TEXT, + created_block BIGINT NOT NULL, + PRIMARY KEY (address) +); + +-- Current state per pool. `sqrt_price_x96` + `tick` come from the latest +-- Swap/Initialize; `liquidity` + `block_number` also update on in-range +-- Mint/Burn events. +CREATE TABLE uniswap_v3_pool_states ( + pool_address BYTEA NOT NULL, + block_number BIGINT NOT NULL, + sqrt_price_x96 NUMERIC NOT NULL, -- uint160 + liquidity NUMERIC NOT NULL, -- uint128 + -- `tick` here means the Uniswap V3 *price tick index* (signed + -- int24), not a database index. See `uniswap_v3_ticks.tick_idx`. + tick INT NOT NULL, + PRIMARY KEY (pool_address), + FOREIGN KEY (pool_address) REFERENCES uniswap_v3_pools(address) +); + +-- Active ticks per pool. Rows with `liquidity_net = 0` are pruned. +-- `tick_idx` is the price tick coordinate (signed int24) — same domain +-- as `uniswap_v3_pool_states.tick`, one row per active tick boundary. +CREATE TABLE uniswap_v3_ticks ( + pool_address BYTEA NOT NULL, + tick_idx INT NOT NULL, + liquidity_net NUMERIC NOT NULL, -- int128 (can be negative) + PRIMARY KEY (pool_address, tick_idx), + FOREIGN KEY (pool_address) REFERENCES uniswap_v3_pools(address) +); + +-- Symbol/decimals backfill hot paths. Partial on `IS NULL` so each +-- index shrinks to near-empty once most rows are populated (real value +-- or the `""` / `-1` "tried, failed" sentinel). +CREATE INDEX ON uniswap_v3_pools (token0) WHERE token0_symbol IS NULL; +CREATE INDEX ON uniswap_v3_pools (token1) WHERE token1_symbol IS NULL; +CREATE INDEX ON uniswap_v3_pools (token0) WHERE token0_decimals IS NULL; +CREATE INDEX ON uniswap_v3_pools (token1) WHERE token1_decimals IS NULL; diff --git a/database/sql/V111__drop_pool_indexer_uniswap_v3.sql b/database/sql/V111__drop_pool_indexer_uniswap_v3.sql new file mode 100644 index 0000000000..5dec00625c --- /dev/null +++ b/database/sql/V111__drop_pool_indexer_uniswap_v3.sql @@ -0,0 +1,13 @@ +-- Removes the pool-indexer tables from databases that run this shared +-- migration set. The indexer's schema (V110) was checked in here by mistake, +-- so every DB's flyway run created these tables even though only the dedicated +-- `*_pool_indexer` DB ever uses them. +-- +-- The indexer's own DB applies its schema from a separate migration location +-- (see `database/sql-pool-indexer/`) and never runs this. +-- Forward-only cleanup: V110 is left untouched so its checksum stays valid on +-- DBs that already applied it. Children (FK to uniswap_v3_pools) drop first. +DROP TABLE IF EXISTS uniswap_v3_pool_states; +DROP TABLE IF EXISTS uniswap_v3_ticks; +DROP TABLE IF EXISTS uniswap_v3_pools; +DROP TABLE IF EXISTS pool_indexer_checkpoints; diff --git a/docker-compose.yaml b/docker-compose.yaml index 75b479a2a5..b2fdf5544a 100644 --- a/docker-compose.yaml +++ b/docker-compose.yaml @@ -36,5 +36,42 @@ services: source: ./database/conf/ target: /flyway/conf + # Flyway can't create databases, so provision the pool-indexer's own DB first. + # Idempotent (mirrors .devcontainer/setup-e2e.sh), so it's safe on an existing + # volume too — no fresh-volume init script required. + create-pool-indexer-db: + image: postgres:16 + restart: on-failure + depends_on: + - db + environment: + PGHOST: db + PGUSER: $USER + command: + - sh + - -c + - "psql -tc \"SELECT 1 FROM pg_database WHERE datname='pool_indexer'\" | grep -q 1 || psql -c \"CREATE DATABASE pool_indexer\"" + + # Same flyway image as `migrations`, but applies the pool-indexer's own + # migration set to its own database (mirrors the per-network prod DB). + migrations-pool-indexer: + build: + context: . + target: migrations + restart: on-failure + command: migrate + depends_on: + create-pool-indexer-db: + condition: service_completed_successfully + environment: + FLYWAY_URL: jdbc:postgresql://db/pool_indexer?user=$USER&password= + volumes: + - type: bind + source: ./database/sql-pool-indexer/ + target: /flyway/sql + - type: bind + source: ./database/conf/ + target: /flyway/conf + volumes: postgres: From 8b03564e86087e76ef6ae294a52f50a970587d6c Mon Sep 17 00:00:00 2001 From: Aryan Godara <65490434+AryanGodara@users.noreply.github.com> Date: Wed, 24 Jun 2026 13:58:44 +0530 Subject: [PATCH 2/5] Split pool-indexer bootstrap from serve via --bootstrap-only flag --- crates/e2e/tests/e2e/pool_indexer.rs | 56 +++++++++++++++++++-- crates/pool-indexer/README.md | 36 +++++++++++-- crates/pool-indexer/src/arguments.rs | 37 ++++++++++++++ crates/pool-indexer/src/lib.rs | 2 +- crates/pool-indexer/src/run.rs | 75 ++++++++++++++++++++++------ 5 files changed, 180 insertions(+), 26 deletions(-) diff --git a/crates/e2e/tests/e2e/pool_indexer.rs b/crates/e2e/tests/e2e/pool_indexer.rs index 6ee3b4c88c..e8555acd44 100644 --- a/crates/e2e/tests/e2e/pool_indexer.rs +++ b/crates/e2e/tests/e2e/pool_indexer.rs @@ -190,10 +190,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: POOL_INDEXER_DB_URL.parse().unwrap(), max_connections: NonZeroU32::new(5).unwrap(), @@ -217,7 +215,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")) @@ -647,3 +651,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(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. + 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" + ); +} diff --git a/crates/pool-indexer/README.md b/crates/pool-indexer/README.md index c5f6ad583c..f810d1f457 100644 --- a/crates/pool-indexer/README.md +++ b/crates/pool-indexer/README.md @@ -9,6 +9,27 @@ 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 ` 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 ` 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. + +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. + ## Running locally Create `crates/pool-indexer/config.local.toml` first (schema = the @@ -17,12 +38,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 +# create + migrate the indexer's own database via flyway (database/sql-pool-indexer) +docker compose up migrations-pool-indexer + +# one process for both phases: +cargo run --release -p pool-indexer -- --config crates/pool-indexer/config.local.toml + +# or split bootstrap from serve: +cargo run --release -p pool-indexer -- --bootstrap-only true --config crates/pool-indexer/config.local.toml cargo run --release -p pool-indexer -- --config crates/pool-indexer/config.local.toml ``` diff --git a/crates/pool-indexer/src/arguments.rs b/crates/pool-indexer/src/arguments.rs index ce5085cf1f..656a776903 100644 --- a/crates/pool-indexer/src/arguments.rs +++ b/crates/pool-indexer/src/arguments.rs @@ -7,6 +7,43 @@ 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 the slow seed run as a separate step ahead of + /// serving. + #[clap(long, env, action = clap::ArgAction::Set, default_value_t = false)] + 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", + "true", + ]); + assert!(bootstrap.bootstrap_only); + + // ArgAction::Set takes a value, so it can be explicitly turned off too. + let explicit_serve = Arguments::parse_from([ + "pool-indexer", + "--config", + "/tmp/c.toml", + "--bootstrap-only", + "false", + ]); + assert!(!explicit_serve.bootstrap_only); + } +} diff --git a/crates/pool-indexer/src/lib.rs b/crates/pool-indexer/src/lib.rs index 7a12692bfb..ab36f57324 100644 --- a/crates/pool-indexer/src/lib.rs +++ b/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; diff --git a/crates/pool-indexer/src/run.rs b/crates/pool-indexer/src/run.rs index b9ce460fcf..60f32fb0f5 100644 --- a/crates/pool-indexer/src/run.rs +++ b/crates/pool-indexer/src/run.rs @@ -21,8 +21,47 @@ pub async fn start(args: impl Iterator) { 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 — meant to run as a separate step +/// ahead of serving. +/// +/// 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); + + // Seed every factory concurrently, like the serve path. + 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 db = db.clone(); + let network = network.clone(); + factory_set.spawn(async move { + bootstrap_factory(&db, &indexer, &network, &factory).await; + }); + } + while let Some(result) = factory_set.join_next().await { + result.expect("bootstrap task panicked"); + } } pub async fn run(config: Configuration) { @@ -125,20 +164,7 @@ async fn run_network_indexer(db: PgPool, network: NetworkConfig, barrier: Arc 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()) From cf905d5d6d4d3f43124492436e11c231d5d4c8a6 Mon Sep 17 00:00:00 2001 From: Aryan Godara <65490434+AryanGodara@users.noreply.github.com> Date: Thu, 25 Jun 2026 04:19:42 +0530 Subject: [PATCH 3/5] Fetch uncached V3 pools on demand so cold quotes find liquidity --- .../src/uniswap_v3/graph_api.rs | 7 + .../src/uniswap_v3/pool_fetching.rs | 319 +++++++++++++++--- 2 files changed, 287 insertions(+), 39 deletions(-) diff --git a/crates/liquidity-sources/src/uniswap_v3/graph_api.rs b/crates/liquidity-sources/src/uniswap_v3/graph_api.rs index 39d0245e2f..db436a3342 100644 --- a/crates/liquidity-sources/src/uniswap_v3/graph_api.rs +++ b/crates/liquidity-sources/src/uniswap_v3/graph_api.rs @@ -260,6 +260,13 @@ impl V3PoolDataSource for UniV3SubgraphClient { ids: &[Address], target_block: u64, ) -> Result { + // `0` means "latest available" (see the V3PoolDataSource contract). The + // subgraph needs a concrete block — `block: { number: 0 }` queries genesis + // — so resolve a recent safe block first. + let target_block = match target_block { + 0 => self.get_safe_block().await?, + n => n, + }; let (pools, ticks) = futures::try_join!( self.get_pools_by_pool_ids(ids, target_block), self.get_ticks_by_pools_ids(ids, target_block) diff --git a/crates/liquidity-sources/src/uniswap_v3/pool_fetching.rs b/crates/liquidity-sources/src/uniswap_v3/pool_fetching.rs index 5ad81425ec..154e72d9d8 100644 --- a/crates/liquidity-sources/src/uniswap_v3/pool_fetching.rs +++ b/crates/liquidity-sources/src/uniswap_v3/pool_fetching.rs @@ -220,18 +220,19 @@ impl PoolsCheckpointHandler { }) } - /// For a given list of token pairs, fetches the pools for the ones that - /// exist in the checkpoint. For the ones that don't exist, flag as - /// missing and expect to exist after the next maintenance run. - fn get(&self, token_pairs: &HashSet) -> (HashMap>, u64) { + /// Returns cached pools for the pairs, plus the ids of any that exist but + /// aren't cached yet. Misses are recorded for the next maintenance run; the + /// quote path fetches them on demand. + fn get( + &self, + token_pairs: &HashSet, + ) -> (HashMap>, Vec
, u64) { let mut pool_ids = token_pairs .iter() .filter_map(|pair| self.pools_by_token_pair.get(pair)) .flatten() .peekable(); - tracing::trace!("get checkpoint for pool_ids: {:?}", pool_ids); - match pool_ids.peek() { Some(_) => { let mut pools_checkpoint = self.pools_checkpoint.lock().unwrap(); @@ -244,6 +245,7 @@ impl PoolsCheckpointHandler { } None => true, }) + .copied() .collect::>(); tracing::trace!( @@ -251,50 +253,85 @@ impl PoolsCheckpointHandler { existing_pools.keys(), missing_pools ); - pools_checkpoint.missing_pools.extend(missing_pools); - (existing_pools, pools_checkpoint.block_number) + pools_checkpoint + .missing_pools + .extend(missing_pools.iter().copied()); + (existing_pools, missing_pools, pools_checkpoint.block_number) } None => Default::default(), } } - /// Fetches state/ticks for missing pools and moves them from - /// `missing_pools` to `pools` - async fn update_missing_pools(&self) -> Result<()> { - let (missing_pools, block_number) = { - let checkpoint = self.pools_checkpoint.lock().unwrap(); - if checkpoint.missing_pools.is_empty() { - return Ok(()); - } - (checkpoint.missing_pools.clone(), checkpoint.block_number) - }; - tracing::debug!("currently missing pools are {:?}", missing_pools); - - let pool_ids = missing_pools.into_iter().collect::>(); - let start = std::time::Instant::now(); + /// Fetches and converts the given pools from the source at `target_block`, + /// skipping any that can't be converted yet (e.g. no ticks). Doesn't touch + /// the cache. Pass the checkpoint block to cache for replay, or 0 to take + /// the indexer's current snapshot without blocking. + async fn fetch_pools( + &self, + pool_ids: &[Address], + target_block: u64, + ) -> Result)>> { + if pool_ids.is_empty() { + return Ok(Vec::new()); + } let pools_with_ticks = self .source - .get_pools_with_ticks_by_ids(&pool_ids, block_number) - .await; - tracing::debug!( - requested_pools = pool_ids.len(), - time = ?start.elapsed(), - request_successful = pools_with_ticks.is_ok(), - "fetched pool ticks" - ); - let pools_with_ticks = pools_with_ticks?; + .get_pools_with_ticks_by_ids(pool_ids, target_block) + .await?; + Ok(pools_with_ticks + .pools + .into_iter() + .filter_map(|pool| { + let id = pool.id; + match PoolInfo::try_from(pool) { + Ok(info) => Some((id, Arc::new(info))), + Err(err) => { + tracing::debug!(?id, ?err, "skipping pool missing tick data"); + None + } + } + }) + .collect()) + } - let mut checkpoint = self.pools_checkpoint.lock().unwrap(); - for pool in pools_with_ticks.pools { - checkpoint.missing_pools.remove(&pool.id); - checkpoint.pools.insert(pool.id, Arc::new(pool.try_into()?)); + /// Fetches the given pools at the indexer's current block, without waiting. + /// The checkpoint block can sit ahead of the indexer (it tracks latest + /// minus the reorg buffer, the indexer serves finalized), so waiting + /// for it would hang the quote. Returns empty on failure so the quote + /// falls back to cached pools. + async fn fetch_current(&self, pool_ids: &[Address]) -> Vec<(Address, Arc)> { + match self.fetch_pools(pool_ids, 0).await { + Ok(pools) => pools, + Err(err) => { + tracing::debug!( + ?err, + "on-demand pool fetch failed; serving cached pools only" + ); + Vec::new() + } } + } - tracing::debug!("number of cached pools is {}", checkpoint.pools.len()); + /// Fetches the pools flagged missing by `get` at the checkpoint block and + /// caches them. Runs from periodic maintenance. + async fn update_missing_pools(&self) -> Result<()> { + let (missing, block_number) = { + let checkpoint = self.pools_checkpoint.lock().unwrap(); + ( + checkpoint.missing_pools.iter().copied().collect::>(), + checkpoint.block_number, + ) + }; + let fetched = self.fetch_pools(&missing, block_number).await?; + let mut checkpoint = self.pools_checkpoint.lock().unwrap(); + for (id, info) in fetched { + checkpoint.missing_pools.remove(&id); + checkpoint.pools.insert(id, info); + } if !checkpoint.missing_pools.is_empty() { tracing::warn!( - "not all missing pools updated: {:?}", - checkpoint.missing_pools + remaining = checkpoint.missing_pools.len(), + "not all missing pools updated" ); } Ok(()) @@ -420,7 +457,13 @@ impl PoolFetching for UniswapV3PoolFetcher { // this is the only place where this function uses checkpoint - no data racing // between maintenance - let (mut checkpoint, checkpoint_block_number) = self.checkpoint.get(token_pairs); + let (mut checkpoint, missing, checkpoint_block_number) = self.checkpoint.get(token_pairs); + + // No pools registered for these pairs: `get` returns block 0, so skip the + // replay below (it would otherwise scan events from block 1) and return. + if checkpoint.is_empty() && missing.is_empty() { + return Ok(Vec::new()); + } if block_number > checkpoint_block_number { let block_range = RangeInclusive::try_new(checkpoint_block_number + 1, block_number)?; @@ -428,6 +471,14 @@ impl PoolFetching for UniswapV3PoolFetcher { append_events(&mut checkpoint, events); } + // The warm cache only holds the top pools by raw liquidity, so plenty of + // real pairs are absent. Fetch those on demand instead of waiting for the + // next maintenance tick. They come back current, so merge them after the + // replay rather than replaying them. + if !missing.is_empty() { + checkpoint.extend(self.checkpoint.fetch_current(&missing).await); + } + // return only pools which current liquidity is positive Ok(checkpoint .into_values() @@ -745,4 +796,194 @@ mod tests { ]) ); } + + /// Serves a fixed set of pools (with ticks) from + /// `get_pools_with_ticks_by_ids` so the on-demand fetch path can be + /// exercised without a real source. `served_block` models the indexer's + /// head: a request for a higher `target_block` fails, mirroring the real + /// client's `wait_until` blocking on a block the indexer hasn't reached. + struct StubSource { + with_ticks: HashMap, + served_block: u64, + } + + impl StubSource { + fn new(pools: impl IntoIterator) -> Self { + Self { + with_ticks: pools.into_iter().map(|p| (p.id, p)).collect(), + served_block: u64::MAX, + } + } + } + + #[async_trait::async_trait] + impl V3PoolDataSource for StubSource { + async fn get_registered_pools( + &self, + _target_block: u64, + ) -> Result { + Ok(Default::default()) + } + + async fn get_pools_with_ticks_by_ids( + &self, + ids: &[Address], + target_block: u64, + ) -> Result { + anyhow::ensure!( + target_block <= self.served_block, + "indexer at {} hasn't reached target block {target_block}", + self.served_block, + ); + let pools = ids + .iter() + .filter_map(|id| self.with_ticks.get(id).cloned()) + .collect(); + Ok(crate::uniswap_v3::graph_api::PoolsWithTicks { + fetched_block_number: self.served_block, + pools, + }) + } + } + + fn pool_with_ticks(id: Address, token0: Address, token1: Address) -> PoolData { + PoolData { + id, + token0: Token { + id: token0, + decimals: 6, + }, + token1: Token { + id: token1, + decimals: 18, + }, + fee_tier: U256::from(3000), + liquidity: U256::from(1_000_000u64), + sqrt_price: U256::from(1u64), + tick: 0, + ticks: Some(vec![crate::uniswap_v3::graph_api::TickData { + tick_idx: -100, + liquidity_net: 1_000, + pool_address: id, + }]), + block_number: 100, + } + } + + fn handler(source: StubSource, checkpoint: PoolsCheckpoint) -> PoolsCheckpointHandler { + PoolsCheckpointHandler { + source: Arc::new(source), + pools_by_token_pair: HashMap::new(), + pools_checkpoint: Mutex::new(checkpoint), + } + } + + /// A pool registered for a pair but absent from the warm cache (the + /// raw-liquidity-ranked top-N) is flagged missing by `get` and resolved by + /// a single on-demand `fetch_current`, without waiting for a + /// maintenance cycle. + #[tokio::test] + async fn on_demand_fetch_serves_uncached_pool() { + let token0 = Address::with_last_byte(1); + let token1 = Address::with_last_byte(2); + let pair = TokenPair::new(token0, token1).unwrap(); + let pool = Address::with_last_byte(9); + + let mut handler = handler( + StubSource::new([pool_with_ticks(pool, token0, token1)]), + PoolsCheckpoint { + pools: HashMap::new(), + block_number: 100, + missing_pools: HashSet::new(), + }, + ); + handler.pools_by_token_pair = HashMap::from([(pair, vec![pool])]); + + // Cold: not cached, flagged missing. + let (cached, missing, _) = handler.get(&HashSet::from([pair])); + assert!(cached.is_empty()); + assert_eq!(missing, vec![pool]); + + // On-demand fetch resolves it. + let fetched = handler.fetch_current(&missing).await; + assert_eq!(fetched.len(), 1); + assert_eq!(fetched[0].0, pool); + } + + /// The on-demand path must not block on the checkpoint block (which can sit + /// persistently ahead of the indexer's served block); it fetches at the + /// indexer's current block. A source that errors for any block above its + /// head still yields the pool via `fetch_current`. + #[tokio::test] + async fn on_demand_does_not_wait_for_future_block() { + let token0 = Address::with_last_byte(1); + let token1 = Address::with_last_byte(2); + let pool = Address::with_last_byte(9); + + let mut source = StubSource::new([pool_with_ticks(pool, token0, token1)]); + source.served_block = 50; // indexer behind the checkpoint below + + let handler = handler( + source, + PoolsCheckpoint { + pools: HashMap::new(), + block_number: 100, // checkpoint ahead of the indexer + missing_pools: HashSet::new(), + }, + ); + + // fetch_current uses target 0, so it succeeds despite served_block < + // checkpoint. + let fetched = handler.fetch_current(&[pool]).await; + assert_eq!(fetched.len(), 1); + assert_eq!(fetched[0].0, pool); + + // Fetching at the checkpoint block would fail (indexer hasn't reached it). + assert!(handler.fetch_pools(&[pool], 100).await.is_err()); + } + + /// Unknown pairs have no registered pools, so `get` reports block 0 with + /// nothing cached or missing — the signal `fetch` uses to skip the replay. + #[test] + fn get_reports_zero_block_for_unknown_pairs() { + let handler = handler( + StubSource::new([]), + PoolsCheckpoint { + pools: HashMap::new(), + block_number: 100, + missing_pools: HashSet::new(), + }, + ); + let pair = TokenPair::new(Address::with_last_byte(1), Address::with_last_byte(2)).unwrap(); + let (cached, missing, block) = handler.get(&HashSet::from([pair])); + assert!(cached.is_empty()); + assert!(missing.is_empty()); + assert_eq!(block, 0); + } + + /// A pool that can't be converted (e.g. ticks not yet available) is skipped + /// rather than failing the whole batch; the convertible pool is returned. + #[tokio::test] + async fn fetch_pools_skips_unconvertible_pool() { + let token0 = Address::with_last_byte(1); + let token1 = Address::with_last_byte(2); + let good = Address::with_last_byte(9); + let bad = Address::with_last_byte(10); + + let mut bad_pool = pool_with_ticks(bad, token0, token1); + bad_pool.ticks = None; // PoolInfo::try_from fails on missing ticks + + let handler = handler( + StubSource::new([pool_with_ticks(good, token0, token1), bad_pool]), + PoolsCheckpoint { + pools: HashMap::new(), + block_number: 100, + missing_pools: HashSet::new(), + }, + ); + + let fetched = handler.fetch_pools(&[good, bad], 0).await.unwrap(); + assert_eq!(fetched.len(), 1); + assert_eq!(fetched[0].0, good); + } } From 7f46f13e308c1c962a0fb745ff58ed1931f835a7 Mon Sep 17 00:00:00 2001 From: Aryan Godara <65490434+AryanGodara@users.noreply.github.com> Date: Thu, 25 Jun 2026 03:42:14 +0530 Subject: [PATCH 4/5] Drop the pool-indexer wait-until-timeout config --- crates/driver/example.toml | 2 +- .../src/boundary/liquidity/uniswap/v3.rs | 2 -- crates/driver/src/infra/config/file/load.rs | 14 +++----- crates/driver/src/infra/config/file/mod.rs | 35 ++++++++++--------- crates/driver/src/infra/liquidity/config.rs | 5 --- .../src/uniswap_v3/pool_indexer.rs | 22 ++++++------ 6 files changed, 35 insertions(+), 45 deletions(-) diff --git a/crates/driver/example.toml b/crates/driver/example.toml index 7925978367..d82b4ee910 100644 --- a/crates/driver/example.toml +++ b/crates/driver/example.toml @@ -98,7 +98,7 @@ max-order-age = "1m" # [[liquidity.uniswap-v3]] # Uniswap V3 configuration (pool-indexer as data source) # preset = "uniswap-v3" -# indexer-config = { pool-indexer = { url = "http://pool-indexer/", wait-until-timeout = "5m" } } +# indexer-config = { pool-indexer = { url = "http://pool-indexer/" } } # max_pools_to_initialize = 100 # [[liquidity.uniswap-v3]] # Custom Uniswap V3 configuration diff --git a/crates/driver/src/boundary/liquidity/uniswap/v3.rs b/crates/driver/src/boundary/liquidity/uniswap/v3.rs index cc8bd148b4..74dc6a44a3 100644 --- a/crates/driver/src/boundary/liquidity/uniswap/v3.rs +++ b/crates/driver/src/boundary/liquidity/uniswap/v3.rs @@ -157,14 +157,12 @@ async fn build_pool_data_source( UniswapV3PoolSource::PoolIndexer(indexer) => { tracing::info!( url = %indexer.url, - wait_until_timeout = ?indexer.wait_until_timeout, "uniswap v3: using pool-indexer as data source", ); Ok(Arc::new(PoolIndexerClient::new( indexer.url.clone(), eth.chain(), http, - indexer.wait_until_timeout, ))) } UniswapV3PoolSource::Subgraph(subgraph) => { diff --git a/crates/driver/src/infra/config/file/load.rs b/crates/driver/src/infra/config/file/load.rs index 68f54c3507..f501de6477 100644 --- a/crates/driver/src/infra/config/file/load.rs +++ b/crates/driver/src/infra/config/file/load.rs @@ -380,15 +380,11 @@ fn uniswap_v3_pool_source( max_pools_per_tick_query, }) } - file::IndexerConfig::PoolIndexer { - url, - wait_until_timeout, - } => liquidity::config::UniswapV3PoolSource::PoolIndexer( - liquidity::config::UniswapV3PoolIndexer { - url, - wait_until_timeout, - }, - ), + file::IndexerConfig::PoolIndexer { url } => { + liquidity::config::UniswapV3PoolSource::PoolIndexer( + liquidity::config::UniswapV3PoolIndexer { url }, + ) + } } } diff --git a/crates/driver/src/infra/config/file/mod.rs b/crates/driver/src/infra/config/file/mod.rs index 69db5aa29b..00496059f6 100644 --- a/crates/driver/src/infra/config/file/mod.rs +++ b/crates/driver/src/infra/config/file/mod.rs @@ -637,16 +637,7 @@ enum IndexerConfig { max_pools_per_tick_query: usize, }, #[serde(rename_all = "kebab-case")] - PoolIndexer { - url: Url, - /// Upper bound on a single `wait_until` call. Size per-network to - /// comfortably exceed the worst-case first-deploy seed time. - #[serde( - with = "humantime_serde", - default = "uniswap_v3::default_pool_indexer_wait_until_timeout" - )] - wait_until_timeout: Duration, - }, + PoolIndexer { url: Url }, } #[derive(Clone, Debug, Deserialize)] @@ -656,8 +647,6 @@ enum UniswapV3Preset { } mod uniswap_v3 { - use std::time::Duration; - pub fn default_max_pools_to_initialize() -> usize { 100 } @@ -665,10 +654,6 @@ mod uniswap_v3 { pub fn default_max_pools_per_tick_query() -> usize { usize::MAX } - - pub fn default_pool_indexer_wait_until_timeout() -> Duration { - Duration::from_secs(300) - } } #[derive(Clone, Debug, Deserialize)] @@ -1205,6 +1190,24 @@ mod tests { assert!(err.to_string().contains("must be signers")); } + #[test] + fn pool_indexer_config_needs_only_a_url() { + let config: IndexerConfig = + toml::from_str(r#"pool-indexer = { url = "http://pool-indexer/" }"#).unwrap(); + assert!(matches!(config, IndexerConfig::PoolIndexer { .. })); + } + + #[test] + fn pool_indexer_config_rejects_wait_until_timeout() { + // The field was dropped (bootstrap is its own initContainer now); a + // stale config still carrying it should fail loudly, not be ignored. + let err = toml::from_str::( + r#"pool-indexer = { url = "http://pool-indexer/", wait-until-timeout = "5m" }"#, + ) + .unwrap_err(); + assert!(err.to_string().contains("wait-until-timeout")); + } + #[test] fn submission_accounts_new_accepts_signers() { let signer = Account::PrivateKey( diff --git a/crates/driver/src/infra/liquidity/config.rs b/crates/driver/src/infra/liquidity/config.rs index 1b7a69df84..b73b893052 100644 --- a/crates/driver/src/infra/liquidity/config.rs +++ b/crates/driver/src/infra/liquidity/config.rs @@ -175,11 +175,6 @@ pub struct UniswapV3PoolIndexer { /// Service root, e.g. `http://pool-indexer/` exposing /// `/api/v1/{network}/uniswap/v3/`. pub url: Url, - - /// Upper bound on a single `wait_until` call. Size per-network to - /// comfortably exceed the worst-case first-deploy seed time (~13 min - /// for mainnet's ~60k pools; tens of seconds for smaller chains). - pub wait_until_timeout: Duration, } /// Where Uniswap V3 pool definitions and tick data are fetched from. Exactly diff --git a/crates/liquidity-sources/src/uniswap_v3/pool_indexer.rs b/crates/liquidity-sources/src/uniswap_v3/pool_indexer.rs index 2717184a79..51cefc9fa0 100644 --- a/crates/liquidity-sources/src/uniswap_v3/pool_indexer.rs +++ b/crates/liquidity-sources/src/uniswap_v3/pool_indexer.rs @@ -30,6 +30,11 @@ use { /// Poll interval for [`PoolIndexerClient::wait_until`]. const WAIT_UNTIL_POLL_INTERVAL: Duration = Duration::from_millis(500); +/// Cap on a single `wait_until` call. Bootstrap now runs as a separate +/// initContainer, so the serve container is up within seconds — this only +/// covers the indexer coming up at startup, not the old cold-bootstrap time. +const WAIT_UNTIL_TIMEOUT: Duration = Duration::from_secs(60); + /// Matches the server-side `MAX_POOL_IDS_PER_REQUEST`. const POOL_IDS_PER_REQUEST: usize = 500; @@ -39,20 +44,13 @@ const LIST_PAGE_SIZE: u64 = 5000; pub struct PoolIndexerClient { base_url: Url, http: Client, - /// Cap on a single `wait_until` call. Pick per network to exceed the - /// worst-case fresh-deploy seed time. - wait_until_timeout: Duration, } impl PoolIndexerClient { - pub fn new(base_url: Url, chain: Chain, http: Client, wait_until_timeout: Duration) -> Self { + pub fn new(base_url: Url, chain: Chain, http: Client) -> Self { let prefix = format!("api/v1/{}/uniswap/v3/", chain.as_str()); let base_url = url_join(&base_url, &prefix); - Self { - base_url, - http, - wait_until_timeout, - } + Self { base_url, http } } fn path(&self, suffix: &str) -> Url { @@ -215,12 +213,12 @@ impl IndexerTick { impl PoolIndexerClient { /// Polls `/pools?limit=1` every [`WAIT_UNTIL_POLL_INTERVAL`] until the /// envelope reports `block_number >= target_block`. Returns - /// immediately if already there; bails after [`Self::wait_until_timeout`]. + /// immediately if already there; bails after [`WAIT_UNTIL_TIMEOUT`]. /// /// `503` is treated as "still bootstrapping" and the loop keeps /// polling. Other non-2xx statuses propagate as errors. async fn wait_until(&self, target_block: u64) -> Result<()> { - let deadline = std::time::Instant::now() + self.wait_until_timeout; + let deadline = std::time::Instant::now() + WAIT_UNTIL_TIMEOUT; let mut last_observed: Option = None; let mut interval = tokio::time::interval(WAIT_UNTIL_POLL_INTERVAL); interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); @@ -247,7 +245,7 @@ impl PoolIndexerClient { anyhow::bail!( "pool-indexer wait_until exceeded {:?} waiting for block {target_block}; last \ observed indexer block: {last_observed:?}", - self.wait_until_timeout, + WAIT_UNTIL_TIMEOUT, ); } } From 17d6582665ac189c8af4f055f7ed33a46b7dbafa Mon Sep 17 00:00:00 2001 From: Aryan Godara <65490434+AryanGodara@users.noreply.github.com> Date: Sun, 28 Jun 2026 00:28:26 +0530 Subject: [PATCH 5/5] Add Balancer V2 mock contracts for the indexer e2e test --- crates/e2e/tests/e2e/balancer_v2_indexer.rs | 186 ++++++++++++++++++++ crates/e2e/tests/e2e/main.rs | 1 + 2 files changed, 187 insertions(+) create mode 100644 crates/e2e/tests/e2e/balancer_v2_indexer.rs diff --git a/crates/e2e/tests/e2e/balancer_v2_indexer.rs b/crates/e2e/tests/e2e/balancer_v2_indexer.rs new file mode 100644 index 0000000000..6a0dc66331 --- /dev/null +++ b/crates/e2e/tests/e2e/balancer_v2_indexer.rs @@ -0,0 +1,186 @@ +//! Mock Balancer V2 contracts the balancer-indexer e2e test deploys on Anvil, +//! plus a smoke test that exercises the discovery surface the indexer relies +//! on: the factory's `PoolCreated`, the pool's `getPoolId` / +//! `getNormalizedWeights`, and the vault's `getPoolTokens`. + +use { + alloy::{ + primitives::{Address, U256}, + providers::Provider, + rpc::types::Filter, + sol, + sol_types::SolEvent, + }, + e2e::setup::run_test, + ethrpc::Web3, +}; + +// Compiled from the Solidity below with solc 0.8.30 --optimize +// --optimize-runs 1000000, evm-version shanghai. Faithful enough for +// discovery: the vault computes canonical poolIds (first 20 bytes = pool +// address) and serves getPoolTokens; the factory emits PoolCreated; the +// pool exposes getPoolId / getNormalizedWeights. +// +// // SPDX-License-Identifier: GPL-3.0-or-later +// pragma solidity ^0.8.17; +// contract MockBalancerV2Vault { +// uint256 private nextNonce; +// mapping(bytes32 => address[]) private tokensOf; +// event PoolRegistered( +// bytes32 indexed poolId, address indexed pool, uint8 spec +// ); +// event TokensRegistered( +// bytes32 indexed poolId, address[] tokens, address[] mgrs +// ); +// // poolId layout matches Balancer: address == poolId[0:20]. +// function registerPool(uint8 spec) external returns (bytes32 id) { +// id = bytes32( +// (uint256(uint160(msg.sender)) << 96) +// | (uint256(spec) << 80) | (nextNonce++) +// ); +// emit PoolRegistered(id, msg.sender, spec); +// } +// function registerTokens(bytes32 id, address[] calldata tokens) +// external +// { +// tokensOf[id] = tokens; +// emit TokensRegistered(id, tokens, new address[](tokens.length)); +// } +// function getPoolTokens(bytes32 id) +// external +// view +// returns (address[] memory t, uint256[] memory b, uint256 lcb) +// { +// t = tokensOf[id]; +// b = new uint256[](t.length); +// lcb = 0; +// } +// } +// contract MockBalancerV2Pool { +// bytes32 public poolId; +// uint256[] private weights; +// constructor( +// MockBalancerV2Vault vault, +// address[] memory tokens, +// uint256[] memory _weights +// ) { +// weights = _weights; +// poolId = vault.registerPool(0); +// vault.registerTokens(poolId, tokens); +// } +// function getPoolId() external view returns (bytes32) { +// return poolId; +// } +// function getNormalizedWeights() +// external +// view +// returns (uint256[] memory) +// { +// return weights; +// } +// } +// contract MockBalancerV2PoolFactory { +// MockBalancerV2Vault public immutable vault; +// event PoolCreated(address indexed pool); +// constructor(MockBalancerV2Vault _vault) { vault = _vault; } +// function create( +// address[] calldata tokens, uint256[] calldata weights +// ) external returns (address pool) { +// pool = address(new MockBalancerV2Pool(vault, tokens, weights)); +// emit PoolCreated(pool); +// } +// } +sol! { + #[allow(missing_docs)] + #[sol(rpc, bytecode = "0x6080604052348015600e575f5ffd5b506105a78061001c5f395ff3fe608060405234801561000f575f5ffd5b506004361061003f575f3560e01c806309b2760f14610043578063200bf51914610069578063f94d46681461007e575b5f5ffd5b61005661005136600461030f565b6100a0565b6040519081526020015b60405180910390f35b61007c610077366004610336565b61010e565b005b61009161008c3660046103b0565b6101aa565b60405161006093929190610417565b5f805481806100ae83610477565b9091555060405160ff8416815233606081901b6aff00000000000000000000605087901b161792909217925082907f3c13bc30b8e878c53fd2a36b679409c073afd75950be43d8858768e956fbc20e9060200160405180910390a3919050565b5f838152600160205260409020610126908383610275565b50827ff5847d3f2197b16cdcd2098ec95d0905cd1abdaf415f07bb7cef2bba8ac5dec483838067ffffffffffffffff811115610164576101646104d3565b60405190808252806020026020018201604052801561018d578160200160208202803683370190505b5060405161019d93929190610500565b60405180910390a2505050565b6060805f60015f8581526020019081526020015f2080548060200260200160405190810160405280929190818152602001828054801561021e57602002820191905f5260205f20905b815473ffffffffffffffffffffffffffffffffffffffff1681526001909101906020018083116101f3575b50505050509250825167ffffffffffffffff81111561023f5761023f6104d3565b604051908082528060200260200182016040528015610268578160200160208202803683370190505b5091505f90509193909250565b828054828255905f5260205f209081019282156102eb579160200282015b828111156102eb5781547fffffffffffffffffffffffff00000000000000000000000000000000000000001673ffffffffffffffffffffffffffffffffffffffff843516178255602090920191600190910190610293565b506102f79291506102fb565b5090565b5b808211156102f7575f81556001016102fc565b5f6020828403121561031f575f5ffd5b813560ff8116811461032f575f5ffd5b9392505050565b5f5f5f60408486031215610348575f5ffd5b83359250602084013567ffffffffffffffff811115610365575f5ffd5b8401601f81018613610375575f5ffd5b803567ffffffffffffffff81111561038b575f5ffd5b8660208260051b840101111561039f575f5ffd5b939660209190910195509293505050565b5f602082840312156103c0575f5ffd5b5035919050565b5f8151808452602084019350602083015f5b8281101561040d57815173ffffffffffffffffffffffffffffffffffffffff168652602095860195909101906001016103d9565b5093949350505050565b606081525f61042960608301866103c7565b82810360208401528085518083526020830191506020870192505f5b81811015610463578351835260209384019390920191600101610445565b505060409390930193909352509392505050565b5f7fffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff82036104cc577f4e487b71000000000000000000000000000000000000000000000000000000005f52601160045260245ffd5b5060010190565b7f4e487b71000000000000000000000000000000000000000000000000000000005f52604160045260245ffd5b604080825281018390525f8460608301825b8681101561055357823573ffffffffffffffffffffffffffffffffffffffff811680821461053e575f5ffd5b83525060209283019290910190600101610512565b50838103602085015261056681866103c7565b97965050505050505056fea2646970667358221220382d38d59c2a76b5cbd3260e689658fc8a242378a758a6af234957185f5c20ee64736f6c634300081e0033")] + contract MockBalancerV2Vault { + function getPoolTokens(bytes32 poolId) external view returns ( + address[] memory tokens, + uint256[] memory balances, + uint256 lastChangeBlock + ); + } +} + +sol! { + #[allow(missing_docs)] + #[sol(rpc, bytecode = "0x60a0604052348015600e575f5ffd5b50604051610868380380610868833981016040819052602b91603b565b6001600160a01b03166080526066565b5f60208284031215604a575f5ffd5b81516001600160a01b0381168114605f575f5ffd5b9392505050565b6080516107e56100835f395f818160790152609e01526107e55ff3fe608060405234801561000f575f5ffd5b5060043610610034575f3560e01c806358d95a9414610038578063fbfa77cf14610074575b5f5ffd5b61004b610046366004610195565b61009b565b60405173ffffffffffffffffffffffffffffffffffffffff909116815260200160405180910390f35b61004b7f000000000000000000000000000000000000000000000000000000000000000081565b5f7f0000000000000000000000000000000000000000000000000000000000000000858585856040516100cd90610140565b6100db959493929190610201565b604051809103905ff0801580156100f4573d5f5f3e3d5ffd5b5060405190915073ffffffffffffffffffffffffffffffffffffffff8216907f83a48fbcfc991335314e74d0496aab6a1987e992ddc85dddbcc4d6dd6ef2e9fc905f90a2949350505050565b6104e8806102c883390190565b5f5f83601f84011261015d575f5ffd5b50813567ffffffffffffffff811115610174575f5ffd5b6020830191508360208260051b850101111561018e575f5ffd5b9250929050565b5f5f5f5f604085870312156101a8575f5ffd5b843567ffffffffffffffff8111156101be575f5ffd5b6101ca8782880161014d565b909550935050602085013567ffffffffffffffff8111156101e9575f5ffd5b6101f58782880161014d565b95989497509550505050565b73ffffffffffffffffffffffffffffffffffffffff8616815260606020820181905281018490525f8560808301825b8781101561027157823573ffffffffffffffffffffffffffffffffffffffff811680821461025c575f5ffd5b83525060209283019290910190600101610230565b5083810360408501525f91508481527f07ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff8511156102ac575f5ffd5b8460051b80876020840137016020019897505050505050505056fe608060405234801561000f575f5ffd5b506040516104e83803806104e883398101604081905261002e91610253565b805161004190600190602084019061010f565b506040516309b2760f60e01b81525f60048201526001600160a01b038416906309b2760f906024016020604051808303815f875af1158015610085573d5f5f3e3d5ffd5b505050506040513d601f19601f820116820180604052508101906100a9919061032a565b5f81905560405163200bf51960e01b81526001600160a01b0385169163200bf519916100da91908690600401610341565b5f604051808303815f87803b1580156100f1575f5ffd5b505af1158015610103573d5f5f3e3d5ffd5b50505050505050610397565b828054828255905f5260205f20908101928215610148579160200282015b8281111561014857825182559160200191906001019061012d565b50610154929150610158565b5090565b5b80821115610154575f8155600101610159565b6001600160a01b0381168114610180575f5ffd5b50565b634e487b7160e01b5f52604160045260245ffd5b604051601f8201601f191681016001600160401b03811182821017156101bf576101bf610183565b604052919050565b5f6001600160401b038211156101df576101df610183565b5060051b60200190565b5f82601f8301126101f8575f5ffd5b815161020b610206826101c7565b610197565b8082825260208201915060208360051b86010192508583111561022c575f5ffd5b602085015b83811015610249578051835260209283019201610231565b5095945050505050565b5f5f5f60608486031215610265575f5ffd5b83516102708161016c565b60208501519093506001600160401b0381111561028b575f5ffd5b8401601f8101861361029b575f5ffd5b80516102a9610206826101c7565b8082825260208201915060208360051b8501019250888311156102ca575f5ffd5b6020840193505b828410156102f55783516102e48161016c565b8252602093840193909101906102d1565b6040880151909550925050506001600160401b03811115610314575f5ffd5b610320868287016101e9565b9150509250925092565b5f6020828403121561033a575f5ffd5b5051919050565b5f60408201848352604060208401528084518083526060850191506020860192505f5b8181101561038b5783516001600160a01b0316835260209384019390920191600101610364565b50909695505050505050565b610144806103a45f395ff3fe608060405234801561000f575f5ffd5b506004361061003f575f3560e01c806338fff2d0146100435780633e0dc34e14610059578063f89f27ed14610061575b5f5ffd5b5f545b6040519081526020015b60405180910390f35b6100465f5481565b610069610076565b60405161005091906100cc565b606060018054806020026020016040519081016040528092919081815260200182805480156100c257602002820191905f5260205f20905b8154815260200190600101908083116100ae575b5050505050905090565b602080825282518282018190525f918401906040840190835b818110156101035783518352602093840193909201916001016100e5565b50909594505050505056fea2646970667358221220731a2e9e654b6bac0ab562f25edd604d513cae5f8f5211bfd8a05e2006f7ec0764736f6c634300081e0033a26469706673582212204184aeb80c6b23da766cee6de765fd97e95a2f3c68407416e85c49d336124d6564736f6c634300081e0033")] + contract MockBalancerV2PoolFactory { + constructor(address vault); + + event PoolCreated(address indexed pool); + + function create( + address[] calldata tokens, + uint256[] calldata weights + ) external returns (address pool); + } +} + +sol! { + #[allow(missing_docs)] + #[sol(rpc)] + contract MockBalancerV2Pool { + function getPoolId() external view returns (bytes32); + function getNormalizedWeights() external view returns (uint256[] memory); + } +} + +#[tokio::test] +#[ignore] +async fn local_node_balancer_indexer_mocks() { + run_test(mocks_smoke).await; +} + +/// Deploys the mock Vault + factory, creates a 2-token pool, and asserts the +/// discovery surface the indexer reads: `PoolCreated` carries the pool address, +/// `getPoolId()` is the address-derived poolId, `getPoolTokens()` returns the +/// registered tokens, and `getNormalizedWeights()` round-trips. +async fn mocks_smoke(web3: Web3) { + let provider = web3.provider.clone().erased(); + + let vault = MockBalancerV2Vault::deploy(provider.clone()).await.unwrap(); + let factory = MockBalancerV2PoolFactory::deploy(provider.clone(), *vault.address()) + .await + .unwrap(); + + let token0 = Address::repeat_byte(1); + let token1 = Address::repeat_byte(2); + let weight = U256::from(500_000_000_000_000_000u64); // 0.5 in Bfp 1e18 + + factory + .create(vec![token0, token1], vec![weight, weight]) + .send() + .await + .unwrap() + .watch() + .await + .unwrap(); + + let block = provider.get_block_number().await.unwrap(); + let logs = provider + .get_logs( + &Filter::new() + .from_block(block) + .to_block(block) + .event_signature(MockBalancerV2PoolFactory::PoolCreated::SIGNATURE_HASH), + ) + .await + .unwrap(); + let pool_addr = MockBalancerV2PoolFactory::PoolCreated::decode_log(&logs[0].inner) + .unwrap() + .data + .pool; + + let pool = MockBalancerV2Pool::MockBalancerV2PoolInstance::new(pool_addr, provider.clone()); + let pool_id = pool.getPoolId().call().await.unwrap(); + + // Balancer invariant: the first 20 bytes of the poolId are the pool address. + assert_eq!(&pool_id.0[..20], pool_addr.as_slice()); + + let weights = pool.getNormalizedWeights().call().await.unwrap(); + assert_eq!(weights, vec![weight, weight]); + + let tokens = vault.getPoolTokens(pool_id).call().await.unwrap().tokens; + assert_eq!(tokens, vec![token0, token1]); +} diff --git a/crates/e2e/tests/e2e/main.rs b/crates/e2e/tests/e2e/main.rs index 630d76d77f..28f1adfc72 100644 --- a/crates/e2e/tests/e2e/main.rs +++ b/crates/e2e/tests/e2e/main.rs @@ -8,6 +8,7 @@ mod api_version; mod app_data; mod app_data_signer; mod autopilot_leader; +mod balancer_v2_indexer; mod banned_users; mod buffers; mod contract_revert;