08fec0b011
rustdoc reads the bare name as both the function and the std macro, which the workspace doc check refuses.
1180 lines
42 KiB
Rust
1180 lines
42 KiB
Rust
//! `/api/schedules` — the operator's scheduled prompts. Each has an id, a
|
|
//! name, a body, a cadence and the agents it goes to.
|
|
//!
|
|
//! **Two buckets.** Every schedule in the swarm is one document under one key
|
|
//! ([`DEFINITIONS_KEY`] in [`DEFINITIONS_BUCKET`]), which only this daemon
|
|
//! reads or writes. Every change is one compare-and-set on that key, so a
|
|
//! schedule exists for all of its agents or for none of them.
|
|
//!
|
|
//! Each agent reads only its own key of `agent-schedules`
|
|
//! (`swarm_queue_client::agent_schedules`), which holds the schedules naming
|
|
//! it, without their names or other agents. That key is a projection of the
|
|
//! definitions: after a change this daemon rewrites the key of each agent the
|
|
//! change touched, and [`spawn_reconcile`] rewrites every agent's key every
|
|
//! [`RECONCILE_INTERVAL`], so a projection write that failed is repaired
|
|
//! within that interval. A projection keeps each schedule's id, which the
|
|
//! agent keys its fired-slot record on, so dropping an agent and adding it
|
|
//! back does not refire a slot.
|
|
|
|
use std::collections::BTreeSet;
|
|
use std::sync::Arc;
|
|
use std::time::Duration;
|
|
|
|
use anyhow::Context as _;
|
|
use async_nats::jetstream::kv::{CreateErrorKind, Operation, Store, UpdateErrorKind};
|
|
use axum::Json;
|
|
use axum::extract::{Path, State};
|
|
use axum::http::StatusCode;
|
|
use problem_details::ProblemDetails;
|
|
use serde::{Deserialize, Deserializer, Serialize};
|
|
use swarm_queue_client::agent_schedules::{self, AgentSchedules, Schedule};
|
|
use utoipa::ToSchema;
|
|
|
|
use crate::agent_renewal::unix_now;
|
|
use crate::wanted::WantedWriter;
|
|
use crate::{AppState, error_problem};
|
|
|
|
/// The bucket holding every schedule definition. Agents hold no grant on it.
|
|
pub const DEFINITIONS_BUCKET: &str = "schedule-definitions";
|
|
|
|
/// The one key in [`DEFINITIONS_BUCKET`].
|
|
pub const DEFINITIONS_KEY: &str = "all";
|
|
|
|
/// How many times a losing writer re-reads and re-applies, as in
|
|
/// `wanted::MAX_ATTEMPTS`.
|
|
const MAX_ATTEMPTS: usize = 5;
|
|
|
|
/// Length of a minted schedule id.
|
|
const ID_LEN: usize = 8;
|
|
|
|
/// How often [`spawn_reconcile`] rewrites every agent's key. Also the
|
|
/// longest a failed projection write leaves an agent behind.
|
|
const RECONCILE_INTERVAL: Duration = Duration::from_mins(1);
|
|
|
|
/// Every schedule in the swarm: the value under [`DEFINITIONS_KEY`].
|
|
#[derive(Debug, Clone, Default, PartialEq, Eq)]
|
|
struct Definitions {
|
|
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.
|
|
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
|
struct Definition {
|
|
id: String,
|
|
name: String,
|
|
agents: BTreeSet<String>,
|
|
body: String,
|
|
anchor_unix: i64,
|
|
#[serde(default, skip_serializing_if = "Option::is_none")]
|
|
interval_seconds: Option<u64>,
|
|
#[serde(default)]
|
|
paused: bool,
|
|
}
|
|
|
|
impl Definition {
|
|
/// What an agent sees of this schedule.
|
|
fn for_agent(&self) -> Schedule {
|
|
Schedule {
|
|
id: self.id.clone(),
|
|
body: self.body.clone(),
|
|
anchor_unix: self.anchor_unix,
|
|
interval_seconds: self.interval_seconds,
|
|
paused: self.paused,
|
|
}
|
|
}
|
|
|
|
/// The next fire time after `now`, or `None` when nothing will fire:
|
|
/// paused, or a one-shot whose time has passed.
|
|
fn next_fire(&self, now: i64) -> Option<i64> {
|
|
if self.paused {
|
|
return None;
|
|
}
|
|
if now < self.anchor_unix {
|
|
return Some(self.anchor_unix);
|
|
}
|
|
let interval = i64::try_from(self.interval_seconds?).ok()?;
|
|
Some(self.for_agent().latest_slot(now)? + interval)
|
|
}
|
|
}
|
|
|
|
/// `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: readable.chain(kept).collect(),
|
|
}
|
|
}
|
|
|
|
/// One schedule as the API shows it.
|
|
#[derive(Debug, Serialize, ToSchema)]
|
|
pub struct ScheduleView {
|
|
id: String,
|
|
/// The operator's title for it.
|
|
name: String,
|
|
/// The agents it fires in, sorted.
|
|
agents: Vec<String>,
|
|
/// The prompt, delivered to each agent verbatim.
|
|
body: String,
|
|
/// The first fire time, unix seconds.
|
|
anchor_unix: i64,
|
|
/// Seconds between fires; `null` for a one-shot.
|
|
interval_seconds: Option<u64>,
|
|
paused: bool,
|
|
/// The next fire time on the schedule's grid, unix seconds, computed when
|
|
/// the response is built; `null` when paused or a one-shot whose time
|
|
/// has passed. Says nothing about whether an agent fired an earlier one.
|
|
next_fire_unix: Option<i64>,
|
|
}
|
|
|
|
impl ScheduleView {
|
|
fn new(d: &Definition, now: i64) -> Self {
|
|
Self {
|
|
id: d.id.clone(),
|
|
name: d.name.clone(),
|
|
agents: d.agents.iter().cloned().collect(),
|
|
body: d.body.clone(),
|
|
anchor_unix: d.anchor_unix,
|
|
interval_seconds: d.interval_seconds,
|
|
paused: d.paused,
|
|
next_fire_unix: d.next_fire(now),
|
|
}
|
|
}
|
|
}
|
|
|
|
/// A new schedule.
|
|
#[derive(Debug, Clone, Deserialize, ToSchema)]
|
|
pub struct AddScheduleRequest {
|
|
/// The title. Must not be empty.
|
|
name: String,
|
|
/// The agents to fire in, at least one. Each must be declared on a hive.
|
|
agents: Vec<String>,
|
|
/// The prompt, delivered verbatim. Must not be empty.
|
|
body: String,
|
|
/// The first fire time, unix seconds. Defaults to now.
|
|
anchor_unix: Option<i64>,
|
|
/// Seconds between fires, at least 1. Absent for a one-shot.
|
|
interval_seconds: Option<u64>,
|
|
#[serde(default)]
|
|
paused: bool,
|
|
}
|
|
|
|
/// A partial change to one schedule; an absent field is left as it is.
|
|
#[derive(Debug, Clone, Default, Deserialize, ToSchema)]
|
|
pub struct EditScheduleRequest {
|
|
name: Option<String>,
|
|
body: Option<String>,
|
|
anchor_unix: Option<i64>,
|
|
/// A number sets the interval; `null` makes the schedule a one-shot.
|
|
#[serde(default)]
|
|
#[schema(value_type = Option<u64>)]
|
|
interval_seconds: IntervalEdit,
|
|
paused: Option<bool>,
|
|
/// Agents to add. Each must be declared on a hive; one already targeted
|
|
/// is left as it is.
|
|
#[serde(default)]
|
|
add_agents: Vec<String>,
|
|
/// Agents to remove. Each must be a current target, and at least one
|
|
/// target must remain.
|
|
#[serde(default)]
|
|
remove_agents: Vec<String>,
|
|
}
|
|
|
|
/// What an edit does to the interval: an absent field keeps it, `null` makes
|
|
/// a one-shot, a number sets it.
|
|
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
|
|
enum IntervalEdit {
|
|
#[default]
|
|
Keep,
|
|
Set(Option<u64>),
|
|
}
|
|
|
|
impl<'de> Deserialize<'de> for IntervalEdit {
|
|
fn deserialize<D: Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
|
|
Option::<u64>::deserialize(d).map(Self::Set)
|
|
}
|
|
}
|
|
|
|
/// Why a change was refused before it was written.
|
|
#[derive(Debug, PartialEq, Eq)]
|
|
enum Refusal {
|
|
Invalid(String),
|
|
NotDeclared(String),
|
|
NoSuchSchedule(String),
|
|
}
|
|
|
|
fn invalid(why: &str) -> Refusal {
|
|
Refusal::Invalid(why.to_owned())
|
|
}
|
|
|
|
fn check_text(value: &str, what: &str) -> Result<(), Refusal> {
|
|
if value.trim().is_empty() {
|
|
return Err(Refusal::Invalid(format!("the {what} is empty")));
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
fn check_interval(interval: Option<u64>) -> Result<(), Refusal> {
|
|
if interval == Some(0) {
|
|
return Err(invalid(
|
|
"interval_seconds must be at least 1; omit it for a one-shot",
|
|
));
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// `names` as agent identifiers, each declared on some hive per `live`.
|
|
fn check_agents(names: &[String], live: &BTreeSet<String>) -> Result<BTreeSet<String>, Refusal> {
|
|
let mut agents = BTreeSet::new();
|
|
for name in names {
|
|
let agent = hive_types::Ident::parse(name)
|
|
.map(hive_types::Ident::into_string)
|
|
.map_err(|why| Refusal::Invalid(format!("agent {name:?}: {why}")))?;
|
|
if !live.contains(&agent) {
|
|
return Err(Refusal::NotDeclared(agent));
|
|
}
|
|
agents.insert(agent);
|
|
}
|
|
Ok(agents)
|
|
}
|
|
|
|
impl AddScheduleRequest {
|
|
fn into_definition(
|
|
self,
|
|
id: String,
|
|
now: i64,
|
|
live: &BTreeSet<String>,
|
|
) -> Result<Definition, Refusal> {
|
|
check_text(&self.name, "name")?;
|
|
check_text(&self.body, "body")?;
|
|
check_interval(self.interval_seconds)?;
|
|
let agents = check_agents(&self.agents, live)?;
|
|
if agents.is_empty() {
|
|
return Err(invalid("a schedule needs at least one agent"));
|
|
}
|
|
Ok(Definition {
|
|
id,
|
|
name: self.name,
|
|
agents,
|
|
body: self.body,
|
|
anchor_unix: self.anchor_unix.unwrap_or(now),
|
|
interval_seconds: self.interval_seconds,
|
|
paused: self.paused,
|
|
})
|
|
}
|
|
}
|
|
|
|
impl EditScheduleRequest {
|
|
/// Apply the edit to schedule `id`, returning it as changed and every
|
|
/// agent it named before or after.
|
|
fn apply(
|
|
self,
|
|
definitions: &mut Definitions,
|
|
id: &str,
|
|
live: &BTreeSet<String>,
|
|
) -> Result<(Definition, BTreeSet<String>), Refusal> {
|
|
if let Some(name) = &self.name {
|
|
check_text(name, "name")?;
|
|
}
|
|
if let Some(body) = &self.body {
|
|
check_text(body, "body")?;
|
|
}
|
|
if let IntervalEdit::Set(interval) = self.interval_seconds {
|
|
check_interval(interval)?;
|
|
}
|
|
let added = check_agents(&self.add_agents, live)?;
|
|
let d = definitions
|
|
.schedules
|
|
.iter_mut()
|
|
.find(|d| d.id == id)
|
|
.ok_or_else(|| Refusal::NoSuchSchedule(id.to_owned()))?;
|
|
let mut agents = d.agents.clone();
|
|
agents.extend(added);
|
|
for name in &self.remove_agents {
|
|
if !agents.remove(name) {
|
|
return Err(Refusal::Invalid(format!(
|
|
"agent {name:?} is not a target of schedule {id:?}"
|
|
)));
|
|
}
|
|
}
|
|
if agents.is_empty() {
|
|
return Err(invalid(
|
|
"this would leave the schedule with no agents; remove the schedule instead",
|
|
));
|
|
}
|
|
let touched = d.agents.union(&agents).cloned().collect();
|
|
d.agents = agents;
|
|
if let Some(name) = self.name {
|
|
d.name = name;
|
|
}
|
|
if let Some(body) = self.body {
|
|
d.body = body;
|
|
}
|
|
if let Some(anchor) = self.anchor_unix {
|
|
d.anchor_unix = anchor;
|
|
}
|
|
if let IntervalEdit::Set(interval) = self.interval_seconds {
|
|
d.interval_seconds = interval;
|
|
}
|
|
if let Some(paused) = self.paused {
|
|
d.paused = paused;
|
|
}
|
|
Ok((d.clone(), touched))
|
|
}
|
|
}
|
|
|
|
/// Remove schedule `id`, returning the agents it named.
|
|
fn remove(definitions: &mut Definitions, id: &str) -> Result<BTreeSet<String>, Refusal> {
|
|
let at = definitions
|
|
.schedules
|
|
.iter()
|
|
.position(|d| d.id == id)
|
|
.ok_or_else(|| Refusal::NoSuchSchedule(id.to_owned()))?;
|
|
Ok(definitions.schedules.remove(at).agents)
|
|
}
|
|
|
|
/// A fresh id: lowercase letters only, so the agent's todo key
|
|
/// `schedule:<id>` can never equal one hive-c0re pushes for its numeric ids.
|
|
fn mint_id(taken: &Definitions) -> anyhow::Result<String> {
|
|
loop {
|
|
let mut bytes = [0u8; ID_LEN];
|
|
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) {
|
|
return Ok(id);
|
|
}
|
|
}
|
|
}
|
|
|
|
/// The failure of one request, as the problem it answers with.
|
|
enum Failure {
|
|
Refused(Refusal),
|
|
Queue(anyhow::Error),
|
|
}
|
|
|
|
impl From<anyhow::Error> for Failure {
|
|
fn from(e: anyhow::Error) -> Self {
|
|
Self::Queue(e)
|
|
}
|
|
}
|
|
|
|
impl From<Refusal> for Failure {
|
|
fn from(r: Refusal) -> Self {
|
|
Self::Refused(r)
|
|
}
|
|
}
|
|
|
|
fn problem(failure: Failure) -> ProblemDetails {
|
|
match failure {
|
|
Failure::Refused(Refusal::Invalid(why)) => error_problem(StatusCode::BAD_REQUEST, &why),
|
|
Failure::Refused(Refusal::NotDeclared(agent)) => error_problem(
|
|
StatusCode::BAD_REQUEST,
|
|
&format!("no hive's wanted state declares an agent {agent:?} that is not destroyed"),
|
|
),
|
|
Failure::Refused(Refusal::NoSuchSchedule(id)) => {
|
|
error_problem(StatusCode::NOT_FOUND, &format!("no schedule {id:?}"))
|
|
}
|
|
Failure::Queue(e) => {
|
|
tracing::warn!(error = %format!("{e:#}"), "schedules: queue failure");
|
|
error_problem(crate::wanted_error_status(&e), &format!("{e:#}"))
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Open a bucket only this daemon writes, creating it if nothing has yet.
|
|
async fn open_or_create_definitions(client: &async_nats::Client) -> anyhow::Result<Store> {
|
|
let js = async_nats::jetstream::new(client.clone());
|
|
if let Ok(store) = js.get_key_value(DEFINITIONS_BUCKET).await {
|
|
return Ok(store);
|
|
}
|
|
js.create_key_value(async_nats::jetstream::kv::Config {
|
|
bucket: DEFINITIONS_BUCKET.to_owned(),
|
|
description: "Every scheduled prompt the swarm controller keeps".to_owned(),
|
|
history: 1,
|
|
..Default::default()
|
|
})
|
|
.await
|
|
.with_context(|| format!("creating the {DEFINITIONS_BUCKET} bucket"))
|
|
}
|
|
|
|
/// 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<T: Default + serde::de::DeserializeOwned>(
|
|
store: &Store,
|
|
key: &str,
|
|
) -> anyhow::Result<(T, Option<u64>, 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)
|
|
}
|
|
})
|
|
})
|
|
.await?;
|
|
Ok((value, revision, undecodable))
|
|
}
|
|
|
|
/// [`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,
|
|
decode: impl FnOnce(&[u8]) -> anyhow::Result<T>,
|
|
) -> anyhow::Result<(T, Option<u64>)> {
|
|
let Some(entry) = store
|
|
.entry(key)
|
|
.await
|
|
.with_context(|| format!("reading {key}"))?
|
|
else {
|
|
return Ok((T::default(), None));
|
|
};
|
|
if entry.operation != Operation::Put {
|
|
return Ok((T::default(), Some(entry.revision)));
|
|
}
|
|
Ok((decode(&entry.value)?, Some(entry.revision)))
|
|
}
|
|
|
|
/// Write `value` to `key` against `revision`. `Ok(false)` when another write
|
|
/// landed first.
|
|
pub(crate) async fn write<T: Serialize>(
|
|
store: &Store,
|
|
key: &str,
|
|
value: &T,
|
|
revision: Option<u64>,
|
|
) -> anyhow::Result<bool> {
|
|
Ok(write_revision(store, key, value, revision).await?.is_some())
|
|
}
|
|
|
|
/// As [`write()`], answering the revision written, or `None` when another write
|
|
/// landed first.
|
|
pub(crate) async fn write_revision<T: Serialize>(
|
|
store: &Store,
|
|
key: &str,
|
|
value: &T,
|
|
revision: Option<u64>,
|
|
) -> anyhow::Result<Option<u64>> {
|
|
let encoded = serde_json::to_vec(value)?;
|
|
match revision {
|
|
Some(revision) => match store.update(key, encoded.into(), revision).await {
|
|
Ok(written) => Ok(Some(written)),
|
|
Err(e) if matches!(e.kind(), UpdateErrorKind::WrongLastRevision) => Ok(None),
|
|
Err(e) => Err(e).with_context(|| format!("writing {key}")),
|
|
},
|
|
None => match store.create(key, encoded.into()).await {
|
|
Ok(written) => Ok(Some(written)),
|
|
Err(e) if matches!(e.kind(), CreateErrorKind::AlreadyExists) => Ok(None),
|
|
Err(e) => Err(e).with_context(|| format!("writing {key}")),
|
|
},
|
|
}
|
|
}
|
|
|
|
/// A write to a key, with what it replaced, so [`Replaced::undo`] can reverse
|
|
/// it.
|
|
pub(crate) struct Replaced<T> {
|
|
/// The value the key held, or `None` when it held none.
|
|
previous: Option<T>,
|
|
/// The revision the write made.
|
|
written: u64,
|
|
}
|
|
|
|
impl<T: Serialize> Replaced<T> {
|
|
pub(crate) fn new(previous: Option<T>, written: u64) -> Self {
|
|
Self { previous, written }
|
|
}
|
|
|
|
/// Put `previous` back, or remove the key if it held none, against the
|
|
/// revision the write made. `Ok(false)` when another write has landed
|
|
/// since: that one is newer than what is being undone and stays.
|
|
pub(crate) async fn undo(self, store: &Store, key: &str) -> anyhow::Result<bool> {
|
|
match self.previous {
|
|
Some(previous) => write(store, key, &previous, Some(self.written)).await,
|
|
None => match store.delete_expect_revision(key, Some(self.written)).await {
|
|
Ok(()) => Ok(true),
|
|
Err(e) if matches!(e.kind(), UpdateErrorKind::WrongLastRevision) => Ok(false),
|
|
Err(e) => Err(e).with_context(|| format!("removing {key}")),
|
|
},
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Apply `change` to the definitions and write them back, re-reading and
|
|
/// re-applying when another write landed in between.
|
|
async fn commit<T>(
|
|
store: &Store,
|
|
mut change: impl FnMut(&mut Definitions) -> Result<T, Failure>,
|
|
) -> Result<T, Failure> {
|
|
for _ in 0..MAX_ATTEMPTS {
|
|
let (mut definitions, revision) = read_definitions(store).await?;
|
|
let out = change(&mut definitions)?;
|
|
if write(store, DEFINITIONS_KEY, &definitions, revision).await? {
|
|
return Ok(out);
|
|
}
|
|
tracing::debug!("schedules: definitions write lost a race, retrying");
|
|
}
|
|
Err(
|
|
anyhow::anyhow!("gave up writing the schedules after {MAX_ATTEMPTS} conflicting writes")
|
|
.into(),
|
|
)
|
|
}
|
|
|
|
/// Rewrite `agent`'s key from the current definitions.
|
|
///
|
|
/// The agent key's revision is read before the definitions, so a write based
|
|
/// on definitions older than another writer's fails its revision check and
|
|
/// 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::<AgentSchedules>(agents, agent).await?;
|
|
let (all, _) = read_definitions(definitions).await?;
|
|
let wanted = projection(&all, agent, ¤t);
|
|
if current == wanted && !undecodable {
|
|
return Ok(());
|
|
}
|
|
if write(agents, agent, &wanted, revision).await? {
|
|
return Ok(());
|
|
}
|
|
}
|
|
anyhow::bail!(
|
|
"gave up writing the schedules of {agent} after {MAX_ATTEMPTS} conflicting writes"
|
|
)
|
|
}
|
|
|
|
/// [`project`] each of `touched`, logging a failure: the definitions are
|
|
/// already written, and [`spawn_reconcile`] repairs the agent's key.
|
|
async fn project_all(client: &async_nats::Client, definitions: &Store, touched: &BTreeSet<String>) {
|
|
let agents = match agent_schedules::open_or_create(client).await {
|
|
Ok(store) => store,
|
|
Err(e) => {
|
|
tracing::warn!(
|
|
error = %swarm_queue_client::chain(&e),
|
|
"schedules: opening the agent-schedules bucket failed; the reconcile pass retries"
|
|
);
|
|
return;
|
|
}
|
|
};
|
|
for agent in touched {
|
|
if let Err(e) = project(definitions, &agents, agent).await {
|
|
tracing::warn!(
|
|
%agent,
|
|
error = %format!("{e:#}"),
|
|
"schedules: rewriting the agent's key failed; the reconcile pass retries"
|
|
);
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Rewrite every agent's key from the definitions, now and every
|
|
/// [`RECONCILE_INTERVAL`] after: the agents the definitions name, and every
|
|
/// live agent, so one dropped from its last schedule is emptied too.
|
|
pub(crate) fn spawn_reconcile(
|
|
client: async_nats::Client,
|
|
wanted: Arc<WantedWriter>,
|
|
hives: Vec<String>,
|
|
) {
|
|
tokio::spawn(async move {
|
|
let mut ticker = tokio::time::interval(RECONCILE_INTERVAL);
|
|
loop {
|
|
ticker.tick().await;
|
|
if swarm_queue_client::ensure_connected(&client).is_err() {
|
|
continue;
|
|
}
|
|
let store = match open_or_create_definitions(&client).await {
|
|
Ok(store) => store,
|
|
Err(e) => {
|
|
tracing::warn!(error = %format!("{e:#}"), "schedules: reconcile could not open the definitions");
|
|
continue;
|
|
}
|
|
};
|
|
let all = match read_definitions(&store).await {
|
|
Ok((all, _)) => all,
|
|
Err(e) => {
|
|
tracing::warn!(error = %format!("{e:#}"), "schedules: reconcile could not read the definitions");
|
|
continue;
|
|
}
|
|
};
|
|
let mut agents = crate::subagent_term::live_agents(&wanted, &hives).await;
|
|
agents.extend(all.schedules.iter().flat_map(|d| d.agents.iter().cloned()));
|
|
project_all(&client, &store, &agents).await;
|
|
}
|
|
});
|
|
}
|
|
|
|
/// The connected queue client, or the problem.
|
|
pub(crate) fn queue(state: &AppState) -> Result<async_nats::Client, (StatusCode, String)> {
|
|
let status = state.status.as_ref().ok_or_else(|| {
|
|
(
|
|
StatusCode::SERVICE_UNAVAILABLE,
|
|
"no swarm queue is configured on this host".to_owned(),
|
|
)
|
|
})?;
|
|
let client = status.queue_client();
|
|
swarm_queue_client::ensure_connected(&client).map_err(|e| {
|
|
(
|
|
StatusCode::SERVICE_UNAVAILABLE,
|
|
swarm_queue_client::chain(&e),
|
|
)
|
|
})?;
|
|
Ok(client)
|
|
}
|
|
|
|
pub(crate) fn refused((status, detail): (StatusCode, String)) -> ProblemDetails {
|
|
error_problem(status, &detail)
|
|
}
|
|
|
|
/// Every agent some hive declares and has not destroyed.
|
|
pub(crate) async fn live(state: &AppState) -> Result<BTreeSet<String>, ProblemDetails> {
|
|
let Some(wanted) = state.wanted.as_ref() else {
|
|
return Err(error_problem(
|
|
StatusCode::SERVICE_UNAVAILABLE,
|
|
"no swarm queue is configured on this host",
|
|
));
|
|
};
|
|
let hives: Vec<String> = state.hives.iter().map(|h| h.name.clone()).collect();
|
|
Ok(crate::subagent_term::live_agents(wanted, &hives).await)
|
|
}
|
|
|
|
#[utoipa::path(
|
|
get,
|
|
path = "/api/schedules",
|
|
responses(
|
|
(status = 200, description = "every schedule, sorted by name then id; empty when \
|
|
there are none", body = Vec<ScheduleView>),
|
|
(status = 503, description = "no swarm queue is configured on this host, or it is \
|
|
not connected (problem+json)", body = String),
|
|
(status = 500, description = "the schedules could not be read (problem+json)",
|
|
body = String),
|
|
),
|
|
tag = "schedules"
|
|
)]
|
|
pub(crate) async fn list_schedules(
|
|
State(state): State<AppState>,
|
|
) -> Result<Json<Vec<ScheduleView>>, ProblemDetails> {
|
|
let client = queue(&state).map_err(refused)?;
|
|
let store = open_or_create_definitions(&client)
|
|
.await
|
|
.map_err(|e| problem(e.into()))?;
|
|
let (mut all, _) = read_definitions(&store)
|
|
.await
|
|
.map_err(|e| problem(e.into()))?;
|
|
all.schedules
|
|
.sort_by(|a, b| (&a.name, &a.id).cmp(&(&b.name, &b.id)));
|
|
let now = unix_now();
|
|
Ok(Json(
|
|
all.schedules
|
|
.iter()
|
|
.map(|d| ScheduleView::new(d, now))
|
|
.collect(),
|
|
))
|
|
}
|
|
|
|
#[utoipa::path(
|
|
post,
|
|
path = "/api/schedules",
|
|
request_body = AddScheduleRequest,
|
|
responses(
|
|
(status = 201, description = "the schedule as stored, with its new id",
|
|
body = ScheduleView),
|
|
(status = 400, description = "the name or body is empty, the interval is 0, no \
|
|
agent is named, or an agent is not an identifier or not declared on any hive \
|
|
(problem+json)", body = String),
|
|
(status = 503, description = "no swarm queue is configured on this host, or it is \
|
|
not connected (problem+json)", body = String),
|
|
(status = 500, description = "the schedules could not be written (problem+json)",
|
|
body = String),
|
|
),
|
|
tag = "schedules"
|
|
)]
|
|
pub(crate) async fn add_schedule(
|
|
State(state): State<AppState>,
|
|
Json(req): Json<AddScheduleRequest>,
|
|
) -> Result<(StatusCode, Json<ScheduleView>), ProblemDetails> {
|
|
let client = queue(&state).map_err(refused)?;
|
|
let live = live(&state).await?;
|
|
let now = unix_now();
|
|
let store = open_or_create_definitions(&client)
|
|
.await
|
|
.map_err(|e| problem(e.into()))?;
|
|
let added = commit(&store, |all| {
|
|
let id = mint_id(all)?;
|
|
let d = req.clone().into_definition(id, now, &live)?;
|
|
all.schedules.push(d.clone());
|
|
Ok(d)
|
|
})
|
|
.await
|
|
.map_err(problem)?;
|
|
tracing::info!(id = %added.id, agents = ?added.agents, "added a schedule");
|
|
project_all(&client, &store, &added.agents).await;
|
|
Ok((StatusCode::CREATED, Json(ScheduleView::new(&added, now))))
|
|
}
|
|
|
|
#[utoipa::path(
|
|
patch,
|
|
path = "/api/schedules/{id}",
|
|
params(("id" = String, Path, description = "schedule to change")),
|
|
request_body = EditScheduleRequest,
|
|
responses(
|
|
(status = 200, description = "the schedule as stored", body = ScheduleView),
|
|
(status = 400, description = "the name or body is empty, the interval is 0, an \
|
|
added agent is not declared on any hive, a removed agent is not a target, or \
|
|
no agent would remain (problem+json)", body = String),
|
|
(status = 404, description = "there is no schedule with this id (problem+json)",
|
|
body = String),
|
|
(status = 503, description = "no swarm queue is configured on this host, or it is \
|
|
not connected (problem+json)", body = String),
|
|
(status = 500, description = "the schedules could not be written (problem+json)",
|
|
body = String),
|
|
),
|
|
tag = "schedules"
|
|
)]
|
|
pub(crate) async fn edit_schedule(
|
|
State(state): State<AppState>,
|
|
Path(id): Path<String>,
|
|
Json(req): Json<EditScheduleRequest>,
|
|
) -> Result<Json<ScheduleView>, ProblemDetails> {
|
|
let client = queue(&state).map_err(refused)?;
|
|
let live = live(&state).await?;
|
|
let store = open_or_create_definitions(&client)
|
|
.await
|
|
.map_err(|e| problem(e.into()))?;
|
|
let (edited, touched) = commit(&store, |all| Ok(req.clone().apply(all, &id, &live)?))
|
|
.await
|
|
.map_err(problem)?;
|
|
tracing::info!(%id, "edited a schedule");
|
|
project_all(&client, &store, &touched).await;
|
|
Ok(Json(ScheduleView::new(&edited, unix_now())))
|
|
}
|
|
|
|
#[utoipa::path(
|
|
delete,
|
|
path = "/api/schedules/{id}",
|
|
params(("id" = String, Path, description = "schedule to remove")),
|
|
responses(
|
|
(status = 204, description = "removed; it fires no more"),
|
|
(status = 404, description = "there is no schedule with this id (problem+json)",
|
|
body = String),
|
|
(status = 503, description = "no swarm queue is configured on this host, or it is \
|
|
not connected (problem+json)", body = String),
|
|
(status = 500, description = "the schedules could not be written (problem+json)",
|
|
body = String),
|
|
),
|
|
tag = "schedules"
|
|
)]
|
|
pub(crate) async fn remove_schedule(
|
|
State(state): State<AppState>,
|
|
Path(id): Path<String>,
|
|
) -> Result<StatusCode, ProblemDetails> {
|
|
let client = queue(&state).map_err(refused)?;
|
|
let store = open_or_create_definitions(&client)
|
|
.await
|
|
.map_err(|e| problem(e.into()))?;
|
|
let touched = commit(&store, |all| Ok(remove(all, &id)?))
|
|
.await
|
|
.map_err(problem)?;
|
|
tracing::info!(%id, "removed a schedule");
|
|
project_all(&client, &store, &touched).await;
|
|
Ok(StatusCode::NO_CONTENT)
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
fn live() -> BTreeSet<String> {
|
|
["atlas", "iris", "scribe"].map(str::to_owned).into()
|
|
}
|
|
|
|
fn standup() -> Definition {
|
|
Definition {
|
|
id: "abc".to_owned(),
|
|
name: "standup".to_owned(),
|
|
agents: ["atlas", "iris"].map(str::to_owned).into(),
|
|
body: "post the standup".to_owned(),
|
|
anchor_unix: 1000,
|
|
interval_seconds: Some(3600),
|
|
paused: false,
|
|
}
|
|
}
|
|
|
|
fn all() -> Definitions {
|
|
Definitions {
|
|
schedules: vec![standup()],
|
|
unreadable: Vec::new(),
|
|
}
|
|
}
|
|
|
|
fn edit(json: &str) -> EditScheduleRequest {
|
|
serde_json::from_str(json).expect("decodes")
|
|
}
|
|
|
|
/// Each agent's key carries the schedules naming it, under the same id,
|
|
/// and nothing of the name or the other agents.
|
|
#[test]
|
|
fn an_agents_key_holds_the_schedules_naming_it() {
|
|
let mut defs = all();
|
|
let mut other = standup();
|
|
"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());
|
|
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()
|
|
);
|
|
let encoded = serde_json::to_string(&atlas).unwrap();
|
|
assert!(!encoded.contains("\"name\""), "{encoded}");
|
|
assert!(!encoded.contains("iris"), "{encoded}");
|
|
}
|
|
|
|
/// A body or name edit leaves anchor and interval, and so the fire
|
|
/// times, where they were, and keeps the id.
|
|
#[test]
|
|
fn a_text_edit_leaves_the_cadence_and_id_alone() {
|
|
let mut defs = all();
|
|
let (out, touched) = edit(r#"{"body":"retro","name":"retro"}"#)
|
|
.apply(&mut defs, "abc", &live())
|
|
.unwrap();
|
|
assert_eq!(
|
|
(out.id.as_str(), out.body.as_str(), out.name.as_str()),
|
|
("abc", "retro", "retro")
|
|
);
|
|
assert_eq!((out.anchor_unix, out.interval_seconds), (1000, Some(3600)));
|
|
assert_eq!(touched, standup().agents);
|
|
}
|
|
|
|
/// Adding and removing targets keeps the id, and every agent on either
|
|
/// side of the change is rewritten.
|
|
#[test]
|
|
fn the_agent_set_changes_without_a_new_id() {
|
|
let mut defs = all();
|
|
let (out, touched) = edit(r#"{"add_agents":["scribe"],"remove_agents":["iris"]}"#)
|
|
.apply(&mut defs, "abc", &live())
|
|
.unwrap();
|
|
assert_eq!(out.id, "abc");
|
|
assert_eq!(out.agents, ["atlas", "scribe"].map(str::to_owned).into());
|
|
assert_eq!(
|
|
touched,
|
|
["atlas", "iris", "scribe"].map(str::to_owned).into()
|
|
);
|
|
assert_eq!(defs.schedules, vec![out]);
|
|
}
|
|
|
|
#[test]
|
|
fn bad_edits_are_refused_before_anything_changes() {
|
|
for (json, id) in [
|
|
(r#"{"interval_seconds":0}"#, "abc"),
|
|
(r#"{"body":" "}"#, "abc"),
|
|
(r#"{"name":""}"#, "abc"),
|
|
(r#"{"add_agents":["nobody"]}"#, "abc"),
|
|
(r#"{"remove_agents":["scribe"]}"#, "abc"),
|
|
(r#"{"remove_agents":["atlas","iris"]}"#, "abc"),
|
|
(r#"{"body":"x"}"#, "nope"),
|
|
] {
|
|
let mut defs = all();
|
|
assert!(edit(json).apply(&mut defs, id, &live()).is_err(), "{json}");
|
|
assert_eq!(defs, all(), "{json}");
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn a_null_interval_makes_a_one_shot_and_an_absent_one_keeps_it() {
|
|
let mut defs = all();
|
|
let (kept, _) = edit(r#"{"paused":true}"#)
|
|
.apply(&mut defs, "abc", &live())
|
|
.unwrap();
|
|
assert_eq!(kept.interval_seconds, Some(3600));
|
|
assert!(kept.paused);
|
|
let (once, _) = edit(r#"{"interval_seconds":null}"#)
|
|
.apply(&mut defs, "abc", &live())
|
|
.unwrap();
|
|
assert_eq!(once.interval_seconds, None);
|
|
}
|
|
|
|
#[test]
|
|
fn removing_drops_only_the_named_schedule() {
|
|
let mut defs = all();
|
|
assert_eq!(
|
|
remove(&mut defs, "zzz"),
|
|
Err(Refusal::NoSuchSchedule("zzz".to_owned()))
|
|
);
|
|
assert_eq!(defs, all());
|
|
assert_eq!(remove(&mut defs, "abc").unwrap(), standup().agents);
|
|
assert!(defs.schedules.is_empty());
|
|
}
|
|
|
|
#[test]
|
|
fn an_add_needs_a_name_and_declared_agents_and_anchors_at_now() {
|
|
let add = |json: &str| {
|
|
serde_json::from_str::<AddScheduleRequest>(json)
|
|
.unwrap()
|
|
.into_definition("id".to_owned(), 42, &live())
|
|
};
|
|
let d = add(r#"{"name":"n","agents":["iris","atlas","iris"],"body":"hi"}"#).unwrap();
|
|
assert_eq!(
|
|
(d.anchor_unix, d.interval_seconds, d.paused),
|
|
(42, None, false)
|
|
);
|
|
assert_eq!(d.agents, ["atlas", "iris"].map(str::to_owned).into());
|
|
assert!(add(r#"{"name":" ","agents":["iris"],"body":"hi"}"#).is_err());
|
|
assert!(add(r#"{"name":"n","agents":[],"body":"hi"}"#).is_err());
|
|
assert_eq!(
|
|
add(r#"{"name":"n","agents":["nobody"],"body":"hi"}"#),
|
|
Err(Refusal::NotDeclared("nobody".to_owned()))
|
|
);
|
|
assert!(add(r#"{"name":"n","agents":["Not An Ident"],"body":"hi"}"#).is_err());
|
|
}
|
|
|
|
#[test]
|
|
fn next_fire_is_the_next_grid_slot_or_none() {
|
|
let mut d = standup();
|
|
assert_eq!(d.next_fire(999), Some(1000));
|
|
assert_eq!(d.next_fire(1000), Some(4600));
|
|
assert_eq!(d.next_fire(4601), Some(8200));
|
|
d.paused = true;
|
|
assert_eq!(d.next_fire(999), None);
|
|
d.paused = false;
|
|
d.interval_seconds = None;
|
|
assert_eq!(d.next_fire(999), Some(1000));
|
|
assert_eq!(d.next_fire(1000), None);
|
|
}
|
|
|
|
#[test]
|
|
fn a_view_shows_the_wire_shape() {
|
|
let v = serde_json::to_value(ScheduleView::new(&standup(), 1500)).unwrap();
|
|
assert_eq!(
|
|
v,
|
|
serde_json::json!({
|
|
"id": "abc",
|
|
"name": "standup",
|
|
"agents": ["atlas", "iris"],
|
|
"body": "post the standup",
|
|
"anchor_unix": 1000,
|
|
"interval_seconds": 3600,
|
|
"paused": false,
|
|
"next_fire_unix": 4600,
|
|
})
|
|
);
|
|
}
|
|
|
|
/// 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]
|
|
fn minted_ids_are_letters_and_avoid_taken_ones() {
|
|
let defs = all();
|
|
for _ in 0..100 {
|
|
let id = mint_id(&defs).unwrap();
|
|
assert_eq!(id.len(), ID_LEN);
|
|
assert!(id.chars().all(|c| c.is_ascii_lowercase()), "{id}");
|
|
assert_ne!(id, "abc");
|
|
}
|
|
}
|
|
}
|