diff --git a/docs/scheduler/coordinator.md b/docs/scheduler/coordinator.md index 043c34b9..51654ebf 100644 --- a/docs/scheduler/coordinator.md +++ b/docs/scheduler/coordinator.md @@ -491,6 +491,9 @@ caps. swarm-controller alone writes it: | `PUT` | `/api/agents//limits` | `{"cpu_quota": "400%", "memory_max": "8G"}`, every field optional, replacing the stored value | | `DELETE` | `/api/agents//limits` | —; answers 204 | +A stored value the controller can't decode, such as one a newer build wrote, +answers `{}`, with a warning in its log, until the next `PUT` replaces it. + `PUT` refuses an agent no hive's wanted state declares, unknown fields, and any value systemd would ignore, which would leave the agent uncapped. A percentage has at most two decimals, no leading zero in its whole part (it diff --git a/docs/swarm/inference.md b/docs/swarm/inference.md index dfc257c9..50115aeb 100644 --- a/docs/swarm/inference.md +++ b/docs/swarm/inference.md @@ -88,7 +88,9 @@ which swarm-controller alone writes: Every field is optional. The agent runs the same with `{}` as with no key; `GET` answers `"configured": null` for no key (or a cleared one) and `{}` for -a stored value with no fields. +a stored value with no fields. A stored value the controller can't decode, +such as one a newer build wrote, also answers `"configured": null`, with a +warning in its log, until the next `PUT` replaces it. | method | path | body | | -------- | ------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------- | diff --git a/docs/swarm/schedules.md b/docs/swarm/schedules.md index a60ed95b..66a502c7 100644 --- a/docs/swarm/schedules.md +++ b/docs/swarm/schedules.md @@ -93,10 +93,15 @@ involved. `schedule-definitions` holds every prompt as one JSON document under the key `all`. Agents hold no grant on this bucket. Each change re-reads the key and writes against its revision, so a prompt exists for every one of its agents or for -none of them, and two changes at once don't lose one. +none of them, and two changes at once don't lose one. The controller leaves a +prompt it can't decode, such as one a newer build wrote, out of the list and +out of every agent's key, with a warning in its log. It keeps the prompt in +the document as it was, after the others, and won't reuse its id. A document +that isn't an object with a list of `schedules` is an error. `agent-schedules` holds one key per agent: the prompts naming that agent, -without their names or other agents. +without their names or other agents. A key the controller can't decode counts +as no prompts, with a warning in its log, and the next rewrite replaces it. ```json { diff --git a/swarm-controller/src/agent_limits.rs b/swarm-controller/src/agent_limits.rs index cad865e0..868f822f 100644 --- a/swarm-controller/src/agent_limits.rs +++ b/swarm-controller/src/agent_limits.rs @@ -19,7 +19,7 @@ use serde::{Deserialize, Serialize}; use swarm_queue_client::agent_limits::{self, AgentCaps}; use utoipa::ToSchema; -use crate::schedules::{live, queue, read, refused, write}; +use crate::schedules::{live, queue, read_or_default, refused, write}; use crate::{AppState, error_problem, wanted_error_status}; /// How many times a losing writer re-reads and re-applies, as in @@ -209,7 +209,7 @@ pub(crate) async fn get_agent_limits( else { return Ok(Json(Caps::default())); }; - let (current, _) = read::(&store, &agent) + let (current, _, _) = read_or_default::(&store, &agent) .await .map_err(|e| queue_problem(&e))?; Ok(Json(current.into())) @@ -255,10 +255,10 @@ pub(crate) async fn put_agent_limits( } let store = open_or_create(&client).await?; for _ in 0..MAX_ATTEMPTS { - let (current, revision) = read::(&store, &agent) + let (current, revision, undecodable) = read_or_default::(&store, &agent) .await .map_err(|e| queue_problem(&e))?; - if current == value + if (current == value && !undecodable) || write(&store, &agent, &value, revision) .await .map_err(|e| queue_problem(&e))? @@ -302,7 +302,7 @@ pub(crate) async fn delete_agent_limits( } #[cfg(test)] -mod tests { +pub(crate) mod tests { use super::{ Caps, cpu_quota, delete_agent_limits, get_agent_limits, memory_max, put_agent_limits, }; @@ -464,7 +464,7 @@ mod tests { } /// A state on a real server whose `pr1ma` declares `atlas`. - async fn declared(client: &async_nats::Client) -> crate::AppState { + pub(crate) async fn declared(client: &async_nats::Client) -> crate::AppState { WantedWriter::new(client.clone()) .set("pr1ma", "atlas", AgentState::Up) .await @@ -698,4 +698,30 @@ mod tests { ); } } + + /// A stored value this build cannot decode, such as one a newer build + /// wrote, reads as no caps, and a `PUT`, even of no caps, replaces it. + #[tokio::test] + async fn an_undecodable_value_reads_as_no_caps_until_a_put() { + let nats = TestNats::start(); + let client = nats.client().await; + let state = declared(&client).await; + let store = swarm_queue_client::agent_limits::open_or_create(&client) + .await + .expect("bucket"); + store + .put( + "atlas", + br#"{"cpu_quota":{"later":"400%"}}"#.to_vec().into(), + ) + .await + .expect("written"); + + assert_eq!(get(&state, "atlas").await, Caps::default()); + let answer = put(&state, "atlas", Caps::default()) + .await + .expect("caps are set"); + assert_eq!(answer.0, Caps::default()); + assert_eq!(published(&client, "atlas").await.as_deref(), Some("{}")); + } } diff --git a/swarm-controller/src/inference.rs b/swarm-controller/src/inference.rs index 25184ed1..8c63b3b6 100644 --- a/swarm-controller/src/inference.rs +++ b/swarm-controller/src/inference.rs @@ -25,7 +25,7 @@ use serde::{Deserialize, Serialize}; use swarm_queue_client::agent_inference::{self, AgentInference, Effective, Effort, Source}; use utoipa::ToSchema; -use crate::schedules::{live, queue, read, refused, write}; +use crate::schedules::{live, queue, read_or_default, refused, write}; use crate::{AppState, error_problem, wanted_error_status}; /// How many times a losing writer re-reads and re-applies, as in @@ -242,9 +242,9 @@ pub(crate) async fn get_agent_inference( let agent = agent_name(&name).map_err(refused)?; let client = queue(&state).map_err(refused)?; let store = open(&client).await?; - // `Option`'s default is the `None` an absent or deleted key reads as; a - // stored `{}` decodes to `Some`. - let (current, _) = read::>(&store, &agent) + // `Option`'s default is the `None` an absent, deleted or undecodable key + // reads as; a stored `{}` decodes to `Some`. + let (current, _, _) = read_or_default::>(&store, &agent) .await .map_err(|e| queue_problem(&e))?; Ok(Json(AgentInferenceView { @@ -296,10 +296,10 @@ pub(crate) async fn put_agent_inference( } let store = open(&client).await?; for _ in 0..MAX_ATTEMPTS { - let (current, revision) = read::(&store, &agent) + let (current, revision, undecodable) = read_or_default::(&store, &agent) .await .map_err(|e| queue_problem(&e))?; - if current == value + if (current == value && !undecodable) || write(&store, &agent, &value, revision) .await .map_err(|e| queue_problem(&e))? @@ -506,4 +506,36 @@ mod tests { Some(AgentInference::default()) ); } + + /// A stored value this build cannot decode, such as an effort a newer + /// build added, reads as no settings, and a `PUT`, even of no fields, + /// replaces it. + #[tokio::test] + async fn an_undecodable_value_reads_as_unset_until_a_put() { + let nats = crate::test_nats::TestNats::start(); + let client = nats.client().await; + let state = crate::agent_limits::tests::declared(&client).await; + let store = agent_inference::open_or_create(&client) + .await + .expect("bucket"); + store + .put("atlas", br#"{"effort":"later"}"#.to_vec().into()) + .await + .expect("written"); + + let view = get_agent_inference(State(state.clone()), Path("atlas".to_owned())) + .await + .expect("read"); + assert_eq!(view.0.configured, None); + let stored = put_agent_inference( + State(state), + Path("atlas".to_owned()), + Json(InferenceSettings::default()), + ) + .await + .expect("set"); + assert_eq!(stored.0, InferenceSettings::default()); + let entry = store.get("atlas").await.expect("read").expect("a value"); + assert_eq!(&entry[..], b"{}"); + } } diff --git a/swarm-controller/src/schedules.rs b/swarm-controller/src/schedules.rs index eb3bda95..d2f79d28 100644 --- a/swarm-controller/src/schedules.rs +++ b/swarm-controller/src/schedules.rs @@ -52,10 +52,79 @@ const ID_LEN: usize = 8; const RECONCILE_INTERVAL: Duration = Duration::from_mins(1); /// Every schedule in the swarm: the value under [`DEFINITIONS_KEY`]. -#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, Default, PartialEq, Eq)] struct Definitions { - #[serde(default)] schedules: Vec, + /// Stored definitions this build cannot decode, such as ones a newer + /// build wrote. Not listed, edited or projected, and written back as they + /// were read, after the others, so a newer build reads them again after a + /// roll-forward. + unreadable: Vec, +} + +impl Serialize for Definitions { + fn serialize(&self, serializer: S) -> Result { + #[derive(Serialize)] + #[serde(untagged)] + enum Entry<'a> { + Readable(&'a Definition), + Unreadable(&'a serde_json::Value), + } + #[derive(Serialize)] + struct Stored<'a> { + schedules: Vec>, + } + let readable = self.schedules.iter().map(Entry::Readable); + let unreadable = self.unreadable.iter().map(Entry::Unreadable); + Stored { + schedules: readable.chain(unreadable).collect(), + } + .serialize(serializer) + } +} + +impl Definitions { + /// `bytes`, the stored document, with a warning when any definition in + /// it does not decode. A document that is not an object with a list of + /// `schedules` is an error: reading it as no schedules would empty every + /// agent's key. + fn decode(bytes: &[u8]) -> anyhow::Result { + #[derive(Deserialize)] + struct Stored { + #[serde(default)] + schedules: Vec, + } + let stored: Stored = serde_json::from_slice(bytes) + .with_context(|| format!("{DEFINITIONS_KEY} is not decodable"))?; + let mut all = Self::default(); + let mut first_error = None; + for value in stored.schedules { + match Definition::deserialize(&value) { + Ok(d) => all.schedules.push(d), + Err(e) => { + first_error.get_or_insert(e); + all.unreadable.push(value); + } + } + } + if let Some(error) = first_error { + tracing::warn!( + bucket = DEFINITIONS_BUCKET, key = DEFINITIONS_KEY, + unreadable = all.unreadable.len(), %error, + "schedules this build cannot decode are left out and kept" + ); + } + Ok(all) + } + + /// Whether a schedule, readable or not, has `id`. + fn taken(&self, id: &str) -> bool { + self.schedules.iter().any(|d| d.id == id) + || self + .unreadable + .iter() + .any(|v| v.get("id").and_then(serde_json::Value::as_str) == Some(id)) + } } /// One schedule as the operator defines it. @@ -345,7 +414,7 @@ fn mint_id(taken: &Definitions) -> anyhow::Result { getrandom::fill(&mut bytes) .map_err(|e| anyhow::anyhow!("drawing a schedule id from the kernel's CSPRNG: {e}"))?; let id: String = bytes.iter().map(|b| char::from(b'a' + b % 26)).collect(); - if !taken.schedules.iter().any(|d| d.id == id) { + if !taken.taken(&id) { return Ok(id); } } @@ -403,18 +472,36 @@ async fn open_or_create_definitions(client: &async_nats::Client) -> anyhow::Resu } /// The value of `key` decoded, with the revision to write it back against. -/// An absent or deleted key is the default value. -pub(crate) async fn read( +/// An absent or deleted key is the default value. So is a document that does +/// not decode, such as one a newer build wrote: it is logged, and the `bool` +/// is `true`, so a writer replaces it even with the default value. +pub(crate) async fn read_or_default( 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")) +) -> anyhow::Result<(T, Option, bool)> { + let ((value, undecodable), revision) = read_with(store, key, |bytes| { + Ok(match serde_json::from_slice(bytes) { + Ok(value) => (value, false), + Err(e) => { + tracing::warn!( + bucket = %store.name, key, error = %e, + "a stored document is not decodable; reading it as unset until the next write" + ); + (T::default(), true) + } + }) }) - .await + .await?; + Ok((value, revision, undecodable)) } -/// As [`read`], with the stored bytes decoded by `decode`. +/// [`DEFINITIONS_KEY`] in `store`, with the revision to write it back against. +async fn read_definitions(store: &Store) -> anyhow::Result<(Definitions, Option)> { + read_with(store, DEFINITIONS_KEY, Definitions::decode).await +} + +/// The value of `key` as `decode` reads its bytes, with the revision to write +/// it back against. An absent or deleted key is the default value. pub(crate) async fn read_with( store: &Store, key: &str, @@ -463,7 +550,7 @@ async fn commit( mut change: impl FnMut(&mut Definitions) -> Result, ) -> Result { for _ in 0..MAX_ATTEMPTS { - let (mut definitions, revision) = read::(store, DEFINITIONS_KEY).await?; + let (mut definitions, revision) = read_definitions(store).await?; let out = change(&mut definitions)?; if write(store, DEFINITIONS_KEY, &definitions, revision).await? { return Ok(out); @@ -483,10 +570,11 @@ async fn commit( /// is redone from fresh ones. async fn project(definitions: &Store, agents: &Store, agent: &str) -> anyhow::Result<()> { for _ in 0..MAX_ATTEMPTS { - let (current, revision) = read::(agents, agent).await?; - let (all, _) = read::(definitions, DEFINITIONS_KEY).await?; + let (current, revision, undecodable) = + read_or_default::(agents, agent).await?; + let (all, _) = read_definitions(definitions).await?; let wanted = projection(&all, agent); - if current == wanted { + if current == wanted && !undecodable { return Ok(()); } if write(agents, agent, &wanted, revision).await? { @@ -544,7 +632,7 @@ pub(crate) fn spawn_reconcile( continue; } }; - let all = match read::(&store, DEFINITIONS_KEY).await { + let all = match read_definitions(&store).await { Ok((all, _)) => all, Err(e) => { tracing::warn!(error = %format!("{e:#}"), "schedules: reconcile could not read the definitions"); @@ -612,7 +700,7 @@ pub(crate) async fn list_schedules( let store = open_or_create_definitions(&client) .await .map_err(|e| problem(e.into()))?; - let (mut all, _) = read::(&store, DEFINITIONS_KEY) + let (mut all, _) = read_definitions(&store) .await .map_err(|e| problem(e.into()))?; all.schedules @@ -757,6 +845,7 @@ mod tests { fn all() -> Definitions { Definitions { schedules: vec![standup()], + unreadable: Vec::new(), } } @@ -912,6 +1001,75 @@ mod tests { ); } + /// A definition from a newer build among readable ones: listing, a change + /// and projecting keep working on the others, and the unreadable one is + /// written back as it was and keeps its id. An agent key that does not + /// decode is rewritten, even when the agent has no schedules. + #[tokio::test] + async fn an_undecodable_definition_is_kept_and_the_others_still_work() { + let nats = crate::test_nats::TestNats::start(); + let client = nats.client().await; + let store = open_or_create_definitions(&client).await.expect("bucket"); + let mut newer = serde_json::to_value(standup()).unwrap(); + newer["id"] = "new".into(); + newer["anchor_unix"] = "a later cadence".into(); + let doc = serde_json::json!({"schedules": [newer, standup()]}); + store + .put(DEFINITIONS_KEY, doc.to_string().into()) + .await + .expect("written"); + + let (read, _) = read_definitions(&store).await.expect("readable"); + assert_eq!(read.schedules, vec![standup()]); + assert!(read.taken("new")); + let mut other = standup(); + "xyz".clone_into(&mut other.id); + commit(&store, |all| { + all.schedules.push(other.clone()); + Ok(()) + }) + .await + .unwrap_or_else(|_| panic!("committed")); + let entry = store + .get(DEFINITIONS_KEY) + .await + .expect("read") + .expect("a doc"); + let after: serde_json::Value = serde_json::from_slice(&entry).unwrap(); + assert_eq!( + after["schedules"], + serde_json::json!([standup(), other, newer]), + "kept untouched" + ); + + let agents = agent_schedules::open_or_create(&client) + .await + .expect("bucket"); + for agent in ["iris", "scribe"] { + agents + .put(agent, br#"{"schedules":"later"}"#.to_vec().into()) + .await + .expect("written"); + project(&store, &agents, agent).await.expect("projected"); + let entry = agents.get(agent).await.expect("read").expect("a key"); + let (all, _) = read_definitions(&store).await.expect("readable"); + assert_eq!( + serde_json::from_slice::(&entry).expect("decodes"), + projection(&all, agent), + "{agent}" + ); + } + } + + /// Reading a document with no list of schedules as none would empty + /// every agent's key. + #[test] + fn a_document_with_no_list_of_schedules_is_an_error() { + assert!(Definitions::decode(br#"{"schedules":{}}"#).is_err()); + assert!(Definitions::decode(br#""later""#).is_err()); + assert_eq!(Definitions::decode(b"{}").unwrap(), Definitions::default()); + } + /// Letters only: hive-c0re keys its todos `schedule:`, and a /// swarm id that could be all digits could collapse into one of them. #[test]