diff --git a/docs/swarm/schedules.md b/docs/swarm/schedules.md index 66a502c7..a9dc7a0e 100644 --- a/docs/swarm/schedules.md +++ b/docs/swarm/schedules.md @@ -94,10 +94,11 @@ involved. `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 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. +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. A key the controller can't decode counts diff --git a/swarm-controller/src/schedules.rs b/swarm-controller/src/schedules.rs index d2f79d28..4c6c0084 100644 --- a/swarm-controller/src/schedules.rs +++ b/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 { + 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::(agents, agent).await?; let (all, _) = read_definitions(definitions).await?; - let wanted = projection(&all, agent); + let wanted = projection(&all, agent, ¤t); 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::(&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::(&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]