diff --git a/Cargo.lock b/Cargo.lock index 9b5dc058..9766390f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3427,6 +3427,7 @@ dependencies = [ "du-bio", "du-domain", "libc", + "md-5 0.10.6", "navigator-align", "navigator-analysis", "navigator-domain", @@ -3437,6 +3438,7 @@ dependencies = [ "reqwest", "serde", "serde_json", + "sha2 0.10.9", "thiserror 2.0.18", "tokio", "uuid", diff --git a/crates/navigator-app/Cargo.toml b/crates/navigator-app/Cargo.toml index f6d17f42..7283b4bf 100644 --- a/crates/navigator-app/Cargo.toml +++ b/crates/navigator-app/Cargo.toml @@ -32,7 +32,15 @@ chrono = { version = "0.4", features = ["serde"] } uuid = { version = "1", features = ["v4"] } # Encode/decode the persisted DM session key (base64-STANDARD, matching navigator-sync's wire). base64 = "0.22" -tokio = { version = "1", features = ["rt-multi-thread", "macros"] } +# ENA publishes an md5 for every run file (`grid::ena`). Already in the lock transitively, so a +# direct dependency adds nothing to the graph. +md-5 = "0.10" +# `grid::canonical_sha256_b64` — the digest hash that the submit signature covers. Must give the +# same answer as `du_db::grid::digest::canonical_sha256_b64` on the AppView. +sha2 = "0.10" +# `time`, for the retry delay in `ena` and the heartbeat interval in `grid_job`. Both already +# compiled through feature unification from another crate, which is not a property to depend on. +tokio = { version = "1", features = ["rt-multi-thread", "macros", "time"] } # Holds the auth HTTP client (built by navigator-sync::dev_http_client). Pinned to 0.12 # with rustls to match du-atproto / navigator-sync. reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] } @@ -41,7 +49,10 @@ reqwest = { version = "0.12", default-features = false, features = ["json", "rus reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] } # `signal`, so the headless realignment harness can cancel through the job's own token on Ctrl-C # and let it clean up its scratch. Dev-only: the shipped crate does not need it. -tokio = { version = "1", features = ["rt-multi-thread", "macros", "signal"] } +# `net`, so the ENA downloader's resume path can be tested against a real HTTP server rather than +# mocked — Range handling and md5-over-a-resumed-prefix are exactly the things a mock would get +# wrong in the same way the code does. Dev-only, like `signal` above. +tokio = { version = "1", features = ["rt-multi-thread", "macros", "signal", "net", "io-util"] } # Build a synthetic mtDNA FASTA from the bundled rCRS for the import test. navigator-analysis = { workspace = true } diff --git a/crates/navigator-app/src/ena.rs b/crates/navigator-app/src/ena.rs new file mode 100644 index 00000000..cae06cb2 --- /dev/null +++ b/crates/navigator-app/src/ena.rs @@ -0,0 +1,519 @@ +//! This module gets the files of a Grid work unit from ENA. +//! +//! A node receives a full manifest with its lease. The manifest holds the URL, the md5 value and +//! the byte size of each file. The AppView makes that list. So a node does not ask ENA to find +//! anything, and this module only gets the files that the manifest names. See +//! `documents/design/distributed-compute-grid.md` §7.1. +//! +//! # Why this module is not `refgenome::download` +//! +//! §7.1 first told us to copy that function. Three properties make a copy impossible. Each one is +//! more important here than for a reference genome. +//! +//! 1. **That function can not continue a transfer.** It sends no `Range` header. An interrupted +//! transfer must start again at zero. +//! +//! A reference genome is about 900 MB. An ENA run file is 10 to 30 GB on the connection of a +//! volunteer. The ability to continue decides if a unit ever completes. +//! 2. **That function calculates SHA-256.** ENA publishes md5. A checksum that you can not compare +//! gives no integrity. +//! 3. **Its one retry is blind.** It does the whole transfer again after each error. Some errors +//! stay after a second try. +//! +//! This module does copy the `.part` file and the rename at the end. A file that is not complete +//! must never look like a complete file. The atomic rename makes that sure. +//! +//! # How a transfer continues, and what the hash must survive +//! +//! When a transfer continues, an earlier process calculated the hash of the bytes on the disk. That +//! process is gone. So the module reads its own `.part` prefix again and builds the md5 state +//! again. Then it asks for the remainder. +//! +//! That costs one sequential read of the bytes that the disk already holds. It is much cheaper than +//! a second download of those bytes. It also occurs while the module waits for the server. +//! +//! The other method is to calculate the hash of the full file at the end. That method costs the +//! same read. But it can not start before the transfer stops. The method here also finds a bad +//! prefix. And the usual case, with no interruption, costs nothing. + +use crate::error::AppError; +use md5::{Digest, Md5}; +use serde::{Deserialize, Serialize}; +use std::path::{Path, PathBuf}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; + +/// How much effort to give one file. A test can supply its own policy. Then the test can examine +/// the failure path and does not wait for the full delay schedule of the application. +/// [`RetryPolicy::default`] is the policy that the application uses. +#[derive(Debug, Clone, Copy)] +pub struct RetryPolicy { + /// How many tries the module makes before it stops. Each try continues at the point where the + /// last try stopped. So this value limits *stalls* and not the total transfer time. The module + /// does not try again while a file continues to make progress. + pub attempts: u32, + /// If the module waits between two tries. `false` removes the delay. A test of the failure + /// path does not need the delay. + pub backoff: bool, +} + +impl Default for RetryPolicy { + fn default() -> Self { + RetryPolicy { + attempts: 5, + backoff: true, + } + } +} + +/// The delay before try *n*, in seconds: 2, 4, 8, 16. The delay has a limit. A node of a volunteer +/// must give a unit back. It must not hold a lease in a retry loop that has no end. +fn backoff_secs(attempt: u32) -> u64 { + 1u64 << attempt.min(4) +} + +/// How much disk space a unit with an alignment needs, as a factor on the size of the manifest. +/// The file arrives ready to read, so the node adds only its own analysis output. +const SPACE_MULTIPLE_ALIGNED: u64 = 3; + +/// How much disk space a unit with reads needs, as a factor on the size of the manifest. +/// +/// The factor is much larger here. The manifest names **compressed** reads, and the node then +/// writes three files that hold the same data in a different form. `mapped.bam` comes from the +/// reads. `sorted.bam` exists while `mapped.bam` is still on the disk, and the sort also spills to +/// the disk. For 30 GB of compressed reads, the peak is far above 90 GB. +/// +/// A value that is too small gives the exact failure that this check prevents. The disk fills in +/// the middle of a unit, after hours of work. +const SPACE_MULTIPLE_READS: u64 = 10; + +/// One file in a work unit's manifest, exactly as the AppView curated it. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct ManifestFile { + #[serde(default)] + pub run_accession: String, + pub url: String, + #[serde(default)] + pub index_url: Option, + #[serde(default)] + pub md5: Option, + #[serde(default)] + pub bytes: Option, + #[serde(default)] + pub format: String, + /// The instrument model that ENA reports, such as `Illumina NovaSeq 6000`. The node chooses a + /// mapper preset from it. Absent on a manifest that an older AppView made. + #[serde(default)] + pub instrument: Option, +} + +impl ManifestFile { + /// The file name this entry lands under, taken from the URL's last segment. + pub fn file_name(&self) -> &str { + self.url.rsplit('/').next().unwrap_or(&self.url) + } +} + +/// ENA gives a location with no scheme, such as `ftp.sra.ebi.ac.uk/vol1/...`. Get it with HTTPS. +/// +/// Do not use FTP. It is more difficult to continue an FTP transfer. A firewall on the network of a +/// volunteer frequently stops FTP. And ENA gives the same paths with HTTPS. This function does not +/// change a URL that already has a scheme. So a manifest can point to a different host. +pub fn to_https(url: &str) -> String { + let u = url.trim(); + if u.starts_with("http://") || u.starts_with("https://") { + u.to_string() + } else { + format!("https://{}", u.trim_start_matches("ftp://")) + } +} + +fn part_path(dest: &Path) -> PathBuf { + let mut s = dest.as_os_str().to_os_string(); + s.push(".part"); + PathBuf::from(s) +} + +fn hex(digest: &[u8]) -> String { + digest.iter().map(|b| format!("{b:02x}")).collect() +} + +/// Whether a downloaded file's checksum is acceptable. +/// +/// An entry with **no** md5 value is acceptable. ENA does not always publish one. If the module +/// refused such work, it would refuse most of the catalogue. The comparison ignores the letter +/// case. Hex letters have the same value in each case, and archives do not use one case only. +pub fn checksum_ok(expected: Option<&str>, actual: &str) -> bool { + match expected.map(str::trim).filter(|s| !s.is_empty()) { + Some(want) => want.eq_ignore_ascii_case(actual), + None => true, + } +} + +/// Refuse a unit that is too large for the disk. The check occurs before the first byte arrives. +/// +/// A full disk in the middle of a run is the worst result. The unit fails. The node holds the lease +/// until the lease ends. And the owner of the machine must remove the files. +/// +/// `free_space` gives zero when it can not measure the disk. A zero lets the try continue. A +/// refusal after a failed measurement is worse than a write that fails. +pub fn preflight_space(dir: &Path, manifest: &[ManifestFile]) -> Result<(), AppError> { + let total: u64 = manifest.iter().filter_map(|f| f.bytes).map(|b| b.max(0) as u64).sum(); + // A unit of reads needs much more room than a unit with an alignment. See the two constants. + let reads = manifest.iter().any(|f| f.format == "FASTQ"); + let multiple = if reads { + SPACE_MULTIPLE_READS + } else { + SPACE_MULTIPLE_ALIGNED + }; + let needed = total.saturating_mul(multiple); + let free = crate::realign_job::free_space(dir); + if !crate::realign_job::has_room(needed, free) { + return Err(AppError::Import(format!( + "not enough room for this work unit: about {} GB is needed and {} GB is free on {}", + needed / 1_000_000_000, + free / 1_000_000_000, + dir.display() + ))); + } + Ok(()) +} + +/// What one try must ask the server for. The bytes on the disk decide this. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Resume { + /// The disk holds nothing that the module can use. Get the full file. + FromStart, + /// Continue from this offset. + From(u64), + /// The `.part` file already has the expected size. Check it, and get no more bytes. + AlreadyComplete, +} + +/// Decide how to continue. The `.part` size and the expected total give the answer. +/// +/// A `.part` file that is **larger** than the expected total starts again at zero. The module does +/// not cut it to the correct length. Such a file is evidence that the disk holds a different file +/// from the one in the manifest. The cause can be a new revision of the file, or two files with the +/// same name. +/// +/// A cut to the correct length gives a file with the correct size but the wrong md5 value. The +/// module finds that only after a second full download. +pub fn resume_from(part_len: u64, expected: Option) -> Resume { + match expected { + Some(total) if part_len == total && total > 0 => Resume::AlreadyComplete, + Some(total) if part_len > total => Resume::FromStart, + _ if part_len == 0 => Resume::FromStart, + _ => Resume::From(part_len), + } +} + +/// Rebuild md5 state over an existing `.part` prefix. +async fn hash_prefix(path: &Path) -> Result<(Md5, u64), AppError> { + let mut file = tokio::fs::File::open(path).await.map_err(|e| io_err(path, e))?; + let mut hasher = Md5::new(); + let mut buf = vec![0u8; 1 << 20]; + let mut total = 0u64; + loop { + let n = file.read(&mut buf).await.map_err(|e| io_err(path, e))?; + if n == 0 { + break; + } + hasher.update(&buf[..n]); + total += n as u64; + } + Ok((hasher, total)) +} + +fn io_err(path: &Path, e: std::io::Error) -> AppError { + AppError::Import(format!("{}: {e}", path.display())) +} + +/// Get one file of the manifest into `dir`. This continues an interrupted transfer, and it checks +/// the md5 value. +/// +/// `progress` receives `(received, total)` as the bytes arrive. The `received` count includes the +/// prefix from the earlier try. A progress bar that starts again at zero would tell the user the +/// opposite of the truth. +pub async fn fetch_file( + client: &reqwest::Client, + dir: &Path, + entry: &ManifestFile, + cancel: &navigator_analysis::CancelToken, + progress: &mut (dyn FnMut(u64, Option) + Send), +) -> Result { + fetch_file_with(client, dir, entry, RetryPolicy::default(), cancel, progress).await +} + +/// [`fetch_file`] with an explicit retry policy. +pub async fn fetch_file_with( + client: &reqwest::Client, + dir: &Path, + entry: &ManifestFile, + retry: RetryPolicy, + cancel: &navigator_analysis::CancelToken, + progress: &mut (dyn FnMut(u64, Option) + Send), +) -> Result { + let dest = dir.join(entry.file_name()); + if dest.exists() { + return Ok(dest); // a completed file is never re-fetched; the rename is what makes it final + } + tokio::fs::create_dir_all(dir).await.map_err(|e| io_err(dir, e))?; + let part = part_path(&dest); + let url = to_https(&entry.url); + let expected = entry.bytes.filter(|b| *b > 0).map(|b| b as u64); + + let mut last: Option = None; + for attempt in 0..retry.attempts { + if cancel.is_cancelled() { + return Err(AppError::Import("cancelled".into())); + } + if attempt > 0 && retry.backoff { + tokio::time::sleep(std::time::Duration::from_secs(backoff_secs(attempt))).await; + } + match fetch_once(client, &url, &part, expected, cancel, progress).await { + Ok(actual) => { + if !checksum_ok(entry.md5.as_deref(), &actual) { + // The module does not try again from the bytes on the disk. Those bytes are + // wrong. A transfer that continues from them gives the same wrong result. So + // the next try must start at zero. + let _ = tokio::fs::remove_file(&part).await; + last = Some(AppError::Import(format!( + "checksum mismatch for {}: expected {}, got {actual}", + entry.file_name(), + entry.md5.as_deref().unwrap_or("?") + ))); + continue; + } + tokio::fs::rename(&part, &dest).await.map_err(|e| io_err(&dest, e))?; + return Ok(dest); + } + Err(e) => last = Some(e), + } + } + Err(last.unwrap_or_else(|| AppError::Import(format!("could not fetch {}", entry.file_name())))) +} + +/// One try. Returns the md5 value of the complete file, in lowercase hex. +async fn fetch_once( + client: &reqwest::Client, + url: &str, + part: &Path, + expected: Option, + cancel: &navigator_analysis::CancelToken, + progress: &mut (dyn FnMut(u64, Option) + Send), +) -> Result { + let part_len = tokio::fs::metadata(part).await.map(|m| m.len()).unwrap_or(0); + let plan = resume_from(part_len, expected); + + let (mut hasher, mut received) = match plan { + Resume::FromStart => (Md5::new(), 0), + Resume::From(_) | Resume::AlreadyComplete => hash_prefix(part).await?, + }; + if plan == Resume::AlreadyComplete { + return Ok(hex(&hasher.finalize())); + } + + let mut req = client.get(url); + if let Resume::From(offset) = plan { + req = req.header(reqwest::header::RANGE, format!("bytes={offset}-")); + } + let resp = req.send().await.map_err(|e| AppError::Import(format!("{url}: {e}")))?; + + // A `416` says that the range which this code asked for does not exist. There is one usual + // cause. The `.part` file already holds the whole file, and the manifest gave no size, so + // `resume_from` could not see that the file was complete. + // + // Remove the `.part` file and report an error. The next try then starts at zero and completes. + // Without this step, each try asks for the same range and receives the same `416`. The file + // never arrives, until a person removes that file by hand. + if resp.status() == reqwest::StatusCode::RANGE_NOT_SATISFIABLE { + let _ = tokio::fs::remove_file(part).await; + return Err(AppError::Import(format!( + "{url}: the server refused the range; the next try starts at the beginning" + ))); + } + let resp = resp + .error_for_status() + .map_err(|e| AppError::Import(format!("{url}: {e}")))?; + + // A server that ignores `Range` answers 200 and sends the full file. Accept that answer. Do + // not add those bytes to the bytes on the disk. Such a file has the correct size only by + // accident, and it fails the checksum. + let restart = matches!(plan, Resume::From(_)) && resp.status() != reqwest::StatusCode::PARTIAL_CONTENT; + if restart { + hasher = Md5::new(); + received = 0; + } + let total = expected.or_else(|| resp.content_length().map(|c| c + received)); + + let mut file = if received > 0 && !restart { + tokio::fs::OpenOptions::new() + .append(true) + .open(part) + .await + .map_err(|e| io_err(part, e))? + } else { + tokio::fs::File::create(part).await.map_err(|e| io_err(part, e))? + }; + + let mut resp = resp; + while let Some(chunk) = resp + .chunk() + .await + .map_err(|e| AppError::Import(format!("{url}: {e}")))? + { + if cancel.is_cancelled() { + // Keep the `.part` file. A later try needs those bytes. If the module removed the + // file, a cancel would discard all the work that the user already paid for. + file.flush().await.map_err(|e| io_err(part, e))?; + return Err(AppError::Import("cancelled".into())); + } + file.write_all(&chunk).await.map_err(|e| io_err(part, e))?; + hasher.update(&chunk); + received += chunk.len() as u64; + progress(received, total); + } + file.flush().await.map_err(|e| io_err(part, e))?; + Ok(hex(&hasher.finalize())) +} + +/// Get each file of a unit manifest into `dir`, one file after the other. +/// +/// The order is sequential by design. The connection of the volunteer is the limit. So parallel +/// transfers do not finish earlier. They also increase the peak disk use, and they put more load on +/// a public archive that helps us at no cost. +pub async fn fetch_unit( + client: &reqwest::Client, + dir: &Path, + manifest: &[ManifestFile], + cancel: &navigator_analysis::CancelToken, + progress: &mut (dyn FnMut(&str, u64, Option) + Send), +) -> Result, AppError> { + preflight_space(dir, manifest)?; + let mut out = Vec::with_capacity(manifest.len()); + for entry in manifest { + let name = entry.file_name().to_string(); + let mut per_file = |recv: u64, total: Option| progress(&name, recv, total); + out.push(fetch_file(client, dir, entry, cancel, &mut per_file).await?); + + // Get the index file beside the alignment, when ENA has one. + // + // The index is some MB, and the alignment is 10 to 30 GB. Without the index, the node + // reads the whole alignment one more time to make its own. So this small transfer removes + // a full pass over the largest file of the unit. + // + // A failure here is not a failure of the unit. The node makes the index itself, which + // costs time and gives the same result. + if let Some(index_url) = entry.index_url.clone().filter(|u| !u.trim().is_empty()) { + let sidecar = ManifestFile { + run_accession: entry.run_accession.clone(), + url: index_url, + index_url: None, + // ENA publishes no checksum for the index file, so there is nothing to compare. + md5: None, + bytes: None, + format: "INDEX".to_string(), + instrument: None, + }; + let name = sidecar.file_name().to_string(); + let mut per_file = |recv: u64, total: Option| progress(&name, recv, total); + let _ = fetch_file(client, dir, &sidecar, cancel, &mut per_file).await; + } + } + Ok(out) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn ena_paths_become_https_and_explicit_schemes_are_left_alone() { + assert_eq!( + to_https("ftp.sra.ebi.ac.uk/vol1/run/ERR/x.cram"), + "https://ftp.sra.ebi.ac.uk/vol1/run/ERR/x.cram" + ); + assert_eq!( + to_https("ftp://ftp.sra.ebi.ac.uk/vol1/x.cram"), + "https://ftp.sra.ebi.ac.uk/vol1/x.cram" + ); + assert_eq!(to_https("https://example.org/x.cram"), "https://example.org/x.cram"); + assert_eq!(to_https("http://example.org/x.cram"), "http://example.org/x.cram"); + } + + #[test] + fn the_file_name_comes_from_the_last_url_segment() { + let f = ManifestFile { + run_accession: "ERR1".into(), + url: "ftp.sra.ebi.ac.uk/vol1/run/ERR1/sample.cram".into(), + index_url: None, + md5: None, + bytes: None, + format: "CRAM".into(), + instrument: None, + }; + assert_eq!(f.file_name(), "sample.cram"); + } + + #[test] + fn resume_continues_from_what_is_already_there() { + assert_eq!(resume_from(0, Some(100)), Resume::FromStart); + assert_eq!(resume_from(40, Some(100)), Resume::From(40)); + assert_eq!(resume_from(100, Some(100)), Resume::AlreadyComplete); + } + + /// A `.part` file larger than the manifest total is evidence that the disk holds a different + /// file. A cut to the correct length gives the correct size and the wrong md5 value. The module + /// finds that only after a second full download. + #[test] + fn an_oversized_part_starts_over_rather_than_being_trimmed() { + assert_eq!(resume_from(140, Some(100)), Resume::FromStart); + } + + /// With no expected size, the module has no value to compare. So it continues the transfer of + /// a partial file. The checksum gives the final answer. + #[test] + fn an_unknown_total_still_resumes() { + assert_eq!(resume_from(40, None), Resume::From(40)); + assert_eq!(resume_from(0, None), Resume::FromStart); + } + + #[test] + fn a_missing_checksum_is_not_a_failure() { + assert!(checksum_ok(None, "d41d8cd98f00b204e9800998ecf8427e")); + assert!(checksum_ok(Some(""), "d41d8cd98f00b204e9800998ecf8427e")); + } + + #[test] + fn checksums_compare_without_regard_to_hex_case() { + assert!(checksum_ok( + Some("D41D8CD98F00B204E9800998ECF8427E"), + "d41d8cd98f00b204e9800998ecf8427e" + )); + assert!(!checksum_ok( + Some("d41d8cd98f00b204e9800998ecf8427e"), + "0bad0bad0bad0bad0bad0bad0bad0bad" + )); + } + + #[test] + fn backoff_grows_and_then_stops_growing() { + assert_eq!((1..=5).map(backoff_secs).collect::>(), vec![2, 4, 8, 16, 16]); + } + + /// The manifest comes from the curation query of the AppView. A FASTQ entry has no + /// `index_url`, because `jsonb_strip_nulls` removes that key. + #[test] + fn a_curated_manifest_entry_decodes() { + let json = r#"{"run_accession":"ERR2000001","url":"ftp.sra.ebi.ac.uk/vol1/s1.cram", + "index_url":"ftp.sra.ebi.ac.uk/vol1/s1.cram.crai", + "md5":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","bytes":12000000000,"format":"CRAM"}"#; + let f: ManifestFile = serde_json::from_str(json).unwrap(); + assert_eq!(f.file_name(), "s1.cram"); + assert_eq!(f.bytes, Some(12_000_000_000)); + + let stripped = r#"{"run_accession":"ERR2","url":"ftp/r_1.fastq.gz","md5":"b","bytes":9,"format":"FASTQ"}"#; + let f: ManifestFile = serde_json::from_str(stripped).unwrap(); + assert!(f.index_url.is_none(), "an absent sidecar must not fail to decode"); + } +} diff --git a/crates/navigator-app/src/grid.rs b/crates/navigator-app/src/grid.rs new file mode 100644 index 00000000..4ed746aa --- /dev/null +++ b/crates/navigator-app/src/grid.rs @@ -0,0 +1,340 @@ +//! The Grid client: `impl App` methods for the signed Grid Edge API of the AppView +//! (`/api/v1/grid/*`). +//! +//! A node announces itself, reserves work units, reports progress, gives a lease back, and sends a +//! signed result. The device key signs each call, as it does for the exchange client and the +//! recruitment client. This module uses the shared [`appview_post`](App::appview_post) and +//! [`appview_get_signed`](App::appview_get_signed) transport. +//! +//! The canonical strings are in [`navigator_sync::grid::messages`], which mirrors +//! `du_db::grid::messages` on the AppView. Design +//! `documents/design/distributed-compute-grid.md` §4.4 and §7.1. +//! +//! # What a signature covers, and why the digest is a hash +//! +//! A call that changes data signs `{ts}\n{base}` through +//! [`DeviceKey::sign_fresh`](navigator_sync::device_key::DeviceKey::sign_fresh). One signature then +//! holds the operation and the time. The AppView keeps each accepted signature for a short period. +//! It then refuses the same bytes a second time. +//! +//! The submit call signs the **hash** of the digest and sends the digest with it. The AppView +//! calculates the hash again from the body that arrives, and refuses a difference. +//! +//! The canonical bytes are what `serde_json` writes. This workspace does not use the +//! `preserve_order` feature. So the keys are in alphabetical order, and the output has no spaces. +//! The AppView uses the same crate with the same setting. That gives the smallest possible +//! agreement between the two repositories. There is no field order to agree, and no number format +//! rules. + +use super::*; +use crate::ena::ManifestFile; +use navigator_sync::grid::messages; + +/// A work unit as the AppView gives it at claim time. It holds everything that a node needs, so the +/// node asks ENA for nothing. +#[derive(Debug, Clone, serde::Deserialize)] +pub struct ClaimedUnit { + pub lease_id: i64, + pub work_unit_id: i64, + pub sample_accession: String, + #[serde(default)] + pub study_accession: Option, + /// `CRAM` for a unit that needs no new alignment, or `FASTQ` for a unit that the node maps. + pub data_kind: String, + #[serde(default)] + pub manifest: Vec, + #[serde(default)] + pub est_bases: Option, + #[serde(default)] + pub total_bytes: Option, + pub expires_at: chrono::DateTime, +} + +/// What a contributor has done, and where the contributor is on the public board. +#[derive(Debug, Clone, Default, serde::Deserialize)] +pub struct GridStanding { + #[serde(default)] + pub leases: Vec, + #[serde(default)] + pub agreed: i64, + #[serde(default)] + pub divergent: i64, + #[serde(default)] + pub cobblestones: f64, + #[serde(default)] + pub units_credited: i64, + /// The position on the board. It is `None` for a contributor with no credit. Such a + /// contributor has no row on the board, so a number here would be an answer to a question that + /// nobody asked. + #[serde(default)] + pub rank: Option, +} + +/// What this node can do. The AppView keeps it and uses it to select work. +#[derive(Debug, Clone, serde::Serialize)] +pub struct NodeCapabilities { + /// `["CRAM"]`, or `["CRAM","FASTQ"]` for a node that can map reads. + pub data_kinds: Vec, + pub threads: u32, + /// The disk space, in bytes, that the user gives to this work. + pub disk_budget: u64, + pub memory_bytes: u64, +} + +impl App { + /// Announce this node, or send its capabilities again. + /// + /// This call is also the heartbeat of the node. The AppView row keeps one `last_heartbeat` + /// value, and this call sets it. A second endpoint that writes the same row would let the two + /// values disagree. + pub async fn grid_register(&self, caps: &NodeCapabilities) -> Result { + let did = self.current_account().ok_or(AppError::NotAuthenticated)?; + let dev = self.ensure_device_key().await?; + let ts = chrono::Utc::now().timestamp(); + let caps_value = serde_json::to_value(caps).map_err(|e| AppError::Import(e.to_string()))?; + let caps_hash = canonical_sha256_b64(&caps_value); + let version = env!("CARGO_PKG_VERSION"); + let sig = dev.sign_fresh(ts, &messages::register(&did, version, &caps_hash)); + let body = serde_json::json!({ + "did": did, + "software_version": version, + "capabilities": caps_value, + "os_info": os_info(), + "ts": ts, + "signature": sig, + }); + let v = self.appview_post("grid/node/register", body).await?; + Ok(v.get("node_id").and_then(|x| x.as_i64()).unwrap_or_default()) + } + + /// Reserve up to `count` work units for `lease_secs` seconds. + /// + /// The node sends only the kinds that it can process. The AppView never gives a FASTQ unit to a + /// node that can not map reads. + /// + /// The result can hold fewer units than `count`, or none. An empty result is the usual answer + /// when the catalogue holds no more work of those kinds. It is not an error. + pub async fn grid_claim( + &self, + data_kinds: &[String], + count: i32, + lease_secs: i64, + ) -> Result, AppError> { + let did = self.current_account().ok_or(AppError::NotAuthenticated)?; + let dev = self.ensure_device_key().await?; + let ts = chrono::Utc::now().timestamp(); + let kinds = messages::normalize_kinds(data_kinds); + if kinds.is_empty() { + return Err(AppError::Import("this node advertises no data kinds".into())); + } + let sig = dev.sign_fresh(ts, &messages::claim(&did, &kinds, count, lease_secs)); + let body = serde_json::json!({ + "did": did, + // The list goes on the wire in the same form that the signature covers. + "data_kinds": kinds.split(',').collect::>(), + "count": count, + "lease_secs": lease_secs, + "ts": ts, + "signature": sig, + }); + let v = self.appview_post("grid/claim", body).await?; + let units = v.get("units").cloned().unwrap_or_else(|| serde_json::json!([])); + serde_json::from_value(units).map_err(|e| AppError::Import(e.to_string())) + } + + /// Report progress on a lease that this node holds. + /// + /// Returns `false` when the lease is no longer the lease of this node. The node must then stop + /// work on that unit. Without this answer, a node can spend hours on a unit that it lost, and + /// it receives no credit for that work. + /// + /// This call does **not** make the lease longer. A node can send a heartbeat and still not + /// finish. Such a node would hold a unit for ever. A limited lease prevents that fault. + pub async fn grid_heartbeat(&self, lease_id: i64, stage: &str, fraction: Option) -> Result { + let did = self.current_account().ok_or(AppError::NotAuthenticated)?; + let dev = self.ensure_device_key().await?; + let ts = chrono::Utc::now().timestamp(); + let sig = dev.sign_fresh(ts, &messages::heartbeat(&did, lease_id, stage)); + let body = serde_json::json!({ + "did": did, + "lease_id": lease_id, + "stage": stage, + "progress": { "stage": stage, "fraction": fraction }, + "ts": ts, + "signature": sig, + }); + let v = self.appview_post("grid/heartbeat", body).await?; + // Only an explicit `false` means that this node lost the lease. + // + // An absent field, a new name for it, or a value of another type gives `None` here. An + // earlier version read each of those as `false`, and the node then stopped work of many + // hours. + // + // The lease has its own time limit. So the safe answer to an unclear reply is to continue. + // At worst, the node finishes a unit that another node also finished. + Ok(v.get("held").and_then(|x| x.as_bool()).unwrap_or(true)) + } + + /// Give a lease back with no result, so another node can take the unit immediately. + /// + /// A second call for the same lease is safe. It returns `false`, and that is not an error: a + /// node that sends the call again after a lost answer did nothing wrong. + pub async fn grid_release(&self, lease_id: i64, reason: &str) -> Result { + let did = self.current_account().ok_or(AppError::NotAuthenticated)?; + let dev = self.ensure_device_key().await?; + let ts = chrono::Utc::now().timestamp(); + let sig = dev.sign_fresh(ts, &messages::release(&did, lease_id, reason)); + let body = serde_json::json!({ + "did": did, + "lease_id": lease_id, + "reason": reason, + "ts": ts, + "signature": sig, + }); + let v = self.appview_post("grid/release", body).await?; + Ok(v.get("released").and_then(|x| x.as_bool()).unwrap_or(false)) + } + + /// Send the result of a unit, and close the lease that made it. + /// + /// `digest` holds **raw** values. The AppView puts the continuous values into groups when it + /// compares two results. A client that made the groups itself would put the group rule into two + /// repositories. A difference between them would then give `DIVERGENT` results against nodes + /// that did nothing wrong, and the message would give no cause. + /// + /// A second call for the same unit replaces the first result. It does not add a second vote. + #[allow(clippy::too_many_arguments)] + pub async fn grid_submit( + &self, + work_unit_id: i64, + lease_id: Option, + digest: &serde_json::Value, + stack_version: &str, + reference_build: &str, + aligner: Option<&str>, + record_refs: &[String], + ) -> Result { + let did = self.current_account().ok_or(AppError::NotAuthenticated)?; + let dev = self.ensure_device_key().await?; + let ts = chrono::Utc::now().timestamp(); + let hash = canonical_sha256_b64(digest); + // Two signatures, for two different purposes. The request signature proves who sent this + // call now. The digest signature stays in the row. A later check can then prove which node + // made this result, after the AppView copies it to other places. + let sig = dev.sign_fresh(ts, &messages::submit(&did, work_unit_id, &hash)); + let digest_sig = dev.sign(&canonical_bytes_string(digest)); + let body = serde_json::json!({ + "did": did, + "work_unit_id": work_unit_id, + "lease_id": lease_id, + "digest": digest, + "digest_sig": digest_sig, + "stack_version": stack_version, + "reference_build": reference_build, + "aligner": aligner, + "record_refs": record_refs, + "ts": ts, + "signature": sig, + }); + let v = self.appview_post("grid/submit", body).await?; + Ok(v.get("submission_id").and_then(|x| x.as_i64()).unwrap_or_default()) + } + + /// The leases, the history and the board position of this node. + pub async fn grid_standing(&self) -> Result { + self.appview_get_signed("grid/mine", messages::poll, &[]).await + } +} + +/// What the digest signature covers: the canonical bytes of the digest, as text. +/// +/// `serde_json` writes the keys in alphabetical order and adds no spaces, because this workspace +/// does not use the `preserve_order` feature. The AppView uses the same crate with the same +/// setting, so both sides make the same bytes with no rules to agree. +fn canonical_bytes_string(value: &serde_json::Value) -> String { + serde_json::to_string(value).unwrap_or_default() +} + +/// The SHA-256 of the canonical bytes of a JSON value, as standard base64. +/// +/// This must give the same answer as `du_db::grid::digest::canonical_sha256_b64` on the AppView. +/// The submit handler calculates it again from the body that arrives, and refuses a difference. +pub(crate) fn canonical_sha256_b64(value: &serde_json::Value) -> String { + use base64::Engine as _; + use sha2::{Digest as _, Sha256}; + let bytes = serde_json::to_vec(value).unwrap_or_default(); + base64::engine::general_purpose::STANDARD.encode(Sha256::digest(bytes)) +} + +/// A short description of this machine, for the fleet view. +fn os_info() -> String { + format!("{} {}", std::env::consts::OS, std::env::consts::ARCH) +} + +#[cfg(test)] +mod tests { + use super::*; + + /// The hash must not change with the order of the keys in the source text. If it did, a node + /// and the AppView could disagree about what the signature covers. + #[test] + fn the_canonical_hash_ignores_key_order() { + let a = serde_json::json!({"calls": {"sex": "XY", "y_terminal": "R-A"}, "unit": "SAMEA1"}); + let b = serde_json::json!({"unit": "SAMEA1", "calls": {"y_terminal": "R-A", "sex": "XY"}}); + assert_eq!(canonical_sha256_b64(&a), canonical_sha256_b64(&b)); + } + + /// A different result must give a different hash, or the check has no value. + #[test] + fn a_different_result_gives_a_different_hash() { + let a = serde_json::json!({"calls": {"sex": "XY"}}); + let b = serde_json::json!({"calls": {"sex": "XX"}}); + assert_ne!(canonical_sha256_b64(&a), canonical_sha256_b64(&b)); + } + + /// The bytes that the digest signature covers are the bytes that go on the wire. + #[test] + fn the_signed_bytes_are_the_bytes_that_are_sent() { + let v = serde_json::json!({"b": 2, "a": 1}); + assert_eq!(canonical_bytes_string(&v), r#"{"a":1,"b":2}"#); + } + + /// A unit as the AppView sends it, with the manifest that `grid-curate` made. + #[test] + fn a_claimed_unit_decodes_with_its_manifest() { + let json = serde_json::json!({ + "lease_id": 7, + "work_unit_id": 12, + "sample_accession": "SAMEA0000001", + "study_accession": "PRJEB00000", + "data_kind": "CRAM", + "manifest": [{ + "run_accession": "ERR0000001", + "url": "ftp.sra.ebi.ac.uk/vol1/s1.cram", + "md5": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "bytes": 12000000000i64, + "format": "CRAM" + }], + "est_bases": 90000000000i64, + "total_bytes": 12000000000i64, + "expires_at": "2026-08-28T00:00:00Z" + }); + let u: ClaimedUnit = serde_json::from_value(json).expect("decode"); + assert_eq!(u.manifest.len(), 1); + assert_eq!(u.manifest[0].file_name(), "s1.cram"); + assert_eq!(u.data_kind, "CRAM"); + } + + /// A contributor with no credit has no board position. The field must arrive as `None` and not + /// as a number. + #[test] + fn an_uncredited_contributor_has_no_rank() { + let json = serde_json::json!({ + "leases": [], "agreed": 0, "divergent": 0, + "cobblestones": 0.0, "units_credited": 0, "rank": null + }); + let s: GridStanding = serde_json::from_value(json).expect("decode"); + assert!(s.rank.is_none()); + assert_eq!(s.agreed, 0); + } +} diff --git a/crates/navigator-app/src/grid_job.rs b/crates/navigator-app/src/grid_job.rs new file mode 100644 index 00000000..3a9d7031 --- /dev/null +++ b/crates/navigator-app/src/grid_job.rs @@ -0,0 +1,1447 @@ +//! The unit driver: the work that a volunteer node does for one Grid work unit. +//! +//! ```text +//! claim ─► preflight ─► fetch from ENA ─► import ─► analyze ─► digest ─► submit ─► clean +//! ``` +//! +//! Each step is a method that already exists somewhere in this crate. This module puts them in +//! order, reports progress, and makes sure that a lease always ends. Design +//! `documents/design/distributed-compute-grid.md` §7.2. +//! +//! # A lease must always end +//! +//! A node that stops in the middle keeps a unit out of the catalogue until the lease time ends. So +//! each path out of [`App::run_grid_unit`] gives the lease back. Success closes the lease through +//! the submit call, and each failure sends a release call. The AppView also has a reaper for the node that +//! disappears. But a node that is still alive must not need it. +//! +//! # What this node can do +//! +//! [`supported_data_kinds`] gives the list that the node advertises, and it is the only place that +//! decides. A node never receives work that this module can not do, because the AppView selects +//! work with that same list. Today the list holds `CRAM` only. See [`map_reads`] for what a FASTQ +//! unit needs. + +use super::*; +use crate::ena::{self, ManifestFile}; +use crate::grid::ClaimedUnit; +use du_domain::fed::Provenance; +use navigator_analysis::CancelToken; +use std::path::{Path, PathBuf}; +use std::sync::{Arc, Mutex}; + +/// The stage that a node is on. The node sends this with each heartbeat, and the fleet view of the +/// AppView shows it. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum GridStage { + Fetch, + Import, + Map, + Analyze, + Ancestry, + Publish, + Submit, +} + +impl GridStage { + pub fn as_str(self) -> &'static str { + match self { + GridStage::Fetch => "fetch", + GridStage::Import => "import", + GridStage::Map => "map", + GridStage::Analyze => "analyze", + GridStage::Ancestry => "ancestry", + GridStage::Publish => "publish", + GridStage::Submit => "submit", + } + } +} + +/// What one unit gave. +#[derive(Debug, Clone)] +pub struct UnitOutcome { + pub sample_accession: String, + /// The submission id, when the node sent a result. + pub submission_id: Option, + /// Why the unit did not finish. The node then gives the lease back. + pub error: Option, +} + +/// How often a node tells the AppView that it is alive, while it works on a unit. +/// +/// The value is far below the shortest lease. A node that stops between two beats is still inside +/// its lease, so a lost beat costs nothing. +const HEARTBEAT_EVERY: std::time::Duration = std::time::Duration::from_secs(60); + +/// How a node contributes. +#[derive(Debug, Clone)] +pub struct GridJobParams { + /// How many units to take in one claim call. + pub max_units: i32, + /// How long to hold each lease, in seconds. + pub lease_secs: i64, + /// Where the files of a unit go. Each unit gets its own directory below this one. + pub scratch_root: PathBuf, + /// The build that the analysis is against, such as `chm13v2.0`. + pub reference_build: String, +} + +/// The data kinds that this node can process. These are the kinds that it advertises. +/// +/// A `CRAM` unit arrives with an alignment that a laboratory already made. The node imports that +/// file and analyzes it. +/// +/// A `FASTQ` unit holds reads only, so the node maps them first. See [`map_unit_reads`]. +/// +/// This list is the one place that decides. The AppView selects work with it, so a node never +/// receives work that this module can not do. +pub fn supported_data_kinds() -> Vec { + vec!["CRAM".to_string(), "FASTQ".to_string()] +} + +/// Sort the read files of a FASTQ unit into the first mate file, the second mate file, and the +/// reads with no mate. +/// +/// ENA gives the mate number in the file name, as `_1` and `_2` before the extension. A run with +/// one file only is a set of reads with no mate, and a long-read run is always such a set. +fn split_mates(files: &[PathBuf]) -> (Option, Option, Vec) { + let (mut r1, mut r2, mut singles) = (None, None, Vec::new()); + for f in files { + let name = f.file_name().and_then(|n| n.to_str()).unwrap_or_default(); + // `_1.` and not `_1`: a run accession such as `ERR1_1.fastq.gz` must not match on the + // accession itself. + if name.contains("_1.") && r1.is_none() { + r1 = Some(f.clone()); + } else if name.contains("_2.") && r2.is_none() { + r2 = Some(f.clone()); + } else { + singles.push(f.clone()); + } + } + (r1, r2, singles) +} + +/// Map the reads of a FASTQ unit to the target build, and give back a finished CRAM file. +/// +/// # Why this does not call the realignment job +/// +/// The realignment job (`realign_job`) does the same four operations, and the first plan was to +/// call it here. A reading of that module changed the plan. That module holds its stages together +/// with the machinery that continues a job which stopped: `Resumed`, `ScratchState`, and the rules +/// about which file each stage may remove. A comment in that file records a fault in exactly those +/// rules. That fault destroyed a 59 GB file and about four hours of work. +/// +/// A Grid unit wants none of that machinery. It has no source alignment, so there is no revert +/// stage. It does not continue a job that stopped, because a unit that fails gives its lease back +/// and another node takes it from the start. And it registers no alignment against a source row. +/// The only common part is the four operations below, and each one is already public. +/// +/// So this function calls those four operations directly. That leaves the realignment module +/// exactly as `v0.1.0-alpha.17` validated it on a full genome. The other method was to divide that +/// module along its most dangerous line, with no way to run that validation again here. +async fn map_unit_reads( + app: &App, + files: &[PathBuf], + manifest: &[ManifestFile], + dir: &Path, + target_build: &str, + cancel: &CancelToken, + report: &mut (dyn FnMut(GridStage, &str) + Send), +) -> Result { + use navigator_analysis::postprocess::{self, MarkDupParams, SortParams}; + + // The map step below decides pairing from whether both mates are present. The preset is a + // separate question, and it comes from the instrument. + let (r1, r2, singles) = split_mates(files); + + // The preset comes from the **instrument**, and not from the count of mates. + // + // An earlier version chose a short-read preset for a set with two mates, and a HiFi preset for + // each other set. A single-end Illumina run has no mate, so that rule mapped short reads under + // a long-read preset. The comment on that rule gave the result. A map under the wrong preset + // does not fail, and it gives alignments that look correct and are wrong. + // + // `Preset::infer` gives an error for an instrument that it does not know, and this function + // passes that error on. A refusal is the correct answer. The unit then goes to another node, + // and no result of unknown quality reaches the quorum. + let instrument = manifest.iter().find_map(|m| m.instrument.clone()); + let preset = navigator_align::Preset::infer(None, instrument.as_deref()).map_err(|e| { + AppError::Import(format!( + "this node can not choose a mapper for the reads of this unit: {e}" + )) + })?; + + report(GridStage::Map, "reference"); + let reference = app.resolve_reference(target_build, &mut |_, _| {}).await?; + + report(GridStage::Map, "index"); + let index = { + let (build, reference) = (target_build.to_string(), reference.clone()); + let batch = navigator_align::batch::BatchSize::for_this_machine(); + tokio::task::spawn_blocking(move || { + navigator_align::index::ensure_index( + &navigator_align::index::cache_root(), + &build, + &reference, + preset, + batch, + &mut |_, _| {}, + ) + }) + .await + .map_err(|e| AppError::Join(e.to_string()))?? + }; + + let mapped = dir.join("mapped.bam"); + let sorted = dir.join("sorted.bam"); + let marked = dir.join("marked.bam"); + let output = dir.join("aligned.cram"); + + report(GridStage::Map, "map"); + { + let (out, work) = (mapped.clone(), dir.join("map")); + let token = cancel.clone(); + let map_params = navigator_align::MapParams { + preset, + threads: 0, + read_group: None, + format: navigator_align::OutputFormat::Bam, + reference: None, + }; + let (r1c, r2c, singlesc) = (r1.clone(), r2.clone(), singles.clone()); + tokio::task::spawn_blocking(move || -> Result<(), AppError> { + let cancelled = move || token.is_cancelled(); + if let (Some(a), Some(b)) = (&r1c, &r2c) { + navigator_align::map_pairs(&index, a, b, &out, &work, &map_params, &cancelled, &mut |_, _, _| {})?; + } else { + let single = singlesc + .first() + .or(r1c.as_ref()) + .ok_or_else(|| AppError::Import("the unit holds no read file".into()))?; + navigator_align::map_reads(&index, single, &out, &work, &map_params, &cancelled, &mut |_, _, _| {})?; + } + Ok(()) + }) + .await + .map_err(|e| AppError::Join(e.to_string()))??; + } + // The reads have no more use, and a set of read files for a whole genome is tens of GB. + for f in files { + let _ = std::fs::remove_file(f); + } + + report(GridStage::Map, "sort"); + { + let (input, out, work) = (mapped.clone(), sorted.clone(), dir.join("sort")); + let token = cancel.clone(); + tokio::task::spawn_blocking(move || { + postprocess::sort_alignment(&input, &out, &work, &SortParams::default(), &token, &mut |_| {}) + }) + .await + .map_err(|e| AppError::Join(e.to_string()))??; + } + let _ = std::fs::remove_file(&mapped); + + report(GridStage::Map, "duplicates"); + { + let (input, out) = (sorted.clone(), marked.clone()); + let token = cancel.clone(); + // A long-read library usually needs no PCR step, and two long reads rarely have the same + // end points. So a mark on those reads removes real coverage. + // A long-read library usually needs no PCR step, and two long reads rarely have the same + // end points. So a mark on those reads removes real coverage. The test is on the preset and + // not on the mate count, because a single-end short-read run still wants the mark. + let md_params = MarkDupParams { + enabled: preset == navigator_align::Preset::ShortRead, + ..Default::default() + }; + tokio::task::spawn_blocking(move || { + postprocess::mark_duplicates(&input, &out, &md_params, &token, &mut |_| {}) + }) + .await + .map_err(|e| AppError::Join(e.to_string()))??; + } + let _ = std::fs::remove_file(&sorted); + + report(GridStage::Map, "compress"); + let finalized = { + let (input, out) = (marked.clone(), output.clone()); + tokio::task::spawn_blocking(move || postprocess::finalize_bam(&input, &out)) + .await + .map_err(|e| AppError::Join(e.to_string()))?? + }; + Ok(finalized.bam) +} + +/// The values that go into the digest of a result. +/// +/// Each one is optional. A sample can have no Y chromosome. The autosomal consensus of a fresh +/// sample can be absent. An absent value is not a failure of the unit, and the AppView compares two +/// absent values as equal. +#[derive(Debug, Clone, Default, PartialEq)] +pub struct UnitResults { + pub sex: Option, + pub y_terminal: Option, + pub ancestry_superpop_argmax: Option, + pub coverage_mean: Option, + pub callable_fraction: Option, +} + +/// Build the digest that the node signs and sends. +/// +/// The values are **raw**. The AppView puts the continuous values into groups when it compares two +/// results. A node that made the groups itself would put the group rule into two repositories. +/// +/// There is no `mt_terminal` field. `App::analyze_biosample` does not give an mtDNA value, because +/// that value is not final on CHM13, and the Grid analyzes against CHM13. A digest can not ask for +/// a value that the analysis does not make. See design §12.3. +pub fn build_digest( + sample_accession: &str, + reference_build: &str, + stack_version: &str, + aligner: Option<&str>, + r: &UnitResults, +) -> serde_json::Value { + let mut calls = serde_json::Map::new(); + if let Some(v) = &r.sex { + calls.insert("sex".into(), serde_json::json!(v)); + } + if let Some(v) = &r.y_terminal { + calls.insert("y_terminal".into(), serde_json::json!(v)); + } + if let Some(v) = &r.ancestry_superpop_argmax { + calls.insert("ancestry_superpop_argmax".into(), serde_json::json!(v)); + } + if let Some(v) = r.coverage_mean { + calls.insert("coverage_mean".into(), serde_json::json!(v)); + } + if let Some(v) = r.callable_fraction { + calls.insert("callable_fraction".into(), serde_json::json!(v)); + } + serde_json::json!({ + "unit": sample_accession, + "reference_build": reference_build, + "stack_version": stack_version, + "aligner": aligner, + "calls": serde_json::Value::Object(calls), + }) +} + +/// The key that the `provenance` block takes in a published record. +/// +/// It is the serde name of the field on the record types of `du_domain::fed`. A test below makes a +/// record with the typed method and then reads the key back. So a test checks this value against +/// the type, and this value is not an assumption. +const PROVENANCE_KEY: &str = "provenance"; + +/// Put the Grid provenance block into a record that a builder already made. +/// +/// The record builders of `publish.rs` serve the ordinary path, where a user publishes a record +/// about their own genome. Such a record carries no provenance, and those builders have thirteen +/// call sites. A new argument on each builder would put a `None` at each of those call sites, for a +/// value that only the Grid supplies. That `None` would mean nothing to any of them. +/// +/// So the Grid adds the block after the builder finishes. The value comes from the typed +/// [`Provenance`] of `du_domain::fed`, so the shape and the field names come from the shared +/// contract. Only the key is a string here, and a test checks that string against the type. +fn attach_provenance(mut value: serde_json::Value, p: &Provenance) -> Result { + let block = serde_json::to_value(p).map_err(|e| AppError::Import(e.to_string()))?; + match &mut value { + serde_json::Value::Object(map) => { + map.insert(PROVENANCE_KEY.to_string(), block); + Ok(value) + } + _ => Err(AppError::Import("a published record must be a JSON object".into())), + } +} + +/// The provenance of a result that this node computed for the Grid. +fn grid_provenance(did: &str, reference_build: &str, aligner: Option<&str>) -> Provenance { + Provenance::new( + did, + "navigator", + env!("CARGO_PKG_VERSION"), + reference_build, + // How the input arrived. It names the public origin, so a later reader can honour any term + // that the study of that sample sets. + "ena:read_run", + ) + .with_aligner(aligner.map(str::to_string)) +} + +/// The accession as a directory name, or `None` when it is not a safe name. +/// +/// The accession arrives from the AppView, and the code makes a path from it and later **removes +/// that path and everything below it**. A value with `..` in it would leave the scratch directory, +/// and the remove would then delete a directory of the user. +/// +/// The AppView is not an attacker. But a value from a server is still a value from outside, and a +/// recursive delete is on the other side of it. Each archive that this code reads gives an +/// accession of the form `[A-Za-z0-9_.-]+`, so this check refuses nothing real. +fn safe_dir_name(accession: &str) -> Option<&str> { + let a = accession.trim(); + let ok = !a.is_empty() + && a.len() <= 64 + && a != "." + && a != ".." + && !a.contains("..") + && a.chars() + .all(|c| c.is_ascii_alphanumeric() || matches!(c, '_' | '.' | '-')); + ok.then_some(a) +} + +/// The value that the digest carries for the sex of a sample. +/// +/// The mapping is explicit, and it does not use the `Debug` form of the enum. The AppView compares +/// this value between two nodes. A new name for a variant would change the value, and no other +/// thing would change. Two versions of Navigator would then disagree about one sample, and no +/// reader could find the cause. +/// +/// An uncertain result gives `None`, and the digest then holds no sex value. That is honest: two +/// nodes that both could not tell agree, and a node that could tell does not agree with one that +/// could not. +fn sex_for_digest(sex: navigator_analysis::sex::InferredSex) -> Option { + use navigator_analysis::sex::InferredSex; + match sex { + InferredSex::Male => Some("XY".to_string()), + InferredSex::Female => Some("XX".to_string()), + InferredSex::Unknown => None, + } +} + +/// How many sequencing runs a manifest holds. +/// +/// The count is of the **run accessions** and not of the files. A single run with two mates gives +/// two files. ENA frequently gives a third file, for the reads of that run that lost their mate. A +/// count of files would refuse such a run as though it held three runs. +/// +/// An entry with no run accession counts as one run. So a manifest with no accession at all is one +/// run. That is the safe reading: the node does the work, and it does not refuse a unit because a +/// field was empty. +fn runs_in(manifest: &[ManifestFile]) -> usize { + let named: std::collections::BTreeSet<&str> = manifest + .iter() + .map(|m| m.run_accession.trim()) + .filter(|a| !a.is_empty()) + .collect(); + named.len().max(1) +} + +/// A short class for the release call, from the error of a unit. +/// +/// The message of an error can hold a local file path. The server keeps this value and the node +/// signs it, so it must hold no name from the machine of the volunteer. +fn release_reason(e: &AppError) -> &'static str { + let text = e.to_string().to_ascii_lowercase(); + if text.contains("cancel") { + "stopped" + } else if text.contains("checksum") { + "checksum" + } else if text.contains("room") || text.contains("space") { + "disk" + } else if text.contains("merge runs") { + "unsupported" + } else if text.contains("analysis") { + "analysis" + } else { + "error" + } +} + +/// The primary data file of a unit: the alignment for a CRAM unit, or the first read file for a +/// FASTQ unit. An index file is never the primary file. +fn primary_file<'a>(manifest: &'a [ManifestFile], files: &'a [PathBuf]) -> Option<&'a PathBuf> { + manifest + .iter() + .zip(files) + .find(|(m, _)| matches!(m.format.as_str(), "CRAM" | "BAM" | "FASTQ")) + .map(|(_, p)| p) +} + +/// The free space of the volume that holds `dir`, in bytes. Zero means that the code can not +/// measure it. +pub fn free_space_for(dir: &Path) -> u64 { + crate::realign_job::free_space(dir) +} + +/// The physical memory of this machine, in bytes. Zero means that the code can not measure it. +pub fn machine_memory_bytes() -> u64 { + navigator_align::batch::detect_memory().map(|m| m.total).unwrap_or(0) +} + +/// The file that marks a directory as one that this node made. +/// +/// [`sweep_old_scratch`] deletes a directory and everything below it. It must delete only what +/// this node created, and this file is the proof of that. +const UNIT_MARKER: &str = ".navigator-grid-unit"; + +/// Make the directory of a unit, and mark it as one that this node made. +async fn make_unit_dir(dir: &Path) -> Result<(), AppError> { + tokio::fs::create_dir_all(dir) + .await + .map_err(|e| AppError::Import(format!("{}: {e}", dir.display())))?; + let _ = tokio::fs::write(dir.join(UNIT_MARKER), b"navigator grid unit\n").await; + Ok(()) +} + +/// Remove each **unit directory** below `scratch_root` that is older than `max_age`. +/// +/// A unit that the user stopped keeps its files, so that the transfer can continue. This removes +/// the directories that no run continued. Call it when a node starts. +/// +/// # What this will not delete +/// +/// `--scratch` takes any path that the user gives. An earlier version of this function removed +/// **each** directory below that path that was old enough. +/// +/// Take `navigator contribute --scratch ~/genomes`. That version deleted each directory in +/// `~/genomes` that nobody had touched for a week. It did that at the start, before the node did +/// any work at all. +/// +/// Two conditions now guard each delete. The name must be a name that this node would make +/// ([`safe_dir_name`]), and the directory must hold the marker file that this node writes. A +/// directory of the user has neither, so this function passes over it. +pub async fn sweep_old_scratch(scratch_root: &Path, max_age: std::time::Duration) -> usize { + let Ok(mut entries) = tokio::fs::read_dir(scratch_root).await else { + return 0; + }; + let mut removed = 0; + while let Ok(Some(entry)) = entries.next_entry().await { + let path = entry.path(); + let named_by_us = path + .file_name() + .and_then(|n| n.to_str()) + .is_some_and(|n| safe_dir_name(n).is_some()); + if !named_by_us || !path.join(UNIT_MARKER).exists() { + continue; + } + let old = entry + .metadata() + .await + .ok() + .and_then(|m| m.modified().ok()) + .and_then(|t| t.elapsed().ok()) + .is_some_and(|age| age > max_age); + if old && tokio::fs::remove_dir_all(&path).await.is_ok() { + removed += 1; + } + } + removed +} + +impl App { + /// Do the work of one unit, and always give the lease back. + /// + /// The node reports each stage through `report`, and the caller sends those to the AppView as a + /// heartbeat. A heartbeat that answers `false` means that this node lost the lease. The node + /// then stops, because it receives no credit for more work on that unit. + pub async fn run_grid_unit( + &self, + unit: &ClaimedUnit, + params: &GridJobParams, + cancel: &CancelToken, + report: &mut (dyn FnMut(GridStage, &str) + Send), + ) -> UnitOutcome { + let mut outcome = UnitOutcome { + sample_accession: unit.sample_accession.clone(), + submission_id: None, + error: None, + }; + // The subject that the unit makes, so the code can remove it at the end. See + // [`discard_unit_subject`]. + let subject = Arc::new(Mutex::new(None)); + let Some(safe_name) = safe_dir_name(&unit.sample_accession) else { + let e = AppError::Import(format!( + "the accession \"{}\" is not a name that this node will make a directory from", + unit.sample_accession + )); + outcome.error = Some(e.to_string()); + let _ = self.grid_release(unit.lease_id, release_reason(&e)).await; + return outcome; + }; + let dir = params.scratch_root.join(safe_name); + if let Err(e) = make_unit_dir(&dir).await { + outcome.error = Some(e.to_string()); + let _ = self.grid_release(unit.lease_id, release_reason(&e)).await; + return outcome; + } + + // The stage that the beat reports. The work writes this value and the beat reads it. So + // the AppView shows the stage that the node is on now, and not the first stage. + let stage = Arc::new(Mutex::new(GridStage::Fetch)); + // Set by the beat when the AppView says that another node holds this lease. It separates + // that event from a stop by the user, and the two want different treatment of the files. + let lease_lost = Arc::new(std::sync::atomic::AtomicBool::new(false)); + + // The work and the beat run together. Neither one can be a separate task, because both use + // `&self`, and a task needs a value that lives for the whole program. `select!` needs no + // such value: it drives two futures that borrow the same data. + let result = { + let stage_for_work = Arc::clone(&stage); + let mut record = |s: GridStage, detail: &str| { + if let Ok(mut cur) = stage_for_work.lock() { + *cur = s; + } + report(s, detail); + }; + let work = self.grid_unit_inner(unit, params, &dir, cancel, &mut record, &subject); + let beat = self.beat_while_working(unit.lease_id, &stage, cancel, &lease_lost); + tokio::pin!(work); + tokio::pin!(beat); + tokio::select! { + r = &mut work => r, + lost = &mut beat => { + // The beat already set the token. Wait for the work to see it and return. + // + // A `select!` that ends here would drop the work in the middle of an `await`. + // The heavy stages run in `spawn_blocking`. A dropped handle does not stop such + // a task. So the sort or the duplicate mark would continue, while the code + // below removes the directory that it writes into. The code waits instead, and + // the work stops at its next test of the token. + // Take the result of the work when it has one. + // + // The work can reach the submit call inside this window and complete. A node + // that reported a failure then would lose the credit for work that it + // finished. It would also call release on a lease that the submit call had + // already closed. + match (&mut work).await { + Ok(id) => Ok(id), + Err(_) => Err(lost), + } + } + } + }; + + match result { + Ok(id) => outcome.submission_id = Some(id), + Err(e) => { + outcome.error = Some(e.to_string()); + // The unit goes back to the catalogue at once. Without this call, it waits for the + // full lease time, and no other node can take it. + // + // The reason that goes to the server is a short class and not the full message. + // The full message can hold a local path, because `ena` puts the path of a file in + // the text of an I/O error. The server keeps the reason, and the node signs it. So + // a path in that text would send the directory names of a volunteer to a public + // service. The full message stays here, in `outcome.error`. + let _ = self.grid_release(unit.lease_id, release_reason(&e)).await; + } + } + // A unit is work, and it is not the data of the user. Remove the subject before the files, + // because the subject names those files. + let made = subject.lock().ok().and_then(|g| *g); + if let Some(guid) = made { + self.discard_unit_subject(guid).await; + } + + // Keep the files of a unit that **the user** stopped. Remove them in each other case. + // + // `ena` can continue a transfer that stopped. It keeps each `.part` file, and it reads the + // md5 state of that prefix again. A remove of the directory here would make all of that + // work impossible: a stop at 25 GB of a 30 GB file would lose those 25 GB. + // + // A unit that failed is different. Another node takes it, and this node may never see it + // again. So those files stay on the disk with no purpose, and a whole genome is tens of GB. + // + // [`sweep_old_scratch`] removes a directory that a stop left, after some days. Without that + // step, a user who stops a run and never continues it keeps those files for ever. + // + // The beat also cancels this token, when another node takes the lease. That is not a stop + // by the user: this node never sees that unit again, so its files have no purpose. The + // caller says which of the two occurred. + let user_stopped = cancel.is_cancelled() && !lease_lost.load(std::sync::atomic::Ordering::Relaxed); + if !user_stopped { + let _ = tokio::fs::remove_dir_all(&dir).await; + } + outcome + } + + /// Remove the subject that a unit made, with each row and each cached result below it. + /// + /// **A Grid unit must leave no subject in the workspace.** The subject exists only because the + /// analysis works on a subject. Its alignment names a file in the temporary directory of the + /// unit, and that directory goes away at the end of the unit. A subject that stayed would name + /// a file that is not there. + /// + /// Without this step, a node that contributes for one week puts some thousands of such + /// subjects among the true subjects of its owner. Each one holds no data that a person can use, + /// and each one is difficult to tell from a real subject. The result of the unit is already + /// safe: the digest went to the AppView, and the records went to the publish queue. + /// + /// The delete of a subject refuses while the subject holds data, so this removes each sequence + /// run first. A failure gives no message to the user, because the unit is already complete. A + /// subject that stays is a fault for a later version to correct. It is not a reason to report + /// a unit as failed. + async fn discard_unit_subject(&self, guid: SampleGuid) { + if let Ok(runs) = self.list_sequence_runs(guid).await { + for run in runs { + let _ = self.delete_sequence_run(run.id).await; + } + } + let _ = self.delete_biosample(guid).await; + } + + /// Tell the AppView that this node is alive, until the unit ends. + /// + /// This future never finishes on its own. It ends when the work beside it finishes, and + /// `select!` then drops it. It returns only when the node **loses** the lease, which is a + /// reason to stop the work at once. + /// + /// A node that lost its lease receives no credit for more work on that unit. Without this + /// check, such a node can spend hours on a unit that another node already finished. The value + /// that the AppView sends back is the only way for the node to learn that. + async fn beat_while_working( + &self, + lease_id: i64, + stage: &Arc>, + cancel: &CancelToken, + lease_lost: &Arc, + ) -> AppError { + loop { + tokio::time::sleep(HEARTBEAT_EVERY).await; + if cancel.is_cancelled() { + // The work stops by itself. This future must not end the unit with an error that + // hides the true reason. + continue; + } + let now = stage.lock().map(|s| *s).unwrap_or(GridStage::Analyze); + match self.grid_heartbeat(lease_id, now.as_str(), None).await { + Ok(true) => {} + Ok(false) => { + lease_lost.store(true, std::sync::atomic::Ordering::Relaxed); + cancel.cancel(); + return AppError::Import("another node now holds this unit".into()); + } + // A beat that did not arrive is not proof that the lease is gone. The network of a + // volunteer is not always available, and the work continues. The lease has its own + // time limit, and the AppView reclaims it if this node truly stopped. + Err(_) => {} + } + } + } + + async fn grid_unit_inner( + &self, + unit: &ClaimedUnit, + params: &GridJobParams, + dir: &Path, + cancel: &CancelToken, + report: &mut (dyn FnMut(GridStage, &str) + Send), + subject: &Arc>>, + ) -> Result { + // A sample with more than one run needs each run mapped and then all of them merged into + // one alignment. There is no merge stage here yet. + // + // The check occurs **before** the fetch, on purpose. The manifest arrives with the claim, + // so the node knows the count at no cost. + // + // An earlier version took the first file and ignored the others. It pulled every one of + // them first, from a public archive that gives us its bandwidth at no charge. It then gave + // a coverage value from one part of the sample, as a value for the whole sample. + // More files than the code reads is the same fault as more runs than the code reads. A + // single run with two files that have no mate gives a coverage value from one of them. + let data_files = unit + .manifest + .iter() + .filter(|m| matches!(m.format.as_str(), "CRAM" | "BAM" | "FASTQ")) + .count(); + let usable = if unit.data_kind == "FASTQ" { 3 } else { 1 }; + if runs_in(&unit.manifest) > 1 || data_files > usable { + return Err(AppError::Import(format!( + "{} holds more data files than this node reads, and it can not merge them yet", + unit.sample_accession + ))); + } + + // ---- fetch ---- + report(GridStage::Fetch, &unit.sample_accession); + let client = self.auth.http.clone(); + // Report only when the whole number of percent changes. + // + // `ena` calls its progress function once for each chunk of the answer, and a chunk is some + // tens of KB. So a file of 30 GB gives some hundreds of thousands of calls. Each + // call here made a line on the screen and took a lock. The screen then held more lines than + // a person can read, and the work went slower. + let mut last_pct = u64::MAX; + let mut on_bytes = |name: &str, recv: u64, total: Option| { + let pct = total.filter(|t| *t > 0).map(|t| recv * 100 / t).unwrap_or(0); + if pct != last_pct { + last_pct = pct; + report(GridStage::Fetch, &format!("{name} {pct}%")); + } + }; + let files = ena::fetch_unit(&client, dir, &unit.manifest, cancel, &mut on_bytes).await?; + let primary = primary_file(&unit.manifest, &files) + .ok_or_else(|| AppError::Import(format!("unit {} has no data file", unit.sample_accession)))?; + + // A FASTQ unit needs a map stage that does not exist yet. The node must never reach this + // point, because `supported_data_kinds` does not advertise FASTQ. The check stays, because + // a wrong advertisement must give a clear message and not a strange failure much later. + let aligned = if unit.data_kind == "FASTQ" { + map_unit_reads( + self, + &files, + &unit.manifest, + dir, + ¶ms.reference_build, + cancel, + report, + ) + .await? + } else { + primary.clone() + }; + + // ---- import ---- + report(GridStage::Import, &unit.sample_accession); + // The ENA accession is the identity of the subject. It is a public catalogue id and not + // personal data, so it is safe as the name that a user sees. + let biosample = self + .add_biosample(None, &unit.sample_accession, Some(unit.sample_accession.clone()), None) + .await?; + // Record the subject at once, and before the import. A failure in any step after this + // point must still remove it. + if let Ok(mut g) = subject.lock() { + *g = Some(biosample.guid); + } + // The ENA accession as an external id. This is what makes the published record *about* + // the public sample. + // + // `BiosampleRecord` carries no accession field. That rule keeps personal data out of a + // published record. The record reads `external_ids` from this table instead. + // + // Without this call, the record goes out with an empty `externalIds`. No reader can then + // connect it to the sample, or join it with the record of a second contributor. + self.add_external_id(biosample.guid, "ENA", &unit.sample_accession) + .await?; + self.add_data(biosample.guid, &aligned).await?; + + // Make the coordinate index when the alignment has none. + // + // The fetch step takes the index of ENA when ENA has one, and that path costs least. ENA + // does not always publish one, and that fetch can fail with no result. + // + // Without an index, each step that asks for a region fails. `analyze_biosample` puts those + // failures in `errors`, and the check below then fails the unit. That occurs **after** the + // walk over the whole file already succeeded. So the index comes first. + report(GridStage::Import, "index"); + for aln in self.list_alignments_for_biosample(biosample.guid).await? { + if let Err(e) = self.ensure_alignment_index(aln.id, |_, _| {}).await { + return Err(AppError::Import(format!( + "{}: no coordinate index, and this node could not make one: {e}", + unit.sample_accession + ))); + } + } + + // ---- analyze ---- + report(GridStage::Analyze, &unit.sample_accession); + let analyzed = self.analyze_biosample(&biosample, cancel.clone()).await?; + if !analyzed.had_alignment { + return Err(AppError::Import(format!( + "no alignment for {} after import", + unit.sample_accession + ))); + } + // `analyze_biosample` puts the failure of one step in `errors` and still gives `Ok`. That + // is correct for a batch over the subjects of a user, where the other steps still give a + // result that a person can use. It is **not** correct here. + // + // A step that failed leaves its value out of the digest. The agreement test compares an + // absent value with an absent value as equal. So two nodes that both failed would agree, + // reach a quorum on a result with no content, and receive credit for it. A unit must fail + // instead, and another node then does the work. + if !analyzed.errors.is_empty() { + return Err(AppError::Import(format!( + "analysis of {} did not complete: {}", + unit.sample_accession, + analyzed.errors.join("; ") + ))); + } + // A stop is **not** a failure that `errors` records. `analyze_biosample` gives `Ok` with an + // empty `errors` when a stop ends it, because a stop is not a fault of the sample. + // + // So the check above passes after a stop, and an earlier version continued. It built the + // autosomal consensus, which is a second pass over the whole genome and which takes no + // token of its own. It then published the records, and it sent a digest with the values of + // the steps that had finished. Two nodes that each stopped at the same step would agree on + // that digest. + // + // A stop also has to stop the node. Without this test, Ctrl-C left the node at work for + // hours, and a lost lease still sent a result. + if cancel.is_cancelled() { + return Err(AppError::Import(format!( + "the analysis of {} stopped before it finished", + unit.sample_accession + ))); + } + + let mut results = UnitResults::default(); + let alignments = self.list_alignments_for_biosample(biosample.guid).await?; + + // The build that the calls are truly against. + // + // A `CRAM` unit is a passthrough. The submitter of that file chose its build, and that + // build is GRCh37 or GRCh38 for most of the archive. Nothing here maps it again. The header + // probe reads the true build during the import, and the row keeps it. + // + // An earlier version reported `params.reference_build` for each unit. That value is the + // default of the command line. So a result on GRCh38 went to the AppView as a result on + // CHM13. + // + // The AppView compares two results only when the build agrees. So such a result + // joined a group of true CHM13 results. It then compared the coverage and the Y value of + // two different references as one measurement. + let reference_build = alignments + .first() + .map(|a| a.reference_build.clone()) + .unwrap_or_else(|| params.reference_build.clone()); + + if let Some(aln) = alignments.first() { + if let Some(cov) = self.cached_coverage(aln.id).await? { + results.coverage_mean = Some(cov.mean_coverage); + if cov.genome_territory > 0 { + results.callable_fraction = Some(cov.callable_bases as f64 / cov.genome_territory as f64); + } + } + if let Some(sex) = self.cached_sex(aln.id).await? { + results.sex = sex_for_digest(sex.inferred_sex); + } + } + // The Y value that the digest carries is the **genome-level** label, and not the call of + // one alignment. + // + // `analyze_biosample` builds the Y profile of the subject, and `consensus_label` on that + // profile is the answer of the application for this sample. `haplogroup_calls(...).first()` + // gives the call of one walk, ordered by row id. That call is a step on the way to the + // label, and it is frequently a node higher in the tree. An earlier version sent that + // shallower value, so the Grid published an answer that this same application would not + // give for the same sample. + // + // A female sample carries no Y value at all. The unit makes its subject with no sex value. + // So the guard that stops a Y placement for a female subject can not fire, and the walk + // then places noise. `results.sex` holds the measured value at this point, and it decides. + let female = results.sex.as_deref() == Some("XX"); + results.y_terminal = if female { + None + } else { + let placed = + navigator_store::consensus_profile::get(self.store.pool(), biosample.guid, DnaType::Y.as_str()) + .await? + .and_then(|p| p.consensus_label) + .filter(|s| !s.is_empty()); + match placed { + Some(label) => Some(label), + // No profile means that the placement did not run, or did not finish. Take the + // call of the walk, because a value is better than none. The check on + // `analyzed.errors` above already failed the unit for a step that gave an error. + None => self + .haplogroup_calls(biosample.guid, DnaType::Y) + .await? + .first() + .map(|c| c.haplogroup.clone()), + } + }; + + // ---- ancestry ---- + // + // The autosomal consensus of a new sample can be absent, and the estimate then fails. That + // is not a failure of the unit. The digest holds no ancestry value, and two results with no + // ancestry value still agree. A unit that failed here would waste the hours of analysis + // that are already complete. + report(GridStage::Ancestry, &unit.sample_accession); + // The autosomal consensus must exist before the estimate can run. Nothing else in a unit + // builds it, so an earlier version left `ancestry_superpop_argmax` absent for **every** + // unit, and this stage only printed a label. + // + // The build genotypes the alignment at the full panel, which is a second pass over the + // whole genome. That is a real cost for a volunteer, and it is not optional. + // + // **The setup of a node must never change the content of a digest.** Two honest nodes that + // analyze one sample must send the same set of keys. + // + // Take a flag that adds or removes the ancestry value. One node then sends a key that the + // other node does not send. The AppView reads two correct results as a disagreement. So + // this step runs for every unit, or the unit fails. + self.build_autosomal_profile(biosample.guid).await.map_err(|e| { + AppError::Import(format!( + "{}: could not build the autosomal consensus: {e}", + unit.sample_accession + )) + })?; + let ancestry = self + .estimate_ancestry_from_consensus(biosample.guid) + .await + .map_err(|e| AppError::Import(format!("{}: could not estimate ancestry: {e}", unit.sample_accession)))?; + results.ancestry_superpop_argmax = ancestry + .super_population_summary + .iter() + .max_by(|x, y| x.percentage.total_cmp(&y.percentage)) + .map(|s| s.super_population.clone()); + + let aligner = (unit.data_kind == "FASTQ").then_some("minimap2-pure-rs"); + + // ---- publish ---- + // + // The records go to the repository of the contributor, and each one carries the provenance + // block. That block is what makes a record *about* a public sample that nobody owns while + // it is *made by* this node. See design §5.1. + // + // A failure here does not fail the unit. The analysis is complete and its digest is the + // thing that the quorum reads. The records are the full result behind that digest, and the + // outbox sends them again later. A unit that failed here would discard hours of work + // because a network call did not answer. + report(GridStage::Publish, &unit.sample_accession); + let record_refs = self + .publish_grid_records(&biosample, &results, &reference_build, aligner) + .await + .unwrap_or_default(); + + // ---- submit ---- + report(GridStage::Submit, &unit.sample_accession); + let stack_version = env!("CARGO_PKG_VERSION"); + let digest = build_digest( + &unit.sample_accession, + &reference_build, + stack_version, + aligner, + &results, + ); + self.grid_submit( + unit.work_unit_id, + Some(unit.lease_id), + &digest, + stack_version, + &reference_build, + aligner, + &record_refs, + ) + .await + } + + /// Put the records of a finished unit in the publish queue, and give back the `at://` address + /// of each one. + /// + /// The queue is the durable path that the rest of the application uses. It repeats a call that + /// failed. A second publish of the same record replaces the first record, and adds no second + /// record. A volunteer machine goes offline, and a direct write would then lose records that a + /// queue keeps. + /// + /// Each record here uses a **fixed** record key. That key gives the address of the record + /// before the write occurs. So this method can give those addresses to the submit call in the + /// same run, and it does not wait for the queue to empty. + async fn publish_grid_records( + &self, + biosample: &Biosample, + results: &UnitResults, + reference_build: &str, + aligner: Option<&str>, + ) -> Result, AppError> { + let did = self.require_account()?; + let prov = grid_provenance(&did, reference_build, aligner); + let mut refs = Vec::new(); + + // The biosample record is the anchor. It carries the ENA accession as an external id. + // That id makes the record about the public sample, and not about this contributor. + let anchor = attach_provenance(self.biosample_record(&did, biosample.guid).await?, &prov)?; + self.enqueue_publish( + "biosample", + &format!("biosample:{}", biosample.guid), + NS_BIOSAMPLE, + Some(&biosample_rkey(biosample.guid)), + anchor, + ) + .await?; + refs.push(biosample_at_uri(&did, biosample.guid)); + + // The coverage record holds the measurements behind the digest. A digest says that two + // nodes agree; this record says what they agree about. + // + // A failure on one record must not discard the records that already went in the queue. + // + // `coverage_record` gives an error in two cases. The first is an alignment with no cached + // coverage. The second is a file that names a whole genome while its reads cover chrY only. + // Both occur on real ENA samples. + // + // An earlier version used `?` here. One such error then gave an empty list, and the + // submission named **no** record at all. It did not even name the anchor, which was + // already in the queue and which the AppView was going to publish. + for aln in self.list_alignments_for_biosample(biosample.guid).await? { + if results.coverage_mean.is_none() { + break; + } + let built = match self.coverage_record(&did, aln.id).await { + Ok(v) => attach_provenance(v, &prov), + Err(e) => Err(e), + }; + let Ok(value) = built else { continue }; + if self + .enqueue_publish( + "coverage", + &format!("alignment:{}", aln.id), + NS_ALIGNMENT, + Some(&alignment_rkey(aln.id)), + value, + ) + .await + .is_ok() + { + refs.push(format!("at://{did}/{NS_ALIGNMENT}/{}", alignment_rkey(aln.id))); + } + } + Ok(refs) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn full() -> UnitResults { + UnitResults { + sex: Some("Male".into()), + y_terminal: Some("R-FGC29071".into()), + ancestry_superpop_argmax: Some("EUR".into()), + coverage_mean: Some(30.4), + callable_fraction: Some(0.9412), + } + } + + #[test] + fn the_digest_holds_the_raw_values() { + let d = build_digest("SAMEA1", "chm13v2.0", "1.7.0", None, &full()); + assert_eq!(d["unit"], "SAMEA1"); + assert_eq!(d["reference_build"], "chm13v2.0"); + assert_eq!(d["calls"]["y_terminal"], "R-FGC29071"); + // Raw, and not in a group. The AppView makes the groups, so the rule has one home. + assert_eq!(d["calls"]["coverage_mean"], 30.4); + assert_eq!(d["calls"]["callable_fraction"], 0.9412); + } + + /// The analysis path does not give an mtDNA value on CHM13, so the digest must not ask for one. + #[test] + fn the_digest_has_no_mtdna_field() { + let d = build_digest("SAMEA1", "chm13v2.0", "1.7.0", None, &full()); + assert!(d["calls"].get("mt_terminal").is_none()); + assert!(!d.to_string().contains("mt_terminal")); + } + + /// An absent value is absent from the digest. It is not `null`. Two results that both have no + /// Y value then agree, and a result with a Y value does not agree with one that has none. + #[test] + fn an_absent_value_is_left_out_and_not_sent_as_null() { + let d = build_digest("SAMEA1", "chm13v2.0", "1.7.0", None, &UnitResults::default()); + assert!(d["calls"].as_object().unwrap().is_empty()); + assert!(!d.to_string().contains("null") || d["aligner"].is_null()); + } + + /// A unit with no new alignment names no mapper. That value records how the node made the + /// result, and a later check reads it. + #[test] + fn a_passthrough_unit_names_no_mapper() { + let d = build_digest("SAMEA1", "chm13v2.0", "1.7.0", None, &full()); + assert!(d["aligner"].is_null()); + let d = build_digest("SAMEA1", "chm13v2.0", "1.7.0", Some("minimap2-pure-rs"), &full()); + assert_eq!(d["aligner"], "minimap2-pure-rs"); + } + + /// This node advertises each kind that it can process, and no other. The AppView selects work + /// with this list, so a wrong entry here gives a node work that it can not do. + #[test] + fn the_node_advertises_each_kind_that_it_can_do() { + let kinds = supported_data_kinds(); + assert!(kinds.contains(&"CRAM".to_string()), "a unit with an alignment"); + assert!(kinds.contains(&"FASTQ".to_string()), "a unit with reads only"); + } + + /// ENA names the two mate files with `_1` and `_2` before the extension. + #[test] + fn the_two_mate_files_are_found_by_name() { + let f = |n: &str| PathBuf::from(format!("/x/{n}")); + let (r1, r2, singles) = split_mates(&[f("ERR1_1.fastq.gz"), f("ERR1_2.fastq.gz")]); + assert_eq!(r1, Some(f("ERR1_1.fastq.gz"))); + assert_eq!(r2, Some(f("ERR1_2.fastq.gz"))); + assert!(singles.is_empty()); + } + + /// A run with one file has reads with no mate, and a long-read run is always such a run. + #[test] + fn one_file_gives_reads_with_no_mate() { + let f = PathBuf::from("/x/ERR1.fastq.gz"); + let (r1, r2, singles) = split_mates(std::slice::from_ref(&f)); + assert!(r1.is_none()); + assert!(r2.is_none()); + assert_eq!(singles, vec![f]); + } + + /// The match is on `_1.` and not on `_1`. A run accession can hold those two characters, and a + /// file that matched on the accession would go to the wrong mate. + #[test] + fn the_mate_match_does_not_read_the_accession() { + let f = PathBuf::from("/x/ERR1_1_1.fastq.gz"); + let (r1, _, singles) = split_mates(std::slice::from_ref(&f)); + assert_eq!(r1, Some(f), "the mate marker is the one before the extension"); + assert!(singles.is_empty()); + } + + /// `PROVENANCE_KEY` must be the serde name of the field on the record types. This test makes a + /// record with the typed method and then reads the key back, so it checks the string against + /// the type. A new name in `du-domain` then fails here. Without this test, it would give a + /// record that the AppView reads and does not understand, with no message. + #[test] + fn the_provenance_key_matches_the_shared_type() { + let rec = du_domain::fed::BiosampleRecord::new(None, None, None, None, "2026-08-25T00:00:00Z") + .with_provenance(Some(grid_provenance("did:plc:x", "chm13v2.0", None))); + let value = serde_json::to_value(&rec).expect("serialize"); + assert!( + value.get(PROVENANCE_KEY).is_some(), + "the typed record wrote its provenance under a different key: {value}" + ); + } + + /// The block that this module adds must equal the block that the typed method writes. If the + /// two differ, a Grid record and an ordinary record carry different shapes for one idea. + #[test] + fn the_added_block_equals_the_block_that_the_type_writes() { + let prov = grid_provenance("did:plc:x", "chm13v2.0", Some("minimap2-pure-rs")); + let typed = du_domain::fed::BiosampleRecord::new(None, None, None, None, "2026-08-25T00:00:00Z") + .with_provenance(Some(prov.clone())); + let from_type = serde_json::to_value(&typed).unwrap()[PROVENANCE_KEY].clone(); + + let plain = serde_json::to_value(du_domain::fed::BiosampleRecord::new( + None, + None, + None, + None, + "2026-08-25T00:00:00Z", + )) + .unwrap(); + let added = attach_provenance(plain, &prov).unwrap()[PROVENANCE_KEY].clone(); + + assert_eq!(added, from_type); + } + + /// A passthrough unit names no mapper, and the block then has no `aligner` key at all. That is + /// a fact about how the node made the result, and not a value that is missing. + #[test] + fn provenance_from_a_passthrough_unit_names_no_mapper() { + let p = grid_provenance("did:plc:x", "chm13v2.0", None); + assert!(p.aligner.is_none()); + let v = serde_json::to_value(&p).unwrap(); + assert!(v.get("aligner").is_none(), "an absent mapper is left out: {v}"); + assert_eq!(v["computedBy"], "did:plc:x"); + assert_eq!(v["source"], "ena:read_run"); + } + + /// A record that is not an object can not take a provenance block. That is a fault in the + /// builder, and it must give an error and not a record with no provenance. + #[test] + fn a_record_that_is_not_an_object_is_refused() { + let p = grid_provenance("did:plc:x", "chm13v2.0", None); + assert!(attach_provenance(serde_json::json!("not a record"), &p).is_err()); + } + + /// The digest must hold the same set of keys for two honest nodes. A value that one node can + /// produce and another can not would read as a disagreement between two correct results. + #[test] + fn the_digest_keys_do_not_depend_on_the_node() { + let full = UnitResults { + sex: Some("XY".into()), + y_terminal: Some("R-A".into()), + ancestry_superpop_argmax: Some("EUR".into()), + coverage_mean: Some(30.0), + callable_fraction: Some(0.94), + }; + let d = build_digest("SAMEA1", "chm13v2.0", "1.7.0", None, &full); + let keys: Vec<&str> = d["calls"].as_object().unwrap().keys().map(|k| k.as_str()).collect(); + assert_eq!( + keys, + vec![ + "ancestry_superpop_argmax", + "callable_fraction", + "coverage_mean", + "sex", + "y_terminal" + ], + "a unit that finishes sends these five keys and no other" + ); + } + + /// A sample with no Y chromosome carries no Y value. An `XX` result and a Y branch name in one + /// digest would be two statements that contradict each other. + #[test] + fn a_female_sample_carries_no_y_value() { + let female = UnitResults { + sex: Some("XX".into()), + y_terminal: None, + ancestry_superpop_argmax: Some("EUR".into()), + coverage_mean: Some(30.0), + callable_fraction: Some(0.94), + }; + let d = build_digest("SAMEA1", "chm13v2.0", "1.7.0", None, &female); + assert_eq!(d["calls"]["sex"], "XX"); + assert!(d["calls"].get("y_terminal").is_none()); + } + + /// The sex value on the wire is an explicit string, and it is not the `Debug` form of the enum. + #[test] + fn the_sex_value_is_explicit_and_uncertain_gives_none() { + use navigator_analysis::sex::InferredSex; + assert_eq!(sex_for_digest(InferredSex::Male).as_deref(), Some("XY")); + assert_eq!(sex_for_digest(InferredSex::Female).as_deref(), Some("XX")); + assert_eq!( + sex_for_digest(InferredSex::Unknown), + None, + "an uncertain result gives no value, and not the word Unknown" + ); + } + + /// This test refuses an accession that would leave the scratch directory. A recursive delete + /// runs on the path that such a name builds. + #[test] + fn an_accession_that_escapes_the_scratch_directory_is_refused() { + assert_eq!(safe_dir_name("SAMEA0000001"), Some("SAMEA0000001")); + assert_eq!(safe_dir_name(" ERR1_1 "), Some("ERR1_1")); + assert!(safe_dir_name("../../etc").is_none()); + assert!(safe_dir_name("..").is_none()); + assert!(safe_dir_name("a/b").is_none()); + assert!(safe_dir_name("").is_none()); + assert!(safe_dir_name(&"x".repeat(65)).is_none()); + } + + /// The sweep must never remove a directory of the user. `--scratch` takes any path, so the + /// sweep runs where the user pointed it. It removes only what this node made. + #[tokio::test] + async fn the_sweep_passes_over_a_directory_that_this_node_did_not_make() { + let root = std::env::temp_dir().join(format!("navigator-sweep-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&root); + std::fs::create_dir_all(&root).unwrap(); + + // A directory of the user: a good name, but no marker. + let theirs = root.join("SAMEA9999999"); + std::fs::create_dir_all(&theirs).unwrap(); + std::fs::write(theirs.join("precious.cram"), b"do not delete").unwrap(); + + // A directory of this node: the same shape, with the marker. + let ours = root.join("SAMEA0000001"); + make_unit_dir(&ours).await.unwrap(); + + // Age zero, so nothing is old enough yet. + assert_eq!(sweep_old_scratch(&root, std::time::Duration::from_secs(3600)).await, 0); + // Now with no minimum age, so each one that the guard permits goes. + let removed = sweep_old_scratch(&root, std::time::Duration::ZERO).await; + assert_eq!(removed, 1, "only the directory of this node"); + assert!(theirs.exists(), "the directory of the user must stay"); + assert!(theirs.join("precious.cram").exists()); + assert!(!ours.exists()); + + let _ = std::fs::remove_dir_all(&root); + } + + /// The mapper preset comes from the instrument. A single-end Illumina run has no mate, and an + /// earlier rule mapped it under a long-read preset for that reason alone. + #[test] + fn the_preset_comes_from_the_instrument_and_not_from_the_mate_count() { + use navigator_align::Preset; + assert_eq!( + Preset::infer(None, Some("Illumina NovaSeq 6000")).unwrap(), + Preset::ShortRead + ); + assert_eq!(Preset::infer(None, Some("PacBio Revio")).unwrap(), Preset::MapHifi); + assert_eq!( + Preset::infer(None, Some("Oxford Nanopore PromethION")).unwrap(), + Preset::MapOnt + ); + } + + /// An instrument that the code does not know gives an error, and the unit then goes to another + /// node. A guess would give alignments that look correct and are wrong. + #[test] + fn an_unknown_instrument_is_refused_and_not_guessed() { + use navigator_align::Preset; + assert!(Preset::infer(None, Some("Some New Sequencer 9000")).is_err()); + assert!(Preset::infer(None, None).is_err()); + } + + /// A run with two mates and a file of reads that lost their mate is **one** run. ENA gives + /// three files for such a run, and a count of files would refuse it. + #[test] + fn three_files_of_one_run_are_one_run() { + let f = |run: &str, name: &str| ManifestFile { + run_accession: run.into(), + url: format!("ftp/{name}"), + index_url: None, + md5: None, + bytes: None, + format: "FASTQ".into(), + instrument: None, + }; + let one_run = vec![ + f("ERR1", "ERR1_1.fastq.gz"), + f("ERR1", "ERR1_2.fastq.gz"), + f("ERR1", "ERR1.fastq.gz"), + ]; + assert_eq!(runs_in(&one_run), 1); + + let two_runs = vec![f("ERR1", "ERR1_1.fastq.gz"), f("ERR2", "ERR2_1.fastq.gz")]; + assert_eq!(runs_in(&two_runs), 2); + } + + /// An empty accession must not refuse the unit. The safe reading is one run. + #[test] + fn a_manifest_with_no_accession_counts_as_one_run() { + let f = ManifestFile { + run_accession: String::new(), + url: "ftp/x.cram".into(), + index_url: None, + md5: None, + bytes: None, + format: "CRAM".into(), + instrument: None, + }; + assert_eq!(runs_in(std::slice::from_ref(&f)), 1); + assert_eq!(runs_in(&[]), 1); + } + + /// The release reason that goes to the server must carry no local path. The message of an + /// error can hold one, because the fetch module puts the path of a file in its error text. + #[test] + fn the_release_reason_carries_no_local_path() { + let leaky = AppError::Import( + "/Users/someone/Library/navigator-grid/SAMEA1/x.cram.part: No space left on device".into(), + ); + let reason = release_reason(&leaky); + assert_eq!(reason, "disk"); + assert!(!reason.contains('/'), "a path must never reach the server"); + assert!(!reason.contains("Users")); + } + + /// Each class is short, and each one tells the operator of the AppView something different. + #[test] + fn each_failure_gives_its_own_short_class() { + let r = |m: &str| release_reason(&AppError::Import(m.into())); + assert_eq!(r("cancelled"), "stopped"); + assert_eq!(r("checksum mismatch for x.cram"), "checksum"); + assert_eq!(r("not enough room for this work unit"), "disk"); + assert_eq!(r("this node can not merge runs yet"), "unsupported"); + assert_eq!(r("analysis of SAMEA1 did not complete"), "analysis"); + assert_eq!(r("something else"), "error"); + } + + /// An index file is never the primary file of a unit. + #[test] + fn the_index_file_is_not_the_primary_file() { + let m = |fmt: &str, url: &str| ManifestFile { + run_accession: "ERR1".into(), + url: url.into(), + index_url: None, + md5: None, + bytes: None, + format: fmt.into(), + instrument: None, + }; + let manifest = vec![m("CRAI", "a.cram.crai"), m("CRAM", "a.cram")]; + let files = vec![PathBuf::from("/x/a.cram.crai"), PathBuf::from("/x/a.cram")]; + assert_eq!(primary_file(&manifest, &files).unwrap(), &PathBuf::from("/x/a.cram")); + } +} diff --git a/crates/navigator-app/src/lib.rs b/crates/navigator-app/src/lib.rs index 9d6ad8d5..8e51a59f 100644 --- a/crates/navigator-app/src/lib.rs +++ b/crates/navigator-app/src/lib.rs @@ -3090,8 +3090,11 @@ pub use blocktree::COLLAPSE_MIN_RUN; mod brief; mod commands; mod dm; +pub mod ena; mod fastpath; mod ftdna_import; +pub mod grid; +pub mod grid_job; mod haplogroup; mod ibd_exchange; mod import_profiles; diff --git a/crates/navigator-app/src/realign_job.rs b/crates/navigator-app/src/realign_job.rs index a1cefc8a..ab90e391 100644 --- a/crates/navigator-app/src/realign_job.rs +++ b/crates/navigator-app/src/realign_job.rs @@ -748,7 +748,7 @@ fn log_buffer(stage: &str, bytes: usize) { /// A `free` value of 0 means that the platform gave no answer, and the function then permits the /// job. A job of many hours must not stop because a call for the free space failed. It is better to /// run that job and let it fail on a real write. -fn has_room(needed: u64, free: u64) -> bool { +pub(crate) fn has_room(needed: u64, free: u64) -> bool { free == 0 || free >= needed } @@ -757,7 +757,7 @@ fn has_room(needed: u64, free: u64) -> bool { /// /// A zero means "unknown", and the preflight then permits the job. A refusal, because a call for the /// free space failed, is worse than a job that runs and then fails on a real write. -fn free_space(path: &Path) -> u64 { +pub(crate) fn free_space(path: &Path) -> u64 { // Walk up to the nearest existing ancestor: the scratch directory itself may not exist yet. let mut probe = path; loop { diff --git a/crates/navigator-app/tests/ena_download.rs b/crates/navigator-app/tests/ena_download.rs new file mode 100644 index 00000000..3085f285 --- /dev/null +++ b/crates/navigator-app/tests/ena_download.rs @@ -0,0 +1,260 @@ +//! Tests of the ENA download module against a true HTTP server. +//! +//! The module exists because `refgenome::download` can not continue an interrupted transfer. A +//! mock server can not test that ability with enough care. The important behaviour is the +//! answer of the code to a `Range` request, to a `206` status and to a `200` status. It is also the +//! md5 value across a prefix that this process did not get. +//! +//! A stub that answers as the caller expects agrees with the code at each point where both are +//! wrong. So these tests use a temporary HTTP/1.1 server on the loopback address. It is about forty +//! lines, it adds no dependency, and it can give a wrong answer on purpose. + +use navigator_analysis::CancelToken; +use navigator_app::ena::{self, ManifestFile, RetryPolicy}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpListener; + +/// A scratch directory with a unique name. It removes itself. +/// +/// The workspace has no `tempfile` dependency. The usual method here is a fixed name under +/// `std::env::temp_dir()`. That is the cause of the flake in `a_present_file_resolves` when two +/// test runs occur together. A unique name for each test costs nothing and prevents that fault. +struct Scratch(std::path::PathBuf); + +impl Scratch { + fn new(tag: &str) -> Self { + static N: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); + let unique = format!( + "navigator-ena-{tag}-{}-{}", + std::process::id(), + N.fetch_add(1, Ordering::Relaxed) + ); + let path = std::env::temp_dir().join(unique); + let _ = std::fs::remove_dir_all(&path); + std::fs::create_dir_all(&path).expect("scratch dir"); + Scratch(path) + } + fn path(&self) -> &std::path::Path { + &self.0 + } +} + +impl Drop for Scratch { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.0); + } +} + +/// How the test server should behave. +#[derive(Clone, Copy, PartialEq)] +enum Mode { + /// Honour `Range` properly: `206` plus the requested tail. + Honest, + /// Ignore `Range` and always send the full body with `200`. Many true servers do this. + IgnoresRange, + /// Serve a body that does not match the advertised md5. + Corrupt, +} + +/// Serve `body` until told to stop. Returns the bound address. +async fn serve(body: Vec, mode: Mode, stop: Arc) -> String { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let addr = listener.local_addr().expect("addr").to_string(); + tokio::spawn(async move { + while !stop.load(Ordering::Relaxed) { + let Ok((mut sock, _)) = listener.accept().await else { + break; + }; + let body = body.clone(); + tokio::spawn(async move { + let mut buf = vec![0u8; 4096]; + let n = sock.read(&mut buf).await.unwrap_or(0); + // Change to lowercase before the match. `reqwest` sends header *names* in + // lowercase. So a server that looks for "Range:" never finds one, and gives no + // message about it. Because of that fault, this stub tested the ignore-range + // path when its purpose was to test a transfer that continues. + let req = String::from_utf8_lossy(&buf[..n]).to_lowercase(); + + // `range: bytes=N-` + let start = req + .lines() + .find_map(|l| l.strip_prefix("range: bytes=")) + .and_then(|r| r.trim().trim_end_matches('-').parse::().ok()) + .unwrap_or(0); + + let send_partial = mode == Mode::Honest && start > 0 && start < body.len(); + let payload = if send_partial { &body[start..] } else { &body[..] }; + let status = if send_partial { + "HTTP/1.1 206 Partial Content" + } else { + "HTTP/1.1 200 OK" + }; + let head = format!( + "{status}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", + payload.len() + ); + let _ = sock.write_all(head.as_bytes()).await; + let _ = sock.write_all(payload).await; + let _ = sock.flush().await; + }); + } + }); + addr +} + +fn md5_hex(bytes: &[u8]) -> String { + use md5::{Digest, Md5}; + let mut h = Md5::new(); + h.update(bytes); + h.finalize().iter().map(|b| format!("{b:02x}")).collect() +} + +fn entry(addr: &str, name: &str, md5: Option, bytes: Option) -> ManifestFile { + ManifestFile { + run_accession: "ERR000001".into(), + url: format!("http://{addr}/{name}"), + index_url: None, + md5, + bytes, + format: "CRAM".into(), + instrument: None, + } +} + +fn body() -> Vec { + // The body is large enough that a transfer continues across some chunks, not inside one. + (0..200_000u32).flat_map(|i| i.to_le_bytes()).collect() +} + +#[tokio::test] +async fn a_whole_file_downloads_and_verifies() { + let data = body(); + let stop = Arc::new(AtomicBool::new(false)); + let addr = serve(data.clone(), Mode::Honest, stop.clone()).await; + let dir = Scratch::new("whole"); + let client = reqwest::Client::new(); + + let e = entry(&addr, "x.cram", Some(md5_hex(&data)), Some(data.len() as i64)); + let mut seen: u64 = 0; + let got = ena::fetch_file(&client, dir.path(), &e, &CancelToken::none(), &mut |r, _| seen = r) + .await + .expect("download"); + + assert_eq!(std::fs::read(&got).unwrap(), data); + assert_eq!(seen, data.len() as u64, "progress ends at the full size"); + assert!( + !got.with_extension("cram.part").exists(), + "the .part is renamed away, never left behind" + ); + stop.store(true, Ordering::Relaxed); +} + +/// This is the purpose of the module. An interrupted transfer continues from the bytes on the +/// disk. The md5 value is still correct, but this process did not get the prefix. +#[tokio::test] +async fn an_interrupted_transfer_resumes_and_still_verifies() { + let data = body(); + let stop = Arc::new(AtomicBool::new(false)); + let addr = serve(data.clone(), Mode::Honest, stop.clone()).await; + let dir = Scratch::new("resume"); + let client = reqwest::Client::new(); + + // Simulate a killed download: the first third is already on disk as a `.part`. + let split = data.len() / 3; + std::fs::write(dir.path().join("x.cram.part"), &data[..split]).unwrap(); + + let e = entry(&addr, "x.cram", Some(md5_hex(&data)), Some(data.len() as i64)); + let mut first_report: Option = None; + let got = ena::fetch_file(&client, dir.path(), &e, &CancelToken::none(), &mut |r, _| { + first_report.get_or_insert(r); + }) + .await + .expect("resumed download"); + + assert_eq!( + std::fs::read(&got).unwrap(), + data, + "the resumed file is byte-identical to the source" + ); + assert!( + first_report.unwrap() > split as u64, + "progress counts the resumed prefix; a bar that restarts at zero tells the user the opposite \ + of what happened" + ); + stop.store(true, Ordering::Relaxed); +} + +/// Many servers ignore `Range` and send the full body with `200`. If the module added those bytes +/// to the prefix, the file would have the correct size only by accident. The checksum would fail. +#[tokio::test] +async fn a_server_that_ignores_range_is_handled_rather_than_trusted() { + let data = body(); + let stop = Arc::new(AtomicBool::new(false)); + let addr = serve(data.clone(), Mode::IgnoresRange, stop.clone()).await; + let dir = Scratch::new("ignores-range"); + let client = reqwest::Client::new(); + + std::fs::write(dir.path().join("x.cram.part"), &data[..data.len() / 3]).unwrap(); + + let e = entry(&addr, "x.cram", Some(md5_hex(&data)), Some(data.len() as i64)); + let got = ena::fetch_file(&client, dir.path(), &e, &CancelToken::none(), &mut |_, _| {}) + .await + .expect("download restarts cleanly"); + assert_eq!( + std::fs::read(&got).unwrap(), + data, + "restarted from zero, not appended to the prefix" + ); + stop.store(true, Ordering::Relaxed); +} + +/// A bad checksum must fail loudly and leave nothing behind that a later run could mistake for a +/// good file. +#[tokio::test] +async fn a_checksum_mismatch_fails_and_leaves_no_part_behind() { + let data = body(); + let stop = Arc::new(AtomicBool::new(false)); + let addr = serve(data.clone(), Mode::Corrupt, stop.clone()).await; + let dir = Scratch::new("corrupt"); + let client = reqwest::Client::new(); + + let e = entry( + &addr, + "x.cram", + Some(md5_hex(b"something else entirely")), + Some(data.len() as i64), + ); + // No delay. This test examines the failure path, and not the true delay schedule. + let policy = RetryPolicy { + attempts: 2, + backoff: false, + }; + let err = ena::fetch_file_with(&client, dir.path(), &e, policy, &CancelToken::none(), &mut |_, _| {}) + .await + .expect_err("must not accept a file that fails its checksum"); + assert!(format!("{err}").contains("checksum mismatch"), "{err}"); + assert!(!dir.path().join("x.cram").exists(), "no finished file"); + assert!( + !dir.path().join("x.cram.part").exists(), + "and no partial one to resume from" + ); + stop.store(true, Ordering::Relaxed); +} + +/// The module never gets a file two times. The rename marks a file as complete. So a second run of +/// a unit after a crash costs nothing for the files that are already complete. +#[tokio::test] +async fn a_completed_file_is_not_downloaded_twice() { + let dir = Scratch::new("existing"); + std::fs::write(dir.path().join("x.cram"), b"already here").unwrap(); + let client = reqwest::Client::new(); + + // The URL points to no server. A request on the network would make this test fail. + let e = entry("127.0.0.1:1", "x.cram", Some("ignored".into()), Some(999)); + let got = ena::fetch_file(&client, dir.path(), &e, &CancelToken::none(), &mut |_, _| {}) + .await + .expect("an existing file short-circuits"); + assert_eq!(std::fs::read(got).unwrap(), b"already here"); +} diff --git a/crates/navigator-sync/src/grid.rs b/crates/navigator-sync/src/grid.rs new file mode 100644 index 00000000..0c12788c --- /dev/null +++ b/crates/navigator-sync/src/grid.rs @@ -0,0 +1,152 @@ +//! Canonical signing strings for the signed Grid Edge API of the AppView (`/api/v1/grid/*`). +//! +//! These mirror `du_db::grid::messages` on the AppView, **exactly**. The server checks the +//! device-key signature against the string that it builds itself. Any difference here gives an +//! immediate 403, and the message tells the user nothing about the cause. +//! +//! Each string starts with its own operation name. So an attacker can not take a signature from one +//! endpoint and use it on a different endpoint. +//! +//! A change to a string here is a change to a published contract. A desktop version that signs the +//! old bytes stops work at the moment the server changes. The tests below hold each string, so an +//! accidental change fails here and not against a released version. +//! +//! Each call that changes data goes through +//! [`DeviceKey::sign_fresh`](crate::device_key::DeviceKey::sign_fresh), which puts the timestamp in +//! front as `{ts}\n{base}`. One signature then holds both the operation and the time. A read poll +//! signs the string here directly and puts its own `ts` in the string. + +pub mod messages { + /// `grid-register\n{did}\n{software_version}\n{caps_sha256_b64}`: announce a node and what it + /// can do. + /// + /// The hash of the capabilities is in the signed string. So a node can not have capabilities + /// that it did not send. That is important, because the claim path filters on them. A false + /// claim of FASTQ ability would give the node work that it can not do. + pub fn register(did: &str, software_version: &str, caps_sha256_b64: &str) -> String { + format!("grid-register\n{did}\n{software_version}\n{caps_sha256_b64}") + } + + /// `grid-poll\n{did}\n{ts}`: a read of what the caller has done, with a replay guard. + pub fn poll(did: &str, ts: i64) -> String { + format!("grid-poll\n{did}\n{ts}") + } + + /// `grid-claim\n{did}\n{kinds}\n{count}\n{lease_secs}`: reserve up to `count` work units. + /// + /// `kinds` is the comma-joined list, in upper case and in alphabetical order. The client makes + /// that form before it signs, and the server makes the same form before it checks. Without one + /// agreed form, `["CRAM","cram"]` and `["cram","CRAM"]` give two different signed strings for + /// one request. + /// + /// The signed values are the values that the node asks for. The server can reduce `count` or + /// `lease_secs` to its own limits after the check. A node does not know those limits, and a + /// signature over a value that the node can not calculate is not possible to make. + pub fn claim(did: &str, kinds: &str, count: i32, lease_secs: i64) -> String { + format!("grid-claim\n{did}\n{kinds}\n{count}\n{lease_secs}") + } + + /// `grid-heartbeat\n{did}\n{lease_id}\n{stage}`: liveness for one lease, with the stage. + pub fn heartbeat(did: &str, lease_id: i64, stage: &str) -> String { + format!("grid-heartbeat\n{did}\n{lease_id}\n{stage}") + } + + /// `grid-release\n{did}\n{lease_id}\n{reason}`: give a lease back with no result. + /// + /// The reason is in the signed string. So the server can not record a release reason that the + /// node did not send. + pub fn release(did: &str, lease_id: i64, reason: &str) -> String { + format!("grid-release\n{did}\n{lease_id}\n{reason}") + } + + /// `grid-submit\n{did}\n{work_unit_id}\n{digest_sha256_b64}`: send a result. + /// + /// The signature covers the **hash of the digest** and not the digest itself. The server + /// calculates that hash again from the body that arrives. A signature over a hash proves only + /// that the signer knew the hash. Without the second calculation, a node could sign the hash of + /// a good result and send a different result. + pub fn submit(did: &str, work_unit_id: i64, digest_sha256_b64: &str) -> String { + format!("grid-submit\n{did}\n{work_unit_id}\n{digest_sha256_b64}") + } + + /// Put the data kinds of a node into the one agreed form: upper case, no repeats, alphabetical + /// order, joined with commas. + /// + /// The client and the server must make the same form. This function is the client half. The + /// AppView handler does the same operation before it checks the signature. + pub fn normalize_kinds(kinds: &[String]) -> String { + let mut k: Vec = kinds.iter().map(|s| s.trim().to_ascii_uppercase()).collect(); + k.sort(); + k.dedup(); + k.join(",") + } +} + +#[cfg(test)] +mod tests { + use super::messages; + + /// The strings match the `du_db::grid::messages` literals of the AppView exactly. A change to + /// one side only gives a 403 with no explanation, so this test is the guard. + #[test] + fn canonical_strings() { + assert_eq!( + messages::poll("did:plc:abc", 1_724_500_000), + "grid-poll\ndid:plc:abc\n1724500000" + ); + assert_eq!( + messages::claim("did:plc:abc", "CRAM,FASTQ", 4, 259_200), + "grid-claim\ndid:plc:abc\nCRAM,FASTQ\n4\n259200" + ); + assert_eq!( + messages::heartbeat("did:plc:abc", 7, "align"), + "grid-heartbeat\ndid:plc:abc\n7\nalign" + ); + assert_eq!( + messages::release("did:plc:abc", 7, "cancelled"), + "grid-release\ndid:plc:abc\n7\ncancelled" + ); + assert_eq!( + messages::submit("did:plc:abc", 12, "3q2+7w=="), + "grid-submit\ndid:plc:abc\n12\n3q2+7w==" + ); + assert_eq!( + messages::register("did:plc:abc", "0.1.0-alpha.18", "3q2+7w=="), + "grid-register\ndid:plc:abc\n0.1.0-alpha.18\n3q2+7w==" + ); + } + + /// Each string starts with a different operation name. A signature from one endpoint is then of + /// no use on a different endpoint. + #[test] + fn each_message_has_its_own_operation_name() { + let all = [ + messages::poll("d", 1), + messages::claim("d", "CRAM", 1, 1), + messages::heartbeat("d", 1, "s"), + messages::release("d", 1, "r"), + messages::submit("d", 1, "h"), + messages::register("d", "v", "h"), + ]; + let names: Vec<&str> = all.iter().map(|m| m.split('\n').next().unwrap()).collect(); + let mut sorted = names.clone(); + sorted.sort_unstable(); + sorted.dedup(); + assert_eq!( + sorted.len(), + names.len(), + "two grid messages share an operation name: {names:?}" + ); + } + + /// The one agreed form of the data kinds. The AppView handler makes the same form, so these + /// results must not change without a change there. + #[test] + fn data_kinds_have_one_agreed_form() { + let k = |v: &[&str]| messages::normalize_kinds(&v.iter().map(|s| s.to_string()).collect::>()); + assert_eq!(k(&["FASTQ", "CRAM"]), "CRAM,FASTQ"); + assert_eq!(k(&["cram", " CRAM ", "FASTQ"]), "CRAM,FASTQ"); + assert_eq!(k(&["CRAM"]), "CRAM"); + assert_eq!(k(&[]), ""); + } +} diff --git a/crates/navigator-sync/src/lib.rs b/crates/navigator-sync/src/lib.rs index d7378190..63c77fe8 100644 --- a/crates/navigator-sync/src/lib.rs +++ b/crates/navigator-sync/src/lib.rs @@ -10,6 +10,7 @@ pub mod device_key; pub mod error; pub mod exchange; +pub mod grid; pub mod oauth; pub mod publish; pub mod records; diff --git a/crates/navigator-ui/Cargo.toml b/crates/navigator-ui/Cargo.toml index 3fb58283..7bc802fb 100644 --- a/crates/navigator-ui/Cargo.toml +++ b/crates/navigator-ui/Cargo.toml @@ -63,7 +63,9 @@ navigator-app = { workspace = true } navigator-domain = { workspace = true } rfd = "0.17.2" serde_json = "1" -tokio = { version = "1.52.3", features = ["rt-multi-thread", "sync", "macros"] } +# `signal`, so `navigator contribute` can catch Ctrl-C. A node that only exited would keep each +# lease that it holds until the lease time ended, and no other node could take that work. +tokio = { version = "1.52.3", features = ["rt-multi-thread", "sync", "macros", "signal"] } [dev-dependencies] navigator-store = { workspace = true } diff --git a/crates/navigator-ui/src/cli.rs b/crates/navigator-ui/src/cli.rs index 1e777ad9..c3c06f8a 100644 --- a/crates/navigator-ui/src/cli.rs +++ b/crates/navigator-ui/src/cli.rs @@ -54,6 +54,19 @@ macro_rules! cli_try { }; } +/// How many units to claim in one call. It is small for two reasons. A node that stops then has +/// few leases to give back. And a node that is new does not take a large part of the catalogue +/// before it proves itself. +const CLAIM_BATCH: i32 = 4; + +/// How long a unit directory that a stop left may stay. A user who stops a run and continues it the +/// same day keeps the transfer. A directory older than this holds files that no run will continue. +const SCRATCH_MAX_AGE: std::time::Duration = std::time::Duration::from_secs(7 * 24 * 3600); + +/// How often the node announces itself, so that the fleet view of the AppView shows it as alive. +/// The value is far below any period that such a view would call dead. +const NODE_LIVENESS_EVERY: std::time::Duration = std::time::Duration::from_secs(5 * 60); + #[derive(Parser)] #[command( name = "navigator", @@ -112,6 +125,14 @@ pub enum Command { /// Ancestry is then ready with no later lazy build. It is heavy: one whole-genome decode for /// each alignment. GenotypePanel(ShowArgs), + /// Give computer time to the DecodingUs Grid. The node takes public ENA samples from the + /// AppView, analyzes them, and sends back a signed result. Agreed results earn compute credit + /// on a public board. + /// + /// The node only takes work that it can do. It advertises its data kinds, and the AppView + /// gives it nothing else. Press Ctrl-C to stop: the node finishes no more units, gives back + /// each lease that it holds, and removes its temporary files. + Contribute(ContributeArgs), /// A branch report for each marker. It gives the genotype of the sample at every marker that /// defines a node in the descendant subtree of a Y or mtDNA tree node. Each row has the /// observed base, the derived or ancestral status, and the evidence. Use it to spot-check a @@ -231,6 +252,35 @@ pub struct ProbeArgs { json: bool, } +#[derive(Args)] +pub struct ContributeArgs { + /// Workspace database path. + #[arg(long)] + db: Option, + /// How many units to do before the node stops. Without this, the node continues until the + /// catalogue has no more work, or until Ctrl-C. + #[arg(long)] + max_units: Option, + /// How long to hold each lease, in days. The AppView reduces a value outside its own limits. + #[arg(long, default_value_t = 3)] + lease_days: i64, + /// Where to put the files of a unit. Each unit gets its own directory below this one, and the + /// node removes that directory when the unit ends. + #[arg(long)] + scratch: Option, + /// The build to report in the result. It must match the build of the analysis. + #[arg(long, default_value = "chm13v2.0")] + reference_build: String, + /// How much disk space, in GB, this node gives to the work. The AppView uses this value to + /// select work that fits. Without it, the node reports the free space of the scratch volume. + #[arg(long)] + disk_gb: Option, + /// Show what the node would take, and then stop. The node claims nothing, gets no file, + /// and analyzes nothing. + #[arg(long)] + dry_run: bool, +} + /// `archaic` takes an optional alignment override, so that a caller can genotype one specific build /// directly. Without it the app picks the best-callable alignment of the subject. The GRCh37 and /// GRCh38 code path is then out of reach on a subject that also has CHM13 data. @@ -570,6 +620,7 @@ pub fn run(command: Command) -> i32 { Command::Archaic(a) => archaic(a).await, Command::ArchaicSegments(a) => archaic_segments(a).await, Command::GenotypePanel(a) => genotype_panel(a).await, + Command::Contribute(a) => contribute(a).await, Command::BranchReport(a) => branch_report(a).await, Command::Doctor(a) => doctor(a).await, Command::Projects(a) => projects(a).await, @@ -2268,3 +2319,213 @@ fn truncate(s: &str, max: usize) -> String { format!("{}…", s.chars().take(max - 1).collect::()) } } + +/// Give computer time to the DecodingUs Grid. +/// +/// The loop is: announce this node, claim a small group of units, do each one, and claim again. It +/// ends when the user presses Ctrl-C, when the catalogue has no more work that this node can do, or +/// when the node reaches `--max-units`. +/// +/// **Ctrl-C must not lose a lease.** The signal sets the cancel token. The unit that is in progress +/// stops at its next step, gives its lease back, and removes its files. A node that only exited +/// would hold each of its units until the lease time ended, and no other node could take them. +async fn contribute(args: ContributeArgs) -> i32 { + use std::time::Instant; + let app = cli_try!(open(args.db).await); + let kinds = navigator_app::grid_job::supported_data_kinds(); + + let scratch = args + .scratch + .unwrap_or_else(|| std::env::temp_dir().join("navigator-grid")); + let params = navigator_app::grid_job::GridJobParams { + max_units: CLAIM_BATCH, + lease_secs: args.lease_days.max(1) * 24 * 3600, + scratch_root: scratch.clone(), + reference_build: args.reference_build.clone(), + }; + + // Real values, and not zeros. + // + // `du_db::grid::claim` filters on the data kind only today. §11 of the design says that it must + // also filter on memory, free disk and thread count. On the day that filter arrives, a node + // that reports zero receives no work. It then prints the same "no more work for this node right + // now" that an empty catalogue gives. Nobody would find the true cause quickly. + let caps = navigator_app::grid::NodeCapabilities { + data_kinds: kinds.clone(), + threads: std::thread::available_parallelism() + .map(|n| n.get() as u32) + .unwrap_or(1), + disk_budget: args + .disk_gb + .map(|gb| gb.saturating_mul(1_000_000_000)) + .unwrap_or_else(|| navigator_app::grid_job::free_space_for(&scratch)), + memory_bytes: navigator_app::grid_job::machine_memory_bytes(), + }; + + println!("DecodingUs Grid — this node offers: {}", kinds.join(", ")); + println!(" scratch: {}", scratch.display()); + println!(" reference: {}", params.reference_build); + println!(" lease: {} day(s)", args.lease_days.max(1)); + + if args.dry_run { + println!("\ndry run: nothing claimed. Remove --dry-run to contribute."); + return 0; + } + + match app.grid_register(&caps).await { + Ok(id) => println!(" node id: {id}"), + Err(e) => { + eprintln!("error: could not announce this node: {e}"); + return ExitCode::exit_code(e); + } + } + + // A directory that an earlier run stopped keeps its files, so that a transfer can continue. + // This removes the ones that no run continued. + let swept = navigator_app::grid_job::sweep_old_scratch(&scratch, SCRATCH_MAX_AGE).await; + if swept > 0 { + println!(" removed {swept} old unit directory(s)"); + } + + // Ctrl-C sets the **session** token. Each unit then gets its own token, because a + // `CancelToken` has no way back: `cancel.rs` states that a token covers exactly one run. + // + // One token for the whole session gave a fault. The beat of a unit cancels its token when the + // AppView reports that another node holds the lease. One token made that one lost lease stop + // the node for the rest of the run. The node then gave back each unit that it still held. A + // lost lease is a normal event, and it must cost one unit and no more. + let session = navigator_app::CancelToken::new(); + let signal_token = session.clone(); + tokio::spawn(async move { + if tokio::signal::ctrl_c().await.is_ok() { + eprintln!("\nstopping: the node gives back each lease that it holds…"); + signal_token.cancel(); + } + }); + + // Tell the fleet view that this node is alive, on a timer of its own. + // + // Only the register call writes `fed.pds_node.last_heartbeat`. The beat of a unit writes the + // row of the lease, which is a different row. + // + // A unit of a whole genome takes hours. So a call between two units, or between two batches, + // leaves the node dead in that view for most of the time that it works. + { + let (app2, caps2, stop) = (app.clone(), caps.clone(), session.clone()); + tokio::spawn(async move { + while !stop.is_cancelled() { + tokio::time::sleep(NODE_LIVENESS_EVERY).await; + if stop.is_cancelled() { + return; + } + let _ = app2.grid_register(&caps2).await; + } + }); + } + + let mut done = 0u32; + let mut failed = 0u32; + loop { + if session.is_cancelled() { + break; + } + let want = match args.max_units { + Some(max) if done + failed >= max => break, + Some(max) => (max - done - failed).min(CLAIM_BATCH as u32) as i32, + None => CLAIM_BATCH, + }; + + let units = match app.grid_claim(&kinds, want, params.lease_secs).await { + Ok(u) => u, + Err(e) => { + eprintln!("error: could not claim work: {e}"); + return ExitCode::exit_code(e); + } + }; + if units.is_empty() { + println!("\nno more work for this node right now."); + break; + } + + for unit in &units { + if session.is_cancelled() { + // Units that this node claimed but did not start still hold a lease. Give each one + // back, so the catalogue does not wait out the lease time for work never begun. + let _ = app.grid_release(unit.lease_id, "stopped by the user").await; + continue; + } + let started = Instant::now(); + println!("\n{} ({})", unit.sample_accession, unit.data_kind); + + // The driver sends the heartbeat itself, next to the work. This callback only draws + // the stage for the user. + let mut report = |stage: navigator_app::grid_job::GridStage, detail: &str| { + println!(" {:<9} {detail}", stage.as_str()); + }; + + // A token for this unit only. A task copies the state of the session token into it, + // so Ctrl-C still stops the work inside a few moments. The unit token can also stop by + // itself, when this node loses the lease, and the session then continues. + let unit_cancel = navigator_app::CancelToken::new(); + // The bridge ends when this value goes out of scope, at the end of the unit. Without + // that signal, a unit that finished with no cancel would leave the task in its loop + // for the life of the process. A node that runs for days would then hold hundreds of + // tasks, and each one wakes four times each second. + let (stop_bridge, mut bridge_ended) = tokio::sync::oneshot::channel::<()>(); + { + let (s, u) = (session.clone(), unit_cancel.clone()); + tokio::spawn(async move { + loop { + if s.is_cancelled() { + u.cancel(); + return; + } + tokio::select! { + _ = &mut bridge_ended => return, + _ = tokio::time::sleep(std::time::Duration::from_millis(250)) => {} + } + } + }); + } + let outcome = app.run_grid_unit(unit, ¶ms, &unit_cancel, &mut report).await; + drop(stop_bridge); + + // Send the records that the unit put in the queue. + // + // The unit gives the AppView the address of each record with its result. Only the + // graphical application empties that queue, on a timer. + // + // So a node that runs with no window named records that stayed in its own database for + // ever. That queue also grew by two rows for each unit. + if let Err(e) = app.drain_outbox().await { + eprintln!(" note the records of this unit are still in the queue: {e}"); + } + match (&outcome.submission_id, &outcome.error) { + (Some(id), _) => { + done += 1; + println!(" done submission #{id} in {:.1?}", started.elapsed()); + } + (None, Some(e)) => { + failed += 1; + eprintln!(" failed {e}"); + } + (None, None) => failed += 1, + } + } + } + + println!("\n{done} unit(s) sent, {failed} failed."); + match app.grid_standing().await { + Ok(s) => { + let rank = s.rank.map(|r| format!("#{r}")).unwrap_or_else(|| "unranked".into()); + println!( + "total: {:.2} cobblestones over {} unit(s), {rank}", + s.cobblestones, s.units_credited + ); + } + // This total is only a courtesy at the end of a run. A failure to read it must not change + // the exit code of work that succeeded. + Err(e) => eprintln!("(could not read your standing: {e})"), + } + i32::from(failed > 0) +}