vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

notify flows: refuse unknown placeholders and require the revision read on every save

Browse source
This commit is contained in:
atlas 2026-10-24 04:38:17 +02:00 • committed by mara
commit ada364a482
7 changed files with 203 additions and 57 deletions
Download patch file Download diff file Expand all files Collapse all files

128
swarm-controller/src/notify_flows.rs
View file

@ -33,9 +33,9 @@ const MAX_ATTEMPTS: usize = 5;
/// What `GET` and a successful `PUT` answer.
#[derive(Debug, Clone, PartialEq, Serialize, ToSchema)]
pub(crate) struct FlowsView {
/// The revision of the stored document, to send back in a `PUT`. `null`
/// when none is stored.
revision: Option<u64>,
/// The revision of the stored document, to send back in a `PUT`; `0` when
/// none is stored.
revision: u64,
/// The ids of the built-in flows that are switched off.
disabled_builtins: Vec<String>,
/// The built-in flows this build ships, all of them, switched off or not.
@ -48,10 +48,10 @@ pub(crate) struct FlowsView {
#[derive(Debug, Clone, Deserialize, ToSchema)]
#[serde(deny_unknown_fields)]
pub(crate) struct PutFlows {
/// The `revision` the client read. When another write has landed since,
/// the `PUT` is refused. Left out, it replaces whatever is stored.
#[serde(default)]
revision: Option<u64>,
/// The `revision` the client read, `0` for "no document stored". The
/// `PUT` is refused when the stored revision is another, which includes a
/// document written since a client read `0`. A KV revision is never `0`.
revision: u64,
/// The ids of the built-in flows to switch off.
#[serde(default)]
disabled_builtins: Vec<String>,
@ -61,7 +61,7 @@ pub(crate) struct PutFlows {
}
impl FlowsView {
fn of(doc: Document, revision: Option<u64>) -> Self {
fn of(doc: Document, revision: u64) -> Self {
Self {
revision,
disabled_builtins: doc.disabled_builtins,
@ -120,7 +120,8 @@ fn queue_problem(e: &anyhow::Error) -> ProblemDetails {
error_problem(wanted_error_status(e), &format!("{e:#}"))
}
/// Read the flows.
/// Read the flows. Creates the `notify-flows` bucket when it is missing, as a
/// `PUT` does, so the router finds it empty rather than absent.
#[utoipa::path(
get,
path = "/api/notify/flows",
@ -144,8 +145,11 @@ pub(crate) async fn get_flows(
let store = open(&client).await.map_err(|e| queue_problem(&e))?;
let (stored, revision) = stored(&store).await.map_err(|e| queue_problem(&e))?;
match stored {
Stored::Nothing => Ok(Json(FlowsView::of(Document::default(), revision))),
Stored::Document(doc) => Ok(Json(FlowsView::of(doc, revision))),
Stored::Nothing => Ok(Json(FlowsView::of(
Document::default(),
revision.unwrap_or(0),
))),
Stored::Document(doc) => Ok(Json(FlowsView::of(doc, revision.unwrap_or(0)))),
Stored::Unreadable(e) => Err(unreadable(&e)),
}
}
@ -173,7 +177,7 @@ fn unreadable(why: &DecodeError) -> ProblemDetails {
letters, digits, `-` or `_`, a source event the router is not sent, a condition \
or recipient field that is no event field, or an upsert with no text \
(problem+json)", body = String),
(status = 409, description = "another write landed since `revision`, or the stored \
(status = 409, description = "the stored document is not at `revision` (another write landed, or one now exists where the client read none), or the stored \
document is a version this build does not read, which a `PUT` would erase \
(problem+json)", body = String),
(status = 503, description = "no swarm queue is configured on this host, or it is \
@ -204,12 +208,12 @@ pub(crate) async fn put_flows(
if let Stored::Unreadable(why @ DecodeError::UnknownVersion(_)) = &current {
return Err(unreadable(why));
}
if put.revision.is_some() && put.revision != revision {
if put.revision != revision.unwrap_or(0) {
return Err(error_problem(
StatusCode::CONFLICT,
&format!(
"the flows changed since revision {}; read them again",
put.revision.unwrap_or_default()
"the flows are not at revision {}; read them again",
put.revision
),
));
}
@ -222,7 +226,7 @@ pub(crate) async fn put_flows(
custom = doc.flows.len(),
"set the notification router's flows"
);
return Ok(Json(FlowsView::of(doc, Some(written))));
return Ok(Json(FlowsView::of(doc, written)));
}
}
Err(queue_problem(&anyhow::anyhow!(
@ -268,7 +272,7 @@ mod tests {
}
}
fn body(revision: Option<u64>, disabled: &[&str], flows: Vec<Flow>) -> Json<PutFlows> {
fn body(revision: u64, disabled: &[&str], flows: Vec<Flow>) -> Json<PutFlows> {
Json(PutFlows {
revision,
disabled_builtins: disabled.iter().map(|s| (*s).to_owned()).collect(),
@ -284,17 +288,18 @@ mod tests {
// Nothing stored: the built-ins alone, none off, no revision.
let Json(view) = get_flows(State(state.clone())).await.expect("get");
assert_eq!(view.revision, None);
assert_eq!(view.revision, 0);
assert!(view.disabled_builtins.is_empty() && view.flows.is_empty());
assert_eq!(view.builtins.len(), 3);
let Json(saved) = put_flows(
State(state.clone()),
Ok(body(None, &["unassigned"], vec![custom("triage")])),
Ok(body(0, &["unassigned"], vec![custom("triage")])),
)
.await
.expect("put");
let revision = saved.revision.expect("a revision");
let revision = saved.revision;
assert_ne!(revision, 0);
let Json(read) = get_flows(State(state.clone())).await.expect("get");
assert_eq!(read, saved);
assert_eq!(read.disabled_builtins, ["unassigned"]);
@ -313,11 +318,11 @@ mod tests {
// A refused save changes nothing: an orphan node, and a built-in that does not exist.
let mut orphan = custom("orphan");
orphan.edges.clear();
let refused = put_flows(State(state.clone()), Ok(body(None, &[], vec![orphan])))
let refused = put_flows(State(state.clone()), Ok(body(revision, &[], vec![orphan])))
.await
.expect_err("an orphan node");
assert_eq!(refused.status, Some(StatusCode::BAD_REQUEST), "{refused:?}");
let unknown = put_flows(State(state.clone()), Ok(body(None, &["nope"], vec![])))
let unknown = put_flows(State(state.clone()), Ok(body(revision, &["nope"], vec![])))
.await
.expect_err("no such built-in");
assert_eq!(unknown.status, Some(StatusCode::BAD_REQUEST), "{unknown:?}");
@ -325,17 +330,14 @@ mod tests {
assert_eq!(after, saved);
// A stale revision is refused; the current one lands.
let outdated = put_flows(
State(state.clone()),
Ok(body(Some(revision + 100), &[], vec![])),
)
.await
.expect_err("stale");
let outdated = put_flows(State(state.clone()), Ok(body(revision + 100, &[], vec![])))
.await
.expect_err("stale");
assert_eq!(outdated.status, Some(StatusCode::CONFLICT), "{outdated:?}");
let Json(next) = put_flows(State(state.clone()), Ok(body(Some(revision), &[], vec![])))
let Json(next) = put_flows(State(state.clone()), Ok(body(revision, &[], vec![])))
.await
.expect("current revision");
assert!(next.revision > Some(revision));
assert!(next.revision > revision);
assert!(next.flows.is_empty());
}
@ -359,7 +361,7 @@ mod tests {
.await
.expect_err("version 2");
assert_eq!(got.status, Some(StatusCode::CONFLICT), "{got:?}");
let refused = put_flows(State(state.clone()), Ok(body(None, &[], vec![])))
let refused = put_flows(State(state.clone()), Ok(body(0, &[], vec![])))
.await
.expect_err("version 2 is kept");
assert_eq!(refused.status, Some(StatusCode::CONFLICT), "{refused:?}");
@ -371,14 +373,74 @@ mod tests {
assert_eq!(&kept[..], br#"{"version":2,"flows":[]}"#);
// Control: garbage that names no version is replaced by a save.
store
let garbage = store
.put(notify_flows::KEY, b"garbage".to_vec().into())
.await
.expect("seed");
let _ = put_flows(State(state.clone()), Ok(body(None, &["closed"], vec![])))
let _ = put_flows(State(state.clone()), Ok(body(garbage, &["closed"], vec![])))
.await
.expect("replaces garbage");
let Json(view) = get_flows(State(state)).await.expect("readable again");
assert_eq!(view.disabled_builtins, ["closed"]);
}
/// A page that read no document cannot erase one written since: its
/// `revision: 0` means "create", and the document exists by then.
#[tokio::test]
async fn a_page_that_read_nothing_cannot_erase_a_flow_written_after() {
let nats = TestNats::start();
let client = nats.client().await;
let state = crate::agent_limits::tests::state(&client);
let Json(page) = get_flows(State(state.clone())).await.expect("get");
assert_eq!(page.revision, 0);
// The API writes a custom flow while the page is open.
let Json(api) = put_flows(
State(state.clone()),
Ok(body(0, &[], vec![custom("triage")])),
)
.await
.expect("the API's write");
// The page's first switch carries what it read.
let erased = put_flows(
State(state.clone()),
Ok(body(page.revision, &["closed"], vec![])),
)
.await
.expect_err("the page read nothing, and a document exists");
assert_eq!(erased.status, Some(StatusCode::CONFLICT), "{erased:?}");
let Json(after) = get_flows(State(state.clone())).await.expect("get");
assert_eq!(after, api);
assert_eq!(after.flows[0].id, "triage");
// Control: re-read, the same switch lands and keeps the flow.
let Json(switched) = put_flows(
State(state),
Ok(body(after.revision, &["closed"], after.flows.clone())),
)
.await
.expect("the switch with the revision it read");
assert_eq!(switched.disabled_builtins, ["closed"]);
assert_eq!(switched.flows[0].id, "triage");
}
#[tokio::test]
async fn a_template_naming_no_event_field_is_refused_on_save() {
let nats = TestNats::start();
let client = nats.client().await;
let state = crate::agent_limits::tests::state(&client);
let mut typo = custom("typo");
if let NodeKind::Todo(sink) = &mut typo.nodes[1].kind {
sink.key = "forge:{repo}#{numbr}:triage".to_owned();
}
let refused = put_flows(State(state.clone()), Ok(body(0, &[], vec![typo])))
.await
.expect_err("{numbr} is no field");
assert_eq!(refused.status, Some(StatusCode::BAD_REQUEST), "{refused:?}");
// Control: the same flow spelled right is stored.
let _ = put_flows(State(state), Ok(body(0, &[], vec![custom("typo")])))
.await
.expect("{number} is a field");
}
}