diff --git a/crates/aster-srv/README.md b/crates/aster-srv/README.md index 5652e16..0e18741 100644 --- a/crates/aster-srv/README.md +++ b/crates/aster-srv/README.md @@ -183,9 +183,8 @@ merged JSON. `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 +by time. The answer for a set of targets and a UTC `day` 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 @@ -201,9 +200,35 @@ empty without history. } ``` -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). +Partitions never ask the server while they run: one shard that could not +reach it would split differently from the rest, and a file would run on two +shards or none. Instead `aster results durations` asks for every partitioned +suite in the workspace and writes `aster-durations.json` at its root, to be +committed. A target that runs a partition of a suite reads its suite there +and runs the files it gets from a longest-first split, so every shard of a +commit splits the same way. Without the file or the suite in it, or with +`ASTER_BALANCE_PARTITIONS=off`, Mix splits as usual. Files new since the +refresh count as the suite's median. Refresh it on a schedule, for example: + +```yaml +on: + schedule: [{ cron: "0 6 * * 1" }] +permissions: { id-token: write, contents: write, pull-requests: write } +jobs: + durations: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - run: aster results durations + env: { ASTER_SRV_URL: https://aster-srv.example.com } + - run: | + git diff --quiet aster-durations.json && exit 0 + git switch -c aster-durations-$(date +%F) + git commit -am "chore: refresh aster-durations.json" + git push -u origin HEAD && gh pr create --fill + env: { GH_TOKEN: "${{ github.token }}" } +``` + 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. diff --git a/docs/specs/test-tracking.html b/docs/specs/test-tracking.html index d03a45e..7b22b9b 100644 --- a/docs/specs/test-tracking.html +++ b/docs/specs/test-tracking.html @@ -455,8 +455,8 @@

