feat: add pool total work per round

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,
@@ -292,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();

src/subcommand/server/database.rs

@@ -1,8 +1,11 @@
 use {
     super::*,
-    rounds::{Round, RoundParticipant},
+    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,20 +58,66 @@ pub struct UpdatePayoutStatusRequest {
     pub failure_reason: Option<String>,
 }
 
+#[derive(Debug)]
+struct CachedRoundSummary {
+    value: Option<RoundSummary>,
+    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<RoundSummary> {
+        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<Postgres>,
+    current_round_summary: Arc<TokioMutex<CachedRoundSummary>>,
 }
 
 impl Database {
     pub async fn new(database_url: String) -> Result<Self> {
+        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<Self> {
         Ok(Self {
             pool: PgPoolOptions::new()
                 .max_connections(5)
                 .acquire_timeout(Duration::from_secs(5))
                 .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,
+            ))),
         })
     }
 
@@ -807,6 +856,131 @@ impl Database {
         }
     }
 
+    pub(crate) async fn get_round_summary(&self, blockheight: Option<i32>) -> Result<RoundSummary> {
+        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<RoundSummary> {
+        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<Option<RoundSummary>, 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<i32>) -> Result<RoundSummary> {
+        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))
+    }
+
+    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,

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<i32>,
+    pub(crate) previous_blockheight: i32,
+    pub(crate) total_diff: f64,
+}
+
 pub(crate) fn rounds_router(config: Arc<ServerConfig>, 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<Database>,
+) -> ServerResult<Response> {
+    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<i32>,
+    Extension(database): Extension<Database>,
+) -> ServerResult<Response> {
+    Ok(Json(database.get_round_summary(Some(blockheight)).await?).into_response())
+}
+
 #[utoipa::path(
     get,
     path = "/participants/{blockheight}",

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);