diff --git a/docs/swarm/review-classifier.md b/docs/swarm/review-classifier.md index 2b2931b9..74946c80 100644 --- a/docs/swarm/review-classifier.md +++ b/docs/swarm/review-classifier.md @@ -135,9 +135,7 @@ 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. 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. +claude's default effort, two reviews at a time. `GET /api/review-classifier/settings` answers the settings in force. `PUT /api/review-classifier/settings` replaces them whole and answers what it @@ -157,8 +155,7 @@ 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. A run request -with no model runs on `haiku`. +place. Lowering the limit lets the runs already going finish. 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`: @@ -390,10 +387,7 @@ 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 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 +there's no time-based retention. 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 c7803f67..d91f2953 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_with, write}; +use crate::schedules::{read, write}; use crate::{AppState, error_problem}; /// The bucket holding the runs. Agents hold no grant on it. @@ -119,76 +119,28 @@ 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 entries, with the revision to write them back against. - fn read(&self) -> impl Future, Option)>> + Send; + /// The runs, 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_with(self, KEY, |bytes| Ok(decode(bytes))).await + async fn read(&self) -> anyhow::Result<(Vec, Option)> { + read(self, KEY).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<()> { @@ -202,9 +154,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`] entries. -fn newest(run: &RunRecord, mut runs: Vec) -> Vec { - runs.insert(0, Stored::Run(run.clone())); +/// `runs` with `run` in front, cut to [`MAX_RUNS`]. +fn newest(run: &RunRecord, mut runs: Vec) -> Vec { + runs.insert(0, run.clone()); runs.truncate(MAX_RUNS); runs } @@ -259,15 +211,7 @@ impl Runs { #[cfg(test)] Self::Fake(doc) => doc.read().await, }; - read.map(|(runs, _)| { - runs.into_iter() - .filter_map(|entry| match entry { - Stored::Run(run) => Some(run), - Stored::Unreadable(_) => None, - }) - .collect() - }) - .map_err(|e| { + read.map(|(runs, _)| runs).map_err(|e| { let detail = format!("{e:#}"); tracing::warn!(error = %detail, "reading the review classifier runs failed"); error_problem(StatusCode::INTERNAL_SERVER_ERROR, &detail) @@ -366,7 +310,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, } @@ -386,37 +330,26 @@ 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, Stored::Run(other)); + state.0.insert(0, other); state.1 += 1; } if revision.unwrap_or(0) != state.1 { @@ -484,72 +417,6 @@ 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 347271a0..c361b4d2 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_with, refused, write}; +use crate::schedules::{queue, read, 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: swarm_queue_client::review_classifier::DEFAULT_MODEL.to_owned(), + model: "haiku".to_owned(), effort: None, max_concurrent_reviews: 2, } @@ -90,42 +90,13 @@ pub(crate) trait SettingsSource { impl SettingsSource for async_nats::Client { async fn settings(&self) -> anyhow::Result { - let (settings, _) = stored(&open(self).await?) + let (settings, _) = read::(&open(self).await?, KEY) .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()); @@ -214,10 +185,13 @@ 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 (_, revision) = stored(&store).await.map_err(|e| queue_problem(&e))?; - if write(&store, KEY, &value, revision) + let (current, revision) = read::(&store, KEY) .await - .map_err(|e| queue_problem(&e))? + .map_err(|e| queue_problem(&e))?; + if current == value + || write(&store, KEY, &value, revision) + .await + .map_err(|e| queue_problem(&e))? { state.reviews.resize(value.max_concurrent_reviews); tracing::info!(%operator, ?value, "set the review classifier's settings"); @@ -251,24 +225,6 @@ 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 [ @@ -358,56 +314,6 @@ 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 eb3bda95..ee080e5a 100644 --- a/swarm-controller/src/schedules.rs +++ b/swarm-controller/src/schedules.rs @@ -407,18 +407,6 @@ 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) @@ -430,7 +418,9 @@ pub(crate) async fn read_with( if entry.operation != Operation::Put { return Ok((T::default(), Some(entry.revision))); } - Ok((decode(&entry.value)?, Some(entry.revision))) + let value = + serde_json::from_slice(&entry.value).with_context(|| format!("{key} is not decodable"))?; + Ok((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 b23ec20f..d3d5620f 100644 --- a/swarm-queue-client/src/review_classifier.rs +++ b/swarm-queue-client/src/review_classifier.rs @@ -33,14 +33,6 @@ 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)] @@ -59,9 +51,6 @@ 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, @@ -187,27 +176,4 @@ 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)) - ); - } }