controller stores: recover from a record they can't decode

After a rollback, a record a newer build wrote (a new enum variant, a
changed required field) stopped decoding, and the strict
`schedules::read` turned that into an error on every read and write of
the key until someone deleted it by hand.

- agent_limits / inference (one document per agent): an undecodable
  document reads as unset, with a WARN naming the bucket, key and error.
  The PUT's "equal to stored, skip the write" shortcut no longer applies
  to an undecodable document, so a PUT of the defaults replaces it too.
- schedule definitions (one document, every schedule): salvaged per
  definition, as the review-classifier run list is. An undecodable
  definition is kept as its raw JSON, hidden from the list, edits and
  projections, written back by every commit after the readable ones, and
  its id is never minted again. A document with no list of schedules
  stays an error: reading it as empty would empty every agent's key.
- per-agent agent-schedules: an undecodable key reads as no schedules
  and is rewritten on the next projection, even when the agent has none.

`schedules::read` is replaced by `read_or_default`; `wanted` keeps its
own strict read.

Closes #5193

docs/scheduler/coordinator.md

@@ -491,6 +491,9 @@ caps. swarm-controller alone writes it:
 | `PUT`    | `/api/agents/<agent>/limits` | `{"cpu_quota": "400%", "memory_max": "8G"}`, every field optional, replacing the stored value |
 | `DELETE` | `/api/agents/<agent>/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

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                                                                                                                                                  |
 | -------- | ------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------- |

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
 {

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::<AgentCaps>(&store, &agent)
+    let (current, _, _) = read_or_default::<AgentCaps>(&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::<AgentCaps>(&store, &agent)
+        let (current, revision, undecodable) = read_or_default::<AgentCaps>(&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("{}"));
+    }
 }

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
+    // `Option`'s default is the `None` an absent, deleted or undecodable key
-    // stored `{}` decodes to `Some`.
+    // reads as; a stored `{}` decodes to `Some`.
-    let (current, _) = read::<Option<AgentInference>>(&store, &agent)
+    let (current, _, _) = read_or_default::<Option<AgentInference>>(&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::<AgentInference>(&store, &agent)
+        let (current, revision, undecodable) = read_or_default::<AgentInference>(&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"{}");
+    }
 }

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<Definition>,
+    /// 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<serde_json::Value>,
+}
+
+impl Serialize for Definitions {
+    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
+        #[derive(Serialize)]
+        #[serde(untagged)]
+        enum Entry<'a> {
+            Readable(&'a Definition),
+            Unreadable(&'a serde_json::Value),
+        }
+        #[derive(Serialize)]
+        struct Stored<'a> {
+            schedules: Vec<Entry<'a>>,
+        }
+        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<Self> {
+        #[derive(Deserialize)]
+        struct Stored {
+            #[serde(default)]
+            schedules: Vec<serde_json::Value>,
+        }
+        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<String> {
         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.
+/// An absent or deleted key is the default value. So is a document that does
-pub(crate) async fn read<T: Default + serde::de::DeserializeOwned>(
+/// 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<T: Default + serde::de::DeserializeOwned>(
     store: &Store,
     key: &str,
-) -> anyhow::Result<(T, Option<u64>)> {
+) -> anyhow::Result<(T, Option<u64>, bool)> {
-    read_with(store, key, |bytes| {
+    let ((value, undecodable), revision) = read_with(store, key, |bytes| {
-        serde_json::from_slice(bytes).with_context(|| format!("{key} is not decodable"))
+        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<u64>)> {
+    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<T: Default>(
     store: &Store,
     key: &str,
@@ -463,7 +550,7 @@ async fn commit<T>(
     mut change: impl FnMut(&mut Definitions) -> Result<T, Failure>,
 ) -> Result<T, Failure> {
     for _ in 0..MAX_ATTEMPTS {
-        let (mut definitions, revision) = read::<Definitions>(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<T>(
 /// 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::<AgentSchedules>(agents, agent).await?;
+        let (current, revision, undecodable) =
-        let (all, _) = read::<Definitions>(definitions, DEFINITIONS_KEY).await?;
+            read_or_default::<AgentSchedules>(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::<Definitions>(&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::<Definitions>(&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::<AgentSchedules>(&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:<number>`, and a
     /// swarm id that could be all digits could collapse into one of them.
     #[test]