Improve benchmark website hydration read path

Signed-off-by: Connor Tsui <connor.tsui20@gmail.com>

Author: connortsui20

Cargo.lock

@@ -26,10 +26,15 @@ path = "src/main.rs"
 anyhow = { workspace = true }
 axum = "0.8"
 base64 = "0.22"
+brotli = "8.0.2"
+bytes = "1.11"
+dashmap = { workspace = true }
 # track vortex-duckdb's bundled engine version (build.rs)
 duckdb = { version = "1.10502", features = ["bundled"] }
+flate2 = "1"
 maud = { version = "0.27", features = ["axum"] }
-serde = { workspace = true, features = ["derive"] }
+parking_lot = { workspace = true }
+serde = { workspace = true, features = ["derive", "rc"] }
 serde_json = { workspace = true }
 subtle = "2.6"
 thiserror = { workspace = true }
@@ -39,9 +44,9 @@ tower-http = { version = "0.6", features = ["compression-br", "compression-gzip"
 tracing = { workspace = true, features = ["std"] }
 tracing-subscriber = { workspace = true, features = ["env-filter", "fmt"] }
 twox-hash = "2.1"
+vortex-utils = { workspace = true }
 
 [dev-dependencies]
-flate2 = "1"
 insta = { workspace = true }
 reqwest = { workspace = true, features = ["json"] }
 tempfile = { workspace = true }

benchmarks-website/server/Cargo.toml

@@ -9,6 +9,7 @@
 //! returns a [`ChartResponse`].
 
 use std::collections::BTreeMap;
+use std::sync::Arc;
 
 use anyhow::Context as _;
 use anyhow::Result;
@@ -100,7 +101,7 @@ pub(crate) fn collect_group_charts(
         charts.push(NamedChartResponse {
             name: link.name,
             slug: link.slug,
-            chart,
+            chart: Arc::new(chart),
         });
     }
     if charts.is_empty() {

benchmarks-website/server/src/api/charts.rs

@@ -10,6 +10,7 @@
 //! ingest side, see [`crate::records`]).
 
 use std::collections::BTreeMap;
+use std::sync::Arc;
 
 use serde::Serialize;
 use serde_json::Value as JsonValue;
@@ -54,17 +55,21 @@ pub fn group_sort_key(name: &str) -> (usize, &str) {
 }
 
 /// Body of `GET /api/groups`: every group with its chart links and summary.
+///
+/// The inner [`Vec`] is held in an [`Arc`] so [`crate::query_cache::QueryCache`]
+/// can serve the same allocation to every concurrent reader without cloning.
+/// `Arc<T>` serialises through to `T`, so the wire shape is unchanged.
 #[derive(Debug, Serialize)]
 pub struct GroupsResponse {
     /// Every group surfaced by the discovery passes, in canonical order.
-    pub groups: Vec<Group>,
+    pub groups: Arc<Vec<Group>>,
 }
 
 /// One group: a display name, a slug for the group permalink, and the chart
 /// links inside it. Optionally carries a v2-compatible rollup summary and a
 /// short editorial description (rendered as a hover tooltip on the
 /// disclosure title).
-#[derive(Debug, Serialize)]
+#[derive(Debug, Clone, Serialize)]
 pub struct Group {
     /// Human-readable group label rendered in the disclosure header.
     pub name: String,
@@ -83,7 +88,7 @@ pub struct Group {
 }
 
 /// All charts in one group, returned by `GET /api/group/{slug}`.
-#[derive(Debug, Serialize)]
+#[derive(Debug, Clone, Serialize)]
 pub struct GroupChartsResponse {
     /// Group display name, matching the `name` field on [`Group`].
     pub name: String,
@@ -98,7 +103,7 @@ pub struct GroupChartsResponse {
 }
 
 /// Server-computed group summary, matching the v2 metadata contract.
-#[derive(Debug, Serialize)]
+#[derive(Debug, Clone, Serialize)]
 #[serde(tag = "type")]
 pub enum Summary {
     /// Random-access format ranking for the latest populated random-access chart.
@@ -161,7 +166,7 @@ pub enum Summary {
 }
 
 /// One random-access summary row.
-#[derive(Debug, Serialize)]
+#[derive(Debug, Clone, Serialize)]
 pub struct RandomAccessRanking {
     /// Series name, normally the physical format.
     pub name: String,
@@ -172,7 +177,7 @@ pub struct RandomAccessRanking {
 }
 
 /// One query benchmark summary row.
-#[derive(Debug, Serialize)]
+#[derive(Debug, Clone, Serialize)]
 pub struct QueryRanking {
     /// Series name, normally `engine:format`.
     pub name: String,
@@ -186,20 +191,24 @@ pub struct QueryRanking {
 /// A single chart inside a [`GroupChartsResponse`]. `name` is the chart's
 /// short label inside the group (e.g. `Q1`); `slug` round-trips through
 /// `/api/chart/{slug}`.
-#[derive(Debug, Serialize)]
+///
+/// `chart` is held in an [`Arc`] so the cache and the landing-page builder
+/// share the same allocation; `Arc<T>` serialises as `T`, so the wire shape
+/// is identical to a plain `ChartResponse`.
+#[derive(Debug, Clone, Serialize)]
 pub struct NamedChartResponse {
     /// Chart label rendered in the chart-card title (e.g. `Q1`).
     pub name: String,
     /// Slug for `/chart/{slug}`. Round-trips through [`crate::slug::ChartKey`].
     pub slug: String,
     /// Inlined chart payload — same shape as `/api/chart/{slug}`.
     #[serde(flatten)]
-    pub chart: ChartResponse,
+    pub chart: Arc<ChartResponse>,
 }
 
 /// One chart's short label inside a group (e.g. `Q1`) plus the slug that
 /// resolves to its `/api/chart/{slug}` payload.
-#[derive(Debug, Serialize)]
+#[derive(Debug, Clone, Serialize)]
 pub struct ChartLink {
     /// Chart label rendered in the chart-card title (e.g. `Q1`).
     pub name: String,

benchmarks-website/server/src/api/dto.rs

@@ -32,7 +32,9 @@ use axum::Json;
 use axum::extract::Path;
 use axum::extract::Query;
 use axum::extract::State;
+use axum::http::HeaderMap;
 use axum::response::IntoResponse;
+use axum::response::Response;
 use duckdb::Connection;
 
 pub(crate) use self::charts::chart_payload;
@@ -62,53 +64,180 @@ pub use self::window::CommitWindow;
 use crate::app::AppState;
 use crate::db;
 use crate::error::ApiError;
+use crate::read_model::ArtifactCachePolicy;
 use crate::slug::ChartKey;
 use crate::slug::GroupKey;
 
+pub(crate) fn read_transaction<T>(
+    conn: &mut Connection,
+    f: impl FnOnce(&Connection) -> Result<T>,
+) -> Result<T> {
+    conn.execute_batch("BEGIN TRANSACTION")?;
+    let result = f(conn);
+    match result {
+        Ok(value) => {
+            conn.execute_batch("COMMIT")?;
+            Ok(value)
+        }
+        Err(err) => {
+            let _ = conn.execute_batch("ROLLBACK");
+            Err(err)
+        }
+    }
+}
+
 /// Handler for `GET /api/groups`.
-pub async fn groups(State(state): State<AppState>) -> Result<impl IntoResponse, ApiError> {
-    let groups = db::run_blocking(&state.db, |conn| collect_groups(conn)).await?;
-    Ok(Json(GroupsResponse { groups }))
+pub async fn groups(
+    State(state): State<AppState>,
+    headers: HeaderMap,
+) -> Result<Response, ApiError> {
+    let generation = state.read_store.active();
+    Ok(generation
+        .groups_artifact()
+        .response(&headers, ArtifactCachePolicy::Revalidate))
 }
 
 /// Handler for `GET /api/chart/{slug}`.
 pub async fn chart(
     State(state): State<AppState>,
     Path(slug): Path<String>,
     Query(q): Query<ChartQuery>,
-) -> Result<impl IntoResponse, ApiError> {
+    headers: HeaderMap,
+) -> Result<Response, ApiError> {
     let key = ChartKey::from_slug(&slug)
         .map_err(|e| ApiError::BadRequest(format!("invalid slug: {e}")))?;
     let window = q.window();
-    let response =
-        db::run_blocking(&state.db, move |conn| chart_payload(conn, &key, &window)).await?;
+    if is_materialized_window(&window) {
+        let generation = state.read_store.active();
+        let response = generation
+            .chart_artifact(&slug)
+            .ok_or_else(|| ApiError::NotFound(format!("no data for slug {slug:?}")))?;
+        return Ok(response.response(&headers, ArtifactCachePolicy::Revalidate));
+    }
+    let response = cached_chart_payload(&state, &slug, &key, &window).await?;
     let response =
         response.ok_or_else(|| ApiError::NotFound(format!("no data for slug {slug:?}")))?;
-    Ok(Json(response))
+    Ok(Json(response).into_response())
 }
 
 /// Handler for `GET /api/group/{slug}`.
 pub async fn group(
     State(state): State<AppState>,
     Path(slug): Path<String>,
     Query(q): Query<ChartQuery>,
-) -> Result<impl IntoResponse, ApiError> {
+    headers: HeaderMap,
+) -> Result<Response, ApiError> {
     let key = GroupKey::from_slug(&slug)
         .map_err(|e| ApiError::BadRequest(format!("invalid group slug: {e}")))?;
     let window = q.window();
-    let response = db::run_blocking(&state.db, move |conn| {
-        collect_group_charts(conn, &key, &window)
-    })
-    .await?;
+    if is_materialized_window(&window) {
+        let generation = state.read_store.active();
+        let response = generation
+            .group_artifact(&slug)
+            .ok_or_else(|| ApiError::NotFound(format!("no data for group slug {slug:?}")))?;
+        return Ok(response.response(&headers, ArtifactCachePolicy::Revalidate));
+    }
+    let response = cached_group_charts(&state, &slug, &key, &window).await?;
     let response =
         response.ok_or_else(|| ApiError::NotFound(format!("no data for group slug {slug:?}")))?;
-    Ok(Json(response))
+    Ok(Json(response).into_response())
+}
+
+/// Handler for versioned latest-100 group shard artifacts.
+pub async fn group_shard_artifact(
+    State(state): State<AppState>,
+    Path((generation_id, group_slug, index)): Path<(String, String, usize)>,
+    headers: HeaderMap,
+) -> Result<Response, ApiError> {
+    let generation = state
+        .read_store
+        .generation(&generation_id)
+        .ok_or_else(|| ApiError::NotFound(format!("unknown generation {generation_id:?}")))?;
+    let artifact = generation
+        .group_shard_artifact(&group_slug, index)
+        .ok_or_else(|| ApiError::NotFound(format!("unknown group shard {group_slug:?}#{index}")))?;
+    Ok(artifact.response(&headers, ArtifactCachePolicy::Immutable))
+}
+
+fn is_materialized_window(window: &CommitWindow) -> bool {
+    matches!(window, CommitWindow::Last(n) if n.get() == DEFAULT_COMMIT_WINDOW)
+}
+
+/// Cache-aware wrapper around `collect_groups`.
+pub async fn cached_groups(state: &AppState) -> Result<std::sync::Arc<Vec<Group>>> {
+    let db = state.db.clone();
+    state
+        .cache
+        .groups(move || async move {
+            db::run_read_blocking(&db, |conn| read_transaction(conn, collect_groups)).await
+        })
+        .await
+}
+
+/// Cache-aware wrapper around [`collect_filter_universe`].
+pub async fn cached_filter_universe(state: &AppState) -> Result<std::sync::Arc<FilterUniverse>> {
+    let db = state.db.clone();
+    state
+        .cache
+        .filter_universe(move || async move {
+            db::run_read_blocking(&db, |conn| read_transaction(conn, collect_filter_universe)).await
+        })
+        .await
+}
+
+/// Cache-aware wrapper around `chart_payload`.
+pub async fn cached_chart_payload(
+    state: &AppState,
+    slug: &str,
+    key: &ChartKey,
+    window: &CommitWindow,
+) -> Result<Option<std::sync::Arc<ChartResponse>>> {
+    let db = state.db.clone();
+    let key_for_compute = key.clone();
+    let window_for_compute = *window;
+    state
+        .cache
+        .chart_payload(slug, window, move || async move {
+            db::run_read_blocking(&db, move |conn| {
+                read_transaction(conn, |conn| {
+                    chart_payload(conn, &key_for_compute, &window_for_compute)
+                })
+            })
+            .await
+        })
+        .await
+}
+
+/// Cache-aware wrapper around `collect_group_charts`.
+pub async fn cached_group_charts(
+    state: &AppState,
+    slug: &str,
+    key: &GroupKey,
+    window: &CommitWindow,
+) -> Result<Option<std::sync::Arc<GroupChartsResponse>>> {
+    let db = state.db.clone();
+    let key_for_compute = key.clone();
+    let window_for_compute = *window;
+    state
+        .cache
+        .group_charts(slug, window, move || async move {
+            db::run_read_blocking(&db, move |conn| {
+                read_transaction(conn, |conn| {
+                    collect_group_charts(conn, &key_for_compute, &window_for_compute)
+                })
+            })
+            .await
+        })
+        .await
 }
 
 /// Handler for `GET /health`.
 pub async fn health(State(state): State<AppState>) -> Result<impl IntoResponse, ApiError> {
     let path = state.db_path.display().to_string();
-    let response = db::run_blocking(&state.db, move |conn| collect_health(conn, path)).await?;
+    let response = db::run_read_blocking(&state.db, move |conn| {
+        read_transaction(conn, |conn| collect_health(conn, path))
+    })
+    .await?;
     Ok(Json(response))
 }
 

benchmarks-website/server/src/api/mod.rs

@@ -17,7 +17,7 @@ use super::dto::DEFAULT_COMMIT_WINDOW;
 ///
 /// `Last(n)` keeps the most recent `n` commits by `commits.timestamp`; `All`
 /// returns every commit ever ingested.
-#[derive(Debug, Clone, Copy)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
 pub enum CommitWindow {
     /// Keep the most recent `n` commits.
     Last(NonZeroU32),