feat(bens): add read replica (#1660)

* refactor(launcher): split postgres connect and SeaORM migrations

* fix(launcher): propagate read replica connect failure from maybe_connect_replica_repo

* feat(bens): add optional replica_database settings and env examples

* refactor(bens): hold Arc<ReadWriteRepo> in SubgraphReader

Drop ReadPoolSource; resolve read/write pools via read_db()/main_db() each call.
Add blockscout-service-launcher dep; adjust tests, benchmark, and server wiring.

* fix(bens): align subgraph error diagnostics with read vs write pool

* fix tests

* set blockscout-launcher-server version to branch

* fix tests

Author: sevenzing

blockscout-ens/Cargo.lock

@@ -17,7 +17,7 @@ anyhow = { version = "1" }
 async-trait = { version = "0.1" }
 bigdecimal = { version = "0.4" }
 blockscout-display-bytes = { version = "1.0" }
-blockscout-service-launcher = { version = "0.21.0" }
+blockscout-service-launcher = { git = "https://github.com/blockscout/blockscout-rs", branch = "ll/bens/read-replica" }
 cached = { version = "0.54.0", default-features = false }
 chrono = "0.4"
 config = { version = "0.14" }
@@ -41,7 +41,8 @@ sea-query = { version = "0.32" }
 serde = { version = "1" }
 serde_json = { version = "1", default-features = false }
 serde_with = { version = "3.11" }
-sqlx = { version = "0.8", default-features = false }
+# Pin to the same sqlx minor as sea-orm / blockscout-service-launcher to avoid duplicate sqlx.
+sqlx = { version = "0.8.6", default-features = false }
 tera = { version = "1.19" }
 thiserror = { version = "2" }
 tokio = { version = "1.23", default-features = false }

blockscout-ens/Cargo.toml

@@ -40,6 +40,9 @@ Service is **multi-chain**, meaning that only one instance of `graph-node`, `pos
 | `BENS__TRACING__ENABLED`                         |                          |                                                               | `true`         |
 | `BENS__TRACING__FORMAT`                          |                          |                                                               | `default`      |
 | `BENS__SUBGRAPHS_READER__REFRESH_CACHE_DISABLED` |                          |                                                               | `false`        |
+| `BENS__REPLICA_DATABASE__CONNECT__URL`           | true                     | e.g. `postgresql://postgres:postgres@localhost:5433/postgres` |                |
+| `BENS__REPLICA_DATABASE__HEALTH_CHECK_INTERVAL`  |                          |                                                               | `10`           |
+| `BENS__REPLICA_DATABASE__MAX_LAG`                |                          |                                                               | `300`          |
 
 [anchor]: <> (anchors.envs.end.envs_main)
 

blockscout-ens/README.md

@@ -11,6 +11,7 @@ alloy-ccip-read = { workspace = true }
 anyhow = { workspace = true }
 async-trait = { workspace = true }
 bigdecimal = { workspace = true }
+blockscout-service-launcher = { workspace = true, features = ["database-1"] }
 cached = { workspace = true, features = [
   "proc_macro",
   "tokio",
@@ -52,6 +53,7 @@ ens-normalize-rs = { workspace = true }
 pretty_assertions = { workspace = true }
 rstest = { workspace = true }
 tracing-subscriber = { workspace = true, features = ["env-filter"] }
+wiremock = { workspace = true }
 
 [[example]]
 name = "resolve_benchmark"

blockscout-ens/bens-logic/Cargo.toml

@@ -5,8 +5,10 @@ use bens_logic::{
     protocols::{Network, ProtocolInfo, Tld},
     subgraph::{BatchResolveAddressNamesInput, SubgraphReader},
 };
+use blockscout_service_launcher::database::{
+    DatabaseConnectSettings, DatabaseSettings, ReadWriteRepo,
+};
 use nonempty::nonempty;
-use sqlx::postgres::PgPoolOptions;
 use std::{collections::HashMap, sync::Arc, time::Instant};
 use tracing_subscriber::fmt::format::FmtSpan;
 
@@ -23,12 +25,13 @@ async fn main() -> Result<(), anyhow::Error> {
         .init();
 
     let url = std::env::var("DATABASE_URL").expect("no database url");
-    let pool = Arc::new(
-        PgPoolOptions::new()
-            .max_connections(40)
-            .connect(&url)
-            .await?,
-    );
+    let settings = DatabaseSettings {
+        connect: DatabaseConnectSettings::Url(url),
+        connect_options: Default::default(),
+        create_database: false,
+        run_migrations: false,
+    };
+    let db = Arc::new(ReadWriteRepo::new_no_migrations(&settings, None).await?);
     let eth_client = BlockscoutClient::new("https://eth.blockscout.com".parse().unwrap(), 5, 30);
     let rootstock_client =
         BlockscoutClient::new("https://rootstock.blockscout.com".parse().unwrap(), 5, 30);
@@ -72,7 +75,7 @@ async fn main() -> Result<(), anyhow::Error> {
             },
         ),
     ]);
-    let reader = SubgraphReader::initialize(pool.clone(), networks, protocol_infos).await?;
+    let reader = SubgraphReader::initialize(db, networks, protocol_infos).await?;
 
     let addresses = vec![
         "0x0292f204513eeafe8c032ffc4cb4c7e10eca908c",

blockscout-ens/bens-logic/examples/resolve_benchmark.rs

@@ -5,7 +5,7 @@ mod metrics;
 pub mod migrations;
 pub mod protocols;
 pub mod subgraph;
-#[cfg(feature = "test-utils")]
+#[cfg(any(test, feature = "test-utils"))]
 pub mod test_utils;
 
 pub use protocols::hash_name::hex;