diff --git a/docs/swarm/notify-router.md b/docs/swarm/notify-router.md index 74cbf779..dea37de4 100644 --- a/docs/swarm/notify-router.md +++ b/docs/swarm/notify-router.md @@ -83,8 +83,9 @@ switched off, and the custom flows. Only the controller writes it, through compare-and-set. The controller answers 400 unless every flow passes the tree check above and reads only what exists: a source event from `issues`, `issue_assign`, `pull_request` or `pull_request_assign`, and condition and -recipient fields from the event's field list. A `PUT` that carries the -`revision` it read gets 409 when another write landed since. Swarm UI's +recipient fields from the event's field list. The same goes for the `{field}` +placeholders in a todo's key and text. A `PUT` must carry the `revision` it read, `0` for no +document, and gets 409 when the stored revision is another. Swarm UI's [`/notifications`](ui.md#notification-flows) page shows the document and switches built-ins. diff --git a/docs/swarm/ui.md b/docs/swarm/ui.md index 106aa794..fb448081 100644 --- a/docs/swarm/ui.md +++ b/docs/swarm/ui.md @@ -226,8 +226,10 @@ restores it. Each **custom** flow shows its name and whether it's on. Every flow draws as an indented tree: the forge events it starts from, the conditions under it, and the todo each branch ends in. -The page edits nothing else. Add custom flows with `PUT /api/notify/flows`. -A switch sends the whole flows document with the revision the page read, so if +A switch sends the whole flows document with the revision the page read, `0` if +it read no document. The controller refuses it when the stored revision is +another, including a document written since the page read none, and the page +shows why. Reload to switch against the current document. another change landed since, the controller refuses it and the page shows why. ### Sending a message to an agent diff --git a/frontend/packages/swarm-ui/src/pages/notifications/FlowTypes.ts b/frontend/packages/swarm-ui/src/pages/notifications/FlowTypes.ts index 8256ffee..2cb58d34 100644 --- a/frontend/packages/swarm-ui/src/pages/notifications/FlowTypes.ts +++ b/frontend/packages/swarm-ui/src/pages/notifications/FlowTypes.ts @@ -31,7 +31,7 @@ export interface Flow { } export interface FlowsView { - revision: number | null; + revision: number; disabled_builtins: string[]; builtins: Flow[]; flows: Flow[]; diff --git a/frontend/packages/swarm-ui/src/pages/notifications/NotificationsPage.tsx b/frontend/packages/swarm-ui/src/pages/notifications/NotificationsPage.tsx index cf7510b8..aae28195 100644 --- a/frontend/packages/swarm-ui/src/pages/notifications/NotificationsPage.tsx +++ b/frontend/packages/swarm-ui/src/pages/notifications/NotificationsPage.tsx @@ -58,7 +58,7 @@ export function NotificationsPage() { const [actionError, setActionError] = useState(null); // Switch one built-in on or off: the whole document goes back with the - // revision this page read, so a change made elsewhere since is refused + // revision this page read (0 for none), so a change made elsewhere since is refused // rather than overwritten. async function setBuiltin(id: string, enabled: boolean) { if (!view) return; diff --git a/swarm-controller/src/notify_flows.rs b/swarm-controller/src/notify_flows.rs index 9f303ee3..3a97a521 100644 --- a/swarm-controller/src/notify_flows.rs +++ b/swarm-controller/src/notify_flows.rs @@ -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, + /// 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, /// 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, + /// 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, @@ -61,7 +61,7 @@ pub(crate) struct PutFlows { } impl FlowsView { - fn of(doc: Document, revision: Option) -> 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(_)) = ¤t { 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, disabled: &[&str], flows: Vec) -> Json { + fn body(revision: u64, disabled: &[&str], flows: Vec) -> Json { 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"); + } } diff --git a/swarm-notify-router/src/flow.rs b/swarm-notify-router/src/flow.rs index f0a262d8..b32f3afa 100644 --- a/swarm-notify-router/src/flow.rs +++ b/swarm-notify-router/src/flow.rs @@ -67,14 +67,25 @@ fn emit(flow: &Flow, node: &Node, sink: &TodoSink, fields: &Fields) -> Vec Vec::new(), }, }; + let rendered = expand(&sink.key, fields).and_then(|key| { + let summary = match sink.op { + Op::Upsert => expand(&sink.text, fields)?, + Op::Clear => String::new(), + }; + Ok((key, summary)) + }); + let (key, summary) = match rendered { + Ok(done) => done, + Err(why) => { + tracing::warn!(flow = %flow.id, node = %node.id, %why, "notify router: a todo node names a placeholder the event lacks; nothing sent"); + return Vec::new(); + } + }; let op = RouterTodoOp { op: sink.op, - key: expand(&sink.key, fields), + key, prefix: sink.op == Op::Clear && sink.prefix, - summary: match sink.op { - Op::Upsert => expand(&sink.text, fields), - Op::Clear => String::new(), - }, + summary, source: format!("router:{}", flow.id), }; recipients @@ -96,9 +107,9 @@ fn holds(condition: &Condition, fields: &Fields) -> bool { } } -/// `template` with each `{field}` replaced by that text field's value. A list -/// field is joined with `, `; an unknown name is left as written. -pub fn expand(template: &str, fields: &Fields) -> String { +/// `template` with each `{field}` replaced by that field's value, a list +/// joined with `, `. A name that is no field is an error naming it. +pub fn expand(template: &str, fields: &Fields) -> Result { let mut out = String::with_capacity(template.len()); let mut rest = template; while let Some(open) = rest.find('{') { @@ -106,22 +117,18 @@ pub fn expand(template: &str, fields: &Fields) -> String { let after = &rest[open + 1..]; let Some(close) = after.find('}') else { out.push_str(&rest[open..]); - return out; + return Ok(out); }; let name = &after[..close]; match fields.get(name) { Some(Value::Text(v)) => out.push_str(v), Some(Value::List(items)) => out.push_str(&items.join(", ")), - None => { - out.push('{'); - out.push_str(name); - out.push('}'); - } + None => return Err(format!("{{{name}}} is no event field")), } rest = &after[close + 1..]; } out.push_str(rest); - out + Ok(out) } #[cfg(test)] @@ -288,11 +295,36 @@ mod tests { } #[test] - fn a_placeholder_expands_to_its_field_and_an_unknown_one_stays() { + fn a_placeholder_expands_to_its_field_and_an_unknown_one_is_an_error() { let fields = event("assigned", &["a", "b"], &[], &[]); assert_eq!( - expand("{repo}#{number} {assignees_added} {nope} {", &fields), - "hyperhive/hyperhive#7 a, b {nope} {" + expand("{repo}#{number} {assignees_added} {", &fields), + Ok("hyperhive/hyperhive#7 a, b {".to_owned()) + ); + assert_eq!( + expand("x {nope}", &fields), + Err("{nope} is no event field".to_owned()) ); } + + /// A sink naming a placeholder the event lacks sends nothing, rather + /// than a todo under a literal `{name}` key. + #[test] + fn a_sink_with_an_unknown_placeholder_emits_nothing() { + let mut f = flow("assigned"); + f.nodes[2].kind = NodeKind::Todo(TodoSink { + to: Recipients::Agent("triage".to_owned()), + op: Op::Upsert, + key: "forge:{repo}#{numbr}:triage".to_owned(), + prefix: false, + text: "t".to_owned(), + }); + let fields = event("assigned", &["atlas"], &[], &[]); + assert!(evaluate(&f, "issue_assign", &fields).is_empty()); + // Control: spelled right, the same flow emits. + if let NodeKind::Todo(s) = &mut f.nodes[2].kind { + s.key = "forge:{repo}#{number}:triage".to_owned(); + } + assert_eq!(evaluate(&f, "issue_assign", &fields).len(), 1); + } } diff --git a/swarm-wire/src/notify_flows.rs b/swarm-wire/src/notify_flows.rs index df019281..4d78682f 100644 --- a/swarm-wire/src/notify_flows.rs +++ b/swarm-wire/src/notify_flows.rs @@ -291,6 +291,21 @@ fn is_field(name: &str) -> bool { FIELDS.contains(&name) } +/// The `{name}` placeholders in `template`, in order. A `{` with no `}` after +/// it is plain text. +#[must_use] +pub fn placeholders(template: &str) -> Vec<&str> { + let mut names = Vec::new(); + let mut rest = template; + while let Some(open) = rest.find('{') { + let after = &rest[open + 1..]; + let Some(close) = after.find('}') else { break }; + names.push(&after[..close]); + rest = &after[close + 1..]; + } + names +} + impl Flow { /// Check the flow is a tree within the caps, and that every node reads /// only what exists. @@ -435,6 +450,11 @@ impl TodoSink { if self.text.len() > MAX_TEXT { return Err(format!("text is longer than {MAX_TEXT} bytes")); } + for (what, template) in [("key", &self.key), ("text", &self.text)] { + if let Some(bad) = placeholders(template).into_iter().find(|n| !is_field(n)) { + return Err(format!("{what} has {{{bad}}}, which is no event field")); + } + } match self.op { Op::Upsert if self.text.trim().is_empty() => Err("an upsert with no text".to_owned()), Op::Upsert if self.prefix => Err("prefix is for a clear".to_owned()), @@ -676,6 +696,35 @@ mod tests { assert_eq!(custom("c").validate(), Ok(())); } + #[test] + fn a_placeholder_must_name_an_event_field() { + let with = |key: &str, text: &str| { + let mut f = custom("c"); + f.nodes[2].kind = NodeKind::Todo(TodoSink { + to: Recipients::Agent("triage".to_owned()), + op: Op::Upsert, + key: key.to_owned(), + prefix: false, + text: text.to_owned(), + }); + f.validate() + }; + assert_eq!( + with("forge:{repo}#{number}:triage", "{title} {url}"), + Ok(()) + ); + assert!( + with("forge:{repo}#{numbr}:triage", "t") + .unwrap_err() + .contains("numbr") + ); + assert!(with("k", "see {ulr}").unwrap_err().contains("ulr")); + assert!(with("k{}", "t").is_err(), "an empty name"); + // An unclosed brace is text. + assert_eq!(with("k", "set {x"), Ok(())); + assert_eq!(placeholders("a{b}c{d} {e"), ["b", "d"]); + } + #[test] fn a_sink_must_say_what_its_op_needs() { let sink = |op, prefix, text: &str| {