From 7fbbe456ff9f183528151280bf1c569e529b7076 Mon Sep 17 00:00:00 2001 From: Aryan Godara Date: Thu, 27 Aug 2026 16:28:48 +0530 Subject: [PATCH 1/2] add bal-v2 pools API to pool-indexer --- .../pool-indexer/src/api/balancer_v2/mod.rs | 118 ++++++++++++++++ .../src/api/balancer_v2/pools_by_ids.rs | 36 +++++ .../src/api/balancer_v2/pools_list.rs | 58 ++++++++ crates/pool-indexer/src/api/mod.rs | 16 ++- crates/pool-indexer/src/api/routes.rs | 8 +- .../src/api/uniswap_v3/bulk_ticks.rs | 2 +- .../src/api/uniswap_v3/pool_ticks.rs | 2 +- .../src/api/uniswap_v3/pools_by_ids.rs | 2 +- .../src/api/uniswap_v3/pools_list.rs | 2 +- crates/pool-indexer/src/db/balancer_v2.rs | 129 +++++++++++++++++- crates/pool-indexer/src/db/mod.rs | 30 +++- crates/pool-indexer/src/db/uniswap_v3.rs | 23 ---- crates/pool-indexer/src/run.rs | 8 +- 13 files changed, 396 insertions(+), 38 deletions(-) create mode 100644 crates/pool-indexer/src/api/balancer_v2/mod.rs create mode 100644 crates/pool-indexer/src/api/balancer_v2/pools_by_ids.rs create mode 100644 crates/pool-indexer/src/api/balancer_v2/pools_list.rs diff --git a/crates/pool-indexer/src/api/balancer_v2/mod.rs b/crates/pool-indexer/src/api/balancer_v2/mod.rs new file mode 100644 index 0000000000..52e03cbf1f --- /dev/null +++ b/crates/pool-indexer/src/api/balancer_v2/mod.rs @@ -0,0 +1,118 @@ +pub mod pools_by_ids; +pub mod pools_list; + +use { + crate::db::balancer_v2 as db, + alloy_primitives::{Address, B256}, + axum::{ + Json, + response::{IntoResponse, Response}, + }, + serde::{Deserialize, Deserializer, Serialize}, +}; +pub use {pools_by_ids::get_pools_by_ids, pools_list::get_pools}; + +/// Max pool ids per bulk lookup. Keeps URLs under proxy limits and caps the DB +/// query size. +pub(super) const MAX_POOL_IDS_PER_REQUEST: usize = 500; + +/// Deserializes `?pool_ids=0x…,0x…` into 32-byte pool ids. Parsing and the cap +/// happen in the extractor so handlers see a `Vec`. +pub(crate) struct PoolIds(pub Vec); + +impl<'de> Deserialize<'de> for PoolIds { + fn deserialize>(de: D) -> Result { + let raw = String::deserialize(de)?; + let out: Vec = raw + .split(',') + .map(str::trim) + .filter(|s| !s.is_empty()) + .map(|entry| { + entry + .parse::() + .map_err(|_| serde::de::Error::custom("invalid pool id")) + }) + .collect::>()?; + if out.len() > MAX_POOL_IDS_PER_REQUEST { + return Err(serde::de::Error::custom(format!( + "too many pool ids; max {MAX_POOL_IDS_PER_REQUEST}" + ))); + } + Ok(PoolIds(out)) + } +} + +/// One token of a pool, in `getPoolTokens` order. +#[derive(Serialize)] +pub struct TokenInfo { + pub address: Address, + pub decimals: u8, + /// Normalized weight as a decimal fraction (e.g. "0.5"); present only for + /// weighted pools. + #[serde(skip_serializing_if = "Option::is_none")] + pub weight: Option, +} + +/// A single Balancer V2 pool. Field names mirror the driver's discovery type. +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +pub struct PoolResponse { + pub pool_type: String, + pub id: B256, + pub address: Address, + pub factory: Address, + /// Discovery-time flag; the driver re-checks it on-chain for LBP pools. + /// Always `true` here, since the indexer stores static metadata only. + pub swap_enabled: bool, + pub tokens: Vec, +} + +#[derive(Serialize)] +pub struct PoolsResponse { + /// Latest block every configured balancer factory is indexed through. + pub block_number: u64, + pub pools: Vec, + /// Pass as `after=` to fetch the next page; `null` on the last page. + pub next_cursor: Option, +} + +impl From for TokenInfo { + fn from(t: db::BalancerTokenRow) -> Self { + Self { + address: t.address, + decimals: t.decimals, + // `normalized()` strips the trailing zeros Postgres NUMERIC pads on + // when decoded through its base-10000 digit groups. Without it a + // weight can render with >18 fractional digits, which the driver's + // fixed-point parser rejects. + weight: t.weight.map(|w| w.normalized().to_string()), + } + } +} + +impl From for PoolResponse { + fn from(row: db::BalancerPoolRow) -> Self { + Self { + pool_type: row.pool_type, + id: row.pool_id, + address: row.address, + factory: row.factory, + swap_enabled: true, + tokens: row.tokens.into_iter().map(TokenInfo::from).collect(), + } + } +} + +/// Shared `PoolsResponse` builder for the balancer listing endpoints. +pub(super) fn pools_response( + block_number: u64, + pools: Vec, + next_cursor: Option, +) -> Response { + Json(PoolsResponse { + block_number, + pools: pools.into_iter().map(PoolResponse::from).collect(), + next_cursor, + }) + .into_response() +} diff --git a/crates/pool-indexer/src/api/balancer_v2/pools_by_ids.rs b/crates/pool-indexer/src/api/balancer_v2/pools_by_ids.rs new file mode 100644 index 0000000000..1a55346798 --- /dev/null +++ b/crates/pool-indexer/src/api/balancer_v2/pools_by_ids.rs @@ -0,0 +1,36 @@ +//! `GET /api/v1/{network}/balancer/v2/pools/by-ids?pool_ids=…` + +use { + super::{PoolIds, pools_response}, + crate::{ + api::{ApiError, AppState, latest_indexed_block}, + db::balancer_v2 as db, + }, + axum::{ + extract::{Query, State}, + response::Response, + }, + serde::Deserialize, + std::sync::Arc, +}; + +#[derive(Deserialize)] +pub struct BulkLookupQuery { + /// Comma-separated 32-byte pool ids. Capped at + /// [`super::MAX_POOL_IDS_PER_REQUEST`]; clients with more should chunk. + pub pool_ids: PoolIds, +} + +/// Pools matching `pool_ids`, sorted by `pool_id`. Unknown ids are skipped +/// silently; treat a partial response as "these are the ones I have". +/// +/// `block_number` is read first so the envelope is never *newer* than the row +/// data (the indexer can advance between the two reads, never regress). +pub async fn get_pools_by_ids( + State(state): State>, + Query(BulkLookupQuery { pool_ids }): Query, +) -> Result { + let block = latest_indexed_block(&state.db, &state.balancer_v2_factories).await?; + let pools = db::get_pools_by_ids(&state.db, &pool_ids.0).await?; + Ok(pools_response(block, pools, None)) +} diff --git a/crates/pool-indexer/src/api/balancer_v2/pools_list.rs b/crates/pool-indexer/src/api/balancer_v2/pools_list.rs new file mode 100644 index 0000000000..75784ee21d --- /dev/null +++ b/crates/pool-indexer/src/api/balancer_v2/pools_list.rs @@ -0,0 +1,58 @@ +//! `GET /api/v1/{network}/balancer/v2/pools` — cursor-paginated pool list. + +use { + super::pools_response, + crate::{ + api::{ApiError, AppState, latest_indexed_block}, + db::balancer_v2 as db, + }, + alloy_primitives::B256, + axum::{ + extract::{Query, State}, + response::Response, + }, + serde::Deserialize, + std::sync::Arc, +}; + +const DEFAULT_PAGE_LIMIT: u64 = 1_000; + +/// Hard server-side cap on `limit`. Applied even if the client asks for more. +const MAX_PAGE_LIMIT: u64 = 5_000; + +#[derive(Deserialize)] +pub struct ListPoolsQuery { + /// Cursor from the previous page (the last-seen `pool_id`); omit to start + /// from the beginning. + pub after: Option, + /// Clamped to `[1, MAX_PAGE_LIMIT]`; defaults to `DEFAULT_PAGE_LIMIT`. + pub limit: Option, +} + +impl ListPoolsQuery { + fn page_limit(&self) -> u64 { + self.limit + .unwrap_or(DEFAULT_PAGE_LIMIT) + .clamp(1, MAX_PAGE_LIMIT) + } +} + +/// All indexed pools, sorted by `pool_id`. +pub async fn get_pools( + State(state): State>, + Query(query): Query, +) -> Result { + let block_number = latest_indexed_block(&state.db, &state.balancer_v2_factories).await?; + let limit = query.page_limit(); + let cursor = query.after.map(|id| id.as_slice().to_vec()); + + let mut rows = db::get_pools(&state.db, cursor, limit + 1).await?; + + let has_next = rows.len() > limit as usize; + rows.truncate(limit as usize); + let next_cursor = has_next + .then(|| rows.last().map(|row| format!("{:#x}", row.pool_id))) + .flatten(); + + Ok(pools_response(block_number, rows, next_cursor)) +} diff --git a/crates/pool-indexer/src/api/mod.rs b/crates/pool-indexer/src/api/mod.rs index 2dc462acc2..d59165ee0b 100644 --- a/crates/pool-indexer/src/api/mod.rs +++ b/crates/pool-indexer/src/api/mod.rs @@ -1,3 +1,4 @@ +pub mod balancer_v2; pub mod routes; pub mod uniswap_v3; @@ -19,9 +20,11 @@ pub struct AppState { /// The network this process indexes. Requests whose `{network}` path /// segment doesn't match get a 404. pub network: NetworkName, - /// Configured factory addresses; the served envelope block is scoped to - /// these so a removed factory's stale checkpoint can't pin it. - pub factories: BTreeSet
, + /// Uniswap V3 factories. The served envelope block is scoped to these so a + /// removed factory's stale checkpoint can't pin it. + pub uniswap_v3_factories: BTreeSet
, + /// Balancer V2 factories, scoping the balancer envelope block likewise. + pub balancer_v2_factories: BTreeSet
, } impl AppState { @@ -62,8 +65,11 @@ impl From for ApiError { } } -pub(super) async fn latest_indexed_block(state: &AppState) -> Result { - crate::db::uniswap_v3::get_latest_indexed_block(&state.db, &state.factories) +pub(super) async fn latest_indexed_block( + db: &PgPool, + factories: &BTreeSet
, +) -> Result { + crate::db::get_latest_indexed_block(db, factories) .await? .ok_or(ApiError::NotReady) } diff --git a/crates/pool-indexer/src/api/routes.rs b/crates/pool-indexer/src/api/routes.rs index 8e95733e20..69fa4e20e4 100644 --- a/crates/pool-indexer/src/api/routes.rs +++ b/crates/pool-indexer/src/api/routes.rs @@ -1,7 +1,7 @@ //! HTTP routing for the pool-indexer API. use { - super::{ApiError, AppState, uniswap_v3}, + super::{ApiError, AppState, balancer_v2, uniswap_v3}, axum::{ Router, extract::{MatchedPath, Path, Request, State}, @@ -26,9 +26,15 @@ pub fn router(state: Arc) -> Router { .route("/pools/{pool_address}/ticks", get(uniswap_v3::get_ticks)) .route_layer(middleware::from_fn_with_state(state.clone(), network_guard)); + let balancer_v2_routes = Router::new() + .route("/pools", get(balancer_v2::get_pools)) + .route("/pools/by-ids", get(balancer_v2::get_pools_by_ids)) + .route_layer(middleware::from_fn_with_state(state.clone(), network_guard)); + Router::new() .route("/health", get(health)) .nest("/api/v1/{network}/uniswap/v3", v3_routes) + .nest("/api/v1/{network}/balancer/v2", balancer_v2_routes) .with_state(state) .layer(middleware::from_fn(record_request_metrics)) .layer( diff --git a/crates/pool-indexer/src/api/uniswap_v3/bulk_ticks.rs b/crates/pool-indexer/src/api/uniswap_v3/bulk_ticks.rs index aaeb8539e3..84a7f867c1 100644 --- a/crates/pool-indexer/src/api/uniswap_v3/bulk_ticks.rs +++ b/crates/pool-indexer/src/api/uniswap_v3/bulk_ticks.rs @@ -44,7 +44,7 @@ pub async fn get_ticks_bulk( Query(BulkTicksQuery { pool_ids }): Query, ) -> Result { let (block, ticks) = tokio::join!( - latest_indexed_block(&state), + latest_indexed_block(&state.db, &state.uniswap_v3_factories), db::get_ticks_for_pools(&state.db, &pool_ids.0), ); diff --git a/crates/pool-indexer/src/api/uniswap_v3/pool_ticks.rs b/crates/pool-indexer/src/api/uniswap_v3/pool_ticks.rs index bc2ebe97a8..9c40756a42 100644 --- a/crates/pool-indexer/src/api/uniswap_v3/pool_ticks.rs +++ b/crates/pool-indexer/src/api/uniswap_v3/pool_ticks.rs @@ -28,7 +28,7 @@ pub async fn get_ticks( Path((_network, pool)): Path<(String, Address)>, ) -> Result { let (block, ticks) = tokio::join!( - latest_indexed_block(&state), + latest_indexed_block(&state.db, &state.uniswap_v3_factories), db::get_ticks(&state.db, &pool), ); diff --git a/crates/pool-indexer/src/api/uniswap_v3/pools_by_ids.rs b/crates/pool-indexer/src/api/uniswap_v3/pools_by_ids.rs index 079418d31e..7a7c268d89 100644 --- a/crates/pool-indexer/src/api/uniswap_v3/pools_by_ids.rs +++ b/crates/pool-indexer/src/api/uniswap_v3/pools_by_ids.rs @@ -31,7 +31,7 @@ pub async fn get_pools_by_ids( State(state): State>, Query(BulkLookupQuery { pool_ids }): Query, ) -> Result { - let block = latest_indexed_block(&state).await?; + let block = latest_indexed_block(&state.db, &state.uniswap_v3_factories).await?; let pools = db::get_pools_by_ids(&state.db, &pool_ids.0).await?; Ok(pools_response(block, &pools, None)) } diff --git a/crates/pool-indexer/src/api/uniswap_v3/pools_list.rs b/crates/pool-indexer/src/api/uniswap_v3/pools_list.rs index 4b4aa3f5ec..55f6561151 100644 --- a/crates/pool-indexer/src/api/uniswap_v3/pools_list.rs +++ b/crates/pool-indexer/src/api/uniswap_v3/pools_list.rs @@ -42,7 +42,7 @@ pub async fn get_pools( State(state): State>, Query(query): Query, ) -> Result { - let block_number = latest_indexed_block(&state).await?; + let block_number = latest_indexed_block(&state.db, &state.uniswap_v3_factories).await?; let limit = query.page_limit(); let cursor = query.after.map(|addr| addr.as_slice().to_vec()); diff --git a/crates/pool-indexer/src/db/balancer_v2.rs b/crates/pool-indexer/src/db/balancer_v2.rs index c30674771d..0c2fa669a4 100644 --- a/crates/pool-indexer/src/db/balancer_v2.rs +++ b/crates/pool-indexer/src/db/balancer_v2.rs @@ -1,9 +1,13 @@ use { - crate::{db::bytes_to_addr, indexer::balancer_v2::NewBalancerPool}, - alloy_primitives::Address, + crate::{ + db::{bytes_to_addr, bytes_to_b256}, + indexer::balancer_v2::NewBalancerPool, + }, + alloy_primitives::{Address, B256}, anyhow::{Context, Result}, bigdecimal::BigDecimal, - sqlx::{PgPool, Postgres, Row, Transaction}, + sqlx::{PgPool, Postgres, Row, Transaction, postgres::PgRow}, + std::collections::HashMap, }; /// Inserts discovered pools and their tokens. Pools are written before tokens @@ -114,3 +118,122 @@ pub async fn batch_set_token_decimals( .context("batch_set_token_decimals")?; Ok(()) } + +/// A discovered pool plus its `getPoolTokens`-ordered tokens, for the read API. +pub struct BalancerPoolRow { + pub pool_id: B256, + pub address: Address, + pub factory: Address, + pub pool_type: String, + pub tokens: Vec, +} + +/// One token of a pool, in registration order. `decimals` is always present: +/// pools with an unresolved-decimals token are excluded from the read path. +pub struct BalancerTokenRow { + pub address: Address, + pub decimals: u8, + pub weight: Option, +} + +/// Pools sorted by `pool_id`, paginated via `cursor` (the last-seen `pool_id`). +/// Only pools whose every token has resolved decimals are returned; a pool with +/// an unresolved token isn't servable (the driver requires `decimals`). +pub async fn get_pools( + pool: &PgPool, + cursor: Option>, + limit: u64, +) -> Result> { + let rows = sqlx::query( + "SELECT pool_id, address, factory, pool_type + FROM balancer_v2_pools p + WHERE ($1::BYTEA IS NULL OR pool_id > $1) + AND NOT EXISTS ( + SELECT 1 FROM balancer_v2_pool_tokens t + WHERE t.pool_id = p.pool_id AND (t.decimals IS NULL OR t.decimals < 0) + ) + ORDER BY pool_id + LIMIT $2", + ) + .bind(cursor) + .bind(limit.cast_signed()) + .fetch_all(pool) + .await + .context("balancer get_pools")?; + + assemble_pools(pool, rows).await +} + +/// Pools matching `pool_ids`, sorted by `pool_id`. Unknown ids and pools with +/// an unresolved-decimals token are skipped. +pub async fn get_pools_by_ids(pool: &PgPool, pool_ids: &[B256]) -> Result> { + if pool_ids.is_empty() { + return Ok(Vec::new()); + } + let ids: Vec<&[u8]> = pool_ids.iter().map(|id| id.as_slice()).collect(); + let rows = sqlx::query( + "SELECT pool_id, address, factory, pool_type + FROM balancer_v2_pools p + WHERE pool_id = ANY($1) + AND NOT EXISTS ( + SELECT 1 FROM balancer_v2_pool_tokens t + WHERE t.pool_id = p.pool_id AND (t.decimals IS NULL OR t.decimals < 0) + ) + ORDER BY pool_id", + ) + .bind(ids) + .fetch_all(pool) + .await + .context("balancer get_pools_by_ids")?; + + assemble_pools(pool, rows).await +} + +/// Loads the tokens for `pool_rows` in one query and attaches them in +/// `position` order. Callers restrict to pools with complete decimals, so each +/// `decimals` decodes as a plain `u8`. +async fn assemble_pools(pool: &PgPool, pool_rows: Vec) -> Result> { + if pool_rows.is_empty() { + return Ok(Vec::new()); + } + let pool_ids: Vec> = pool_rows.iter().map(|r| r.get("pool_id")).collect(); + let ids: Vec<&[u8]> = pool_ids.iter().map(|v| v.as_slice()).collect(); + + let token_rows = sqlx::query( + "SELECT pool_id, token, decimals, weight + FROM balancer_v2_pool_tokens + WHERE pool_id = ANY($1) + ORDER BY pool_id, position", + ) + .bind(ids) + .fetch_all(pool) + .await + .context("balancer pool tokens")?; + + let mut tokens: HashMap, Vec> = HashMap::new(); + for row in token_rows { + let decimals: i16 = row.get("decimals"); + tokens + .entry(row.get("pool_id")) + .or_default() + .push(BalancerTokenRow { + address: bytes_to_addr(row.get("token"))?, + decimals: u8::try_from(decimals).context("token decimals out of range")?, + weight: row.get("weight"), + }); + } + + pool_rows + .into_iter() + .map(|r| { + let pool_id: Vec = r.get("pool_id"); + Ok(BalancerPoolRow { + tokens: tokens.remove(&pool_id).unwrap_or_default(), + pool_id: bytes_to_b256(&pool_id)?, + address: bytes_to_addr(r.get("address"))?, + factory: bytes_to_addr(r.get("factory"))?, + pool_type: r.get("pool_type"), + }) + }) + .collect() +} diff --git a/crates/pool-indexer/src/db/mod.rs b/crates/pool-indexer/src/db/mod.rs index 433ed0f816..d8d5165957 100644 --- a/crates/pool-indexer/src/db/mod.rs +++ b/crates/pool-indexer/src/db/mod.rs @@ -2,9 +2,10 @@ pub mod balancer_v2; pub mod uniswap_v3; use { - alloy_primitives::Address, + alloy_primitives::{Address, B256}, anyhow::{Context, Result}, sqlx::{PgPool, Postgres, Row, Transaction}, + std::collections::BTreeSet, }; /// Decodes a Postgres `BYTEA` column into an [`Address`]. @@ -12,6 +13,11 @@ pub(crate) fn bytes_to_addr(b: Vec) -> Result
{ Address::try_from(b.as_slice()).context("invalid address bytes") } +/// Decodes a Postgres `BYTEA` column into a 32-byte [`B256`] Balancer pool id. +pub(crate) fn bytes_to_b256(b: &[u8]) -> Result { + B256::try_from(b).context("invalid pool_id bytes") +} + /// Highest block scanned for a factory's `PoolCreated` events. Shared by both /// indexers: `pool_indexer_checkpoints` is keyed by factory address, which is /// unique across protocols, so their rows never collide. @@ -44,3 +50,25 @@ pub async fn set_checkpoint( .context("set_checkpoint")?; Ok(()) } + +/// The block every configured factory is indexed through, so every served pool +/// is current at least to here. Scoped to `factories` so a decommissioned +/// factory's leftover checkpoint row can't pin the value. +pub async fn get_latest_indexed_block( + pool: &PgPool, + factories: &BTreeSet
, +) -> Result> { + let factories: Vec<&[u8]> = factories.iter().map(|f| f.as_slice()).collect(); + let row = sqlx::query( + "SELECT MIN(block_number) AS block FROM pool_indexer_checkpoints + WHERE contract_address = ANY($1)", + ) + .bind(factories) + .fetch_one(pool) + .await + .context("get_latest_indexed_block")?; + + Ok(row + .get::, _>("block") + .map(|b| b.cast_unsigned())) +} diff --git a/crates/pool-indexer/src/db/uniswap_v3.rs b/crates/pool-indexer/src/db/uniswap_v3.rs index cb195428e0..25efd5dcfa 100644 --- a/crates/pool-indexer/src/db/uniswap_v3.rs +++ b/crates/pool-indexer/src/db/uniswap_v3.rs @@ -9,7 +9,6 @@ use { num::ToPrimitive, number::conversions::u160_to_big_decimal, sqlx::{PgPool, Postgres, Row, Transaction, postgres::PgRow}, - std::collections::BTreeSet, }; fn address_bytes_list(addresses: &[Address]) -> Vec<&[u8]> { @@ -580,25 +579,3 @@ pub async fn batch_set_token_symbols( Ok(()) } - -/// The block every configured factory is indexed through, so every served pool -/// is current at least to here. Scoped to `factories` so a decommissioned -/// factory's leftover checkpoint row can't pin the value. -pub async fn get_latest_indexed_block( - pool: &PgPool, - factories: &BTreeSet
, -) -> Result> { - let factories: Vec<&[u8]> = factories.iter().map(|f| f.as_slice()).collect(); - let row = sqlx::query( - "SELECT MIN(block_number) AS block FROM pool_indexer_checkpoints - WHERE contract_address = ANY($1)", - ) - .bind(factories) - .fetch_one(pool) - .await - .context("get_latest_indexed_block")?; - - Ok(row - .get::, _>("block") - .map(|b| b.cast_unsigned())) -} diff --git a/crates/pool-indexer/src/run.rs b/crates/pool-indexer/src/run.rs index 0810445a91..5bf28119b7 100644 --- a/crates/pool-indexer/src/run.rs +++ b/crates/pool-indexer/src/run.rs @@ -182,12 +182,18 @@ fn build_api_state(db: &PgPool, network: &NetworkConfig) -> Arc { Arc::new(AppState { db: db.clone(), network: network.name.clone(), - factories: network + uniswap_v3_factories: network .uniswap_v3 .iter() .flat_map(|u| &u.factories) .map(|f| f.address) .collect(), + balancer_v2_factories: network + .balancer_v2 + .iter() + .flat_map(balancer_v2::configured_factories) + .map(|(_, factory)| factory.address) + .collect(), }) } From fed2f488f870df02451cb9c4a1ccc85091339754 Mon Sep 17 00:00:00 2001 From: Aryan Godara Date: Thu, 27 Aug 2026 17:57:19 +0530 Subject: [PATCH 2/2] replace balancer-v2 subgraph with pool-indexer in driver --- crates/driver/example.toml | 3 +- .../src/boundary/liquidity/balancer/v2/mod.rs | 8 +- crates/driver/src/infra/config/file/load.rs | 8 +- crates/driver/src/infra/config/file/mod.rs | 8 +- crates/driver/src/infra/liquidity/config.rs | 8 +- .../src/balancer_v2/graph_api.rs | 481 ------------------ .../liquidity-sources/src/balancer_v2/mod.rs | 19 +- .../src/balancer_v2/models.rs | 283 +++++++++++ .../src/balancer_v2/pool_fetching/mod.rs | 11 +- .../src/balancer_v2/pool_indexer.rs | 147 ++++++ .../src/balancer_v2/pool_init.rs | 27 - .../src/balancer_v2/pools/common.rs | 4 +- .../balancer_v2/pools/composable_stable.rs | 4 +- .../pools/liquidity_bootstrapping.rs | 4 +- .../src/balancer_v2/pools/mod.rs | 2 +- .../src/balancer_v2/pools/stable.rs | 4 +- .../src/balancer_v2/pools/weighted.rs | 4 +- crates/liquidity-sources/src/lib.rs | 1 - crates/liquidity-sources/src/macros.rs | 12 - crates/liquidity-sources/src/subgraph.rs | 303 ----------- 20 files changed, 479 insertions(+), 862 deletions(-) delete mode 100644 crates/liquidity-sources/src/balancer_v2/graph_api.rs create mode 100644 crates/liquidity-sources/src/balancer_v2/models.rs create mode 100644 crates/liquidity-sources/src/balancer_v2/pool_indexer.rs delete mode 100644 crates/liquidity-sources/src/balancer_v2/pool_init.rs delete mode 100644 crates/liquidity-sources/src/subgraph.rs diff --git a/crates/driver/example.toml b/crates/driver/example.toml index aeb4580ea1..397cd13a28 100644 --- a/crates/driver/example.toml +++ b/crates/driver/example.toml @@ -75,7 +75,7 @@ max-order-age = "1m" # [[liquidity.balancer-v2]] # Balancer V2 configuration # preset = "balancer-v2" -# graph-url = "http://localhost:1234" # which subgraph url to fetch the data from +# indexer-url = "http://localhost:7777" # pool-indexer base URL to fetch pool data from # pool-deny-list = [] # optional # [[liquidity.balancer-v2]] # Custom Balancer V2 configuration @@ -83,6 +83,7 @@ max-order-age = "1m" # weighted = [] # weighted pool factory addresses # stable = [] # stable pool factory addresses # liquidity-bootstrapping = [] # liquidity bootstrapping pool factory addresses +# indexer-url = "http://localhost:7777" # pool-indexer base URL to fetch pool data from # pool-deny-list = [] # which pools to ignore # [[liquidity.uniswap-v3]] # Uniswap V3 configuration (pool-indexer as data source) diff --git a/crates/driver/src/boundary/liquidity/balancer/v2/mod.rs b/crates/driver/src/boundary/liquidity/balancer/v2/mod.rs index 350a687555..6d61a96397 100644 --- a/crates/driver/src/boundary/liquidity/balancer/v2/mod.rs +++ b/crates/driver/src/boundary/liquidity/balancer/v2/mod.rs @@ -22,6 +22,7 @@ use { liquidity_sources::balancer_v2::{ BalancerPoolFetcher, pool_fetching::{BalancerContracts, BalancerFactoryInstance}, + pool_indexer::BalancerIndexerClient, }, shared::http_solver::model::TokenAmount, solver::{ @@ -171,12 +172,15 @@ async fn init_liquidity( let balancer_pool_fetcher = Arc::new( BalancerPoolFetcher::new( - &config.graph_url, + Box::new(BalancerIndexerClient::new( + config.indexer_url.clone(), + eth.chain(), + boundary::liquidity::http_client(), + )), block_retriever.clone(), token_info_fetcher.clone(), boundary::liquidity::cache_config(), block_stream.clone(), - boundary::liquidity::http_client(), web3.clone(), &contracts, config.pool_deny_list.to_vec(), diff --git a/crates/driver/src/infra/config/file/load.rs b/crates/driver/src/infra/config/file/load.rs index 4b53756e4e..5f096d2fda 100644 --- a/crates/driver/src/infra/config/file/load.rs +++ b/crates/driver/src/infra/config/file/load.rs @@ -265,14 +265,14 @@ pub async fn load(chain: Chain, path: &Path) -> infra::Config { file::BalancerV2Config::Preset { preset, pool_deny_list, - graph_url, + indexer_url, reinit_interval, } => liquidity::config::BalancerV2 { pool_deny_list: pool_deny_list.clone(), reinit_interval, ..match preset { file::BalancerV2Preset::BalancerV2 => { - liquidity::config::BalancerV2::balancer_v2(&graph_url, chain) + liquidity::config::BalancerV2::balancer_v2(&indexer_url, chain) } } .expect("no Balancer V2 preset for current network") @@ -285,7 +285,7 @@ pub async fn load(chain: Chain, path: &Path) -> infra::Config { liquidity_bootstrapping, composable_stable, pool_deny_list, - graph_url, + indexer_url, reinit_interval, } => liquidity::config::BalancerV2 { vault: vault.into(), @@ -295,7 +295,7 @@ pub async fn load(chain: Chain, path: &Path) -> infra::Config { liquidity_bootstrapping, composable_stable, pool_deny_list: pool_deny_list.clone(), - graph_url, + indexer_url, reinit_interval, }, }) diff --git a/crates/driver/src/infra/config/file/mod.rs b/crates/driver/src/infra/config/file/mod.rs index fdd3f9502a..006a48ce09 100644 --- a/crates/driver/src/infra/config/file/mod.rs +++ b/crates/driver/src/infra/config/file/mod.rs @@ -636,8 +636,8 @@ enum BalancerV2Config { #[serde(default)] pool_deny_list: Vec, - /// The URL used to connect to balancer v2 subgraph client. - graph_url: Url, + /// Base URL of the pool-indexer that seeds the pool registry. + indexer_url: Url, /// How often the liquidity source should be reinitialized to get /// access to new pools. @@ -677,8 +677,8 @@ enum BalancerV2Config { #[serde(default)] pool_deny_list: Vec, - /// The URL used to connect to balancer v2 subgraph client. - graph_url: Url, + /// Base URL of the pool-indexer that seeds the pool registry. + indexer_url: Url, /// How often the liquidity source should be reinitialized to get /// access to new pools. diff --git a/crates/driver/src/infra/liquidity/config.rs b/crates/driver/src/infra/liquidity/config.rs index fd48020c2b..260c5d60d5 100644 --- a/crates/driver/src/infra/liquidity/config.rs +++ b/crates/driver/src/infra/liquidity/config.rs @@ -224,8 +224,8 @@ pub struct BalancerV2 { /// ignored. pub pool_deny_list: Vec, - /// The base URL used to connect to balancer v2 subgraph client. - pub graph_url: Url, + /// Base URL of the pool-indexer that seeds the pool registry. + pub indexer_url: Url, /// How often the liquidty source should be re-initialized to become /// aware of new pools. @@ -235,7 +235,7 @@ pub struct BalancerV2 { impl BalancerV2 { /// Returns the liquidity configuration for Balancer V2. #[expect(clippy::self_named_constructors)] - pub fn balancer_v2(graph_url: &Url, chain: Chain) -> Option { + pub fn balancer_v2(indexer_url: &Url, chain: Chain) -> Option { macro_rules! address_for { ( $chain:expr, [ $( $($p:ident)::+ ),* $(,)? ] ) => {{ let arr = [ $({ @@ -282,7 +282,7 @@ impl BalancerV2 { ] ), pool_deny_list: Vec::new(), - graph_url: graph_url.clone(), + indexer_url: indexer_url.clone(), reinit_interval: None, }) } diff --git a/crates/liquidity-sources/src/balancer_v2/graph_api.rs b/crates/liquidity-sources/src/balancer_v2/graph_api.rs deleted file mode 100644 index 149338ad47..0000000000 --- a/crates/liquidity-sources/src/balancer_v2/graph_api.rs +++ /dev/null @@ -1,481 +0,0 @@ -//! Module containing The Graph API client used for retrieving Balancer weighted -//! pools from the Balancer V2 subgraph. -//! -//! The pools retrieved from this client are used to prime the graph event store -//! to reduce start-up time. We do not use this in general for retrieving pools -//! as to: -//! - not rely on external services -//! - ensure that we are using the latest up-to-date pool data by using events -//! from the node - -use { - super::swap::fixed_point::Bfp, - crate::{json_map, subgraph::SubgraphClient}, - alloy::primitives::{Address, B256}, - anyhow::Result, - event_indexing::event_handler::MAX_REORG_BLOCK_COUNT, - reqwest::{Client, Url}, - serde::Deserialize, - serde_json::json, - serde_with::{DisplayFromStr, serde_as}, - std::collections::HashMap, -}; - -/// The page size when querying pools. -#[cfg(not(test))] -const QUERY_PAGE_SIZE: usize = 1000; -#[cfg(test)] -const QUERY_PAGE_SIZE: usize = 10; - -/// A client to the Balancer V2 subgraph. -/// -/// This client is not implemented to allow general GraphQL queries, but instead -/// implements high-level methods that perform GraphQL queries under the hood. -pub struct BalancerSubgraphClient(SubgraphClient); - -impl BalancerSubgraphClient { - /// Creates a new Balancer subgraph client with full subgraph URL. - pub fn from_subgraph_url(subgraph_url: &Url, client: Client) -> Result { - Ok(Self(SubgraphClient::try_new(subgraph_url.clone(), client)?)) - } - - /// Retrieves the list of registered pools from the subgraph. - pub async fn get_registered_pools(&self) -> Result { - use self::pools_query::*; - - let block_number = self.get_safe_block().await?; - - let mut pools = Vec::new(); - let mut last_id = B256::default(); - - // We do paging by last ID instead of using `skip`. This is the - // suggested approach to paging best performance: - // - loop { - let page = self - .0 - .query::( - QUERY, - Some(json_map! { - "block" => block_number, - "pageSize" => QUERY_PAGE_SIZE, - "lastId" => json!(last_id), - }), - ) - .await? - .pools; - let no_more_pages = page.len() != QUERY_PAGE_SIZE; - if let Some(last_pool) = page.last() { - last_id = last_pool.id; - } - - pools.extend(page); - - if no_more_pages { - break; - } - } - - Ok(RegisteredPools { - fetched_block_number: block_number, - pools, - }) - } - - /// Retrieves a recent block number for which it is safe to assume no - /// reorgs will happen. - async fn get_safe_block(&self) -> Result { - // Ideally we would want to use block hash here so that we can check - // that there indeed is no reorg. However, it does not seem possible to - // retrieve historic block hashes just from the subgraph (it always - // returns `null`). - Ok(self - .0 - .query::(block_number_query::QUERY, None) - .await? - .meta - .block - .number - .saturating_sub(MAX_REORG_BLOCK_COUNT)) - } -} - -/// Result of the registered stable pool query. -#[derive(Debug, Default, Eq, PartialEq)] -pub struct RegisteredPools { - /// The block number that the data was fetched, and for which the registered - /// weighted pools can be considered up to date. - pub fetched_block_number: u64, - /// The registered Pools - pub pools: Vec, -} - -impl RegisteredPools { - /// Creates an empty collection of registered pools for the specified block - /// number. - pub fn empty(fetched_block_number: u64) -> Self { - Self { - fetched_block_number, - ..Default::default() - } - } - - /// Groups registered pools by factory addresses. - pub fn group_by_factory(self) -> HashMap { - let fetched_block_number = self.fetched_block_number; - self.pools - .into_iter() - .fold(HashMap::new(), |mut grouped, pool| { - grouped - .entry(pool.factory) - .or_insert(RegisteredPools { - fetched_block_number, - ..Default::default() - }) - .pools - .push(pool); - grouped - }) - } -} - -/// Pool data from the Balancer V2 subgraph. -#[derive(Debug, Deserialize, Eq, PartialEq)] -#[serde(rename_all = "camelCase")] -pub struct PoolData { - pub pool_type: PoolType, - pub id: B256, - pub address: Address, - pub factory: Address, - pub swap_enabled: bool, - pub tokens: Vec, -} - -/// Supported pool kinds. -#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Hash)] -pub enum PoolType { - Stable, - Weighted, - LiquidityBootstrapping, - ComposableStable, -} - -/// Token data for pools. -#[serde_as] -#[derive(Debug, Deserialize, Eq, PartialEq)] -pub struct Token { - pub address: Address, - pub decimals: u8, - #[serde_as(as = "Option")] - #[serde(default)] - pub weight: Option, -} - -mod pools_query { - use {super::PoolData, serde::Deserialize}; - - pub const QUERY: &str = r#" - query Pools($block: Int, $pageSize: Int, $lastId: ID) { - pools( - block: { number: $block } - first: $pageSize - where: { - id_gt: $lastId - poolType_in: [ - "Stable", - "Weighted", - "LiquidityBootstrapping", - "ComposableStable", - ] - totalLiquidity_gt: "1" # 1$ value of tokens - } - ) { - poolType - id - address - factory - swapEnabled - tokens { - address - decimals - weight - } - } - } - "#; - - #[derive(Debug, Deserialize, Eq, PartialEq)] - pub struct Data { - pub pools: Vec, - } -} - -mod block_number_query { - use serde::Deserialize; - - pub const QUERY: &str = r#"{ - _meta { - block { number } - } - }"#; - - #[derive(Debug, Deserialize, Eq, PartialEq)] - pub struct Data { - #[serde(rename = "_meta")] - pub meta: Meta, - } - - #[derive(Debug, Deserialize, Eq, PartialEq)] - pub struct Meta { - pub block: Block, - } - - #[derive(Debug, Deserialize, Eq, PartialEq)] - pub struct Block { - pub number: u64, - } -} - -#[cfg(test)] -mod tests { - use { - super::*, - crate::balancer_v2::swap::fixed_point::Bfp, - alloy::primitives::U256, - maplit::hashmap, - }; - - #[test] - fn decode_pools_data() { - use pools_query::*; - - assert_eq!( - serde_json::from_value::(json!({ - "pools": [ - { - "poolType": "Weighted", - "address": "0x2222222222222222222222222222222222222222", - "id": "0x1111111111111111111111111111111111111111111111111111111111111111", - "factory": "0x5555555555555555555555555555555555555555", - "swapEnabled": true, - "tokens": [ - { - "address": "0x3333333333333333333333333333333333333333", - "decimals": 3, - "weight": "0.5" - }, - { - "address": "0x4444444444444444444444444444444444444444", - "decimals": 4, - "weight": "0.5" - }, - ], - }, - { - "poolType": "Stable", - "address": "0x2222222222222222222222222222222222222222", - "id": "0x1111111111111111111111111111111111111111111111111111111111111111", - "factory": "0x5555555555555555555555555555555555555555", - "swapEnabled": true, - "tokens": [ - { - "address": "0x3333333333333333333333333333333333333333", - "decimals": 3, - }, - { - "address": "0x4444444444444444444444444444444444444444", - "decimals": 4, - }, - ], - }, - { - "poolType": "LiquidityBootstrapping", - "address": "0x2222222222222222222222222222222222222222", - "id": "0x1111111111111111111111111111111111111111111111111111111111111111", - "factory": "0x5555555555555555555555555555555555555555", - "swapEnabled": true, - "tokens": [ - { - "address": "0x3333333333333333333333333333333333333333", - "decimals": 3, - "weight": "0.5" - }, - { - "address": "0x4444444444444444444444444444444444444444", - "decimals": 4, - "weight": "0.5" - }, - ], - }, - { - "poolType": "ComposableStable", - "address": "0x2222222222222222222222222222222222222222", - "id": "0x1111111111111111111111111111111111111111111111111111111111111111", - "factory": "0x5555555555555555555555555555555555555555", - "swapEnabled": true, - "tokens": [ - { - "address": "0x3333333333333333333333333333333333333333", - "decimals": 3, - }, - { - "address": "0x4444444444444444444444444444444444444444", - "decimals": 4, - }, - ], - }, - ], - })) - .unwrap(), - Data { - pools: vec![ - PoolData { - pool_type: PoolType::Weighted, - id: B256::repeat_byte(0x11), - address: Address::repeat_byte(0x22), - factory: Address::repeat_byte(0x55), - swap_enabled: true, - tokens: vec![ - Token { - address: Address::repeat_byte(0x33), - decimals: 3, - weight: Some(Bfp::from_wei(U256::from( - 500_000_000_000_000_000_u128 - ))), - }, - Token { - address: Address::repeat_byte(0x44), - decimals: 4, - weight: Some(Bfp::from_wei(U256::from( - 500_000_000_000_000_000_u128 - ))), - }, - ], - }, - PoolData { - pool_type: PoolType::Stable, - id: B256::repeat_byte(0x11), - address: Address::repeat_byte(0x22), - factory: Address::repeat_byte(0x55), - swap_enabled: true, - tokens: vec![ - Token { - address: Address::repeat_byte(0x33), - decimals: 3, - weight: None, - }, - Token { - address: Address::repeat_byte(0x44), - decimals: 4, - weight: None, - }, - ], - }, - PoolData { - pool_type: PoolType::LiquidityBootstrapping, - id: B256::repeat_byte(0x11), - address: Address::repeat_byte(0x22), - factory: Address::repeat_byte(0x55), - swap_enabled: true, - tokens: vec![ - Token { - address: Address::repeat_byte(0x33), - decimals: 3, - weight: Some(Bfp::from_wei(U256::from( - 500_000_000_000_000_000_u128 - ))), - }, - Token { - address: Address::repeat_byte(0x44), - decimals: 4, - weight: Some(Bfp::from_wei(U256::from( - 500_000_000_000_000_000_u128 - ))), - }, - ], - }, - PoolData { - pool_type: PoolType::ComposableStable, - id: B256::repeat_byte(0x11), - address: Address::repeat_byte(0x22), - factory: Address::repeat_byte(0x55), - swap_enabled: true, - tokens: vec![ - Token { - address: Address::repeat_byte(0x33), - decimals: 3, - weight: None, - }, - Token { - address: Address::repeat_byte(0x44), - decimals: 4, - weight: None, - }, - ], - }, - ], - } - ); - } - - #[test] - fn decode_block_number_data() { - use block_number_query::*; - - assert_eq!( - serde_json::from_value::(json!({ - "_meta": { - "block": { - "number": 42, - }, - }, - })) - .unwrap(), - Data { - meta: Meta { - block: Block { number: 42 } - } - } - ); - } - - #[test] - fn groups_pools_by_factory() { - let pool = |factory: Address, id: u8| PoolData { - id: B256::repeat_byte(id), - factory, - pool_type: PoolType::Weighted, - address: Default::default(), - swap_enabled: true, - tokens: Default::default(), - }; - - let registered_pools = RegisteredPools { - pools: vec![ - pool(Address::repeat_byte(1), 1), - pool(Address::repeat_byte(1), 2), - pool(Address::repeat_byte(2), 3), - ], - fetched_block_number: 42, - }; - - assert_eq!( - registered_pools.group_by_factory(), - hashmap! { - Address::repeat_byte(1) => RegisteredPools { - pools: vec![ - pool(Address::repeat_byte(1), 1), - pool(Address::repeat_byte(1), 2), - ], - fetched_block_number: 42, - }, - Address::repeat_byte(2) => RegisteredPools { - pools: vec![ - pool(Address::repeat_byte(2), 3), - ], - fetched_block_number: 42, - }, - } - ) - } -} diff --git a/crates/liquidity-sources/src/balancer_v2/mod.rs b/crates/liquidity-sources/src/balancer_v2/mod.rs index b9b4a9de5c..dcbdd8610b 100644 --- a/crates/liquidity-sources/src/balancer_v2/mod.rs +++ b/crates/liquidity-sources/src/balancer_v2/mod.rs @@ -30,20 +30,29 @@ //! respectively along with the current balances of each of the pool's tokens //! (aka the pool's "reserves"). //! -//! For this reason, only the `event_handler`, `pool_cache`, `pool_fetching` and -//! `swap` are declared as public, others merely contain internal logic -//! regarding how information is collected and stored. +//! For this reason, only the `models`, `pool_fetching`, `pool_indexer`, `pools` +//! and `swap` modules are declared as public; others merely contain internal +//! logic regarding how information is collected and stored. //! //! Once should think of `PoolStorage` as a type of Database for which one is //! not concerned with how it maintains itself. -mod graph_api; +pub mod models; pub mod pool_fetching; -mod pool_init; +pub mod pool_indexer; pub mod pools; pub mod swap; +use {anyhow::Result, models::RegisteredPools}; + pub use self::{ pool_fetching::{BalancerPoolFetcher, BalancerPoolFetching}, pools::{Pool, PoolKind}, }; + +/// Seeds the balancer pool registry at start-up with the currently registered +/// pools. Implemented by [`pool_indexer::BalancerIndexerClient`]. +#[async_trait::async_trait] +pub trait PoolInitializing: Send + Sync + 'static { + async fn initialize_pools(&self) -> Result; +} diff --git a/crates/liquidity-sources/src/balancer_v2/models.rs b/crates/liquidity-sources/src/balancer_v2/models.rs new file mode 100644 index 0000000000..868a62b126 --- /dev/null +++ b/crates/liquidity-sources/src/balancer_v2/models.rs @@ -0,0 +1,283 @@ +//! Static Balancer V2 pool data — the registry seed consumed by the aggregate +//! pool fetcher. Populated by the pool-indexer client. + +use { + super::swap::fixed_point::Bfp, + alloy::primitives::{Address, B256}, + serde::Deserialize, + serde_with::{DisplayFromStr, serde_as}, + std::collections::HashMap, +}; + +/// A set of registered pools, up to date as of `fetched_block_number`. +#[derive(Debug, Default, Eq, PartialEq)] +pub struct RegisteredPools { + /// The block the pools were fetched for and can be considered current at. + pub fetched_block_number: u64, + /// The registered pools. + pub pools: Vec, +} + +impl RegisteredPools { + /// Creates an empty collection for the specified block number. + pub fn empty(fetched_block_number: u64) -> Self { + Self { + fetched_block_number, + ..Default::default() + } + } + + /// Groups registered pools by factory address. + pub fn group_by_factory(self) -> HashMap { + let fetched_block_number = self.fetched_block_number; + self.pools + .into_iter() + .fold(HashMap::new(), |mut grouped, pool| { + grouped + .entry(pool.factory) + .or_insert(RegisteredPools { + fetched_block_number, + ..Default::default() + }) + .pools + .push(pool); + grouped + }) + } +} + +/// Static data for a Balancer V2 pool. +#[derive(Debug, Deserialize, Eq, PartialEq)] +#[serde(rename_all = "camelCase")] +pub struct PoolData { + pub pool_type: PoolType, + pub id: B256, + pub address: Address, + pub factory: Address, + pub swap_enabled: bool, + pub tokens: Vec, +} + +/// Supported pool kinds. +#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Hash)] +pub enum PoolType { + Stable, + Weighted, + LiquidityBootstrapping, + ComposableStable, +} + +/// Token data for pools. `weight` is present only for weighted pools. +#[serde_as] +#[derive(Debug, Deserialize, Eq, PartialEq)] +pub struct Token { + pub address: Address, + pub decimals: u8, + #[serde_as(as = "Option")] + #[serde(default)] + pub weight: Option, +} + +#[cfg(test)] +mod tests { + use {super::*, alloy::primitives::U256, maplit::hashmap, serde_json::json}; + + #[test] + fn decode_pools_data() { + assert_eq!( + serde_json::from_value::>(json!([ + { + "poolType": "Weighted", + "address": "0x2222222222222222222222222222222222222222", + "id": "0x1111111111111111111111111111111111111111111111111111111111111111", + "factory": "0x5555555555555555555555555555555555555555", + "swapEnabled": true, + "tokens": [ + { + "address": "0x3333333333333333333333333333333333333333", + "decimals": 3, + "weight": "0.5" + }, + { + "address": "0x4444444444444444444444444444444444444444", + "decimals": 4, + "weight": "0.5" + }, + ], + }, + { + "poolType": "Stable", + "address": "0x2222222222222222222222222222222222222222", + "id": "0x1111111111111111111111111111111111111111111111111111111111111111", + "factory": "0x5555555555555555555555555555555555555555", + "swapEnabled": true, + "tokens": [ + { + "address": "0x3333333333333333333333333333333333333333", + "decimals": 3, + }, + { + "address": "0x4444444444444444444444444444444444444444", + "decimals": 4, + }, + ], + }, + { + "poolType": "LiquidityBootstrapping", + "address": "0x2222222222222222222222222222222222222222", + "id": "0x1111111111111111111111111111111111111111111111111111111111111111", + "factory": "0x5555555555555555555555555555555555555555", + "swapEnabled": true, + "tokens": [ + { + "address": "0x3333333333333333333333333333333333333333", + "decimals": 3, + "weight": "0.5" + }, + { + "address": "0x4444444444444444444444444444444444444444", + "decimals": 4, + "weight": "0.5" + }, + ], + }, + { + "poolType": "ComposableStable", + "address": "0x2222222222222222222222222222222222222222", + "id": "0x1111111111111111111111111111111111111111111111111111111111111111", + "factory": "0x5555555555555555555555555555555555555555", + "swapEnabled": true, + "tokens": [ + { + "address": "0x3333333333333333333333333333333333333333", + "decimals": 3, + }, + { + "address": "0x4444444444444444444444444444444444444444", + "decimals": 4, + }, + ], + }, + ])) + .unwrap(), + vec![ + PoolData { + pool_type: PoolType::Weighted, + id: B256::repeat_byte(0x11), + address: Address::repeat_byte(0x22), + factory: Address::repeat_byte(0x55), + swap_enabled: true, + tokens: vec![ + Token { + address: Address::repeat_byte(0x33), + decimals: 3, + weight: Some(Bfp::from_wei(U256::from(500_000_000_000_000_000_u128))), + }, + Token { + address: Address::repeat_byte(0x44), + decimals: 4, + weight: Some(Bfp::from_wei(U256::from(500_000_000_000_000_000_u128))), + }, + ], + }, + PoolData { + pool_type: PoolType::Stable, + id: B256::repeat_byte(0x11), + address: Address::repeat_byte(0x22), + factory: Address::repeat_byte(0x55), + swap_enabled: true, + tokens: vec![ + Token { + address: Address::repeat_byte(0x33), + decimals: 3, + weight: None, + }, + Token { + address: Address::repeat_byte(0x44), + decimals: 4, + weight: None, + }, + ], + }, + PoolData { + pool_type: PoolType::LiquidityBootstrapping, + id: B256::repeat_byte(0x11), + address: Address::repeat_byte(0x22), + factory: Address::repeat_byte(0x55), + swap_enabled: true, + tokens: vec![ + Token { + address: Address::repeat_byte(0x33), + decimals: 3, + weight: Some(Bfp::from_wei(U256::from(500_000_000_000_000_000_u128))), + }, + Token { + address: Address::repeat_byte(0x44), + decimals: 4, + weight: Some(Bfp::from_wei(U256::from(500_000_000_000_000_000_u128))), + }, + ], + }, + PoolData { + pool_type: PoolType::ComposableStable, + id: B256::repeat_byte(0x11), + address: Address::repeat_byte(0x22), + factory: Address::repeat_byte(0x55), + swap_enabled: true, + tokens: vec![ + Token { + address: Address::repeat_byte(0x33), + decimals: 3, + weight: None, + }, + Token { + address: Address::repeat_byte(0x44), + decimals: 4, + weight: None, + }, + ], + }, + ] + ); + } + + #[test] + fn groups_pools_by_factory() { + let pool = |factory: Address, id: u8| PoolData { + id: B256::repeat_byte(id), + factory, + pool_type: PoolType::Weighted, + address: Default::default(), + swap_enabled: true, + tokens: Default::default(), + }; + + let registered_pools = RegisteredPools { + pools: vec![ + pool(Address::repeat_byte(1), 1), + pool(Address::repeat_byte(1), 2), + pool(Address::repeat_byte(2), 3), + ], + fetched_block_number: 42, + }; + + assert_eq!( + registered_pools.group_by_factory(), + hashmap! { + Address::repeat_byte(1) => RegisteredPools { + pools: vec![ + pool(Address::repeat_byte(1), 1), + pool(Address::repeat_byte(1), 2), + ], + fetched_block_number: 42, + }, + Address::repeat_byte(2) => RegisteredPools { + pools: vec![ + pool(Address::repeat_byte(2), 3), + ], + fetched_block_number: 42, + }, + } + ) + } +} diff --git a/crates/liquidity-sources/src/balancer_v2/pool_fetching/mod.rs b/crates/liquidity-sources/src/balancer_v2/pool_fetching/mod.rs index 1dfdaab5d2..d0ada41630 100644 --- a/crates/liquidity-sources/src/balancer_v2/pool_fetching/mod.rs +++ b/crates/liquidity-sources/src/balancer_v2/pool_fetching/mod.rs @@ -12,8 +12,8 @@ use { registry::Registry, }, super::{ - graph_api::{BalancerSubgraphClient, RegisteredPools}, - pool_init::PoolInitializing, + PoolInitializing, + models::RegisteredPools, pools::{ FactoryIndexing, Pool, @@ -50,7 +50,6 @@ use { ethrpc::{Web3, alloy::ProviderLabelingExt, block_stream::CurrentBlockWatcher}, event_indexing::block_retriever::BlockRetrieving, model::TokenPair, - reqwest::{Client, Url}, std::{ collections::{BTreeMap, HashSet}, sync::Arc, @@ -234,17 +233,15 @@ pub struct BalancerContracts { impl BalancerPoolFetcher { #[expect(clippy::too_many_arguments)] pub async fn new( - subgraph_url: &Url, + pool_initializer: Box, block_retriever: Arc, token_infos: Arc, config: CacheConfig, block_stream: CurrentBlockWatcher, - client: Client, web3: Web3, contracts: &BalancerContracts, deny_listed_pool_ids: Vec, ) -> Result { - let pool_initializer = BalancerSubgraphClient::from_subgraph_url(subgraph_url, client)?; let web3 = web3.labeled("balancerV2"); let fetcher = Arc::new(Cache::new( create_aggregate_pool_fetcher( @@ -316,7 +313,7 @@ impl BalancerPoolFetching for BalancerPoolFetcher { /// Creates an aggregate fetcher for all supported pool factories. async fn create_aggregate_pool_fetcher( web3: Web3, - pool_initializer: impl PoolInitializing, + pool_initializer: Box, block_retriever: Arc, token_infos: Arc, contracts: &BalancerContracts, diff --git a/crates/liquidity-sources/src/balancer_v2/pool_indexer.rs b/crates/liquidity-sources/src/balancer_v2/pool_indexer.rs new file mode 100644 index 0000000000..a37e69bb7f --- /dev/null +++ b/crates/liquidity-sources/src/balancer_v2/pool_indexer.rs @@ -0,0 +1,147 @@ +//! HTTP client for the Balancer V2 pools served by the pool-indexer service. +//! +//! Implements [`PoolInitializing`] so the pool-indexer can stand in for the +//! subgraph as the source that seeds the pool registry at start-up. The +//! indexer's `PoolData` response shares the subgraph's wire shape, so the +//! pages deserialize straight into [`PoolData`] and need no remapping. + +use { + super::{ + PoolInitializing, + models::{PoolData, RegisteredPools}, + }, + anyhow::{Context, Result}, + chain::Chain, + reqwest::{Client, Url}, + serde::Deserialize, + std::time::Duration, +}; + +/// Matches the server-side `MAX_PAGE_LIMIT`. +const LIST_PAGE_SIZE: u64 = 5000; + +/// Poll interval while the indexer is still bootstrapping (503). +const READY_POLL_INTERVAL: Duration = Duration::from_millis(500); + +/// Cap on the start-up wait for the indexer to serve its first checkpoint. +const READY_TIMEOUT: Duration = Duration::from_secs(60); + +pub struct BalancerIndexerClient { + base_url: Url, + http: Client, +} + +impl BalancerIndexerClient { + pub fn new(base_url: Url, chain: Chain, http: Client) -> Self { + let prefix = format!("api/v1/{}/balancer/v2/", chain.as_str()); + Self { + base_url: url_join(&base_url, &prefix), + http, + } + } + + fn path(&self, suffix: &str) -> Url { + url_join(&self.base_url, suffix) + } + + /// `GET /pools?limit=N[&after=cursor]`. `None` means the indexer replied + /// 503 — still bootstrapping, no checkpoint yet. + async fn fetch_pools_page( + &self, + limit: u64, + cursor: Option<&str>, + ) -> Result> { + let mut url = self.path("pools"); + url.query_pairs_mut() + .append_pair("limit", &limit.to_string()); + if let Some(c) = cursor { + url.query_pairs_mut().append_pair("after", c); + } + let resp = self.http.get(url).send().await.context("GET /pools")?; + if resp.status() == reqwest::StatusCode::SERVICE_UNAVAILABLE { + return Ok(None); + } + let page = resp + .error_for_status() + .context("/pools HTTP status")? + .json() + .await + .context("/pools body")?; + Ok(Some(page)) + } + + /// Polls `/pools` until the indexer is past bootstrap (not 503), bounded by + /// [`READY_TIMEOUT`]. Covers the serve container coming up moments after + /// the driver; a cold bootstrap is expected to have run before then. + async fn wait_until_ready(&self) -> Result<()> { + let deadline = std::time::Instant::now() + READY_TIMEOUT; + loop { + if self.fetch_pools_page(1, None).await?.is_some() { + return Ok(()); + } + if std::time::Instant::now() >= deadline { + anyhow::bail!("balancer pool-indexer not ready after {READY_TIMEOUT:?}"); + } + tracing::debug!("balancer pool-indexer not ready yet (503); waiting"); + tokio::time::sleep(READY_POLL_INTERVAL).await; + } + } +} + +#[async_trait::async_trait] +impl PoolInitializing for BalancerIndexerClient { + async fn initialize_pools(&self) -> Result { + self.wait_until_ready().await?; + + let mut cursor: Option = None; + let mut pools: Vec = Vec::new(); + let mut fetched_block_number: Option = None; + loop { + let page = self + .fetch_pools_page(LIST_PAGE_SIZE, cursor.as_deref()) + .await? + .context("balancer pool-indexer returned 503 after readiness check")?; + fetched_block_number.get_or_insert(page.block_number); + pools.extend(page.pools); + match page.next_cursor { + Some(c) => cursor = Some(c), + None => break, + } + } + + let registered_pools = RegisteredPools { + fetched_block_number: fetched_block_number + .context("balancer pool-indexer returned no pages")?, + pools, + }; + tracing::debug!( + block = %registered_pools.fetched_block_number, + pools = %registered_pools.pools.len(), + "initialized registered pools from indexer", + ); + Ok(registered_pools) + } +} + +/// Wire form of a `/pools` page. `pools` reuses the subgraph's [`PoolData`]. +#[derive(Deserialize)] +struct PoolsResponse { + block_number: u64, + pools: Vec, + #[serde(default)] + next_cursor: Option, +} + +/// Joins `path` onto `url` with exactly one slash between them. `Url::join` +/// drops a base's last path segment when it lacks a trailing slash (RFC 3986 +/// path resolution), and the operator-supplied indexer URL may omit one. +fn url_join(url: &Url, mut path: &str) -> Url { + let mut url = url.to_string(); + while url.ends_with('/') { + url.pop(); + } + while path.starts_with('/') { + path = &path[1..]; + } + Url::parse(&format!("{url}/{path}")).expect("constructed URL is valid") +} diff --git a/crates/liquidity-sources/src/balancer_v2/pool_init.rs b/crates/liquidity-sources/src/balancer_v2/pool_init.rs deleted file mode 100644 index 29e349c27c..0000000000 --- a/crates/liquidity-sources/src/balancer_v2/pool_init.rs +++ /dev/null @@ -1,27 +0,0 @@ -//! Balancer pool registry initialization. -//! -//! This module contains a component used to initialize Balancer pool registries -//! with existing data in order to reduce the "cold start" time of the service. - -use { - super::graph_api::{BalancerSubgraphClient, RegisteredPools}, - anyhow::Result, -}; - -#[async_trait::async_trait] -pub trait PoolInitializing: Send + Sync { - async fn initialize_pools(&self) -> Result; -} - -#[async_trait::async_trait] -impl PoolInitializing for BalancerSubgraphClient { - async fn initialize_pools(&self) -> Result { - let registered_pools = self.get_registered_pools().await?; - tracing::debug!( - block = %registered_pools.fetched_block_number, pools = %registered_pools.pools.len(), - "initialized registered pools", - ); - - Ok(registered_pools) - } -} diff --git a/crates/liquidity-sources/src/balancer_v2/pools/common.rs b/crates/liquidity-sources/src/balancer_v2/pools/common.rs index c9c6c831d3..e795915a6b 100644 --- a/crates/liquidity-sources/src/balancer_v2/pools/common.rs +++ b/crates/liquidity-sources/src/balancer_v2/pools/common.rs @@ -3,7 +3,7 @@ use { super::{FactoryIndexing, Pool, PoolIndexing as _, PoolStatus}, crate::balancer_v2::{ - graph_api::{PoolData, PoolType}, + models::{PoolData, PoolType}, swap::fixed_point::Bfp, }, alloy::{ @@ -357,7 +357,7 @@ mod tests { crate::{ balancer_v2::{ PoolKind, - graph_api::{PoolType, Token}, + models::{PoolType, Token}, pools::{MockFactoryIndexing, weighted}, }, bfp, diff --git a/crates/liquidity-sources/src/balancer_v2/pools/composable_stable.rs b/crates/liquidity-sources/src/balancer_v2/pools/composable_stable.rs index 5f5164fa37..5a10ece7de 100644 --- a/crates/liquidity-sources/src/balancer_v2/pools/composable_stable.rs +++ b/crates/liquidity-sources/src/balancer_v2/pools/composable_stable.rs @@ -3,7 +3,7 @@ use { super::{FactoryIndexing, PoolIndexing, common}, crate::balancer_v2::{ - graph_api::{PoolData, PoolType}, + models::{PoolData, PoolType}, swap::fixed_point::Bfp, }, alloy::eips::BlockId, @@ -112,7 +112,7 @@ impl FactoryIndexing for BalancerV2ComposableStablePoolFactory::Instance { mod tests { use { super::*, - crate::balancer_v2::graph_api::Token, + crate::balancer_v2::models::Token, alloy::primitives::{Address, B256}, }; diff --git a/crates/liquidity-sources/src/balancer_v2/pools/liquidity_bootstrapping.rs b/crates/liquidity-sources/src/balancer_v2/pools/liquidity_bootstrapping.rs index 535cfdfef2..b8a1b64f43 100644 --- a/crates/liquidity-sources/src/balancer_v2/pools/liquidity_bootstrapping.rs +++ b/crates/liquidity-sources/src/balancer_v2/pools/liquidity_bootstrapping.rs @@ -3,7 +3,7 @@ use { super::{FactoryIndexing, PoolIndexing, common}, crate::balancer_v2::{ - graph_api::{PoolData, PoolType}, + models::{PoolData, PoolType}, swap::fixed_point::Bfp, }, alloy::eips::BlockId, @@ -118,7 +118,7 @@ impl FactoryIndexing for BalancerV2LiquidityBootstrappingPoolFactory::Instance { mod tests { use { super::*, - crate::balancer_v2::graph_api::Token, + crate::balancer_v2::models::Token, alloy::primitives::{Address, B256}, }; diff --git a/crates/liquidity-sources/src/balancer_v2/pools/mod.rs b/crates/liquidity-sources/src/balancer_v2/pools/mod.rs index 2d3138bd85..d757999c12 100644 --- a/crates/liquidity-sources/src/balancer_v2/pools/mod.rs +++ b/crates/liquidity-sources/src/balancer_v2/pools/mod.rs @@ -14,7 +14,7 @@ pub mod stable; pub mod weighted; use { - super::graph_api::PoolData, + super::models::PoolData, alloy::{eips::BlockId, primitives::B256}, anyhow::Result, futures::future::BoxFuture, diff --git a/crates/liquidity-sources/src/balancer_v2/pools/stable.rs b/crates/liquidity-sources/src/balancer_v2/pools/stable.rs index 21ad5dd031..060fa309ce 100644 --- a/crates/liquidity-sources/src/balancer_v2/pools/stable.rs +++ b/crates/liquidity-sources/src/balancer_v2/pools/stable.rs @@ -3,7 +3,7 @@ use { super::{FactoryIndexing, PoolIndexing, common}, crate::balancer_v2::{ - graph_api::{PoolData, PoolType}, + models::{PoolData, PoolType}, swap::fixed_point::Bfp, }, alloy::{ @@ -127,7 +127,7 @@ impl FactoryIndexing for BalancerV2StablePoolFactoryV2::Instance { #[cfg(test)] mod tests { - use {super::*, crate::balancer_v2::graph_api::Token, alloy::primitives::B256}; + use {super::*, crate::balancer_v2::models::Token, alloy::primitives::B256}; #[test] fn errors_when_converting_wrong_pool_type() { diff --git a/crates/liquidity-sources/src/balancer_v2/pools/weighted.rs b/crates/liquidity-sources/src/balancer_v2/pools/weighted.rs index 2e654de6a5..ac10de96df 100644 --- a/crates/liquidity-sources/src/balancer_v2/pools/weighted.rs +++ b/crates/liquidity-sources/src/balancer_v2/pools/weighted.rs @@ -3,7 +3,7 @@ use { super::{FactoryIndexing, PoolIndexing, common}, crate::balancer_v2::{ - graph_api::{PoolData, PoolType}, + models::{PoolData, PoolType}, swap::fixed_point::Bfp, }, alloy::{eips::BlockId, primitives::Address}, @@ -145,7 +145,7 @@ fn pool_state( mod tests { use { super::*, - crate::{balancer_v2::graph_api::Token, bfp}, + crate::{balancer_v2::models::Token, bfp}, alloy::{ primitives::{Address, B256, U256}, providers::{Provider, ProviderBuilder, mock::Asserter}, diff --git a/crates/liquidity-sources/src/lib.rs b/crates/liquidity-sources/src/lib.rs index d6c106ff32..002dc40062 100644 --- a/crates/liquidity-sources/src/lib.rs +++ b/crates/liquidity-sources/src/lib.rs @@ -5,7 +5,6 @@ pub mod base_tokens; pub mod baseline_solvable; mod macros; pub mod recent_block_cache; -pub mod subgraph; pub mod swapr; pub mod uniswap_v2; pub mod uniswap_v3; diff --git a/crates/liquidity-sources/src/macros.rs b/crates/liquidity-sources/src/macros.rs index c9fbf51f3a..a0ceb8e365 100644 --- a/crates/liquidity-sources/src/macros.rs +++ b/crates/liquidity-sources/src/macros.rs @@ -6,15 +6,3 @@ macro_rules! bfp { .unwrap() }; } - -#[macro_export] -macro_rules! json_map { - ($($key:expr_2021 => $value:expr_2021),* $(,)?) => {{ - #[allow(unused_mut)] - let mut map = ::serde_json::Map::::new(); - $( - map.insert(($key).into(), ($value).into()); - )* - map - }} -} diff --git a/crates/liquidity-sources/src/subgraph.rs b/crates/liquidity-sources/src/subgraph.rs deleted file mode 100644 index 61fed055f6..0000000000 --- a/crates/liquidity-sources/src/subgraph.rs +++ /dev/null @@ -1,303 +0,0 @@ -//! A module implementing a client for querying subgraphs. - -use { - crate::json_map, - anyhow::{Result, bail}, - reqwest::{Client, Url}, - serde::{Deserialize, Serialize, de::DeserializeOwned}, - serde_json::{Map, Value, json}, - thiserror::Error, -}; - -pub const QUERY_PAGE_SIZE: usize = 1000; -const MAX_NUMBER_OF_ATTEMPTS_DEFAULT: usize = 10; - -/// A general client for querying subgraphs. -pub struct SubgraphClient { - client: Client, - subgraph_url: Url, - max_number_of_attempts: usize, -} - -pub trait ContainsId { - fn get_id(&self) -> String; -} - -#[derive(Debug, Deserialize, Eq, PartialEq)] -pub struct Data { - #[serde(alias = "pools", alias = "ticks")] - pub inner: Vec, -} - -impl SubgraphClient { - /// Creates a new subgraph client from the specified organization and name. - pub fn try_new(subgraph_url: Url, client: Client) -> Result { - Ok(Self { - client, - subgraph_url, - max_number_of_attempts: MAX_NUMBER_OF_ATTEMPTS_DEFAULT, - }) - } - - /// Performs the specified GraphQL query on the current subgraph. - pub async fn query(&self, query: &str, variables: Option>) -> Result - where - T: DeserializeOwned, - { - // for long lasting queries subgraph call might randomly fail - // introduced retry mechanism that should efficiently help since failures are - // quick and we need 1 or 2 retries to succeed. - let mut error: Option = None; - for _ in 0..self.max_number_of_attempts { - match self.query_without_retry(query, &variables).await { - Ok(result) => return Ok(result), - Err(err) => error = Some(err), - } - } - Err(anyhow::anyhow!(format!( - "failed to execute query on subgraph: {}", - error.unwrap() - ))) - } - - pub async fn query_without_retry( - &self, - query: &str, - variables: &Option>, - ) -> Result - where - T: DeserializeOwned, - { - match self - .client - .post(self.subgraph_url.clone()) - .json(&Query { - query, - variables: variables.clone(), - }) - .send() - .await? - .json::>() - .await? - .into_result() - { - Ok(result) => Ok(result), - Err(err) => { - tracing::warn!("failed to query subgraph: {}", err); - Err(anyhow::anyhow!(format!( - "failed to execute query on subgraph: {}", - err - ))) - } - } - } - - /// Performs the specified GraphQL query on the current subgraph. - /// This function should be called for queries that return very - /// long(paginated) result. - pub async fn paginated_query( - &self, - query: &str, - mut variables: Map, - ) -> Result> - where - T: ContainsId + DeserializeOwned, - { - let mut result = Vec::new(); - - // We do paging by last ID instead of using `skip`. This is the - // suggested approach to paging best performance: - // - variables.extend(json_map! { - "pageSize" => QUERY_PAGE_SIZE, - "lastId" => json!(String::default()), - }); - loop { - let page = self - .query::>(query, Some(variables.clone())) - .await? - .inner; - let no_more_pages = page.len() != QUERY_PAGE_SIZE; - if let Some(last_elem) = page.last() { - variables.insert("lastId".to_string(), json!(last_elem.get_id())); - } - - result.extend(page); - - if no_more_pages { - break; - } - } - - Ok(result) - } -} - -/// A GraphQL query. -#[derive(Serialize)] -struct Query<'a> { - query: &'a str, - variables: Option>, -} - -/// A GraphQL query response. -/// -/// This type gets converted into a Rust `Result` type, while handling invalid -/// responses (with missing data and errors). -#[derive(Debug, Deserialize)] -struct QueryResponse { - #[serde(default = "empty_data")] - data: Option, - #[serde(default)] - errors: Option>, -} - -impl QueryResponse { - fn into_result(self) -> Result { - match self { - Self { - data: Some(data), - errors: None, - } => Ok(data), - Self { - errors: Some(errors), - data: None, - } if !errors.is_empty() => { - // Make sure to log additional errors if there are more than - // one, and just bubble up the first error. - for error in &errors[1..] { - tracing::warn!("additional GraphQL error: {}", error.message); - } - bail!("{}", errors[0]) - } - _ => bail!("invalid GraphQL response"), - } - } -} - -#[derive(Debug, Deserialize, Error)] -#[error("{}", .message)] -struct QueryError { - message: String, -} - -/// Function to work around the fact that `#[serde(default)]` on an `Option` -/// requires `T: Default`. -fn empty_data() -> Option { - None -} - -#[cfg(test)] -mod tests { - use { - super::*, - serde_json::{Value, json}, - }; - - #[test] - fn serialize_query() { - assert_eq!( - serde_json::to_value(Query { - query: r#"foo { - }"#, - variables: Some(json_map! { - "foo" => "bar", - "baz" => 42, - "thing" => false, - }), - }) - .unwrap(), - json!({ - "query": "foo {\n }", - "variables": { - "foo": "bar", - "baz": 42, - "thing": false, - }, - }), - ); - } - - fn response_from_json(value: Value) -> Result - where - T: DeserializeOwned, - { - serde_json::from_value::>(value) - .unwrap() - .into_result() - } - - #[test] - fn deserialize_successful_response() { - assert!(response_from_json::(json!({ "data": true })).unwrap()); - } - - #[test] - fn deserialize_error_response() { - assert_eq!( - response_from_json::(json!({ - "data": null, - "errors": [{"message": "foo"}], - })) - .unwrap_err() - .to_string(), - "foo", - ); - assert_eq!( - response_from_json::(json!({ - "errors": [{"message": "bar"}], - })) - .unwrap_err() - .to_string(), - "bar", - ); - } - - #[test] - fn deserialize_multi_error_response() { - assert_eq!( - response_from_json::(json!({ - "data": null, - "errors": [ - {"message": "foo"}, - {"message": "bar"}, - ], - })) - .unwrap_err() - .to_string(), - "foo", - ); - } - - #[test] - fn deserialize_invalid_response() { - assert!( - response_from_json::(json!({ - "data": null, - "errors": null, - })) - .is_err() - ); - assert!( - response_from_json::(json!({ - "data": null, - "errors": [], - })) - .is_err() - ); - assert!( - response_from_json::(json!({ - "data": true, - "errors": [], - })) - .is_err() - ); - assert!( - response_from_json::(json!({ - "data": true, - "errors": [{"message":"bad"}], - })) - .is_err() - ); - } -}