diff --git a/docs/scheduler/coordinator.md b/docs/scheduler/coordinator.md index 51654ebf..043c34b9 100644 --- a/docs/scheduler/coordinator.md +++ b/docs/scheduler/coordinator.md @@ -491,9 +491,6 @@ 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 50115aeb..dfc257c9 100644 --- a/docs/swarm/inference.md +++ b/docs/swarm/inference.md @@ -88,9 +88,7 @@ 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 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. +a stored value with no fields. | method | path | body | | -------- | ------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------- | diff --git a/docs/swarm/schedules.md b/docs/swarm/schedules.md index a9dc7a0e..a60ed95b 100644 --- a/docs/swarm/schedules.md +++ b/docs/swarm/schedules.md @@ -93,16 +93,10 @@ 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. The controller leaves a -prompt it can't decode, such as one a newer build wrote, out of the list, with -a warning in its log. An agent's key that already held the prompt keeps it as -it was, so it goes on firing, and no other key gets it. The controller 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. +none of them, and two changes at once don't lose one. `agent-schedules` holds one key per agent: the prompts naming that agent, -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. +without their names or other agents. ```json { diff --git a/swarm-controller/src/agent_limits.rs b/swarm-controller/src/agent_limits.rs index 868f822f..cad865e0 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_or_default, refused, write}; +use crate::schedules::{live, queue, read, 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_or_default::(&store, &agent) + let (current, _) = read::(&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, undecodable) = read_or_default::(&store, &agent) + let (current, revision) = read::(&store, &agent) .await .map_err(|e| queue_problem(&e))?; - if (current == value && !undecodable) + if current == value || 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)] -pub(crate) mod tests { +mod tests { use super::{ Caps, cpu_quota, delete_agent_limits, get_agent_limits, memory_max, put_agent_limits, }; @@ -464,7 +464,7 @@ pub(crate) mod tests { } /// A state on a real server whose `pr1ma` declares `atlas`. - pub(crate) async fn declared(client: &async_nats::Client) -> crate::AppState { + async fn declared(client: &async_nats::Client) -> crate::AppState { WantedWriter::new(client.clone()) .set("pr1ma", "atlas", AgentState::Up) .await @@ -698,30 +698,4 @@ pub(crate) 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 8c63b3b6..25184ed1 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_or_default, refused, write}; +use crate::schedules::{live, queue, read, 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, deleted or undecodable key - // reads as; a stored `{}` decodes to `Some`. - let (current, _, _) = read_or_default::>(&store, &agent) + // `Option`'s default is the `None` an absent or deleted key reads as; a + // stored `{}` decodes to `Some`. + let (current, _) = read::>(&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, undecodable) = read_or_default::(&store, &agent) + let (current, revision) = read::(&store, &agent) .await .map_err(|e| queue_problem(&e))?; - if (current == value && !undecodable) + if current == value || write(&store, &agent, &value, revision) .await .map_err(|e| queue_problem(&e))? @@ -506,36 +506,4 @@ 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 4c6c0084..eb3bda95 100644 --- a/swarm-controller/src/schedules.rs +++ b/swarm-controller/src/schedules.rs @@ -52,82 +52,10 @@ 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)] +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] 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_ids().any(|i| i == id) - } - - /// The ids of the [`Self::unreadable`] definitions that have one. - fn unreadable_ids(&self) -> impl Iterator { - self.unreadable - .iter() - .filter_map(|v| v.get("id").and_then(serde_json::Value::as_str)) - } } /// One schedule as the operator defines it. @@ -170,23 +98,15 @@ impl Definition { } } -/// `agent`'s key: the schedules that name it, in definition order, then the -/// entries of `current`, its key as stored, whose definition this build -/// cannot decode. Those stay as they were, so they keep firing, and are never -/// added to a key that did not hold them. -fn projection(definitions: &Definitions, agent: &str, current: &AgentSchedules) -> AgentSchedules { - let readable = definitions - .schedules - .iter() - .filter(|d| d.agents.contains(agent)) - .map(Definition::for_agent); - let kept = current - .schedules - .iter() - .filter(|s| definitions.unreadable_ids().any(|id| id == s.id)) - .cloned(); +/// `agent`'s key: the schedules that name it, in definition order. +fn projection(definitions: &Definitions, agent: &str) -> AgentSchedules { AgentSchedules { - schedules: readable.chain(kept).collect(), + schedules: definitions + .schedules + .iter() + .filter(|d| d.agents.contains(agent)) + .map(Definition::for_agent) + .collect(), } } @@ -425,7 +345,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.taken(&id) { + if !taken.schedules.iter().any(|d| d.id == id) { return Ok(id); } } @@ -483,36 +403,18 @@ 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. 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( +/// An absent or deleted key is the default value. +pub(crate) async fn read( store: &Store, key: &str, -) -> 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) - } - }) +) -> anyhow::Result<(T, Option)> { + read_with(store, key, |bytes| { + serde_json::from_slice(bytes).with_context(|| format!("{key} is not decodable")) }) - .await?; - Ok((value, revision, undecodable)) + .await } -/// [`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. +/// As [`read`], with the stored bytes decoded by `decode`. pub(crate) async fn read_with( store: &Store, key: &str, @@ -561,7 +463,7 @@ async fn commit( mut change: impl FnMut(&mut Definitions) -> Result, ) -> Result { for _ in 0..MAX_ATTEMPTS { - let (mut definitions, revision) = read_definitions(store).await?; + let (mut definitions, revision) = read::(store, DEFINITIONS_KEY).await?; let out = change(&mut definitions)?; if write(store, DEFINITIONS_KEY, &definitions, revision).await? { return Ok(out); @@ -581,11 +483,10 @@ 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, undecodable) = - read_or_default::(agents, agent).await?; - let (all, _) = read_definitions(definitions).await?; - let wanted = projection(&all, agent, ¤t); - if current == wanted && !undecodable { + let (current, revision) = read::(agents, agent).await?; + let (all, _) = read::(definitions, DEFINITIONS_KEY).await?; + let wanted = projection(&all, agent); + if current == wanted { return Ok(()); } if write(agents, agent, &wanted, revision).await? { @@ -643,7 +544,7 @@ pub(crate) fn spawn_reconcile( continue; } }; - let all = match read_definitions(&store).await { + let all = match read::(&store, DEFINITIONS_KEY).await { Ok((all, _)) => all, Err(e) => { tracing::warn!(error = %format!("{e:#}"), "schedules: reconcile could not read the definitions"); @@ -711,7 +612,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_definitions(&store) + let (mut all, _) = read::(&store, DEFINITIONS_KEY) .await .map_err(|e| problem(e.into()))?; all.schedules @@ -856,7 +757,6 @@ mod tests { fn all() -> Definitions { Definitions { schedules: vec![standup()], - unreadable: Vec::new(), } } @@ -873,17 +773,10 @@ mod tests { "xyz".clone_into(&mut other.id); other.agents = ["scribe"].map(str::to_owned).into(); defs.schedules.push(other); - let atlas = projection(&defs, "atlas", &AgentSchedules::default()); + let atlas = projection(&defs, "atlas"); assert_eq!(atlas.schedules, vec![standup().for_agent()]); - assert_eq!( - projection(&defs, "scribe", &AgentSchedules::default()).schedules[0].id, - "xyz" - ); - assert!( - projection(&defs, "argus", &AgentSchedules::default()) - .schedules - .is_empty() - ); + assert_eq!(projection(&defs, "scribe").schedules[0].id, "xyz"); + assert!(projection(&defs, "argus").schedules.is_empty()); let encoded = serde_json::to_string(&atlas).unwrap(); assert!(!encoded.contains("\"name\""), "{encoded}"); assert!(!encoded.contains("iris"), "{encoded}"); @@ -1019,112 +912,6 @@ 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, &AgentSchedules::default()), - "{agent}" - ); - } - } - - /// A projected schedule whose definition stops decoding stays in the - /// agent's key through a reconcile, so it keeps firing; one deleted from - /// the definitions leaves it. - #[tokio::test] - async fn a_projected_schedule_that_stops_decoding_keeps_firing() { - let nats = crate::test_nats::TestNats::start(); - let client = nats.client().await; - let store = open_or_create_definitions(&client).await.expect("bucket"); - let agents = agent_schedules::open_or_create(&client) - .await - .expect("bucket"); - let mut deleted = standup(); - "xyz".clone_into(&mut deleted.id); - let doc = serde_json::json!({"schedules": [standup(), deleted]}); - store - .put(DEFINITIONS_KEY, doc.to_string().into()) - .await - .expect("written"); - let iris = BTreeSet::from(["iris".to_owned()]); - let key = || async { - let entry = agents.get("iris").await.expect("read").expect("a key"); - serde_json::from_slice::(&entry).expect("decodes") - }; - project_all(&client, &store, &iris).await; - assert_eq!(key().await.schedules.len(), 2); - - let mut newer = serde_json::to_value(standup()).unwrap(); - newer["anchor_unix"] = "a later cadence".into(); - let doc = serde_json::json!({"schedules": [newer]}); - store - .put(DEFINITIONS_KEY, doc.to_string().into()) - .await - .expect("written"); - project_all(&client, &store, &iris).await; - assert_eq!(key().await.schedules, vec![standup().for_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]