diff --git a/docs/swarm/review-classifier.md b/docs/swarm/review-classifier.md index 74946c80..2b2931b9 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. 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 d91f2953..c7803f67 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. @@ -119,28 +119,76 @@ 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)> { - read(self, KEY).await + 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 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) => { + tracing::warn!( + bucket = BUCKET, key = KEY, error = %e, + "the review classifier runs are not decodable; reading them as none" + ); + return Vec::new(); + } + }; + let mut unreadable = 0_usize; + let mut first_error = None; + let entries = values + .into_iter() + .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(); + 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 /// re-applying when another write landed in between. async fn push(doc: &impl RunDoc, run: &RunRecord) -> anyhow::Result<()> { @@ -154,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 } @@ -211,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) @@ -310,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, } @@ -330,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 { @@ -417,6 +484,72 @@ 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. 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_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 { + 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("a JSON array") + }; + + 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(&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()) + .await + .expect("written"); + assert!(runs.list().await.expect("listed").is_empty()); + runs.record(record("d")).await; + 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] 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..347271a0 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,42 @@ 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 +} + +/// 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 +/// 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| { + if let Some(fields) = value.as_object_mut() { + fields.retain(|name, _| FIELDS.contains(&name.as_str())); + } + 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 +214,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"); @@ -225,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 [ @@ -314,6 +358,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)) + ); + } }