Aster Test Tracking

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-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)"], + 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 aster-durations.json, committed at the workspace root and refreshed from GET /v1/test-results/durations by aster results durations (on a schedule). Shards never ask the server while running, so every shard of a commit reads the same durations. 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, on every shard alike: no aster-durations.json, the suite missing from it, a malformed file, 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.", diff --git a/src/cli/commands.rs b/src/cli/commands.rs index d0556e8..e92c75f 100644 --- a/src/cli/commands.rs +++ b/src/cli/commands.rs @@ -626,6 +626,20 @@ pub enum ResultsCommands { #[arg(long)] url: Option, }, + /// Write aster-durations.json from aster-srv, for balanced partitions + /// + /// Finds every `MIX_TEST_PARTITION=K mix test --partitions N` suite in + /// the workspace and writes each test file's median duration, from the + /// partitions' recent runs, to aster-durations.json at the workspace + /// root. Commit it: each partition then splits the suite by those + /// durations, from the checkout, so all partitions of a commit agree. + /// Uses the credentials `upload` uses. Leaves the file unchanged and + /// exits 1 when any suite cannot be fetched. + Durations { + /// The aster-srv URL [default: $ASTER_SRV_URL] + #[arg(long)] + url: Option, + }, } /// Project-level subcommands diff --git a/src/executor/runner.rs b/src/executor/runner.rs index 78cd799..f667dff 100644 --- a/src/executor/runner.rs +++ b/src/executor/runner.rs @@ -37,6 +37,7 @@ use crate::test_results::{ use crate::ui::ProgressDisplay; mod partitions; +pub use partitions::partitioned_suites; mod retry; use super::command::parse_command; @@ -802,7 +803,6 @@ impl<'a> Executor<'a> { &commands, &working_dir, self.workspace_root, - project_map, &|name| std::env::var(name).ok(), ) { Some(balanced) => vec![balanced], diff --git a/src/executor/runner/partitions.rs b/src/executor/runner/partitions.rs index 49d8c65..30b7dfd 100644 --- a/src/executor/runner/partitions.rs +++ b/src/executor/runner/partitions.rs @@ -3,14 +3,15 @@ //! 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. +//! And when the workspace's durations file (written from aster-srv by +//! `aster results durations`) times the suite, 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::plugins::mix_partitions; use crate::test_results::balance; /// `ASTER_BALANCE_PARTITIONS=off` keeps Mix's own split. @@ -31,24 +32,48 @@ pub(super) fn test_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}")) - }) +/// Every partitioned Mix suite in the workspace: its directory relative to +/// `workspace_root` and the targets that run a partition of it, sorted. +pub fn partitioned_suites( + workspace_root: &Path, + project_map: &HashMap, +) -> Vec<(String, Vec)> { + let mut suites: std::collections::BTreeMap> = Default::default(); + for (address, project) in project_map { + for (name, target) in &project.targets { + let working_dir = target.working_dir.as_deref().unwrap_or(&project.root); + let Some(found) = mix_partitions::parse(working_dir, &target.command) else { + continue; + }; + if let Some(key) = suite_key(workspace_root, &found.dir) { + suites + .entry(key) + .or_default() + .push(format!("{address}:{name}")); + } + } + } + suites + .into_iter() + .map(|(key, mut targets)| { + targets.sort(); + (key, targets) }) + .collect() +} + +/// A suite's key in the durations file: its directory relative to the +/// workspace root, with `/` separators. +fn suite_key(workspace_root: &Path, dir: &Path) -> Option { + let relative = canonical(dir) + .strip_prefix(canonical(workspace_root)) + .ok()? + .to_path_buf(); + let parts: Option> = relative + .components() + .map(|c| c.as_os_str().to_str()) .collect(); - peers.sort(); - peers + Some(parts?.join("/")) } fn canonical(path: &Path) -> std::path::PathBuf { @@ -63,7 +88,6 @@ pub(super) fn balance( commands: &[String], working_dir: &Path, workspace_root: &Path, - project_map: &HashMap, env: &dyn Fn(&str) -> Option, ) -> Option { let [command] = commands else { @@ -81,33 +105,23 @@ pub(super) fn balance( 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(suite) = suite_key(workspace_root, &found.dir) else { + return keep("its suite is outside the workspace"); + }; + let durations = match balance::read(workspace_root, &suite) { + Ok(Some(durations)) => durations, + Ok(None) => return keep(&format!("no durations for {suite} in {}", balance::FILE)), + Err(error) => return keep(&format!("{error:#}")), + }; 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"); + return keep(&format!("none of its files are timed in {}", balance::FILE)); } let bins = balance::split(&files, &durations, found.total); let mine = &bins[found.partition - 1]; @@ -119,11 +133,12 @@ pub(super) fn balance( .unwrap_or(0); eprintln!( "{label} balanced by duration: {} of {} files, ~{}s of a slowest ~{}s \ - ({known} files timed, medians before {day})", + ({known} files timed in {})", mine.len(), files.len(), expected / 1000, slowest / 1000, + balance::FILE, ); Some(found.render(mine)) } @@ -188,137 +203,108 @@ mod tests { ); 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"] + partitioned_suites(ws.path(), &map), + [( + "app".to_string(), + vec![ + "//shard-1:test-ci".to_string(), + "//shard-2:test-ci".to_string() + ] + )] ); } #[test] - fn without_a_server_or_when_switched_off_mix_splits() { + fn without_durations_or_when_switched_off_mix_splits() { let ws = tempfile::tempdir().unwrap(); + std::fs::create_dir_all(ws.path().join("app/test")).unwrap(); + for file in ["a", "b"] { + std::fs::write(ws.path().join(format!("app/test/{file}_test.exs")), "").unwrap(); + } + std::fs::write(ws.path().join("app/mix.exs"), "[app: :app]").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; + // No durations file. assert_eq!( - balance(&address, &command, &s1.root, ws.path(), &map, &none), + balance(&address, &command, &s1.root, ws.path(), &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, - }; + // A file without this suite. + write_durations(ws.path(), "other", &[("test/a_test.exs", 5)]); assert_eq!( - balance(&address, &command, &s1.root, ws.path(), &map, &off), + balance(&address, &command, &s1.root, ws.path(), &none), None ); - // Not a partitioned command at all. - let plain = vec!["mix test".to_string()]; + // A broken file. + std::fs::write(ws.path().join(balance::FILE), "{").unwrap(); assert_eq!( - balance(&address, &plain, &s1.root, ws.path(), &map, &none), + balance(&address, &command, &s1.root, ws.path(), &none), None ); + // Timed, but switched off. + write_durations(ws.path(), "app", &[("test/a_test.exs", 5)]); + let off = |name: &str| (name == OFF_SWITCH).then(|| "off".to_string()); + assert_eq!(balance(&address, &command, &s1.root, ws.path(), &off), None); + assert!(balance(&address, &command, &s1.root, ws.path(), &none).is_some()); + // Not a partitioned command at all. + let plain = vec!["mix test".to_string()]; + assert_eq!(balance(&address, &plain, &s1.root, ws.path(), &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) + fn write_durations(root: &Path, suite: &str, files: &[(&str, u64)]) { + let file = balance::DurationsFile { + version: 1, + day: "2026-10-15".into(), + suites: [( + suite.to_string(), + balance::Suite { + targets: vec![], + files: files.iter().map(|(f, ms)| (f.to_string(), *ms)).collect(), + }, + )] + .into_iter() + .collect(), + }; + balance::write(root, &file).unwrap(); } #[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 (_, s1) = shard(ws.path(), 1); + let (_, s2) = shard(ws.path(), 2); + write_durations( + ws.path(), + "app", + &[ + ("test/a_test.exs", 100), + ("test/b_test.exs", 10), + ("test/c_test.exs", 10), + ("test/d_test.exs", 80), + ("test/gone_test.exs", 999), + ], ); - let env = move |name: &str| match name { - "ASTER_SRV_URL" => Some(url.clone()), - "ASTER_SRV_TOKEN" => Some("t".to_string()), - _ => None, - }; + // No server and no network: the checkout alone decides. + let none = |_: &str| 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() + balance(address, &command, &shard.root, ws.path(), &none).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, + run(&s1, "//shard-1:test-ci"), "bash -c 'cd ../app && MIX_TEST_PARTITION=1 exec mix test test/a_test.exs'" ); assert_eq!( - second, + run(&s2, "//shard-2:test-ci"), "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/main.rs b/src/main.rs index 93e9ee5..110f9da 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1733,6 +1733,42 @@ fn handle_results(command: &ResultsCommands, output_mode: OutputMode) -> Result< println!("{report}"); Ok(()) } + ResultsCommands::Durations { url } => { + use aster::test_results::balance; + let url = srv_url(url.as_deref()) + .ok_or_else(|| anyhow::anyhow!("give --url or set ASTER_SRV_URL"))?; + let cwd = env::current_dir().context("Failed to get current directory")?; + let workspace_root = find_workspace_root(&cwd) + .ok_or_else(|| anyhow::anyhow!("not in an aster workspace"))?; + let registry = PluginRegistry::with_all_plugins(); + let projects = discover_projects(&workspace_root, ®istry) + .context("Failed to discover projects")?; + let project_map: HashMap = projects + .iter() + .map(|project| (project_address(project), project)) + .collect(); + let suites = aster::executor::runner::partitioned_suites(&workspace_root, &project_map); + if suites.is_empty() { + anyhow::bail!("no partitioned `mix test --partitions` suites in this workspace"); + } + let day = chrono::Utc::now().date_naive(); + let file = balance::fetch_all(&url, &suites, day, &|name| env::var(name).ok())?; + balance::write(&workspace_root, &file)?; + if output_mode != OutputMode::Quiet { + for (dir, suite) in &file.suites { + eprintln!( + "[aster] {dir}: {} files timed from {} partitions", + suite.files.len(), + suite.targets.len() + ); + } + eprintln!( + "[aster] Wrote {}", + workspace_root.join(balance::FILE).display() + ); + } + Ok(()) + } } } diff --git a/src/test_results/balance.rs b/src/test_results/balance.rs index e6aaa12..8fdc01c 100644 --- a/src/test_results/balance.rs +++ b/src/test_results/balance.rs @@ -2,17 +2,18 @@ //! //! 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. +//! inputs: the file list and the durations in [`FILE`], both from the +//! checkout. Nothing is fetched while the suite runs, so no partition can +//! see different durations from another. `aster results durations` writes +//! the file from aster-srv's medians, to be committed like any change. 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. +//! Without the file, or without the suite in it, the command runs as +//! written and Mix deals the files out itself, on every partition alike. -use std::collections::HashMap; -use std::path::{Path, PathBuf}; +use std::collections::{BTreeMap, HashMap}; +use std::path::Path; use std::time::Duration; use anyhow::{bail, Context, Result}; @@ -80,93 +81,114 @@ pub fn expected_ms(files: &[String], all: &[String], durations: &Durations) -> u .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()) -} +/// The durations file, at the workspace root. +pub const FILE: &str = "aster-durations.json"; +/// The [`DurationsFile::version`] this Aster writes and reads. +const VERSION: u32 = 1; -#[derive(Serialize, Deserialize)] -struct Cached { - targets: Vec, - /// `YYYY-MM-DD`. - day: String, - files: Vec, +/// [`FILE`]: each partitioned suite's file durations. +#[derive(Debug, Default, PartialEq, Serialize, Deserialize)] +pub struct DurationsFile { + pub version: u32, + /// The UTC day the medians were computed as of, `YYYY-MM-DD`. + pub day: String, + /// By the suite's directory, relative to the workspace root. + pub suites: BTreeMap, } -#[derive(Serialize, Deserialize)] -struct CachedFile { - file: String, - p50_ms: u64, +/// One partitioned suite in [`FILE`]. +#[derive(Debug, Default, PartialEq, Serialize, Deserialize)] +pub struct Suite { + /// The partition targets whose runs the medians come from. + pub targets: Vec, + /// Each test file's median duration in milliseconds. + pub files: BTreeMap, } -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]); +/// The durations [`FILE`] records for the suite in `suite` (relative to +/// the workspace root). `Ok(None)`: no file, or the suite is not in it. +pub fn read(workspace_root: &Path, suite: &str) -> Result> { + let path = workspace_root.join(FILE); + let bytes = match std::fs::read(&path) { + Ok(bytes) => bytes, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None), + Err(error) => return Err(error).with_context(|| format!("cannot read {FILE}")), + }; + let file: DurationsFile = + serde_json::from_slice(&bytes).with_context(|| format!("cannot parse {FILE}"))?; + if file.version != VERSION { + bail!( + "{FILE} is version {}, this Aster reads {VERSION}", + file.version + ); } - let digest: String = hasher - .finalize() - .iter() - .map(|b| format!("{b:02x}")) - .collect(); - workspace_root - .join(".aster") - .join("durations") - .join(format!("{day}-{}.json", &digest[..16])) + Ok(file + .suites + .get(suite) + .map(|suite| suite.files.iter().map(|(f, ms)| (f.clone(), *ms)).collect())) } -/// 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], +/// Write `file` as [`FILE`], sorted and pretty, so a refresh diffs cleanly. +pub fn write(workspace_root: &Path, file: &DurationsFile) -> Result<()> { + let mut text = serde_json::to_string_pretty(file)?; + text.push('\n'); + std::fs::write(workspace_root.join(FILE), text).with_context(|| format!("cannot write {FILE}")) +} + +/// A [`DurationsFile`] for `suites` (each a directory and its partition +/// targets), with aster-srv's medians as of `day`. +pub fn fetch_all( + url: &str, + suites: &[(String, Vec)], 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(), - )); - } +) -> Result { let Some(credentials) = Credentials::from_env(env) else { - bail!("no credentials for {url}"); + bail!("no credentials for {url}: in GitHub Actions grant `id-token: write`; elsewhere set ASTER_SRV_TOKEN"); }; + let mut file = DurationsFile { + version: VERSION, + day: day.to_string(), + suites: BTreeMap::new(), + }; + for (dir, targets) in suites { + let mut targets = targets.clone(); + targets.sort(); + targets.dedup(); + if targets.len() > MAX_TARGETS { + bail!( + "{dir} has {} partition targets; aster-srv takes at most {MAX_TARGETS}", + targets.len() + ); + } + let files = fetch(url, &credentials, env, &targets, day) + .with_context(|| format!("durations of {dir}"))?; + file.suites.insert(dir.clone(), Suite { targets, files }); + } + Ok(file) +} + +/// The medians aster-srv reports for the files `targets` ran, as of `day`. +fn fetch( + url: &str, + credentials: &Credentials, + env: &dyn Fn(&str) -> Option, + targets: &[String], + day: NaiveDate, +) -> Result> { + #[derive(Deserialize)] + struct File { + file: String, + p50_ms: u64, + } let mut path = format!("/v1/test-results/durations?day={day}"); - for target in &targets { + 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) + 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, @@ -178,27 +200,9 @@ pub fn fetch( } } }; - let files: Vec = serde_json::from_value(body["files"].clone()) + 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(), - )) + Ok(files.into_iter().map(|f| (f.file, f.p50_ms)).collect()) } #[cfg(test)] @@ -260,39 +264,49 @@ mod tests { } #[test] - fn no_server_means_no_durations() { + fn the_file_round_trips_and_names_suites_by_directory() { 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()); + assert_eq!(read(dir.path(), "services/platform").unwrap(), None); + let file = DurationsFile { + version: VERSION, + day: "2026-10-15".into(), + suites: [( + "services/platform".to_string(), + Suite { + targets: names(&["//services/platform-test-shard-1:test-ci"]), + files: [("test/a_test.exs".to_string(), 42)].into_iter().collect(), + }, + )] + .into_iter() + .collect(), + }; + write(dir.path(), &file).unwrap(); + let found = read(dir.path(), "services/platform").unwrap().unwrap(); + assert_eq!(found["test/a_test.exs"], 42); + assert_eq!(read(dir.path(), "services/other").unwrap(), None); + let text = std::fs::read_to_string(dir.path().join(FILE)).unwrap(); + assert!(text.ends_with("}\n"), "{text}"); } #[test] - fn a_cached_answer_is_used_without_the_server() { + fn a_broken_or_newer_file_is_an_error() { 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(dir.path().join(FILE), "{").unwrap(); + assert!(read(dir.path(), "a").is_err()); 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(), + dir.path().join(FILE), + r#"{"version":2,"day":"x","suites":{}}"#, ) .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); + let error = read(dir.path(), "a").unwrap_err().to_string(); + assert!(error.contains("version 2"), "{error}"); + } + + #[test] + fn fetching_needs_credentials() { + let day = NaiveDate::from_ymd_opt(2026, 10, 15).unwrap(); + let suites = vec![("app".to_string(), names(&["//a:test"]))]; + let error = fetch_all("http://127.0.0.1:9", &suites, day, &|_| None).unwrap_err(); + assert!(error.to_string().contains("no credentials"), "{error}"); } } diff --git a/tests/test_results.rs b/tests/test_results.rs index 0153766..32365f7 100644 --- a/tests/test_results.rs +++ b/tests/test_results.rs @@ -561,3 +561,144 @@ fn results_failures_exits_nonzero_when_it_cannot_ask() { assert!(stderr.contains(message), "{stderr}"); } } + +/// Answers one request with `body` and hands back 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) +} + +/// Two shard projects that each run a partition of `app`'s suite, and a +/// `mix` on PATH that writes what it was asked to run to +/// `app/ran-.txt`. +#[cfg(unix)] +fn partitioned_workspace() -> TempDir { + use std::os::unix::fs::PermissionsExt; + let tmp = TempDir::new().unwrap(); + fs::create_dir(tmp.path().join(".git")).unwrap(); + write(tmp.path(), "app/mix.exs", "[app: :app]\n"); + for file in ["a", "b", "c", "d"] { + write(tmp.path(), &format!("app/test/{file}_test.exs"), ""); + } + for k in [1, 2] { + write( + tmp.path(), + &format!("shard-{k}/package.json"), + &format!(r#"{{"name":"shard-{k}","private":true}}"#), + ); + write( + tmp.path(), + &format!("shard-{k}/aster.toml"), + &format!( + "[targets.test-ci]\ncommand = \"bash -c 'cd ../app && MIX_TEST_PARTITION={k} exec mix test --partitions 2'\"\n" + ), + ); + } + write( + tmp.path(), + "bin/mix", + "#!/bin/sh\necho \"$*\" > \"ran-$MIX_TEST_PARTITION.txt\"\n", + ); + let mix = tmp.path().join("bin/mix"); + fs::set_permissions(&mix, fs::Permissions::from_mode(0o755)).unwrap(); + tmp +} + +#[cfg(unix)] +fn aster_with_mix(root: &Path, args: &[&str]) -> Output { + let path = format!( + "{}:{}", + root.join("bin").display(), + std::env::var("PATH").unwrap_or_default() + ); + Command::new(env!("CARGO_BIN_EXE_aster")) + .current_dir(root) + .env("PATH", path) + .env_remove("ASTER_SRV_URL") + .env_remove("ASTER_BALANCE_PARTITIONS") + .env("ASTER_SRV_TOKEN", "t") + .args(args) + .output() + .unwrap() +} + +#[test] +#[cfg(unix)] +fn shards_split_by_the_committed_durations_file_without_a_server() { + let ws = partitioned_workspace(); + let root = ws.path(); + // Without the file, Mix deals the files out. + let output = aster_with_mix(root, &["run", "//shard-1:test-ci", "--no-cache"]); + let log = String::from_utf8_lossy(&output.stderr).to_string() + + &String::from_utf8_lossy(&output.stdout); + assert!(output.status.success(), "{log}"); + let ran = |k: u32| fs::read_to_string(root.join(format!("app/ran-{k}.txt"))).unwrap(); + assert_eq!(ran(1), "test --partitions 2\n"); + assert!( + log.contains("split by Mix (no durations for app in aster-durations.json)"), + "{log}" + ); + + // `aster results durations` writes the file from the server. + 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}]}"#, + ); + let output = aster_with_mix(root, &["results", "durations", "--url", &url]); + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + let request = server.join().unwrap(); + assert!( + request.contains("target=%2F%2Fshard-1%3Atest-ci&target=%2F%2Fshard-2%3Atest-ci"), + "{request}" + ); + let file: serde_json::Value = + serde_json::from_str(&fs::read_to_string(root.join("aster-durations.json")).unwrap()) + .unwrap(); + assert_eq!(file["suites"]["app"]["files"]["test/a_test.exs"], 100); + + // Each shard now runs its share by name, with no server configured. + let run = |shard: &str| { + let output = aster_with_mix(root, &["run", &format!("//{shard}:test-ci"), "--no-cache"]); + let log = String::from_utf8_lossy(&output.stderr).to_string() + + &String::from_utf8_lossy(&output.stdout); + assert!(output.status.success(), "{log}"); + log + }; + let first = run("shard-1"); + assert!( + first.contains("balanced by duration: 1 of 4 files"), + "{first}" + ); + assert_eq!(ran(1), "test test/a_test.exs\n"); + run("shard-2"); + assert_eq!( + ran(2), + "test test/b_test.exs test/c_test.exs test/d_test.exs\n" + ); +}