From f45d0ccf1bf43026900e053799beb31acbb9cc25 Mon Sep 17 00:00:00 2001 From: calvin-archastro Date: Thu, 15 Oct 2026 09:41:39 -0400 Subject: [PATCH 1/2] feat(aster-srv): median test-file durations for splitting partitioned suites GET /v1/test-results/durations?target=&day= returns each test file's median duration over the named targets' latest 20 complete, uncached runs in the 7 days before `day` began. Every shard of one commit asks with the same targets and day and gets the same answer; a day that has not begun is 409. Answers are kept per replica since they cannot change. Co-Authored-By: Claude Opus 5.5 (1M context) --- crates/aster-srv/src/http.rs | 11 +- crates/aster-srv/src/lib.rs | 3 + .../aster-srv/src/test_results/durations.rs | 245 +++++++++++++++++ crates/aster-srv/src/test_results/mod.rs | 1 + crates/aster-srv/tests/durations.rs | 259 ++++++++++++++++++ 5 files changed, 517 insertions(+), 2 deletions(-) create mode 100644 crates/aster-srv/src/test_results/durations.rs create mode 100644 crates/aster-srv/tests/durations.rs diff --git a/crates/aster-srv/src/http.rs b/crates/aster-srv/src/http.rs index 26e2bca..79449cb 100644 --- a/crates/aster-srv/src/http.rs +++ b/crates/aster-srv/src/http.rs @@ -19,7 +19,7 @@ use tokio::net::TcpListener; use crate::auth::{AuthError, Identity}; use crate::test_results::ingest::{self, Upload}; -use crate::test_results::query; +use crate::test_results::{durations, query}; use crate::App; /// A response before it is sent. @@ -135,6 +135,13 @@ async fn route(app: &App, request: Request) -> Reply { }; ingest::handle(app, &identity, Upload { body, gzip }).await } + (&Method::GET, "/v1/test-results/durations") => { + let identity = match authenticate(app, &request).await { + Ok(identity) => identity, + Err(reply) => return reply, + }; + durations::handle(app, &identity, request.uri().query()).await + } (&Method::GET, path) if query::parse_route(path).is_some() => { let identity = match authenticate(app, &request).await { Ok(identity) => identity, @@ -151,7 +158,7 @@ async fn route(app: &App, request: Request) -> Reply { .map(str::to_string); crate::ui::handle(app, method, path, request.uri().query(), assertion).await } - (_, "/healthz" | "/readyz" | "/v1/test-results/runs") => { + (_, "/healthz" | "/readyz" | "/v1/test-results/runs" | "/v1/test-results/durations") => { Reply::error(405, "method not allowed") } (_, path) if query::parse_route(path).is_some() => Reply::error(405, "method not allowed"), diff --git a/crates/aster-srv/src/lib.rs b/crates/aster-srv/src/lib.rs index ee357bf..25cb3eb 100644 --- a/crates/aster-srv/src/lib.rs +++ b/crates/aster-srv/src/lib.rs @@ -38,6 +38,8 @@ pub struct App { pub verifier: auth::Verifier, pub iap: iap::IapVerifier, pub raw_store: raw_store::RawStore, + /// Test durations already answered (they cannot change). + pub durations: test_results::durations::Answers, } /// Every feature's migrations, in order. @@ -79,6 +81,7 @@ impl App { verifier, iap, raw_store, + durations: Default::default(), })) } } diff --git a/crates/aster-srv/src/test_results/durations.rs b/crates/aster-srv/src/test_results/durations.rs new file mode 100644 index 0000000..c3b9ec4 --- /dev/null +++ b/crates/aster-srv/src/test_results/durations.rs @@ -0,0 +1,245 @@ +//! `GET /v1/test-results/durations?target=&target=&day=`: how long each +//! test file takes, so Aster can deal a partitioned suite out by time +//! instead of by count. +//! +//! Every shard of one commit asks the same question and must get the same +//! answer, or a file runs on no shard or on two. So the answer depends only +//! on the request: `day` (the UTC day of the commit, the same on every +//! shard) fixes a window of runs received in the 7 days before that day +//! began, less a 15-minute margin for uploads still committing when it +//! began. A window that may still be filling (`day` is in the future) is +//! refused with 409, never answered from partial data. +//! +//! Per file: the sum of its tests' first-attempt durations in a run, then +//! the median of that over the most recent [`RUNS_PER_TARGET`] uncached, +//! completely captured runs of each target in the window. Files are named +//! as the runner reported them, relative to the directory the tests ran in. + +use std::collections::HashMap; +use std::sync::Mutex; + +use chrono::{DateTime, Duration, NaiveDate, Utc}; +use serde_json::{json, Value}; + +use super::query::parse_query; +use crate::auth::Identity; +use crate::http::Reply; +use crate::App; + +/// The most targets one request names: the shards of one suite. +pub const MAX_TARGETS: usize = 32; +/// Runs of each target the median is taken over. +pub const RUNS_PER_TARGET: i64 = 20; +/// Days of runs before `day` that count. +pub const WINDOW_DAYS: i64 = 7; +/// Answers a replica keeps; past this it forgets them all. +const CACHED_ANSWERS: usize = 256; + +/// Answers already given, by repository, targets and day. +pub type Answers = Mutex, NaiveDate), Value>>; + +/// Runs received this close before `day` began are left out: their upload +/// may still have been committing. +const SETTLE: Duration = Duration::minutes(15); + +/// The window for `day`: runs with `received_at` in `[from, to)`. +pub fn window(day: NaiveDate) -> (DateTime, DateTime) { + let start = day.and_hms_opt(0, 0, 0).expect("midnight").and_utc(); + let to = start - SETTLE; + (to - Duration::days(WINDOW_DAYS), to) +} + +/// The query: repo `$1`, targets `$2`, window `$3`..`$4`, days `$5`..`$6` +/// (the partitions that can hold the window), runs per target `$7`. +/// +/// The picked runs (at most [`MAX_TARGETS`] × [`RUNS_PER_TARGET`]) drive +/// the read of their results through `test_results_run`, one partition +/// each; `OFFSET 0` keeps the planner from turning that into a scan of +/// every partition. On a seeded 14M-row copy, 4 targets of 3,000 tests +/// take 1 to 2.5 s (the scan took over 10 minutes), so an answer is kept +/// once computed: it cannot change. +pub const SQL: &str = "\ +WITH recent AS ( \ + SELECT t.run_id, t.address, t.received_on, \ + row_number() OVER (PARTITION BY t.address \ + ORDER BY r.received_at DESC, t.run_id DESC) AS n \ + FROM test_results.target_runs t \ + JOIN test_results.runs r ON r.run_id = t.run_id \ + WHERE t.repo = $1 AND t.address = ANY($2) AND NOT t.cached \ + AND t.confidence = 'complete' \ + AND t.received_on BETWEEN $5 AND $6 \ + AND r.received_at >= $3 AND r.received_at < $4), \ +picked AS MATERIALIZED (SELECT run_id, address, received_on FROM recent WHERE n <= $7), \ +per_test AS MATERIALIZED ( \ + SELECT p.run_id, tr.repo, tr.test_id, tr.duration_ms \ + FROM picked p \ + CROSS JOIN LATERAL ( \ + SELECT tr.repo, tr.test_id, tr.duration_ms FROM test_results.test_results tr \ + WHERE tr.run_id = p.run_id AND tr.received_on = p.received_on \ + AND tr.target = p.address AND tr.attempt = 1 AND tr.outcome <> 'skip' \ + AND tr.duration_ms IS NOT NULL \ + OFFSET 0) tr), \ +per_run AS ( \ + SELECT t.run_id, s.file, sum(t.duration_ms) AS ms \ + FROM per_test t \ + JOIN test_results.tests s ON s.repo = t.repo AND s.test_id = t.test_id \ + WHERE s.file IS NOT NULL AND s.file <> '' \ + GROUP BY t.run_id, s.file) \ +SELECT file, \ + round(percentile_cont(0.5) WITHIN GROUP (ORDER BY ms))::bigint AS p50_ms, \ + count(*) AS runs \ +FROM per_run GROUP BY file ORDER BY file"; + +pub async fn handle(app: &App, identity: &Identity, query: Option<&str>) -> Reply { + let (targets, day) = match parse(identity, query) { + Ok(found) => found, + Err(reply) => return reply, + }; + if day > Utc::now().date_naive() { + return Reply::error( + 409, + format!("{day} has not begun, so its window may still be filling"), + ); + } + let key = (identity.repository.clone(), targets, day); + if let Some(body) = app.durations.lock().expect("answers").get(&key) { + return Reply::json(200, body.clone()); + } + match read(app, &key.0, &key.1, day).await { + Ok(body) => { + let mut kept = app.durations.lock().expect("answers"); + if kept.len() >= CACHED_ANSWERS { + kept.clear(); + } + kept.insert(key, body.clone()); + Reply::json(200, body) + } + Err(error) => { + crate::log::error( + "cannot read test durations", + json!({"error": format!("{error:#}")}), + ); + Reply::error(503, "the durations could not be read; retry later") + } + } +} + +fn parse(identity: &Identity, query: Option<&str>) -> Result<(Vec, NaiveDate), Reply> { + let pairs = parse_query(query).ok_or_else(|| Reply::error(400, "the query is not valid"))?; + let mut targets = Vec::new(); + let mut day = None; + for (key, value) in pairs { + match key.as_str() { + "repo" if !value.eq_ignore_ascii_case(&identity.repository) => { + return Err(Reply::error( + 403, + format!( + "the token is for {} but the request is for {value}", + identity.repository + ), + )); + } + "repo" => {} + "target" if value.starts_with("//") && value.contains(':') => { + if !targets.contains(&value) { + targets.push(value); + } + } + "target" => return Err(Reply::error(400, "target must be a target address")), + "day" => { + day = Some( + NaiveDate::parse_from_str(&value, "%Y-%m-%d") + .map_err(|_| Reply::error(400, "day must be a date (YYYY-MM-DD)"))?, + ); + } + other => return Err(Reply::error(400, format!("unknown parameter {other}"))), + } + } + if targets.is_empty() || targets.len() > MAX_TARGETS { + return Err(Reply::error( + 400, + format!("name from 1 to {MAX_TARGETS} targets"), + )); + } + let day = day.ok_or_else(|| Reply::error(400, "day is required"))?; + targets.sort(); + Ok((targets, day)) +} + +async fn read(app: &App, repo: &str, targets: &[String], day: NaiveDate) -> anyhow::Result { + let (from, to) = window(day); + let (first_day, last_day) = (from.date_naive(), to.date_naive()); + let client = app.pool.get().await?; + let rows = client + .query( + SQL, + &[ + &repo, + &targets, + &from, + &to, + &first_day, + &last_day, + &RUNS_PER_TARGET, + ], + ) + .await?; + let files: Vec = rows + .iter() + .map(|row| { + json!({ + "file": row.get::<_, String>("file"), + "p50_ms": row.get::<_, i64>("p50_ms"), + "runs": row.get::<_, i64>("runs"), + }) + }) + .collect(); + Ok(json!({ + "repo": repo, + "targets": targets, + "day": day.to_string(), + "files": files, + })) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn the_window_ends_before_the_day_begins() { + let day = NaiveDate::from_ymd_opt(2026, 10, 15).unwrap(); + let (from, to) = window(day); + assert_eq!(to.to_rfc3339(), "2026-10-14T23:45:00+00:00"); + assert_eq!(from.to_rfc3339(), "2026-10-07T23:45:00+00:00"); + } + + #[test] + fn parses_targets_and_day() { + let identity = Identity { + repository: "ArchAstro/firstlanding".into(), + }; + let (targets, day) = parse( + &identity, + Some("target=%2F%2Fb%3Atest&target=%2F%2Fa%3Atest&day=2026-10-15&repo=archastro%2Ffirstlanding"), + ) + .unwrap(); + assert_eq!(targets, ["//a:test", "//b:test"]); + assert_eq!(day.to_string(), "2026-10-15"); + for bad in [ + "day=2026-10-15", + "target=%2F%2Fa%3Atest", + "target=a&day=2026-10-15", + "target=%2F%2Fa%3Atest&day=yesterday", + "target=%2F%2Fa%3Atest&day=2026-10-15&limit=3", + ] { + assert_eq!( + parse(&identity, Some(bad)).unwrap_err().status, + 400, + "{bad}" + ); + } + let other = "target=%2F%2Fa%3Atest&day=2026-10-15&repo=ArchAstro%2Fother"; + assert_eq!(parse(&identity, Some(other)).unwrap_err().status, 403); + } +} diff --git a/crates/aster-srv/src/test_results/mod.rs b/crates/aster-srv/src/test_results/mod.rs index 4ac1e20..ae9b6a0 100644 --- a/crates/aster-srv/src/test_results/mod.rs +++ b/crates/aster-srv/src/test_results/mod.rs @@ -10,6 +10,7 @@ //! classifier's evidence. pub mod classifier; +pub mod durations; pub mod ingest; pub mod query; pub mod retention; diff --git a/crates/aster-srv/tests/durations.rs b/crates/aster-srv/tests/durations.rs new file mode 100644 index 0000000..b31a6c5 --- /dev/null +++ b/crates/aster-srv/tests/durations.rs @@ -0,0 +1,259 @@ +//! `GET /v1/test-results/durations` against a real Postgres. + +mod common; + +use aster_srv::test_results::durations::window; +use chrono::{DateTime, Duration, NaiveDate, Utc}; +use common::{start, Server, REPO}; +use serde_json::json; + +/// A run of `target` received at `at` whose tests are `(file, test, ms, +/// attempt)`. +async fn seed( + server: &Server, + run_id: &str, + target: &str, + at: DateTime, + cached: bool, + confidence: &str, + tests: &[(&str, &str, i64, i32)], +) { + let client = server.client().await; + let day = at.date_naive(); + aster_srv::test_results::ensure_partitions(&client, day) + .await + .unwrap(); + client + .execute( + "INSERT INTO test_results.runs (run_id, repo, sha, aster_version, started_at, received_at) \ + VALUES ($1, $2, 'sha', '0', $3, $3)", + &[&run_id, &REPO, &at], + ) + .await + .unwrap(); + client + .execute( + "INSERT INTO test_results.target_runs \ + (received_on, run_id, repo, address, status, cached, duration_ms, confidence) \ + VALUES ($1, $2, $3, $4, 'passed', $5, 1, $6)", + &[&day, &run_id, &REPO, &target, &cached, &confidence], + ) + .await + .unwrap(); + for (file, test, ms, attempt) in tests { + let test_id = format!("{file}#{test}"); + client + .execute( + "INSERT INTO test_results.test_results \ + (received_on, run_id, repo, target, test_id, outcome, attempt, duration_ms) \ + VALUES ($1, $2, $3, $4, $5, 'pass', $6, $7)", + &[&day, &run_id, &REPO, &target, &test_id, attempt, ms], + ) + .await + .unwrap(); + client + .execute( + "INSERT INTO test_results.tests (repo, test_id, target, file, name, first_seen, last_seen) \ + VALUES ($1, $2, $3, $4, $5, $6, $6) ON CONFLICT DO NOTHING", + &[&REPO, &test_id, &target, file, test, &at], + ) + .await + .unwrap(); + } +} + +const QUERY: &str = + "/v1/test-results/durations?target=%2F%2Fshard-1%3Atest&target=%2F%2Fshard-2%3Atest"; + +#[tokio::test] +async fn medians_each_file_over_the_shards_runs_in_the_window() { + let Some(server) = start(&[]).await else { + return; + }; + let today = Utc::now().date_naive(); + let (from, to) = window(today); + let inside = to - Duration::hours(1); + // a.exs ran on shard 1 (two tests: 10+20, then 30+40) and moved to shard + // 2 (50+60): per run 30, 70, 110, median 70. The retry attempt is left out. + seed( + &server, + "r1", + "//shard-1:test", + inside, + false, + "complete", + &[ + ("test/a_test.exs", "x", 10, 1), + ("test/a_test.exs", "y", 20, 1), + ("test/a_test.exs", "y", 999, 2), + ], + ) + .await; + seed( + &server, + "r2", + "//shard-1:test", + inside - Duration::hours(1), + false, + "complete", + &[ + ("test/a_test.exs", "x", 30, 1), + ("test/a_test.exs", "y", 40, 1), + ("test/b_test.exs", "z", 5, 1), + ], + ) + .await; + seed( + &server, + "r3", + "//shard-2:test", + inside, + false, + "complete", + &[ + ("test/a_test.exs", "x", 50, 1), + ("test/a_test.exs", "y", 60, 1), + ], + ) + .await; + // Left out: cached, partial, another target, before and after the window. + seed( + &server, + "c", + "//shard-1:test", + inside, + true, + "complete", + &[("test/c_test.exs", "c", 1, 1)], + ) + .await; + seed( + &server, + "p", + "//shard-1:test", + inside, + false, + "partial", + &[("test/c_test.exs", "c", 1, 1)], + ) + .await; + seed( + &server, + "o", + "//other:test", + inside, + false, + "complete", + &[("test/c_test.exs", "c", 1, 1)], + ) + .await; + seed( + &server, + "old", + "//shard-1:test", + from - Duration::minutes(1), + false, + "complete", + &[("test/c_test.exs", "c", 1, 1)], + ) + .await; + seed( + &server, + "new", + "//shard-1:test", + to, + false, + "complete", + &[("test/c_test.exs", "c", 1, 1)], + ) + .await; + + let (status, body) = server + .get(&format!("{QUERY}&day={today}"), Some(REPO)) + .await; + assert_eq!(status, 200, "{body}"); + assert_eq!(body["targets"], json!(["//shard-1:test", "//shard-2:test"])); + assert_eq!( + body["files"], + json!([ + {"file": "test/a_test.exs", "p50_ms": 70, "runs": 3}, + {"file": "test/b_test.exs", "p50_ms": 5, "runs": 1}, + ]) + ); + + // The same question asked later gets the same answer. + seed( + &server, + "later", + "//shard-2:test", + Utc::now(), + false, + "complete", + &[("test/a_test.exs", "x", 1, 1)], + ) + .await; + let (_, again) = server + .get(&format!("{QUERY}&day={today}"), Some(REPO)) + .await; + assert_eq!(again, body); + + // A day that has not begun is refused; another repository is forbidden. + let tomorrow = today.succ_opt().unwrap(); + let (status, _) = server + .get(&format!("{QUERY}&day={tomorrow}"), Some(REPO)) + .await; + assert_eq!(status, 409); + let (status, _) = server + .get( + &format!("{QUERY}&day={today}&repo=ArchAstro%2Fother"), + Some(REPO), + ) + .await; + assert_eq!(status, 403); + let (status, _) = server.get(&format!("{QUERY}&day={today}"), None).await; + assert_eq!(status, 401); + let (status, empty) = server + .get( + &format!( + "{QUERY}&day={}", + NaiveDate::from_ymd_opt(2020, 1, 1).unwrap() + ), + Some(REPO), + ) + .await; + assert_eq!((status, &empty["files"]), (200, &json!([]))); + server.stop().await; +} + +#[tokio::test] +async fn keeps_only_each_targets_latest_runs() { + let Some(server) = start(&[]).await else { + return; + }; + let today = Utc::now().date_naive(); + let (_, to) = window(today); + // 25 runs: the 5 oldest take 1000 ms, the 20 newest 10 ms. + for i in 0..25 { + let ms = if i < 5 { 1000 } else { 10 }; + let at = to - Duration::minutes(60 - i); + seed( + &server, + &format!("r{i:02}"), + "//shard-1:test", + at, + false, + "complete", + &[("test/a_test.exs", "x", ms, 1)], + ) + .await; + } + let (status, body) = server + .get(&format!("{QUERY}&day={today}"), Some(REPO)) + .await; + assert_eq!(status, 200, "{body}"); + assert_eq!( + body["files"], + json!([{"file": "test/a_test.exs", "p50_ms": 10, "runs": 20}]) + ); + server.stop().await; +} From 0befcae100d8871a2325285826066959f6dc5cb8 Mon Sep 17 00:00:00 2001 From: calvin-archastro Date: Thu, 15 Oct 2026 09:41:39 -0400 Subject: [PATCH 2/2] feat(results): split mix test partitions by recorded duration A `MIX_TEST_PARTITION=K mix test --partitions N` target (a firstlanding shard) now runs an explicit file list: the files Mix would deal out, split longest-first by their median durations from aster-srv as of the commit's UTC day, so every shard reaches the same split. No server, no history, an error, an unknowable file set or ASTER_BALANCE_PARTITIONS=off keep the command as written. MIX_TEST_PARTITION stays set for per-partition databases; retry and result capture work on the rewritten command. Tests a shard runs for another project (`cd ../platform && mix test`) are now identified by that project, so a test keeps its id whichever shard runs it. This changes the ids of shard tests once. Co-Authored-By: Claude Opus 5.5 (1M context) --- crates/aster-srv/README.md | 31 +++ docs/specs/test-tracking.html | 12 +- src/cli/output.rs | 1 + src/executor/runner.rs | 25 ++ src/executor/runner/partitions.rs | 325 ++++++++++++++++++++++++++ src/executor/runner/retry.rs | 3 + src/plugins/mix_partitions.rs | 369 ++++++++++++++++++++++++++++++ src/plugins/mod.rs | 1 + src/plugins/test_reports.rs | 37 +++ src/plugins/test_retry.rs | 17 ++ src/test_results/balance.rs | 298 ++++++++++++++++++++++++ src/test_results/client.rs | 12 +- src/test_results/mod.rs | 7 + src/test_results/stream.rs | 39 +++- 14 files changed, 1166 insertions(+), 11 deletions(-) create mode 100644 src/executor/runner/partitions.rs create mode 100644 src/plugins/mix_partitions.rs create mode 100644 src/test_results/balance.rs diff --git a/crates/aster-srv/README.md b/crates/aster-srv/README.md index eb8ee0f..5652e16 100644 --- a/crates/aster-srv/README.md +++ b/crates/aster-srv/README.md @@ -21,6 +21,7 @@ from this server's state, looked up with the same credentials. | `GET /v1/test-results/tests/:test_id` | One test's state and history. | | `GET /v1/test-results/runs/:run_id/failures` | The failing tests of a run, with their state. | | `GET /v1/test-results/failures?github_run=` | The failing tests of every Aster run in one GitHub Actions run. | +| `GET /v1/test-results/durations?target=&day=` | Each test file's median duration, for splitting a partitioned suite. | | `GET /healthz` | The process is up. | | `GET /readyz` | The database answers. | @@ -177,6 +178,36 @@ name. `aster results failures --github-run ` fetches every page and prints the merged JSON. +### Durations + +`GET /v1/test-results/durations?target=
&target=…&day=` +returns each test file's median duration over the targets' recent runs, for +splitting a partitioned suite (`MIX_TEST_PARTITION=K mix test --partitions N`) +by time. Every partition of one commit asks with the same targets and the +commit's UTC `day`, and must get the same answer, so the answer depends on +nothing else: it covers runs received in the 7 days before `day` began (less +15 minutes for uploads still committing), and a `day` that has not begun is +`409`. Per file, the first-attempt durations of its tests are summed in each +of the latest 20 uncached, completely captured runs of each target, and the +median of those sums is `p50_ms`. 1 to 32 targets; `files` is sorted and +empty without history. + +```json +{ + "repo": "ArchAstro/firstlanding", + "targets": ["//services/platform-test-shard-1:test-ci", "…"], + "day": "2026-10-15", + "files": [ { "file": "test/platform/agents_test.exs", "p50_ms": 18234, "runs": 60 } ] +} +``` + +Aster asks when a target runs a partition of a suite and `ASTER_SRV_URL` is +set, then runs the files that partition gets from a longest-first split +(`ASTER_BALANCE_PARTITIONS=off` keeps Mix's own split; so does any error). +The shards' tests are recorded under the project whose suite it is (the +directory the command `cd`s to), so a test's id does not depend on the shard +that ran it. + ## Authentication CI authenticates with its GitHub Actions OIDC token for the `aster-srv` diff --git a/docs/specs/test-tracking.html b/docs/specs/test-tracking.html index 52098f9..d03a45e 100644 --- a/docs/specs/test-tracking.html +++ b/docs/specs/test-tracking.html @@ -309,7 +309,7 @@

Aster Test Tracking

RetentionA daily job drops partitions older than 30 days. A GCS lifecycle rule deletes raw objects at 30 days. tests rows not seen for 30 days are deleted. DeployImage ghcr.io/archastro/aster-srv from this repo's release workflow. Pulumi stack in firstlanding/infra/services/aster-srv: Deployment, Cloud SQL, bucket, Workload Identity. NotificationsConfigured webhook endpoints, each with its own subscribed event types: test.quarantined, test.released, test.expired, test.broken. The JSON payload is signed with HMAC-SHA256 (X-Aster-Signature) and retried with backoff. Slack, issue trackers and other receivers live outside the service. - RoutesPOST /v1/test-results/runs · GET /v1/test-results/quarantine?repo= · GET /v1/test-results/tests?state=flaky|broken|quarantined · GET /v1/test-results/tests/:id · GET /v1/test-results/runs/:run_id/failures · /healthz + RoutesPOST /v1/test-results/runs · GET /v1/test-results/quarantine?repo= · GET /v1/test-results/tests?state=flaky|broken|quarantined · GET /v1/test-results/tests/:id · GET /v1/test-results/runs/:run_id/failures · GET /v1/test-results/durations?target=&day= · /healthz @@ -352,7 +352,7 @@

Aster Test Tracking

{ id: "P1", name: "Capture + ingest", note: "Per-test records uploaded from every firstlanding test-ci job." }, { id: "P2", name: "Retry + classify", note: "Per-test retry replaces retry_failed_targets.py, and flakes are scored." }, { id: "P3", name: "Reporting", note: "PR runs say which failures are known flakes and which are new, and a dashboard shows the history." }, - { id: "P4", name: "Quarantine + sharding", note: "Known flakes stop failing builds, and lanes are balanced by duration." } + { id: "P4", name: "Quarantine + sharding", note: "Known flakes stop failing builds, and partitioned suites are split by duration." } ]; const T = [ { id: "T01", p: "P1", area: "client", title: "Test result model + NDJSON writer", sub: "crates/aster-results-model, shared by the CLI and the service", @@ -454,10 +454,10 @@

Aster Test Tracking

obj: "Config sets a list of webhook endpoints, each with a URL, a secret and the event types it receives (test.quarantined, test.released, test.expired, test.broken). Deliveries are queued in Postgres and retried with exponential backoff.", acc: ["Payload signed as X-Aster-Signature: sha256=", "A failing endpoint never blocks ingest or classification", "Delivery log kept 30 days", "No endpoints configured means nothing is sent"], files: ["crates/aster-srv/src/webhooks.rs", "crates/aster-srv/src/config.rs"], deps: ["T22"] }, - { id: "T24", p: "P4", area: "client", title: "Duration-based lane balancing", sub: "Uses p50 test durations", - obj: "--lane and mix --partitions split by recorded duration instead of file count, using the duration data from the service.", - acc: ["Slowest lane's wall time drops on firstlanding (measured)", "Falls back to the current split when the service is unavailable"], - files: ["src/git/lane.rs", "src/plugins/elixir.rs"], deps: ["T17"] }, + { id: "T24", p: "P4", area: "client", title: "Duration-balanced mix partitions", sub: "p50 file durations, longest first", + obj: "A MIX_TEST_PARTITION=K mix test --partitions N target (the firstlanding shard projects) runs an explicit file list: Mix's file set, split longest-first by each file's median duration from GET /v1/test-results/durations, as of the commit's UTC day so every shard gets the same answer. Lanes stay configured selectors. Shard tests are recorded under the project that owns them (the cd ../platform target), so a test keeps its id on any shard.", + acc: ["Every shard of one commit gets the same split: no file runs on no shard or on two", "Falls back to Mix's own split, unchanged, with no ASTER_SRV_URL, no history, an error, or ASTER_BALANCE_PARTITIONS=off", "Retry and result capture work on the rewritten command", "Slowest partition's time drops on firstlanding (measured after rollout; logs show up to ~20% headroom)"], + files: ["src/plugins/mix_partitions.rs", "src/test_results/balance.rs", "src/executor/runner/partitions.rs", "crates/aster-srv/src/test_results/durations.rs"], deps: ["T17"] }, { id: "T25", p: "P4", area: "docs", title: "Docs", sub: "README, GETTING_STARTED, service runbook", obj: "Document capture per runner, the env vars (ASTER_SRV_URL, ASTER_QUARANTINE), the quarantine rules and self-hosting.", acc: ["Doc examples run against a fixture workspace"], diff --git a/src/cli/output.rs b/src/cli/output.rs index e5d78f7..5282490 100644 --- a/src/cli/output.rs +++ b/src/cli/output.rs @@ -503,6 +503,7 @@ mod tests { duration_ms: 1, exit_code: Some(1), tests: Some(TestReport { + project: None, capture: "junit".to_string(), confidence: Confidence::Partial, cases: vec![ diff --git a/src/executor/runner.rs b/src/executor/runner.rs index 07a3339..d282f95 100644 --- a/src/executor/runner.rs +++ b/src/executor/runner.rs @@ -36,6 +36,7 @@ use crate::test_results::{ }; use crate::ui::ProgressDisplay; +mod partitions; mod retry; use super::command::parse_command; @@ -792,6 +793,26 @@ impl<'a> Executor<'a> { .get(&target_name) .and_then(|t| t.working_dir.clone()) .unwrap_or_else(|| project_root.clone()); + // A partition of a suite runs the share recorded durations give + // it. Like test capture, this rewrites the commands only after + // the cache key and input hash were computed from them as + // written. + let commands = match partitions::balance( + target_addr, + &commands, + &working_dir, + self.workspace_root, + project_map, + &|name| std::env::var(name).ok(), + ) { + Some(balanced) => vec![balanced], + None => commands, + }; + // From the command as configured, so an override that narrows + // it never changes whose tests these are. + let test_project = project.targets.get(&target_name).and_then(|target| { + partitions::test_owner(project, target, &target.command, project_map) + }); // Test capture rewrites the commands only now, after the cache // key was computed from the commands as written. let captures = if self.capture_tests || self.retry_failed > 0 { @@ -913,6 +934,9 @@ impl<'a> Executor<'a> { ); } } + if let Some(report) = &mut result.tests { + report.project = test_project; + } if let Some(report) = &result.tests { if std::env::var("GITHUB_ACTIONS").as_deref() == Ok("true") { for line in @@ -1733,6 +1757,7 @@ fn read_test_report( let _ = std::fs::remove_dir_all(dir); } Some(TestReport { + project: None, capture: names.join("+"), confidence: if complete { Confidence::Complete diff --git a/src/executor/runner/partitions.rs b/src/executor/runner/partitions.rs new file mode 100644 index 0000000..49d8c65 --- /dev/null +++ b/src/executor/runner/partitions.rs @@ -0,0 +1,325 @@ +//! Partitioned test suites: whose tests a shard runs, and which of them. +//! +//! A shard project (`//services/platform-test-shard-2`) runs one partition +//! of another project's suite. Its tests are recorded under the project +//! that owns them, so a test keeps its identity whichever shard runs it. +//! And when aster-srv knows how long each file takes, the shard runs a +//! share picked by time instead of the one Mix deals out by count. + +use std::collections::HashMap; +use std::path::Path; + +use crate::discovery::DiscoveredProject; +use crate::plugins::mix_partitions::{self, PartitionedMix}; +use crate::test_results::balance; + +/// `ASTER_BALANCE_PARTITIONS=off` keeps Mix's own split. +pub(super) const OFF_SWITCH: &str = "ASTER_BALANCE_PARTITIONS"; + +/// The address of the project whose tests `command` runs, when that is not +/// `project` (a `cd ../platform && mix test` one-liner). +pub(super) fn test_owner( + project: &DiscoveredProject, + target: &crate::plugins::Target, + command: &str, + project_map: &HashMap, +) -> Option { + let owner = super::tests_run_elsewhere(project, target, command, project_map)??; + project_map + .iter() + .find(|(_, candidate)| std::ptr::eq(**candidate, owner)) + .map(|(address, _)| address.clone()) +} + +/// The targets that run a partition of the same suite as `found`, `found`'s +/// own included, sorted. +fn peers(found: &PartitionedMix, project_map: &HashMap) -> Vec { + let dir = canonical(&found.dir); + let dir = &dir; + let mut peers: Vec = project_map + .iter() + .flat_map(|(address, project)| { + project.targets.iter().filter_map(move |(name, target)| { + let working_dir = target.working_dir.as_deref().unwrap_or(&project.root); + let other = mix_partitions::parse(working_dir, &target.command)?; + (other.total == found.total && canonical(&other.dir) == *dir) + .then(|| format!("{address}:{name}")) + }) + }) + .collect(); + peers.sort(); + peers +} + +fn canonical(path: &Path) -> std::path::PathBuf { + std::fs::canonicalize(path).unwrap_or_else(|_| path.to_path_buf()) +} + +/// The command that runs `address`'s share of a partitioned suite by +/// recorded duration, or `None` to run `commands` as written. Says which +/// split it used. +pub(super) fn balance( + address: &str, + commands: &[String], + working_dir: &Path, + workspace_root: &Path, + project_map: &HashMap, + env: &dyn Fn(&str) -> Option, +) -> Option { + let [command] = commands else { + return None; + }; + let found = mix_partitions::parse(working_dir, command)?; + let label = format!( + "[aster] {address}: partition {}/{}", + found.partition, found.total + ); + let keep = |why: &str| { + eprintln!("{label} split by Mix ({why})"); + None + }; + if env(OFF_SWITCH).is_some_and(|value| value.eq_ignore_ascii_case("off")) { + return keep(&format!("{OFF_SWITCH}=off")); + } + if env("ASTER_SRV_URL").is_none_or(|url| url.is_empty()) { + return keep("no ASTER_SRV_URL for durations"); + } + let Some(files) = found.files() else { + return keep("cannot list its files for sure"); + }; + if files.len() < found.total { + return keep("fewer files than partitions"); + } + let peers = peers(&found, project_map); + if !peers.iter().any(|peer| peer == address) { + return keep("this target is not one of the suite's partitions"); + } + if peers.len() > crate::test_results::balance::MAX_TARGETS { + return keep("too many partition targets"); + } + let Some(day) = balance::commit_day(workspace_root) else { + return keep("no commit date"); + }; + let durations = match balance::fetch(workspace_root, &peers, day, env) { + Ok(Some(durations)) => durations, + Ok(None) => return keep("no ASTER_SRV_URL for durations"), + Err(error) => return keep(&format!("durations unavailable: {error:#}")), + }; + let known = files.iter().filter(|f| durations.contains_key(*f)).count(); + if known == 0 { + return keep("no recorded durations"); + } + let bins = balance::split(&files, &durations, found.total); + let mine = &bins[found.partition - 1]; + let expected = balance::expected_ms(mine, &files, &durations); + let slowest = bins + .iter() + .map(|bin| balance::expected_ms(bin, &files, &durations)) + .max() + .unwrap_or(0); + eprintln!( + "{label} balanced by duration: {} of {} files, ~{}s of a slowest ~{}s \ + ({known} files timed, medians before {day})", + mine.len(), + files.len(), + expected / 1000, + slowest / 1000, + ); + Some(found.render(mine)) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::plugins::{ProjectMetadata, Target}; + + fn project(root: std::path::PathBuf, targets: HashMap) -> DiscoveredProject { + DiscoveredProject { + config_path: root.join("mix.exs"), + root, + metadata: ProjectMetadata { + name: "p".to_string(), + version: None, + }, + dependencies: vec![], + targets, + plugin_name: "elixir".to_string(), + languages: vec![], + build_system: None, + relative_path: std::path::PathBuf::new(), + } + } + + fn shard(root: &Path, partition: usize) -> (String, DiscoveredProject) { + let dir = root.join(format!("shard-{partition}")); + std::fs::create_dir_all(&dir).unwrap(); + let mut targets = HashMap::new(); + targets.insert( + "test-ci".to_string(), + Target { + command: format!( + "bash -c 'cd ../app && MIX_TEST_PARTITION={partition} exec mix test --partitions 2'" + ), + ..Default::default() + }, + ); + (format!("//shard-{partition}"), project(dir, targets)) + } + + #[test] + fn shards_name_their_suites_owner_and_find_each_other() { + let ws = tempfile::tempdir().unwrap(); + let app_dir = ws.path().join("app"); + std::fs::create_dir_all(app_dir.join("test")).unwrap(); + let app = project(app_dir, HashMap::new()); + let (a1, s1) = shard(ws.path(), 1); + let (a2, s2) = shard(ws.path(), 2); + let map: HashMap = [ + ("//app".to_string(), &app), + (a1.clone(), &s1), + (a2.clone(), &s2), + ] + .into_iter() + .collect(); + let target = &s1.targets["test-ci"]; + assert_eq!( + test_owner(&s1, target, &target.command, &map).as_deref(), + Some("//app") + ); + assert_eq!(test_owner(&app, target, "mix test", &map), None); + + let found = mix_partitions::parse(&s1.root, &target.command).unwrap(); + assert_eq!( + peers(&found, &map), + ["//shard-1:test-ci", "//shard-2:test-ci"] + ); + } + + #[test] + fn without_a_server_or_when_switched_off_mix_splits() { + let ws = tempfile::tempdir().unwrap(); + let (address, s1) = shard(ws.path(), 1); + let map: HashMap = + [(address.clone(), &s1)].into_iter().collect(); + let command = vec![s1.targets["test-ci"].command.clone()]; + let address = format!("{address}:test-ci"); + let none = |_: &str| None; + assert_eq!( + balance(&address, &command, &s1.root, ws.path(), &map, &none), + None + ); + let off = |name: &str| match name { + OFF_SWITCH => Some("off".to_string()), + "ASTER_SRV_URL" => Some("http://127.0.0.1:9".to_string()), + _ => None, + }; + assert_eq!( + balance(&address, &command, &s1.root, ws.path(), &map, &off), + None + ); + // Not a partitioned command at all. + let plain = vec!["mix test".to_string()]; + assert_eq!( + balance(&address, &plain, &s1.root, ws.path(), &map, &none), + None + ); + } + + /// Answers one request with `body`; returns the request line. + fn serve_once(body: &'static str) -> (String, std::thread::JoinHandle) { + use std::io::{BufRead, BufReader, Write}; + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let url = format!("http://{}", listener.local_addr().unwrap()); + let handle = std::thread::spawn(move || { + let (stream, _) = listener.accept().unwrap(); + let mut reader = BufReader::new(stream.try_clone().unwrap()); + let mut request_line = String::new(); + reader.read_line(&mut request_line).unwrap(); + loop { + let mut line = String::new(); + reader.read_line(&mut line).unwrap(); + if line == "\r\n" || line.is_empty() { + break; + } + } + let mut stream = stream; + write!( + stream, + "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}", + body.len() + ) + .unwrap(); + request_line + }); + (url, handle) + } + + #[test] + fn every_shard_takes_a_disjoint_share_of_the_suite_by_duration() { + let ws = tempfile::tempdir().unwrap(); + let vcs = |args: &[&str]| { + let status = std::process::Command::new("git") + .args(args) + .current_dir(ws.path()) + .env("GIT_COMMITTER_DATE", "2026-10-14T23:30:00-02:00") + .env("GIT_AUTHOR_DATE", "2026-10-14T23:30:00-02:00") + .status() + .unwrap(); + assert!(status.success()); + }; + vcs(&["init", "-q"]); + vcs(&[ + "-c", + "user.name=t", + "-c", + "user.email=t@t", + "commit", + "-q", + "--allow-empty", + "-m", + "x", + ]); + let app_dir = ws.path().join("app"); + std::fs::create_dir_all(app_dir.join("test")).unwrap(); + for file in ["a", "b", "c", "d"] { + std::fs::write(app_dir.join(format!("test/{file}_test.exs")), "").unwrap(); + } + std::fs::write(app_dir.join("mix.exs"), "[app: :app]").unwrap(); + let app = project(app_dir, HashMap::new()); + let (a1, s1) = shard(ws.path(), 1); + let (a2, s2) = shard(ws.path(), 2); + let map: HashMap = + [("//app".to_string(), &app), (a1, &s1), (a2, &s2)] + .into_iter() + .collect(); + let (url, server) = serve_once( + r#"{"files":[{"file":"test/a_test.exs","p50_ms":100,"runs":3},{"file":"test/b_test.exs","p50_ms":10,"runs":3},{"file":"test/c_test.exs","p50_ms":10,"runs":3},{"file":"test/d_test.exs","p50_ms":80,"runs":3},{"file":"test/gone_test.exs","p50_ms":999,"runs":3}]}"#, + ); + let env = move |name: &str| match name { + "ASTER_SRV_URL" => Some(url.clone()), + "ASTER_SRV_TOKEN" => Some("t".to_string()), + _ => None, + }; + let run = |shard: &DiscoveredProject, address: &str| { + let command = vec![shard.targets["test-ci"].command.clone()]; + balance(address, &command, &shard.root, ws.path(), &map, &env).unwrap() + }; + let first = run(&s1, "//shard-1:test-ci"); + // The second shard reads the first's cached answer: one request. + let second = run(&s2, "//shard-2:test-ci"); + let request = server.join().unwrap(); + assert!(request.contains("day=2026-10-15"), "{request}"); + assert!( + request.contains("target=%2F%2Fshard-1%3Atest-ci&target=%2F%2Fshard-2%3Atest-ci"), + "{request}" + ); + assert_eq!( + first, + "bash -c 'cd ../app && MIX_TEST_PARTITION=1 exec mix test test/a_test.exs'" + ); + assert_eq!( + second, + "bash -c 'cd ../app && MIX_TEST_PARTITION=2 exec mix test test/b_test.exs test/c_test.exs test/d_test.exs'" + ); + } +} diff --git a/src/executor/runner/retry.rs b/src/executor/runner/retry.rs index 57cc92e..506a65d 100644 --- a/src/executor/runner/retry.rs +++ b/src/executor/runner/retry.rs @@ -149,6 +149,7 @@ pub(super) fn rerun_failures(job: &RetryJob, result: &mut ExecutionResult) -> bo if ran_any { result.tests = Some(TestReport { + project: report.project, capture: report.capture, confidence, cases, @@ -315,6 +316,7 @@ mod tests { duration_ms: 10, exit_code, tests: Some(TestReport { + project: None, capture: "go-json".into(), confidence: Confidence::Complete, cases, @@ -427,6 +429,7 @@ mod tests { #[test] fn annotations_name_the_file_and_escape_the_message() { let report = TestReport { + project: None, capture: "go-json".into(), confidence: Confidence::Complete, cases: vec![ diff --git a/src/plugins/mix_partitions.rs b/src/plugins/mix_partitions.rs new file mode 100644 index 0000000..499bbb8 --- /dev/null +++ b/src/plugins/mix_partitions.rs @@ -0,0 +1,369 @@ +//! `MIX_TEST_PARTITION=K mix test --partitions N`, run as an explicit list +//! of test files instead. +//! +//! Mix deals the sorted test files out in turn, so a partition's share of +//! the suite's time is luck. Aster can instead pick partition K's files +//! itself (see [`crate::test_results::balance`]) and run them by name. This +//! module recognises such a command, lists the files Mix would deal out, +//! and writes the command that runs a chosen set of them. +//! +//! Every partition must list exactly the files Mix would, or a file runs on +//! no partition. So anything that changes Mix's own selection (custom +//! `test_paths` or `test_pattern`, umbrella apps, line filters, `--failed`, +//! `--stale`, a symlink under a test directory) makes [`parse`] or +//! [`PartitionedMix::files`] say `None`, and the command runs unchanged. +//! +//! `--partitions` is dropped but `MIX_TEST_PARTITION` is kept: projects +//! name per-partition databases after it, and without `--partitions` Mix +//! ignores it. + +use std::path::{Path, PathBuf}; + +use super::related_tests::{Argv, Delegated}; + +const MIX_VALUE_FLAGS: &[&str] = &[ + "--only", + "--exclude", + "--include", + "--seed", + "--max-failures", + "--partitions", + "--timeout", + "--formatter", + "--slowest", + "--max-cases", + "--cover-filter", + "--exit-status", +]; + +/// `mix.exs` settings that change which files `mix test` runs. +const SELECTION_SETTINGS: &[&str] = &[ + "test_paths", + "test_pattern", + "test_load_filters", + "test_ignore_filters", + "apps_path", +]; + +/// A partitioned `mix test`, possibly delegated (`bash -c 'cd ../core && +/// MIX_TEST_PARTITION=1 exec mix test --partitions 4'`). +pub(crate) struct PartitionedMix { + delegated: Option, + argv: Argv, + test: usize, + /// `N` of `--partitions N`. + pub(crate) total: usize, + /// `K` of `MIX_TEST_PARTITION=K`, from 1. + pub(crate) partition: usize, + /// Where `mix test` runs. + pub(crate) dir: PathBuf, +} + +/// The partitioned `mix test` that `command` runs in `working_dir`. +pub(crate) fn parse(working_dir: &Path, command: &str) -> Option { + let (delegated, inner) = match Argv::parse(command) { + Some(argv) => (None, argv), + None => { + let delegated = Delegated::parse(command)?; + let inner = Argv::parse(&delegated.command)?; + (Some(delegated), inner) + } + }; + let argv = inner; + if argv.parts[argv.program] != "mix" || argv.parts.iter().any(|p| p.ends_with(',')) { + return None; + } + let test = argv.program + + 1 + + argv.parts[argv.program + 1..] + .iter() + .position(|p| p == "test")?; + // `mix test` must be the task, not an argument of another one. + if test != argv.program + 1 { + return None; + } + let arguments = &argv.parts[test + 1..]; + if arguments + .iter() + .any(|p| p == "--failed" || p == "--stale" || p == "--") + { + return None; + } + let total: usize = arguments.iter().enumerate().find_map(|(i, part)| { + match part.strip_prefix("--partitions") { + Some("") => arguments.get(i + 1)?.parse().ok(), + Some(rest) => rest.strip_prefix('=')?.parse().ok(), + None => None, + } + })?; + let mut assignments: Vec = argv.parts[..argv.program].to_vec(); + if let Some(delegated) = &delegated { + assignments.extend( + delegated + .environment() + .split_whitespace() + .map(str::to_string), + ); + } + let partition: usize = assignments + .iter() + .rev() + .find_map(|part| part.strip_prefix("MIX_TEST_PARTITION="))? + .parse() + .ok()?; + if total < 2 || !(1..=total).contains(&partition) { + return None; + } + let dir = match delegated.as_ref().and_then(|d| d.dir.as_deref()) { + Some(dir) => working_dir.join(dir), + None => working_dir.to_path_buf(), + }; + Some(PartitionedMix { + delegated, + argv, + test, + total, + partition, + dir, + }) +} + +impl PartitionedMix { + fn listed(&self) -> Vec { + self.argv.positionals(self.test + 1, MIX_VALUE_FLAGS) + } + + /// Every file the partitions deal out, sorted, relative to [`Self::dir`] + /// as ExUnit names them. `None` when that cannot be known for sure. + pub(crate) fn files(&self) -> Option> { + let mix_exs = std::fs::read_to_string(self.dir.join("mix.exs")).ok()?; + if SELECTION_SETTINGS.iter().any(|s| mix_exs.contains(s)) { + return None; + } + let listed: Vec = self + .listed() + .into_iter() + .map(|i| self.argv.parts[i].clone()) + .collect(); + let roots = if listed.is_empty() { + vec!["test".to_string()] + } else { + listed + }; + let mut files = Vec::new(); + for root in roots { + let root = root.trim_start_matches("./").trim_end_matches('/'); + if root.is_empty() + || root == "." + || root.contains(':') + || root.starts_with('/') + || root.split('/').any(|part| part == "..") + { + return None; + } + let path = self.dir.join(root); + let meta = std::fs::symlink_metadata(&path).ok()?; + if meta.is_file() { + if !root.ends_with(".exs") { + return None; + } + files.push(root.to_string()); + } else if meta.is_dir() { + // Path.wildcard skips hidden entries. + let entries = walkdir::WalkDir::new(&path) + .sort_by_file_name() + .into_iter() + .filter_entry(|e| { + e.depth() == 0 || !e.file_name().to_string_lossy().starts_with('.') + }); + for entry in entries { + let entry = entry.ok()?; + if entry.path_is_symlink() { + return None; + } + let name = entry.file_name().to_string_lossy(); + let relative = entry.path().strip_prefix(&self.dir).ok()?; + let relative = relative.to_str()?.replace('\\', "/"); + if entry.file_type().is_file() && name.ends_with("_test.exs") { + files.push(relative); + } + } + } else { + return None; + } + } + files.sort(); + files.dedup(); + Some(files) + } + + /// The command that runs `files` (relative to [`Self::dir`]) without + /// `--partitions`, otherwise as written. + pub(crate) fn render(&self, files: &[String]) -> String { + let mut drop = self.listed(); + let parts = &self.argv.parts; + let mut i = self.test + 1; + while i < parts.len() { + if parts[i] == "--partitions" { + drop.extend([i, i + 1]); + i += 1; + } else if parts[i].starts_with("--partitions=") { + drop.push(i); + } + i += 1; + } + let inner = self.argv.render(&drop, files); + match &self.delegated { + Some(delegated) => delegated.render(&inner), + None => inner, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn project(files: &[&str], mix_exs: &str) -> tempfile::TempDir { + let dir = tempfile::tempdir().unwrap(); + std::fs::write(dir.path().join("mix.exs"), mix_exs).unwrap(); + for file in files { + let path = dir.path().join(file); + std::fs::create_dir_all(path.parent().unwrap()).unwrap(); + std::fs::write(path, "").unwrap(); + } + dir + } + + const MIX_EXS: &str = "defmodule P.MixProject do\n def project, do: [app: :p]\nend\n"; + + #[test] + fn reads_the_partition_from_a_delegated_one_liner() { + let ws = tempfile::tempdir().unwrap(); + let shard = ws.path().join("shard"); + let found = parse( + &shard, + "bash -c 'cd ../platform && MIX_TEST_PARTITION=3 exec mix test --warnings-as-errors --partitions 4'", + ) + .unwrap(); + assert_eq!((found.partition, found.total), (3, 4)); + assert_eq!(found.dir, shard.join("../platform")); + assert_eq!( + found.render(&["test/a_test.exs".into(), "test/b c_test.exs".into()]), + "bash -c 'cd ../platform && MIX_TEST_PARTITION=3 exec mix test --warnings-as-errors test/a_test.exs '\\''test/b c_test.exs'\\'''" + ); + + let plain = parse( + ws.path(), + "MIX_TEST_PARTITION=1 mix test --partitions=2 --seed 0", + ) + .unwrap(); + assert_eq!((plain.partition, plain.total), (1, 2)); + assert_eq!( + plain.render(&["test/a_test.exs".into()]), + "MIX_TEST_PARTITION=1 mix test --seed 0 test/a_test.exs" + ); + } + + #[test] + fn leaves_other_commands_alone() { + let dir = Path::new("/ws"); + for command in [ + "mix test", + "mix test --partitions 4", + "MIX_TEST_PARTITION=1 mix test", + "MIX_TEST_PARTITION=5 mix test --partitions 4", + "MIX_TEST_PARTITION=1 mix test --partitions 1", + "MIX_TEST_PARTITION=1 mix test --partitions 2 --failed", + "MIX_TEST_PARTITION=1 mix test --partitions 2 --stale", + "MIX_TEST_PARTITION=1 mix do compile, test --partitions 2", + "MIX_TEST_PARTITION=1 mix coveralls test --partitions 2", + "MIX_TEST_PARTITION=1 go test --partitions 2", + ] { + assert!(parse(dir, command).is_none(), "{command}"); + } + } + + #[test] + fn lists_the_files_mix_would_deal_out() { + let dir = project( + &[ + "test/b_test.exs", + "test/a/z_test.exs", + "test/support/helper.exs", + "test/test_helper.exs", + "test/.hidden/x_test.exs", + "test/.y_test.exs", + "other/q_test.exs", + ], + MIX_EXS, + ); + let all = parse(dir.path(), "MIX_TEST_PARTITION=1 mix test --partitions 2").unwrap(); + assert_eq!( + all.files().unwrap(), + ["test/a/z_test.exs", "test/b_test.exs"] + ); + let listed = parse( + dir.path(), + "MIX_TEST_PARTITION=1 mix test --partitions 2 --only slow ./test/a/ other/q_test.exs", + ) + .unwrap(); + assert_eq!( + listed.files().unwrap(), + ["other/q_test.exs", "test/a/z_test.exs"] + ); + assert_eq!( + listed.render(&["test/a/z_test.exs".into()]), + "MIX_TEST_PARTITION=1 mix test --only slow test/a/z_test.exs" + ); + } + + #[test] + fn unknowable_selections_are_none() { + let custom = project(&["test/a_test.exs"], "[test_paths: [\"t\"]]"); + let found = parse( + custom.path(), + "MIX_TEST_PARTITION=1 mix test --partitions 2", + ) + .unwrap(); + assert!(found.files().is_none()); + + let dir = project(&["test/a_test.exs"], MIX_EXS); + for listed in [ + "test/a_test.exs:3", + "../x", + "/abs", + "missing", + "test/a_test.ex", + ] { + let command = format!("MIX_TEST_PARTITION=1 mix test --partitions 2 {listed}"); + let found = parse(dir.path(), &command).unwrap(); + assert!(found.files().is_none(), "{listed}"); + } + + #[cfg(unix)] + { + let linked = project(&["elsewhere/x_test.exs"], MIX_EXS); + std::fs::create_dir_all(linked.path().join("test")).unwrap(); + std::os::unix::fs::symlink( + linked.path().join("elsewhere"), + linked.path().join("test/link"), + ) + .unwrap(); + let found = parse( + linked.path(), + "MIX_TEST_PARTITION=1 mix test --partitions 2", + ) + .unwrap(); + assert!(found.files().is_none()); + } + + let no_mix = tempfile::tempdir().unwrap(); + let found = parse( + no_mix.path(), + "MIX_TEST_PARTITION=1 mix test --partitions 2", + ) + .unwrap(); + assert!(found.files().is_none()); + } +} diff --git a/src/plugins/mod.rs b/src/plugins/mod.rs index b7caabf..e35aea7 100644 --- a/src/plugins/mod.rs +++ b/src/plugins/mod.rs @@ -408,6 +408,7 @@ pub mod go; pub mod gradle; mod jvm; pub mod maven; +pub(crate) mod mix_partitions; pub mod nodejs; pub mod python; pub mod registry; diff --git a/src/plugins/test_reports.rs b/src/plugins/test_reports.rs index ce3d748..52d2452 100644 --- a/src/plugins/test_reports.rs +++ b/src/plugins/test_reports.rs @@ -893,6 +893,43 @@ test("parent", async (t) => { await t.test("child", () => {}); }); assert_eq!(plan.files, ReportFiles::Stdout); } + #[test] + fn a_balanced_partition_is_captured_like_the_partition_it_replaces() { + let tmp = tempfile::tempdir().unwrap(); + let shard = tmp.path().join("platform-test-shard-2"); + let platform = tmp.path().join("platform"); + std::fs::create_dir_all(&shard).unwrap(); + std::fs::create_dir_all(platform.join("test")).unwrap(); + std::fs::write(platform.join("test/test_helper.exs"), "ExUnit.start()\n").unwrap(); + let report_dir = tmp.path().join("reports"); + std::fs::create_dir_all(&report_dir).unwrap(); + let original = + "bash -c 'cd ../platform && MIX_TEST_PARTITION=2 exec mix test --partitions 4'"; + let command = crate::plugins::mix_partitions::parse(&shard, original) + .unwrap() + .render(&["test/a_test.exs".into(), "test/b_test.exs".into()]); + let ctx = ReportContext { + project_dir: &shard, + working_dir: &shard, + target_name: "test-ci", + command: &command, + report_dir: &report_dir, + }; + let plan = delegating(&ctx, mix).unwrap(); + let script = shell_words::split(&plan.command).unwrap().pop().unwrap(); + assert!( + script.starts_with("cd ../platform && MIX_TEST_PARTITION=2 exec elixir -r "), + "{script}" + ); + assert!( + script.ends_with( + "-S mix test --formatter AsterTestResults.ExUnitFormatter --formatter ExUnit.CLIFormatter test/a_test.exs test/b_test.exs" + ), + "{script}" + ); + assert_eq!(plan.confidence, Confidence::Complete); + } + fn js_fixture(manifest: &str, files: &[(&str, &str)]) -> (tempfile::TempDir, Fixture) { let tmp = tempfile::tempdir().unwrap(); std::fs::write(tmp.path().join("package.json"), manifest).unwrap(); diff --git a/src/plugins/test_retry.rs b/src/plugins/test_retry.rs index e8ec795..2b166c7 100644 --- a/src/plugins/test_retry.rs +++ b/src/plugins/test_retry.rs @@ -725,6 +725,23 @@ mod tests { assert_eq!(script, "cd ../core && exec mix test test/a_test.exs"); } + #[test] + fn mix_retries_a_balanced_partition_with_only_its_failing_files() { + let dir = Path::new("/w/shard"); + let original = + "bash -c 'cd ../core && MIX_TEST_PARTITION=2 exec mix test --seed 0 --partitions 4'"; + let balanced = crate::plugins::mix_partitions::parse(dir, original) + .unwrap() + .render(&["test/a_test.exs".into(), "test/b_test.exs".into()]); + let failures = [failure(Some("test/b_test.exs"), None, "test one")]; + let found = one(dir, &balanced, &failures); + let script = shell_words::split(&found).unwrap().pop().unwrap(); + assert_eq!( + script, + "cd ../core && exec mix test --seed 0 test/b_test.exs" + ); + } + #[test] fn go_selects_tests_per_package() { let dir = Path::new("/w"); diff --git a/src/test_results/balance.rs b/src/test_results/balance.rs new file mode 100644 index 0000000..e6aaa12 --- /dev/null +++ b/src/test_results/balance.rs @@ -0,0 +1,298 @@ +//! Dealing a partitioned test suite out by recorded duration. +//! +//! Each partition of a suite runs on its own runner and works out its own +//! share, so every partition must reach the same split from the same +//! inputs: the file list (from the checkout), and each file's median +//! duration as aster-srv reported it for the UTC day of the commit (which +//! the server answers the same way however often it is asked). The split +//! is greedy longest-processing-time: the longest file first, each to the +//! partition with the least time so far. +//! +//! Durations are a hint. Without them (no server, no history, an error) +//! the command runs as written, and Mix deals the files out itself. + +use std::collections::HashMap; +use std::path::{Path, PathBuf}; +use std::time::Duration; + +use anyhow::{bail, Context, Result}; +use chrono::NaiveDate; +use serde::{Deserialize, Serialize}; + +use super::client::{percent_encode, Server}; +use super::upload::{proxy_token_from_env, Credentials}; + +/// The most partition targets one request names (the server's limit). +pub const MAX_TARGETS: usize = 32; +/// Requests made before giving up on the server. +const ATTEMPTS: u32 = 3; +/// How long one request may take. +const TIMEOUT: Duration = Duration::from_secs(10); + +/// Each file's median duration in milliseconds. +pub type Durations = HashMap; + +/// Split `files` into `partitions` sets. A file without a duration counts +/// as the median of the known ones. Ties go to the earlier path and the +/// lower partition, so the split depends only on its inputs. Every +/// partition gets a file when there are at least as many files. +pub fn split(files: &[String], durations: &Durations, partitions: usize) -> Vec> { + let mut known: Vec = files + .iter() + .filter_map(|file| durations.get(file).copied()) + .collect(); + known.sort_unstable(); + let fallback = known.get(known.len() / 2).copied().unwrap_or(1); + let mut weighted: Vec<(u64, &String)> = files + .iter() + .map(|file| (durations.get(file).copied().unwrap_or(fallback), file)) + .collect(); + weighted.sort_by(|a, b| b.0.cmp(&a.0).then_with(|| a.1.cmp(b.1))); + let mut bins: Vec<(u64, Vec)> = vec![(0, Vec::new()); partitions]; + for (weight, file) in weighted { + let (_, bin) = bins + .iter_mut() + .enumerate() + .min_by_key(|(index, (load, files))| (*load, files.len(), *index)) + .expect("at least one partition"); + bin.0 += weight; + bin.1.push(file.clone()); + } + bins.into_iter() + .map(|(_, mut files)| { + files.sort(); + files + }) + .collect() +} + +/// The expected time of `files`, by the same weights as [`split`]. +pub fn expected_ms(files: &[String], all: &[String], durations: &Durations) -> u64 { + let mut known: Vec = all + .iter() + .filter_map(|f| durations.get(f).copied()) + .collect(); + known.sort_unstable(); + let fallback = known.get(known.len() / 2).copied().unwrap_or(1); + files + .iter() + .map(|f| durations.get(f).copied().unwrap_or(fallback)) + .sum() +} + +/// The UTC day of `HEAD`'s commit, the same on every runner of that commit. +pub fn commit_day(workspace_root: &Path) -> Option { + let output = std::process::Command::new("git") + .args(["log", "-1", "--format=%cI", "HEAD"]) + .current_dir(workspace_root) + .output() + .ok()?; + if !output.status.success() { + return None; + } + let text = String::from_utf8(output.stdout).ok()?; + let at = chrono::DateTime::parse_from_rfc3339(text.trim()).ok()?; + Some(at.with_timezone(&chrono::Utc).date_naive()) +} + +#[derive(Serialize, Deserialize)] +struct Cached { + targets: Vec, + /// `YYYY-MM-DD`. + day: String, + files: Vec, +} + +#[derive(Serialize, Deserialize)] +struct CachedFile { + file: String, + p50_ms: u64, +} + +fn cache_path(workspace_root: &Path, targets: &[String], day: NaiveDate) -> PathBuf { + use sha2::{Digest, Sha256}; + let mut hasher = Sha256::new(); + for target in targets { + hasher.update(target.as_bytes()); + hasher.update([0]); + } + let digest: String = hasher + .finalize() + .iter() + .map(|b| format!("{b:02x}")) + .collect(); + workspace_root + .join(".aster") + .join("durations") + .join(format!("{day}-{}.json", &digest[..16])) +} + +/// The durations of the files `targets` ran, as of `day`. `Ok(None)`: no +/// server is configured. Answers are kept under `.aster/durations/`, since +/// the server's answer for a day does not change. +pub fn fetch( + workspace_root: &Path, + targets: &[String], + day: NaiveDate, + env: &dyn Fn(&str) -> Option, +) -> Result> { + let Some(url) = env("ASTER_SRV_URL").filter(|url| !url.is_empty()) else { + return Ok(None); + }; + let mut targets = targets.to_vec(); + targets.sort(); + targets.dedup(); + let cache = cache_path(workspace_root, &targets, day); + if let Some(found) = std::fs::read(&cache) + .ok() + .and_then(|bytes| serde_json::from_slice::(&bytes).ok()) + .filter(|cached| cached.targets == targets && cached.day == day.to_string()) + { + return Ok(Some( + found + .files + .into_iter() + .map(|f| (f.file, f.p50_ms)) + .collect(), + )); + } + let Some(credentials) = Credentials::from_env(env) else { + bail!("no credentials for {url}"); + }; + let mut path = format!("/v1/test-results/durations?day={day}"); + for target in &targets { + path.push_str(&format!("&target={}", percent_encode(target))); + } + let mut attempt = 1; + let body = loop { + let result = + Server::connect_with_timeout(&url, &credentials, proxy_token_from_env(env), TIMEOUT) + .and_then(|server| server.get_json(&path)); + match result { + Ok(Some(body)) => break body, + Ok(None) => bail!("{url} does not serve durations"), + Err(error) if attempt >= ATTEMPTS => return Err(error), + Err(_) => { + std::thread::sleep(Duration::from_secs(u64::from(attempt))); + attempt += 1; + } + } + }; + let files: Vec = serde_json::from_value(body["files"].clone()) + .context("cannot read the server's durations")?; + let cached = Cached { + targets, + day: day.to_string(), + files, + }; + if let Some(parent) = cache.parent() { + if std::fs::create_dir_all(parent).is_ok() { + if let Ok(bytes) = serde_json::to_vec(&cached) { + let _ = std::fs::write(&cache, bytes); + } + } + } + Ok(Some( + cached + .files + .into_iter() + .map(|f| (f.file, f.p50_ms)) + .collect(), + )) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn names(list: &[&str]) -> Vec { + list.iter().map(|s| s.to_string()).collect() + } + + #[test] + fn longest_files_first_to_the_least_loaded_partition() { + let files = names(&["a", "b", "c", "d", "e", "f"]); + let durations: Durations = [ + ("a", 70), + ("b", 50), + ("c", 40), + ("d", 30), + ("e", 20), + ("f", 10), + ] + .into_iter() + .map(|(f, ms)| (f.to_string(), ms)) + .collect(); + let bins = split(&files, &durations, 3); + // a(70) | b(50)+e(20) | c(40)+d(30), then f(10) to the first at 70. + assert_eq!( + bins, + [names(&["a", "f"]), names(&["b", "e"]), names(&["c", "d"])] + ); + let all: Vec = bins.concat(); + assert_eq!(all.len(), files.len()); + assert_eq!(expected_ms(&bins[1], &files, &durations), 70); + } + + #[test] + fn unknown_files_count_as_the_median_and_ties_are_stable() { + let files = names(&["a", "b", "c", "new1", "new2"]); + let durations: Durations = [("a", 10), ("b", 20), ("c", 30)] + .into_iter() + .map(|(f, ms)| (f.to_string(), ms)) + .collect(); + assert_eq!(expected_ms(&names(&["new1"]), &files, &durations), 20); + let first = split(&files, &durations, 2); + let mut shuffled = files.clone(); + shuffled.reverse(); + assert_eq!(split(&shuffled, &durations, 2), first); + // With no durations at all every file weighs the same. + let even = split(&files, &Durations::new(), 2); + assert_eq!(even, [names(&["a", "c", "new2"]), names(&["b", "new1"])]); + } + + #[test] + fn every_partition_gets_a_file_even_at_zero_cost() { + let files = names(&["a", "b", "c"]); + let zero: Durations = files.iter().map(|f| (f.clone(), 0)).collect(); + let bins = split(&files, &zero, 3); + assert!(bins.iter().all(|bin| bin.len() == 1), "{bins:?}"); + } + + #[test] + fn no_server_means_no_durations() { + let dir = tempfile::tempdir().unwrap(); + let day = NaiveDate::from_ymd_opt(2026, 10, 15).unwrap(); + let found = fetch(dir.path(), &names(&["//a:test"]), day, &|_| None).unwrap(); + assert!(found.is_none()); + } + + #[test] + fn a_cached_answer_is_used_without_the_server() { + let dir = tempfile::tempdir().unwrap(); + let day = NaiveDate::from_ymd_opt(2026, 10, 15).unwrap(); + let targets = names(&["//b:test", "//a:test"]); + let mut sorted = targets.clone(); + sorted.sort(); + let path = cache_path(dir.path(), &sorted, day); + std::fs::create_dir_all(path.parent().unwrap()).unwrap(); + std::fs::write( + &path, + serde_json::to_vec(&Cached { + targets: sorted, + day: day.to_string(), + files: vec![CachedFile { + file: "test/a_test.exs".into(), + p50_ms: 42, + }], + }) + .unwrap(), + ) + .unwrap(); + // The URL is unreachable and there are no credentials: only the + // cache can answer. + let env = |name: &str| (name == "ASTER_SRV_URL").then(|| "http://127.0.0.1:9".to_string()); + let found = fetch(dir.path(), &targets, day, &env).unwrap().unwrap(); + assert_eq!(found["test/a_test.exs"], 42); + } +} diff --git a/src/test_results/client.rs b/src/test_results/client.rs index 8242294..beb4ae9 100644 --- a/src/test_results/client.rs +++ b/src/test_results/client.rs @@ -27,7 +27,17 @@ impl Server { credentials: &Credentials, proxy_token: Option, ) -> Result { - let agent = agent(Duration::from_secs(20)); + Self::connect_with_timeout(url, credentials, proxy_token, Duration::from_secs(20)) + } + + /// [`Server::connect`] whose requests give up after `timeout`. + pub fn connect_with_timeout( + url: &str, + credentials: &Credentials, + proxy_token: Option, + timeout: Duration, + ) -> Result { + let agent = agent(timeout); let token = bearer(&agent, credentials)?; Ok(Self { agent, diff --git a/src/test_results/mod.rs b/src/test_results/mod.rs index 4a0406c..ce53e22 100644 --- a/src/test_results/mod.rs +++ b/src/test_results/mod.rs @@ -9,6 +9,7 @@ //! The plan is applied after the cache key is computed from the original //! command, so capturing results never changes what is cached. +pub mod balance; pub mod client; mod context; pub(crate) mod exunit; @@ -57,6 +58,11 @@ pub struct TestReport { pub capture: String, pub confidence: Confidence, pub cases: Vec, + /// The project whose tests these are, when it is not the target's own: + /// a shard project that runs another project's suite + /// (`cd ../platform && mix test`). Test identities use it, so a test + /// keeps its id whichever shard runs it. + pub project: Option, } /// A test that failed in one attempt and passed in a later one. @@ -473,6 +479,7 @@ mod tests { #[test] fn counts_failures() { let report = TestReport { + project: None, capture: "junit".into(), confidence: Confidence::Complete, cases: vec![ diff --git a/src/test_results/stream.rs b/src/test_results/stream.rs index 1979592..0baaed0 100644 --- a/src/test_results/stream.rs +++ b/src/test_results/stream.rs @@ -35,10 +35,12 @@ pub fn records(run: RunRecord, results: &[ExecutionResult]) -> Vec { let Some(report) = &result.tests else { continue; }; - let project = result - .address - .rsplit_once(':') - .map_or(result.address.as_str(), |(project, _)| project); + let project = report.project.as_deref().unwrap_or_else(|| { + result + .address + .rsplit_once(':') + .map_or(result.address.as_str(), |(project, _)| project) + }); for case in &report.cases { let identity = TestIdentity::new( project, @@ -148,6 +150,7 @@ mod tests { #[test] fn writes_run_then_targets_then_their_tests() { let report = TestReport { + project: None, capture: "junit".into(), confidence: Confidence::Complete, cases: vec![ @@ -200,6 +203,34 @@ mod tests { assert_eq!(records.len(), 5); } + #[test] + fn a_shards_tests_are_named_for_the_project_that_owns_them() { + let report = |shard: &str| { + let report = TestReport { + project: Some("//services/platform".into()), + capture: "exunit-formatter".into(), + confidence: Confidence::Complete, + cases: vec![case("fails", TestOutcome::Fail, Some("boom"))], + }; + records(run(), &[result(shard, false, Some(report))]) + }; + let ids: Vec = [ + "//services/platform-test-shard-1:test-ci", + "//services/platform-test-shard-3:test-ci", + ] + .into_iter() + .map(|shard| match &report(shard)[2] { + Record::Test(test) => { + assert_eq!(test.target, shard); + test.test_id.clone() + } + _ => panic!("expected test"), + }) + .collect(); + let owned = TestIdentity::new("//services/platform", "test/a_test.exs", "AppTest", "fails"); + assert_eq!(ids, [owned.test_id(), owned.test_id()]); + } + #[test] fn writes_ndjson_that_reads_back() { let tmp = tempfile::tempdir().unwrap();