From 2fbeb14c5e853c88223b90afacaf26fb7df21aba Mon Sep 17 00:00:00 2001 From: mrv777 Date: Sat, 16 May 2026 07:38:42 -0500 Subject: [PATCH 1/2] Add round summary endpoints --- src/subcommand/server.rs | 3 ++ src/subcommand/server/database.rs | 31 +++++++++++++- src/subcommand/server/rounds.rs | 43 ++++++++++++++++++++ tests/server_with_db.rs | 67 +++++++++++++++++++++++++++++++ 4 files changed, 143 insertions(+), 1 deletion(-) diff --git a/src/subcommand/server.rs b/src/subcommand/server.rs index 2d27006c..0857c489 100644 --- a/src/subcommand/server.rs +++ b/src/subcommand/server.rs @@ -152,7 +152,9 @@ impl Modify for SecurityAddon { // Round endpoints rounds::rounds, rounds::round_current, + rounds::round_current_summary, rounds::round, + rounds::round_summary, rounds::participants, // Sync endpoints sync_routes::sync_batch, @@ -181,6 +183,7 @@ impl Modify for SecurityAddon { // Round schemas rounds::Round, rounds::RoundParticipant, + rounds::RoundSummary, // Server schemas Payment, SatSplit, diff --git a/src/subcommand/server/database.rs b/src/subcommand/server/database.rs index 7cb24b65..27274aa7 100644 --- a/src/subcommand/server/database.rs +++ b/src/subcommand/server/database.rs @@ -1,6 +1,6 @@ use { super::*, - rounds::{Round, RoundParticipant}, + rounds::{Round, RoundParticipant, RoundSummary}, }; #[derive(sqlx::FromRow, Deserialize, Serialize, Debug, Clone, PartialEq, ToSchema)] @@ -807,6 +807,35 @@ impl Database { } } + pub(crate) async fn get_round_summary(&self, blockheight: Option) -> Result { + sqlx::query_as::<_, RoundSummary>( + " + WITH previous_block AS ( + SELECT COALESCE(MAX(blockheight), 0) AS previous_blockheight + FROM blocks + WHERE ($1::INTEGER IS NULL OR blockheight < $1) + ), + round_total AS ( + SELECT COALESCE(SUM(rs.diff), 0.0)::FLOAT8 AS total_diff + FROM remote_shares rs, previous_block pb + WHERE rs.blockheight > pb.previous_blockheight + AND ($1::INTEGER IS NULL OR rs.blockheight <= $1) + AND rs.reject_reason IS NULL + ) + SELECT + $1::INTEGER AS blockheight, + pb.previous_blockheight, + rt.total_diff + FROM previous_block pb + CROSS JOIN round_total rt + ", + ) + .bind(blockheight) + .fetch_one(&self.pool) + .await + .map_err(|err| anyhow!(err)) + } + async fn get_round_participation_history( &self, blockheight: i32, diff --git a/src/subcommand/server/rounds.rs b/src/subcommand/server/rounds.rs index c7a424a4..12b4beb1 100644 --- a/src/subcommand/server/rounds.rs +++ b/src/subcommand/server/rounds.rs @@ -16,11 +16,20 @@ pub(crate) struct RoundParticipant { pub(crate) top_diff: f64, } +#[derive(sqlx::FromRow, Serialize, Deserialize, Debug, Clone, ToSchema)] +pub(crate) struct RoundSummary { + pub(crate) blockheight: Option, + pub(crate) previous_blockheight: i32, + pub(crate) total_diff: f64, +} + pub(crate) fn rounds_router(config: Arc, database: Database) -> axum::Router { let mut router = axum::Router::new() .route("/rounds", get(rounds)) .route("/rounds/current", get(round_current)) + .route("/rounds/current/summary", get(round_current_summary)) .route("/rounds/{blockheight}", get(round)) + .route("/rounds/{blockheight}/summary", get(round_summary)) .route("/participants/{blockheight}", get(participants)); if let Some(token) = config.api_token() { @@ -58,6 +67,21 @@ pub(crate) async fn round_current( Ok(Json(database.get_round_participation(None).await?).into_response()) } +#[utoipa::path( + get, + path = "/rounds/current/summary", + security(("api_token" = [])), + responses( + (status = 200, description = "Current in-progress round summary", body = RoundSummary), + ), + tag = "rounds" +)] +pub(crate) async fn round_current_summary( + Extension(database): Extension, +) -> ServerResult { + Ok(Json(database.get_round_summary(None).await?).into_response()) +} + #[utoipa::path( get, path = "/rounds/{blockheight}", @@ -77,6 +101,25 @@ pub(crate) async fn round( Ok(Json(database.get_round_participation(Some(blockheight)).await?).into_response()) } +#[utoipa::path( + get, + path = "/rounds/{blockheight}/summary", + security(("api_token" = [])), + params( + ("blockheight" = i32, Path, description = "Block height of the found block ending this round") + ), + responses( + (status = 200, description = "Round summary", body = RoundSummary), + ), + tag = "rounds" +)] +pub(crate) async fn round_summary( + Path(blockheight): Path, + Extension(database): Extension, +) -> ServerResult { + Ok(Json(database.get_round_summary(Some(blockheight)).await?).into_response()) +} + #[utoipa::path( get, path = "/participants/{blockheight}", diff --git a/tests/server_with_db.rs b/tests/server_with_db.rs index 59db3e51..977b602d 100644 --- a/tests/server_with_db.rs +++ b/tests/server_with_db.rs @@ -13,6 +13,13 @@ struct RoundParticipant { top_diff: f64, } +#[derive(Deserialize, Debug)] +struct RoundSummary { + blockheight: Option, + previous_blockheight: i32, + total_diff: f64, +} + async fn insert_test_shares_for_round( database_url: String, users: Vec<(&str, f64)>, @@ -1100,6 +1107,36 @@ async fn test_round_first_includes_all_prior_shares() { assert_eq!(participants[0].top_diff, 300.0); } +#[tokio::test] +async fn test_round_summary_totals_shares_since_previous_round() { + let server = TestServer::spawn_with_db().await; + let db_url = server.database_url().unwrap(); + setup_test_schema(db_url.clone()).await.unwrap(); + + insert_test_block(db_url.clone(), 5).await.unwrap(); + insert_test_block(db_url.clone(), 10).await.unwrap(); + + insert_test_shares_for_round(db_url.clone(), vec![("old", 9000.0)], 4, 100) + .await + .unwrap(); + insert_test_shares_for_round( + db_url.clone(), + vec![("foo", 1000.0), ("bar", 2000.0)], + 7, + 200, + ) + .await + .unwrap(); + insert_test_shares_for_round(db_url.clone(), vec![("baz", 3000.0)], 10, 300) + .await + .unwrap(); + + let summary: RoundSummary = server.get_json_async("/rounds/10/summary").await; + assert_eq!(summary.blockheight, Some(10)); + assert_eq!(summary.previous_blockheight, 5); + assert_eq!(summary.total_diff, 6000.0); +} + #[tokio::test] async fn test_round_empty() { let server = TestServer::spawn_with_db().await; @@ -1161,6 +1198,32 @@ async fn test_current_round_no_blocks_found() { assert_eq!(participants.len(), 2); } +#[tokio::test] +async fn test_current_round_summary_totals_since_latest_block() { + let server = TestServer::spawn_with_db().await; + let db_url = server.database_url().unwrap(); + setup_test_schema(db_url.clone()).await.unwrap(); + + insert_test_block(db_url.clone(), 5).await.unwrap(); + + insert_test_shares_for_round(db_url.clone(), vec![("old", 1000.0)], 3, 100) + .await + .unwrap(); + insert_test_shares_for_round( + db_url.clone(), + vec![("foo", 2000.0), ("bar", 500.0)], + 7, + 200, + ) + .await + .unwrap(); + + let summary: RoundSummary = server.get_json_async("/rounds/current/summary").await; + assert_eq!(summary.blockheight, None); + assert_eq!(summary.previous_blockheight, 5); + assert_eq!(summary.total_diff, 2500.0); +} + #[tokio::test] async fn test_participants_for_blockheight() { let server = TestServer::spawn_with_db().await; @@ -1203,6 +1266,7 @@ async fn test_participants_excludes_rejected_shares() { let server = TestServer::spawn_with_db().await; let db_url = server.database_url().unwrap(); setup_test_schema(db_url.clone()).await.unwrap(); + insert_test_block(db_url.clone(), 50).await.unwrap(); let pool = sqlx::PgPool::connect(&db_url).await.unwrap(); @@ -1238,4 +1302,7 @@ async fn test_participants_excludes_rejected_shares() { let participants: Vec = server.get_json_async("/participants/50").await; assert_eq!(participants.len(), 1); assert_eq!(participants[0], "bar"); + + let summary: RoundSummary = server.get_json_async("/rounds/50/summary").await; + assert_eq!(summary.total_diff, 100.0); } From 19eabe6095808e66641da2202d22c89d8ec0a474 Mon Sep 17 00:00:00 2001 From: mrv777 Date: Sat, 16 May 2026 08:13:59 -0500 Subject: [PATCH 2/2] Persist round summaries --- src/subcommand/server.rs | 2 +- src/subcommand/server/database.rs | 145 +++++++++++++++++++++++++++ src/subcommand/server/sync_routes.rs | 7 ++ tests/server_with_db.rs | 91 +++++++++++++++++ tests/test_psql.rs | 12 +++ 5 files changed, 256 insertions(+), 1 deletion(-) diff --git a/src/subcommand/server.rs b/src/subcommand/server.rs index 0857c489..ed2fc893 100644 --- a/src/subcommand/server.rs +++ b/src/subcommand/server.rs @@ -295,7 +295,7 @@ impl Server { ); } - match Database::new(config.database_url()).await { + match Database::new_with_round_summary_ttl(config.database_url(), config.ttl()).await { Ok(database) => { if config.migrate_accounts() { let pool = database.pool.clone(); diff --git a/src/subcommand/server/database.rs b/src/subcommand/server/database.rs index 27274aa7..ab3ff63a 100644 --- a/src/subcommand/server/database.rs +++ b/src/subcommand/server/database.rs @@ -1,8 +1,11 @@ use { super::*, rounds::{Round, RoundParticipant, RoundSummary}, + tokio::sync::Mutex as TokioMutex, }; +const DEFAULT_ROUND_SUMMARY_TTL: Duration = Duration::from_secs(30); + #[derive(sqlx::FromRow, Deserialize, Serialize, Debug, Clone, PartialEq, ToSchema)] pub struct HighestDiff { pub blockheight: i32, @@ -55,13 +58,56 @@ pub struct UpdatePayoutStatusRequest { pub failure_reason: Option, } +#[derive(Debug)] +struct CachedRoundSummary { + value: Option, + last_updated: Instant, + ttl: Duration, +} + +impl CachedRoundSummary { + fn new(ttl: Duration) -> Self { + Self { + value: None, + last_updated: Instant::now() - ttl, + ttl, + } + } + + fn value(&self) -> Option { + if self.last_updated.elapsed() < self.ttl { + self.value.clone() + } else { + None + } + } + + fn update(&mut self, summary: RoundSummary) { + self.value = Some(summary); + self.last_updated = Instant::now(); + } + + fn invalidate(&mut self) { + self.value = None; + self.last_updated = Instant::now() - self.ttl; + } +} + #[derive(Debug, Clone)] pub struct Database { pub(crate) pool: Pool, + current_round_summary: Arc>, } impl Database { pub async fn new(database_url: String) -> Result { + Self::new_with_round_summary_ttl(database_url, DEFAULT_ROUND_SUMMARY_TTL).await + } + + pub async fn new_with_round_summary_ttl( + database_url: String, + round_summary_ttl: Duration, + ) -> Result { Ok(Self { pool: PgPoolOptions::new() .max_connections(5) @@ -69,6 +115,9 @@ impl Database { .connect(&database_url) .await .with_context(|| format!("failed to connect to database at `{database_url}`"))?, + current_round_summary: Arc::new(TokioMutex::new(CachedRoundSummary::new( + round_summary_ttl, + ))), }) } @@ -808,6 +857,62 @@ impl Database { } pub(crate) async fn get_round_summary(&self, blockheight: Option) -> Result { + match blockheight { + None => self.get_current_round_summary().await, + Some(h) => match self.get_round_summary_history(h).await { + Ok(Some(summary)) => Ok(summary), + Ok(None) => { + if let Err(e) = self.snapshot_round_summary(h).await { + warn!("Failed to snapshot round summary {h}: {e}"); + return self.query_round_summary(Some(h)).await; + } + + match self.get_round_summary_history(h).await { + Ok(Some(summary)) => Ok(summary), + Ok(None) => self.query_round_summary(Some(h)).await, + Err(_) => self.query_round_summary(Some(h)).await, + } + } + Err(_) => self.query_round_summary(Some(h)).await, + }, + } + } + + async fn get_current_round_summary(&self) -> Result { + let mut cached = self.current_round_summary.lock().await; + if let Some(summary) = cached.value() { + return Ok(summary); + } + + let summary = self.query_round_summary(None).await?; + cached.update(summary.clone()); + Ok(summary) + } + + pub(crate) async fn invalidate_current_round_summary(&self) { + self.current_round_summary.lock().await.invalidate(); + } + + async fn get_round_summary_history( + &self, + blockheight: i32, + ) -> Result, sqlx::Error> { + sqlx::query_as::<_, RoundSummary>( + " + SELECT + blockheight AS blockheight, + previous_blockheight, + total_diff + FROM round_summary_history + WHERE blockheight = $1 + ", + ) + .bind(blockheight) + .fetch_optional(&self.pool) + .await + } + + async fn query_round_summary(&self, blockheight: Option) -> Result { sqlx::query_as::<_, RoundSummary>( " WITH previous_block AS ( @@ -836,6 +941,46 @@ impl Database { .map_err(|err| anyhow!(err)) } + pub(crate) async fn snapshot_round_summary(&self, blockheight: i32) -> Result<()> { + sqlx::query( + " + INSERT INTO round_summary_history (blockheight, previous_blockheight, total_diff) + WITH target_block AS ( + SELECT blockheight + FROM blocks + WHERE blockheight = $1 + ), + previous_block AS ( + SELECT COALESCE(MAX(b.blockheight), 0) AS previous_blockheight + FROM blocks b + JOIN target_block tb ON b.blockheight < tb.blockheight + ), + round_total AS ( + SELECT COALESCE(SUM(rs.diff), 0.0)::FLOAT8 AS total_diff + FROM remote_shares rs, previous_block pb, target_block tb + WHERE rs.blockheight > pb.previous_blockheight + AND rs.blockheight <= tb.blockheight + AND rs.reject_reason IS NULL + ) + SELECT + tb.blockheight, + pb.previous_blockheight, + rt.total_diff + FROM target_block tb + CROSS JOIN previous_block pb + CROSS JOIN round_total rt + WHERE TRUE + ON CONFLICT (blockheight) DO NOTHING + ", + ) + .bind(blockheight) + .execute(&self.pool) + .await + .map_err(|err| anyhow!(err))?; + + Ok(()) + } + async fn get_round_participation_history( &self, blockheight: i32, diff --git a/src/subcommand/server/sync_routes.rs b/src/subcommand/server/sync_routes.rs index b3f44455..6c515565 100644 --- a/src/subcommand/server/sync_routes.rs +++ b/src/subcommand/server/sync_routes.rs @@ -116,6 +116,13 @@ pub(crate) async fn sync_batch( height, e ); } + if let Err(e) = database.snapshot_round_summary(height).await { + error!( + "Failed to snapshot round summary for block {}: {}", + height, e + ); + } + database.invalidate_current_round_summary().await; } else { if let Err(e) = database.refresh_current_round_participation().await { error!("Failed to refresh current round participation: {}", e); diff --git a/tests/server_with_db.rs b/tests/server_with_db.rs index 977b602d..1c46b4d8 100644 --- a/tests/server_with_db.rs +++ b/tests/server_with_db.rs @@ -1137,6 +1137,75 @@ async fn test_round_summary_totals_shares_since_previous_round() { assert_eq!(summary.total_diff, 6000.0); } +#[tokio::test] +async fn test_round_summary_uses_accounting_diff_not_share_diff() { + let server = TestServer::spawn_with_db().await; + let db_url = server.database_url().unwrap(); + setup_test_schema(db_url.clone()).await.unwrap(); + + insert_test_block(db_url.clone(), 5).await.unwrap(); + insert_test_block(db_url.clone(), 10).await.unwrap(); + insert_test_shares_with_diff( + db_url.clone(), + vec![("foo".to_string(), 11_400_000_000_000.0)], + 7, + ) + .await + .unwrap(); + + let summary: RoundSummary = server.get_json_async("/rounds/10/summary").await; + assert_eq!(summary.total_diff, 1.0); + + let participants: Vec = server.get_json_async("/rounds/10").await; + assert_eq!(participants[0].top_diff, 11_400_000_000_000.0); +} + +#[tokio::test] +async fn test_round_summary_uses_persisted_historical_snapshot() { + let server = TestServer::spawn_with_db().await; + let db_url = server.database_url().unwrap(); + setup_test_schema(db_url.clone()).await.unwrap(); + + insert_test_block(db_url.clone(), 5).await.unwrap(); + insert_test_block(db_url.clone(), 10).await.unwrap(); + insert_test_shares_for_round(db_url.clone(), vec![("foo", 1000.0)], 7, 100) + .await + .unwrap(); + + let summary: RoundSummary = server.get_json_async("/rounds/10/summary").await; + assert_eq!(summary.total_diff, 1000.0); + + insert_test_shares_for_round(db_url.clone(), vec![("bar", 9000.0)], 8, 200) + .await + .unwrap(); + + let summary: RoundSummary = server.get_json_async("/rounds/10/summary").await; + assert_eq!(summary.total_diff, 1000.0); +} + +#[tokio::test] +async fn test_round_summary_does_not_persist_missing_blockheight() { + let server = TestServer::spawn_with_db().await; + let db_url = server.database_url().unwrap(); + setup_test_schema(db_url.clone()).await.unwrap(); + + insert_test_block(db_url.clone(), 5).await.unwrap(); + insert_test_shares_for_round(db_url.clone(), vec![("foo", 1000.0)], 7, 100) + .await + .unwrap(); + + let summary: RoundSummary = server.get_json_async("/rounds/10/summary").await; + assert_eq!(summary.total_diff, 1000.0); + + insert_test_block(db_url.clone(), 10).await.unwrap(); + insert_test_shares_for_round(db_url.clone(), vec![("bar", 9000.0)], 8, 200) + .await + .unwrap(); + + let summary: RoundSummary = server.get_json_async("/rounds/10/summary").await; + assert_eq!(summary.total_diff, 10000.0); +} + #[tokio::test] async fn test_round_empty() { let server = TestServer::spawn_with_db().await; @@ -1224,6 +1293,28 @@ async fn test_current_round_summary_totals_since_latest_block() { assert_eq!(summary.total_diff, 2500.0); } +#[tokio::test] +async fn test_current_round_summary_uses_ttl_cache() { + let server = TestServer::spawn_with_db_args("--ttl 60").await; + let db_url = server.database_url().unwrap(); + setup_test_schema(db_url.clone()).await.unwrap(); + + insert_test_block(db_url.clone(), 5).await.unwrap(); + insert_test_shares_for_round(db_url.clone(), vec![("foo", 1000.0)], 7, 100) + .await + .unwrap(); + + let summary: RoundSummary = server.get_json_async("/rounds/current/summary").await; + assert_eq!(summary.total_diff, 1000.0); + + insert_test_shares_for_round(db_url.clone(), vec![("bar", 9000.0)], 8, 200) + .await + .unwrap(); + + let summary: RoundSummary = server.get_json_async("/rounds/current/summary").await; + assert_eq!(summary.total_diff, 1000.0); +} + #[tokio::test] async fn test_participants_for_blockheight() { let server = TestServer::spawn_with_db().await; diff --git a/tests/test_psql.rs b/tests/test_psql.rs index de3986d3..eef3d2e2 100644 --- a/tests/test_psql.rs +++ b/tests/test_psql.rs @@ -321,6 +321,18 @@ pub(crate) async fn setup_test_schema(db_url: String) -> Result<(), Box