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.

swarm-controller/src/schedules.rs

@@ -119,11 +119,14 @@ impl Definitions {
 
     /// 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))
+        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))
     }
 }
 
@@ -167,15 +170,23 @@ impl Definition {
     }
 }
 
-/// `agent`'s key: the schedules that name it, in definition order.
-fn projection(definitions: &Definitions, agent: &str) -> AgentSchedules {
+/// `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();
     AgentSchedules {
-        schedules: definitions
-            .schedules
-            .iter()
-            .filter(|d| d.agents.contains(agent))
-            .map(Definition::for_agent)
-            .collect(),
+        schedules: readable.chain(kept).collect(),
     }
 }
 
@@ -573,7 +584,7 @@ async fn project(definitions: &Store, agents: &Store, agent: &str) -> anyhow::Re
         let (current, revision, undecodable) =
             read_or_default::<AgentSchedules>(agents, agent).await?;
         let (all, _) = read_definitions(definitions).await?;
-        let wanted = projection(&all, agent);
+        let wanted = projection(&all, agent, &current);
         if current == wanted && !undecodable {
             return Ok(());
         }
@@ -862,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!(projection(&defs, "argus").schedules.is_empty());
+        assert_eq!(
+            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}");
@@ -1055,12 +1073,49 @@ mod tests {
             let (all, _) = read_definitions(&store).await.expect("readable");
             assert_eq!(
                 serde_json::from_slice::<AgentSchedules>(&entry).expect("decodes"),
-                projection(&all, agent),
+                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]