diff --git a/rust/Cargo.lock b/rust/Cargo.lock index 37f3bb3..90fa2e9 100644 --- a/rust/Cargo.lock +++ b/rust/Cargo.lock @@ -1360,7 +1360,7 @@ dependencies = [ [[package]] name = "du-atproto" version = "0.1.0" -source = "git+https://github.com/JamesKane/decodingus-shared.git?rev=b02884c95e98c9dcb585a231d140c9841c3d37a6#b02884c95e98c9dcb585a231d140c9841c3d37a6" +source = "git+https://github.com/JamesKane/decodingus-shared.git?rev=b11ba77956275c48ad4fe32e3a58c84961565c85#b11ba77956275c48ad4fe32e3a58c84961565c85" dependencies = [ "base64", "du-domain", @@ -1378,7 +1378,7 @@ dependencies = [ [[package]] name = "du-bio" version = "0.1.0" -source = "git+https://github.com/JamesKane/decodingus-shared.git?rev=b02884c95e98c9dcb585a231d140c9841c3d37a6#b02884c95e98c9dcb585a231d140c9841c3d37a6" +source = "git+https://github.com/JamesKane/decodingus-shared.git?rev=b11ba77956275c48ad4fe32e3a58c84961565c85#b11ba77956275c48ad4fe32e3a58c84961565c85" dependencies = [ "du-domain", "serde", @@ -1390,11 +1390,13 @@ dependencies = [ name = "du-db" version = "0.1.0" dependencies = [ + "base64", "chrono", "du-bio", "du-domain", "serde", "serde_json", + "sha2 0.10.9", "sqlx", "thiserror 2.0.18", "tokio", @@ -1405,7 +1407,7 @@ dependencies = [ [[package]] name = "du-domain" version = "0.1.0" -source = "git+https://github.com/JamesKane/decodingus-shared.git?rev=b02884c95e98c9dcb585a231d140c9841c3d37a6#b02884c95e98c9dcb585a231d140c9841c3d37a6" +source = "git+https://github.com/JamesKane/decodingus-shared.git?rev=b11ba77956275c48ad4fe32e3a58c84961565c85#b11ba77956275c48ad4fe32e3a58c84961565c85" dependencies = [ "chrono", "rust_decimal", diff --git a/rust/Cargo.toml b/rust/Cargo.toml index 8d6b6cc..bef766d 100644 --- a/rust/Cargo.toml +++ b/rust/Cargo.toml @@ -26,9 +26,9 @@ repository = "https://github.com/decodingus/decodingus" # To update: push decodingus-shared, then bump `rev` (or switch to a pushed tag, # e.g. `tag = "v0.1.0"`). For local co-dev against working-tree changes, add a # [patch] section pointing these back at ../../decodingus-shared/crates/*. -du-domain = { git = "https://github.com/JamesKane/decodingus-shared.git", rev = "b02884c95e98c9dcb585a231d140c9841c3d37a6" } -du-atproto = { git = "https://github.com/JamesKane/decodingus-shared.git", rev = "b02884c95e98c9dcb585a231d140c9841c3d37a6" } -du-bio = { git = "https://github.com/JamesKane/decodingus-shared.git", rev = "b02884c95e98c9dcb585a231d140c9841c3d37a6" } +du-domain = { git = "https://github.com/JamesKane/decodingus-shared.git", rev = "b11ba77956275c48ad4fe32e3a58c84961565c85" } +du-atproto = { git = "https://github.com/JamesKane/decodingus-shared.git", rev = "b11ba77956275c48ad4fe32e3a58c84961565c85" } +du-bio = { git = "https://github.com/JamesKane/decodingus-shared.git", rev = "b11ba77956275c48ad4fe32e3a58c84961565c85" } # Internal (decodingus-only) crates du-db = { path = "crates/du-db" } du-external = { path = "crates/du-external" } diff --git a/rust/crates/du-db/Cargo.toml b/rust/crates/du-db/Cargo.toml index b437640..6df94fc 100644 --- a/rust/crates/du-db/Cargo.toml +++ b/rust/crates/du-db/Cargo.toml @@ -15,6 +15,10 @@ sqlx = { workspace = true } tokio = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } +# `grid::digest::canonical_sha256_b64` — the submit handler recomputes the digest hash a node +# signed, so a node cannot sign one result and send another. +sha2 = { workspace = true } +base64 = { workspace = true } uuid = { workspace = true } chrono = { workspace = true } thiserror = { workspace = true } diff --git a/rust/crates/du-db/src/grid/digest.rs b/rust/crates/du-db/src/grid/digest.rs new file mode 100644 index 0000000..3dbcdbb --- /dev/null +++ b/rust/crates/du-db/src/grid/digest.rs @@ -0,0 +1,226 @@ +//! Result-digest comparison — the agreement test that gates canonicalization. +//! +//! Realignment is not byte-deterministic across thread counts and builds, so validation never +//! hashes the BAM. It compares a small canonical digest of **discrete calls** plus **bucketed** +//! continuous metrics. Design §5.2. +//! +//! # The node signs raw values; the AppView buckets +//! +//! This is an amendment to §5.2, which had the node submit values already bucketed. Bucketing on +//! the client would make the bucket function a **cross-repo contract**: every Navigator release +//! would have to round coverage exactly as the AppView expects, and a divergence between the two +//! would surface as unexplained `DIVERGENT` verdicts against honest nodes. Bucketing here instead +//! means the rule exists once, and it can be re-tuned without redeploying a single desktop client. +//! +//! The signature still covers what the node actually computed, because the node signs the raw +//! digest it sends. Boundary sensitivity — two honest nodes landing either side of a bucket edge — +//! is inherent to bucketing and is unchanged by where it happens. +//! +//! # What is compared +//! +//! Exact match, after trimming: `sex`, `y_terminal`, `ancestry_superpop_argmax`. Two digests that +//! both omit a field agree on it; one that has it and one that does not, disagree. That is the +//! wanted behaviour — a sample with no Y call and a sample called `R-FGC29071` are not the same +//! result. +//! +//! Bucketed: `coverage_mean` to the nearest 2×, `callable_fraction` to two decimals. +//! +//! **`mt_terminal` is deliberately absent.** `App::analyze_biosample` declines to assign mtDNA +//! because that value "is not final on CHM13", and the Grid realigns to CHM13 — so the digest +//! cannot require a value the analysis path does not produce. mt is still published in the full +//! records; it simply does not gate canonicalization. Design §12.3. + +use serde_json::Value; + +/// Coverage agrees within this bucket width, in fold-coverage. +const COVERAGE_BUCKET: f64 = 2.0; +/// Callable fraction agrees to this many decimal places. +const CALLABLE_DECIMALS: f64 = 100.0; + +/// The comparable projection of one submitted digest. +/// +/// Anything not listed here is carried in the full records but does not gate canonicalization. +#[derive(Debug, Clone, PartialEq)] +pub struct Comparable { + pub sex: Option, + pub y_terminal: Option, + pub ancestry_superpop_argmax: Option, + /// `coverage_mean` snapped to [`COVERAGE_BUCKET`]; `None` when the digest omitted it. + pub coverage_bucket: Option, + /// `callable_fraction` snapped to two decimals. + pub callable_bucket: Option, +} + +fn text(calls: &Value, key: &str) -> Option { + calls + .get(key) + .and_then(Value::as_str) + .map(str::trim) + .filter(|s| !s.is_empty()) + .map(str::to_string) +} + +fn bucketed(calls: &Value, key: &str, scale: f64) -> Option { + let v = calls.get(key).and_then(Value::as_f64)?; + if !v.is_finite() { + return None; + } + Some((v * scale).round() as i64) +} + +impl Comparable { + /// Project a submitted digest. A digest whose shape is wrong yields all-`None`, which agrees + /// only with another equally empty digest — so a malformed submission can never canonicalize a + /// unit on its own. + pub fn from_digest(digest: &Value) -> Self { + let calls = digest.get("calls").cloned().unwrap_or(Value::Null); + Comparable { + sex: text(&calls, "sex"), + y_terminal: text(&calls, "y_terminal"), + ancestry_superpop_argmax: text(&calls, "ancestry_superpop_argmax"), + coverage_bucket: bucketed(&calls, "coverage_mean", 1.0 / COVERAGE_BUCKET), + callable_bucket: bucketed(&calls, "callable_fraction", CALLABLE_DECIMALS), + } + } +} + +/// The major component of a semver-ish stack version: `"1.7.0"` → `"1"`. +/// +/// Only submissions from a compatible major are compared. A minor release that refactors a walker +/// should not invalidate a canonical result; a major one may legitimately change what a call means. +pub fn stack_major(version: &str) -> &str { + version.split('.').next().unwrap_or(version).trim() +} + +/// Whether two submissions are even *comparable* — same reference build and same stack major. +/// +/// Results on different references are not in disagreement; they are answers to different +/// questions, and clustering them together would manufacture divergence out of nothing. +pub fn comparable(a_build: &str, a_stack: &str, b_build: &str, b_stack: &str) -> bool { + a_build.trim().eq_ignore_ascii_case(b_build.trim()) + && stack_major(a_stack) == stack_major(b_stack) +} + +/// The SHA-256 of a digest's canonical bytes, base64 (standard alphabet). +/// +/// "Canonical" is `serde_json` with **sorted keys and no whitespace**, which is simply what +/// `serde_json::to_vec` produces: this workspace does not enable the `preserve_order` feature, so +/// `serde_json::Map` is a `BTreeMap` and serialization is key-sorted and deterministic. Navigator +/// uses the same crate under the same default, so both sides hash identical bytes without either +/// having to implement a canonicalization scheme. +/// +/// **This is the one remaining byte-level contract between the repos**, and it is deliberately the +/// smallest one available: no field order to agree on, no float formatting rules, no separator +/// choices. The test below pins it. If `preserve_order` is ever switched on anywhere, that test +/// fails here rather than as unexplained 400s against desktop clients. +/// +/// The submit handler recomputes this from the digest it received and rejects a mismatch, so a node +/// cannot sign the hash of one result and send another. +pub fn canonical_sha256_b64(digest: &Value) -> String { + use base64::Engine as _; + use sha2::{Digest as _, Sha256}; + let bytes = serde_json::to_vec(digest).unwrap_or_default(); + base64::engine::general_purpose::STANDARD.encode(Sha256::digest(bytes)) +} + +/// Whether two digests **agree**, per §5.2. +pub fn agree(a: &Value, b: &Value) -> bool { + Comparable::from_digest(a) == Comparable::from_digest(b) +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + fn d(sex: &str, y: &str, pop: &str, cov: f64, callable: f64) -> Value { + json!({"calls": { + "sex": sex, "y_terminal": y, "ancestry_superpop_argmax": pop, + "coverage_mean": cov, "callable_fraction": callable + }}) + } + + #[test] + fn identical_digests_agree() { + assert!(agree( + &d("XY", "R-FGC29071", "EUR", 30.4, 0.9412), + &d("XY", "R-FGC29071", "EUR", 30.4, 0.9412) + )); + } + + /// The point of bucketing: two honest nodes never produce the same float, and must still agree. + #[test] + fn continuous_metrics_agree_within_their_bucket() { + assert!(agree( + &d("XY", "R-A", "EUR", 30.4, 0.9412), + &d("XY", "R-A", "EUR", 30.9, 0.9448) + )); + } + + #[test] + fn a_different_discrete_call_is_a_disagreement() { + assert!(!agree( + &d("XY", "R-A", "EUR", 30.0, 0.94), + &d("XY", "R-B", "EUR", 30.0, 0.94) + )); + assert!(!agree( + &d("XY", "R-A", "EUR", 30.0, 0.94), + &d("XX", "R-A", "EUR", 30.0, 0.94) + )); + assert!(!agree( + &d("XY", "R-A", "EUR", 30.0, 0.94), + &d("XY", "R-A", "AFR", 30.0, 0.94) + )); + } + + /// Far apart in coverage is a real disagreement — one of the two analysed different data. + #[test] + fn coverage_far_apart_is_a_disagreement() { + assert!(!agree( + &d("XY", "R-A", "EUR", 30.0, 0.94), + &d("XY", "R-A", "EUR", 12.0, 0.94) + )); + } + + /// A sample with no Y call agrees with another that also has none, and never with one that does. + #[test] + fn a_missing_field_agrees_only_with_a_missing_field() { + let absent = json!({"calls": {"sex": "XX", "ancestry_superpop_argmax": "EUR"}}); + let present = + json!({"calls": {"sex": "XX", "ancestry_superpop_argmax": "EUR", "y_terminal": "R-A"}}); + assert!(agree(&absent, &absent.clone())); + assert!(!agree(&absent, &present)); + } + + /// A malformed digest must never canonicalize anything by matching a well-formed one. + #[test] + fn a_shapeless_digest_agrees_with_nothing_real() { + let junk = json!({"unexpected": true}); + assert!(!agree(&junk, &d("XY", "R-A", "EUR", 30.0, 0.94))); + assert!( + agree(&junk, &json!(null)), + "two empties are consistent, and equally uninformative" + ); + } + + /// The cross-repo byte contract: key order in the source JSON must not change the hash, or a + /// node and the AppView would disagree about what was signed. + #[test] + fn the_canonical_hash_ignores_key_order() { + let a = json!({"calls": {"sex": "XY", "y_terminal": "R-A"}, "unit": "SAMEA1"}); + let b = json!({"unit": "SAMEA1", "calls": {"y_terminal": "R-A", "sex": "XY"}}); + assert_eq!(canonical_sha256_b64(&a), canonical_sha256_b64(&b)); + assert_ne!( + canonical_sha256_b64(&a), + canonical_sha256_b64(&json!({"calls": {"sex": "XX"}, "unit": "SAMEA1"})), + "a different result must hash differently, or the check is worthless" + ); + } + + #[test] + fn only_a_matching_build_and_stack_major_are_compared() { + assert!(comparable("chm13v2.0", "1.7.0", "CHM13v2.0", "1.9.3")); + assert!(!comparable("chm13v2.0", "1.7.0", "GRCh38", "1.7.0")); + assert!(!comparable("chm13v2.0", "1.7.0", "chm13v2.0", "2.0.0")); + } +} diff --git a/rust/crates/du-db/src/grid/mod.rs b/rust/crates/du-db/src/grid/mod.rs new file mode 100644 index 0000000..5860bf1 --- /dev/null +++ b/rust/crates/du-db/src/grid/mod.rs @@ -0,0 +1,874 @@ +//! DecodingUs Grid — work-unit coordination (AppView side). +//! +//! The AppView publishes public-ENA work units; volunteer Navigator instances lease one, fetch +//! it, (re)align to CHM13, run the analysis stack, and submit a signed digest. This module is the +//! storage and the state transitions. Signature verification lives in the `du-web` handler, which +//! is where async DID resolution belongs — the same split as [`crate::exchange`]. +//! +//! Design: `documents/design/distributed-compute-grid.md` in the DUNavigator repo. +//! +//! **Claimability is derived, not stored.** `work_unit.state` holds only the exclusive lifecycle +//! milestones (AVAILABLE / CANONICAL / CONTESTED / RETIRED). Whether a unit can be claimed *now* +//! is a function of how many replicas are in flight against `required_replicas`, which the lease +//! and submission tables already answer. Nothing counts replicas into a column, so no count can +//! drift out of step with the rows it summarises. See `0075_grid.sql` for the full reasoning. +//! +//! **Canonical signed messages** ([`messages`]) are a cross-repo contract: the Navigator edge +//! signs byte-identical strings. Keep them stable. + +pub mod digest; + +use crate::DbError; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use sqlx::PgPool; +use uuid::Uuid; + +/// The exact bytes each grid request's Ed25519 signature covers. A cross-repo contract with the +/// Navigator edge — do not reorder or reformat. +/// +/// Mutating calls go through `du_web::sig::verify_signed_fresh`, which frames these as +/// `{ts}\n{base}` before verifying, so the timestamp binds to the operation in one signature. +/// Read polls carry their own `ts` and go through `verify_signed`. +pub mod messages { + /// Announce a node and its capabilities. `caps_sha256_b64` is over the canonical JSON of the + /// capabilities object, so a node cannot claim RAM it did not advertise at registration. + 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}") + } + /// Replay-guarded poll for the catalogue view (what is claimable, and the caller's standing). + pub fn poll(did: &str, ts: i64) -> String { + format!("grid-poll\n{did}\n{ts}") + } + /// Reserve up to `count` units of the given kinds for `lease_secs`. `kinds` is the + /// comma-joined, ascending-sorted list the caller sent, so the server cannot widen it. + 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}") + } + /// Liveness for one held lease, with the stage the node is on. + pub fn heartbeat(did: &str, lease_id: i64, stage: &str) -> String { + format!("grid-heartbeat\n{did}\n{lease_id}\n{stage}") + } + /// Give a lease back without a result. `reason` is free text and is signed, so a node cannot + /// have a release attributed to it that it did not send. + pub fn release(did: &str, lease_id: i64, reason: &str) -> String { + format!("grid-release\n{did}\n{lease_id}\n{reason}") + } + /// Submit a result. The signature covers the digest **bytes** the node computed, not a + /// re-serialisation of them: `digest_sha256_b64` is the SHA-256 of the exact canonical JSON + /// the node signed and sent, so no JSON round-trip on either side can change what was signed. + 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}") + } +} + +/// A unit as a node receives it on claim — everything needed to do the work without asking the +/// AppView (or ENA) anything else. +#[derive(Debug, Clone, Serialize, Deserialize, sqlx::FromRow)] +pub struct ClaimedUnit { + pub lease_id: i64, + pub work_unit_id: i64, + pub sample_accession: String, + pub study_accession: Option, + pub data_kind: String, + /// `[{run_accession, url, md5, bytes, format}]`. + pub manifest: Value, + pub est_bases: Option, + pub total_bytes: Option, + pub expires_at: chrono::DateTime, +} + +/// One row of the public leaderboard. +/// +/// `cobblestones_milli` is thousandths, matching the ledger column — see `0075_grid.sql`. Divide +/// by 1000 at the point of rendering, never before, so the sum stays exact. +#[derive(Debug, Clone, Serialize, sqlx::FromRow)] +pub struct LeaderboardRow { + pub handle: Option, + pub did: String, + pub cobblestones_milli: i64, + pub units: i64, +} + +/// One cobblestone, in the ledger's units. +pub const COBBLESTONE: i64 = 1_000; + +/// A work unit as the curation job upserts it. +#[derive(Debug, Clone)] +pub struct NewWorkUnit { + pub sample_accession: String, + pub study_accession: Option, + pub data_kind: String, + pub manifest: Value, + pub est_bases: Option, + pub total_bytes: Option, +} + +/// Publish (or refresh) a work unit. Idempotent at `sample_accession`, so a re-crawl of the same +/// study is safe and cheap. +/// +/// A refresh updates the manifest and the size estimates — ENA does re-publish files — but never +/// touches `state`, `required_replicas` or the canonical digest. Curation describes the *input*; +/// validation owns the *lifecycle*, and a re-crawl must not silently un-canonicalise a unit or +/// reset a contested one back to claimable. +pub async fn upsert_work_unit(pool: &PgPool, u: &NewWorkUnit) -> Result { + let id: (i64,) = sqlx::query_as( + "INSERT INTO grid.work_unit (sample_accession, study_accession, data_kind, manifest, est_bases, total_bytes) \ + VALUES ($1, $2, $3, $4, $5, $6) \ + ON CONFLICT (sample_accession) DO UPDATE SET \ + study_accession = EXCLUDED.study_accession, \ + data_kind = EXCLUDED.data_kind, \ + manifest = EXCLUDED.manifest, \ + est_bases = EXCLUDED.est_bases, \ + total_bytes = EXCLUDED.total_bytes, \ + updated_at = now() \ + RETURNING id", + ) + .bind(&u.sample_accession) + .bind(&u.study_accession) + .bind(&u.data_kind) + .bind(&u.manifest) + .bind(u.est_bases) + .bind(u.total_bytes) + .fetch_one(pool) + .await?; + Ok(id.0) +} + +/// Reserve up to `count` units for `did`, for `lease_secs`. +/// +/// This is the one piece of genuinely new concurrency in the Grid, and it is one statement: +/// +/// * `FOR UPDATE SKIP LOCKED` — two nodes claiming at the same instant take *different* units +/// instead of one of them blocking or both taking the same one. +/// * The replica arithmetic — active leases plus submissions that have not been ruled divergent — +/// is what stops a unit being handed out beyond `required_replicas`. +/// * The two `NOT EXISTS` clauses stop a node replicating *itself*: a contributor that already +/// holds a lease on a unit, or already submitted for it, is not offered it again. Quorum means +/// independent results, so self-replication is not a rate limit but a correctness rule. +/// * `ON CONFLICT … DO NOTHING` on the partial unique index makes a retried claim idempotent +/// rather than an error, which matters because the node retries on any transport failure. +/// +/// Returns the units actually reserved, which may be fewer than `count` (or none) when the +/// catalogue is exhausted for those kinds. `count` is an `i64` because Postgres `LIMIT` takes a +/// bigint; binding a narrower integer there is a type error, not a widening. +pub async fn claim( + pool: &PgPool, + did: &str, + node_id: Option, + data_kinds: &[String], + count: i64, + lease_secs: i64, +) -> Result, DbError> { + let rows = sqlx::query_as::<_, ClaimedUnit>( + "WITH candidate AS ( \ + SELECT w.id FROM grid.work_unit w \ + WHERE w.state IN ('AVAILABLE', 'CONTESTED') \ + AND w.data_kind = ANY($3) \ + AND ( (SELECT count(*) FROM grid.lease l \ + WHERE l.work_unit_id = w.id AND l.released_at IS NULL AND l.expires_at > now()) \ + + (SELECT count(*) FROM grid.submission s \ + WHERE s.work_unit_id = w.id AND s.status <> 'DIVERGENT') \ + ) < w.required_replicas \ + AND NOT EXISTS (SELECT 1 FROM grid.lease l2 \ + WHERE l2.work_unit_id = w.id AND l2.did = $1 AND l2.released_at IS NULL) \ + AND NOT EXISTS (SELECT 1 FROM grid.submission s2 \ + WHERE s2.work_unit_id = w.id AND s2.did = $1) \ + ORDER BY w.id \ + LIMIT $4 \ + FOR UPDATE SKIP LOCKED \ + ), ins AS ( \ + INSERT INTO grid.lease (work_unit_id, did, node_id, expires_at) \ + SELECT c.id, $1, $2, now() + make_interval(secs => $5) FROM candidate c \ + ON CONFLICT (work_unit_id, did) WHERE released_at IS NULL DO NOTHING \ + RETURNING id, work_unit_id, expires_at \ + ) \ + SELECT i.id AS lease_id, i.work_unit_id, w.sample_accession, w.study_accession, \ + w.data_kind, w.manifest, w.est_bases, w.total_bytes, i.expires_at \ + FROM ins i JOIN grid.work_unit w ON w.id = i.work_unit_id \ + ORDER BY i.work_unit_id", + ) + .bind(did) + .bind(node_id) + .bind(data_kinds) + .bind(count) + // `make_interval(secs => …)` takes an int4. Clamping rather than erroring is right: a lease + // longer than 68 years is a caller bug, and the server's own bound is what matters anyway. + .bind(i32::try_from(lease_secs).unwrap_or(i32::MAX)) + .fetch_all(pool) + .await?; + Ok(rows) +} + +/// Record liveness on a held lease. Returns false when the lease is not the caller's, or is +/// already released — the node then knows to stop working rather than finish a unit it lost. +/// +/// A heartbeat deliberately does **not** extend `expires_at`. A node that can heartbeat but never +/// finish would otherwise hold a unit forever; the lease is a bounded promise, and a node that +/// needs longer re-claims. +pub async fn heartbeat( + pool: &PgPool, + did: &str, + lease_id: i64, + progress: Option<&Value>, +) -> Result { + let r = sqlx::query( + "UPDATE grid.lease SET heartbeat_at = now(), progress = COALESCE($3, progress) \ + WHERE id = $1 AND did = $2 AND released_at IS NULL", + ) + .bind(lease_id) + .bind(did) + .bind(progress) + .execute(pool) + .await?; + Ok(r.rows_affected() > 0) +} + +/// Give a lease back without a result. Idempotent: releasing an already-released lease is a +/// no-op that reports false. +pub async fn release( + pool: &PgPool, + did: &str, + lease_id: i64, + outcome: &str, +) -> Result { + let r = sqlx::query( + "UPDATE grid.lease SET released_at = now(), outcome = $3 \ + WHERE id = $1 AND did = $2 AND released_at IS NULL", + ) + .bind(lease_id) + .bind(did) + .bind(outcome) + .execute(pool) + .await?; + Ok(r.rows_affected() > 0) +} + +/// Close every lease whose bound has passed. Run by the `grid-reap` job. +/// +/// **This is not what frees the replica slot** — [`claim`] already ignores any lease past +/// `expires_at`, so a unit held by a node that crashed becomes claimable by *another* contributor +/// the moment the lease lapses, with no job run in between. That is what makes a lease honest: a +/// vanished node costs the catalogue one lease duration and nothing more, even if the reaper is +/// down. +/// +/// What the reaper actually does is the other two things, both of which need a row write: +/// +/// 1. **Records the outcome** (`EXPIRED`), so a node's history distinguishes "timed out" from +/// "gave it back" — which trust tiering (§6.1) needs and a derived query cannot recover. +/// 2. **Lets the *same* node claim the unit again.** The self-replication guard in [`claim`] keys +/// on `released_at IS NULL` with no expiry test, deliberately: relaxing it would let a node +/// re-claim a unit it already holds a row for, and the partial unique index would then make +/// `ON CONFLICT DO NOTHING` swallow the insert and hand back an empty result with no +/// explanation. So a node that overran its lease waits for the reaper before it can retry — +/// which is the honest ordering, since its first attempt is genuinely over. +/// +/// Returns how many leases were closed. +pub async fn reap_expired(pool: &PgPool) -> Result { + let r = sqlx::query( + "UPDATE grid.lease SET released_at = now(), outcome = 'EXPIRED' \ + WHERE released_at IS NULL AND expires_at <= now()", + ) + .execute(pool) + .await?; + Ok(r.rows_affected()) +} + +/// Record a signed result and close the lease that produced it, in one transaction — the node +/// must never be able to lose its lease without its submission landing, or vice versa. +/// +/// Re-submitting for the same unit **updates** the row rather than adding a second vote (the +/// `(work_unit_id, did)` unique index), so a retry after a dropped response is safe and a +/// contributor still cannot pad its own quorum. +#[allow(clippy::too_many_arguments)] +pub async fn submit( + pool: &PgPool, + did: &str, + work_unit_id: i64, + lease_id: Option, + digest: &Value, + digest_sig: &str, + stack_version: &str, + reference_build: &str, + aligner: Option<&str>, + record_refs: &Value, +) -> Result { + let mut tx = pool.begin().await?; + let id: (i64,) = sqlx::query_as( + "INSERT INTO grid.submission \ + (work_unit_id, did, lease_id, digest, digest_sig, stack_version, reference_build, aligner, record_refs) \ + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) \ + ON CONFLICT (work_unit_id, did) DO UPDATE SET \ + lease_id = EXCLUDED.lease_id, \ + digest = EXCLUDED.digest, \ + digest_sig = EXCLUDED.digest_sig, \ + stack_version = EXCLUDED.stack_version, \ + reference_build = EXCLUDED.reference_build, \ + aligner = EXCLUDED.aligner, \ + record_refs = EXCLUDED.record_refs, \ + status = 'PENDING', \ + submitted_at = now(), \ + validated_at = NULL \ + RETURNING id", + ) + .bind(work_unit_id) + .bind(did) + .bind(lease_id) + .bind(digest) + .bind(digest_sig) + .bind(stack_version) + .bind(reference_build) + .bind(aligner) + .bind(record_refs) + .fetch_one(&mut *tx) + .await?; + + if let Some(lease) = lease_id { + sqlx::query( + "UPDATE grid.lease SET released_at = now(), outcome = 'SUBMITTED' \ + WHERE id = $1 AND did = $2 AND released_at IS NULL", + ) + .bind(lease) + .bind(did) + .execute(&mut *tx) + .await?; + } + tx.commit().await?; + Ok(id.0) +} + +/// Award credit for an agreed submission. Unique per `(work_unit_id, did)`, so a re-validation +/// cannot pay the same contributor twice for the same unit; the second award is dropped. +/// +/// `user_id` is resolved here rather than at claim time because a contributor may have linked an +/// account between claiming and being validated. +pub async fn award_credit( + pool: &PgPool, + did: &str, + work_unit_id: i64, + submission_id: i64, + cobblestones_milli: i64, + kind: &str, +) -> Result { + let r = sqlx::query( + "INSERT INTO grid.credit (did, user_id, work_unit_id, submission_id, cobblestones_milli, kind) \ + SELECT $1, (SELECT id FROM ident.users WHERE did = $1), $2, $3, $4, $5 \ + ON CONFLICT (work_unit_id, did) DO NOTHING", + ) + .bind(did) + .bind(work_unit_id) + .bind(submission_id) + .bind(cobblestones_milli) + .bind(kind) + .execute(pool) + .await?; + Ok(r.rows_affected() > 0) +} + +/// The public leaderboard: total cobblestones per contributor, best first. +/// +/// `since_days = None` is all-time; `Some(30)` is the rolling window. Contributors with no linked +/// account still appear, by DID and without a handle — the work was done and the board should say +/// so, even though the account link is what `ident.users` would give it a name from. +pub async fn leaderboard( + pool: &PgPool, + since_days: Option, + limit: i64, +) -> Result, DbError> { + let rows = sqlx::query_as::<_, LeaderboardRow>( + "SELECT u.handle, c.did, SUM(c.cobblestones_milli)::bigint AS cobblestones_milli, count(*) AS units \ + FROM grid.credit c \ + LEFT JOIN ident.users u ON u.id = c.user_id \ + WHERE $1::int IS NULL OR c.awarded_at >= now() - ($1::int * interval '1 day') \ + GROUP BY u.handle, c.did \ + ORDER BY cobblestones_milli DESC \ + LIMIT $2", + ) + .bind(since_days) + .bind(limit) + .fetch_all(pool) + .await?; + Ok(rows) +} + +/// A sample the crawl has already resolved, shaped as a work unit. +#[derive(Debug, Clone, sqlx::FromRow)] +pub struct CurationCandidate { + pub sample_accession: String, + pub study_accession: Option, + pub data_kind: String, + pub manifest: Value, + pub est_bases: Option, + pub total_bytes: Option, +} + +/// Samples that could become work units, projected out of what `crawl_project` already stored. +/// +/// The Grid does **not** talk to ENA to build its catalogue. `EnaClient::run_files` and +/// `du_jobs::crawl_project` already resolve every run of a study, group the runs by sample, and +/// materialise the files into `genomics.sequence_file` with their URLs, md5s and sizes. Curation is +/// therefore a projection of tables we already have, not a second ENA integration — which also +/// keeps the fleet off the ENA portal (design §6.2, fair use). +/// +/// **`data_kind` is decided per sample, and the manifest is filtered to match it.** A sample with +/// any CRAM/BAM is a passthrough unit and its manifest carries only the aligned files; otherwise it +/// is a FASTQ unit carrying only reads. `crawl_project::build_libraries` already prefers aligned +/// over FASTQ per sample, so in practice the two agree — but deciding it here as well means the +/// manifest can never carry a file the data kind says the node will not use. +/// +/// With `only_new`, samples that already have a work unit are skipped. That is the nightly path. +/// Passing `false` re-projects everything, which refreshes manifests after a re-crawl. +/// +/// **`est_bases` prefers the measured `base_count`** that ENA publishes on `read_run`, and falls +/// back to `reads × read_length` where a row predates that column. The fallback is only ever a +/// mean-length approximation and is wrong outright for variable-length long reads, so it is a +/// fallback and not the primary. +/// +/// It can still be `NULL`, for a crawled row that has neither — `du-jobs run-once ena-base-count` +/// backfills those. A NULL is deliberate: this figure sets what a contributor is *paid* (§6.3), and +/// a fabricated estimate in a ledger is worse than an honest absence. `grid-curate` warns with a +/// count of the units it published without one. +pub async fn curation_candidates( + pool: &PgPool, + only_new: bool, + limit: i64, +) -> Result, DbError> { + let rows = sqlx::query_as::<_, CurationCandidate>( + "WITH sample AS ( \ + SELECT b.sample_guid, b.accession, \ + CASE WHEN bool_or(sf.file_format IN ('CRAM', 'BAM')) THEN 'CRAM' ELSE 'FASTQ' END AS data_kind \ + FROM core.biosample b \ + JOIN genomics.sequence_library sl ON sl.sample_guid = b.sample_guid \ + JOIN genomics.sequence_file sf ON sf.library_id = sl.id \ + WHERE b.deleted = false \ + AND b.accession IS NOT NULL \ + AND sl.atproto->>'source' = 'ENA' \ + AND sf.file_format IN ('CRAM', 'BAM', 'FASTQ') \ + AND sf.http_locations->0->>'file_url' IS NOT NULL \ + GROUP BY b.sample_guid, b.accession \ + ) \ + SELECT s.accession AS sample_accession, \ + ( SELECT gs.accession FROM pubs.publication_biosample pb \ + JOIN pubs.publication_study ps ON ps.publication_id = pb.publication_id \ + JOIN pubs.genomic_study gs ON gs.id = ps.study_id \ + WHERE pb.sample_guid = s.sample_guid \ + ORDER BY gs.accession LIMIT 1 ) AS study_accession, \ + s.data_kind, \ + jsonb_agg(jsonb_strip_nulls(jsonb_build_object( \ + 'run_accession', sl.atproto->>'run_accession', \ + 'url', sf.http_locations->0->>'file_url', \ + 'index_url', sf.http_locations->0->>'file_index_url', \ + 'md5', sf.checksums->0->>'checksum', \ + 'bytes', sf.file_size_bytes, \ + 'format', sf.file_format \ + )) ORDER BY sl.id, sf.id) AS manifest, \ + ( SELECT SUM(COALESCE(l2.base_count, l2.reads::bigint * l2.read_length::bigint))::bigint \ + FROM genomics.sequence_library l2 WHERE l2.sample_guid = s.sample_guid ) AS est_bases, \ + SUM(sf.file_size_bytes)::bigint AS total_bytes \ + FROM sample s \ + JOIN genomics.sequence_library sl ON sl.sample_guid = s.sample_guid \ + JOIN genomics.sequence_file sf ON sf.library_id = sl.id \ + WHERE ( (s.data_kind = 'CRAM' AND sf.file_format IN ('CRAM', 'BAM')) \ + OR (s.data_kind = 'FASTQ' AND sf.file_format = 'FASTQ') ) \ + AND ( NOT $1 OR NOT EXISTS (SELECT 1 FROM grid.work_unit w WHERE w.sample_accession = s.accession) ) \ + GROUP BY s.sample_guid, s.accession, s.data_kind \ + ORDER BY s.accession \ + LIMIT $2", + ) + .bind(only_new) + .bind(limit) + .fetch_all(pool) + .await?; + Ok(rows) +} + +/// One submission awaiting validation, with everything the agreement test needs. +#[derive(Debug, Clone, sqlx::FromRow)] +pub struct PendingSubmission { + pub id: i64, + pub did: String, + pub digest: Value, + pub stack_version: String, + pub reference_build: String, + pub submitted_at: chrono::DateTime, +} + +/// A unit with unvalidated submissions. +#[derive(Debug, Clone, sqlx::FromRow)] +pub struct PendingUnit { + pub id: i64, + pub sample_accession: String, + pub required_replicas: i16, + pub est_bases: Option, + pub data_kind: String, +} + +/// A contributor's grid history, which is what trust tiering is derived from (§6.1). +/// +/// Deliberately *not* the social reputation score: grid trust must be earned by grid work, or a +/// well-regarded community member could canonicalize bad results on reputation alone. +#[derive(Debug, Clone, Copy, sqlx::FromRow)] +pub struct GridHistory { + pub agreed: i64, + pub divergent: i64, +} + +/// Units carrying at least one `PENDING` submission, oldest first. +pub async fn units_awaiting_validation( + pool: &PgPool, + limit: i64, +) -> Result, DbError> { + let rows = sqlx::query_as::<_, PendingUnit>( + "SELECT w.id, w.sample_accession, w.required_replicas, w.est_bases, w.data_kind \ + FROM grid.work_unit w \ + WHERE w.state IN ('AVAILABLE', 'CONTESTED') \ + AND EXISTS (SELECT 1 FROM grid.submission s \ + WHERE s.work_unit_id = w.id AND s.status = 'PENDING') \ + ORDER BY w.id \ + LIMIT $1", + ) + .bind(limit) + .fetch_all(pool) + .await?; + Ok(rows) +} + +/// Every submission on a unit that has not been ruled divergent, oldest first. +/// +/// `AGREED` rows are included as well as `PENDING` ones: a unit re-opened by a shadow check is +/// judged over its whole history, not just the newest arrival. +pub async fn submissions_for_validation( + pool: &PgPool, + work_unit_id: i64, +) -> Result, DbError> { + let rows = sqlx::query_as::<_, PendingSubmission>( + "SELECT id, did, digest, stack_version, reference_build, submitted_at \ + FROM grid.submission \ + WHERE work_unit_id = $1 AND status <> 'DIVERGENT' \ + ORDER BY submitted_at, id", + ) + .bind(work_unit_id) + .fetch_all(pool) + .await?; + Ok(rows) +} + +/// A contributor's agreed and divergent counts. +pub async fn grid_history(pool: &PgPool, did: &str) -> Result { + let row: GridHistory = sqlx::query_as( + "SELECT COUNT(*) FILTER (WHERE status = 'AGREED') AS agreed, \ + COUNT(*) FILTER (WHERE status = 'DIVERGENT') AS divergent \ + FROM grid.submission WHERE did = $1", + ) + .bind(did) + .fetch_one(pool) + .await?; + Ok(row) +} + +/// Promote a unit to `CANONICAL`: store the agreed digest, mark the winning submissions `AGREED` +/// and the rest `DIVERGENT`, all in one transaction. +/// +/// A partial application here would be the worst possible state — a canonical unit whose +/// submissions still read `PENDING` would be re-judged on the next pass and could be credited +/// twice, which the `grid.credit` unique index would then silently swallow. So it is all or none. +pub async fn canonicalize( + pool: &PgPool, + work_unit_id: i64, + canonical: &Value, + agreed_ids: &[i64], + divergent_ids: &[i64], +) -> Result<(), DbError> { + let mut tx = pool.begin().await?; + sqlx::query( + "UPDATE grid.work_unit \ + SET state = 'CANONICAL', canonical_digest = $2, canonical_at = now(), updated_at = now() \ + WHERE id = $1", + ) + .bind(work_unit_id) + .bind(canonical) + .execute(&mut *tx) + .await?; + sqlx::query( + "UPDATE grid.submission SET status = 'AGREED', validated_at = now() WHERE id = ANY($1)", + ) + .bind(agreed_ids) + .execute(&mut *tx) + .await?; + sqlx::query( + "UPDATE grid.submission SET status = 'DIVERGENT', validated_at = now() WHERE id = ANY($1)", + ) + .bind(divergent_ids) + .execute(&mut *tx) + .await?; + tx.commit().await?; + Ok(()) +} + +/// Mark a unit contested and raise the bar: the submissions on it disagree and none of them has +/// earned the right to be believed, so the unit needs another independent result. +/// +/// Nothing is marked `DIVERGENT` here. With two conflicting clusters and no quorum there is no +/// evidence about *which* is wrong, and penalising a contributor on a coin-flip would punish +/// honest work. The next replica breaks the tie, and that pass assigns blame. +pub async fn contest(pool: &PgPool, work_unit_id: i64, note: &str) -> Result<(), DbError> { + sqlx::query( + "UPDATE grid.work_unit \ + SET state = 'CONTESTED', required_replicas = required_replicas + 1, \ + note = $2, updated_at = now() \ + WHERE id = $1", + ) + .bind(work_unit_id) + .bind(note) + .execute(pool) + .await?; + Ok(()) +} + +/// With probability `rate`, ask for one more independent result without contesting anything — +/// the shadow spot-check. Returns whether the shadow was requested. +/// +/// Used when a trusted node's lone submission would otherwise canonicalize, so that trust is +/// re-earned rather than assumed indefinitely. The unit simply stays claimable with a higher +/// replica bar: the shadow then arrives through the ordinary claim path and the next validation +/// pass either confirms or contests it. No `SHADOW` state, no schema column, no second code path. +/// +/// **The draw happens in the database, deliberately.** §6.2 requires spot-checks to be +/// AppView-chosen rather than self-selected, so the rule must be one a contributor cannot compute +/// in advance — which rules out anything derived from the unit id or the digest. Doing it in SQL +/// also avoids pulling a random-number crate into the workspace for a single coin flip. +pub async fn maybe_request_shadow( + pool: &PgPool, + work_unit_id: i64, + rate: f64, +) -> Result { + let r = sqlx::query( + "UPDATE grid.work_unit \ + SET required_replicas = GREATEST(required_replicas, 2), \ + note = 'shadow spot-check', updated_at = now() \ + WHERE id = $1 AND state = 'AVAILABLE' AND random() < $2", + ) + .bind(work_unit_id) + .bind(rate) + .execute(pool) + .await?; + Ok(r.rows_affected() > 0) +} + +/// Register (or refresh) a contributing node in the shared fleet registry. +/// +/// Reuses `fed.pds_node` rather than adding a `grid.node`: the columns the Grid needs — DID, +/// capabilities, heartbeat, software version — are the columns that table was built with and +/// never wired to anything. +pub async fn register_node( + pool: &PgPool, + did: &str, + software_version: &str, + capabilities: &Value, + os_info: Option<&str>, +) -> Result { + let id: (i64,) = sqlx::query_as( + "INSERT INTO fed.pds_node (did, software_version, capabilities, os_info, status, last_heartbeat) \ + VALUES ($1, $2, $3, $4, 'ONLINE', now()) \ + ON CONFLICT (did) DO UPDATE SET \ + software_version = EXCLUDED.software_version, \ + capabilities = EXCLUDED.capabilities, \ + os_info = EXCLUDED.os_info, \ + status = 'ONLINE', \ + last_heartbeat = now(), \ + updated_at = now() \ + RETURNING id", + ) + .bind(did) + .bind(software_version) + .bind(capabilities) + .bind(os_info) + .fetch_one(pool) + .await?; + Ok(id.0) +} + +/// The registry id of a node, if it has registered. `claim` records it on the lease so the fleet +/// view can attribute work to a machine; a node that never registered still gets to work, it is +/// simply anonymous in that view. +pub async fn node_id_for_did(pool: &PgPool, did: &str) -> Result, DbError> { + let row: Option<(i64,)> = sqlx::query_as("SELECT id FROM fed.pds_node WHERE did = $1") + .bind(did) + .fetch_optional(pool) + .await?; + Ok(row.map(|r| r.0)) +} + +/// A work unit as the public `/grid/work/{accession}` endpoint shows it. +#[derive(Debug, Clone, Serialize, sqlx::FromRow)] +pub struct PublicWorkUnit { + pub sample_accession: String, + pub study_accession: Option, + pub data_kind: String, + pub state: String, + pub canonical_digest: Option, + pub canonical_at: Option>, + /// How many independent contributors agreed. The number is the reason to believe the digest, + /// so publishing the result without it would be publishing a claim with its evidence removed. + pub replicas_agreed: i64, +} + +/// One unit by ENA accession, for the public result page. +pub async fn work_unit_public( + pool: &PgPool, + sample_accession: &str, +) -> Result, DbError> { + let row = sqlx::query_as::<_, PublicWorkUnit>( + "SELECT w.sample_accession, w.study_accession, w.data_kind, w.state, \ + w.canonical_digest, w.canonical_at, \ + (SELECT count(*) FROM grid.submission s \ + WHERE s.work_unit_id = w.id AND s.status = 'AGREED') AS replicas_agreed \ + FROM grid.work_unit w WHERE w.sample_accession = $1", + ) + .bind(sample_accession) + .fetch_optional(pool) + .await?; + Ok(row) +} + +/// Grid throughput, for the public stats endpoint. +/// +/// One round trip rather than five: these are counts over small indexed sets, and a stats endpoint +/// that costs five queries is one that gets called on every page load and then blamed for load. +pub async fn stats(pool: &PgPool) -> Result { + let row: (i64, i64, i64, i64, i64, i64) = sqlx::query_as( + "SELECT (SELECT count(*) FROM grid.work_unit), \ + (SELECT count(*) FROM grid.work_unit WHERE state = 'CANONICAL'), \ + (SELECT count(*) FROM grid.work_unit WHERE state = 'CONTESTED'), \ + (SELECT count(*) FROM grid.lease WHERE released_at IS NULL AND expires_at > now()), \ + (SELECT count(DISTINCT did) FROM grid.submission), \ + (SELECT COALESCE(SUM(cobblestones_milli), 0)::bigint FROM grid.credit)", + ) + .fetch_one(pool) + .await?; + Ok(serde_json::json!({ + "units_total": row.0, + "units_canonical": row.1, + "units_contested": row.2, + "leases_active": row.3, + "contributors": row.4, + "cobblestones_awarded": row.5 as f64 / COBBLESTONE as f64, + })) +} + +/// What one contributor's own grid participation looks like: the leases it holds right now, its +/// agreed/divergent history, its credit, and where it sits on the board. +/// +/// Authenticated rather than public, because it is the caller's own work — the leaderboard shows +/// totals, and this shows the rows behind one contributor's total. +pub async fn standing(pool: &PgPool, did: &str) -> Result { + let held = sqlx::query_as::<_, ClaimedUnit>( + "SELECT l.id AS lease_id, l.work_unit_id, w.sample_accession, w.study_accession, \ + w.data_kind, w.manifest, w.est_bases, w.total_bytes, l.expires_at \ + FROM grid.lease l JOIN grid.work_unit w ON w.id = l.work_unit_id \ + WHERE l.did = $1 AND l.released_at IS NULL AND l.expires_at > now() \ + ORDER BY l.expires_at", + ) + .bind(did) + .fetch_all(pool) + .await?; + + let h = grid_history(pool, did).await?; + let (milli, units, rank): (i64, i64, i64) = sqlx::query_as( + "SELECT COALESCE(SUM(c.cobblestones_milli), 0)::bigint, \ + COUNT(c.id)::bigint, \ + COALESCE(( SELECT count(*) + 1 FROM ( \ + SELECT did, SUM(cobblestones_milli) AS t FROM grid.credit GROUP BY did \ + ) b WHERE b.t > COALESCE((SELECT SUM(cobblestones_milli) FROM grid.credit WHERE did = $1), 0) \ + ), 1)::bigint \ + FROM grid.credit c WHERE c.did = $1", + ) + .bind(did) + .fetch_one(pool) + .await?; + + // A contributor with no credit has no row on the leaderboard, so it has no rank either. + // Reporting the position it *would* hold reads as "you are last" to someone who has simply not + // finished their first unit yet — a discouraging answer to a question they did not ask. `null` + // says "unranked", which is what is true. + let rank = (units > 0).then_some(rank); + + Ok(serde_json::json!({ + "leases": held, + "agreed": h.agreed, + "divergent": h.divergent, + "cobblestones": milli as f64 / COBBLESTONE as f64, + "units_credited": units, + "rank": rank, + })) +} + +/// The user account a contributing DID resolves to, if any. Sybil resistance leans on this: an +/// untrusted submission cannot canonicalise alone, so a lone account-less attacker cannot inject +/// a canonical result. +pub async fn user_for_did(pool: &PgPool, did: &str) -> Result, DbError> { + let row: Option<(Uuid,)> = sqlx::query_as("SELECT id FROM ident.users WHERE did = $1") + .bind(did) + .fetch_optional(pool) + .await?; + Ok(row.map(|r| r.0)) +} + +#[cfg(test)] +mod tests { + use super::messages; + + /// The signed strings are a cross-repo contract with the Navigator edge. This test exists to + /// make an accidental reformat fail here, in the repo that defines them, rather than as a 403 + /// against a released desktop build that signs the old bytes. + #[test] + fn canonical_messages_are_stable() { + 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==" + ); + } + + /// Every message starts with its own operation tag, so a signature harvested from one + /// endpoint cannot be replayed against another. + #[test] + fn each_message_is_domain_separated() { + 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 tags: Vec<&str> = all.iter().map(|m| m.split('\n').next().unwrap()).collect(); + let mut sorted = tags.clone(); + sorted.sort_unstable(); + sorted.dedup(); + assert_eq!( + sorted.len(), + tags.len(), + "two grid messages share an operation tag: {tags:?}" + ); + } +} diff --git a/rust/crates/du-db/src/lib.rs b/rust/crates/du-db/src/lib.rs index d865de4..3b38c20 100644 --- a/rust/crates/du-db/src/lib.rs +++ b/rust/crates/du-db/src/lib.rs @@ -24,6 +24,7 @@ pub mod exchange; pub mod fed; pub mod fed_subject; pub mod genome_region; +pub mod grid; pub mod haplogroup; pub mod ibd; pub mod identifier; diff --git a/rust/crates/du-db/src/sequence.rs b/rust/crates/du-db/src/sequence.rs index adefc2b..3fcd148 100644 --- a/rust/crates/du-db/src/sequence.rs +++ b/rust/crates/du-db/src/sequence.rs @@ -37,6 +37,8 @@ pub struct NewSeqLibrary { pub instrument: Option, pub reads: Option, pub read_length: Option, + /// Total bases in the run, as ENA reports it. The Grid weighs a work unit by this (§6.3). + pub base_count: Option, pub paired_end: Option, pub run_date: Option, /// Source run/analysis accession (e.g. ENA `ERR...`), stored as provenance. @@ -89,7 +91,11 @@ pub async fn ingest_libraries( .fetch_one(&mut *tx) .await?; if existing > 0 { - return Ok(IngestReport { library_ids: Vec::new(), files_created: 0, skipped_existing: true }); + return Ok(IngestReport { + library_ids: Vec::new(), + files_created: 0, + skipped_existing: true, + }); } let mut library_ids = Vec::with_capacity(libs.len()); @@ -97,14 +103,15 @@ pub async fn ingest_libraries( for lib in libs { let lib_id: i64 = sqlx::query_scalar( "INSERT INTO genomics.sequence_library \ - (sample_guid, run_date, instrument, reads, read_length, paired_end, atproto) \ - VALUES ($1, $2, $3, $4, $5, $6, $7) RETURNING id", + (sample_guid, run_date, instrument, reads, read_length, base_count, paired_end, atproto) \ + VALUES ($1, $2, $3, $4, $5, $6, $7, $8) RETURNING id", ) .bind(guid.0) .bind(lib.run_date) .bind(&lib.instrument) .bind(lib.reads) .bind(lib.read_length) + .bind(lib.base_count) .bind(lib.paired_end) // No ATP record for academic runs; reuse the JSONB slot for source provenance. .bind(json!({ "source": "ENA", "run_accession": lib.external_run_ref })) @@ -143,5 +150,61 @@ pub async fn ingest_libraries( } tx.commit().await?; - Ok(IngestReport { library_ids, files_created, skipped_existing: false }) + Ok(IngestReport { + library_ids, + files_created, + skipped_existing: false, + }) +} + +/// One crawled ENA run that has no `base_count` yet — the backfill job's work list. +#[derive(Debug, Clone, sqlx::FromRow)] +pub struct MissingBaseCount { + pub id: i64, + /// The ENA run accession, from the `atproto` provenance slot the crawl writes. + pub run_accession: String, +} + +/// Crawled ENA runs still missing a measured `base_count`, oldest first. +/// +/// Only rows the crawl created: the accession comes from the `{source: "ENA", run_accession}` +/// provenance the crawl writes, so a hand-loaded library with no ENA origin is never queried +/// against ENA — there would be nothing there to find. +pub async fn runs_missing_base_count( + pool: &PgPool, + limit: i64, +) -> Result, DbError> { + let rows = sqlx::query_as::<_, MissingBaseCount>( + "SELECT id, atproto->>'run_accession' AS run_accession \ + FROM genomics.sequence_library \ + WHERE base_count IS NULL \ + AND atproto->>'source' = 'ENA' \ + AND COALESCE(atproto->>'run_accession', '') <> '' \ + ORDER BY id \ + LIMIT $1", + ) + .bind(limit) + .fetch_all(pool) + .await?; + Ok(rows) +} + +/// Record a measured `base_count` against one library. +/// +/// Only ever fills a NULL. A row that already carries a figure is left alone, so re-running the +/// backfill cannot overwrite a measurement with a later, differently-reported one and silently +/// change what a contributor was paid for work already credited. +pub async fn set_base_count( + pool: &PgPool, + library_id: i64, + base_count: i64, +) -> Result { + let r = sqlx::query( + "UPDATE genomics.sequence_library SET base_count = $2 WHERE id = $1 AND base_count IS NULL", + ) + .bind(library_id) + .bind(base_count) + .execute(pool) + .await?; + Ok(r.rows_affected() > 0) } diff --git a/rust/crates/du-db/tests/grid.rs b/rust/crates/du-db/tests/grid.rs new file mode 100644 index 0000000..7011c43 --- /dev/null +++ b/rust/crates/du-db/tests/grid.rs @@ -0,0 +1,1267 @@ +//! Integration test: the Grid's work-unit coordination against a live Postgres. +//! +//! The claim path is the one piece of genuinely new concurrency in the Grid, and it is the piece +//! that unit tests cannot reach: `FOR UPDATE SKIP LOCKED`, a partial unique index, and replica +//! arithmetic over two other tables only mean anything inside a real transaction. So these tests +//! run against a real database or not at all. +//! +//! Skips (passes) when `DATABASE_URL` is unset, so `cargo test` stays green without one. To run: +//! DATABASE_URL=postgres://…/postgres cargo test -p du-db --test grid -- --nocapture + +use du_db::grid::{self, NewWorkUnit, COBBLESTONE}; +use serde_json::json; + +fn database_url() -> Option { + std::env::var("DATABASE_URL").ok().filter(|s| !s.is_empty()) +} + +const DID_A: &str = "did:plc:aaaaaaaaaaaaaaaaaaaaaaaa"; +const DID_B: &str = "did:plc:bbbbbbbbbbbbbbbbbbbbbbbb"; +const DID_C: &str = "did:plc:cccccccccccccccccccccccc"; + +fn unit(acc: &str, kind: &str) -> NewWorkUnit { + NewWorkUnit { + sample_accession: acc.into(), + study_accession: Some("PRJEB00000".into()), + data_kind: kind.into(), + manifest: json!([{ "run_accession": "ERR0000000", "url": "ftp://example/f.cram", + "md5": "d41d8cd98f00b204e9800998ecf8427e", "bytes": 1024, "format": "CRAM" }]), + est_bases: Some(90_000_000_000), + total_bytes: Some(1024), + } +} + +fn kinds(v: &[&str]) -> Vec { + v.iter().map(|s| s.to_string()).collect() +} + +#[tokio::test] +async fn claim_respects_replicas_and_never_self_replicates() { + let Some(url) = database_url() else { + eprintln!("DATABASE_URL unset — skipping live-DB test"); + return; + }; + let db = du_db::testing::ephemeral_db(&url) + .await + .expect("ephemeral db"); + let pool = db.pool().clone(); + + let id = grid::upsert_work_unit(&pool, &unit("SAMEA0000001", "CRAM")) + .await + .unwrap(); + + // A takes it. required_replicas defaults to 2, so one slot remains. + let a = grid::claim(&pool, DID_A, None, &kinds(&["CRAM"]), 5, 3600) + .await + .unwrap(); + assert_eq!(a.len(), 1, "A should get the only unit"); + assert_eq!(a[0].work_unit_id, id); + assert_eq!(a[0].data_kind, "CRAM"); + assert_eq!( + a[0].manifest[0]["format"], "CRAM", + "the node gets the fetch manifest with the claim" + ); + + // A again: a contributor must never replicate itself, so there is nothing left for it. + let a2 = grid::claim(&pool, DID_A, None, &kinds(&["CRAM"]), 5, 3600) + .await + .unwrap(); + assert!( + a2.is_empty(), + "the same DID must not be handed the same unit twice" + ); + + // B takes the second replica slot. + let b = grid::claim(&pool, DID_B, None, &kinds(&["CRAM"]), 5, 3600) + .await + .unwrap(); + assert_eq!(b.len(), 1, "B should get the second replica"); + + // C finds nothing: both replicas are in flight. + let c = grid::claim(&pool, DID_C, None, &kinds(&["CRAM"]), 5, 3600) + .await + .unwrap(); + assert!( + c.is_empty(), + "a unit must not be handed out beyond required_replicas" + ); +} + +#[tokio::test] +async fn data_kind_filter_is_honoured() { + let Some(url) = database_url() else { + return; + }; + let db = du_db::testing::ephemeral_db(&url) + .await + .expect("ephemeral db"); + let pool = db.pool().clone(); + + grid::upsert_work_unit(&pool, &unit("SAMEA0000010", "CRAM")) + .await + .unwrap(); + grid::upsert_work_unit(&pool, &unit("SAMEA0000011", "FASTQ")) + .await + .unwrap(); + + // A node that only wants passthrough work is never offered a FASTQ unit. + let only_cram = grid::claim(&pool, DID_A, None, &kinds(&["CRAM"]), 10, 3600) + .await + .unwrap(); + assert_eq!(only_cram.len(), 1); + assert_eq!(only_cram[0].sample_accession, "SAMEA0000010"); + + let both = grid::claim(&pool, DID_B, None, &kinds(&["CRAM", "FASTQ"]), 10, 3600) + .await + .unwrap(); + assert_eq!(both.len(), 2, "a node advertising both kinds gets both"); +} + +#[tokio::test] +async fn a_lapsed_lease_frees_the_slot_immediately_and_the_reaper_records_the_outcome() { + let Some(url) = database_url() else { + return; + }; + let db = du_db::testing::ephemeral_db(&url) + .await + .expect("ephemeral db"); + let pool = db.pool().clone(); + + grid::upsert_work_unit(&pool, &unit("SAMEA0000020", "CRAM")) + .await + .unwrap(); + + // A and B take both replica slots on leases that have already lapsed. + for did in [DID_A, DID_B] { + let got = grid::claim(&pool, did, None, &kinds(&["CRAM"]), 1, -1) + .await + .unwrap(); + assert_eq!(got.len(), 1); + } + + // C can claim RIGHT NOW, with no reaper run in between: `claim` ignores any lease past its + // bound, so a unit held by a node that crashed frees itself. This is what makes the lease an + // honest promise — a vanished node costs the catalogue one lease duration even if the reaper + // is down. (The first version of this test asserted the opposite, encoding the design's story + // that reclamation is what frees the slot. The code is right and the story was wrong.) + let c = grid::claim(&pool, DID_C, None, &kinds(&["CRAM"]), 1, 3600) + .await + .unwrap(); + assert_eq!(c.len(), 1, "a lapsed lease does not hold a replica slot"); + + // The reaper closes the two lapsed leases and leaves C's live one alone. + assert_eq!( + grid::reap_expired(&pool).await.unwrap(), + 2, + "only the lapsed leases are closed" + ); + assert_eq!( + grid::reap_expired(&pool).await.unwrap(), + 0, + "reaping is idempotent" + ); + + let outcomes: Vec<(String, Option)> = + sqlx::query_as("SELECT did, outcome FROM grid.lease ORDER BY did") + .fetch_all(&pool) + .await + .unwrap(); + assert_eq!(outcomes.len(), 3); + assert_eq!(outcomes[0].1.as_deref(), Some("EXPIRED"), "A timed out"); + assert_eq!(outcomes[1].1.as_deref(), Some("EXPIRED"), "B timed out"); + assert_eq!(outcomes[2].1, None, "C is still working"); +} + +/// The reaper's *other* job: until it runs, a node that overran its own lease cannot re-claim the +/// unit, because the self-replication guard keys on an unreleased lease regardless of expiry. +/// Relaxing that guard would collide with the partial unique index and hand back a silently empty +/// result instead, so the wait is deliberate. +#[tokio::test] +async fn a_node_that_overran_its_lease_can_retry_only_after_the_reaper_runs() { + let Some(url) = database_url() else { + return; + }; + let db = du_db::testing::ephemeral_db(&url) + .await + .expect("ephemeral db"); + let pool = db.pool().clone(); + + grid::upsert_work_unit(&pool, &unit("SAMEA0000021", "CRAM")) + .await + .unwrap(); + assert_eq!( + grid::claim(&pool, DID_A, None, &kinds(&["CRAM"]), 1, -1) + .await + .unwrap() + .len(), + 1 + ); + + assert!( + grid::claim(&pool, DID_A, None, &kinds(&["CRAM"]), 1, 3600) + .await + .unwrap() + .is_empty(), + "its own lapsed-but-open lease still blocks it" + ); + + grid::reap_expired(&pool).await.unwrap(); + + assert_eq!( + grid::claim(&pool, DID_A, None, &kinds(&["CRAM"]), 1, 3600) + .await + .unwrap() + .len(), + 1, + "once the outcome is recorded, the node may take a fresh lease" + ); +} + +#[tokio::test] +async fn submit_closes_the_lease_and_a_resubmit_updates_rather_than_duplicating() { + let Some(url) = database_url() else { + return; + }; + let db = du_db::testing::ephemeral_db(&url) + .await + .expect("ephemeral db"); + let pool = db.pool().clone(); + + let unit_id = grid::upsert_work_unit(&pool, &unit("SAMEA0000030", "CRAM")) + .await + .unwrap(); + let claimed = grid::claim(&pool, DID_A, None, &kinds(&["CRAM"]), 1, 3600) + .await + .unwrap(); + let lease = claimed[0].lease_id; + + assert!( + grid::heartbeat(&pool, DID_A, lease, Some(&json!({"stage": "download"}))) + .await + .unwrap() + ); + + let digest = + json!({"unit": "SAMEA0000030", "calls": {"sex": "XY", "y_terminal": "R-FGC29071"}}); + let s1 = grid::submit( + &pool, + DID_A, + unit_id, + Some(lease), + &digest, + "sig-1", + "1.7.0", + "chm13v2.0", + None, + &json!([]), + ) + .await + .unwrap(); + + // The lease closed with the submission, in the same transaction. A heartbeat on it now fails, + // which is how a node learns it no longer holds the unit. + assert!(!grid::heartbeat(&pool, DID_A, lease, None).await.unwrap()); + + // A retry after a dropped response must not become a second vote. + let s2 = grid::submit( + &pool, + DID_A, + unit_id, + None, + &digest, + "sig-2", + "1.7.0", + "chm13v2.0", + None, + &json!([]), + ) + .await + .unwrap(); + assert_eq!( + s1, s2, + "resubmitting updates the same row rather than adding a vote" + ); + + let n: (i64,) = sqlx::query_as("SELECT count(*) FROM grid.submission WHERE work_unit_id = $1") + .bind(unit_id) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(n.0, 1, "one contributor, one submission"); + + // Having submitted, A is not offered the unit again even though a replica slot is open. + assert!(grid::claim(&pool, DID_A, None, &kinds(&["CRAM"]), 1, 3600) + .await + .unwrap() + .is_empty()); + assert_eq!( + grid::claim(&pool, DID_B, None, &kinds(&["CRAM"]), 1, 3600) + .await + .unwrap() + .len(), + 1, + "the second replica slot is still open to a different contributor" + ); +} + +#[tokio::test] +async fn credit_is_awarded_once_per_unit_and_totals_on_the_leaderboard() { + let Some(url) = database_url() else { + return; + }; + let db = du_db::testing::ephemeral_db(&url) + .await + .expect("ephemeral db"); + let pool = db.pool().clone(); + + let mut submissions = Vec::new(); + for acc in ["SAMEA0000040", "SAMEA0000041"] { + let unit_id = grid::upsert_work_unit(&pool, &unit(acc, "CRAM")) + .await + .unwrap(); + let claimed = grid::claim(&pool, DID_A, None, &kinds(&["CRAM"]), 1, 3600) + .await + .unwrap(); + let sub = grid::submit( + &pool, + DID_A, + unit_id, + Some(claimed[0].lease_id), + &json!({}), + "sig", + "1.7.0", + "chm13v2.0", + None, + &json!([]), + ) + .await + .unwrap(); + submissions.push((unit_id, sub)); + } + + for (unit_id, sub) in &submissions { + assert!(grid::award_credit( + &pool, + DID_A, + *unit_id, + *sub, + 5 * COBBLESTONE, + "QUORUM_AGREE" + ) + .await + .unwrap()); + } + // A re-validation must not pay twice for the same unit. + let (unit_id, sub) = submissions[0]; + assert!( + !grid::award_credit(&pool, DID_A, unit_id, sub, 5 * COBBLESTONE, "QUORUM_AGREE") + .await + .unwrap(), + "a second award for the same (unit, DID) is dropped" + ); + + let board = grid::leaderboard(&pool, None, 10).await.unwrap(); + assert_eq!(board.len(), 1); + assert_eq!(board[0].did, DID_A); + assert_eq!(board[0].units, 2); + assert_eq!( + board[0].cobblestones_milli, + 10 * COBBLESTONE, + "two units at five cobblestones each" + ); + // No linked account, so the board shows the DID and no handle — the work still counts. + assert!(board[0].handle.is_none()); +} + +/// Two nodes claiming at the same instant must take *different* units. This is the whole point of +/// `SKIP LOCKED`: without it one claimer blocks on the other's row lock, and a fleet serialises. +#[tokio::test] +async fn concurrent_claims_take_disjoint_units() { + let Some(url) = database_url() else { + return; + }; + let db = du_db::testing::ephemeral_db(&url) + .await + .expect("ephemeral db"); + let pool = db.pool().clone(); + + for i in 0..20 { + grid::upsert_work_unit(&pool, &unit(&format!("SAMEA10000{i:02}"), "CRAM")) + .await + .unwrap(); + } + + let (p1, p2) = (pool.clone(), pool.clone()); + let (r1, r2) = tokio::join!( + tokio::spawn( + async move { grid::claim(&p1, DID_A, None, &kinds(&["CRAM"]), 10, 3600).await } + ), + tokio::spawn( + async move { grid::claim(&p2, DID_B, None, &kinds(&["CRAM"]), 10, 3600).await } + ), + ); + let a = r1.unwrap().unwrap(); + let b = r2.unwrap().unwrap(); + assert_eq!(a.len(), 10); + assert_eq!(b.len(), 10); + + // Distinct DIDs may legitimately share a unit as replicas, so overlap is not an error here — + // what matters is that neither claimer blocked and both got a full batch. + let total: std::collections::HashSet = + a.iter().chain(b.iter()).map(|u| u.work_unit_id).collect(); + assert!( + total.len() >= 10, + "claims resolved without serialising; {} distinct units", + total.len() + ); +} + +// ── curation ────────────────────────────────────────────────────────────────── +// +// These seed through the real ingest path (`biosample::upsert_by_accession` + +// `sequence::ingest_libraries`) rather than hand-written INSERTs, on purpose: the curation query +// reads `http_locations->0->>'file_url'` and `checksums->0->>'checksum'`, which are JSONB shapes +// that only `ingest_libraries` defines. A test that wrote its own rows could agree with the query +// while both disagreed with what the crawl actually stores. + +use du_db::sequence::{NewSeqFile, NewSeqLibrary}; + +fn seq_file( + name: &str, + fmt: &str, + url: &str, + idx: Option<&str>, + md5: &str, + bytes: i64, +) -> NewSeqFile { + NewSeqFile { + file_name: name.into(), + file_format: Some(fmt.into()), + file_size_bytes: Some(bytes), + file_url: url.into(), + file_index_url: idx.map(Into::into), + md5: Some(md5.into()), + aligner: None, + target_reference: None, + } +} + +fn seq_lib( + run: &str, + reads: Option, + read_length: Option, + files: Vec, +) -> NewSeqLibrary { + seq_lib_bases(run, reads, read_length, None, files) +} + +fn seq_lib_bases( + run: &str, + reads: Option, + read_length: Option, + base_count: Option, + files: Vec, +) -> NewSeqLibrary { + NewSeqLibrary { + instrument: Some("Illumina NovaSeq 6000".into()), + reads, + read_length, + base_count, + paired_end: Some(true), + run_date: None, + external_run_ref: run.into(), + files, + } +} + +/// Seed one crawled ENA sample and return nothing — the accession is the handle. +async fn seed_sample(pool: &sqlx::PgPool, accession: &str, libs: Vec) { + let (guid, _) = du_db::biosample::upsert_by_accession(pool, accession, "EXTERNAL", None) + .await + .expect("upsert biosample"); + du_db::sequence::ingest_libraries(pool, guid, &libs) + .await + .expect("ingest"); +} + +#[tokio::test] +async fn curation_projects_crawled_samples_into_work_units() { + let Some(url) = database_url() else { + return; + }; + let db = du_db::testing::ephemeral_db(&url) + .await + .expect("ephemeral db"); + let pool = db.pool().clone(); + + seed_sample( + &pool, + "SAMEA2000001", + vec![seq_lib( + "ERR2000001", + Some(400_000_000), + Some(150), + vec![seq_file( + "s1.cram", + "CRAM", + "ftp.sra.ebi.ac.uk/vol1/run/ERR200/s1.cram", + Some("ftp.sra.ebi.ac.uk/vol1/run/ERR200/s1.cram.crai"), + "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + 12_000_000_000, + )], + )], + ) + .await; + seed_sample( + &pool, + "SAMEA2000002", + vec![seq_lib_bases( + "ERR2000002", + Some(300_000_000), + None, // ENA never reports read_length + Some(45_000_000_000), // …but it does report base_count, which is what we weigh by + vec![ + seq_file( + "r_1.fastq.gz", + "FASTQ", + "ftp/r_1.fastq.gz", + None, + "b".repeat(32).as_str(), + 9_000_000_000, + ), + seq_file( + "r_2.fastq.gz", + "FASTQ", + "ftp/r_2.fastq.gz", + None, + "c".repeat(32).as_str(), + 9_100_000_000, + ), + ], + )], + ) + .await; + + let got = du_db::grid::curation_candidates(&pool, true, 100) + .await + .unwrap(); + assert_eq!(got.len(), 2, "both crawled samples are candidates"); + + let cram = got + .iter() + .find(|c| c.sample_accession == "SAMEA2000001") + .unwrap(); + assert_eq!( + cram.data_kind, "CRAM", + "a sample with an aligned file is a passthrough unit" + ); + assert_eq!(cram.manifest.as_array().unwrap().len(), 1); + assert_eq!(cram.manifest[0]["format"], "CRAM"); + assert_eq!( + cram.manifest[0]["run_accession"], "ERR2000001", + "the run accession survives the crawl's atproto slot" + ); + assert_eq!( + cram.manifest[0]["url"], + "ftp.sra.ebi.ac.uk/vol1/run/ERR200/s1.cram" + ); + assert_eq!( + cram.manifest[0]["index_url"], + "ftp.sra.ebi.ac.uk/vol1/run/ERR200/s1.cram.crai" + ); + assert_eq!(cram.manifest[0]["md5"], "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"); + assert_eq!(cram.total_bytes, Some(12_000_000_000)); + assert_eq!( + cram.est_bases, + Some(400_000_000 * 150), + "reads × read_length when both are known" + ); + + let fq = got + .iter() + .find(|c| c.sample_accession == "SAMEA2000002") + .unwrap(); + assert_eq!(fq.data_kind, "FASTQ", "no aligned file ⇒ the node realigns"); + assert_eq!( + fq.manifest.as_array().unwrap().len(), + 2, + "both mates are in the manifest" + ); + assert!( + fq.manifest[0].get("index_url").is_none(), + "jsonb_strip_nulls drops the absent sidecar" + ); + assert_eq!(fq.total_bytes, Some(18_100_000_000)); + assert_eq!( + fq.est_bases, + Some(45_000_000_000), + "the measured base_count is what the per-Gbp credit term weighs a realignment by" + ); +} + +/// A sample carrying both an aligned file and its FASTQ is one *passthrough* unit, and its +/// manifest must not also list reads the node will never open. +#[tokio::test] +async fn the_manifest_is_filtered_to_the_chosen_data_kind() { + let Some(url) = database_url() else { + return; + }; + let db = du_db::testing::ephemeral_db(&url) + .await + .expect("ephemeral db"); + let pool = db.pool().clone(); + + seed_sample( + &pool, + "SAMEA2000010", + vec![ + seq_lib( + "ERR2000010", + Some(1), + Some(1), + vec![seq_file( + "a.cram", + "CRAM", + "ftp/a.cram", + None, + &"a".repeat(32), + 100, + )], + ), + seq_lib( + "ERR2000011", + Some(1), + Some(1), + vec![seq_file( + "a_1.fastq.gz", + "FASTQ", + "ftp/a_1.fastq.gz", + None, + &"b".repeat(32), + 200, + )], + ), + ], + ) + .await; + + let got = du_db::grid::curation_candidates(&pool, true, 100) + .await + .unwrap(); + assert_eq!(got.len(), 1); + assert_eq!(got[0].data_kind, "CRAM"); + let formats: Vec<&str> = got[0] + .manifest + .as_array() + .unwrap() + .iter() + .map(|f| f["format"].as_str().unwrap()) + .collect(); + assert_eq!( + formats, + vec!["CRAM"], + "the FASTQ is excluded, not merely deprioritised" + ); + assert_eq!( + got[0].total_bytes, + Some(100), + "and it is not counted in the download budget either" + ); +} + +#[tokio::test] +async fn only_new_skips_samples_that_already_have_a_unit() { + let Some(url) = database_url() else { + return; + }; + let db = du_db::testing::ephemeral_db(&url) + .await + .expect("ephemeral db"); + let pool = db.pool().clone(); + + seed_sample( + &pool, + "SAMEA2000020", + vec![seq_lib( + "ERR2000020", + Some(1), + Some(1), + vec![seq_file( + "a.cram", + "CRAM", + "ftp/a.cram", + None, + &"a".repeat(32), + 100, + )], + )], + ) + .await; + + let first = du_db::grid::curation_candidates(&pool, true, 100) + .await + .unwrap(); + assert_eq!(first.len(), 1); + let unit = NewWorkUnit { + sample_accession: first[0].sample_accession.clone(), + study_accession: first[0].study_accession.clone(), + data_kind: first[0].data_kind.clone(), + manifest: first[0].manifest.clone(), + est_bases: first[0].est_bases, + total_bytes: first[0].total_bytes, + }; + let unit_id = grid::upsert_work_unit(&pool, &unit).await.unwrap(); + + // Incremental: nothing new to publish. + assert!(du_db::grid::curation_candidates(&pool, true, 100) + .await + .unwrap() + .is_empty()); + // Full re-projection still offers it, which is how a manifest gets refreshed after a re-crawl. + assert_eq!( + du_db::grid::curation_candidates(&pool, false, 100) + .await + .unwrap() + .len(), + 1 + ); + + // A refresh must not disturb the lifecycle. Canonicalise the unit, re-upsert, and check. + sqlx::query( + "UPDATE grid.work_unit SET state = 'CANONICAL', required_replicas = 5 WHERE id = $1", + ) + .bind(unit_id) + .execute(&pool) + .await + .unwrap(); + grid::upsert_work_unit(&pool, &unit).await.unwrap(); + let (state, reps): (String, i16) = + sqlx::query_as("SELECT state, required_replicas FROM grid.work_unit WHERE id = $1") + .bind(unit_id) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!( + state, "CANONICAL", + "curation describes the input; it must not un-canonicalise a unit" + ); + assert_eq!(reps, 5, "nor reset a replica count validation raised"); +} + +// ── validation ──────────────────────────────────────────────────────────────── + +/// Claim, then submit `y` as the Y call. Returns the submission id. +async fn submit_as(pool: &sqlx::PgPool, did: &str, unit_id: i64, y: &str, build: &str) -> i64 { + let claimed = grid::claim(pool, did, None, &kinds(&["CRAM"]), 1, 3600) + .await + .unwrap(); + let lease = claimed.first().map(|c| c.lease_id); + grid::submit( + pool, + did, + unit_id, + lease, + &json!({"calls": {"sex": "XY", "y_terminal": y, "coverage_mean": 30.0}}), + "sig", + "1.7.0", + build, + None, + &json!([]), + ) + .await + .unwrap() +} + +#[tokio::test] +async fn canonicalizing_marks_the_winners_agreed_and_the_rest_divergent() { + let Some(url) = database_url() else { + return; + }; + let db = du_db::testing::ephemeral_db(&url) + .await + .expect("ephemeral db"); + let pool = db.pool().clone(); + + let unit_id = grid::upsert_work_unit(&pool, &unit("SAMEA0000100", "CRAM")) + .await + .unwrap(); + // Three replicas so all three contributors can hold a slot at once. + sqlx::query("UPDATE grid.work_unit SET required_replicas = 3 WHERE id = $1") + .bind(unit_id) + .execute(&pool) + .await + .unwrap(); + + let a = submit_as(&pool, DID_A, unit_id, "R-A", "chm13v2.0").await; + let b = submit_as(&pool, DID_B, unit_id, "R-A", "chm13v2.0").await; + let c = submit_as(&pool, DID_C, unit_id, "R-WRONG", "chm13v2.0").await; + + let pending = grid::units_awaiting_validation(&pool, 10).await.unwrap(); + assert_eq!(pending.len(), 1, "the unit is waiting on validation"); + assert_eq!( + grid::submissions_for_validation(&pool, unit_id) + .await + .unwrap() + .len(), + 3 + ); + + let canonical = json!({"calls": {"sex": "XY", "y_terminal": "R-A", "coverage_mean": 30.0}}); + grid::canonicalize(&pool, unit_id, &canonical, &[a, b], &[c]) + .await + .unwrap(); + + let statuses: Vec<(i64, String)> = sqlx::query_as( + "SELECT id, status FROM grid.submission WHERE work_unit_id = $1 ORDER BY id", + ) + .bind(unit_id) + .fetch_all(&pool) + .await + .unwrap(); + assert_eq!( + statuses, + vec![ + (a, "AGREED".into()), + (b, "AGREED".into()), + (c, "DIVERGENT".into()) + ] + ); + + let (state, digest): (String, serde_json::Value) = + sqlx::query_as("SELECT state, canonical_digest FROM grid.work_unit WHERE id = $1") + .bind(unit_id) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(state, "CANONICAL"); + assert_eq!(digest["calls"]["y_terminal"], "R-A"); + + // A canonical unit is off the work list, and no longer awaiting validation. + assert!( + grid::claim(&pool, "did:plc:zzzz", None, &kinds(&["CRAM"]), 1, 3600) + .await + .unwrap() + .is_empty() + ); + assert!(grid::units_awaiting_validation(&pool, 10) + .await + .unwrap() + .is_empty()); + + // And the history that trust tiering reads reflects it. + assert_eq!(grid::grid_history(&pool, DID_A).await.unwrap().agreed, 1); + assert_eq!(grid::grid_history(&pool, DID_C).await.unwrap().divergent, 1); +} + +/// A contested unit must go back on the work list — otherwise a disagreement deadlocks the unit +/// forever, since the tie-breaker can never be claimed. +#[tokio::test] +async fn contesting_raises_the_bar_and_reopens_the_unit_for_a_tie_breaker() { + let Some(url) = database_url() else { + return; + }; + let db = du_db::testing::ephemeral_db(&url) + .await + .expect("ephemeral db"); + let pool = db.pool().clone(); + + let unit_id = grid::upsert_work_unit(&pool, &unit("SAMEA0000110", "CRAM")) + .await + .unwrap(); + submit_as(&pool, DID_A, unit_id, "R-A", "chm13v2.0").await; + submit_as(&pool, DID_B, unit_id, "R-B", "chm13v2.0").await; + + grid::contest(&pool, unit_id, "submissions disagree") + .await + .unwrap(); + + let (state, reps): (String, i16) = + sqlx::query_as("SELECT state, required_replicas FROM grid.work_unit WHERE id = $1") + .bind(unit_id) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(state, "CONTESTED"); + assert_eq!(reps, 3, "the bar rises by one"); + + // Two submissions exist and the bar is now three, so a third contributor can claim. + let c = grid::claim(&pool, DID_C, None, &kinds(&["CRAM"]), 1, 3600) + .await + .unwrap(); + assert_eq!(c.len(), 1, "a contested unit is claimable again"); + + // Nobody was blamed: with two conflicting answers there is no evidence about which is wrong. + let divergent: i64 = + sqlx::query_scalar("SELECT count(*) FROM grid.submission WHERE status = 'DIVERGENT'") + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(divergent, 0, "a coin-flip penalty would punish honest work"); +} + +#[tokio::test] +async fn the_shadow_spot_check_holds_a_unit_back_for_a_second_opinion() { + let Some(url) = database_url() else { + return; + }; + let db = du_db::testing::ephemeral_db(&url) + .await + .expect("ephemeral db"); + let pool = db.pool().clone(); + + let unit_id = grid::upsert_work_unit(&pool, &unit("SAMEA0000120", "CRAM")) + .await + .unwrap(); + sqlx::query("UPDATE grid.work_unit SET required_replicas = 1 WHERE id = $1") + .bind(unit_id) + .execute(&pool) + .await + .unwrap(); + + assert!( + !grid::maybe_request_shadow(&pool, unit_id, 0.0) + .await + .unwrap(), + "rate 0 never fires" + ); + assert!( + grid::maybe_request_shadow(&pool, unit_id, 1.0) + .await + .unwrap(), + "rate 1 always fires" + ); + + let (reps, note): (i16, Option) = + sqlx::query_as("SELECT required_replicas, note FROM grid.work_unit WHERE id = $1") + .bind(unit_id) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(reps, 2, "one lone result is no longer enough for this unit"); + assert_eq!(note.as_deref(), Some("shadow spot-check")); + + // The shadow arrives through the ordinary claim path — no special state, no second code path. + assert_eq!( + grid::claim(&pool, DID_A, None, &kinds(&["CRAM"]), 1, 3600) + .await + .unwrap() + .len(), + 1 + ); + + // A canonical unit is never held for a shadow: the check is for a result about to be trusted. + grid::canonicalize(&pool, unit_id, &json!({}), &[], &[]) + .await + .unwrap(); + assert!(!grid::maybe_request_shadow(&pool, unit_id, 1.0) + .await + .unwrap()); +} + +// ── the public + standing read paths ────────────────────────────────────────── + +#[tokio::test] +async fn the_public_result_page_shows_the_evidence_behind_a_digest() { + let Some(url) = database_url() else { + return; + }; + let db = du_db::testing::ephemeral_db(&url) + .await + .expect("ephemeral db"); + let pool = db.pool().clone(); + + let unit_id = grid::upsert_work_unit(&pool, &unit("SAMEA0000200", "CRAM")) + .await + .unwrap(); + + // Before a quorum: the unit is known but has no answer yet. "Working on it" and "never heard + // of it" must be distinguishable, so this is a row and not a 404. + let pending = grid::work_unit_public(&pool, "SAMEA0000200") + .await + .unwrap() + .unwrap(); + assert_eq!(pending.state, "AVAILABLE"); + assert!(pending.canonical_digest.is_none()); + assert_eq!(pending.replicas_agreed, 0); + assert!(grid::work_unit_public(&pool, "SAMEA_NOPE") + .await + .unwrap() + .is_none()); + + let a = submit_as(&pool, DID_A, unit_id, "R-A", "chm13v2.0").await; + let b = submit_as(&pool, DID_B, unit_id, "R-A", "chm13v2.0").await; + let canonical = json!({"calls": {"sex": "XY", "y_terminal": "R-A"}}); + grid::canonicalize(&pool, unit_id, &canonical, &[a, b], &[]) + .await + .unwrap(); + + let done = grid::work_unit_public(&pool, "SAMEA0000200") + .await + .unwrap() + .unwrap(); + assert_eq!(done.state, "CANONICAL"); + assert_eq!(done.canonical_digest.unwrap()["calls"]["y_terminal"], "R-A"); + assert_eq!( + done.replicas_agreed, 2, + "the replica count is the reason to believe the digest; publishing one without the other \ + would be publishing a claim with its evidence removed" + ); +} + +#[tokio::test] +async fn a_contributor_sees_its_own_leases_history_and_rank() { + let Some(url) = database_url() else { + return; + }; + let db = du_db::testing::ephemeral_db(&url) + .await + .expect("ephemeral db"); + let pool = db.pool().clone(); + + let u1 = grid::upsert_work_unit(&pool, &unit("SAMEA0000210", "CRAM")) + .await + .unwrap(); + let u2 = grid::upsert_work_unit(&pool, &unit("SAMEA0000211", "CRAM")) + .await + .unwrap(); + + // A finishes one unit and is credited; B is credited twice, so B outranks A. + let a = submit_as(&pool, DID_A, u1, "R-A", "chm13v2.0").await; + grid::canonicalize(&pool, u1, &json!({}), &[a], &[]) + .await + .unwrap(); + grid::award_credit(&pool, DID_A, u1, a, 3 * COBBLESTONE, "CANONICAL_FIRST") + .await + .unwrap(); + let b = submit_as(&pool, DID_B, u2, "R-A", "chm13v2.0").await; + grid::canonicalize(&pool, u2, &json!({}), &[b], &[]) + .await + .unwrap(); + grid::award_credit(&pool, DID_B, u2, b, 9 * COBBLESTONE, "CANONICAL_FIRST") + .await + .unwrap(); + + // C holds a live lease on a third unit and has been credited nothing. + let u3 = grid::upsert_work_unit(&pool, &unit("SAMEA0000212", "FASTQ")) + .await + .unwrap(); + grid::claim(&pool, DID_C, None, &kinds(&["FASTQ"]), 1, 3600) + .await + .unwrap(); + + let a_standing = grid::standing(&pool, DID_A).await.unwrap(); + assert_eq!(a_standing["agreed"], 1); + assert_eq!(a_standing["divergent"], 0); + assert_eq!( + a_standing["cobblestones"], 3.0, + "rendered from the ledger's thousandths" + ); + assert_eq!( + a_standing["rank"], 2, + "B has more cobblestones, so A is second" + ); + assert_eq!( + a_standing["leases"].as_array().unwrap().len(), + 0, + "A holds nothing now" + ); + + let c_standing = grid::standing(&pool, DID_C).await.unwrap(); + assert_eq!(c_standing["cobblestones"], 0.0); + assert!( + c_standing["rank"].is_null(), + "a contributor with no credit has no row on the board, so it has no rank — reporting the \ + position it would hold reads as \"you are last\" to someone who has not finished their \ + first unit yet" + ); + let leases = c_standing["leases"].as_array().unwrap(); + assert_eq!(leases.len(), 1, "the live lease is visible to its holder"); + assert_eq!(leases[0]["work_unit_id"], u3); + assert_eq!(leases[0]["data_kind"], "FASTQ"); +} + +#[tokio::test] +async fn stats_counts_units_leases_and_contributors() { + let Some(url) = database_url() else { + return; + }; + let db = du_db::testing::ephemeral_db(&url) + .await + .expect("ephemeral db"); + let pool = db.pool().clone(); + + let u1 = grid::upsert_work_unit(&pool, &unit("SAMEA0000220", "CRAM")) + .await + .unwrap(); + grid::upsert_work_unit(&pool, &unit("SAMEA0000221", "CRAM")) + .await + .unwrap(); + let a = submit_as(&pool, DID_A, u1, "R-A", "chm13v2.0").await; + let b = submit_as(&pool, DID_B, u1, "R-A", "chm13v2.0").await; + grid::canonicalize(&pool, u1, &json!({}), &[a, b], &[]) + .await + .unwrap(); + grid::award_credit(&pool, DID_A, u1, a, 4 * COBBLESTONE, "CANONICAL_FIRST") + .await + .unwrap(); + grid::claim(&pool, DID_C, None, &kinds(&["CRAM"]), 1, 3600) + .await + .unwrap(); + + let s = grid::stats(&pool).await.unwrap(); + assert_eq!(s["units_total"], 2); + assert_eq!(s["units_canonical"], 1); + assert_eq!(s["units_contested"], 0); + assert_eq!(s["leases_active"], 1, "only C still holds one"); + assert_eq!(s["contributors"], 2, "distinct submitters"); + assert_eq!(s["cobblestones_awarded"], 4.0); +} + +/// `est_bases` prefers the measured `base_count` and only falls back to `reads × read_length`, +/// which is a mean-length approximation and wrong outright for variable-length long reads. +#[tokio::test] +async fn est_bases_prefers_the_measured_count_over_the_approximation() { + let Some(url) = database_url() else { + return; + }; + let db = du_db::testing::ephemeral_db(&url) + .await + .expect("ephemeral db"); + let pool = db.pool().clone(); + + let f = || { + vec![seq_file( + "a.cram", + "CRAM", + "ftp/a.cram", + None, + &"a".repeat(32), + 100, + )] + }; + + // Both figures present and disagreeing: the measurement wins. + seed_sample( + &pool, + "SAMEA3000001", + vec![seq_lib_bases( + "ERR1", + Some(100), + Some(150), + Some(42_000), + f(), + )], + ) + .await; + // Only the approximation available — a row from before migration 0076. + seed_sample( + &pool, + "SAMEA3000002", + vec![seq_lib_bases("ERR2", Some(100), Some(150), None, f())], + ) + .await; + // Neither. The backfill has not reached it and ENA may never have published one. + seed_sample( + &pool, + "SAMEA3000003", + vec![seq_lib_bases("ERR3", Some(100), None, None, f())], + ) + .await; + + let got = du_db::grid::curation_candidates(&pool, true, 100) + .await + .unwrap(); + let by = |acc: &str| { + got.iter() + .find(|c| c.sample_accession == acc) + .unwrap() + .est_bases + }; + assert_eq!( + by("SAMEA3000001"), + Some(42_000), + "the measurement wins over reads × read_length" + ); + assert_eq!( + by("SAMEA3000002"), + Some(15_000), + "falling back where there is no measurement" + ); + assert_eq!( + by("SAMEA3000003"), + None, + "no honest figure available, so no figure — this sets what a contributor is paid, and a \ + fabricated number in a ledger is worse than an absent one" + ); +} + +/// The backfill only ever fills a NULL. Re-running it must not overwrite a measurement and +/// silently change what a contributor was already paid for. +#[tokio::test] +async fn the_base_count_backfill_never_overwrites_a_measurement() { + let Some(url) = database_url() else { + return; + }; + let db = du_db::testing::ephemeral_db(&url) + .await + .expect("ephemeral db"); + let pool = db.pool().clone(); + + let f = || { + vec![seq_file( + "a.cram", + "CRAM", + "ftp/a.cram", + None, + &"a".repeat(32), + 100, + )] + }; + seed_sample( + &pool, + "SAMEA3000010", + vec![seq_lib_bases("ERR10", Some(1), None, None, f())], + ) + .await; + seed_sample( + &pool, + "SAMEA3000011", + vec![seq_lib_bases("ERR11", Some(1), None, Some(999), f())], + ) + .await; + + let pending = du_db::sequence::runs_missing_base_count(&pool, 100) + .await + .unwrap(); + assert_eq!( + pending.len(), + 1, + "only the row without a measurement is work" + ); + assert_eq!( + pending[0].run_accession, "ERR10", + "the accession comes from the crawl's provenance slot" + ); + + assert!(du_db::sequence::set_base_count(&pool, pending[0].id, 5_000) + .await + .unwrap()); + assert!( + !du_db::sequence::set_base_count(&pool, pending[0].id, 7_000) + .await + .unwrap(), + "a second run finds nothing to fill" + ); + + let filled: Option = + sqlx::query_scalar("SELECT base_count FROM genomics.sequence_library WHERE id = $1") + .bind(pending[0].id) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(filled, Some(5_000), "the first measurement stands"); + assert!(du_db::sequence::runs_missing_base_count(&pool, 100) + .await + .unwrap() + .is_empty()); +} diff --git a/rust/crates/du-external/src/ena.rs b/rust/crates/du-external/src/ena.rs index ec36912..a4a00bc 100644 --- a/rust/crates/du-external/src/ena.rs +++ b/rust/crates/du-external/src/ena.rs @@ -12,7 +12,7 @@ const DEFAULT_BASE: &str = "https://www.ebi.ac.uk/ena/portal/api"; /// `link-ena-sequence-files.py`). Multi-file columns arrive `;`-joined. const RUN_FIELDS: &str = "run_accession,sample_accession,submitted_ftp,submitted_md5,\ submitted_bytes,submitted_format,fastq_ftp,fastq_md5,fastq_bytes,instrument_model,\ -library_layout,read_count,first_public"; +library_layout,read_count,base_count,first_public"; /// One ENA `read_run` row for a project — the raw fields as ENA returns them /// (`;`-joined for multi-file runs; the file-selection policy lives in the crawl job). @@ -30,6 +30,10 @@ pub struct EnaRunRow { pub instrument_model: String, pub library_layout: String, pub read_count: String, + /// Total bases in the run. The Grid's per-gigabase credit term weighs a work unit by this; it + /// is a measured total rather than `read_count × read_length`, which is a mean-length + /// approximation and simply wrong for variable-length long reads. + pub base_count: String, pub first_public: String, } @@ -77,7 +81,7 @@ fn parse_run_report(tsv: &str) -> Vec { }; let cols: Vec<&str> = header.split('\t').collect(); let col = |name: &str| cols.iter().position(|c| *c == name); - let (run, samp, sftp, smd5, sbytes, sfmt, fq, fqmd5, fqbytes, instr, layout, reads, pubd) = ( + let (run, samp, sftp, smd5, sbytes, sfmt, fq, fqmd5, fqbytes, instr, layout, reads, bases, pubd) = ( col("run_accession"), col("sample_accession"), col("submitted_ftp"), @@ -90,6 +94,7 @@ fn parse_run_report(tsv: &str) -> Vec { col("instrument_model"), col("library_layout"), col("read_count"), + col("base_count"), col("first_public"), ); let get = |f: &[&str], i: Option| i.and_then(|i| f.get(i)).map(|s| s.trim().to_string()).unwrap_or_default(); @@ -109,6 +114,7 @@ fn parse_run_report(tsv: &str) -> Vec { instrument_model: get(&f, instr), library_layout: get(&f, layout), read_count: get(&f, reads), + base_count: get(&f, bases), first_public: get(&f, pubd), } }) @@ -156,6 +162,10 @@ impl EnaClient { /// Enumerate every sequencing run for a study / project accession. Works for /// both ENA studies (`PRJEB…` → `ERR…`/`SAMEA…`) and NCBI BioProjects /// (`PRJNA…` → `SRR…`/`SAMN…`), since ENA mirrors the INSDC collaboration. + /// + /// A **run** accession (`ERR…`/`SRR…`) works here too and returns that single run — the + /// endpoint filters on whatever accession it is given. The `ena-base-count` backfill job + /// relies on that, rather than re-crawling whole studies to fill one column. /// Returns one row per run across all samples; an empty vec when the project /// has no reads (204/404 — e.g. genotype-only or unknown accession). pub async fn run_files(&self, accession: &str) -> Result, ExternalError> { diff --git a/rust/crates/du-jobs/src/crawl_project.rs b/rust/crates/du-jobs/src/crawl_project.rs index a2976a5..e662868 100644 --- a/rust/crates/du-jobs/src/crawl_project.rs +++ b/rust/crates/du-jobs/src/crawl_project.rs @@ -34,6 +34,13 @@ fn basename(url: &str) -> String { /// Parse a non-negative integer field; blanks / non-numeric → None (matches the /// python driver's `isdigit` guard). +/// +/// Shared with the `ena-base-count` backfill, which needs exactly this leniency: ENA leaves a +/// column empty when the submitter never supplied it, and that is not an error. +pub fn parse_count(v: &str) -> Option { + to_int(v) +} + fn to_int(v: &str) -> Option { let v = v.trim(); (!v.is_empty() && v.bytes().all(|b| b.is_ascii_digit())).then(|| v.parse().ok()).flatten() @@ -67,6 +74,7 @@ fn mk_lib(r: &EnaRunRow, files: Vec) -> NewSeqLibrary { instrument: opt(&r.instrument_model), reads: to_int(&r.read_count), read_length: None, + base_count: to_int(&r.base_count), paired_end: opt(&r.library_layout).map(|l| l.eq_ignore_ascii_case("PAIRED")), run_date: chrono::NaiveDate::parse_from_str(r.first_public.trim(), "%Y-%m-%d").ok(), external_run_ref: r.run_accession.trim().to_string(), diff --git a/rust/crates/du-jobs/src/ena.rs b/rust/crates/du-jobs/src/ena.rs index 3b0f24d..1251988 100644 --- a/rust/crates/du-jobs/src/ena.rs +++ b/rust/crates/du-jobs/src/ena.rs @@ -34,6 +34,71 @@ pub async fn enrich_studies(pool: &PgPool, client: &EnaClient) -> anyhow::Result Err(e) => tracing::warn!(accession = %c.accession, error = %e, "ena fetch failed"), } } - tracing::info!(candidates = candidates.len(), enriched, "ena-study-enrichment done"); + tracing::info!( + candidates = candidates.len(), + enriched, + "ena-study-enrichment done" + ); Ok(()) } + +/// Per-run batch cap for the base-count backfill. One ENA call per run, so this bounds both the +/// job's runtime and its politeness footprint; run it repeatedly until it reports zero. +const BACKFILL_BATCH: i64 = 200; + +/// Politeness gap between per-run ENA calls, matching the study crawl's. +const BACKFILL_GAP: std::time::Duration = std::time::Duration::from_millis(150); + +/// Backfill `genomics.sequence_library.base_count` from ENA for crawled runs that predate the +/// column (migration `0076`). +/// +/// A re-crawl cannot do this: `sequence::ingest_libraries` is idempotent at *sample* granularity +/// and skips a sample that already has files, which is the property that keeps re-crawls cheap and +/// which we do not want to weaken for one column. So this fills the column directly, one run at a +/// time — `filereport` filters on whatever accession it is given, so a run accession returns just +/// that run. +/// +/// Why it matters: the Grid pays per gigabase realigned (design §6.3), and a unit with no +/// `est_bases` pays the flat base rate — so until this has run, a 90 Gbp realignment earns what a +/// CRAM passthrough earns. +pub async fn backfill_base_counts(pool: &PgPool, client: &EnaClient) -> anyhow::Result { + let pending = du_db::sequence::runs_missing_base_count(pool, BACKFILL_BATCH).await?; + if pending.is_empty() { + tracing::debug!("ena-base-count: nothing to backfill"); + return Ok(0); + } + let (mut filled, mut absent) = (0usize, 0usize); + for row in &pending { + match client.run_files(&row.run_accession).await { + Ok(runs) => { + // ENA reports `base_count` as a possibly-empty string; an empty one means the + // submitter never supplied it, which is not an error and not something a retry + // will fix. Leave the NULL rather than writing a zero that would read as "this + // run sequenced nothing" and quietly pay a contributor for it. + let bases = runs + .iter() + .find(|r| r.run_accession.trim() == row.run_accession) + .and_then(|r| super::crawl_project::parse_count(&r.base_count)); + match bases { + Some(n) if n > 0 => { + if du_db::sequence::set_base_count(pool, row.id, n).await? { + filled += 1; + } + } + _ => absent += 1, + } + } + Err(e) => { + tracing::warn!(run = %row.run_accession, error = %e, "ena-base-count: lookup failed") + } + } + tokio::time::sleep(BACKFILL_GAP).await; + } + tracing::info!( + examined = pending.len(), + filled, + absent, + "ena-base-count: batch done (re-run until examined is 0)" + ); + Ok(filled) +} diff --git a/rust/crates/du-jobs/src/grid_curate.rs b/rust/crates/du-jobs/src/grid_curate.rs new file mode 100644 index 0000000..d00d0f7 --- /dev/null +++ b/rust/crates/du-jobs/src/grid_curate.rs @@ -0,0 +1,94 @@ +//! Grid curation: publish the claimable work list. +//! +//! This job turns samples the ENA crawl already resolved into `grid.work_unit` rows. It makes **no +//! network calls at all** — `crawl_project` did that, and the file URLs, md5s and sizes are already +//! in `genomics.sequence_file`. Curating centrally is also what keeps a volunteer fleet off the ENA +//! portal: a node is handed a finished manifest and never goes discovering files for itself. +//! +//! Design: `documents/design/distributed-compute-grid.md` §4.5 in the DUNavigator repo. +//! +//! Idempotent. A refresh updates a unit's manifest and sizes but never its `state`, +//! `required_replicas` or canonical digest — curation describes the *input*, validation owns the +//! *lifecycle*, and a re-crawl must not un-canonicalise a finished unit. + +use du_db::grid::{self, NewWorkUnit}; +use du_db::PgPool; + +/// Per-run cap. Curation is cheap (one query, then one upsert per sample), but a bounded batch +/// keeps a first run over a large catalogue from holding the job lock for an unbounded time. +const BATCH: i64 = 500; + +/// What one curation pass did. +pub struct CurateOutcome { + pub examined: usize, + pub published: usize, + /// Units that carry no `est_bases`. The per-Gbp credit term has nothing to weigh these by; + /// see [`grid::curation_candidates`] for why, and `documents/…/distributed-compute-grid.md` + /// §12.4 for the fix. + pub without_est_bases: usize, + pub cram: usize, + pub fastq: usize, +} + +/// Publish work units for samples that do not have one yet. +/// +/// `only_new = false` re-projects every eligible sample, which refreshes manifests after a +/// re-crawl. That is the ops path; the timer runs the incremental one. +pub async fn curate(pool: &PgPool, only_new: bool) -> anyhow::Result { + let candidates = grid::curation_candidates(pool, only_new, BATCH).await?; + let mut out = CurateOutcome { + examined: candidates.len(), + published: 0, + without_est_bases: 0, + cram: 0, + fastq: 0, + }; + if candidates.is_empty() { + tracing::debug!("grid-curate: nothing to publish"); + return Ok(out); + } + + for c in &candidates { + if c.est_bases.is_none() { + out.without_est_bases += 1; + } + match c.data_kind.as_str() { + "CRAM" => out.cram += 1, + _ => out.fastq += 1, + } + let unit = NewWorkUnit { + sample_accession: c.sample_accession.clone(), + study_accession: c.study_accession.clone(), + data_kind: c.data_kind.clone(), + manifest: c.manifest.clone(), + est_bases: c.est_bases, + total_bytes: c.total_bytes, + }; + // One bad sample must not sink the batch: the catalogue is a best-effort projection, and a + // sample that fails to publish is simply picked up by the next run. + match grid::upsert_work_unit(pool, &unit).await { + Ok(_) => out.published += 1, + Err(e) => { + tracing::warn!(sample = %c.sample_accession, error = %e, "grid-curate: upsert failed") + } + } + } + + tracing::info!( + examined = out.examined, + published = out.published, + cram = out.cram, + fastq = out.fastq, + without_est_bases = out.without_est_bases, + "grid-curate: done" + ); + // Say the quiet part out loud rather than leaving a silent hole in the credit formula. + if out.without_est_bases > 0 { + tracing::warn!( + count = out.without_est_bases, + "grid-curate: units published with no est_bases — the per-Gbp credit term cannot weigh \ + them. ENA exposes base_count; du-external's RUN_FIELDS does not request it." + ); + } + Ok(out) +} diff --git a/rust/crates/du-jobs/src/grid_validate.rs b/rust/crates/du-jobs/src/grid_validate.rs new file mode 100644 index 0000000..894cfe9 --- /dev/null +++ b/rust/crates/du-jobs/src/grid_validate.rs @@ -0,0 +1,348 @@ +//! Grid validation — adaptive replication (design §6.1), as a `run-once` job. +//! +//! Per unit with unvalidated submissions: cluster the digests by agreement, and if one cluster +//! satisfies the unit's replica bar *and* the trust policy, promote the unit to `CANONICAL` and pay +//! the contributors who agreed. +//! +//! # Why trust is derived from grid work only +//! +//! Tiers come from a contributor's `grid.submission` history, never from the social reputation +//! score. Those are different claims: social standing says a person participates well in the +//! community, and grid trust has to say their *machine produces correct results*. Letting the first +//! stand in for the second would let a well-regarded member canonicalize bad output on reputation +//! alone, which is precisely the attack adaptive replication exists to stop. +//! +//! # The constants are placeholders and are meant to be +//! +//! Design §9 asks for conservative placeholders until real throughput data exists. `AGREED_FOR_*` +//! and the credit weights are exactly that. They are deliberately strict: too-slow promotion costs +//! duplicated compute, while too-fast promotion costs a wrong canonical result, and only one of +//! those is recoverable. + +use du_db::grid::{self, digest, GridHistory, PendingSubmission}; +use du_db::PgPool; +use std::collections::HashSet; + +/// Units examined per run. +const BATCH: i64 = 200; + +/// Agreed units needed to leave `Untrusted`. +const AGREED_FOR_PROVISIONAL: i64 = 5; +/// Agreed units needed to reach `Trusted`, where one submission can canonicalize alone. +const AGREED_FOR_TRUSTED: i64 = 25; +/// A contributor with any divergence in its history cannot be `Trusted`. Blunt on purpose: with no +/// real data yet, the safe reading of a divergence is the pessimistic one. +const MAX_DIVERGENCE_FOR_TRUSTED: i64 = 0; + +/// Fraction of would-be lone canonicalizations that are held for a shadow replica instead. +pub const SPOT_CHECK_RATE: f64 = 0.05; + +/// Cobblestones (in milli-units) for running the stack on one unit, whatever its size. +const BASE_CREDIT: i64 = 2 * grid::COBBLESTONE; +/// Additional cobblestones per gigabase realigned. Only a FASTQ unit earns this. +const REALIGN_PER_GBP: i64 = grid::COBBLESTONE / 2; + +/// What a contributor's history has earned it. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Tier { + /// New, or still proving itself. Can contribute to a quorum but never satisfy one alone. + Untrusted, + /// Established. Counts toward quorum and may pair with an untrusted node to reach it. + Provisional, + /// Sustained agreement. One submission canonicalizes, subject to a random shadow re-check. + Trusted, +} + +fn tier_of(h: GridHistory) -> Tier { + if h.agreed >= AGREED_FOR_TRUSTED && h.divergent <= MAX_DIVERGENCE_FOR_TRUSTED { + Tier::Trusted + } else if h.agreed >= AGREED_FOR_PROVISIONAL { + Tier::Provisional + } else { + Tier::Untrusted + } +} + +/// Partition submissions into clusters that agree, keeping only mutually comparable ones together. +/// +/// Comparability comes first: results against different references, or different stack majors, are +/// answers to different questions rather than a disagreement, and clustering them together would +/// manufacture divergence out of nothing. +fn cluster(subs: &[PendingSubmission]) -> Vec> { + let mut clusters: Vec> = Vec::new(); + 'next: for (i, s) in subs.iter().enumerate() { + for c in clusters.iter_mut() { + let head = &subs[c[0]]; + if digest::comparable( + &head.reference_build, + &head.stack_version, + &s.reference_build, + &s.stack_version, + ) && digest::agree(&head.digest, &s.digest) + { + c.push(i); + continue 'next; + } + } + clusters.push(vec![i]); + } + // Biggest first, and among equals the one that got there first — so `CANONICAL_FIRST` goes to + // whoever actually arrived first rather than to an accident of iteration order. + clusters.sort_by(|a, b| b.len().cmp(&a.len()).then(a[0].cmp(&b[0]))); + clusters +} + +/// Whether `cluster` may canonicalize the unit, given who is in it. +/// +/// A lone submission canonicalizes only from a `Trusted` contributor. Two or more agreeing from +/// **distinct** contributors always suffice: independent agreement is the evidence, and requiring +/// a trusted node on top of it would stall a young fleet where nobody is trusted yet. +fn may_canonicalize(tiers: &[Tier], required: i16) -> bool { + match tiers.len() { + 0 => false, + 1 => tiers[0] == Tier::Trusted, + n => n >= required.max(2) as usize, + } +} + +/// Credit for one agreed unit, in milli-cobblestones. Passthrough units earn the base only; a +/// realigned unit earns per gigabase on top, which is where the real compute went. +fn credit_for(data_kind: &str, est_bases: Option) -> i64 { + if data_kind == "FASTQ" { + let gbp = est_bases.unwrap_or(0) as f64 / 1e9; + BASE_CREDIT + (gbp * REALIGN_PER_GBP as f64).round() as i64 + } else { + BASE_CREDIT + } +} + +/// Outcome of one validation pass. +#[derive(Debug, Default)] +pub struct ValidateOutcome { + pub examined: usize, + pub canonicalized: usize, + pub contested: usize, + pub shadow_requested: usize, + pub credited: usize, +} + +pub async fn validate(pool: &PgPool, spot_check_rate: f64) -> anyhow::Result { + let units = grid::units_awaiting_validation(pool, BATCH).await?; + let mut out = ValidateOutcome { + examined: units.len(), + ..Default::default() + }; + + for unit in &units { + let subs = grid::submissions_for_validation(pool, unit.id).await?; + if subs.is_empty() { + continue; + } + let clusters = cluster(&subs); + let winner = &clusters[0]; + + // Distinct DIDs, not distinct rows. The unique index already makes them the same thing; + // relying on it silently would leave this correct only by coincidence. + let dids: HashSet<&str> = winner.iter().map(|&i| subs[i].did.as_str()).collect(); + let mut tiers = Vec::with_capacity(dids.len()); + for did in &dids { + tiers.push(tier_of(grid::grid_history(pool, did).await?)); + } + + if !may_canonicalize(&tiers, unit.required_replicas) { + // Several mutually contradictory answers and nothing decisive: raise the bar and let + // the next replica break the tie. With two conflicting clusters there is no evidence + // about which is wrong, so nobody is marked divergent yet. + if clusters.len() > 1 { + grid::contest( + pool, + unit.id, + "submissions disagree; awaiting a tie-breaker", + ) + .await?; + out.contested += 1; + } + continue; + } + + // A trusted node about to canonicalize alone is sometimes held for a shadow replica, so + // that trust is re-earned rather than assumed indefinitely. The draw is made in the + // database (see `maybe_request_shadow`) because a rule derived from the unit id or the + // digest would be one a contributor could compute in advance and cheat around — which is + // exactly what §6.2 says a spot-check must not be. + if tiers.len() == 1 && grid::maybe_request_shadow(pool, unit.id, spot_check_rate).await? { + out.shadow_requested += 1; + continue; + } + + let agreed_ids: Vec = winner.iter().map(|&i| subs[i].id).collect(); + let divergent_ids: Vec = clusters[1..] + .iter() + .flatten() + .map(|&i| subs[i].id) + .collect(); + let canonical = subs[winner[0]].digest.clone(); + grid::canonicalize(pool, unit.id, &canonical, &agreed_ids, &divergent_ids).await?; + out.canonicalized += 1; + + let amount = credit_for(&unit.data_kind, unit.est_bases); + for (rank, &i) in winner.iter().enumerate() { + let kind = if rank == 0 { + "CANONICAL_FIRST" + } else { + "QUORUM_AGREE" + }; + match grid::award_credit(pool, &subs[i].did, unit.id, subs[i].id, amount, kind).await { + Ok(true) => out.credited += 1, + Ok(false) => {} // already paid for this unit; the ledger's unique index held + Err(e) => { + tracing::warn!(did = %subs[i].did, unit = unit.id, error = %e, "grid-validate: credit failed") + } + } + } + tracing::debug!(unit = %unit.sample_accession, replicas = winner.len(), "grid-validate: canonical"); + } + + tracing::info!( + examined = out.examined, + canonicalized = out.canonicalized, + contested = out.contested, + shadow_requested = out.shadow_requested, + credited = out.credited, + "grid-validate: done" + ); + Ok(out) +} + +/// Close every lapsed lease. See `du_db::grid::reap_expired` for what this is and is not for — in +/// particular, it is **not** what returns a unit to the claimable pool. +pub async fn reap(pool: &PgPool) -> anyhow::Result { + let n = grid::reap_expired(pool).await?; + if n > 0 { + tracing::info!(closed = n, "grid-reap: lapsed leases closed"); + } + Ok(n) +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + fn sub(id: i64, did: &str, y: &str, build: &str, stack: &str) -> PendingSubmission { + PendingSubmission { + id, + did: did.into(), + digest: json!({"calls": {"sex": "XY", "y_terminal": y, "coverage_mean": 30.0}}), + stack_version: stack.into(), + reference_build: build.into(), + submitted_at: chrono::DateTime::from_timestamp(1_724_500_000 + id, 0).unwrap(), + } + } + + #[test] + fn agreeing_submissions_form_one_cluster() { + let subs = vec![ + sub(1, "did:a", "R-A", "chm13v2.0", "1.7.0"), + sub(2, "did:b", "R-A", "chm13v2.0", "1.8.1"), + ]; + assert_eq!( + cluster(&subs), + vec![vec![0, 1]], + "a minor version bump does not split a quorum" + ); + } + + #[test] + fn disagreeing_submissions_split_and_the_larger_cluster_leads() { + let subs = vec![ + sub(1, "did:a", "R-WRONG", "chm13v2.0", "1.7.0"), + sub(2, "did:b", "R-A", "chm13v2.0", "1.7.0"), + sub(3, "did:c", "R-A", "chm13v2.0", "1.7.0"), + ]; + let c = cluster(&subs); + assert_eq!(c[0], vec![1, 2], "the majority answer leads"); + assert_eq!(c[1], vec![0]); + } + + /// Different references are not a disagreement — they are different questions. + #[test] + fn incomparable_submissions_never_share_a_cluster() { + let subs = vec![ + sub(1, "did:a", "R-A", "chm13v2.0", "1.7.0"), + sub(2, "did:b", "R-A", "GRCh38", "1.7.0"), + sub(3, "did:c", "R-A", "chm13v2.0", "2.0.0"), + ]; + assert_eq!( + cluster(&subs).len(), + 3, + "same call, three incompatible contexts" + ); + } + + #[test] + fn only_a_trusted_contributor_canonicalizes_alone() { + assert!(may_canonicalize(&[Tier::Trusted], 2)); + assert!(!may_canonicalize(&[Tier::Provisional], 2)); + assert!(!may_canonicalize(&[Tier::Untrusted], 2)); + assert!(!may_canonicalize(&[], 2)); + } + + /// Two independent untrusted contributors agreeing is evidence, and must be enough — otherwise + /// a fleet where nobody is trusted yet can never canonicalize anything at all. + #[test] + fn two_independent_contributors_suffice_whatever_their_tier() { + assert!(may_canonicalize(&[Tier::Untrusted, Tier::Untrusted], 2)); + assert!( + !may_canonicalize(&[Tier::Untrusted, Tier::Untrusted], 3), + "a contested unit needs more" + ); + } + + #[test] + fn tiers_come_from_agreement_and_any_divergence_blocks_trust() { + assert_eq!( + tier_of(GridHistory { + agreed: 0, + divergent: 0 + }), + Tier::Untrusted + ); + assert_eq!( + tier_of(GridHistory { + agreed: 5, + divergent: 0 + }), + Tier::Provisional + ); + assert_eq!( + tier_of(GridHistory { + agreed: 25, + divergent: 0 + }), + Tier::Trusted + ); + assert_eq!( + tier_of(GridHistory { + agreed: 100, + divergent: 1 + }), + Tier::Provisional, + "one bad result costs trust, however much good work surrounds it" + ); + } + + #[test] + fn only_a_realigned_unit_earns_per_gigabase() { + assert_eq!(credit_for("CRAM", Some(90_000_000_000)), BASE_CREDIT); + assert_eq!( + credit_for("FASTQ", Some(90_000_000_000)), + BASE_CREDIT + 90 * REALIGN_PER_GBP + ); + assert_eq!( + credit_for("FASTQ", None), + BASE_CREDIT, + "an unknown size pays the base rather than nothing — the work was still done" + ); + } +} diff --git a/rust/crates/du-jobs/src/main.rs b/rust/crates/du-jobs/src/main.rs index 8d32da0..c956693 100644 --- a/rust/crates/du-jobs/src/main.rs +++ b/rust/crates/du-jobs/src/main.rs @@ -9,6 +9,8 @@ mod coord_lift; mod crawl_project; mod ena; +mod grid_curate; +mod grid_validate; mod faidx; mod ftdna_str; mod gzio; @@ -372,6 +374,36 @@ async fn main() -> anyhow::Result<()> { None => crawl_project::crawl_pending(&pool, &ena).await?, } } + // Backfill `sequence_library.base_count` from ENA for runs that predate migration + // 0076. A re-crawl cannot do it — `ingest_libraries` skips samples that already have + // files — and until it has run, the Grid pays a 90 Gbp realignment what it pays a + // CRAM passthrough. Bounded batch; re-run until it reports nothing examined. + "ena-base-count" => { + let client = du_external::ena::EnaClient::new(); + ena::backfill_base_counts(&pool, &client).await?; + } + // Grid curation: project the samples `crawl-project` already resolved into the + // claimable work list (`grid.work_unit`). Makes no network calls — the file URLs, + // md5s and sizes are already in `genomics.sequence_file`, and curating centrally is + // what keeps a volunteer fleet off the ENA portal. `all` re-projects every eligible + // sample to refresh manifests after a re-crawl; the bare form is incremental and is + // what the timer runs. + "grid-curate" => { + let only_new = argv.next().as_deref() != Some("all"); + grid_curate::curate(&pool, only_new).await?; + } + // Close lapsed leases. NOT what returns a unit to the claimable pool — `claim` + // already ignores any lease past its bound, so the catalogue keeps flowing even while + // this is down. What it does is record the EXPIRED outcome that trust tiering needs, + // and let a node that overran take a fresh lease. Design §12.5. + "grid-reap" => { + grid_validate::reap(&pool).await?; + } + // Adaptive replication: cluster the submitted digests, canonicalize a unit when one + // cluster satisfies both the replica bar and the trust policy, and pay whoever agreed. + "grid-validate" => { + grid_validate::validate(&pool, grid_validate::SPOT_CHECK_RATE).await?; + } // External enrichment (formerly scheduled; now nightly run-once). OpenAlex // by-DOI refresh, date-sorted discovery, and PubMed by-PMID gap-fill. "publication-update" => { diff --git a/rust/crates/du-web/src/routes/grid_edge.rs b/rust/crates/du-web/src/routes/grid_edge.rs new file mode 100644 index 0000000..c661000 --- /dev/null +++ b/rust/crates/du-web/src/routes/grid_edge.rs @@ -0,0 +1,365 @@ +//! Signed Edge API for the DecodingUs Grid (`/api/v1/grid/*`) — design §4.4. +//! +//! Volunteer Navigator instances register, lease public-ENA work units, report progress, and submit +//! signed result digests. Authentication is the same Ed25519 device-key path the D1 exchange and the +//! recruitment Edge already use: `verify_signed_fresh` for anything that mutates, `verify_signed` +//! plus `ensure_fresh_ts` for reads. The canonical signed strings live in `du_db::grid::messages` +//! and are a cross-repo contract with Navigator. +//! +//! PII-free throughout: DIDs, signatures, ENA accessions and computed calls. Nothing here touches a +//! donor. +//! +//! # Two deviations from §4.4, both to avoid a second way to do one thing +//! +//! **No `/grid/node/heartbeat`.** The table §4.4 gives it lists `POST /grid/node/register` as an +//! upsert of `fed.pds_node` and `/grid/node/heartbeat` as node liveness — but `register_node` +//! already sets `last_heartbeat` and is idempotent, so re-registering *is* the node heartbeat. Two +//! endpoints writing the same row is a drift waiting to happen. +//! +//! **`/grid/heartbeat` does not extend the lease.** §4.4 offers an "optional TTL extension"; a node +//! that can heartbeat but never finish would then hold a unit forever, which is the exact failure a +//! bounded lease exists to prevent. See design §4.3 and §12.5. + +use crate::error::AppError; +use crate::sig::{ensure_fresh_ts, verify_signed, verify_signed_fresh}; +use crate::state::AppState; +use axum::extract::{Path, Query, State}; +use axum::routing::{get, post}; +use axum::{Json, Router}; +use du_db::grid::{self, digest, messages}; +use serde::Deserialize; +use serde_json::{json, Value}; + +/// Upper bound on units handed out in one call, whatever the node asks for. A node that wants more +/// comes back; a node that asks for ten thousand does not get to empty the catalogue. +const MAX_CLAIM: i64 = 32; +/// Lease bounds in seconds — 1 hour to 14 days (§4.3, "AppView clamps"). +const MIN_LEASE_SECS: i64 = 3_600; +const MAX_LEASE_SECS: i64 = 14 * 24 * 3_600; + +const DEFAULT_LEADERBOARD_LIMIT: i64 = 100; +const MAX_LEADERBOARD_LIMIT: i64 = 500; + +pub fn router() -> Router { + Router::new() + .route("/api/v1/grid/node/register", post(node_register)) + .route("/api/v1/grid/claim", post(claim)) + .route("/api/v1/grid/heartbeat", post(heartbeat)) + .route("/api/v1/grid/release", post(release)) + .route("/api/v1/grid/submit", post(submit)) + .route("/api/v1/grid/mine", get(mine)) + .route("/api/v1/grid/leaderboard", get(leaderboard)) + .route("/api/v1/grid/work/:accession", get(work_unit)) + .route("/api/v1/grid/stats", get(stats)) +} + +// ── signed: node lifecycle ──────────────────────────────────────────────────── + +#[derive(Deserialize)] +struct RegisterBody { + did: String, + software_version: String, + /// Free-form: data kinds, threads, disk budget, OS/arch. + capabilities: Value, + os_info: Option, + ts: i64, + signature: String, +} + +/// Register a node, or refresh what it advertises. Doubles as the node-level heartbeat. +/// +/// The capabilities hash is inside the signed message, so a node cannot have capabilities +/// attributed to it that it did not send — which matters because `claim` filters on them, and a +/// forged claim of FASTQ capability would hand a node work it cannot do. +async fn node_register( + State(st): State, + Json(b): Json, +) -> Result, AppError> { + let caps_hash = digest::canonical_sha256_b64(&b.capabilities); + verify_signed_fresh( + &st.pool, + &b.did, + b.ts, + &messages::register(&b.did, &b.software_version, &caps_hash), + &b.signature, + ) + .await?; + let id = grid::register_node( + &st.pool, + &b.did, + &b.software_version, + &b.capabilities, + b.os_info.as_deref(), + ) + .await?; + Ok(Json(json!({ "node_id": id }))) +} + +// ── signed: the work loop ───────────────────────────────────────────────────── + +#[derive(Deserialize)] +struct ClaimBody { + did: String, + /// Data kinds this node can actually process, e.g. `["CRAM","FASTQ"]`. + data_kinds: Vec, + count: i64, + lease_secs: i64, + ts: i64, + signature: String, +} + +/// Lease up to `count` units the node can handle. +/// +/// `data_kinds` is normalised — uppercased, deduplicated, sorted — *before* the signature is +/// checked, and the signed message covers the normalised form. Otherwise `["CRAM","cram"]` and +/// `["cram","CRAM"]` would be different signed strings for the same request, and a node whose +/// ordering differed from ours would get an unexplained 403. +async fn claim( + State(st): State, + Json(b): Json, +) -> Result, AppError> { + let mut kinds: Vec = b + .data_kinds + .iter() + .map(|k| k.trim().to_ascii_uppercase()) + .collect(); + kinds.sort(); + kinds.dedup(); + if kinds.is_empty() { + return Err(AppError::BadRequest("no data kinds advertised".into())); + } + let count = b.count.clamp(1, MAX_CLAIM); + let lease = b.lease_secs.clamp(MIN_LEASE_SECS, MAX_LEASE_SECS); + + // Sign what the node asked for, not what we clamped it to: the node cannot know our bounds, and + // making it guess them to produce a valid signature would be an unusable API. + verify_signed_fresh( + &st.pool, + &b.did, + b.ts, + &messages::claim(&b.did, &kinds.join(","), b.count as i32, b.lease_secs), + &b.signature, + ) + .await?; + + let node_id = grid::node_id_for_did(&st.pool, &b.did).await?; + let units = grid::claim(&st.pool, &b.did, node_id, &kinds, count, lease).await?; + Ok(Json(json!({ "units": units }))) +} + +#[derive(Deserialize)] +struct HeartbeatBody { + did: String, + lease_id: i64, + stage: String, + progress: Option, + ts: i64, + signature: String, +} + +/// Report progress on a held lease. Does **not** extend it (see the module note). +/// +/// `held: false` tells a node it no longer owns the lease — expired, released, or never its own — +/// so it can stop work rather than spend hours finishing a unit it will not be credited for. +async fn heartbeat( + State(st): State, + Json(b): Json, +) -> Result, AppError> { + verify_signed_fresh( + &st.pool, + &b.did, + b.ts, + &messages::heartbeat(&b.did, b.lease_id, &b.stage), + &b.signature, + ) + .await?; + let held = grid::heartbeat(&st.pool, &b.did, b.lease_id, b.progress.as_ref()).await?; + Ok(Json(json!({ "held": held }))) +} + +#[derive(Deserialize)] +struct ReleaseBody { + did: String, + lease_id: i64, + reason: String, + ts: i64, + signature: String, +} + +/// Give a lease back without a result, so the unit recycles immediately instead of waiting out its +/// bound. Idempotent: releasing an already-closed lease reports `released: false` and is not an +/// error, because a node retrying after a dropped response has done nothing wrong. +async fn release( + State(st): State, + Json(b): Json, +) -> Result, AppError> { + verify_signed_fresh( + &st.pool, + &b.did, + b.ts, + &messages::release(&b.did, b.lease_id, &b.reason), + &b.signature, + ) + .await?; + let released = grid::release(&st.pool, &b.did, b.lease_id, "RELEASED").await?; + Ok(Json(json!({ "released": released }))) +} + +#[derive(Deserialize)] +struct SubmitBody { + did: String, + work_unit_id: i64, + lease_id: Option, + /// The result digest (§5.2), carrying **raw** values — the AppView buckets at comparison time. + digest: Value, + /// The node's own Ed25519 signature over its digest, stored for later audit. + digest_sig: String, + stack_version: String, + reference_build: String, + aligner: Option, + /// `at://` URIs of the fed records the node published. + record_refs: Value, + ts: i64, + signature: String, +} + +/// Record a signed result and close the lease that produced it. +/// +/// The request signature covers the **hash of the digest**, and that hash is recomputed here from +/// what actually arrived. Without that check a node could sign the hash of a good result and post a +/// different one, and the stored `digest_sig` would still look valid to a later auditor. +async fn submit( + State(st): State, + Json(b): Json, +) -> Result, AppError> { + let hash = digest::canonical_sha256_b64(&b.digest); + verify_signed_fresh( + &st.pool, + &b.did, + b.ts, + &messages::submit(&b.did, b.work_unit_id, &hash), + &b.signature, + ) + .await?; + + let id = grid::submit( + &st.pool, + &b.did, + b.work_unit_id, + b.lease_id, + &b.digest, + &b.digest_sig, + &b.stack_version, + &b.reference_build, + b.aligner.as_deref(), + &b.record_refs, + ) + .await?; + Ok(Json(json!({ "submission_id": id }))) +} + +#[derive(Deserialize)] +struct PollQuery { + did: String, + ts: i64, + sig: String, +} + +/// The caller's own grid standing: leases held right now, agreed/divergent history, cobblestones +/// and board rank. This is what §7.1's Grid panel renders. +/// +/// The one **signed read** in this API — everything else either mutates or is public. It is +/// authenticated because it is the caller's own work: the leaderboard publishes totals, and this +/// publishes the rows behind one contributor's total, which is theirs to see and nobody else's. +async fn mine( + State(st): State, + Query(q): Query, +) -> Result, AppError> { + ensure_fresh_ts(q.ts)?; + verify_signed(&st.pool, &q.did, &messages::poll(&q.did, q.ts), &q.sig).await?; + Ok(Json(grid::standing(&st.pool, &q.did).await?)) +} + +// ── public ──────────────────────────────────────────────────────────────────── + +#[derive(Deserialize)] +struct LeaderboardQuery { + /// `30` for the rolling window; omitted for all-time. + days: Option, + limit: Option, +} + +/// Ranked contributors (§6.4). Public and unauthenticated — a leaderboard nobody can read is not a +/// leaderboard. +/// +/// Cobblestones are rendered from the ledger's integer thousandths here, at the last possible +/// moment, so the sum stays exact all the way up (§12.2). +async fn leaderboard( + State(st): State, + Query(q): Query, +) -> Result, AppError> { + let limit = q + .limit + .unwrap_or(DEFAULT_LEADERBOARD_LIMIT) + .clamp(1, MAX_LEADERBOARD_LIMIT); + let rows = grid::leaderboard(&st.pool, q.days, limit).await?; + let items: Vec = rows + .into_iter() + .map(|r| { + json!({ + "handle": r.handle, + "did": r.did, + "cobblestones": r.cobblestones_milli as f64 / grid::COBBLESTONE as f64, + "units": r.units, + }) + }) + .collect(); + Ok(Json(json!({ "items": items }))) +} + +/// The canonical community result for one ENA sample, once a quorum has agreed on it. +/// +/// A unit that exists but has not reached quorum returns its state and no digest, rather than a +/// 404: "we are working on it" and "we have never heard of it" are different answers, and a +/// consumer needs to tell them apart. +async fn work_unit( + State(st): State, + Path(accession): Path, +) -> Result, AppError> { + let row = grid::work_unit_public(&st.pool, accession.trim()) + .await? + .ok_or_else(|| AppError::NotFound(format!("no grid work unit for {accession}")))?; + Ok(Json(json!({ + "sample_accession": row.sample_accession, + "study_accession": row.study_accession, + "data_kind": row.data_kind, + "state": row.state, + "canonical_digest": row.canonical_digest, + "canonical_at": row.canonical_at, + "replicas_agreed": row.replicas_agreed, + }))) +} + +/// Grid throughput: units by state, active leases, contributing nodes. +async fn stats(State(st): State) -> Result, AppError> { + Ok(Json(grid::stats(&st.pool).await?)) +} + +#[cfg(test)] +mod tests { + /// The claim endpoint normalises `data_kinds` before signing, so that the same request written + /// two ways produces the same signed bytes. This mirrors the handler's normalisation; if one + /// changes without the other, honest nodes start getting 403s. + #[test] + fn data_kinds_normalise_to_one_signed_form() { + let norm = |v: &[&str]| { + let mut k: Vec = v.iter().map(|s| s.trim().to_ascii_uppercase()).collect(); + k.sort(); + k.dedup(); + k.join(",") + }; + assert_eq!(norm(&["FASTQ", "CRAM"]), "CRAM,FASTQ"); + assert_eq!(norm(&["cram", " CRAM ", "FASTQ"]), "CRAM,FASTQ"); + assert_eq!(norm(&["CRAM"]), "CRAM"); + } +} diff --git a/rust/crates/du-web/src/routes/mod.rs b/rust/crates/du-web/src/routes/mod.rs index 49c03b6..dc1db15 100644 --- a/rust/crates/du-web/src/routes/mod.rs +++ b/rust/crates/du-web/src/routes/mod.rs @@ -19,6 +19,7 @@ pub mod auth_routes; pub mod change_sets; pub mod coverage; pub mod exchange; +pub mod grid_edge; pub mod ibd; pub mod curation; pub mod curator; @@ -89,6 +90,7 @@ pub fn app(state: AppState) -> Router { .merge(projects::router()) .merge(recruitment::router()) .merge(recruitment_edge::router()) + .merge(grid_edge::router()) .merge(exchange::router()) .merge(ibd::router()) .merge(research::router()) diff --git a/rust/crates/du-web/src/routes/samples.rs b/rust/crates/du-web/src/routes/samples.rs index 21a699e..3cc6491 100644 --- a/rust/crates/du-web/src/routes/samples.rs +++ b/rust/crates/du-web/src/routes/samples.rs @@ -602,6 +602,10 @@ struct SeqLibraryIn { instrument: Option, reads: Option, read_length: Option, + /// Total bases in the run. Optional: this ops endpoint predates the column, and a caller that + /// does not know the figure should send nothing rather than a guess — the Grid's per-gigabase + /// credit reads it, and an invented number there pays someone the wrong amount. + base_count: Option, paired_end: Option, /// ISO date (`YYYY-MM-DD`); tolerated absent. run_date: Option, @@ -634,6 +638,7 @@ async fn ingest_sequence_libraries( instrument: l.instrument, reads: l.reads, read_length: l.read_length, + base_count: l.base_count, paired_end: l.paired_end, run_date: l.run_date, external_run_ref: l.external_run_ref, diff --git a/rust/migrations/0075_grid.sql b/rust/migrations/0075_grid.sql new file mode 100644 index 0000000..43ee303 --- /dev/null +++ b/rust/migrations/0075_grid.sql @@ -0,0 +1,155 @@ +-- DecodingUs Grid — the coordination substrate for community realignment & analysis. +-- Design: `documents/design/distributed-compute-grid.md` in the DUNavigator repo (§4, §11). +-- +-- WHY THIS EXISTS. The AppView publishes a list of public-ENA work units; volunteer Navigator +-- instances lease one, fetch it, (re)align it to CHM13, run the analysis stack, and submit a +-- signed digest. Validated contributions earn compute credit on a public leaderboard. The payoff +-- is a uniformly hs1-aligned community corpus derived once and verified, at no central compute +-- cost. This migration is the coordination half: the catalogue, the lease, the submission, and +-- the credit ledger. +-- +-- WHAT IS DELIBERATELY REUSED, not rebuilt: +-- * `fed.pds_node` — the node registry (capabilities, heartbeat, software_version) from +-- `0008_fed.sql`, built for almost exactly this and never wired. +-- * `fed.device_key` — Ed25519 keys a node published to its own repo. Every grid endpoint +-- authenticates through `du_web::sig::verify_signed{,_fresh}`, the same +-- path the D1 exchange and the recruitment Edge already use. +-- * `ident.users` — credit attribution, joined on the DID. +-- `fed.pds_submission` is NOT reused: its status lifecycle means curator review of a proposed +-- variant call, which is a different thing from digest quorum, and overloading it would make both +-- meanings unreadable. +-- +-- THE STATE MODEL, and why it is not the one the design sketched. The design's §3 diagram gives +-- the work unit the states AVAILABLE → LEASED → SUBMITTED → CANONICAL. That cannot express what +-- the same document requires two paragraphs later: `required_replicas` defaults to 2, so a unit +-- routinely needs a second independent result while a first node still holds a lease. "LEASED" +-- and "SUBMITTED" would each have to mean "…and also still claimable", which is not a state. +-- +-- So `work_unit.state` carries only the lifecycle milestones that are genuinely exclusive +-- (AVAILABLE / CANONICAL / CONTESTED / RETIRED), and **claimability is derived**: +-- +-- claimable ⇔ state IN ('AVAILABLE','CONTESTED') +-- AND (active leases + non-divergent submissions) < required_replicas +-- AND the calling DID holds no lease or submission on the unit +-- +-- One SELECT … FOR UPDATE SKIP LOCKED answers that, which is also the design's §4.2 requirement. +-- The lease and submission tables are the source of truth for "how many replicas are in flight"; +-- no counter needs maintaining, so no counter can drift. +-- +-- P1 SCOPE (decided 2026-08-24, amending the design's D3). D3 staged CRAM-passthrough first to +-- retire aligner risk before the coordination loop. That risk evaporated when the realignment +-- module shipped on a pure-Rust mapper, so P1 now carries BOTH data kinds — hence `data_kind` on +-- the work unit from the first migration rather than added later, and `est_bases` from the start +-- because the per-Gbp credit factor is live immediately. + +CREATE SCHEMA IF NOT EXISTS grid; -- distributed community compute: work units, leases, credit + +-- The catalogue of claimable work. One row = one ENA **sample** (decided 2026-08-24 over +-- per-run), because that is the grain Navigator analyses at: `App::analyze_biosample` is the unit +-- of work on the edge, consensus haplogroups are per-biosample, and `du_jobs::crawl_project` +-- already groups ENA runs by sample. A multi-run sample is one lease whose manifest lists every +-- run; the node merges them, which is work it must do anyway before consensus means anything. +CREATE TABLE grid.work_unit ( + id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + sample_accession TEXT NOT NULL UNIQUE, -- SAMEA…/SAMN… — the unit's identity + study_accession TEXT, -- PRJEB…/PRJNA… — provenance + curation filter + data_kind TEXT NOT NULL, -- CRAM/FASTQ — decides whether the node realigns + -- Everything the node needs to fetch without asking ENA anything itself. One entry per file: + -- {run_accession, url, md5, bytes, format}. Curated centrally so a fleet does not hammer the + -- ENA portal rediscovering the same file list (design §6.2, ENA fair-use). + manifest JSONB NOT NULL DEFAULT '[]'::jsonb, + est_bases BIGINT, -- read_count × read_length; drives per-Gbp credit + total_bytes BIGINT, -- the node's download budget preflight + state TEXT NOT NULL DEFAULT 'AVAILABLE', -- AVAILABLE/CANONICAL/CONTESTED/RETIRED + required_replicas SMALLINT NOT NULL DEFAULT 2, + canonical_digest JSONB, -- the agreed digest, once a quorum forms + canonical_at TIMESTAMPTZ, + note TEXT, -- why RETIRED, or what CONTESTED it + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +-- The claim path's index. Partial, because a mature catalogue is mostly CANONICAL and the claim +-- query never looks at those rows. +CREATE INDEX work_unit_claimable_idx ON grid.work_unit (data_kind, id) + WHERE state IN ('AVAILABLE', 'CONTESTED'); +CREATE INDEX work_unit_study_idx ON grid.work_unit (study_accession); + +-- A reservation of one unit by one node, for a bounded time. Separate from the work unit because +-- `required_replicas` > 1 means several nodes legitimately hold concurrent leases on the same +-- unit — the thing a `leased_by` column on the unit itself cannot represent. +CREATE TABLE grid.lease ( + id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + work_unit_id BIGINT NOT NULL REFERENCES grid.work_unit(id) ON DELETE CASCADE, + did TEXT NOT NULL, -- the contributor; authenticated per request + node_id BIGINT REFERENCES fed.pds_node(id) ON DELETE SET NULL, + claimed_at TIMESTAMPTZ NOT NULL DEFAULT now(), + expires_at TIMESTAMPTZ NOT NULL, -- honesty bound; the reaper reclaims past this + heartbeat_at TIMESTAMPTZ, -- extends nothing by itself; evidence of liveness + progress JSONB, -- {stage, fraction} for the fleet view + released_at TIMESTAMPTZ, -- NULL ⇒ active + outcome TEXT -- SUBMITTED/RELEASED/EXPIRED +); + +-- One *active* lease per (unit, DID). A node that re-claims a unit it already holds gets its +-- existing lease back rather than a second row, which makes `claim` idempotent under retry. +-- Partial, so the history of released leases on the same unit stays queryable. +CREATE UNIQUE INDEX lease_active_unit_did_idx ON grid.lease (work_unit_id, did) + WHERE released_at IS NULL; +CREATE INDEX lease_expiry_idx ON grid.lease (expires_at) WHERE released_at IS NULL; +CREATE INDEX lease_did_idx ON grid.lease (did); + +-- A signed result. The digest — not the BAM — is what validation compares: realignment is not +-- byte-deterministic across thread counts and builds, so the agreement test runs over discrete +-- calls plus bucketed continuous metrics (design §5.2). +-- +-- NOTE on `mt_terminal`: the design listed it as an exact-match field. It is NOT (decided +-- 2026-08-24). `App::analyze_biosample` declines to assign mtDNA because that value "is not final +-- on CHM13", and the Grid realigns *to* CHM13 — so the digest cannot require a value the analysis +-- path deliberately does not produce. mt is still published in the full records; it just does not +-- gate canonicalization. +CREATE TABLE grid.submission ( + id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + work_unit_id BIGINT NOT NULL REFERENCES grid.work_unit(id) ON DELETE CASCADE, + did TEXT NOT NULL, + lease_id BIGINT REFERENCES grid.lease(id) ON DELETE SET NULL, + digest JSONB NOT NULL, -- the canonical digest object, as signed + digest_sig TEXT NOT NULL, -- Ed25519 over the canonical digest bytes + stack_version TEXT NOT NULL, -- only compatible majors are compared + reference_build TEXT NOT NULL, -- pins what the calls are even about + aligner TEXT, -- NULL for CRAM passthrough + record_refs JSONB NOT NULL DEFAULT '[]'::jsonb, -- at:// URIs of the published fed records + status TEXT NOT NULL DEFAULT 'PENDING', -- PENDING/AGREED/DIVERGENT/SUPERSEDED + submitted_at TIMESTAMPTZ NOT NULL DEFAULT now(), + validated_at TIMESTAMPTZ +); + +-- One submission per (unit, DID): a contributor cannot pad a quorum with its own repeats, and a +-- resubmit is an update rather than a second vote (design §6.2, free-riding). +CREATE UNIQUE INDEX submission_unit_did_idx ON grid.submission (work_unit_id, did); +CREATE INDEX submission_pending_idx ON grid.submission (work_unit_id) WHERE status = 'PENDING'; +CREATE INDEX submission_did_idx ON grid.submission (did); + +-- The cobblestone ledger. Append-only, awarded only on AGREED/canonical, and unique per +-- (unit, DID) so a re-validation cannot pay twice. The leaderboard is a SUM over this joined to +-- `ident.users` on the DID. +CREATE TABLE grid.credit ( + id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + did TEXT NOT NULL, + -- Resolved at award time. Nullable because a `did:key` node need not have an account yet; + -- such a row is still an honest record of work done, it just cannot appear on the board. + user_id UUID REFERENCES ident.users(id) ON DELETE SET NULL, + work_unit_id BIGINT NOT NULL REFERENCES grid.work_unit(id) ON DELETE CASCADE, + submission_id BIGINT REFERENCES grid.submission(id) ON DELETE SET NULL, + -- Milli-cobblestones, as an exact integer. NUMERIC would need a decimal feature on sqlx that + -- this workspace does not build, and f64 is the wrong shape for a ledger that gets SUMmed over + -- every contribution ever made. Three decimal places was the intended precision anyway, so the + -- integer *is* the value — 1 cobblestone = 1000 here. Never render this number raw. + cobblestones_milli BIGINT NOT NULL, + kind TEXT NOT NULL, -- CANONICAL_FIRST/QUORUM_AGREE/SPOTCHECK_PASS + awarded_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE UNIQUE INDEX credit_unit_did_idx ON grid.credit (work_unit_id, did); +CREATE INDEX credit_did_idx ON grid.credit (did); +CREATE INDEX credit_user_idx ON grid.credit (user_id) WHERE user_id IS NOT NULL; diff --git a/rust/migrations/0076_sequence_library_base_count.sql b/rust/migrations/0076_sequence_library_base_count.sql new file mode 100644 index 0000000..84969e1 --- /dev/null +++ b/rust/migrations/0076_sequence_library_base_count.sql @@ -0,0 +1,34 @@ +-- `genomics.sequence_library.base_count` — total bases sequenced in one run. +-- +-- WHY THIS EXISTS. The DecodingUs Grid pays contributors per gigabase realigned +-- (`documents/design/distributed-compute-grid.md` §6.3), and it had nothing to weigh a FASTQ unit +-- by. `grid.work_unit.est_bases` was computed as `reads × read_length`, but the ENA crawl never +-- sets `read_length`: `du_jobs::crawl_project::mk_lib` writes `read_length: None`, because +-- `du_external::ena::RUN_FIELDS` did not request a length field at all. So `est_bases` was NULL for +-- essentially every crawled sample, and `grid-validate` paid a 90 Gbp realignment exactly what it +-- paid a CRAM passthrough. +-- +-- ENA's `filereport` has published `base_count` on `read_run` all along. This column is where it +-- lands. It is the measured total, not `reads × read_length` — which is only ever a mean-length +-- approximation, and wrong outright for variable-length reads (which is to say, for every long-read +-- platform). +-- +-- WHY NOT THE `atproto` JSONB SLOT, which already carries `{source, run_accession}` for crawled +-- runs. That slot is provenance — where this row came from. `base_count` is a measurement of the +-- library itself, the same kind of fact as the `reads` and `read_length` columns beside it, and it +-- gets summed in an aggregate query that feeds a ledger paying real people. That belongs in a +-- typed column where a NULL is visible, not inside a JSON blob where a missing key reads the same +-- as a zero. +-- +-- BACKFILL. Existing rows stay NULL: `sequence::ingest_libraries` is idempotent at *sample* +-- granularity and skips a sample that already has files, so a re-crawl will not fill them in. Run +-- `du-jobs run-once ena-base-count` to populate them from ENA, a bounded batch at a time. +-- `grid::curation_candidates` falls back to `reads × read_length` where `base_count` is still +-- absent, and `grid-curate` warns with a count of units it published with no estimate at all. + +ALTER TABLE genomics.sequence_library + ADD COLUMN base_count BIGINT; -- total bases in the run, as ENA reports it + +-- The backfill job's work list: crawled ENA runs that have no measurement yet. +CREATE INDEX sequence_library_base_count_backfill_idx ON genomics.sequence_library (id) + WHERE base_count IS NULL;