diff --git a/docs/swarm/review-classifier.md b/docs/swarm/review-classifier.md index b75b2c64..2b2931b9 100644 --- a/docs/swarm/review-classifier.md +++ b/docs/swarm/review-classifier.md @@ -391,9 +391,9 @@ controller reads or writes. The controller creates the bucket on first use and adds each run with a compare-and-set write, so the list survives a controller restart or a move to another host. Recording the 101st run drops the oldest; there's no time-based retention. The controller leaves a run it can't decode, such -as one a newer build recorded, out of the list with a warning in its log, and -the next recorded run writes the list back without it; a document that isn't a -list reads as no runs. The controller logs a failed write, and it +as one a newer build recorded, out of the list with a warning in its log. It +keeps the run in the store as it was and counts it toward the 100. A document +that isn't a list reads as no runs. The controller logs a failed write, and it changes nothing on the forge. `GET /api/review-classifier/runs` returns the list, newest first. Like the diff --git a/swarm-controller/src/review_classifier/runs.rs b/swarm-controller/src/review_classifier/runs.rs index 097647ae..c7803f67 100644 --- a/swarm-controller/src/review_classifier/runs.rs +++ b/swarm-controller/src/review_classifier/runs.rs @@ -119,33 +119,44 @@ pub(super) fn answer(reply: &Reply) -> (Option>, Option) { } } +/// One entry of the stored list. +#[derive(Debug, Clone, PartialEq, Serialize)] +#[serde(untagged)] +pub(crate) enum Stored { + Run(RunRecord), + /// A record this build cannot decode, such as one a newer build wrote. + /// Left out of the list but kept, counted toward [`MAX_RUNS`] and written + /// back as it was read, so a newer build reads it again after a + /// roll-forward. + Unreadable(serde_json::Value), +} + /// Where the runs are kept: [`BUCKET`] in production, a fake in the tests. pub(crate) trait RunDoc { - /// The runs, with the revision to write them back against. - fn read(&self) -> impl Future, Option)>> + Send; + /// The entries, with the revision to write them back against. + fn read(&self) -> impl Future, Option)>> + Send; /// `Ok(false)` when another write landed since `revision`. fn write( &self, - runs: Vec, + runs: Vec, revision: Option, ) -> impl Future> + Send; } impl RunDoc for Store { - async fn read(&self) -> anyhow::Result<(Vec, Option)> { + async fn read(&self) -> anyhow::Result<(Vec, Option)> { read_with(self, KEY, |bytes| Ok(decode(bytes))).await } - async fn write(&self, runs: Vec, revision: Option) -> anyhow::Result { + async fn write(&self, runs: Vec, revision: Option) -> anyhow::Result { write(self, KEY, &runs, revision).await } } -/// The records in `bytes`, the stored document. A record this build cannot -/// decode, such as one a newer build wrote, is dropped with a warning, and a -/// document that is not a JSON array reads as no runs; either way the next -/// write replaces the document with one holding only decodable records. -fn decode(bytes: &[u8]) -> Vec { +/// The entries in `bytes`, the stored document, with a warning when any is +/// [`Stored::Unreadable`]. A document that is not a JSON array reads as no +/// entries, with a warning, and the next write replaces it. +fn decode(bytes: &[u8]) -> Vec { let values: Vec = match serde_json::from_slice(bytes) { Ok(values) => values, Err(e) => { @@ -156,20 +167,26 @@ fn decode(bytes: &[u8]) -> Vec { return Vec::new(); } }; - values + let mut unreadable = 0_usize; + let mut first_error = None; + let entries = values .into_iter() - .enumerate() - .filter_map(|(index, value)| { - serde_json::from_value(value) - .inspect_err(|e| { - tracing::warn!( - bucket = BUCKET, key = KEY, index, error = %e, - "dropping a review classifier run that is not decodable" - ); - }) - .ok() + .map(|value| match RunRecord::deserialize(&value) { + Ok(run) => Stored::Run(run), + Err(e) => { + unreadable += 1; + first_error.get_or_insert(e); + Stored::Unreadable(value) + } }) - .collect() + .collect(); + if let Some(error) = first_error { + tracing::warn!( + bucket = BUCKET, key = KEY, unreadable, %error, + "review classifier runs this build cannot decode are left out of the list and kept" + ); + } + entries } /// Put `run` first in `doc`, keeping the newest [`MAX_RUNS`], re-reading and @@ -185,9 +202,9 @@ async fn push(doc: &impl RunDoc, run: &RunRecord) -> anyhow::Result<()> { anyhow::bail!("gave up recording the run after {MAX_ATTEMPTS} conflicting writes") } -/// `runs` with `run` in front, cut to [`MAX_RUNS`]. -fn newest(run: &RunRecord, mut runs: Vec) -> Vec { - runs.insert(0, run.clone()); +/// `runs` with `run` in front, cut to [`MAX_RUNS`] entries. +fn newest(run: &RunRecord, mut runs: Vec) -> Vec { + runs.insert(0, Stored::Run(run.clone())); runs.truncate(MAX_RUNS); runs } @@ -242,7 +259,15 @@ impl Runs { #[cfg(test)] Self::Fake(doc) => doc.read().await, }; - read.map(|(runs, _)| runs).map_err(|e| { + read.map(|(runs, _)| { + runs.into_iter() + .filter_map(|entry| match entry { + Stored::Run(run) => Some(run), + Stored::Unreadable(_) => None, + }) + .collect() + }) + .map_err(|e| { let detail = format!("{e:#}"); tracing::warn!(error = %detail, "reading the review classifier runs failed"); error_problem(StatusCode::INTERNAL_SERVER_ERROR, &detail) @@ -341,7 +366,7 @@ pub(crate) mod tests { /// every read. #[derive(Default)] pub(crate) struct FakeDoc { - state: Mutex<(Vec, u64)>, + state: Mutex<(Vec, u64)>, conflicts: Mutex, fail: bool, } @@ -361,26 +386,37 @@ pub(crate) mod tests { } } + /// The decodable runs, as the list shows them. pub(crate) fn runs(&self) -> Vec { + self.entries() + .into_iter() + .filter_map(|entry| match entry { + Stored::Run(run) => Some(run), + Stored::Unreadable(_) => None, + }) + .collect() + } + + fn entries(&self) -> Vec { self.state.lock().expect("unpoisoned").0.clone() } } impl RunDoc for FakeDoc { - async fn read(&self) -> anyhow::Result<(Vec, Option)> { + async fn read(&self) -> anyhow::Result<(Vec, Option)> { anyhow::ensure!(!self.fail, "the store is down"); let (runs, revision) = self.state.lock().expect("unpoisoned").clone(); Ok((runs, Some(revision).filter(|r| *r > 0))) } - async fn write(&self, runs: Vec, revision: Option) -> anyhow::Result { + async fn write(&self, runs: Vec, revision: Option) -> anyhow::Result { let mut state = self.state.lock().expect("unpoisoned"); let mut conflicts = self.conflicts.lock().expect("unpoisoned"); if *conflicts > 0 { *conflicts -= 1; let mut other = record(&format!("other-{conflicts}")); other.trigger = Trigger::Review; - state.0.insert(0, other); + state.0.insert(0, Stored::Run(other)); state.1 += 1; } if revision.unwrap_or(0) != state.1 { @@ -449,10 +485,11 @@ pub(crate) mod tests { } /// A record from a newer build among readable ones, and a document that - /// is no list at all: listing and recording both keep working, and the - /// next write leaves a document this build decodes whole. + /// is no list at all: listing and recording both keep working. The newer + /// record is left out of the list and written back as it was; the + /// non-list document is replaced by one this build decodes. #[tokio::test] - async fn an_undecodable_record_or_document_is_dropped_not_fatal() { + async fn an_undecodable_record_is_kept_and_a_non_list_replaced() { let nats = crate::test_nats::TestNats::start(); let client = nats.client().await; let Ok(store) = open(Some(&client)).await else { @@ -461,7 +498,7 @@ pub(crate) mod tests { let runs = Runs::Live(Some(client)); let stored = || async { let entry = store.get(KEY).await.expect("readable").expect("a document"); - serde_json::from_slice::>(&entry).expect("decodes whole") + serde_json::from_slice::>(&entry).expect("a JSON array") }; let mut newer = serde_json::to_value(record("newer")).expect("encodes"); @@ -473,7 +510,10 @@ pub(crate) mod tests { .expect("written"); assert_eq!(ids(&runs.list().await.expect("listed")), ["b", "a"]); runs.record(record("c")).await; - assert_eq!(ids(&stored().await), ["c", "b", "a"]); + assert_eq!(ids(&runs.list().await.expect("listed")), ["c", "b", "a"]); + let after = stored().await; + assert_eq!(after.len(), 4, "{after:?}"); + assert_eq!(after[2], newer, "kept untouched, in its place"); store .put(KEY, br#"{"not":"a list"}"#.to_vec().into()) @@ -481,7 +521,33 @@ pub(crate) mod tests { .expect("written"); assert!(runs.list().await.expect("listed").is_empty()); runs.record(record("d")).await; - assert_eq!(ids(&stored().await), ["d"]); + let after = stored().await; + let decoded: Vec = serde_json::from_value(after.into()).expect("decodes whole"); + assert_eq!(ids(&decoded), ["d"]); + } + + #[tokio::test] + async fn an_unreadable_entry_counts_toward_the_hundred() { + let doc = FakeDoc::default(); + { + let mut state = doc.state.lock().expect("unpoisoned"); + state.0 = (0..MAX_RUNS - 1) + .map(|i| Stored::Run(record(&format!("r{i}")))) + .collect(); + state + .0 + .insert(50, Stored::Unreadable(serde_json::json!({"from": "later"}))); + state.1 = 1; + } + push(&doc, &record("new")).await.expect("stored"); + let entries = doc.entries(); + assert_eq!(entries.len(), MAX_RUNS); + assert!(entries.iter().any(|e| matches!(e, Stored::Unreadable(_)))); + assert_eq!( + doc.runs().len(), + MAX_RUNS - 1, + "the unreadable entry took a run's place" + ); } #[tokio::test] diff --git a/swarm-controller/src/review_classifier/settings.rs b/swarm-controller/src/review_classifier/settings.rs index 7b747a21..347271a0 100644 --- a/swarm-controller/src/review_classifier/settings.rs +++ b/swarm-controller/src/review_classifier/settings.rs @@ -102,16 +102,18 @@ async fn stored(store: &Store) -> anyhow::Result<(Settings, Option)> { read_with(store, KEY, |bytes| Ok(decode(bytes))).await } +/// Every field of [`Settings`]. `a_full_document_round_trips` fails when one +/// is missing. +const FIELDS: &[&str] = &["model", "effort", "max_concurrent_reviews"]; + /// `bytes`, the stored document, as settings. Unlike a `PUT` body, a field -/// this build does not know, such as one a newer build added, is ignored; -/// a document that still does not decode is [`Settings::default`], with a -/// warning, until the next `PUT` replaces it. The known fields are those -/// [`Settings::default`] encodes, which is every field. +/// not in [`FIELDS`], such as one a newer build added, is ignored; a document +/// that still does not decode is [`Settings::default`], with a warning, +/// until the next `PUT` replaces it. fn decode(bytes: &[u8]) -> Settings { let decoded = serde_json::from_slice(bytes).and_then(|mut value: serde_json::Value| { - let known = serde_json::to_value(Settings::default())?; - if let (Some(fields), Some(known)) = (value.as_object_mut(), known.as_object()) { - fields.retain(|name, _| known.contains_key(name)); + if let Some(fields) = value.as_object_mut() { + fields.retain(|name, _| FIELDS.contains(&name.as_str())); } serde_json::from_value(value) }); @@ -249,6 +251,24 @@ mod tests { ); } + /// Every field set, every `Option` `Some`: a stored document keeps all of + /// them, so a field missing from [`FIELDS`] fails here. + #[test] + fn a_full_document_round_trips() { + let full = Settings { + model: "sonnet".to_owned(), + effort: Some(Effort::High), + max_concurrent_reviews: 5, + }; + let encoded = serde_json::to_value(&full).unwrap(); + let mut keys: Vec<_> = encoded.as_object().unwrap().keys().collect(); + let mut fields = FIELDS.to_vec(); + keys.sort_unstable(); + fields.sort_unstable(); + assert_eq!(keys, fields); + assert_eq!(decode(&serde_json::to_vec(&full).unwrap()), full); + } + #[test] fn bad_settings_are_refused() { for json in [