diff --git a/crates/server/src/routes/dashboard/live.rs b/crates/server/src/routes/dashboard/live.rs index efaeabf1..6027011c 100644 --- a/crates/server/src/routes/dashboard/live.rs +++ b/crates/server/src/routes/dashboard/live.rs @@ -30,26 +30,33 @@ use crate::{permissions::UiPermissions, state::AppState}; /// Bursts of build updates within this window rerender once. const SETTLE: Duration = Duration::from_millis(500); +/// Resolving the viewer touches its session row, so a ticking view re-checks +/// access every this many ticks rather than every second. +const RECHECK_TICKS: u32 = 30; + /// Bumped whenever a build is inserted or changes status. #[derive(Clone)] -pub(super) struct BuildsChanged(watch::Receiver); +pub(super) struct BuildsChanged { + receiver: watch::Receiver, + ticks: u32, +} /// Why a live build view woke up. pub(super) enum Wake { - /// A build changed, so reload. - Changed, + /// A build changed, so reload as the viewer resolved again. + Changed(DashboardContext), /// A second passed, so redraw running clocks. Tick, } impl BuildsChanged { pub(super) fn subscribe(cx: &Cx) -> Self { - let mut receiver = app_context::(cx).0.clone(); + let mut receiver = app_context::(cx).receiver.clone(); receiver.mark_unchanged(); - Self(receiver) + Self { receiver, ticks: 0 } } - /// Waits for a build change, or one second when `ticking`, then resolves the + /// Waits for a build change, or one second when `ticking`, and resolves the /// viewer again, since a run outlives the request that started it. `None` /// once the viewer has gone, and the redirect once they lost access. pub(super) async fn next( @@ -57,7 +64,7 @@ impl BuildsChanged { cx: &Cx, page: DashboardPage, ticking: bool, - ) -> Result> { + ) -> Result> { if !connected(cx) { return Ok(None); } @@ -70,20 +77,24 @@ impl BuildsChanged { } }; - let wake = tokio::select! { - changed = self.0.changed() => { + tokio::select! { + changed = self.receiver.changed() => { if changed.is_err() { return Ok(None); } tokio::time::sleep(SETTLE).await; - self.0.mark_unchanged(); - Wake::Changed + self.receiver.mark_unchanged(); + Ok(Some(Wake::Changed(shard_viewer(cx, page).await?))) }, - () = tick => Wake::Tick, - }; - - Ok(Some((wake, shard_viewer(cx, page).await?))) + () = tick => { + self.ticks += 1; + if self.ticks.is_multiple_of(RECHECK_TICKS) { + shard_viewer(cx, page).await?; + } + Ok(Some(Wake::Tick)) + }, + } } } @@ -105,7 +116,7 @@ pub fn service(state: AppState, assets: AssetBundle) -> TowerService { let router = Router::builder() .app_context(state) - .app_context(BuildsChanged(receiver)) + .app_context(BuildsChanged { receiver, ticks: 0 }) .discover() .assets(assets) .runtime() @@ -193,13 +204,13 @@ async fn queue_tables( tables(data: &data, permissions: &ctx.permissions, csrf_token: &ctx.csrf_token) }?; - let Some((wake, current)) = changed.next(cx, DashboardPage::Queue, ticking).await? else { - break Ok(token); - }; - ctx = current; - - if matches!(wake, Wake::Changed) { - data = load_queue(state, &filter).await; + match changed.next(cx, DashboardPage::Queue, ticking).await? { + None => break Ok(token), + Some(Wake::Changed(current)) => { + ctx = current; + data = load_queue(state, &filter).await; + }, + Some(Wake::Tick) => {}, } } }) diff --git a/crates/server/src/routes/dashboard/views/builds.rs b/crates/server/src/routes/dashboard/views/builds.rs index f053f2cc..a64c4a34 100644 --- a/crates/server/src/routes/dashboard/views/builds.rs +++ b/crates/server/src/routes/dashboard/views/builds.rs @@ -1,14 +1,13 @@ //! The dashboard home, the build list, a build, and its log. -use std::{collections::HashMap, path::Path, time::Duration}; +use std::{collections::HashMap, path::Path}; use circus_common::models::{BuildProduct, NewsItem}; -use jiff::Timestamp; use topcoat::{ Result, context::{Cx, app_context}, router::{error::RouterErrorExt, page, path_param, query_params}, - runtime::{connected, shard}, + runtime::shard, view::{View, component, emit, live, view}, }; use uuid::Uuid; @@ -16,10 +15,11 @@ use uuid::Uuid; use super::super::{ build_log::{BuildLogView, parse_build_log}, layout::{document, shard_viewer, viewer}, + live::{BuildsChanged, Wake}, pages::{ dashboard_system_filters, derivation_name, - format_elapsed, + elapsed_since, is_failed_status, }, shared::{ @@ -74,26 +74,6 @@ async fn home_page(cx: &Cx) -> Result { let announcements = circus_common::repo::news::list(&state.pool, 3, 0) .await .unwrap_or_default(); - let needle = search.trim().to_lowercase(); - let recent_builds: Vec = overview - .recent_builds - .iter() - .map(BuildView::from) - .filter(|build| status.is_empty() || build.status_class == status) - .filter(|build| { - needle.is_empty() - || [&build.project_name, &build.job_name, &build.system] - .iter() - .any(|field| field.to_lowercase().contains(&needle)) - }) - .collect(); - let totals = Totals { - total: overview.total_builds, - succeeded: overview.completed_builds, - failed: overview.failed_builds, - running: overview.running_builds, - pending: overview.pending_builds, - }; Ok(view! { document(title: "Dashboard", viewer: &viewer, topbar: false, @@ -132,14 +112,14 @@ async fn home_page(cx: &Cx) -> Result { announcement_lines(items: announcements) - metric_strip(totals: totals) + live_totals()
projects_panel( projects: overview.projects.iter().map(ProjectSummaryView::from).collect(), is_admin: viewer.is_admin, ) - recent_builds_panel(builds: recent_builds, search: search, status: status) + recent_builds_panel(search: search, status: status) agents_panel( workers: overview.workers.iter().map(WorkerSummaryView::from).collect(), show_names: !viewer.auth_name.is_empty(), @@ -179,6 +159,7 @@ async fn announcement_lines(items: Vec) -> Result { }) } +#[derive(Clone, Copy)] struct Totals { total: i64, succeeded: i64, @@ -308,9 +289,42 @@ async fn projects_panel( }) } +/// The home page's build counts, reloaded whenever a build changes. +#[shard] +async fn live_totals(cx: &Cx) -> Result { + shard_viewer(cx, DashboardPage::Home).await?; + let state = app_context::(cx); + let load = async || { + let stats = circus_common::repo::builds::get_stats(&state.pool).await?; + Ok::<_, circus_common::error::CiError>(Totals { + total: stats.total_builds.unwrap_or(0), + succeeded: stats.completed_builds.unwrap_or(0), + failed: stats.failed_builds.unwrap_or(0), + running: stats.running_builds.unwrap_or(0), + pending: stats.pending_builds.unwrap_or(0), + }) + }; + let mut changed = BuildsChanged::subscribe(cx); + let mut totals = load().await?; + + Ok(live! { + loop { + let token = emit! { metric_strip(totals: totals) }?; + + if changed.next(cx, DashboardPage::Home, false).await?.is_none() { + break Ok(token); + } + + match load().await { + Ok(fresh) => totals = fresh, + Err(error) => tracing::warn!("Failed to reload build totals: {error}"), + } + } + }) +} + #[component] async fn recent_builds_panel( - builds: Vec, search: String, status: String, ) -> Result { @@ -332,7 +346,7 @@ async fn recent_builds_panel( @@ -343,6 +357,40 @@ async fn recent_builds_panel( + recent_builds_table(search: search, status: status.clone()) +
+ }) +} + +/// Recent builds matching the home filter, reloaded whenever a build changes. +#[shard] +async fn recent_builds_table( + cx: &Cx, + search: String, + status: String, +) -> Result { + shard_viewer(cx, DashboardPage::Home).await?; + let state = app_context::(cx); + let needle = search.trim().to_lowercase(); + let mut changed = BuildsChanged::subscribe(cx); + let mut recent = operator::recent_builds(state) + .await + .map_err(|error| error.0)?; + + Ok(live! { + loop { + let builds: Vec = recent + .iter() + .map(BuildView::from) + .filter(|build| status.is_empty() || build.status_class == status) + .filter(|build| { + needle.is_empty() + || [&build.project_name, &build.job_name, &build.system] + .iter() + .any(|field| field.to_lowercase().contains(&needle)) + }) + .collect(); + let token = emit! { if builds.is_empty() {
"No builds yet"
@@ -384,7 +432,17 @@ async fn recent_builds_panel(
} - + }?; + + if changed.next(cx, DashboardPage::Home, false).await?.is_none() { + break Ok(token); + } + + match operator::recent_builds(state).await { + Ok(fresh) => recent = fresh, + Err(error) => tracing::warn!("Failed to reload recent builds: {}", error.0), + } + } }) } @@ -556,92 +614,10 @@ pub(in crate::routes::dashboard) struct BuildFilterParams { #[page("/builds")] async fn builds_page(cx: &Cx) -> Result { let viewer = viewer(cx, DashboardPage::Builds).await?; - let state = app_context::(cx); let params = query_params::(cx)?; - let limit = params.limit.unwrap_or(50).clamp(1, 200); - let offset = params.offset.unwrap_or(0).max(0); - let items = circus_common::repo::builds::list_filtered( - &state.pool, - None, - params.status.as_deref(), - params.system.as_deref(), - params.job_name.as_deref(), - limit, - offset, - ) - .await - .unwrap_or_default(); - let total = circus_common::repo::builds::count_filtered( - &state.pool, - None, - params.status.as_deref(), - params.system.as_deref(), - params.job_name.as_deref(), - ) - .await - .unwrap_or(0); - let pagination = Pagination::new(total, offset, limit); - - let mut context_by_eval = HashMap::new(); - for item in &items { - if context_by_eval.contains_key(&item.evaluation_id) { - continue; - } - - let Ok(eval) = - circus_common::repo::evaluations::get(&state.pool, item.evaluation_id) - .await - else { - continue; - }; - let Ok(jobset) = - circus_common::repo::jobsets::get(&state.pool, eval.jobset_id).await - else { - continue; - }; - let Ok(project) = - circus_common::repo::projects::get(&state.pool, jobset.project_id).await - else { - continue; - }; - context_by_eval.insert( - item.evaluation_id, - (project.id, project.name, jobset.id, jobset.name), - ); - } - - let rows: Vec = items - .iter() - .map(|item| { - context_by_eval.get(&item.evaluation_id).map_or_else( - || build_view(item), - |(project_id, project_name, jobset_id, jobset_name)| { - build_view_with_context( - item, - *project_id, - project_name, - *jobset_id, - jobset_name, - ) - }, - ) - }) - .collect(); let status = params.status.clone().unwrap_or_default(); let system = params.system.clone().unwrap_or_default(); let job = params.job_name.clone().unwrap_or_default(); - let page_href = |page_offset: i64| { - let query: String = url::form_urlencoded::Serializer::new(String::new()) - .append_pair("offset", &page_offset.to_string()) - .append_pair("limit", &limit.to_string()) - .append_pair("status", &status) - .append_pair("system", &system) - .append_pair("job_name", &job) - .finish(); - format!("/builds?{query}") - }; - let prev_href = page_href(pagination.prev_offset); - let next_href = page_href(pagination.next_offset); let statuses = [ ("", "All"), ("pending", "Pending"), @@ -672,6 +648,115 @@ async fn builds_page(cx: &Cx) -> Result { + builds_table( + status: params.status.clone(), + system: params.system.clone(), + job_name: params.job_name.clone(), + limit: params.limit, + offset: params.offset, + ) + ) + }) +} + +/// The filtered build list, reloaded whenever a build changes. +#[shard] +async fn builds_table( + cx: &Cx, + status: Option, + system: Option, + job_name: Option, + limit: Option, + offset: Option, +) -> Result { + shard_viewer(cx, DashboardPage::Builds).await?; + let state = app_context::(cx); + let limit = limit.unwrap_or(50).clamp(1, 200); + let offset = offset.unwrap_or(0).max(0); + let mut changed = BuildsChanged::subscribe(cx); + + Ok(live! { + loop { + let items = circus_common::repo::builds::list_filtered( + &state.pool, + None, + status.as_deref(), + system.as_deref(), + job_name.as_deref(), + limit, + offset, + ) + .await + .unwrap_or_default(); + let total = circus_common::repo::builds::count_filtered( + &state.pool, + None, + status.as_deref(), + system.as_deref(), + job_name.as_deref(), + ) + .await + .unwrap_or(0); + let pagination = Pagination::new(total, offset, limit); + + let mut context_by_eval = HashMap::new(); + for item in &items { + if context_by_eval.contains_key(&item.evaluation_id) { + continue; + } + + let Ok(eval) = + circus_common::repo::evaluations::get(&state.pool, item.evaluation_id) + .await + else { + continue; + }; + let Ok(jobset) = + circus_common::repo::jobsets::get(&state.pool, eval.jobset_id).await + else { + continue; + }; + let Ok(project) = + circus_common::repo::projects::get(&state.pool, jobset.project_id).await + else { + continue; + }; + context_by_eval.insert( + item.evaluation_id, + (project.id, project.name, jobset.id, jobset.name), + ); + } + + let rows: Vec = items + .iter() + .map(|item| { + context_by_eval.get(&item.evaluation_id).map_or_else( + || build_view(item), + |(project_id, project_name, jobset_id, jobset_name)| { + build_view_with_context( + item, + *project_id, + project_name, + *jobset_id, + jobset_name, + ) + }, + ) + }) + .collect(); + let page_href = |page_offset: i64| { + let query: String = url::form_urlencoded::Serializer::new(String::new()) + .append_pair("offset", &page_offset.to_string()) + .append_pair("limit", &limit.to_string()) + .append_pair("status", status.as_deref().unwrap_or_default()) + .append_pair("system", system.as_deref().unwrap_or_default()) + .append_pair("job_name", job_name.as_deref().unwrap_or_default()) + .finish(); + format!("/builds?{query}") + }; + let prev_href = page_href(pagination.prev_offset); + let next_href = page_href(pagination.next_offset); + let token = emit! { if rows.is_empty() {
"No builds match filters"
@@ -730,7 +815,12 @@ async fn builds_page(cx: &Cx) -> Result { } } - ) + }?; + + if changed.next(cx, DashboardPage::Builds, false).await?.is_none() { + break Ok(token); + } + } }) } @@ -802,37 +892,6 @@ async fn build_page(cx: &Cx) -> Result { .iter() .map(build_view) .collect(); - let broke_in = if is_failed_status(build.status) { - circus_common::repo::builds::broke_in(&state.pool, id) - .await - .unwrap_or_else(|error| { - tracing::warn!(build_id = %id, "Failed to find where the job broke: {error}"); - None - }) - .map(|found| { - BrokeInView { - build_id: found.build_id, - commit_short: found.commit_hash.chars().take(12).collect(), - commit_subject: found.commit_subject.unwrap_or_default(), - last_success_build_id: found.last_success_build_id, - last_success_short: found - .last_success_commit - .chars() - .take(12) - .collect(), - } - }) - } else { - None - }; - let builder_label = match build.agent_machine_id { - Some(machine_id) => { - circus_common::repo::builder_sessions::get(&state.pool, machine_id) - .await - .map_or_else(|_| "local".to_owned(), |session| session.name) - }, - None => "local".to_owned(), - }; let build = build_view(&build); let title = format!("Build {}", build.job_name); @@ -850,7 +909,7 @@ async fn build_page(cx: &Cx) -> Result { (lineage.commit_short.clone())

"Build: " (build.job_name.clone())

- build_overview(details: &build, builder_label: builder_label, broke_in: broke_in) + live_overview(id: id.to_string())

"Reproduce This Build"

@@ -892,12 +951,92 @@ async fn build_page(cx: &Cx) -> Result { }) } -#[component] -async fn build_overview( - details: &BuildView, +/// What the overview panel shows, reloaded whenever a build changes. +struct Overview { + details: BuildView, builder_label: String, - broke_in: Option, -) -> Result { + broke_in: Option, +} + +async fn load_overview(state: &AppState, id: Uuid) -> Option { + let build = circus_common::repo::builds::get(&state.pool, id) + .await + .ok()?; + let broke_in = if is_failed_status(build.status) { + circus_common::repo::builds::broke_in(&state.pool, id) + .await + .unwrap_or_else(|error| { + tracing::warn!(build_id = %id, "Failed to find where the job broke: {error}"); + None + }) + .map(|found| { + BrokeInView { + build_id: found.build_id, + commit_short: found.commit_hash.chars().take(12).collect(), + commit_subject: found.commit_subject.unwrap_or_default(), + last_success_build_id: found.last_success_build_id, + last_success_short: found + .last_success_commit + .chars() + .take(12) + .collect(), + } + }) + } else { + None + }; + let builder_label = match build.agent_machine_id { + Some(machine_id) => { + circus_common::repo::builder_sessions::get(&state.pool, machine_id) + .await + .map_or_else(|_| "local".to_owned(), |session| session.name) + }, + None => "local".to_owned(), + }; + + Some(Overview { + details: build_view(&build), + builder_label, + broke_in, + }) +} + +/// The overview panel, reloaded when the build changes and ticking while it +/// runs. +#[shard] +async fn live_overview(cx: &Cx, id: String) -> Result { + shard_viewer(cx, DashboardPage::Build).await?; + let id = id.parse::().ok().ok_or_not_found()?; + let state = app_context::(cx); + let mut changed = BuildsChanged::subscribe(cx); + let mut overview = load_overview(state, id).await.ok_or_not_found()?; + + Ok(live! { + loop { + let running = overview.details.started_epoch.is_some() + && overview.details.duration.is_empty(); + let token = emit! { build_overview(overview: &overview) }?; + + match changed.next(cx, DashboardPage::Build, running).await? { + None => break Ok(token), + Some(Wake::Changed(_)) => { + if let Some(fresh) = load_overview(state, id).await { + overview = fresh; + } + }, + Some(Wake::Tick) => {}, + } + } + }) +} + +#[component] +async fn build_overview(overview: &Overview) -> Result { + let Overview { + details, + builder_label, + broke_in, + } = overview; let signed = if details.signed { "Yes" } else { "No" }; Ok(view! { @@ -908,21 +1047,21 @@ async fn build_overview(
meta_item(label: "Status", - (details.status_text.clone()) + (details.status_text.as_str()) ) - meta_item(label: "System", (details.system.clone())) - meta_item(label: "Builder", (builder_label)) - meta_item(label: "Created", (details.created_at.clone())) + meta_item(label: "System", (details.system.as_str())) + meta_item(label: "Builder", (builder_label.as_str())) + meta_item(label: "Created", (details.created_at.as_str())) if !details.started_at.is_empty() { - meta_item(label: "Started", (details.started_at.clone())) + meta_item(label: "Started", (details.started_at.as_str())) } if !details.completed_at.is_empty() { - meta_item(label: "Completed", (details.completed_at.clone())) + meta_item(label: "Completed", (details.completed_at.as_str())) } match (details.started_epoch, details.duration.is_empty()) { - (Some(started), true) => meta_item(label: "Elapsed", elapsed(started: started)), - (Some(started), false) => meta_item(label: "Duration", elapsed(started: started)), - (None, false) => meta_item(label: "Duration", (details.duration.clone())), + (Some(started), true) => meta_item(label: "Elapsed", (elapsed_since(started))), + (Some(started), false) => meta_item(label: "Duration", (elapsed_since(started))), + (None, false) => meta_item(label: "Duration", (details.duration.as_str())), (None, true) => {}, } meta_item(label: "Priority", (details.priority)) @@ -930,9 +1069,9 @@ async fn build_overview( if details.is_aggregate { meta_item(label: "Aggregate", "Yes") } - meta_item(label: "Derivation", (details.drv_path.clone())) + meta_item(label: "Derivation", (details.drv_path.as_str())) if !details.output_path.is_empty() { - meta_item(label: "Output", (details.output_path.clone())) + meta_item(label: "Output", (details.output_path.as_str())) } if let Some(broke) = broke_in {
@@ -941,14 +1080,14 @@ async fn build_overview( if broke.build_id == details.id { "This evaluation" } else { - (broke.commit_short) + (broke.commit_short.as_str()) } if !broke.commit_subject.is_empty() { - (broke.commit_subject) + (broke.commit_subject.as_str()) } "Last passed at " - (broke.last_success_short) + (broke.last_success_short.as_str())
@@ -959,14 +1098,14 @@ async fn build_overview(

"Errors"

    for line in &details.error_lines { -
  • (line.text.clone())
    • +
  • (line.text.as_str())
    • }
} else if !details.error_message.is_empty() {

"Errors"

- 
(details.error_message.clone())
+ 
(details.error_message.as_str())
} if details.has_log { @@ -992,27 +1131,6 @@ async fn meta_item( }) } -/// A running build's elapsed time, ticking once a second while the page is -/// open. -#[shard] -async fn elapsed(cx: &Cx, started: i64) -> Result { - shard_viewer(cx, DashboardPage::Build).await?; - - Ok(live! { - let mut tick = tokio::time::interval(Duration::from_secs(1)); - - loop { - tick.tick().await; - let secs = Timestamp::now().as_second().saturating_sub(started).max(0); - let token = emit! { (format_elapsed(secs)) }?; - - if !connected(cx) { - break Ok(token); - } - } - }) -} - #[component] async fn related_builds( title: &str, diff --git a/crates/server/src/routes/dashboard/views/evaluations.rs b/crates/server/src/routes/dashboard/views/evaluations.rs index 25502bce..aea53da3 100644 --- a/crates/server/src/routes/dashboard/views/evaluations.rs +++ b/crates/server/src/routes/dashboard/views/evaluations.rs @@ -30,7 +30,7 @@ use super::super::{ components::{confirm_button, local_time}, layout::{document, shard_viewer, viewer}, pages::{ - format_elapsed, + elapsed_since, is_failed_derivation_status, is_failed_status, is_job_name, @@ -89,10 +89,6 @@ fn non_empty(value: &str) -> Option<&str> { Some(value).filter(|value| !value.is_empty()) } -fn elapsed_since(epoch: i64) -> String { - format_elapsed((Timestamp::now().as_second() - epoch).max(0)) -} - #[component] async fn status_badges( class: &str,