From 6f0a72b24629708aa22fef70e10be8bb898282e4 Mon Sep 17 00:00:00 2001 From: atlas Date: Tue, 13 Oct 2026 02:24:26 +0200 Subject: [PATCH 1/2] review classifier: recover from a stored record it can't decode The controller keeps the classifier's newest runs as one JSON array and its settings as one JSON object, each under one KV key. A document this build could not decode, as after a rollback past a build that added an outcome or a settings field, failed every read and every write until someone deleted the key by hand. A classifier newer than its controller refused every run request for the missing `model`. The run store now decodes the array one record at a time: a record that does not decode is dropped with a WARN naming the bucket, key, index and error, and a document that is not an array reads as no runs. Listing keeps working, and the next recorded run writes back only the records that decoded. The stored settings ignore a field this build does not know; one that still does not decode reads as the defaults with a WARN. The PUT body keeps `deny_unknown_fields`. A PUT now always writes, so a PUT of the defaults also replaces an undecodable document. A run request without `model` runs on haiku, the swarm-wide setting's default, now one constant in swarm-queue-client. `schedules::read` gains a sibling, `read_with`, taking the decoder. Closes #5162 --- docs/swarm/review-classifier.md | 12 ++- .../src/review_classifier/runs.rs | 71 +++++++++++++- .../src/review_classifier/settings.rs | 92 +++++++++++++++++-- swarm-controller/src/schedules.rs | 16 +++- swarm-queue-client/src/review_classifier.rs | 34 +++++++ 5 files changed, 208 insertions(+), 17 deletions(-) diff --git a/docs/swarm/review-classifier.md b/docs/swarm/review-classifier.md index 74946c80..b75b2c64 100644 --- a/docs/swarm/review-classifier.md +++ b/docs/swarm/review-classifier.md @@ -135,7 +135,9 @@ The model, the effort and how many requested reviews run at once are one document for the whole swarm. `swarm-controller` keeps it in its `review-classifier-settings` KV bucket under the key `all`, and no other service reads it. With no document stored, the classifier runs on `haiku` at -claude's default effort, two reviews at a time. +claude's default effort, two reviews at a time. The controller +ignores a stored field it doesn't know, and a document it still can't decode +counts as none, with a warning in its log, until the next PUT replaces it. `GET /api/review-classifier/settings` answers the settings in force. `PUT /api/review-classifier/settings` replaces them whole and answers what it @@ -155,7 +157,8 @@ stored. Both take and give one shape: The controller reads the settings for every run and sends the model and effort in the run request, so a change applies from the next run with no restart. A try-out reads them when it's posted, a review when it gets its -place. Lowering the limit lets the runs already going finish. +place. Lowering the limit lets the runs already going finish. A run request +with no model runs on `haiku`. The PUT needs the `Remote-User` header the gateway sets, as the try-out POST does. Every error is problem+json with the reason in `detail`: @@ -387,7 +390,10 @@ The newest 100 runs are one JSON array under the key `recent` in the 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 logs a failed write, and it +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 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 d91f2953..097647ae 100644 --- a/swarm-controller/src/review_classifier/runs.rs +++ b/swarm-controller/src/review_classifier/runs.rs @@ -23,7 +23,7 @@ use serde::{Deserialize, Serialize}; use swarm_queue_client::review_classifier::{Overall, Reply, Row}; use utoipa::ToSchema; -use crate::schedules::{read, write}; +use crate::schedules::{read_with, write}; use crate::{AppState, error_problem}; /// The bucket holding the runs. Agents hold no grant on it. @@ -133,7 +133,7 @@ pub(crate) trait RunDoc { impl RunDoc for Store { async fn read(&self) -> anyhow::Result<(Vec, Option)> { - read(self, KEY).await + read_with(self, KEY, |bytes| Ok(decode(bytes))).await } async fn write(&self, runs: Vec, revision: Option) -> anyhow::Result { @@ -141,6 +141,37 @@ impl RunDoc for Store { } } +/// 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 { + let values: Vec = match serde_json::from_slice(bytes) { + Ok(values) => values, + Err(e) => { + tracing::warn!( + bucket = BUCKET, key = KEY, error = %e, + "the review classifier runs are not decodable; reading them as none" + ); + return Vec::new(); + } + }; + 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() + }) + .collect() +} + /// Put `run` first in `doc`, keeping the newest [`MAX_RUNS`], re-reading and /// re-applying when another write landed in between. async fn push(doc: &impl RunDoc, run: &RunRecord) -> anyhow::Result<()> { @@ -417,6 +448,42 @@ pub(crate) mod tests { assert_eq!(ids(&listed), ["b", "a"]); } + /// 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. + #[tokio::test] + async fn an_undecodable_record_or_document_is_dropped_not_fatal() { + let nats = crate::test_nats::TestNats::start(); + let client = nats.client().await; + let Ok(store) = open(Some(&client)).await else { + panic!("the bucket did not open"); + }; + 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") + }; + + let mut newer = serde_json::to_value(record("newer")).expect("encodes"); + newer["outcome"] = "a-later-outcome".into(); + let doc = serde_json::json!([record("b"), newer, record("a")]); + store + .put(KEY, doc.to_string().into()) + .await + .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"]); + + store + .put(KEY, br#"{"not":"a list"}"#.to_vec().into()) + .await + .expect("written"); + assert!(runs.list().await.expect("listed").is_empty()); + runs.record(record("d")).await; + assert_eq!(ids(&stored().await), ["d"]); + } + #[tokio::test] async fn the_route_lists_the_runs_newest_first_and_a_down_store_is_an_error() { let doc = Arc::new(FakeDoc::default()); diff --git a/swarm-controller/src/review_classifier/settings.rs b/swarm-controller/src/review_classifier/settings.rs index c361b4d2..7b747a21 100644 --- a/swarm-controller/src/review_classifier/settings.rs +++ b/swarm-controller/src/review_classifier/settings.rs @@ -22,7 +22,7 @@ use serde::{Deserialize, Serialize}; use swarm_queue_client::agent_inference::Effort; use utoipa::ToSchema; -use crate::schedules::{queue, read, refused, write}; +use crate::schedules::{queue, read_with, refused, write}; use crate::{AppState, error_problem, wanted_error_status}; /// The bucket holding the settings. No other service holds a grant on it. @@ -59,7 +59,7 @@ pub(crate) struct Settings { impl Default for Settings { fn default() -> Self { Self { - model: "haiku".to_owned(), + model: swarm_queue_client::review_classifier::DEFAULT_MODEL.to_owned(), effort: None, max_concurrent_reviews: 2, } @@ -90,13 +90,40 @@ pub(crate) trait SettingsSource { impl SettingsSource for async_nats::Client { async fn settings(&self) -> anyhow::Result { - let (settings, _) = read::(&open(self).await?, KEY) + let (settings, _) = stored(&open(self).await?) .await .context("reading the review classifier's settings")?; Ok(settings) } } +/// The stored settings, with the revision to write them back against. +async fn stored(store: &Store) -> anyhow::Result<(Settings, Option)> { + read_with(store, KEY, |bytes| Ok(decode(bytes))).await +} + +/// `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. +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)); + } + serde_json::from_value(value) + }); + decoded.unwrap_or_else(|e| { + tracing::warn!( + bucket = BUCKET, key = KEY, error = %e, + "the review classifier settings are not decodable; using the defaults" + ); + Settings::default() + }) +} + /// Open [`BUCKET`], creating it if nothing has yet. async fn open(client: &async_nats::Client) -> anyhow::Result { let js = async_nats::jetstream::new(client.clone()); @@ -185,13 +212,10 @@ pub(crate) async fn put_settings( let client = queue(&state).map_err(refused)?; let store = open(&client).await.map_err(|e| queue_problem(&e))?; for _ in 0..MAX_ATTEMPTS { - let (current, revision) = read::(&store, KEY) + let (_, revision) = stored(&store).await.map_err(|e| queue_problem(&e))?; + if write(&store, KEY, &value, revision) .await - .map_err(|e| queue_problem(&e))?; - if current == value - || write(&store, KEY, &value, revision) - .await - .map_err(|e| queue_problem(&e))? + .map_err(|e| queue_problem(&e))? { state.reviews.resize(value.max_concurrent_reviews); tracing::info!(%operator, ?value, "set the review classifier's settings"); @@ -314,6 +338,56 @@ mod tests { .expect("a request") } + /// A stored field this build does not know is ignored, and a document + /// that still does not decode reads as the defaults; either way the GET + /// and a run's read keep working, and a PUT, even of the defaults, + /// replaces it with one that decodes. + #[tokio::test] + async fn an_undecodable_document_reads_as_the_defaults_until_a_put() { + let nats = TestNats::start(); + let (app, _) = app(&nats).await; + let client = nats.client().await; + let store = open(&client).await.expect("the bucket"); + let defaults = serde_json::to_value(Settings::default()).unwrap(); + + let newer = serde_json::json!({ + "model": "sonnet", "effort": "high", "max_concurrent_reviews": 3, "later": true, + }); + store + .put(KEY, newer.to_string().into()) + .await + .expect("written"); + let (status, read, _) = send(&app, get()).await; + assert_eq!(status, StatusCode::OK, "{read}"); + assert_eq!( + read, + serde_json::json!({"model": "sonnet", "effort": "high", "max_concurrent_reviews": 3}) + ); + + store + .put( + KEY, + br#"{"model":"sonnet","effort":"later"}"#.to_vec().into(), + ) + .await + .expect("written"); + let (status, read, _) = send(&app, get()).await; + assert_eq!(status, StatusCode::OK, "{read}"); + assert_eq!(read, defaults); + assert_eq!( + client.settings().await.expect("readable"), + Settings::default() + ); + + let (status, stored, _) = send(&app, put(&defaults.to_string())).await; + assert_eq!(status, StatusCode::OK, "{stored}"); + let entry = store.get(KEY).await.expect("readable").expect("a document"); + assert_eq!( + serde_json::from_slice::(&entry).expect("decodes strictly"), + Settings::default() + ); + } + #[tokio::test] async fn a_put_is_what_the_get_and_the_next_run_read() { let nats = TestNats::start(); diff --git a/swarm-controller/src/schedules.rs b/swarm-controller/src/schedules.rs index ee080e5a..eb3bda95 100644 --- a/swarm-controller/src/schedules.rs +++ b/swarm-controller/src/schedules.rs @@ -407,6 +407,18 @@ async fn open_or_create_definitions(client: &async_nats::Client) -> anyhow::Resu pub(crate) async fn read( store: &Store, key: &str, +) -> anyhow::Result<(T, Option)> { + read_with(store, key, |bytes| { + serde_json::from_slice(bytes).with_context(|| format!("{key} is not decodable")) + }) + .await +} + +/// As [`read`], with the stored bytes decoded by `decode`. +pub(crate) async fn read_with( + store: &Store, + key: &str, + decode: impl FnOnce(&[u8]) -> anyhow::Result, ) -> anyhow::Result<(T, Option)> { let Some(entry) = store .entry(key) @@ -418,9 +430,7 @@ pub(crate) async fn read( if entry.operation != Operation::Put { return Ok((T::default(), Some(entry.revision))); } - let value = - serde_json::from_slice(&entry.value).with_context(|| format!("{key} is not decodable"))?; - Ok((value, Some(entry.revision))) + Ok((decode(&entry.value)?, Some(entry.revision))) } /// Write `value` to `key` against `revision`. `Ok(false)` when another write diff --git a/swarm-queue-client/src/review_classifier.rs b/swarm-queue-client/src/review_classifier.rs index d3d5620f..b23ec20f 100644 --- a/swarm-queue-client/src/review_classifier.rs +++ b/swarm-queue-client/src/review_classifier.rs @@ -33,6 +33,14 @@ pub const CLASSIFICATION_DEADLINE: Duration = Duration::from_mins(3); /// this long. pub const RUN_DEADLINE: Duration = Duration::from_mins(31); +/// The model a run uses when its request names none, and the swarm-wide +/// setting's default. +pub const DEFAULT_MODEL: &str = "haiku"; + +fn default_model() -> String { + DEFAULT_MODEL.to_owned() +} + /// One run: the PR as the controller read it, and the classifications to /// answer about it. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] @@ -51,6 +59,9 @@ pub struct TryOut { pub diff: String, pub classifications: Vec, /// The model every classification runs on, as claude's `--model` takes it. + /// Absent is [`DEFAULT_MODEL`], so a controller that sends none still gets + /// its run answered. + #[serde(default = "default_model")] pub model: String, /// The effort every classification runs at; `None` passes no `--effort`. pub effort: Option, @@ -176,4 +187,27 @@ mod tests { let all = CLASSIFICATION_DEADLINE * u32::try_from(MAX_CLASSIFICATIONS).expect("small"); assert!(RUN_DEADLINE > all); } + + #[test] + fn a_request_without_a_model_runs_on_the_default() { + let mut request = serde_json::json!({ + "operator": "alice", + "repo": "o/r", + "pr": 7, + "head_sha": "abc123", + "commits": [], + "diff": "", + "classifications": [], + }); + let run: TryOut = serde_json::from_value(request.clone()).expect("decodes"); + assert_eq!((run.model.as_str(), run.effort), (DEFAULT_MODEL, None)); + // Control: a named model and effort are the ones used. + request["model"] = "sonnet".into(); + request["effort"] = "high".into(); + let run: TryOut = serde_json::from_value(request).expect("decodes"); + assert_eq!( + (run.model.as_str(), run.effort), + ("sonnet", Some(Effort::High)) + ); + } } From daaf920e06a0a3c9c326b4677c6bb0b2234a00aa Mon Sep 17 00:00:00 2001 From: atlas Date: Tue, 13 Oct 2026 02:54:07 +0200 Subject: [PATCH 2/2] review classifier: keep stored runs this build can't decode A run record this build couldn't decode was dropped from the list it read, and the next recorded run wrote the list back without it, so a rollback followed by a roll-forward lost the newer build's runs for good. The stored list is now a list of entries, each a decoded run or the raw JSON of one that didn't decode. The route still shows only the decoded runs. An undecodable entry keeps its place, counts toward the 100, and is written back as it was read. A read that finds any logs one WARN with their count and the first error. The settings decode derived its known fields from what `Settings::default()` serializes, so a later field skipped when unset would have been stripped from a valid stored document and lost on the next PUT. The known fields are now an explicit list. A test serializes a fully populated `Settings`, checks that its keys match the list, and checks that `decode` round-trips it. --- docs/swarm/review-classifier.md | 6 +- .../src/review_classifier/runs.rs | 138 +++++++++++++----- .../src/review_classifier/settings.rs | 34 ++++- 3 files changed, 132 insertions(+), 46 deletions(-) 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 [