From mboxrd@z Thu Jan 1 00:00:00 1970 Return-Path: Received: from gate001.proxmox.com (gate001.proxmox.com [45.144.208.40]) by lore.proxmox.com (Postfix) with ESMTPS id 24DB01FF0C2 for ; Thu, 22 Oct 2026 16:22:43 +0200 (CEST) Received: from gate001.proxmox.com (localhost.localdomain [127.0.0.1]) by gate001.proxmox.com (Proxmox) with ESMTP id 44CB1216D7; Thu, 22 Oct 2026 16:22:15 +0200 (CEST) From: Hannes Laimer To: pve-devel@lists.proxmox.com Subject: [PATCH pve-cluster 03/12] rust: notify: add the wire protocol Date: Thu, 22 Oct 2026 16:21:56 +0200 Message-ID: <20261022142205.930536-4-h.laimer@proxmox.com> X-Mailer: git-send-email 2.47.3 In-Reply-To: <20261022142205.930536-1-h.laimer@proxmox.com> References: <20261022142205.930536-1-h.laimer@proxmox.com> MIME-Version: 1.0 Content-Transfer-Encoding: 8bit X-Bm-Milter-Handled: 55990f41-d878-4baa-be0a-ee34c49e34d2 X-Bm-Transport-Timestamp: 1792678930784 X-SPAM-LEVEL: Spam detection results: 0 AWL 0.349 Adjusted score from AWL reputation of From: address DMARC_MISSING 0.1 Missing DMARC policy KAM_DMARC_STATUS 0.01 Test Rule for DKIM or SPF Failure with Strict Alignment (newer systems) RCVD_IN_DNSWL_MED -2.3 Sender listed at https://www.dnswl.org/, medium trust SPF_HELO_NONE 0.001 SPF: HELO does not publish an SPF Record SPF_PASS -0.001 SPF: sender matches SPF record Message-ID-Hash: JDXOQI33JYHERH32L7UUYIBXNENHMUIH X-Message-ID-Hash: JDXOQI33JYHERH32L7UUYIBXNENHMUIH X-MailFrom: h.laimer@proxmox.com X-Mailman-Rule-Misses: dmarc-mitigation; no-senders; approved; loop; banned-address; emergency; member-moderation; nonmember-moderation; administrivia; implicit-dest; max-recipients; max-size; news-moderation; no-subject; digests; suspicious-header X-Mailman-Version: 3.3.10 Precedence: list List-Id: Proxmox VE development discussion List-Help: List-Owner: List-Post: List-Subscribe: List-Unsubscribe: Requests, replies and events are all JSON objects, one per line. So a client needs a single parser, and the stream stays readable as text. A request is a command with its arguments, and a reply carries either a result or an error. An event carries a mutation's type, the path, the rename target and the captured values per subscription name. Signed-off-by: Hannes Laimer --- src/rust/pmxcfs-notify/src/lib.rs | 8 ++ src/rust/pmxcfs-notify/src/protocol.rs | 185 +++++++++++++++++++++++++ 2 files changed, 193 insertions(+) create mode 100644 src/rust/pmxcfs-notify/src/protocol.rs diff --git a/src/rust/pmxcfs-notify/src/lib.rs b/src/rust/pmxcfs-notify/src/lib.rs index 889fe50..d35d3c5 100644 --- a/src/rust/pmxcfs-notify/src/lib.rs +++ b/src/rust/pmxcfs-notify/src/lib.rs @@ -1,5 +1,12 @@ //! Change notification socket for pmxcfs. //! +//! Clients connect to a Unix stream socket and exchange newline delimited JSON objects. A request +//! is `{"command": "...", "args": {...}}` and is answered with `{"ok": }` or +//! `{"error": "..."}`. Once a client has sent `subscribe` with named path templates, every +//! matching memdb mutation is pushed as `{"event": {...}}` with the event type, the path, the +//! rename target, and per subscription name the parameter sets captured from the path and the +//! rename target. +//! //! A template is a memdb path whose components may each hold one `{name}` placeholder for all or //! part of that component. The last component may instead be `{name...}` for the rest of the path. //! `nodes/{node}/qemu-server/{vmid}.conf` captures node and vmid, and `{path...}` matches every @@ -7,6 +14,7 @@ #![allow(dead_code)] +mod protocol; mod template; /// Events and templates name memdb paths without a leading slash. diff --git a/src/rust/pmxcfs-notify/src/protocol.rs b/src/rust/pmxcfs-notify/src/protocol.rs new file mode 100644 index 0000000..d28c9dc --- /dev/null +++ b/src/rust/pmxcfs-notify/src/protocol.rs @@ -0,0 +1,185 @@ +use std::collections::BTreeMap; + +use serde::{Deserialize, Serialize}; +use serde_json::Value; + +use crate::template::{Captures, Template}; + +pub(crate) const PROTOCOL_VERSION: u32 = 1; + +/// Keyed by subscription name, with one capture set per side of the event that matched. +pub(crate) type Params<'a> = BTreeMap<&'a str, Vec>>; + +#[non_exhaustive] +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "lowercase")] +pub enum EventKind { + Create, + Write, + Mtime, + Rename, + Delete, + Mkdir, + Resync, +} + +#[derive(Serialize)] +pub(crate) struct Event<'a> { + #[serde(rename = "type")] + pub kind: EventKind, + #[serde(skip_serializing_if = "Option::is_none")] + pub path: Option<&'a str>, + #[serde(skip_serializing_if = "Option::is_none")] + pub to: Option<&'a str>, + #[serde(skip_serializing_if = "Option::is_none")] + pub params: Option>, +} + +/// Everything the server sends. The variant name is the single key of the JSON object on the wire. +#[derive(Serialize)] +#[serde(rename_all = "lowercase")] +pub(crate) enum Message<'a> { + Ok(Value), + Error(String), + Event(Event<'a>), +} + +impl Message<'static> { + pub(crate) fn resync() -> Self { + Message::Event(Event { + kind: EventKind::Resync, + path: None, + to: None, + params: None, + }) + } +} + +impl Message<'_> { + pub(crate) fn line(&self) -> String { + let mut line = serde_json::to_string(self).expect("message serialization cannot fail"); + line.push('\n'); + line + } +} + +#[derive(Debug, PartialEq, Eq, Deserialize)] +#[serde(tag = "command", content = "args", rename_all = "lowercase")] +pub(crate) enum Command { + Hello, + Subscribe { + #[serde(default)] + patterns: BTreeMap, + }, + Unsubscribe, +} + +#[cfg(test)] +mod tests { + use super::*; + + fn parse(line: &str) -> Result { + serde_json::from_str(line) + } + + fn template(text: &str) -> Template { + text.parse().unwrap() + } + + #[test] + fn parses_commands() { + assert_eq!(parse(r#"{"command":"hello"}"#).unwrap(), Command::Hello); + assert_eq!( + parse( + r#"{"command":"subscribe","args":{"patterns":{"guest":"nodes/{node}/qemu-server/{vmid}.conf","dc":"/datacenter.cfg"}}}"# + ) + .unwrap(), + Command::Subscribe { + patterns: BTreeMap::from([ + ("dc".to_owned(), template("datacenter.cfg")), + ( + "guest".to_owned(), + template("nodes/{node}/qemu-server/{vmid}.conf") + ), + ]), + } + ); + assert_eq!( + parse(r#"{"command":"subscribe","args":{}}"#).unwrap(), + Command::Subscribe { + patterns: BTreeMap::new(), + } + ); + assert_eq!( + parse(r#"{"command":"unsubscribe"}"#).unwrap(), + Command::Unsubscribe + ); + } + + #[test] + fn rejects_bad_requests() { + assert!(parse("not json").is_err()); + assert!(parse(r#"{"args":{}}"#).is_err()); + assert!(parse(r#"{"command":"subscribe"}"#).is_err()); + assert!(parse(r#"{"command":"subscribe","args":{"patterns":["a.cfg"]}}"#).is_err()); + assert!(parse(r#"{"command":"subscribe","args":{"patterns":{"x":1}}}"#).is_err()); + let err = + parse(r#"{"command":"subscribe","args":{"patterns":{"x":"{a}{b}"}}}"#).unwrap_err(); + assert!( + err.to_string() + .starts_with("only one placeholder per component in '{a}{b}'"), + "{err}" + ); + let err = parse(r#"{"command":"bogus"}"#).unwrap_err(); + assert!( + err.to_string().starts_with("unknown variant `bogus`"), + "{err}" + ); + } + + #[test] + fn formats_events() { + let params = Params::from([( + "guest", + vec![Captures::from([("node", "n1"), ("vmid", "100")])], + )]); + let write = Message::Event(Event { + kind: EventKind::Write, + path: Some("nodes/n1/qemu-server/100.conf"), + to: None, + params: Some(params), + }); + assert_eq!( + write.line(), + "{\"event\":{\"type\":\"write\",\"path\":\"nodes/n1/qemu-server/100.conf\",\"params\":{\"guest\":[{\"node\":\"n1\",\"vmid\":\"100\"}]}}}\n" + ); + let params = Params::from([("dc", vec![Captures::new()])]); + let rename = Message::Event(Event { + kind: EventKind::Rename, + path: Some("a"), + to: Some("datacenter.cfg"), + params: Some(params), + }); + assert_eq!( + rename.line(), + "{\"event\":{\"type\":\"rename\",\"path\":\"a\",\"to\":\"datacenter.cfg\",\"params\":{\"dc\":[{}]}}}\n" + ); + assert_eq!( + Message::resync().line(), + "{\"event\":{\"type\":\"resync\"}}\n" + ); + } + + #[test] + fn formats_replies() { + assert_eq!(Message::Ok(Value::Null).line(), "{\"ok\":null}\n"); + assert_eq!( + Message::Ok(serde_json::json!({"protocol": 1})).line(), + "{\"ok\":{\"protocol\":1}}\n" + ); + assert_eq!( + Message::Error("multi\nline".to_owned()).line(), + "{\"error\":\"multi\\nline\"}\n" + ); + } +} -- 2.47.3