schedules: keep projecting a schedule whose definition stops decoding

A reconcile projected only the definitions this build decodes, so after
a rollback an agent's already-projected schedule whose definition a newer
build rewrote was dropped from its key within a minute and stopped
firing. The projection now keeps an agent's stored entries whose id
belongs to an unreadable definition, as they were; it never adds one to
a key that did not hold it. A deleted readable definition still leaves
every key.

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,16 @@ 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, 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.
 
 `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,82 @@ 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_ids().any(|i| i == id)
+    }
+
+    /// The ids of the [`Self::unreadable`] definitions that have one.
+    fn unreadable_ids(&self) -> impl Iterator<Item = &str> {
+        self.unreadable
+            .iter()
+            .filter_map(|v| v.get("id").and_then(serde_json::Value::as_str))
+    }
 }
 
 /// One schedule as the operator defines it.
@@ -98,15 +170,23 @@ impl Definition {
     }
 }
 
-/// `agent`'s key: the schedules that name it, in definition order.
+/// `agent`'s key: the schedules that name it, in definition order, then the
-fn projection(definitions: &Definitions, agent: &str) -> AgentSchedules {
+/// entries of `current`, its key as stored, whose definition this build
-    AgentSchedules {
+/// cannot decode. Those stay as they were, so they keep firing, and are never
-        schedules: definitions
+/// 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)
+        .map(Definition::for_agent);
-            .collect(),
+    let kept = current
+        .schedules
+        .iter()
+        .filter(|s| definitions.unreadable_ids().any(|id| id == s.id))
+        .cloned();
+    AgentSchedules {
+        schedules: readable.chain(kept).collect(),
     }
 }
 
@@ -345,7 +425,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 +483,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 +561,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 +581,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 wanted = projection(&all, agent);
+        let (all, _) = read_definitions(definitions).await?;
-        if current == wanted {
+        let wanted = projection(&all, agent, &current);
+        if current == wanted && !undecodable {
             return Ok(());
         }
         if write(agents, agent, &wanted, revision).await? {
@@ -544,7 +643,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 +711,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 +856,7 @@ mod tests {
     fn all() -> Definitions {
         Definitions {
             schedules: vec![standup()],
+            unreadable: Vec::new(),
         }
     }
 
@@ -773,10 +873,17 @@ 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");
+        let atlas = projection(&defs, "atlas", &AgentSchedules::default());
         assert_eq!(atlas.schedules, vec![standup().for_agent()]);
-        assert_eq!(projection(&defs, "scribe").schedules[0].id, "xyz");
+        assert_eq!(
-        assert!(projection(&defs, "argus").schedules.is_empty());
+            projection(&defs, "scribe", &AgentSchedules::default()).schedules[0].id,
+            "xyz"
+        );
+        assert!(
+            projection(&defs, "argus", &AgentSchedules::default())
+                .schedules
+                .is_empty()
+        );
         let encoded = serde_json::to_string(&atlas).unwrap();
         assert!(!encoded.contains("\"name\""), "{encoded}");
         assert!(!encoded.contains("iris"), "{encoded}");
@@ -912,6 +1019,112 @@ 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, &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::<AgentSchedules>(&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:<number>`, and a
     /// swarm id that could be all digits could collapse into one of them.
     #[test]