Skip to content

Commit c628569

Browse files
committed
Split pool-indexer bootstrap from serve via --bootstrap-only flag
1 parent 40d2b09 commit c628569

5 files changed

Lines changed: 171 additions & 26 deletions

File tree

crates/e2e/tests/e2e/pool_indexer.rs

Lines changed: 51 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -225,10 +225,8 @@ async fn seed_checkpoint(db: &PgPool, factory: Address, block: u64) {
225225
.unwrap();
226226
}
227227

228-
/// Spawns the pool-indexer task and waits for its `/health` endpoint to come
229-
/// up.
230-
async fn spawn_pool_indexer(factory: Address, metrics_port: u16) -> tokio::task::JoinHandle<()> {
231-
let config = Configuration {
228+
fn pool_indexer_config(factory: Address, metrics_port: u16) -> Configuration {
229+
Configuration {
232230
database: DatabaseConfig {
233231
url: LOCAL_DB_URL.parse().unwrap(),
234232
max_connections: NonZeroU32::new(5).unwrap(),
@@ -252,7 +250,13 @@ async fn spawn_pool_indexer(factory: Address, metrics_port: u16) -> tokio::task:
252250
metrics: MetricsConfig {
253251
bind_address: SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::LOCALHOST, metrics_port)),
254252
},
255-
};
253+
}
254+
}
255+
256+
/// Spawns the pool-indexer task and waits for its `/health` endpoint to come
257+
/// up.
258+
async fn spawn_pool_indexer(factory: Address, metrics_port: u16) -> tokio::task::JoinHandle<()> {
259+
let config = pool_indexer_config(factory, metrics_port);
256260
let handle = tokio::task::spawn(pool_indexer::run(config));
257261
wait_for_condition(TIMEOUT, || async {
258262
reqwest::get(format!("{POOL_INDEXER_HOST}/health"))
@@ -682,3 +686,45 @@ async fn pagination(web3: Web3) {
682686
})
683687
.await;
684688
}
689+
690+
#[tokio::test]
691+
#[ignore]
692+
async fn local_node_pool_indexer_bootstrap_idempotent() {
693+
run_test(bootstrap_idempotent).await;
694+
}
695+
696+
/// `--bootstrap-only` on an already-seeded DB must be a fast no-op: detect the
697+
/// existing checkpoint, skip the (here unreachable) subgraph seeder, and return
698+
/// without binding any ports — mirroring a re-run of the bootstrap
699+
/// initContainer on a pod restart.
700+
async fn bootstrap_idempotent(web3: Web3) {
701+
let db = PgPool::connect(LOCAL_DB_URL).await.unwrap();
702+
clear_pool_indexer_tables(&db).await;
703+
704+
// A pre-seeded checkpoint marks the DB as already bootstrapped. No on-chain
705+
// factory is needed: bootstrap reads the checkpoint and returns before any
706+
// seeding or catch-up. The RPC is only used for the chain_id sanity check.
707+
let factory = Address::repeat_byte(0x11);
708+
let head = web3.provider.get_block_number().await.unwrap();
709+
seed_checkpoint(&db, factory, head).await;
710+
711+
tokio::time::timeout(
712+
TIMEOUT,
713+
pool_indexer::bootstrap(pool_indexer_config(factory, POOL_INDEXER_METRICS_PORT)),
714+
)
715+
.await
716+
.expect("bootstrap-only did not exit on an already-seeded DB");
717+
718+
let checkpoint: i64 = sqlx::query_scalar(
719+
"SELECT block_number FROM pool_indexer_checkpoints WHERE contract_address = $1",
720+
)
721+
.bind(factory.as_slice())
722+
.fetch_one(&db)
723+
.await
724+
.unwrap();
725+
assert_eq!(
726+
checkpoint,
727+
head.cast_signed(),
728+
"bootstrap mutated an already-seeded checkpoint"
729+
);
730+
}

crates/pool-indexer/README.md

Lines changed: 43 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,38 @@ at a fixed block, catches up to the chain tip via RPC events, then stays
99
live by polling new blocks. Drivers consume it via the `pool-indexer-url`
1010
field in their Uniswap V3 liquidity config.
1111

12+
## Bootstrap and serve
13+
14+
Startup has two phases:
15+
16+
- **Bootstrap** — initial subgraph seed plus catch-up to the finalized head.
17+
One-time and slow (minutes on a large chain).
18+
- **Serve** — live block polling and the HTTP API. No long startup cost.
19+
20+
`pool-indexer --config <toml>` runs both in one process: it bootstraps when the
21+
DB has no checkpoint, then serves. This is the single-container deployment.
22+
23+
`pool-indexer --bootstrap-only --config <toml>` runs only the bootstrap phase
24+
and then exits 0, binding no HTTP ports. It is **idempotent**: on a DB that
25+
already has a checkpoint it skips the seed and catch-up entirely (never touching
26+
the subgraph) and returns immediately, so re-running it — for example a
27+
restarted bootstrap initContainer — is a fast, safe no-op.
28+
29+
This lets K8s run bootstrap as an initContainer and apply a tight startupProbe
30+
to the serve container, which finds the checkpoint already present and flips
31+
`/startup` ready almost immediately:
32+
33+
```yaml
34+
initContainers:
35+
- name: init-db # flyway, schema from the indexer's own location
36+
command: ["flyway", "-locations=filesystem:/flyway/sql-pool-indexer", "migrate"]
37+
- name: bootstrap # one-time seed + catch-up; idempotent on restart
38+
command: ["pool-indexer", "--bootstrap-only", "--config", "/etc/config/pool-indexer.toml"]
39+
containers:
40+
- name: pool-indexer # serve; DB already seeded, so startup is fast
41+
command: ["pool-indexer", "--config", "/etc/config/pool-indexer.toml"]
42+
```
43+
1244
## Running locally
1345
1446
Create `crates/pool-indexer/config.local.toml` first (schema = the
@@ -17,12 +49,19 @@ Create `crates/pool-indexer/config.local.toml` first (schema = the
1749
sections. String fields accept `%ENV_VAR` so secrets can come from the
1850
environment instead of being written into the file.
1951

20-
Then, from the repository root, reset the local stack and start the indexer:
52+
The indexer uses its own database, migrated from `database/sql-pool-indexer`
53+
(separate from the shared autopilot/orderbook set in `database/sql`). From the
54+
repository root:
2155

2256
```bash
23-
# wipes the local DB — dev machines only
24-
docker compose down --volumes
2557
docker compose up -d db
26-
docker compose run --rm migrations
58+
# apply the indexer schema (database/sql-pool-indexer) to the indexer's DB
59+
psql "$POOL_INDEXER_DB_URL" -f database/sql-pool-indexer/V110__pool_indexer_uniswap_v3.sql
60+
61+
# one process for both phases:
62+
cargo run --release -p pool-indexer -- --config crates/pool-indexer/config.local.toml
63+
64+
# or split bootstrap from serve, as the K8s deployment does:
65+
cargo run --release -p pool-indexer -- --bootstrap-only --config crates/pool-indexer/config.local.toml
2766
cargo run --release -p pool-indexer -- --config crates/pool-indexer/config.local.toml
2867
```

crates/pool-indexer/src/arguments.rs

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,32 @@ pub struct Arguments {
77
#[clap(long, env)]
88
pub config: PathBuf,
99

10+
/// Run only the bootstrap phase (initial seed + catch-up to the finalized
11+
/// head), then exit; bind no HTTP ports. Idempotent: a fast no-op when the
12+
/// DB is already seeded. Lets K8s run bootstrap as an initContainer and
13+
/// apply tight startup probes to the serve container.
14+
#[clap(long, env)]
15+
pub bootstrap_only: bool,
16+
1017
#[clap(flatten)]
1118
pub logging: LoggingArguments,
1219
}
20+
21+
#[cfg(test)]
22+
mod tests {
23+
use {super::*, clap::Parser};
24+
25+
#[test]
26+
fn bootstrap_only_flag_parses() {
27+
let serve = Arguments::parse_from(["pool-indexer", "--config", "/tmp/c.toml"]);
28+
assert!(!serve.bootstrap_only);
29+
30+
let bootstrap = Arguments::parse_from([
31+
"pool-indexer",
32+
"--config",
33+
"/tmp/c.toml",
34+
"--bootstrap-only",
35+
]);
36+
assert!(bootstrap.bootstrap_only);
37+
}
38+
}

crates/pool-indexer/src/lib.rs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
pub mod config;
2-
pub use run::{run, start};
2+
pub use run::{bootstrap, run, start};
33

44
mod api;
55
mod arguments;

crates/pool-indexer/src/run.rs

Lines changed: 50 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -21,8 +21,38 @@ pub async fn start(args: impl Iterator<Item = String>) {
2121
initialize_observability(&args);
2222
observe::metrics::setup_registry(None, None);
2323
let config = Configuration::from_path(&args.config).expect("failed to load configuration");
24-
tracing::info!("pool-indexer starting");
25-
run(config).await;
24+
if args.bootstrap_only {
25+
tracing::info!("pool-indexer bootstrap-only starting");
26+
bootstrap(config).await;
27+
tracing::info!("pool-indexer bootstrap complete, exiting");
28+
} else {
29+
tracing::info!("pool-indexer starting");
30+
run(config).await;
31+
}
32+
}
33+
34+
/// Runs the bootstrap phase (seed + catch-up to the finalized head) for every
35+
/// factory, then returns. Binds no HTTP ports — this is migration-style work
36+
/// meant to run as a K8s initContainer ahead of the serve container.
37+
///
38+
/// Idempotent: each factory with an existing checkpoint is skipped (see
39+
/// [`bootstrap_factory`]), so re-running on an already-seeded DB is a fast
40+
/// no-op that never touches the subgraph. On return, a subsequent `run` finds
41+
/// the checkpoints present and flips `/startup` ready almost immediately.
42+
pub async fn bootstrap(config: Configuration) {
43+
let db = connect_db(&config).await;
44+
let network = config.network;
45+
let provider = build_provider_checked(&network).await;
46+
let network = Arc::new(network);
47+
48+
for factory in network.factories.iter().copied() {
49+
let indexer = UniswapV3Indexer::new(
50+
provider.clone(),
51+
db.clone(),
52+
&network.indexer_config(factory.address),
53+
);
54+
bootstrap_factory(&db, &indexer, &network, &factory).await;
55+
}
2656
}
2757

2858
pub async fn run(config: Configuration) {
@@ -125,20 +155,7 @@ async fn run_network_indexer(db: PgPool, network: NetworkConfig, barrier: Arc<St
125155
"starting network indexer",
126156
);
127157

128-
let provider = build_provider(&network);
129-
130-
// Catch misconfigured RPC-vs-network pairings (e.g. chain_id=1 pointed
131-
// at an Arbitrum node) before we index the wrong chain into the DB.
132-
let actual_chain_id = provider
133-
.get_chain_id()
134-
.await
135-
.expect("failed to fetch chain_id from RPC");
136-
assert_eq!(
137-
actual_chain_id, network.chain_id,
138-
"chain_id mismatch for network {}: config says {}, RPC reports {}",
139-
network.name, network.chain_id, actual_chain_id,
140-
);
141-
158+
let provider = build_provider_checked(&network).await;
142159
let network = Arc::new(network);
143160

144161
// One task per factory. Provider + DB pool are shared; checkpoints are
@@ -255,6 +272,23 @@ fn build_provider(network: &NetworkConfig) -> AlloyProvider {
255272
.clone()
256273
}
257274

275+
/// Builds the RPC provider and asserts the node's chain_id matches config.
276+
/// Catches misconfigured RPC-vs-network pairings (e.g. chain_id=1 pointed at
277+
/// an Arbitrum node) before we index the wrong chain into the DB.
278+
async fn build_provider_checked(network: &NetworkConfig) -> AlloyProvider {
279+
let provider = build_provider(network);
280+
let actual_chain_id = provider
281+
.get_chain_id()
282+
.await
283+
.expect("failed to fetch chain_id from RPC");
284+
assert_eq!(
285+
actual_chain_id, network.chain_id,
286+
"chain_id mismatch for network {}: config says {}, RPC reports {}",
287+
network.name, network.chain_id, actual_chain_id,
288+
);
289+
provider
290+
}
291+
258292
async fn connect_db(config: &Configuration) -> sqlx::PgPool {
259293
PgPoolOptions::new()
260294
.max_connections(config.database.max_connections.get())

0 commit comments

Comments
 (0)