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)) + ); + } }