* [PATCH pve-cluster 01/12] buildsys: add rust workspace under src/rust
2026-10-22 14:21 [PATCH cluster 00/12] pmxcfs: add a change notification socket Hannes Laimer
@ 2026-10-22 14:21 ` Hannes Laimer
2026-10-22 14:21 ` [PATCH pve-cluster 02/12] rust: notify: add path templates with placeholder capture Hannes Laimer
` (10 subsequent siblings)
11 siblings, 0 replies; 13+ messages in thread
From: Hannes Laimer @ 2026-10-22 14:21 UTC (permalink / raw)
To: pve-devel
Add a Cargo workspace under src/rust to hold Rust code for pmxcfs. Its
only member is an empty crate for the change notification socket
server. Crates come from the Debian cargo registry as in pve-rs. The
package build picks the Debian cargo wrapper and a release build in
d/rules, as in proxmox-backup. A plain make in the tree uses the cargo
in PATH and builds in debug mode. The workspace stays in the
src/Makefile recursion so that make check runs its tests and make
clean removes its target directory.
Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
.gitignore | 2 ++
Makefile | 1 +
debian/control | 22 +++++++++++++++++++++-
debian/rules | 17 +++++++++++++++++
src/Makefile | 2 +-
src/defines.mk | 11 +++++++++++
src/rust/.cargo/config.toml | 8 ++++++++
src/rust/Cargo.toml | 23 +++++++++++++++++++++++
src/rust/Makefile | 20 ++++++++++++++++++++
src/rust/pmxcfs-notify/Cargo.toml | 11 +++++++++++
src/rust/pmxcfs-notify/src/lib.rs | 1 +
src/rust/rustfmt.toml | 2 ++
12 files changed, 118 insertions(+), 2 deletions(-)
create mode 100644 src/defines.mk
create mode 100644 src/rust/.cargo/config.toml
create mode 100644 src/rust/Cargo.toml
create mode 100644 src/rust/Makefile
create mode 100644 src/rust/pmxcfs-notify/Cargo.toml
create mode 100644 src/rust/pmxcfs-notify/src/lib.rs
create mode 100644 src/rust/rustfmt.toml
diff --git a/.gitignore b/.gitignore
index 84bb09a..bb6e54b 100644
--- a/.gitignore
+++ b/.gitignore
@@ -5,3 +5,5 @@
/*.dsc
/*.tar*
/pve-cluster-*/
+/src/rust/target/
+/src/rust/Cargo.lock
diff --git a/Makefile b/Makefile
index 5fa96ab..3e1c5b9 100644
--- a/Makefile
+++ b/Makefile
@@ -23,6 +23,7 @@ tidy:
$(BUILDDIR):
rm -rf $@ $@.tmp
cp -a src $@.tmp
+ rm -rf $@.tmp/rust/target $@.tmp/rust/Cargo.lock
cp -a debian $@.tmp/
echo "git clone git://git.proxmox.com/git/pve-cluster.git\\ngit checkout $(GITVERSION)" > $@.tmp/debian/SOURCE
mv $@.tmp $@
diff --git a/debian/control b/debian/control
index 995e2d8..db858b6 100644
--- a/debian/control
+++ b/debian/control
@@ -2,8 +2,10 @@ Source: pve-cluster
Section: admin
Priority: optional
Maintainer: Proxmox Support Team <support@proxmox.com>
-Build-Depends: check,
+Build-Depends: cargo:native,
+ check,
debhelper-compat (= 13),
+ dh-cargo (>= 25),
libcmap-dev (>= 0.17.1),
libcorosync-common-dev,
libcpg-dev (>= 2.3.4),
@@ -17,11 +19,29 @@ Build-Depends: check,
libquorum-dev (>= 2.3.4),
librrd-dev,
librrds-perl,
+ librust-anyhow-1+default-dev,
+ librust-libc-0.2+default-dev,
+ librust-log-0.4+default-dev,
+ librust-nix-0.29+default-dev,
+ librust-nix-0.29+fs-dev,
+ librust-nix-0.29+socket-dev,
+ librust-serde-1+default-dev,
+ librust-serde-1+derive-dev,
+ librust-serde-json-1+default-dev,
+ librust-tokio-1+default-dev,
+ librust-tokio-1+io-util-dev,
+ librust-tokio-1+macros-dev,
+ librust-tokio-1+net-dev,
+ librust-tokio-1+rt-dev,
+ librust-tokio-1+sync-dev,
+ librust-tokio-1+time-dev,
libsqlite3-dev,
+ libstd-rust-dev,
libtest-mockmodule-perl,
libuuid-perl,
pve-doc-generator (>= 6.0-9~),
rrdcached,
+ rustc:native (>= 1.88),
sqlite3,
Standards-Version: 4.6.2
diff --git a/debian/rules b/debian/rules
index 01efb49..f4d8905 100755
--- a/debian/rules
+++ b/debian/rules
@@ -1,12 +1,29 @@
#!/usr/bin/make -f
# -*- makefile -*-
+include /usr/share/dpkg/pkg-info.mk
+include /usr/share/rustc/architecture.mk
+
# Uncomment this to turn on verbose mode.
#export DH_VERBOSE=1
+export CARGO=/usr/share/cargo/bin/cargo
+export BUILD_MODE=release
+
+export CFLAGS CXXFLAGS CPPFLAGS LDFLAGS
+export DEB_HOST_RUST_TYPE DEB_HOST_GNU_TYPE
+export CARGO_HOME = $(CURDIR)/debian/cargo_home
+
+export DEB_CARGO_CRATE=pmxcfs-notify_$(DEB_VERSION_UPSTREAM)
+export DEB_CARGO_PACKAGE=pmxcfs-notify
+
%:
dh $@
+override_dh_auto_configure:
+ $(CARGO) prepare-debian $(CURDIR)/debian/cargo_registry --link-from-system
+ dh_auto_configure
+
override_dh_installinit:
override_dh_missing:
diff --git a/src/Makefile b/src/Makefile
index 50dd6aa..ac78c89 100644
--- a/src/Makefile
+++ b/src/Makefile
@@ -1,4 +1,4 @@
-SUBDIRS := PVE pmxcfs test
+SUBDIRS := PVE rust pmxcfs test
export LD_LIBRARY_PATH+=$(CURDIR)/PVE
export PERLLIB+=$(CURDIR)/PVE
diff --git a/src/defines.mk b/src/defines.mk
new file mode 100644
index 0000000..f539f5d
--- /dev/null
+++ b/src/defines.mk
@@ -0,0 +1,11 @@
+include /usr/share/rustc/architecture.mk
+
+CARGO ?= cargo
+
+ifeq ($(BUILD_MODE), release)
+CARGO_BUILD_ARGS += --release --target $(DEB_HOST_RUST_TYPE)
+RUST_COMPILE_DIR := target/$(DEB_HOST_RUST_TYPE)/release
+else
+CARGO_BUILD_ARGS += --target $(DEB_HOST_RUST_TYPE)
+RUST_COMPILE_DIR := target/$(DEB_HOST_RUST_TYPE)/debug
+endif
diff --git a/src/rust/.cargo/config.toml b/src/rust/.cargo/config.toml
new file mode 100644
index 0000000..7b442dc
--- /dev/null
+++ b/src/rust/.cargo/config.toml
@@ -0,0 +1,8 @@
+[source]
+[source.debian-packages]
+directory = "/usr/share/cargo/registry"
+[source.crates-io]
+replace-with = "debian-packages"
+
+[profile.release]
+debug = true
diff --git a/src/rust/Cargo.toml b/src/rust/Cargo.toml
new file mode 100644
index 0000000..8cf928e
--- /dev/null
+++ b/src/rust/Cargo.toml
@@ -0,0 +1,23 @@
+[workspace]
+members = [
+ "pmxcfs-notify",
+]
+resolver = "3"
+
+[workspace.package]
+authors = ["Proxmox Support Team <support@proxmox.com>"]
+edition = "2024"
+license = "AGPL-3"
+homepage = "https://proxmox.com"
+rust-version = "1.88"
+
+[workspace.dependencies]
+anyhow = "1.0"
+libc = "0.2"
+log = "0.4"
+nix = { version = "0.29", features = ["fs", "socket"] }
+serde = { version = "1.0", features = ["derive"] }
+serde_json = "1.0"
+tokio = { version = "1", features = ["io-util", "macros", "net", "rt", "sync", "time"] }
+
+pmxcfs-notify = { path = "pmxcfs-notify" }
diff --git a/src/rust/Makefile b/src/rust/Makefile
new file mode 100644
index 0000000..be2701a
--- /dev/null
+++ b/src/rust/Makefile
@@ -0,0 +1,20 @@
+include ../defines.mk
+
+all:
+ $(CARGO) build $(CARGO_BUILD_ARGS)
+
+.PHONY: check
+check:
+ $(CARGO) test $(CARGO_BUILD_ARGS)
+
+.PHONY: fmt
+fmt:
+ cargo fmt
+
+.PHONY: install
+install:
+
+.PHONY: clean
+clean:
+ $(CARGO) clean
+ rm -f Cargo.lock
diff --git a/src/rust/pmxcfs-notify/Cargo.toml b/src/rust/pmxcfs-notify/Cargo.toml
new file mode 100644
index 0000000..3de30b5
--- /dev/null
+++ b/src/rust/pmxcfs-notify/Cargo.toml
@@ -0,0 +1,11 @@
+[package]
+name = "pmxcfs-notify"
+version = "0.1.0"
+description = "Change notification socket server for pmxcfs"
+authors.workspace = true
+edition.workspace = true
+license.workspace = true
+homepage.workspace = true
+rust-version.workspace = true
+
+[dependencies]
diff --git a/src/rust/pmxcfs-notify/src/lib.rs b/src/rust/pmxcfs-notify/src/lib.rs
new file mode 100644
index 0000000..8b13789
--- /dev/null
+++ b/src/rust/pmxcfs-notify/src/lib.rs
@@ -0,0 +1 @@
+
diff --git a/src/rust/rustfmt.toml b/src/rust/rustfmt.toml
new file mode 100644
index 0000000..f3e454b
--- /dev/null
+++ b/src/rust/rustfmt.toml
@@ -0,0 +1,2 @@
+edition = "2024"
+style_edition = "2024"
--
2.47.3
^ permalink raw reply related [flat|nested] 13+ messages in thread* [PATCH pve-cluster 02/12] rust: notify: add path templates with placeholder capture
2026-10-22 14:21 [PATCH cluster 00/12] pmxcfs: add a change notification socket Hannes Laimer
2026-10-22 14:21 ` [PATCH pve-cluster 01/12] buildsys: add rust workspace under src/rust Hannes Laimer
@ 2026-10-22 14:21 ` Hannes Laimer
2026-10-22 14:21 ` [PATCH pve-cluster 03/12] rust: notify: add the wire protocol Hannes Laimer
` (9 subsequent siblings)
11 siblings, 0 replies; 13+ messages in thread
From: Hannes Laimer @ 2026-10-22 14:21 UTC (permalink / raw)
To: pve-devel
A template such as nodes/{node}/qemu-server/{vmid}.conf matches memdb
paths and captures the parts its placeholders stand for. A component
holds at most one placeholder, and a placeholder may take the rest of
the path only as the whole last component. With these limits a match
takes a single pass over the path without backtracking.
Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
src/rust/pmxcfs-notify/Cargo.toml | 3 +
src/rust/pmxcfs-notify/src/lib.rs | 14 ++
src/rust/pmxcfs-notify/src/template.rs | 285 +++++++++++++++++++++++++
3 files changed, 302 insertions(+)
create mode 100644 src/rust/pmxcfs-notify/src/template.rs
diff --git a/src/rust/pmxcfs-notify/Cargo.toml b/src/rust/pmxcfs-notify/Cargo.toml
index 3de30b5..5a49c75 100644
--- a/src/rust/pmxcfs-notify/Cargo.toml
+++ b/src/rust/pmxcfs-notify/Cargo.toml
@@ -9,3 +9,6 @@ homepage.workspace = true
rust-version.workspace = true
[dependencies]
+anyhow.workspace = true
+serde.workspace = true
+serde_json.workspace = true
diff --git a/src/rust/pmxcfs-notify/src/lib.rs b/src/rust/pmxcfs-notify/src/lib.rs
index 8b13789..889fe50 100644
--- a/src/rust/pmxcfs-notify/src/lib.rs
+++ b/src/rust/pmxcfs-notify/src/lib.rs
@@ -1 +1,15 @@
+//! Change notification socket for pmxcfs.
+//!
+//! 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
+//! path.
+#![allow(dead_code)]
+
+mod template;
+
+/// Events and templates name memdb paths without a leading slash.
+pub(crate) fn normalize_path(path: &str) -> &str {
+ path.trim_start_matches('/')
+}
diff --git a/src/rust/pmxcfs-notify/src/template.rs b/src/rust/pmxcfs-notify/src/template.rs
new file mode 100644
index 0000000..e8feb8c
--- /dev/null
+++ b/src/rust/pmxcfs-notify/src/template.rs
@@ -0,0 +1,285 @@
+use std::collections::BTreeMap;
+use std::str::FromStr;
+
+use anyhow::{Error, bail};
+use serde::Deserialize;
+
+use crate::normalize_path;
+
+pub(crate) type Captures<'a> = BTreeMap<&'a str, &'a str>;
+
+#[derive(Debug, PartialEq, Eq)]
+enum Component {
+ Literal(String),
+ Param {
+ prefix: String,
+ name: String,
+ suffix: String,
+ },
+ Rest(String),
+}
+
+// One placeholder per component and the spanning form only at the end let a match run in a single
+// pass over the path without backtracking. It runs for every event, template and connection.
+#[derive(Debug, PartialEq, Eq, Deserialize)]
+#[serde(try_from = "String")]
+pub(crate) struct Template {
+ components: Vec<Component>,
+}
+
+fn valid_name(name: &str) -> bool {
+ let mut chars = name.chars();
+ chars
+ .next()
+ .is_some_and(|c| c.is_ascii_alphabetic() || c == '_')
+ && chars.all(|c| c.is_ascii_alphanumeric() || c == '_')
+}
+
+impl FromStr for Template {
+ type Err = Error;
+
+ fn from_str(text: &str) -> Result<Self, Error> {
+ let text = normalize_path(text);
+ if text.is_empty() {
+ bail!("empty template");
+ }
+ let parts: Vec<&str> = text.split('/').collect();
+ let last = parts.len() - 1;
+ let mut components = Vec::with_capacity(parts.len());
+ let mut names: Vec<&str> = Vec::new();
+ for (index, part) in parts.into_iter().enumerate() {
+ if part.is_empty() {
+ bail!("empty path component");
+ }
+ let Some((prefix, rest)) = part.split_once('{') else {
+ if part.contains('}') {
+ bail!("malformed placeholder in '{part}'");
+ }
+ components.push(Component::Literal(part.to_owned()));
+ continue;
+ };
+ let Some((body, suffix)) = rest.split_once('}') else {
+ bail!("unterminated placeholder in '{part}'");
+ };
+ if suffix.contains('{') {
+ bail!("only one placeholder per component in '{part}'");
+ }
+ if prefix.contains('}') || body.contains('{') || suffix.contains('}') {
+ bail!("malformed placeholder in '{part}'");
+ }
+ let (name, spans) = match body.strip_suffix("...") {
+ Some(name) => (name, true),
+ None => (body, false),
+ };
+ if !valid_name(name) {
+ bail!("invalid placeholder name '{name}'");
+ }
+ if names.contains(&name) {
+ bail!("placeholder '{name}' used twice");
+ }
+ names.push(name);
+ if spans {
+ if index != last || !prefix.is_empty() || !suffix.is_empty() {
+ bail!("'{{{name}...}}' must be the whole last component");
+ }
+ components.push(Component::Rest(name.to_owned()));
+ } else {
+ components.push(Component::Param {
+ prefix: prefix.to_owned(),
+ name: name.to_owned(),
+ suffix: suffix.to_owned(),
+ });
+ }
+ }
+ Ok(Self { components })
+ }
+}
+
+impl TryFrom<String> for Template {
+ type Error = Error;
+
+ fn try_from(text: String) -> Result<Self, Error> {
+ text.parse()
+ }
+}
+
+impl Template {
+ pub(crate) fn matches<'a>(&'a self, path: &'a str) -> Option<Captures<'a>> {
+ let mut captures = Captures::new();
+ let mut rest = path;
+ let last = self.components.len() - 1;
+ for (index, component) in self.components.iter().enumerate() {
+ if let Component::Rest(name) = component {
+ if rest.is_empty() {
+ return None;
+ }
+ captures.insert(name.as_str(), rest);
+ return Some(captures);
+ }
+ let (head, tail) = match rest.split_once('/') {
+ Some((head, tail)) => (head, Some(tail)),
+ None => (rest, None),
+ };
+ match component {
+ Component::Literal(literal) if head == literal.as_str() => {}
+ Component::Param {
+ prefix,
+ name,
+ suffix,
+ } => {
+ let value = head
+ .strip_prefix(prefix.as_str())?
+ .strip_suffix(suffix.as_str())?;
+ if value.is_empty() {
+ return None;
+ }
+ captures.insert(name.as_str(), value);
+ }
+ _ => return None,
+ }
+ match (tail, index == last) {
+ (None, true) => return Some(captures),
+ (Some(tail), false) => rest = tail,
+ _ => return None,
+ }
+ }
+ unreachable!("the last component returns")
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ fn parse(text: &str) -> Result<Template, Error> {
+ text.parse()
+ }
+
+ fn captures<'a>(pairs: &[(&'a str, &'a str)]) -> Captures<'a> {
+ pairs.iter().copied().collect()
+ }
+
+ #[test]
+ fn parses_templates() {
+ let literal = |text: &str| Component::Literal(text.to_owned());
+ let param = |prefix: &str, name: &str, suffix: &str| Component::Param {
+ prefix: prefix.to_owned(),
+ name: name.to_owned(),
+ suffix: suffix.to_owned(),
+ };
+ assert_eq!(
+ parse("/datacenter.cfg").unwrap(),
+ Template {
+ components: vec![literal("datacenter.cfg")]
+ }
+ );
+ assert_eq!(
+ parse("nodes/{node}/qemu-server/{vmid}.conf").unwrap(),
+ Template {
+ components: vec![
+ literal("nodes"),
+ param("", "node", ""),
+ literal("qemu-server"),
+ param("", "vmid", ".conf"),
+ ]
+ }
+ );
+ assert_eq!(
+ parse("{path...}").unwrap(),
+ Template {
+ components: vec![Component::Rest("path".to_owned())]
+ }
+ );
+ assert_eq!(
+ parse("priv/lock/{path...}").unwrap(),
+ Template {
+ components: vec![
+ literal("priv"),
+ literal("lock"),
+ Component::Rest("path".to_owned())
+ ]
+ }
+ );
+ }
+
+ #[test]
+ fn rejects_bad_templates() {
+ for (text, message) in [
+ ("", "empty template"),
+ ("/", "empty template"),
+ ("a//b", "empty path component"),
+ ("a/", "empty path component"),
+ ("{a}{b}", "only one placeholder per component"),
+ ("{a", "unterminated placeholder"),
+ ("a}", "malformed placeholder"),
+ ("{a}}", "malformed placeholder"),
+ ("{a{b}", "malformed placeholder"),
+ ("{}", "invalid placeholder name ''"),
+ ("{1a}", "invalid placeholder name '1a'"),
+ ("{a-b}", "invalid placeholder name 'a-b'"),
+ ("a/{x}/{x}", "placeholder 'x' used twice"),
+ ("{a...}/b", "must be the whole last component"),
+ ("x{a...}", "must be the whole last component"),
+ ("{a...}.conf", "must be the whole last component"),
+ ] {
+ let err = parse(text).unwrap_err().to_string();
+ assert!(err.contains(message), "{text:?}: {err}");
+ }
+ }
+
+ #[test]
+ fn deserializes_from_a_string() {
+ let template: Template = serde_json::from_str("\"/nodes/{node}/host.fw\"").unwrap();
+ assert_eq!(template, parse("nodes/{node}/host.fw").unwrap());
+ let err = serde_json::from_str::<Template>("\"{a}{b}\"").unwrap_err();
+ assert!(
+ err.to_string()
+ .starts_with("only one placeholder per component in '{a}{b}'"),
+ "{err}"
+ );
+ assert!(serde_json::from_str::<Template>("1").is_err());
+ }
+
+ #[test]
+ fn matches_paths() {
+ let guest = parse("nodes/{node}/qemu-server/{vmid}.conf").unwrap();
+ assert_eq!(
+ guest.matches("nodes/n1/qemu-server/100.conf"),
+ Some(captures(&[("node", "n1"), ("vmid", "100")]))
+ );
+ assert_eq!(guest.matches("nodes/n1/qemu-server/100.conf.tmp.1"), None);
+ assert_eq!(guest.matches("nodes/n1/qemu-server/.conf"), None);
+ assert_eq!(guest.matches("nodes/n1/lxc/100.conf"), None);
+ assert_eq!(guest.matches("nodes/n1/qemu-server"), None);
+ assert_eq!(guest.matches("nodes/n1/qemu-server/100.conf/x"), None);
+ assert_eq!(guest.matches("nodes//qemu-server/100.conf"), None);
+ assert_eq!(guest.matches(""), None);
+
+ let host = parse("nodes/{node}/host-{id}.cfg").unwrap();
+ assert_eq!(
+ host.matches("nodes/n1/host-7.cfg"),
+ Some(captures(&[("node", "n1"), ("id", "7")]))
+ );
+ assert_eq!(host.matches("nodes/n1/host-.cfg"), None);
+ assert_eq!(host.matches("nodes/n1/host-7.cfgx"), None);
+
+ let exact = parse("datacenter.cfg").unwrap();
+ assert_eq!(exact.matches("datacenter.cfg"), Some(captures(&[])));
+ assert_eq!(exact.matches("datacenter.cfg/x"), None);
+ assert_eq!(exact.matches("xdatacenter.cfg"), None);
+
+ let all = parse("{path...}").unwrap();
+ assert_eq!(all.matches("a"), Some(captures(&[("path", "a")])));
+ assert_eq!(all.matches("a/b/c"), Some(captures(&[("path", "a/b/c")])));
+ assert_eq!(all.matches(""), None);
+
+ let locks = parse("priv/lock/{path...}").unwrap();
+ assert_eq!(
+ locks.matches("priv/lock/file-storage_cfg/a"),
+ Some(captures(&[("path", "file-storage_cfg/a")]))
+ );
+ assert_eq!(locks.matches("priv/lock"), None);
+ assert_eq!(locks.matches("priv/lock/"), None);
+ assert_eq!(locks.matches("priv/locks/a"), None);
+ }
+}
--
2.47.3
^ permalink raw reply related [flat|nested] 13+ messages in thread* [PATCH pve-cluster 03/12] rust: notify: add the wire protocol
2026-10-22 14:21 [PATCH cluster 00/12] pmxcfs: add a change notification socket Hannes Laimer
2026-10-22 14:21 ` [PATCH pve-cluster 01/12] buildsys: add rust workspace under src/rust Hannes Laimer
2026-10-22 14:21 ` [PATCH pve-cluster 02/12] rust: notify: add path templates with placeholder capture Hannes Laimer
@ 2026-10-22 14:21 ` Hannes Laimer
2026-10-22 14:21 ` [PATCH pve-cluster 04/12] rust: notify: add subscription matching Hannes Laimer
` (8 subsequent siblings)
11 siblings, 0 replies; 13+ messages in thread
From: Hannes Laimer @ 2026-10-22 14:21 UTC (permalink / raw)
To: pve-devel
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 <h.laimer@proxmox.com>
---
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": <json>}` 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<Captures<'a>>>;
+
+#[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<Params<'a>>,
+}
+
+/// 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<String, Template>,
+ },
+ Unsubscribe,
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ fn parse(line: &str) -> Result<Command, serde_json::Error> {
+ 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
^ permalink raw reply related [flat|nested] 13+ messages in thread* [PATCH pve-cluster 04/12] rust: notify: add subscription matching
2026-10-22 14:21 [PATCH cluster 00/12] pmxcfs: add a change notification socket Hannes Laimer
` (2 preceding siblings ...)
2026-10-22 14:21 ` [PATCH pve-cluster 03/12] rust: notify: add the wire protocol Hannes Laimer
@ 2026-10-22 14:21 ` Hannes Laimer
2026-10-22 14:21 ` [PATCH pve-cluster 05/12] rust: notify: add the socket server Hannes Laimer
` (7 subsequent siblings)
11 siblings, 0 replies; 13+ messages in thread
From: Hannes Laimer @ 2026-10-22 14:21 UTC (permalink / raw)
To: pve-devel
A subscription is a set of named templates. An event matches when its
path or its rename target matches one of them. A rename carries the
captures of each side that matches. A connection without root
privileges gets no event for a private path, just as IPC and FUSE
hide the content from it.
Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
src/rust/pmxcfs-notify/src/conn.rs | 140 +++++++++++++++++++++++++++++
src/rust/pmxcfs-notify/src/lib.rs | 4 +-
2 files changed, 143 insertions(+), 1 deletion(-)
create mode 100644 src/rust/pmxcfs-notify/src/conn.rs
diff --git a/src/rust/pmxcfs-notify/src/conn.rs b/src/rust/pmxcfs-notify/src/conn.rs
new file mode 100644
index 0000000..451504c
--- /dev/null
+++ b/src/rust/pmxcfs-notify/src/conn.rs
@@ -0,0 +1,140 @@
+use std::collections::BTreeMap;
+
+use crate::protocol::Params;
+use crate::template::Template;
+
+pub(crate) struct Conn {
+ privileged: bool,
+ templates: BTreeMap<String, Template>,
+}
+
+impl Conn {
+ pub(crate) fn new(privileged: bool) -> Self {
+ Self {
+ privileged,
+ templates: BTreeMap::new(),
+ }
+ }
+
+ pub(crate) fn matches<'a>(&'a self, path: &'a str, to: Option<&'a str>) -> Option<Params<'a>> {
+ if !self.privileged && (is_private(path) || to.is_some_and(is_private)) {
+ return None;
+ }
+ let mut params = Params::new();
+ for (name, template) in &self.templates {
+ let mut sets = Vec::new();
+ if let Some(captures) = template.matches(path) {
+ sets.push(captures);
+ }
+ // equal captures mean the same path, so a rename onto itself matches once
+ if let Some(to) = to
+ && let Some(captures) = template.matches(to)
+ && sets.first() != Some(&captures)
+ {
+ sets.push(captures);
+ }
+ if !sets.is_empty() {
+ params.insert(name.as_str(), sets);
+ }
+ }
+ (!params.is_empty()).then_some(params)
+ }
+}
+
+// Same rule as path_is_private in the daemon.
+fn is_private(path: &str) -> bool {
+ fn priv_component(path: &str) -> bool {
+ path == "priv" || path.starts_with("priv/")
+ }
+ let path = crate::normalize_path(path);
+ priv_component(path)
+ || path
+ .strip_prefix("nodes/")
+ .and_then(|rest| rest.split_once('/'))
+ .is_some_and(|(_, below)| priv_component(below))
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ fn subscribed(privileged: bool, specs: &[(&str, &str)]) -> Conn {
+ let mut conn = Conn::new(privileged);
+ conn.templates = specs
+ .iter()
+ .map(|(name, text)| ((*name).to_owned(), text.parse().unwrap()))
+ .collect();
+ conn
+ }
+
+ #[test]
+ fn matches_per_template() {
+ let conn = subscribed(
+ true,
+ &[
+ ("guest", "nodes/{node}/qemu-server/{vmid}.conf"),
+ ("dc", "datacenter.cfg"),
+ ],
+ );
+ let params = conn.matches("nodes/n1/qemu-server/100.conf", None).unwrap();
+ assert_eq!(params.len(), 1);
+ assert_eq!(
+ params["guest"],
+ vec![BTreeMap::from([("node", "n1"), ("vmid", "100")])]
+ );
+ assert_eq!(
+ conn.matches("datacenter.cfg", None).unwrap()["dc"],
+ vec![BTreeMap::new()]
+ );
+ assert!(conn.matches("datacenter.cfg.tmp.1", None).is_none());
+ assert!(conn.matches("other", None).is_none());
+ assert!(Conn::new(true).matches("datacenter.cfg", None).is_none());
+ }
+
+ #[test]
+ fn rename_matches_on_either_side() {
+ let conn = subscribed(true, &[("guest", "nodes/{node}/qemu-server/{vmid}.conf")]);
+ let n1 = "nodes/n1/qemu-server/100.conf";
+ let n2 = "nodes/n2/qemu-server/100.conf";
+
+ let params = conn.matches(n1, Some(n2)).unwrap();
+ assert_eq!(params["guest"].len(), 2);
+ assert_eq!(params["guest"][0]["node"], "n1");
+ assert_eq!(params["guest"][1]["node"], "n2");
+
+ let params = conn
+ .matches("nodes/n2/qemu-server/100.conf.tmp.1", Some(n2))
+ .unwrap();
+ assert_eq!(params["guest"].len(), 1);
+ assert_eq!(params["guest"][0]["node"], "n2");
+
+ assert_eq!(conn.matches(n1, Some(n1)).unwrap()["guest"].len(), 1);
+ }
+
+ #[test]
+ fn private_paths_need_privileges() {
+ let conn = subscribed(false, &[("all", "{path...}")]);
+ assert!(conn.matches("datacenter.cfg", None).is_some());
+ assert!(conn.matches("priv/lock/x", None).is_none());
+ assert!(conn.matches("nodes/n1/priv/ssl.key", None).is_none());
+ assert!(conn.matches("datacenter.cfg", Some("priv/moved")).is_none());
+ assert!(conn.matches("priv/old", Some("datacenter.cfg")).is_none());
+
+ let conn = subscribed(true, &[("all", "{path...}")]);
+ assert!(conn.matches("priv/lock/x", None).is_some());
+ assert!(conn.matches("priv/old", Some("datacenter.cfg")).is_some());
+ }
+
+ #[test]
+ fn private_rules() {
+ assert!(is_private("priv"));
+ assert!(is_private("priv/lock/x"));
+ assert!(is_private("/priv/tfa.cfg"));
+ assert!(is_private("nodes/n1/priv"));
+ assert!(is_private("nodes/n1/priv/ssl.key"));
+ assert!(!is_private("private"));
+ assert!(!is_private("nodes/n1/qemu-server/100.conf"));
+ assert!(!is_private("nodes/priv"));
+ assert!(!is_private("datacenter.cfg"));
+ }
+}
diff --git a/src/rust/pmxcfs-notify/src/lib.rs b/src/rust/pmxcfs-notify/src/lib.rs
index d35d3c5..18e7d2f 100644
--- a/src/rust/pmxcfs-notify/src/lib.rs
+++ b/src/rust/pmxcfs-notify/src/lib.rs
@@ -10,10 +10,12 @@
//! 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
-//! path.
+//! path. Connections admitted by their group id rather than as root never receive events that
+//! mention a private path, just as IPC and FUSE hide private files from them.
#![allow(dead_code)]
+mod conn;
mod protocol;
mod template;
--
2.47.3
^ permalink raw reply related [flat|nested] 13+ messages in thread* [PATCH pve-cluster 05/12] rust: notify: add the socket server
2026-10-22 14:21 [PATCH cluster 00/12] pmxcfs: add a change notification socket Hannes Laimer
` (3 preceding siblings ...)
2026-10-22 14:21 ` [PATCH pve-cluster 04/12] rust: notify: add subscription matching Hannes Laimer
@ 2026-10-22 14:21 ` Hannes Laimer
2026-10-22 14:21 ` [PATCH pve-cluster 06/12] rust: ffi: add C ABI staticlib for pmxcfs Hannes Laimer
` (6 subsequent siblings)
11 siblings, 0 replies; 13+ messages in thread
From: Hannes Laimer @ 2026-10-22 14:21 UTC (permalink / raw)
To: pve-devel
Daemons learn about changes in /etc/pve by polling the file versions
over IPC, and inotify only sees writes made on the local node. Add a
server that pushes each mutation it is handed to the connections whose
subscription matches it. It runs on its own thread, and only root and
one configured group can connect to its Unix socket.
A memdb mutation holds the database mutex, so the server must never
wait on a client when it is handed an event. It just puts the event
into a bounded channel that every client reads at its own pace. A
client that falls behind by more than the channel holds gets one
resync in place of what it missed. If a write to a client does not
complete within 30 seconds, its connection is dropped.
No event history is kept beyond the channel. A new subscription starts
at the current state, so the server answers a subscribe with a resync
and the client reads everything once. A resync from a lagged channel
asks for the same.
Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
src/rust/pmxcfs-notify/Cargo.toml | 3 +
src/rust/pmxcfs-notify/src/conn.rs | 196 +++++++++-
src/rust/pmxcfs-notify/src/lib.rs | 11 +-
src/rust/pmxcfs-notify/src/server.rs | 542 +++++++++++++++++++++++++++
4 files changed, 749 insertions(+), 3 deletions(-)
create mode 100644 src/rust/pmxcfs-notify/src/server.rs
diff --git a/src/rust/pmxcfs-notify/Cargo.toml b/src/rust/pmxcfs-notify/Cargo.toml
index 5a49c75..bd29936 100644
--- a/src/rust/pmxcfs-notify/Cargo.toml
+++ b/src/rust/pmxcfs-notify/Cargo.toml
@@ -10,5 +10,8 @@ rust-version.workspace = true
[dependencies]
anyhow.workspace = true
+log.workspace = true
+nix.workspace = true
serde.workspace = true
serde_json.workspace = true
+tokio.workspace = true
diff --git a/src/rust/pmxcfs-notify/src/conn.rs b/src/rust/pmxcfs-notify/src/conn.rs
index 451504c..60319f5 100644
--- a/src/rust/pmxcfs-notify/src/conn.rs
+++ b/src/rust/pmxcfs-notify/src/conn.rs
@@ -1,13 +1,36 @@
use std::collections::BTreeMap;
+use std::io;
+use std::time::Duration;
-use crate::protocol::Params;
+use serde_json::{Value, json};
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::UnixStream;
+use tokio::sync::broadcast::{self, error::RecvError};
+use tokio::time::timeout;
+
+use crate::protocol::{Command, Event, Message, PROTOCOL_VERSION, Params};
+use crate::server::{Entry, Slot};
use crate::template::Template;
+const MAX_LINE: usize = 64 * 1024;
+#[cfg(not(test))]
+const WRITE_TIMEOUT: Duration = Duration::from_secs(30);
+#[cfg(test)]
+const WRITE_TIMEOUT: Duration = Duration::from_secs(2);
+
pub(crate) struct Conn {
privileged: bool,
templates: BTreeMap<String, Template>,
}
+#[derive(Debug)]
+enum ReadError {
+ Eof,
+ TooLong,
+ NotUtf8,
+ Io(io::Error),
+}
+
impl Conn {
pub(crate) fn new(privileged: bool) -> Self {
Self {
@@ -41,6 +64,118 @@ impl Conn {
}
}
+pub(crate) async fn serve(mut stream: UnixStream, privileged: bool, slot: Slot) {
+ let mut conn = Conn::new(privileged);
+ let mut events: Option<broadcast::Receiver<Entry>> = None;
+ let mut inbuf = Vec::new();
+ loop {
+ let line = tokio::select! {
+ line = read_line(&mut stream, &mut inbuf) => {
+ let line = match line {
+ Ok(line) => line,
+ Err(ReadError::Eof) => return,
+ Err(ReadError::TooLong) => {
+ log::warn!("closing connection, line too long");
+ return;
+ }
+ Err(ReadError::NotUtf8) => {
+ log::warn!("closing connection, line is not UTF-8");
+ return;
+ }
+ Err(ReadError::Io(err)) => {
+ log::debug!("closing connection: {err}");
+ return;
+ }
+ };
+ match serde_json::from_str::<Command>(&line) {
+ Ok(Command::Hello) => {
+ Message::Ok(json!({ "protocol": PROTOCOL_VERSION })).line()
+ }
+ Ok(Command::Subscribe { patterns }) => {
+ conn.templates = patterns;
+ events = Some(slot.events.subscribe());
+ Message::Ok(Value::Null).line() + &Message::resync().line()
+ }
+ Ok(Command::Unsubscribe) => {
+ conn.templates.clear();
+ events = None;
+ Message::Ok(Value::Null).line()
+ }
+ Err(err) => Message::Error(err.to_string()).line(),
+ }
+ }
+ event = next_event(&mut events) => match event {
+ Ok(Entry::Resync) => Message::resync().line(),
+ Ok(Entry::Change { kind, path, to }) => {
+ match conn.matches(&path, to.as_deref()) {
+ Some(params) => Message::Event(Event {
+ kind,
+ path: Some(&path),
+ to: to.as_deref(),
+ params: Some(params),
+ })
+ .line(),
+ None => continue,
+ }
+ }
+ // the resync stands for everything up to now, so a fresh receiver starting there
+ // replaces one that still holds older entries
+ Err(RecvError::Lagged(_)) => {
+ events = Some(slot.events.subscribe());
+ Message::resync().line()
+ }
+ Err(RecvError::Closed) => return,
+ },
+ };
+ if !write(&mut stream, &line).await {
+ return;
+ }
+ }
+}
+
+async fn next_event(events: &mut Option<broadcast::Receiver<Entry>>) -> Result<Entry, RecvError> {
+ match events {
+ Some(events) => events.recv().await,
+ None => std::future::pending().await,
+ }
+}
+
+async fn read_line(stream: &mut UnixStream, inbuf: &mut Vec<u8>) -> Result<String, ReadError> {
+ loop {
+ if let Some(end) = inbuf.iter().position(|&b| b == b'\n') {
+ let line = inbuf[..end].to_vec();
+ inbuf.drain(..=end);
+ return String::from_utf8(line).map_err(|_| ReadError::NotUtf8);
+ }
+ if inbuf.len() > MAX_LINE {
+ return Err(ReadError::TooLong);
+ }
+ let mut chunk = [0u8; 4096];
+ match stream.read(&mut chunk).await {
+ Ok(0) => return Err(ReadError::Eof),
+ Ok(n) => inbuf.extend_from_slice(&chunk[..n]),
+ Err(err) if err.kind() == io::ErrorKind::Interrupted => {}
+ Err(err) => return Err(ReadError::Io(err)),
+ }
+ }
+}
+
+/// Returns whether the line went out. A client whose socket stays full for the whole timeout is
+/// taken to have stopped reading and is cut off.
+async fn write(stream: &mut UnixStream, line: &str) -> bool {
+ match timeout(WRITE_TIMEOUT, stream.write_all(line.as_bytes())).await {
+ Ok(Ok(())) => true,
+ Ok(Err(err)) => {
+ log::debug!("closing connection: {err}");
+ false
+ }
+ Err(_) => {
+ log::warn!("closing connection, client stopped reading");
+ false
+ }
+ }
+}
+
// Same rule as path_is_private in the daemon.
fn is_private(path: &str) -> bool {
fn priv_component(path: &str) -> bool {
@@ -67,6 +202,65 @@ mod tests {
conn
}
+ async fn line_or_pending(
+ stream: &mut UnixStream,
+ inbuf: &mut Vec<u8>,
+ ) -> Option<Result<String, ReadError>> {
+ timeout(Duration::from_millis(100), read_line(stream, inbuf))
+ .await
+ .ok()
+ }
+
+ #[tokio::test]
+ async fn reassembles_partial_lines() {
+ let (mut server, mut client) = UnixStream::pair().unwrap();
+ let mut inbuf = Vec::new();
+ client.write_all(b"{\"command\":").await.unwrap();
+ assert!(line_or_pending(&mut server, &mut inbuf).await.is_none());
+ client
+ .write_all(b"\"hello\"}\n{\"command\":\"unsubscribe\"}\n")
+ .await
+ .unwrap();
+ assert_eq!(
+ read_line(&mut server, &mut inbuf).await.unwrap(),
+ "{\"command\":\"hello\"}"
+ );
+ assert_eq!(
+ read_line(&mut server, &mut inbuf).await.unwrap(),
+ "{\"command\":\"unsubscribe\"}"
+ );
+ assert!(line_or_pending(&mut server, &mut inbuf).await.is_none());
+ assert!(inbuf.is_empty());
+ }
+
+ #[tokio::test]
+ async fn rejects_overlong_and_non_utf8_lines() {
+ let (mut server, mut client) = UnixStream::pair().unwrap();
+ let mut inbuf = Vec::new();
+ tokio::spawn(async move {
+ client.write_all(&vec![b'x'; MAX_LINE + 1]).await.unwrap();
+ std::future::pending::<()>().await;
+ });
+ assert!(matches!(
+ read_line(&mut server, &mut inbuf).await,
+ Err(ReadError::TooLong)
+ ));
+
+ let (mut server, mut client) = UnixStream::pair().unwrap();
+ let mut inbuf = Vec::new();
+ client.write_all(b"\xff\xfe\n").await.unwrap();
+ assert!(matches!(
+ read_line(&mut server, &mut inbuf).await,
+ Err(ReadError::NotUtf8)
+ ));
+
+ drop(client);
+ assert!(matches!(
+ read_line(&mut server, &mut inbuf).await,
+ Err(ReadError::Eof)
+ ));
+ }
+
#[test]
fn matches_per_template() {
let conn = subscribed(
diff --git a/src/rust/pmxcfs-notify/src/lib.rs b/src/rust/pmxcfs-notify/src/lib.rs
index 18e7d2f..c152d43 100644
--- a/src/rust/pmxcfs-notify/src/lib.rs
+++ b/src/rust/pmxcfs-notify/src/lib.rs
@@ -12,13 +12,20 @@
//! `nodes/{node}/qemu-server/{vmid}.conf` captures node and vmid, and `{path...}` matches every
//! path. Connections admitted by their group id rather than as root never receive events that
//! mention a private path, just as IPC and FUSE hide private files from them.
-
-#![allow(dead_code)]
+//!
+//! Nothing is kept for a client beyond a bounded backlog. A client that falls further behind gets
+//! one resync event in place of what it missed, the same event every client gets when the daemon
+//! replaces its state. A subscription starts at the current state, so its reply is followed by a
+//! resync. A resync means reading everything again.
mod conn;
mod protocol;
+mod server;
mod template;
+pub use protocol::EventKind;
+pub use server::{Config, Server};
+
/// Events and templates name memdb paths without a leading slash.
pub(crate) fn normalize_path(path: &str) -> &str {
path.trim_start_matches('/')
diff --git a/src/rust/pmxcfs-notify/src/server.rs b/src/rust/pmxcfs-notify/src/server.rs
new file mode 100644
index 0000000..fe5defc
--- /dev/null
+++ b/src/rust/pmxcfs-notify/src/server.rs
@@ -0,0 +1,542 @@
+use std::fs;
+use std::io;
+use std::os::unix::fs::PermissionsExt;
+use std::os::unix::net::UnixListener as StdUnixListener;
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::thread::JoinHandle;
+use std::time::Duration;
+
+use anyhow::{Context, Error};
+use nix::sys::socket::{setsockopt, sockopt};
+use nix::sys::stat::{Mode, umask};
+use tokio::net::{UnixListener, UnixStream};
+use tokio::sync::{Notify, broadcast};
+
+use crate::conn;
+use crate::normalize_path;
+use crate::protocol::EventKind;
+
+const MAX_CONNECTIONS: usize = 128;
+const SNDBUF: usize = 2 * 1024 * 1024;
+
+pub struct Config {
+ pub path: PathBuf,
+ pub gid: u32,
+ /// Events kept for a client that reads slower than they arrive.
+ pub capacity: usize,
+}
+
+// Reference counted strings, so handing a copy to every connection clones only pointers.
+#[derive(Clone)]
+pub(crate) enum Entry {
+ Resync,
+ Change {
+ kind: EventKind,
+ path: Arc<str>,
+ to: Option<Arc<str>>,
+ },
+}
+
+pub(crate) struct Shared {
+ pub(crate) events: broadcast::Sender<Entry>,
+ stop: Notify,
+ connections: AtomicUsize,
+ gid: u32,
+}
+
+/// A connection's place under the cap, held for as long as its task runs.
+pub(crate) struct Slot(Arc<Shared>);
+
+impl std::ops::Deref for Slot {
+ type Target = Shared;
+
+ fn deref(&self) -> &Shared {
+ &self.0
+ }
+}
+
+impl Drop for Slot {
+ fn drop(&mut self) {
+ self.0.connections.fetch_sub(1, Ordering::Relaxed);
+ }
+}
+
+pub struct Server {
+ shared: Arc<Shared>,
+ thread: Option<JoinHandle<()>>,
+}
+
+impl Server {
+ /// Changes the process umask around the bind, so no other thread may be creating files while
+ /// this runs.
+ pub fn start(config: Config) -> Result<Server, Error> {
+ let listener = bind(&config.path, config.gid)?;
+ let runtime = tokio::runtime::Builder::new_current_thread()
+ .enable_io()
+ .enable_time()
+ .build()?;
+ let listener = {
+ let _runtime = runtime.enter();
+ UnixListener::from_std(listener)?
+ };
+ let (events, _) = broadcast::channel(config.capacity);
+ let shared = Arc::new(Shared {
+ events,
+ stop: Notify::new(),
+ connections: AtomicUsize::new(0),
+ gid: config.gid,
+ });
+ let thread = {
+ let shared = Arc::clone(&shared);
+ std::thread::Builder::new()
+ .name("pmxcfs-notify".to_owned())
+ .spawn(move || {
+ let _socket = RemoveOnExit(config.path);
+ runtime.block_on(accept_loop(listener, shared));
+ })?
+ };
+ Ok(Server {
+ shared,
+ thread: Some(thread),
+ })
+ }
+
+ /// Hands the mutation to every subscribed connection without waiting on any of them, so it
+ /// never blocks the caller. Every kind but a resync needs a path.
+ pub fn emit(&self, kind: EventKind, path: Option<&str>, to: Option<&str>) {
+ let entry = match (kind, path) {
+ (EventKind::Resync, _) => Entry::Resync,
+ (kind, Some(path)) => Entry::Change {
+ kind,
+ path: normalize_path(path).into(),
+ to: to.map(|to| normalize_path(to).into()),
+ },
+ (kind, None) => {
+ log::warn!("ignoring {kind:?} event without a path");
+ return;
+ }
+ };
+ // an error only says that no connection is subscribed
+ let _ = self.shared.events.send(entry);
+ }
+
+ pub fn shutdown(mut self) {
+ self.stop_thread();
+ }
+
+ fn stop_thread(&mut self) {
+ let Some(thread) = self.thread.take() else {
+ return;
+ };
+ self.shared.stop.notify_one();
+ let _ = thread.join();
+ }
+
+ #[cfg(test)]
+ pub(crate) fn connection_count(&self) -> usize {
+ self.shared.connections.load(Ordering::Relaxed)
+ }
+}
+
+impl Drop for Server {
+ fn drop(&mut self) {
+ self.stop_thread();
+ }
+}
+
+// The socket file goes away with the thread however the thread ends, a panic included, so no stale
+// socket is left behind.
+struct RemoveOnExit(PathBuf);
+
+impl Drop for RemoveOnExit {
+ fn drop(&mut self) {
+ let _ = fs::remove_file(&self.0);
+ }
+}
+
+// The socket is created with mode 0700, so between bind and chmod nobody but its owner can connect.
+fn bind(path: &Path, gid: u32) -> Result<StdUnixListener, Error> {
+ match fs::remove_file(path) {
+ Ok(()) => {}
+ Err(err) if err.kind() == io::ErrorKind::NotFound => {}
+ Err(err) => {
+ return Err(err).with_context(|| format!("failed to remove stale {}", path.display()));
+ }
+ }
+ let mask = umask(Mode::from_bits_truncate(0o077));
+ let listener = StdUnixListener::bind(path);
+ umask(mask);
+ let listener = listener.with_context(|| format!("failed to bind {}", path.display()))?;
+ listener.set_nonblocking(true)?;
+ std::os::unix::fs::chown(path, None, Some(gid))?;
+ fs::set_permissions(path, fs::Permissions::from_mode(0o660))?;
+ Ok(listener)
+}
+
+async fn accept_loop(listener: UnixListener, shared: Arc<Shared>) {
+ let mut accept_failing = false;
+ loop {
+ let stream = tokio::select! {
+ accepted = listener.accept() => match accepted {
+ Ok((stream, _)) => {
+ accept_failing = false;
+ stream
+ }
+ Err(err) => {
+ // a failed accept leaves the connection queued and the listener readable, so
+ // an immediate retry would spin. A lasting cause like exhausted descriptors is
+ // logged once, not on every round
+ if !accept_failing {
+ log::warn!("accept failed: {err}");
+ accept_failing = true;
+ }
+ tokio::time::sleep(Duration::from_millis(100)).await;
+ continue;
+ }
+ },
+ () = shared.stop.notified() => return,
+ };
+ admit(stream, &shared);
+ }
+}
+
+fn admit(stream: UnixStream, shared: &Arc<Shared>) {
+ let cred = match stream.peer_cred() {
+ Ok(cred) => cred,
+ Err(err) => {
+ log::warn!("failed to get peer credentials: {err}");
+ return;
+ }
+ };
+ let privileged = cred.uid() == 0 && cred.gid() == 0;
+ if !privileged && cred.gid() != shared.gid {
+ log::warn!(
+ "connection from bad user {}/{} rejected",
+ cred.uid(),
+ cred.gid()
+ );
+ return;
+ }
+ if shared.connections.load(Ordering::Relaxed) >= MAX_CONNECTIONS {
+ log::warn!("connection limit of {MAX_CONNECTIONS} reached, refusing");
+ return;
+ }
+ // root may raise the send buffer past wmem_max, which gives a slow client room for thousands
+ // of events before it falls behind
+ if setsockopt(&stream, sockopt::SndBufForce, &SNDBUF).is_err()
+ && setsockopt(&stream, sockopt::SndBuf, &SNDBUF).is_err()
+ {
+ log::debug!("could not raise the send buffer of a connection");
+ }
+ shared.connections.fetch_add(1, Ordering::Relaxed);
+ tokio::spawn(conn::serve(stream, privileged, Slot(Arc::clone(shared))));
+}
+
+#[cfg(test)]
+mod tests {
+ use std::io::{BufRead, BufReader, Write};
+ use std::os::unix::fs::MetadataExt;
+ use std::os::unix::net::UnixStream;
+ use std::time::Instant;
+
+ use serde_json::{Value, json};
+
+ use super::*;
+
+ static COUNTER: AtomicUsize = AtomicUsize::new(0);
+
+ struct TempDir(PathBuf);
+
+ impl Drop for TempDir {
+ fn drop(&mut self) {
+ let _ = fs::remove_dir_all(&self.0);
+ }
+ }
+
+ fn start(capacity: usize) -> (Server, PathBuf, TempDir) {
+ let n = COUNTER.fetch_add(1, Ordering::SeqCst);
+ let dir = std::env::temp_dir().join(format!("pmxcfs-notify-{}-{n}", std::process::id()));
+ fs::create_dir_all(&dir).unwrap();
+ let gid = fs::metadata(&dir).unwrap().gid();
+ let path = dir.join("sock");
+ let server = Server::start(Config {
+ path: path.clone(),
+ gid,
+ capacity,
+ })
+ .unwrap();
+ (server, path, TempDir(dir))
+ }
+
+ fn connect(path: &PathBuf) -> (UnixStream, BufReader<UnixStream>) {
+ let stream = UnixStream::connect(path).unwrap();
+ stream
+ .set_read_timeout(Some(Duration::from_secs(5)))
+ .unwrap();
+ let reader = BufReader::new(stream.try_clone().unwrap());
+ (stream, reader)
+ }
+
+ fn request(stream: &mut UnixStream, reader: &mut BufReader<UnixStream>, line: &str) -> String {
+ stream.write_all(line.as_bytes()).unwrap();
+ stream.write_all(b"\n").unwrap();
+ let mut reply = String::new();
+ reader.read_line(&mut reply).unwrap();
+ reply
+ }
+
+ fn subscribe(stream: &mut UnixStream, reader: &mut BufReader<UnixStream>, patterns: &str) {
+ let line = format!("{{\"command\":\"subscribe\",\"args\":{{\"patterns\":{patterns}}}}}");
+ assert_eq!(request(stream, reader, &line), "{\"ok\":null}\n");
+ assert_eq!(event(reader), json!({ "type": "resync" }));
+ }
+
+ fn event(reader: &mut BufReader<UnixStream>) -> Value {
+ let mut line = String::new();
+ reader.read_line(&mut line).unwrap();
+ serde_json::from_str::<Value>(&line).unwrap()["event"].clone()
+ }
+
+ /// The next event if one arrives within a short timeout, None otherwise.
+ fn pending(reader: &mut BufReader<UnixStream>) -> Option<Value> {
+ let stream = reader.get_ref();
+ stream
+ .set_read_timeout(Some(Duration::from_millis(300)))
+ .unwrap();
+ let mut line = String::new();
+ let result = reader.read_line(&mut line);
+ reader
+ .get_ref()
+ .set_read_timeout(Some(Duration::from_secs(5)))
+ .unwrap();
+ match result {
+ Ok(n) if n > 0 => Some(serde_json::from_str::<Value>(&line).unwrap()["event"].clone()),
+ _ => None,
+ }
+ }
+
+ fn wait_for(mut condition: impl FnMut() -> bool) {
+ let deadline = Instant::now() + Duration::from_secs(5);
+ while !condition() {
+ assert!(Instant::now() < deadline, "condition not met in time");
+ std::thread::sleep(Duration::from_millis(10));
+ }
+ }
+
+ #[test]
+ fn subscribe_and_receive() {
+ let (server, path, _dir) = start(64);
+ let meta = fs::metadata(&path).unwrap();
+ assert_eq!(meta.mode() & 0o777, 0o660);
+
+ let (mut stream, mut reader) = connect(&path);
+ assert_eq!(
+ request(&mut stream, &mut reader, r#"{"command":"hello"}"#),
+ "{\"ok\":{\"protocol\":1}}\n"
+ );
+ assert_eq!(server.connection_count(), 1);
+ let reply = request(&mut stream, &mut reader, r#"{"command":"bogus"}"#);
+ assert!(
+ reply.starts_with("{\"error\":\"unknown variant `bogus`"),
+ "{reply}"
+ );
+ server.emit(EventKind::Write, Some("datacenter.cfg"), None);
+ subscribe(
+ &mut stream,
+ &mut reader,
+ r#"{"guest":"/nodes/{node}/qemu-server/{vmid}.conf","dc":"datacenter.cfg"}"#,
+ );
+ // a rejected subscription leaves the previous one in place
+ let reply = request(
+ &mut stream,
+ &mut reader,
+ r#"{"command":"subscribe","args":{"patterns":{"bad":"{a}{b}"}}}"#,
+ );
+ assert!(
+ reply.starts_with("{\"error\":\"only one placeholder per component in '{a}{b}'"),
+ "{reply}"
+ );
+
+ server.emit(EventKind::Write, Some("storage.cfg"), None);
+ server.emit(
+ EventKind::Create,
+ Some("/nodes/n1/qemu-server/100.conf"),
+ None,
+ );
+ server.emit(
+ EventKind::Write,
+ Some("nodes/n1/qemu-server/100.conf.tmp.1"),
+ None,
+ );
+ server.emit(
+ EventKind::Rename,
+ Some("nodes/n1/qemu-server/100.conf.tmp.1"),
+ Some("datacenter.cfg"),
+ );
+ server.emit(EventKind::Resync, None, None);
+
+ let create = event(&mut reader);
+ assert_eq!(
+ create,
+ json!({
+ "type": "create",
+ "path": "nodes/n1/qemu-server/100.conf",
+ "params": { "guest": [{ "node": "n1", "vmid": "100" }] },
+ })
+ );
+ let rename = event(&mut reader);
+ assert_eq!(rename["type"], "rename");
+ assert_eq!(rename["to"], "datacenter.cfg");
+ assert_eq!(rename["params"], json!({ "dc": [{}] }));
+ assert_eq!(event(&mut reader), json!({ "type": "resync" }));
+
+ assert_eq!(
+ request(&mut stream, &mut reader, r#"{"command":"unsubscribe"}"#),
+ "{\"ok\":null}\n"
+ );
+ server.emit(EventKind::Write, Some("datacenter.cfg"), None);
+ server.emit(EventKind::Resync, None, None);
+ // nothing emitted while unsubscribed shows up after subscribing again
+ subscribe(&mut stream, &mut reader, r#"{"dc":"datacenter.cfg"}"#);
+ server.emit(EventKind::Write, Some("datacenter.cfg"), None);
+ assert_eq!(event(&mut reader)["type"], "write");
+ assert!(pending(&mut reader).is_none());
+
+ server.shutdown();
+ let mut line = String::new();
+ assert_eq!(reader.read_line(&mut line).unwrap(), 0);
+ assert!(!path.exists());
+ }
+
+ #[test]
+ fn private_paths_follow_credentials() {
+ let (server, path, _dir) = start(64);
+ let meta = fs::metadata(&path).unwrap();
+ let privileged = meta.uid() == 0 && meta.gid() == 0;
+
+ let (mut stream, mut reader) = connect(&path);
+ subscribe(&mut stream, &mut reader, r#"{"all":"{path...}"}"#);
+ server.emit(EventKind::Write, Some("nodes/n1/priv/ssl.key"), None);
+ server.emit(EventKind::Write, Some("nodes/n1/pve-ssl.pem"), None);
+
+ if privileged {
+ assert_eq!(event(&mut reader)["path"], "nodes/n1/priv/ssl.key");
+ }
+ let public = event(&mut reader);
+ assert_eq!(public["path"], "nodes/n1/pve-ssl.pem");
+ assert_eq!(
+ public["params"],
+ json!({ "all": [{ "path": "nodes/n1/pve-ssl.pem" }] })
+ );
+ }
+
+ #[test]
+ fn client_close_is_noticed() {
+ let (server, path, _dir) = start(64);
+ let (mut stream, mut reader) = connect(&path);
+ request(&mut stream, &mut reader, r#"{"command":"hello"}"#);
+ assert_eq!(server.connection_count(), 1);
+ drop(reader);
+ drop(stream);
+ wait_for(|| server.connection_count() == 0);
+ }
+
+ #[test]
+ fn connections_are_capped() {
+ let (server, path, _dir) = start(64);
+ let mut streams: Vec<UnixStream> = (0..MAX_CONNECTIONS)
+ .map(|_| UnixStream::connect(&path).unwrap())
+ .collect();
+ wait_for(|| server.connection_count() == MAX_CONNECTIONS);
+
+ let (_stream, mut reader) = connect(&path);
+ let mut line = String::new();
+ assert_eq!(reader.read_line(&mut line).unwrap(), 0, "one too many");
+
+ drop(streams.pop());
+ wait_for(|| server.connection_count() == MAX_CONNECTIONS - 1);
+ let (mut stream, mut reader) = connect(&path);
+ assert_eq!(
+ request(&mut stream, &mut reader, r#"{"command":"hello"}"#),
+ "{\"ok\":{\"protocol\":1}}\n"
+ );
+ }
+
+ fn numbered(n: usize) -> String {
+ format!("{}/f{n}", "x".repeat(3000))
+ }
+
+ /// Fills the client's socket while it is not reading. Returns the number of events emitted,
+ /// numbered from 1 upward.
+ fn burst(server: &Server) -> usize {
+ let count = 2000;
+ for n in 1..=count {
+ server.emit(EventKind::Write, Some(&numbered(n)), None);
+ }
+ count
+ }
+
+ #[test]
+ fn a_slow_client_is_resynced() {
+ let (server, path, _dir) = start(100);
+ let (mut stream, mut reader) = connect(&path);
+ subscribe(&mut stream, &mut reader, r#"{"all":"{filler}/f{n}"}"#);
+ let count = burst(&server);
+
+ // where the client falls behind depends on scheduling. It always gets at least one resync
+ // and no event twice, and after the last resync a gapless run of events up to the newest
+ // one, if any
+ let mut events = vec![event(&mut reader)];
+ while let Some(ev) = pending(&mut reader) {
+ events.push(ev);
+ }
+ let number = |ev: &Value| -> usize {
+ ev["params"]["all"][0]["n"]
+ .as_str()
+ .unwrap()
+ .parse()
+ .unwrap()
+ };
+ let writes: Vec<usize> = events
+ .iter()
+ .filter(|ev| ev["type"] == "write")
+ .map(number)
+ .collect();
+ assert!(writes.len() < count, "nothing was missed");
+ assert!(
+ writes.windows(2).all(|w| w[0] < w[1]),
+ "an event arrived twice"
+ );
+ let last_resync = events
+ .iter()
+ .rposition(|ev| ev["type"] == "resync")
+ .expect("the client never fell behind");
+ let tail: Vec<usize> = events[last_resync + 1..].iter().map(number).collect();
+ assert!(
+ tail.windows(2).all(|w| w[1] == w[0] + 1),
+ "gap after the last resync"
+ );
+ if let Some(last) = tail.last() {
+ assert_eq!(*last, count, "the newest event never arrived");
+ }
+
+ server.emit(EventKind::Write, Some(&numbered(0)), None);
+ assert_eq!(number(&event(&mut reader)), 0);
+ assert_eq!(server.connection_count(), 1);
+ }
+
+ #[test]
+ fn a_client_that_stopped_reading_is_cut_off() {
+ // room for the whole burst, so the socket fills instead of the channel
+ let (server, path, _dir) = start(4096);
+ let (mut stream, mut reader) = connect(&path);
+ subscribe(&mut stream, &mut reader, r#"{"all":"{path...}"}"#);
+ burst(&server);
+ wait_for(|| server.connection_count() == 0);
+ // what the socket still held is readable, then EOF rather than a reset
+ let mut sink = Vec::new();
+ assert!(std::io::Read::read_to_end(&mut stream, &mut sink).is_ok());
+ }
+}
--
2.47.3
^ permalink raw reply related [flat|nested] 13+ messages in thread* [PATCH pve-cluster 06/12] rust: ffi: add C ABI staticlib for pmxcfs
2026-10-22 14:21 [PATCH cluster 00/12] pmxcfs: add a change notification socket Hannes Laimer
` (4 preceding siblings ...)
2026-10-22 14:21 ` [PATCH pve-cluster 05/12] rust: notify: add the socket server Hannes Laimer
@ 2026-10-22 14:21 ` Hannes Laimer
2026-10-22 14:22 ` [PATCH pve-cluster 07/12] pmxcfs: memdb: add change notification hook Hannes Laimer
` (5 subsequent siblings)
11 siblings, 0 replies; 13+ messages in thread
From: Hannes Laimer @ 2026-10-22 14:21 UTC (permalink / raw)
To: pve-devel
pmxcfs is a C daemon, so the notification socket gets a C interface
in a static library the daemon can link. Three functions start the
socket, emit an event and shut the socket down. Log messages go with
a syslog priority to a callback from the caller, so they can end up
in the daemon's own log. The event type crosses the boundary as the
Rust enum itself, laid out as a C enum, so nothing translates between
the two sides.
Rust aborts the process when a panic reaches the C caller. In pmxcfs
that would take /etc/pve down because of a bug in the notification
socket. So every entry point catches a panic. A panic while starting
or emitting disables notifications for the rest of the daemon's life
and shuts the socket down. Clients then see EOF and the socket file
gone, instead of waiting on a connection nobody serves.
Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
src/rust/Cargo.toml | 1 +
src/rust/pmxcfs-ffi/Cargo.toml | 17 ++
src/rust/pmxcfs-ffi/include/pmxcfs-notify.h | 24 ++
src/rust/pmxcfs-ffi/src/lib.rs | 306 ++++++++++++++++++++
src/rust/pmxcfs-notify/src/protocol.rs | 6 +-
5 files changed, 353 insertions(+), 1 deletion(-)
create mode 100644 src/rust/pmxcfs-ffi/Cargo.toml
create mode 100644 src/rust/pmxcfs-ffi/include/pmxcfs-notify.h
create mode 100644 src/rust/pmxcfs-ffi/src/lib.rs
diff --git a/src/rust/Cargo.toml b/src/rust/Cargo.toml
index 8cf928e..55e0565 100644
--- a/src/rust/Cargo.toml
+++ b/src/rust/Cargo.toml
@@ -1,5 +1,6 @@
[workspace]
members = [
+ "pmxcfs-ffi",
"pmxcfs-notify",
]
resolver = "3"
diff --git a/src/rust/pmxcfs-ffi/Cargo.toml b/src/rust/pmxcfs-ffi/Cargo.toml
new file mode 100644
index 0000000..8264cde
--- /dev/null
+++ b/src/rust/pmxcfs-ffi/Cargo.toml
@@ -0,0 +1,17 @@
+[package]
+name = "pmxcfs-ffi"
+version = "0.1.0"
+description = "C ABI bridge exposing pmxcfs-notify to the pmxcfs daemon"
+authors.workspace = true
+edition.workspace = true
+license.workspace = true
+homepage.workspace = true
+rust-version.workspace = true
+
+[lib]
+crate-type = ["staticlib"]
+
+[dependencies]
+libc.workspace = true
+log.workspace = true
+pmxcfs-notify.workspace = true
diff --git a/src/rust/pmxcfs-ffi/include/pmxcfs-notify.h b/src/rust/pmxcfs-ffi/include/pmxcfs-notify.h
new file mode 100644
index 0000000..8e0f0ce
--- /dev/null
+++ b/src/rust/pmxcfs-ffi/include/pmxcfs-notify.h
@@ -0,0 +1,24 @@
+#ifndef PMXCFS_NOTIFY_H
+#define PMXCFS_NOTIFY_H
+
+#include <sys/types.h>
+
+/* mirrored by EventKind on the Rust side, the values must stay the same */
+enum pmxcfs_notify_type {
+ PMXCFS_NOTIFY_CREATE = 0,
+ PMXCFS_NOTIFY_WRITE = 1,
+ PMXCFS_NOTIFY_MTIME = 2,
+ PMXCFS_NOTIFY_RENAME = 3,
+ PMXCFS_NOTIFY_DELETE = 4,
+ PMXCFS_NOTIFY_MKDIR = 5,
+ PMXCFS_NOTIFY_RESYNC = 6,
+};
+
+/* priority is a syslog(3) level */
+typedef void (*pmxcfs_notify_log_fn)(int priority, const char *msg);
+
+int pmxcfs_notify_init(const char *socket_path, gid_t gid, pmxcfs_notify_log_fn log_cb);
+void pmxcfs_notify_emit(enum pmxcfs_notify_type type, const char *path, const char *to);
+void pmxcfs_notify_shutdown(void);
+
+#endif /* PMXCFS_NOTIFY_H */
diff --git a/src/rust/pmxcfs-ffi/src/lib.rs b/src/rust/pmxcfs-ffi/src/lib.rs
new file mode 100644
index 0000000..7291d7d
--- /dev/null
+++ b/src/rust/pmxcfs-ffi/src/lib.rs
@@ -0,0 +1,306 @@
+//! C ABI for the pmxcfs daemon, declared in include/pmxcfs-notify.h.
+//!
+//! Rust aborts the process when a panic reaches an `extern "C"` frame, which must never happen to
+//! pmxcfs. Every entry point therefore runs under `catch_unwind`, and a panic in one switches the
+//! notifier off for the rest of the daemon's life instead of taking it down.
+
+use std::ffi::{CStr, CString, OsStr, c_char, c_int};
+use std::os::unix::ffi::OsStrExt;
+use std::panic::{UnwindSafe, catch_unwind};
+use std::path::Path;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Mutex, MutexGuard, Once};
+
+use log::{Level, LevelFilter, Log, Metadata, Record};
+use pmxcfs_notify::{Config, EventKind, Server};
+
+pub type LogFn = unsafe extern "C" fn(c_int, *const c_char);
+
+const CAPACITY: usize = 8192;
+
+static SERVER: Mutex<Option<Server>> = Mutex::new(None);
+static DISABLED: AtomicBool = AtomicBool::new(false);
+static LOG_CB: Mutex<Option<LogFn>> = Mutex::new(None);
+static LOGGER: CLogger = CLogger;
+static LOGGER_INSTALLED: Once = Once::new();
+
+struct CLogger;
+
+impl Log for CLogger {
+ fn enabled(&self, _: &Metadata) -> bool {
+ true
+ }
+
+ fn log(&self, record: &Record) {
+ let Some(callback) = *lock(&LOG_CB) else {
+ return;
+ };
+ let priority = match record.level() {
+ Level::Error => libc::LOG_ERR,
+ Level::Warn => libc::LOG_WARNING,
+ Level::Info => libc::LOG_INFO,
+ Level::Debug | Level::Trace => libc::LOG_DEBUG,
+ };
+ let msg = CString::new(record.args().to_string().replace('\0', " "))
+ .expect("no NUL left after the replace");
+ unsafe { callback(priority, msg.as_ptr()) };
+ }
+
+ fn flush(&self) {}
+}
+
+fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
+ mutex
+ .lock()
+ .unwrap_or_else(|poisoned| poisoned.into_inner())
+}
+
+fn guard<R>(what: &str, fallback: R, f: impl FnOnce() -> R + UnwindSafe) -> R {
+ if DISABLED.load(Ordering::Relaxed) {
+ return fallback;
+ }
+ match catch_unwind(f) {
+ Ok(result) => result,
+ Err(payload) => {
+ DISABLED.store(true, Ordering::Relaxed);
+ let reason = payload
+ .downcast_ref::<&str>()
+ .copied()
+ .or_else(|| payload.downcast_ref::<String>().map(String::as_str))
+ .unwrap_or("<unexpected panic value>");
+ let _ = catch_unwind(|| {
+ log::error!("{what} panicked, notifications disabled: {reason}");
+ });
+ let _ = catch_unwind(stop_server);
+ fallback
+ }
+ }
+}
+
+fn stop_server() {
+ let server = lock(&SERVER).take();
+ if let Some(server) = server {
+ server.shutdown();
+ }
+}
+
+unsafe fn c_str<'a>(ptr: *const c_char) -> Option<&'a CStr> {
+ if ptr.is_null() {
+ return None;
+ }
+ Some(unsafe { CStr::from_ptr(ptr) })
+}
+
+/// # Safety
+///
+/// `socket_path` must be NULL or point to a NUL terminated string.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn pmxcfs_notify_init(
+ socket_path: *const c_char,
+ gid: libc::gid_t,
+ log_cb: Option<LogFn>,
+) -> c_int {
+ guard("init", -1, || {
+ let Some(path) = (unsafe { c_str(socket_path) }) else {
+ return -1;
+ };
+ let path = Path::new(OsStr::from_bytes(path.to_bytes()));
+ *lock(&LOG_CB) = log_cb;
+ LOGGER_INSTALLED.call_once(|| {
+ let _ = log::set_logger(&LOGGER);
+ log::set_max_level(LevelFilter::Debug);
+ });
+
+ let mut server = lock(&SERVER);
+ if server.is_some() {
+ log::warn!("notification socket is already running");
+ return 0;
+ }
+ match Server::start(Config {
+ path: path.to_owned(),
+ gid,
+ capacity: CAPACITY,
+ }) {
+ Ok(started) => {
+ *server = Some(started);
+ log::info!("listening on {}", path.display());
+ 0
+ }
+ Err(err) => {
+ log::error!("failed to start notification socket: {err:#}");
+ -1
+ }
+ }
+ })
+}
+
+/// # Safety
+///
+/// `kind` must be a value of enum pmxcfs_notify_type, and `path` and `to` must each be NULL or
+/// point to a NUL terminated string.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn pmxcfs_notify_emit(
+ kind: EventKind,
+ path: *const c_char,
+ to: *const c_char,
+) {
+ guard("emit", (), || {
+ let server = lock(&SERVER);
+ let Some(server) = server.as_ref() else {
+ return;
+ };
+ match kind {
+ EventKind::Create
+ | EventKind::Write
+ | EventKind::Mtime
+ | EventKind::Rename
+ | EventKind::Delete
+ | EventKind::Mkdir
+ | EventKind::Resync => {}
+ // #[non_exhaustive] requires this arm. It cannot catch a bad value from C, which is
+ // undefined behavior before this point
+ _ => {
+ log::warn!("ignoring event with unknown type");
+ return;
+ }
+ }
+ let path = unsafe { c_str(path) }.map(CStr::to_str).transpose();
+ let to = unsafe { c_str(to) }.map(CStr::to_str).transpose();
+ match (path, to) {
+ (Ok(path), Ok(to)) => server.emit(kind, path, to),
+ // a JSON string holds only Unicode, so a path that is not UTF-8 is dropped rather than
+ // sent mangled
+ _ => log::warn!("ignoring {kind:?} event for a path that is not UTF-8"),
+ }
+ })
+}
+
+#[unsafe(no_mangle)]
+pub extern "C" fn pmxcfs_notify_shutdown() {
+ let _ = catch_unwind(|| {
+ stop_server();
+ *lock(&LOG_CB) = None;
+ });
+}
+
+#[cfg(test)]
+mod tests {
+ use std::io::{BufRead, BufReader, Read, Write};
+ use std::os::unix::fs::MetadataExt;
+ use std::os::unix::net::UnixStream;
+ use std::time::Duration;
+
+ use super::*;
+
+ static LINES: Mutex<Vec<(c_int, String)>> = Mutex::new(Vec::new());
+ // the tests share the global notifier state, so they run one at a time
+ static SERIAL: Mutex<()> = Mutex::new(());
+
+ unsafe extern "C" fn collect(priority: c_int, msg: *const c_char) {
+ let msg = unsafe { CStr::from_ptr(msg) }
+ .to_string_lossy()
+ .into_owned();
+ lock(&LINES).push((priority, msg));
+ }
+
+ fn warnings_containing(text: &str) -> usize {
+ lock(&LINES)
+ .iter()
+ .filter(|(prio, msg)| *prio == libc::LOG_WARNING && msg.contains(text))
+ .count()
+ }
+
+ #[test]
+ fn init_emit_shutdown_cycle() {
+ let _serial = lock(&SERIAL);
+ lock(&LINES).clear();
+ let dir = std::env::temp_dir().join(format!("pmxcfs-ffi-{}", std::process::id()));
+ std::fs::create_dir_all(&dir).unwrap();
+ let gid = std::fs::metadata(&dir).unwrap().gid();
+ let socket = CString::new(dir.join("sock").to_str().unwrap()).unwrap();
+
+ for _ in 0..2 {
+ let rc = unsafe { pmxcfs_notify_init(socket.as_ptr(), gid, Some(collect)) };
+ assert_eq!(rc, 0);
+ let path = CString::new("nodes/n1/qemu-server/100.conf").unwrap();
+ let odd = CString::new(b"nodes/n1/\xff.conf".to_vec()).unwrap();
+ unsafe {
+ pmxcfs_notify_emit(EventKind::Write, path.as_ptr(), std::ptr::null());
+ pmxcfs_notify_emit(EventKind::Resync, std::ptr::null(), std::ptr::null());
+ pmxcfs_notify_emit(EventKind::Delete, std::ptr::null(), std::ptr::null());
+ pmxcfs_notify_emit(EventKind::Rename, std::ptr::null(), path.as_ptr());
+ pmxcfs_notify_emit(EventKind::Write, odd.as_ptr(), std::ptr::null());
+ pmxcfs_notify_emit(EventKind::Rename, path.as_ptr(), odd.as_ptr());
+ }
+ pmxcfs_notify_shutdown();
+ }
+
+ let rc = unsafe { pmxcfs_notify_init(std::ptr::null(), gid, Some(collect)) };
+ assert_eq!(rc, -1);
+ let _ = std::fs::remove_dir_all(&dir);
+
+ let listening = lock(&LINES)
+ .iter()
+ .filter(|(prio, msg)| *prio == libc::LOG_INFO && msg.starts_with("listening on "))
+ .count();
+ assert_eq!(listening, 2);
+ assert_eq!(warnings_containing("Delete event without a path"), 2);
+ assert_eq!(warnings_containing("Rename event without a path"), 2);
+ assert_eq!(
+ warnings_containing("Write event for a path that is not UTF-8"),
+ 2
+ );
+ assert_eq!(
+ warnings_containing("Rename event for a path that is not UTF-8"),
+ 2
+ );
+ }
+
+ #[test]
+ fn panic_closes_the_connections() {
+ let _serial = lock(&SERIAL);
+ lock(&LINES).clear();
+ let dir = std::env::temp_dir().join(format!("pmxcfs-ffi-panic-{}", std::process::id()));
+ std::fs::create_dir_all(&dir).unwrap();
+ let gid = std::fs::metadata(&dir).unwrap().gid();
+ let path = dir.join("sock");
+ let socket = CString::new(path.to_str().unwrap()).unwrap();
+ assert_eq!(
+ unsafe { pmxcfs_notify_init(socket.as_ptr(), gid, Some(collect)) },
+ 0
+ );
+ let mut client = UnixStream::connect(&path).unwrap();
+ client
+ .set_read_timeout(Some(Duration::from_secs(5)))
+ .unwrap();
+ // a reply proves the server accepted the connection. Stopping the server after the panic
+ // below then closes an established socket, not a pending one the peer would see as a reset
+ client.write_all(b"{\"command\":\"hello\"}\n").unwrap();
+ let mut reply = String::new();
+ BufReader::new(client.try_clone().unwrap())
+ .read_line(&mut reply)
+ .unwrap();
+ assert!(reply.starts_with("{\"ok\""), "{reply}");
+
+ assert_eq!(guard("test", -1, || -> c_int { panic!("injected") }), -1);
+
+ assert!(DISABLED.load(Ordering::Relaxed));
+ assert!(lock(&SERVER).is_none());
+ assert!(!path.exists());
+ let mut sink = Vec::new();
+ assert_eq!(client.read_to_end(&mut sink).unwrap(), 0, "client sees EOF");
+ assert!(
+ lock(&LINES)
+ .iter()
+ .any(|(prio, msg)| *prio == libc::LOG_ERR && msg.contains("test panicked"))
+ );
+
+ // a disabled notifier ignores everything, including a new init
+ assert_eq!(
+ unsafe { pmxcfs_notify_init(socket.as_ptr(), gid, Some(collect)) },
+ -1
+ );
+ DISABLED.store(false, Ordering::Relaxed);
+ let _ = std::fs::remove_dir_all(&dir);
+ }
+}
diff --git a/src/rust/pmxcfs-notify/src/protocol.rs b/src/rust/pmxcfs-notify/src/protocol.rs
index d28c9dc..e706cda 100644
--- a/src/rust/pmxcfs-notify/src/protocol.rs
+++ b/src/rust/pmxcfs-notify/src/protocol.rs
@@ -10,6 +10,9 @@ 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<Captures<'a>>>;
+// Crosses the C ABI by value, so the order has to match enum pmxcfs_notify_type in the pmxcfs-ffi
+// header.
+#[repr(C)]
#[non_exhaustive]
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
#[serde(rename_all = "lowercase")]
@@ -35,7 +38,8 @@ pub(crate) struct Event<'a> {
pub params: Option<Params<'a>>,
}
-/// Everything the server sends. The variant name is the single key of the JSON object on the wire.
+/// 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> {
--
2.47.3
^ permalink raw reply related [flat|nested] 13+ messages in thread* [PATCH pve-cluster 07/12] pmxcfs: memdb: add change notification hook
2026-10-22 14:21 [PATCH cluster 00/12] pmxcfs: add a change notification socket Hannes Laimer
` (5 preceding siblings ...)
2026-10-22 14:21 ` [PATCH pve-cluster 06/12] rust: ffi: add C ABI staticlib for pmxcfs Hannes Laimer
@ 2026-10-22 14:22 ` Hannes Laimer
2026-10-22 14:22 ` [PATCH pve-cluster 08/12] buildsys: link pmxcfs against the rust notify staticlib Hannes Laimer
` (4 subsequent siblings)
11 siblings, 0 replies; 13+ messages in thread
From: Hannes Laimer @ 2026-10-22 14:22 UTC (permalink / raw)
To: pve-devel
Every change to the cluster file system, local or delivered from
another node, goes through a memdb mutation. Add a hook to memdb and
call it as the last step of each mutation, after the change is stored
and the guest list updated. An observer is then told only about stored
changes, and whatever it reads next already shows them. During a
synchronization a node can take the whole state from the cluster. It
never saw the mutations behind that state, so the hook gets a resync
instead.
The hook is a function pointer instead of a direct call into the
notifier, since the database tool and the tests link the same static
C library as the daemon. They should not pull in the socket server.
The event types come from the notifier's header, so memdb needs no
mapping of its own.
Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
src/pmxcfs/Makefile | 1 +
src/pmxcfs/database.c | 2 ++
src/pmxcfs/memdb.c | 22 ++++++++++++++++++++++
src/pmxcfs/memdb.h | 8 ++++++++
4 files changed, 33 insertions(+)
diff --git a/src/pmxcfs/Makefile b/src/pmxcfs/Makefile
index 547f258..90bc6a5 100644
--- a/src/pmxcfs/Makefile
+++ b/src/pmxcfs/Makefile
@@ -8,6 +8,7 @@ CFLAGS += -Wpedantic
CFLAGS += -g -O2
CFLAGS += -I.
CFLAGS += $(shell pkg-config --cflags $(DEPENDENCIES))
+CFLAGS += -I../rust/pmxcfs-ffi/include
LDFLAGS += -Wl,-z,relro $(shell pkg-config --libs $(DEPENDENCIES))
diff --git a/src/pmxcfs/database.c b/src/pmxcfs/database.c
index 4ee2389..36fc1d7 100644
--- a/src/pmxcfs/database.c
+++ b/src/pmxcfs/database.c
@@ -611,6 +611,8 @@ gboolean bdb_backend_commit_update(
memdb_update_locks(memdb);
+ memdb_notify(PMXCFS_NOTIFY_RESYNC, NULL, NULL);
+
result = TRUE;
ret:
diff --git a/src/pmxcfs/memdb.c b/src/pmxcfs/memdb.c
index 77d7652..3f60593 100644
--- a/src/pmxcfs/memdb.c
+++ b/src/pmxcfs/memdb.c
@@ -41,6 +41,18 @@
#define CFS_LOCK_TIMEOUT (60 * 2)
+// Set once before any other thread runs, so no locking is needed here.
+// The hook runs under memdb->mutex and must not call back into memdb.
+static memdb_notify_fn notify_hook;
+
+void memdb_set_notify_hook(memdb_notify_fn hook) { notify_hook = hook; }
+
+void memdb_notify(enum pmxcfs_notify_type type, const char *path, const char *to) {
+ if (notify_hook) {
+ notify_hook(type, path, to);
+ }
+}
+
memdb_tree_entry_t *memdb_tree_entry_new(const char *name) {
g_return_val_if_fail(name != NULL, NULL);
@@ -621,6 +633,8 @@ int memdb_mkdir(memdb_t *memdb, const char *path, guint32 writer, guint32 mtime)
}
}
+ memdb_notify(PMXCFS_NOTIFY_MKDIR, path, NULL);
+
ret = 0;
ret:
@@ -824,6 +838,8 @@ static int memdb_pwrite(
vmlist_register_vm(vmtype, vmid, nodename);
}
+ memdb_notify(old ? PMXCFS_NOTIFY_WRITE : PMXCFS_NOTIFY_CREATE, path, NULL);
+
ret = count;
ret:
@@ -929,6 +945,8 @@ int memdb_mtime(memdb_t *memdb, const char *path, guint32 writer, guint32 mtime)
}
}
+ memdb_notify(PMXCFS_NOTIFY_MTIME, path, NULL);
+
ret = 0;
ret:
@@ -1191,6 +1209,8 @@ int memdb_rename(memdb_t *memdb, const char *from, const char *to, guint32 write
/* directories are alwayse empty (see unlink_tree_entry) */
}
+ memdb_notify(PMXCFS_NOTIFY_RENAME, from, to);
+
ret = 0;
ret:
@@ -1264,6 +1284,8 @@ int memdb_delete(memdb_t *memdb, const char *path, guint32 writer, guint32 mtime
vmlist_delete_vm(vmid);
}
+ memdb_notify(PMXCFS_NOTIFY_DELETE, path, NULL);
+
ret = 0;
ret:
diff --git a/src/pmxcfs/memdb.h b/src/pmxcfs/memdb.h
index bca8967..28e7c28 100644
--- a/src/pmxcfs/memdb.h
+++ b/src/pmxcfs/memdb.h
@@ -27,6 +27,8 @@
#include <glib.h>
#include <sys/statvfs.h>
+#include "pmxcfs-notify.h"
+
#define MEMDB_MAX_FILE_SIZE (1024 * 1024) // 1 MiB
#define MEMDB_MAX_FSSIZE (128 * 1024 * 1024) // 128 MiB
#define MEMDB_MAX_INODES (256 * 1024) // 256k
@@ -83,6 +85,8 @@ typedef struct {
db_backend_t *bdb;
} memdb_t;
+typedef void (*memdb_notify_fn)(enum pmxcfs_notify_type type, const char *path, const char *to);
+
memdb_t *memdb_open(const char *dbfilename);
void memdb_close(memdb_t *memdb);
@@ -138,6 +142,10 @@ memdb_tree_entry_t *memdb_getattr(memdb_t *memdb, const char *path);
int memdb_rename(memdb_t *memdb, const char *from, const char *to, guint32 writer, guint32 mtime);
+void memdb_set_notify_hook(memdb_notify_fn hook);
+
+void memdb_notify(enum pmxcfs_notify_type type, const char *path, const char *to);
+
void memdb_dump(memdb_t *memdb);
gboolean
--
2.47.3
^ permalink raw reply related [flat|nested] 13+ messages in thread* [PATCH pve-cluster 08/12] buildsys: link pmxcfs against the rust notify staticlib
2026-10-22 14:21 [PATCH cluster 00/12] pmxcfs: add a change notification socket Hannes Laimer
` (6 preceding siblings ...)
2026-10-22 14:22 ` [PATCH pve-cluster 07/12] pmxcfs: memdb: add change notification hook Hannes Laimer
@ 2026-10-22 14:22 ` Hannes Laimer
2026-10-22 14:22 ` [PATCH pve-cluster 09/12] pmxcfs: notify: emit change events over the notification socket Hannes Laimer
` (3 subsequent siblings)
11 siblings, 0 replies; 13+ messages in thread
From: Hannes Laimer @ 2026-10-22 14:22 UTC (permalink / raw)
To: pve-devel
The C convenience library calls nothing in the staticlib, so the
database tool and the test programs link as before. The pmxcfs
Makefile builds the staticlib itself, so a standalone make in
src/pmxcfs keeps working. The staticlib's prerequisites come from the
dependency file cargo writes, so a changed Rust source rebuilds it.
Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
src/pmxcfs/Makefile | 19 +++++++++++++++++--
1 file changed, 17 insertions(+), 2 deletions(-)
diff --git a/src/pmxcfs/Makefile b/src/pmxcfs/Makefile
index 90bc6a5..e05b4e3 100644
--- a/src/pmxcfs/Makefile
+++ b/src/pmxcfs/Makefile
@@ -1,3 +1,4 @@
+include ../defines.mk
DEPENDENCIES=libcpg libcmap libcorosync_common libquorum libqb glib-2.0 fuse sqlite3 librrd
@@ -12,9 +13,18 @@ CFLAGS += -I../rust/pmxcfs-ffi/include
LDFLAGS += -Wl,-z,relro $(shell pkg-config --libs $(DEPENDENCIES))
+NOTIFY_LIB := $(abspath ../rust/$(RUST_COMPILE_DIR)/libpmxcfs_ffi.a)
+# from `cargo rustc -p pmxcfs-ffi --release -- --print native-static-libs`
+NOTIFY_LIBS := -lgcc_s -lutil -lrt -lpthread -lm -ldl -lc
+
AR = ar
ARFLAGS = crs
+-include $(NOTIFY_LIB:.a=.d)
+# a source file deleted since cargo wrote the dep file has no rule, and make
+# would stop before the rebuild that drops it from the list
+%.rs: ;
+
.c.o:
$(CC) $(CFLAGS) -c -o $@ $< -MMD -MT $@ -MF $@.d
@@ -26,9 +36,14 @@ libpmxcfs.a: cfs-utils.o memdb.o database.o
libpmxcfs.a:
$(AR) $(ARFLAGS) $@ $^
-pmxcfs: pmxcfs.o libpmxcfs.a
- $(CC) -o $@ $^ $(LDFLAGS)
+pmxcfs: pmxcfs.o libpmxcfs.a $(NOTIFY_LIB)
+ $(CC) -o $@ $^ $(LDFLAGS) $(NOTIFY_LIBS)
+$(NOTIFY_LIB): ../rust/Cargo.toml ../rust/.cargo/config.toml \
+ ../rust/pmxcfs-ffi/Cargo.toml ../rust/pmxcfs-notify/Cargo.toml
+ $(MAKE) -C ../rust all
+.PHONY: notify-lib
+notify-lib: $(NOTIFY_LIB)
create_pmxcfs_db: create_pmxcfs_db.o libpmxcfs.a
$(CC) $(CFLAGS) -o $@ $^ $(LDFLAGS)
--
2.47.3
^ permalink raw reply related [flat|nested] 13+ messages in thread* [PATCH pve-cluster 09/12] pmxcfs: notify: emit change events over the notification socket
2026-10-22 14:21 [PATCH cluster 00/12] pmxcfs: add a change notification socket Hannes Laimer
` (7 preceding siblings ...)
2026-10-22 14:22 ` [PATCH pve-cluster 08/12] buildsys: link pmxcfs against the rust notify staticlib Hannes Laimer
@ 2026-10-22 14:22 ` Hannes Laimer
2026-10-22 14:22 ` [PATCH pve-cluster 10/12] cfs: add perl client for the change " Hannes Laimer
` (2 subsequent siblings)
11 siblings, 0 replies; 13+ messages in thread
From: Hannes Laimer @ 2026-10-22 14:22 UTC (permalink / raw)
To: pve-devel
The daemon serves the socket at /run/pve-cluster/pmxcfs.sock to root
and the www-data group. The filesystem works without the socket, so
the daemon only logs a failure to start it. The notifier runs on a
thread of its own, so it starts after the daemonize fork.
The notifier's log messages go through the daemon's logging under
their own domain, so they land in the same journal stream as the
daemon's other messages.
Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
src/pmxcfs/cfs-utils.h | 4 ++++
src/pmxcfs/pmxcfs.c | 24 ++++++++++++++++++++++++
2 files changed, 28 insertions(+)
diff --git a/src/pmxcfs/cfs-utils.h b/src/pmxcfs/cfs-utils.h
index ed67264..29f844e 100644
--- a/src/pmxcfs/cfs-utils.h
+++ b/src/pmxcfs/cfs-utils.h
@@ -78,6 +78,10 @@ void ipc_log_fn(const char *file, int32_t line, int32_t severity, const char *ms
cfs_log(G_LOG_DOMAIN, G_LOG_LEVEL_CRITICAL, __FILE__, __LINE__, G_STRFUNC, __VA_ARGS__)
#define cfs_dom_critical(domain, ...) \
cfs_log(domain, G_LOG_LEVEL_CRITICAL, __FILE__, __LINE__, G_STRFUNC, __VA_ARGS__)
+#define cfs_warn(...) \
+ cfs_log(G_LOG_DOMAIN, G_LOG_LEVEL_WARNING, __FILE__, __LINE__, G_STRFUNC, __VA_ARGS__)
+#define cfs_dom_warn(domain, ...) \
+ cfs_log(domain, G_LOG_LEVEL_WARNING, __FILE__, __LINE__, G_STRFUNC, __VA_ARGS__)
#define cfs_message(...) \
cfs_log(G_LOG_DOMAIN, G_LOG_LEVEL_MESSAGE, __FILE__, __LINE__, G_STRFUNC, __VA_ARGS__)
#define cfs_dom_message(domain, ...) \
diff --git a/src/pmxcfs/pmxcfs.c b/src/pmxcfs/pmxcfs.c
index 14f1168..34bd4be 100644
--- a/src/pmxcfs/pmxcfs.c
+++ b/src/pmxcfs/pmxcfs.c
@@ -39,6 +39,7 @@
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/utsname.h>
+#include <syslog.h>
#include <unistd.h>
#include <qb/qbdefs.h>
@@ -51,6 +52,7 @@
#include "confdb.h"
#include "dcdb.h"
#include "dfsm.h"
+#include "pmxcfs-notify.h"
#include "quorum.h"
#include "server.h"
#include "status.h"
@@ -58,6 +60,7 @@
#define DBFILENAME VARLIBDIR "/config.db"
#define LOCKFILE VARLIBDIR "/.pmxcfs.lockfile"
#define RESTART_FLAG_FILE RUNDIR "/cfs-restart-flag"
+#define NOTIFY_SOCKET RUNDIR "/pmxcfs.sock"
#define CFSDIR "/etc/pve"
@@ -78,6 +81,18 @@ static void glib_log_handler(
cfs_log(log_domain, log_level, NULL, 0, NULL, "%s", message);
}
+static void notify_log_cb(int priority, const char *msg) {
+ if (priority <= LOG_ERR) {
+ cfs_dom_critical("notify", "%s", msg);
+ } else if (priority <= LOG_WARNING) {
+ cfs_dom_warn("notify", "%s", msg);
+ } else if (priority <= LOG_INFO) {
+ cfs_dom_message("notify", "%s", msg);
+ } else {
+ cfs_dom_debug("notify", "%s", msg);
+ }
+}
+
static gboolean write_pidfile(pid_t pid) {
char *strpid = g_strdup_printf("%d\n", pid);
gboolean res = atomic_write_file(CFS_PID_FN, strpid, strlen(strpid), 0644, getgid());
@@ -1032,6 +1047,13 @@ int main(int argc, char *argv[]) {
cfs_loop_add_service(corosync_loop, service_status, QB_LOOP_LOW);
}
+ memdb_set_notify_hook(pmxcfs_notify_emit);
+
+ // must run before any other thread starts, since the bind changes the process umask
+ if (pmxcfs_notify_init(NOTIFY_SOCKET, cfs.gid, notify_log_cb) != 0) {
+ cfs_critical("running without change notification socket");
+ }
+
cfs_loop_start_worker(corosync_loop);
server_start(memdb);
@@ -1053,6 +1075,8 @@ int main(int argc, char *argv[]) {
server_stop();
+ pmxcfs_notify_shutdown();
+
fuse_unmount(CFSDIR, fuse_chan);
fuse_destroy(fuse);
--
2.47.3
^ permalink raw reply related [flat|nested] 13+ messages in thread* [PATCH pve-cluster 10/12] cfs: add perl client for the change notification socket
2026-10-22 14:21 [PATCH cluster 00/12] pmxcfs: add a change notification socket Hannes Laimer
` (8 preceding siblings ...)
2026-10-22 14:22 ` [PATCH pve-cluster 09/12] pmxcfs: notify: emit change events over the notification socket Hannes Laimer
@ 2026-10-22 14:22 ` Hannes Laimer
2026-10-22 14:22 ` [PATCH pve-cluster 11/12] cfs: add hook registry for change notification consumers Hannes Laimer
2026-10-22 14:22 ` [PATCH pve-cluster 12/12] cfs: add runner executing cluster change hooks in children Hannes Laimer
11 siblings, 0 replies; 13+ messages in thread
From: Hannes Laimer @ 2026-10-22 14:22 UTC (permalink / raw)
To: pve-devel
Daemons that want to react to cluster-wide config changes all need to
connect to the pmxcfs notification socket, subscribe with path
templates, wait for events, and recover when pmxcfs restarts. A shared
client module handles that, so every consumer gets the reconnect
handling right instead of reimplementing it.
The client retries a lost connection and a failed connect on its own,
with a growing backoff between failed attempts. An error reply from
the daemon or an unsupported protocol version makes the call die
instead, since a retry would get the same answer. Two processes
reading one stream would split its lines between them, so a forked
child opens a connection of its own.
Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
debian/pve-cluster.install | 1 +
src/PVE/Cluster/Makefile | 2 +-
src/PVE/Cluster/Watch.pm | 325 ++++++++++++++++++++++++++++++++++
src/test/Makefile | 6 +-
src/test/watch_client_test.pl | 229 ++++++++++++++++++++++++
5 files changed, 561 insertions(+), 2 deletions(-)
create mode 100644 src/PVE/Cluster/Watch.pm
create mode 100644 src/test/watch_client_test.pl
diff --git a/debian/pve-cluster.install b/debian/pve-cluster.install
index f66cd06..77e3244 100644
--- a/debian/pve-cluster.install
+++ b/debian/pve-cluster.install
@@ -5,4 +5,5 @@ usr/lib/
usr/share/man/man8/pmxcfs.8
usr/share/perl5/PVE/Cluster.pm
usr/share/perl5/PVE/Cluster/IPCConst.pm
+usr/share/perl5/PVE/Cluster/Watch.pm
usr/share/perl5/PVE/IPCC.pm
diff --git a/src/PVE/Cluster/Makefile b/src/PVE/Cluster/Makefile
index 3f920cb..7beb976 100644
--- a/src/PVE/Cluster/Makefile
+++ b/src/PVE/Cluster/Makefile
@@ -1,6 +1,6 @@
PVEDIR=$(DESTDIR)/usr/share/perl5/PVE
-SOURCES=IPCConst.pm Setup.pm
+SOURCES=IPCConst.pm Setup.pm Watch.pm
.PHONY: install
install: $(SOURCES)
diff --git a/src/PVE/Cluster/Watch.pm b/src/PVE/Cluster/Watch.pm
new file mode 100644
index 0000000..dde8711
--- /dev/null
+++ b/src/PVE/Cluster/Watch.pm
@@ -0,0 +1,325 @@
+package PVE::Cluster::Watch;
+
+use strict;
+use warnings;
+
+use IO::Select;
+use IO::Socket::UNIX;
+use JSON;
+use Socket qw(SOCK_STREAM MSG_NOSIGNAL);
+use Time::HiRes qw(time sleep);
+
+my $default_socket = '/run/pve-cluster/pmxcfs.sock';
+my $protocol_version = 1;
+my $io_timeout = 10;
+my $min_backoff = 1;
+my $max_backoff = 30;
+
+sub new {
+ my ($class, %param) = @_;
+
+ my $self = bless {
+ socket_path => $param{socket} // $default_socket,
+ patterns => {},
+ pid => $$,
+ }, $class;
+ $self->reset();
+
+ return $self;
+}
+
+# Disconnected, with a connect due right away.
+sub reset {
+ my ($self) = @_;
+
+ $self->{sock} = undef;
+ $self->{buf} = '';
+ $self->{queue} = [];
+ $self->{established} = 0;
+ $self->{next_connect} = 0;
+ $self->{backoff} = $min_backoff;
+ $self->{warned} = 0;
+
+ return;
+}
+
+# Two processes reading one connection would split its lines between them,
+# so a forked child connects on its own, as IPCC does. It closes the
+# inherited copy without a shutdown, which would cut off the parent as well.
+# A child that keeps running should close the watch, otherwise that copy
+# stays open until the child next calls a method of the watch.
+sub check_fork {
+ my ($self) = @_;
+
+ return if $self->{pid} == $$;
+
+ $self->{sock}->close() if $self->{sock};
+ $self->{pid} = $$;
+ $self->reset();
+
+ return;
+}
+
+# Takes effect when connected and is replayed on every reconnect. Dies if
+# the daemon refuses it. A connection lost while sending it is dropped, and
+# the next wait reconnects.
+sub subscribe {
+ my ($self, %param) = @_;
+
+ $self->check_fork();
+
+ my $previous = $self->{patterns};
+ $self->{patterns} = { ($param{patterns} // {})->%* };
+
+ return if !$self->{sock};
+
+ my $error = eval { $self->send_subscription() };
+ if ($@) {
+ chomp(my $reason = $@);
+ $self->disconnect($reason) if $self->{sock};
+ return;
+ }
+ # the daemon keeps serving the previous subscription, so keep its
+ # patterns for the next reconnect
+ if (defined($error)) {
+ $self->{patterns} = $previous;
+ die "subscription refused: $error\n";
+ }
+
+ return;
+}
+
+sub connected {
+ my ($self) = @_;
+
+ $self->check_fork();
+
+ return defined($self->{sock});
+}
+
+# A reconnect replaces the socket, possibly under the same number, so a
+# caller watching it has to look it up again after every wait. It is undef
+# while disconnected, and only wait reconnects.
+sub fd {
+ my ($self) = @_;
+
+ $self->check_fork();
+
+ return $self->{sock} ? fileno($self->{sock}) : undef;
+}
+
+sub close {
+ my ($self) = @_;
+
+ $self->check_fork();
+
+ $self->{sock}->close() if $self->{sock};
+ $self->{sock} = undef;
+ $self->{established} = 0;
+ $self->{buf} = '';
+
+ return;
+}
+
+# Returns the queued events as soon as there are any, or none once $timeout
+# seconds have passed or a signal has arrived. A due reconnect also happens
+# here and is the one way a zero timeout can block. Dies if the daemon
+# refuses the handshake, since a retry would not change that.
+sub wait {
+ my ($self, $timeout) = @_;
+
+ $self->check_fork();
+
+ my $deadline = time() + ($timeout // 0);
+
+ while (1) {
+ if (scalar($self->{queue}->@*)) {
+ my @events = $self->{queue}->@*;
+ $self->{queue} = [];
+ return @events;
+ }
+
+ my $now = time();
+ if (!$self->{sock}) {
+ if ($now >= $self->{next_connect}) {
+ $self->connect();
+ next;
+ }
+ my $until = $self->{next_connect} < $deadline ? $self->{next_connect} : $deadline;
+ return if $until <= $now;
+ sleep($until - $now);
+ # a signal cut the sleep short, and a caller whose handler only
+ # sets a flag needs control back to act on it
+ return if time() < $until;
+ next;
+ }
+
+ my $remaining = $deadline - $now;
+ $remaining = 0 if $remaining < 0;
+
+ return if !IO::Select->new($self->{sock})->can_read($remaining);
+ $self->read_messages();
+ }
+}
+
+sub connect {
+ my ($self) = @_;
+
+ # connect blocks while the daemon's backlog is full, as when it cannot
+ # accept for a while. The timeout bounds that, and a failed attempt is
+ # retried later
+ my $sock = IO::Socket::UNIX->new(
+ Type => SOCK_STREAM,
+ Peer => $self->{socket_path},
+ Timeout => $io_timeout,
+ );
+ if (!$sock) {
+ $self->connect_failed("connect to $self->{socket_path} failed: $!");
+ return;
+ }
+
+ $self->{sock} = $sock;
+
+ my $error = eval { $self->handshake() };
+ if (my $reason = $@) {
+ $self->close();
+ $self->connect_failed("handshake with $self->{socket_path} failed: $reason");
+ return;
+ }
+ if (defined($error)) {
+ $self->close();
+ $self->delay_retry();
+ die "handshake with $self->{socket_path} refused: $error\n";
+ }
+
+ $self->{established} = 1;
+ $self->{backoff} = $min_backoff;
+ $self->{warned} = 0;
+
+ return;
+}
+
+sub handshake {
+ my ($self) = @_;
+
+ my ($hello, $error) = $self->request('hello');
+ return $error if defined($error);
+
+ my $version = ref($hello) eq 'HASH' ? $hello->{protocol} : undef;
+ return "unsupported protocol version " . ($version // 'unknown')
+ if !defined($version) || $version != $protocol_version;
+
+ return $self->send_subscription();
+}
+
+sub connect_failed {
+ my ($self, $msg) = @_;
+
+ if (!$self->{warned}) {
+ chomp $msg;
+ warn "$msg, retrying\n";
+ $self->{warned} = 1;
+ }
+
+ $self->delay_retry();
+
+ return;
+}
+
+sub delay_retry {
+ my ($self) = @_;
+
+ $self->{next_connect} = time() + $self->{backoff};
+ $self->{backoff} *= 2;
+ $self->{backoff} = $max_backoff if $self->{backoff} > $max_backoff;
+
+ return;
+}
+
+# A connection lost before its handshake completed is reported by the
+# connect attempt, so a daemon that accepts and then closes causes one
+# warning, not two.
+sub disconnect {
+ my ($self, $reason) = @_;
+
+ my $established = $self->{established};
+ $self->close();
+ return if !$established;
+
+ warn "notification socket $reason, reconnecting\n";
+ $self->{next_connect} = time();
+ $self->{backoff} = $min_backoff;
+
+ return;
+}
+
+sub send_subscription {
+ my ($self) = @_;
+
+ my (undef, $error) = $self->request('subscribe', { patterns => $self->{patterns} });
+
+ return $error;
+}
+
+# Sends a request and returns its reply as (data, error), queueing any events
+# that arrive meanwhile. Dies when no reply comes in time or the connection
+# is lost.
+sub request {
+ my ($self, $command, $args) = @_;
+
+ my $line = encode_json({ command => $command, ($args ? (args => $args) : ()) }) . "\n";
+ my $written = 0;
+ while ($written < length($line)) {
+ my $len = send($self->{sock}, substr($line, $written), MSG_NOSIGNAL);
+ next if !defined($len) && $!{EINTR};
+ die "write failed: $!\n" if !defined($len);
+ $written += $len;
+ }
+
+ my $deadline = time() + $io_timeout;
+ while (1) {
+ my $remaining = $deadline - time();
+ die "timeout waiting for reply to '$command'\n" if $remaining <= 0;
+ # bounds the wait, since the read itself would block for as long as
+ # the daemon stays silent
+ next if !IO::Select->new($self->{sock})->can_read($remaining);
+
+ for my $msg ($self->read_messages()) {
+ return ($msg->{ok}, undef) if exists $msg->{ok};
+ return (undef, $msg->{error} // 'unknown error') if exists $msg->{error};
+ }
+ die "connection lost while waiting for reply to '$command'\n" if !$self->{sock};
+ }
+}
+
+sub read_messages {
+ my ($self) = @_;
+
+ my $len;
+ do {
+ # appended to what the last read left unfinished
+ $len = sysread($self->{sock}, $self->{buf}, 65536, length($self->{buf}));
+ } while (!defined($len) && $!{EINTR});
+ if (!$len) {
+ $self->disconnect(defined($len) ? 'closed' : "read failed: $!");
+ return;
+ }
+
+ my @replies;
+ while ($self->{buf} =~ s/^([^\n]*)\n//) {
+ my $msg = eval { decode_json($1) };
+ if (ref($msg) ne 'HASH') {
+ $self->disconnect('sent a malformed line');
+ return;
+ }
+ if (my $event = $msg->{event}) {
+ push $self->{queue}->@*, $event;
+ } else {
+ push @replies, $msg;
+ }
+ }
+
+ return @replies;
+}
+
+1;
diff --git a/src/test/Makefile b/src/test/Makefile
index cdd37d0..7b36fca 100644
--- a/src/test/Makefile
+++ b/src/test/Makefile
@@ -4,7 +4,7 @@ cpgtest: cpgtest.c
gcc -Wall cpgtest.c $(shell pkg-config --cflags --libs libcpg libqb) -o cpgtest
.PHONY: check install clean distclean
-check: corosync-parser-test test-mac-prefix
+check: corosync-parser-test test-mac-prefix watch-client-test
.PHONY: corosync-parser-test
corosync-parser-test:
@@ -14,5 +14,9 @@ corosync-parser-test:
test-mac-prefix:
perl test_mac_prefix.pl
+.PHONY: watch-client-test
+watch-client-test:
+ perl watch_client_test.pl
+
distclean: clean
clean:
diff --git a/src/test/watch_client_test.pl b/src/test/watch_client_test.pl
new file mode 100644
index 0000000..ef8159e
--- /dev/null
+++ b/src/test/watch_client_test.pl
@@ -0,0 +1,229 @@
+#!/usr/bin/perl
+
+use lib '..';
+
+use strict;
+use warnings;
+
+use File::Temp qw(tempdir);
+use IO::Socket::UNIX;
+use JSON;
+use Socket qw(SOCK_STREAM);
+use Test::More;
+use Time::HiRes qw(sleep time);
+
+use PVE::Cluster::Watch;
+
+my $dir = tempdir(CLEANUP => 1);
+my $path = "$dir/sock";
+
+my $listener = IO::Socket::UNIX->new(Type => SOCK_STREAM, Local => $path, Listen => 1)
+ or die "listen failed: $!\n";
+
+# A stand-in for pmxcfs that serves ten connections in a row. The fourth
+# refuses the subscription, the fifth hangs up after answering the hello,
+# the seventh refuses a second subscription on a live connection and the
+# last three hang up before answering anything. The others answer the
+# handshake, send the resync, mirror the subscription back inside two
+# events and hang up, which forces the client through its reconnect path.
+my $server = fork() // die "fork failed: $!\n";
+if (!$server) {
+ for my $round (1 .. 10) {
+ my $conn = $listener->accept() or die "accept failed: $!\n";
+ if ($round >= 8) {
+ close($conn);
+ next;
+ }
+ my $subscribed = 0;
+ while (my $line = <$conn>) {
+ my $req = decode_json($line);
+ if ($req->{command} eq 'hello') {
+ print $conn encode_json({ ok => { protocol => 1 } }), "\n";
+ } elsif ($round == 5) {
+ last;
+ } elsif ($req->{command} eq 'subscribe') {
+ if ($round == 4) {
+ print $conn encode_json({ error => "pattern 'guest': no good" }), "\n";
+ last;
+ }
+ if ($round == 7 && $subscribed) {
+ print $conn encode_json({ error => "pattern 'bad': no good" }), "\n";
+ next;
+ }
+ $subscribed = 1;
+ my $patterns = $req->{args}->{patterns};
+ print $conn encode_json({ ok => undef }), "\n";
+ print $conn encode_json({ event => { type => 'resync' } }), "\n";
+ next if $round == 7;
+ print $conn encode_json({
+ event => {
+ type => 'write',
+ path => 'x',
+ params =>
+ { map { $_ => [{ template => $patterns->{$_} }] } keys %$patterns },
+ },
+ }),
+ "\n";
+ print $conn encode_json({
+ event => {
+ type => 'rename',
+ path => 'old',
+ to => join(',', sort keys %$patterns),
+ },
+ }),
+ "\n";
+ last;
+ } else {
+ print $conn encode_json({ error => "unknown command '$req->{command}'" }), "\n";
+ }
+ }
+ close($conn);
+ }
+ exit(0);
+}
+close($listener);
+
+my $collect = sub {
+ my ($watch, $count, $timeout) = @_;
+ my @events;
+ my $deadline = time() + 10;
+ while (scalar(@events) < $count && time() < $deadline) {
+ push @events, $watch->wait($timeout);
+ sleep(0.05) if !$timeout;
+ }
+ return \@events;
+};
+
+my $watch = PVE::Cluster::Watch->new(socket => $path);
+ok(!$watch->connected(), 'not connected before the first wait');
+
+my $template = 'nodes/{node}/qemu-server/{vmid}.conf';
+$watch->subscribe(patterns => { guest => $template });
+
+my $write = {
+ type => 'write',
+ path => 'x',
+ params => { guest => [{ template => $template }] },
+};
+my $rename = { type => 'rename', path => 'old', to => 'guest' };
+my $expected = [{ type => 'resync' }, $write, $rename];
+
+is_deeply($collect->($watch, 3, 1), $expected, 'connect gives a resync followed by the events');
+ok(defined($watch->fd()), 'a descriptor is available while connected');
+
+$watch->{pid} = -1; # pretend the object was inherited across a fork
+is_deeply(
+ $collect->($watch, 3, 0),
+ $expected,
+ 'a forked child gets a connection of its own, a zero timeout drains it',
+);
+
+is_deeply(
+ $collect->($watch, 3, 1),
+ $expected,
+ 'reconnect after hangup replays the subscription and gives a resync again',
+);
+
+# the server has hung up on that connection by now. Writing to it must
+# neither raise SIGPIPE nor die, and the next wait reconnects
+sleep(0.2);
+$watch->subscribe(patterns => { guest => $template });
+ok(!$watch->connected(), 'a subscription on a lost connection drops it');
+
+eval { $watch->wait(1) };
+like($@, qr/refused: pattern 'guest': no good/, 'a refused subscription makes the wait die');
+ok(!$watch->connected(), 'and its connection is closed');
+
+# a handshake cut after the hello, rounds five and six
+{
+ my @warnings;
+ local $SIG{__WARN__} = sub { push @warnings, $_[0] };
+ my $half = PVE::Cluster::Watch->new(socket => $path);
+ $half->subscribe(patterns => { guest => $template });
+ is_deeply(
+ $collect->($half, 3, 1),
+ $expected,
+ 'a connect that is cut after the hello is retried and gives a resync',
+ );
+ is(scalar(@warnings), 1, 'and the cut handshake is reported');
+}
+
+# a refused subscription on a live connection, round seven. The connection
+# stays up after the refusal, as it does with pmxcfs
+{
+ my $live = PVE::Cluster::Watch->new(socket => $path);
+ $live->subscribe(patterns => { guest => $template });
+ is_deeply(
+ $collect->($live, 1, 1),
+ [{ type => 'resync' }],
+ 'connected with the first subscription',
+ );
+ is_deeply(
+ [$live->request('bogus')],
+ [undef, "unknown command 'bogus'"],
+ 'an error reply outside the handshake comes back as the error',
+ );
+ eval { $live->subscribe(patterns => { bad => 'x/{y}' }) };
+ like($@, qr/subscription refused: pattern 'bad': no good/, 'a refusal makes subscribe die');
+ is_deeply($live->{patterns}, { guest => $template }, 'and the previous patterns stay');
+ ok($live->connected(), 'on a connection that stays up');
+}
+
+# a daemon that accepts and hangs up immediately, rounds eight to ten
+{
+ my @warnings;
+ local $SIG{__WARN__} = sub { push @warnings, $_[0] };
+ my $cut = PVE::Cluster::Watch->new(socket => $path);
+ $cut->subscribe(patterns => { guest => $template });
+ my @none = $cut->wait(5);
+ is(scalar(@none), 0, 'nothing arrives from a daemon that hangs up immediately');
+ is(scalar(@warnings), 1, 'which is reported once and retried with backoff');
+ ok(!$cut->connected(), 'and leaves the client disconnected');
+}
+
+# a signal that cuts a wait short while the client waits for its next
+# connect attempt
+{
+ my @warnings;
+ local $SIG{__WARN__} = sub { push @warnings, $_[0] };
+ my $flagged = 0;
+ local $SIG{USR1} = sub { $flagged = 1 };
+
+ my $parent = $$;
+ my $signaller = fork() // die "fork failed: $!\n";
+ if (!$signaller) {
+ sleep(0.3);
+ kill 'USR1', $parent;
+ exit(0);
+ }
+
+ my $gone = PVE::Cluster::Watch->new(socket => "$dir/gone");
+ my $started = time();
+ my @none = $gone->wait(3);
+ my $elapsed = time() - $started;
+ waitpid($signaller, 0);
+
+ is(scalar(@none), 0, 'a wait cut short by a signal returns no events');
+ ok($flagged, 'the handler of the caller has run');
+ # a signal that lands after the first backoff sleep ended only cuts the
+ # second one short, so the bound has to sit above one backoff and still
+ # below the deadline
+ ok(
+ $elapsed < 2.5,
+ sprintf('and the wait gave up after %.2fs, short of its deadline', $elapsed),
+ );
+ is(scalar(@warnings), 1, 'the socket that is not there is reported once');
+}
+
+# a shortfall in connections must fail the test, not hang it
+local $SIG{ALRM} = sub { kill 'KILL', $server; die "fake server did not finish\n" };
+alarm(30);
+waitpid($server, 0);
+alarm(0);
+is($?, 0, 'fake server ran all rounds cleanly');
+
+my @none = $watch->wait(1);
+is(scalar(@none), 0, 'no events once the server is gone');
+ok(!$watch->connected(), 'disconnected after the server went away');
+
+done_testing();
--
2.47.3
^ permalink raw reply related [flat|nested] 13+ messages in thread* [PATCH pve-cluster 11/12] cfs: add hook registry for change notification consumers
2026-10-22 14:21 [PATCH cluster 00/12] pmxcfs: add a change notification socket Hannes Laimer
` (9 preceding siblings ...)
2026-10-22 14:22 ` [PATCH pve-cluster 10/12] cfs: add perl client for the change " Hannes Laimer
@ 2026-10-22 14:22 ` Hannes Laimer
2026-10-22 14:22 ` [PATCH pve-cluster 12/12] cfs: add runner executing cluster change hooks in children Hannes Laimer
11 siblings, 0 replies; 13+ messages in thread
From: Hannes Laimer @ 2026-10-22 14:22 UTC (permalink / raw)
To: pve-devel
A hook has a name, a path template and code, registered like an API
method. The registry collects the templates into one subscription and
turns each event into runs, one per hook and set of parameters, keyed
by both. Runs with the same key differ only in the event behind them,
so a burst of changes to one file needs no more than one run.
The registry filters on the event type, since a subscription only
carries templates. It checks templates at registration against the
daemon's rules, so a bad one fails where the hook is written. A resync
carries no path, so every hook gets one run without parameters.
Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
debian/pve-cluster.install | 1 +
src/PVE/Cluster/Hooks.pm | 135 +++++++++++++++++++++++
src/PVE/Cluster/Makefile | 2 +-
src/test/Makefile | 6 +-
src/test/hooks_test.pl | 219 +++++++++++++++++++++++++++++++++++++
5 files changed, 361 insertions(+), 2 deletions(-)
create mode 100644 src/PVE/Cluster/Hooks.pm
create mode 100644 src/test/hooks_test.pl
diff --git a/debian/pve-cluster.install b/debian/pve-cluster.install
index 77e3244..b8ebff7 100644
--- a/debian/pve-cluster.install
+++ b/debian/pve-cluster.install
@@ -4,6 +4,7 @@ usr/bin/pmxcfs
usr/lib/
usr/share/man/man8/pmxcfs.8
usr/share/perl5/PVE/Cluster.pm
+usr/share/perl5/PVE/Cluster/Hooks.pm
usr/share/perl5/PVE/Cluster/IPCConst.pm
usr/share/perl5/PVE/Cluster/Watch.pm
usr/share/perl5/PVE/IPCC.pm
diff --git a/src/PVE/Cluster/Hooks.pm b/src/PVE/Cluster/Hooks.pm
new file mode 100644
index 0000000..a5c8452
--- /dev/null
+++ b/src/PVE/Cluster/Hooks.pm
@@ -0,0 +1,135 @@
+package PVE::Cluster::Hooks;
+
+use strict;
+use warnings;
+
+use JSON;
+
+# A hook subscribes to a memdb path template and runs its code with the
+# placeholder values captured from a matching change. Runs are keyed by hook
+# and parameters, and one run can stand for several events that lead to the
+# same key. A hook therefore reads the state when it runs and takes the event
+# it is handed as the latest of those. Runs of different keys may overlap, so
+# a hook that writes shared state has to take the same locks an API handler
+# would.
+
+my $hooks = {};
+my $order = [];
+
+# a change of mtime alone leaves the content as it was
+my $default_types = [qw(create write rename delete mkdir)];
+my $known_types = { map { $_ => 1 } qw(create write mtime rename delete mkdir) };
+my $canonical = JSON->new->canonical;
+
+sub register_hook {
+ my ($class, $info) = @_;
+
+ my $name = $info->{name} // die "hook registration without a name\n";
+ die "hook '$name': invalid name\n" if $name !~ m/^[a-zA-Z][\w-]*$/;
+ die "hook '$name': already registered\n" if $hooks->{$name};
+ die "hook '$name': no path\n" if !defined($info->{path});
+ die "hook '$name': no code\n" if ref($info->{code}) ne 'CODE';
+
+ my $template = validate_template($name, $info->{path});
+
+ my $types = $info->{types} // $default_types;
+ for my $type ($types->@*) {
+ die "hook '$name': unknown event type '$type'\n" if !$known_types->{$type};
+ }
+
+ $hooks->{$name} = {
+ name => $name,
+ code => $info->{code},
+ types => { map { $_ => 1 } $types->@* },
+ template => $template,
+ };
+ push $order->@*, $name;
+
+ return;
+}
+
+# Same rules as the template parser in the daemon. The daemon refuses a
+# subscription as a whole if one template is bad, so a bad path fails here
+# at registration.
+sub validate_template {
+ my ($name, $path) = @_;
+
+ $path =~ s!^/+!!;
+ die "hook '$name': empty path\n" if $path eq '';
+
+ my @comps = split(m!/!, $path, -1);
+ my $seen = {};
+
+ for my $i (0 .. $#comps) {
+ my $comp = $comps[$i];
+ die "hook '$name': path component has zero length\n" if $comp eq '';
+ next if $comp !~ m/[{}]/;
+
+ my ($prefix, $body, $suffix) = $comp =~ m/^([^{}]*)\{([^{}]*)\}([^{}]*)$/
+ or die "hook '$name': malformed path component '$comp'\n";
+
+ if ($body =~ m/^([A-Za-z_][A-Za-z0-9_]*)\.\.\.$/) {
+ die "hook '$name': '{$1...}' must be the last component\n"
+ if $i != $#comps || $prefix ne '' || $suffix ne '';
+ die "hook '$name': placeholder '$1' used twice\n" if $seen->{$1}++;
+ next;
+ }
+
+ my ($pname) = $body =~ m/^([A-Za-z_][A-Za-z0-9_]*)$/
+ or die "hook '$name': malformed path component '$comp'\n";
+ die "hook '$name': placeholder '$pname' used twice\n" if $seen->{$pname}++;
+ }
+
+ return $path;
+}
+
+sub run_key {
+ my ($name, $param) = @_;
+
+ return $name if !scalar(keys $param->%*);
+
+ return "$name:" . $canonical->encode($param);
+}
+
+# pmxcfs already matched the templates and delivers the parameter sets per
+# hook name, so only the event type is filtered here.
+sub dispatch {
+ my ($class, $event) = @_;
+
+ my $change = { map { defined($event->{$_}) ? ($_ => $event->{$_}) : () } qw(type path to) };
+ my @runs;
+
+ if ($event->{type} eq 'resync') {
+ for my $name ($order->@*) {
+ push @runs, { hook => $hooks->{$name}, param => {}, key => $name, event => $change };
+ }
+ return \@runs;
+ }
+
+ my $params = $event->{params} // {};
+ for my $name ($order->@*) {
+ my $hook = $hooks->{$name};
+ next if !$hook->{types}->{ $event->{type} };
+
+ for my $param (($params->{$name} // [])->@*) {
+ my $key = run_key($name, $param);
+ push @runs, { hook => $hook, param => $param, key => $key, event => $change };
+ }
+ }
+
+ return \@runs;
+}
+
+sub subscription {
+ my ($class) = @_;
+
+ return { patterns => { map { $_ => $hooks->{$_}->{template} } $order->@* } };
+}
+
+sub hooks {
+ my ($class) = @_;
+
+ return [map { $hooks->{$_} } $order->@*];
+}
+
+1;
diff --git a/src/PVE/Cluster/Makefile b/src/PVE/Cluster/Makefile
index 7beb976..0da7838 100644
--- a/src/PVE/Cluster/Makefile
+++ b/src/PVE/Cluster/Makefile
@@ -1,6 +1,6 @@
PVEDIR=$(DESTDIR)/usr/share/perl5/PVE
-SOURCES=IPCConst.pm Setup.pm Watch.pm
+SOURCES=Hooks.pm IPCConst.pm Setup.pm Watch.pm
.PHONY: install
install: $(SOURCES)
diff --git a/src/test/Makefile b/src/test/Makefile
index 7b36fca..de07608 100644
--- a/src/test/Makefile
+++ b/src/test/Makefile
@@ -4,7 +4,7 @@ cpgtest: cpgtest.c
gcc -Wall cpgtest.c $(shell pkg-config --cflags --libs libcpg libqb) -o cpgtest
.PHONY: check install clean distclean
-check: corosync-parser-test test-mac-prefix watch-client-test
+check: corosync-parser-test test-mac-prefix watch-client-test hooks-test
.PHONY: corosync-parser-test
corosync-parser-test:
@@ -18,5 +18,9 @@ test-mac-prefix:
watch-client-test:
perl watch_client_test.pl
+.PHONY: hooks-test
+hooks-test:
+ perl hooks_test.pl
+
distclean: clean
clean:
diff --git a/src/test/hooks_test.pl b/src/test/hooks_test.pl
new file mode 100644
index 0000000..e49fbd9
--- /dev/null
+++ b/src/test/hooks_test.pl
@@ -0,0 +1,219 @@
+#!/usr/bin/perl
+
+use lib '..';
+
+use strict;
+use warnings;
+
+use Test::More;
+
+use PVE::Cluster::Hooks;
+
+my $noop = sub { };
+
+PVE::Cluster::Hooks->register_hook({
+ name => 'guest',
+ path => 'nodes/{node}/qemu-server/{vmid}.conf',
+ code => $noop,
+});
+
+PVE::Cluster::Hooks->register_hook({
+ name => 'dc',
+ path => '/datacenter.cfg',
+ types => ['write', 'rename'],
+ code => $noop,
+});
+
+PVE::Cluster::Hooks->register_hook({
+ name => 'sdn',
+ path => 'sdn/{file...}',
+ code => $noop,
+});
+
+PVE::Cluster::Hooks->register_hook({
+ name => 'locks',
+ path => 'priv/lock/{path...}',
+ types => ['mtime', 'create', 'delete'],
+ code => $noop,
+});
+
+is_deeply(
+ [map { $_->{name} } PVE::Cluster::Hooks->hooks()->@*],
+ ['guest', 'dc', 'sdn', 'locks'],
+ 'hooks are kept in registration order',
+);
+
+is_deeply(
+ PVE::Cluster::Hooks->subscription(),
+ {
+ patterns => {
+ guest => 'nodes/{node}/qemu-server/{vmid}.conf',
+ dc => 'datacenter.cfg',
+ sdn => 'sdn/{file...}',
+ locks => 'priv/lock/{path...}',
+ },
+ },
+ 'the subscription holds every template',
+);
+
+my $runs = sub {
+ my ($event) = @_;
+ return [map { { name => $_->{hook}->{name}, param => $_->{param}, key => $_->{key} } }
+ PVE::Cluster::Hooks->dispatch($event)->@*];
+};
+
+my $guest = { node => 'n1', vmid => '100' };
+
+is_deeply(
+ $runs->({
+ type => 'write',
+ path => 'nodes/n1/qemu-server/100.conf',
+ params => { guest => [$guest] },
+ }),
+ [{ name => 'guest', param => $guest, key => 'guest:{"node":"n1","vmid":"100"}' }],
+ 'delivered parameters become the run and its key',
+);
+
+is_deeply(
+ $runs->({
+ type => 'rename',
+ path => 'nodes/n1/qemu-server/100.conf',
+ to => 'nodes/n2/qemu-server/100.conf',
+ params => { guest => [$guest, { node => 'n2', vmid => '100' }] },
+ }),
+ [
+ { name => 'guest', param => $guest, key => 'guest:{"node":"n1","vmid":"100"}' },
+ {
+ name => 'guest',
+ param => { node => 'n2', vmid => '100' },
+ key => 'guest:{"node":"n2","vmid":"100"}',
+ },
+ ],
+ 'every delivered parameter set is a run of its own',
+);
+
+is_deeply(
+ $runs->({ type => 'write', path => 'datacenter.cfg', params => { dc => [{}] } }),
+ [{ name => 'dc', param => {}, key => 'dc' }],
+ 'a parameterless hook is keyed by its name',
+);
+
+is_deeply(
+ $runs->({ type => 'create', path => 'datacenter.cfg', params => { dc => [{}] } }),
+ [],
+ 'event types outside the list are ignored',
+);
+
+is_deeply(
+ $runs->({
+ type => 'mtime',
+ path => 'nodes/n1/qemu-server/100.conf',
+ params => { guest => [$guest] },
+ }),
+ [],
+ 'mtime is not delivered by default',
+);
+
+is_deeply(
+ $runs->({
+ type => 'mtime',
+ path => 'priv/lock/file-storage_cfg/a',
+ params => { locks => [{ path => 'file-storage_cfg/a' }] },
+ }),
+ [{
+ name => 'locks',
+ param => { path => 'file-storage_cfg/a' },
+ key => 'locks:{"path":"file-storage_cfg/a"}',
+ }],
+ 'a spanning placeholder carries the rest of the path',
+);
+
+is_deeply(
+ $runs->({
+ type => 'write',
+ path => 'sdn/zones.cfg',
+ params => { sdn => [{ file => 'zones.cfg' }], other => [{}] },
+ }),
+ [{ name => 'sdn', param => { file => 'zones.cfg' }, key => 'sdn:{"file":"zones.cfg"}' }],
+ 'parameters for unknown hooks are ignored',
+);
+
+is_deeply(
+ $runs->({ type => 'write', path => 'datacenter.cfg' }),
+ [],
+ 'an event without parameters runs nothing',
+);
+
+my $untouched = { type => 'delete', path => 'datacenter.cfg', params => { dc => [{}] } };
+PVE::Cluster::Hooks->dispatch($untouched);
+is_deeply(
+ $untouched,
+ { type => 'delete', path => 'datacenter.cfg', params => { dc => [{}] } },
+ 'dispatch leaves the event alone',
+);
+
+is_deeply(
+ [sort map { $_->{key} } $runs->({ type => 'resync' })->@*],
+ ['dc', 'guest', 'locks', 'sdn'],
+ 'a resync runs every hook without parameters',
+);
+
+my $fails = sub {
+ my ($info, $like, $desc) = @_;
+ eval { PVE::Cluster::Hooks->register_hook($info) };
+ like($@, $like, $desc);
+};
+
+$fails->(
+ { name => 'guest', path => 'x', code => $noop },
+ qr/already registered/,
+ 'duplicate names are rejected',
+);
+$fails->(
+ { name => '1bad', path => 'x', code => $noop },
+ qr/invalid name/,
+ 'names must start with a letter',
+);
+$fails->({ name => 'nocode', path => 'x' }, qr/no code/, 'a hook needs code');
+$fails->(
+ { name => 'badtype', path => 'x', types => ['chmod'], code => $noop },
+ qr/unknown event type/,
+ 'event types are validated',
+);
+$fails->(
+ { name => 'twice', path => 'a/{x}/{x}', code => $noop },
+ qr/used twice/,
+ 'placeholder names must be unique',
+);
+$fails->(
+ { name => 'broken', path => 'a/{x', code => $noop },
+ qr/malformed path component/,
+ 'unterminated placeholders are rejected',
+);
+$fails->(
+ { name => 'double', path => 'a/{x}{y}', code => $noop },
+ qr/malformed path component/,
+ 'a component holds one placeholder at most',
+);
+$fails->(
+ { name => 'typed', path => 'sdn/{file:x}', code => $noop },
+ qr/malformed path component/,
+ 'a placeholder holds a name and nothing else',
+);
+$fails->(
+ { name => 'middle', path => '{x...}/b', code => $noop },
+ qr/must be the last component/,
+ 'the spanning placeholder comes last',
+);
+$fails->(
+ { name => 'slashes', path => 'a//b', code => $noop },
+ qr/zero length/,
+ 'empty components are rejected',
+);
+$fails->(
+ { name => 'empty', path => '/', code => $noop },
+ qr/empty path/,
+ 'an empty template is rejected',
+);
+
+done_testing();
--
2.47.3
^ permalink raw reply related [flat|nested] 13+ messages in thread* [PATCH pve-cluster 12/12] cfs: add runner executing cluster change hooks in children
2026-10-22 14:21 [PATCH cluster 00/12] pmxcfs: add a change notification socket Hannes Laimer
` (10 preceding siblings ...)
2026-10-22 14:22 ` [PATCH pve-cluster 11/12] cfs: add hook registry for change notification consumers Hannes Laimer
@ 2026-10-22 14:22 ` Hannes Laimer
11 siblings, 0 replies; 13+ messages in thread
From: Hannes Laimer @ 2026-10-22 14:22 UTC (permalink / raw)
To: pve-devel
The runner subscribes to the hook registry's path templates over one
notification socket connection and executes the runs the registry
derives from each event. Every run gets a forked child, so a hook that
dies or hangs cannot take down the runner or stall its connection.
The registry keys a run by its hook and parameters, and one run per
key is in flight at a time. Events for that key that arrive meanwhile
collapse into one rerun after the run ends. Runs of different keys go
in parallel up to a worker limit.
A run that runs too long stops counting against that limit, so the
other runs keep going. The next event for its key kills it and starts
a fresh run. A cap on all children keeps such runs from piling up. A
stop terminates its children and kills what has not exited after a
grace period. A child that survives even the kill is logged and left
behind, since waiting on it any longer has no point.
Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
debian/pve-cluster.install | 1 +
src/PVE/Cluster/HookRunner.pm | 268 ++++++++++++++++++++++++++++++
src/PVE/Cluster/Makefile | 2 +-
src/test/Makefile | 6 +-
src/test/hook_runner_test.pl | 302 ++++++++++++++++++++++++++++++++++
5 files changed, 577 insertions(+), 2 deletions(-)
create mode 100644 src/PVE/Cluster/HookRunner.pm
create mode 100644 src/test/hook_runner_test.pl
diff --git a/debian/pve-cluster.install b/debian/pve-cluster.install
index b8ebff7..8321f70 100644
--- a/debian/pve-cluster.install
+++ b/debian/pve-cluster.install
@@ -4,6 +4,7 @@ usr/bin/pmxcfs
usr/lib/
usr/share/man/man8/pmxcfs.8
usr/share/perl5/PVE/Cluster.pm
+usr/share/perl5/PVE/Cluster/HookRunner.pm
usr/share/perl5/PVE/Cluster/Hooks.pm
usr/share/perl5/PVE/Cluster/IPCConst.pm
usr/share/perl5/PVE/Cluster/Watch.pm
diff --git a/src/PVE/Cluster/HookRunner.pm b/src/PVE/Cluster/HookRunner.pm
new file mode 100644
index 0000000..f6fe34a
--- /dev/null
+++ b/src/PVE/Cluster/HookRunner.pm
@@ -0,0 +1,268 @@
+package PVE::Cluster::HookRunner;
+
+use strict;
+use warnings;
+
+use POSIX qw(WNOHANG);
+use Time::HiRes qw(sleep time);
+
+use PVE::Cluster;
+use PVE::Cluster::Hooks;
+use PVE::Cluster::Watch;
+use PVE::SafeSyslog;
+
+# Executes the runs the hook registry derives from cluster changes, one run
+# per key at a time in a forked child.
+
+my $stop_grace = 5;
+
+sub new {
+ my ($class, %param) = @_;
+
+ my $self = bless {
+ watch => $param{watch} // PVE::Cluster::Watch->new(),
+ max_workers => $param{max_workers} // 4,
+ stall_age => $param{stall_age} // 120,
+ running => {},
+ by_key => {},
+ pending => {},
+ queue => [],
+ queued => {},
+ terminate => 0,
+ fork_failed => 0,
+ }, $class;
+
+ return $self;
+}
+
+sub run {
+ my ($self) = @_;
+
+ local $SIG{TERM} = sub { $self->{terminate} = 1 };
+ local $SIG{INT} = $SIG{TERM};
+ # a child's exit has to interrupt the wait, under the default
+ # disposition the wait runs on to its timeout
+ local $SIG{CHLD} = sub { };
+
+ # a warning on stderr is lost in a daemon, so the watch's warnings about
+ # its connection and those of hooks in the children go to syslog
+ local $SIG{__WARN__} = sub {
+ my ($msg) = @_;
+ chomp $msg;
+ syslog('warning', $msg);
+ };
+
+ $self->{watch}->subscribe(PVE::Cluster::Hooks->subscription()->%*);
+
+ # the wait dies on a refused handshake. The children are stopped before
+ # the error propagates, so none of them outlives the runner
+ eval {
+ # with nothing running or queued only an event or a signal needs
+ # handling, and both end the wait early. Runs in flight need the short
+ # wait so a stall is noticed
+ while (!$self->{terminate}) {
+ $self->run_once($self->idle() ? 60 : 1);
+ }
+ };
+ my $err = $@;
+ $self->stop_children();
+ die $err if $err;
+
+ return;
+}
+
+sub run_once {
+ my ($self, $timeout) = @_;
+
+ for my $event ($self->{watch}->wait($timeout)) {
+ $self->schedule($_) for PVE::Cluster::Hooks->dispatch($event)->@*;
+ }
+
+ $self->reap();
+ $self->check_stalls();
+ $self->start_queued() if !$self->{terminate};
+
+ return;
+}
+
+sub stalled {
+ my ($self, $run) = @_;
+
+ return time() - $run->{started} >= $self->{stall_age};
+}
+
+sub check_stalls {
+ my ($self) = @_;
+
+ for my $pid (keys $self->{running}->%*) {
+ my $run = $self->{running}->{$pid};
+ if ($run->{killed_at} && time() - $run->{killed_at} > $stop_grace && !$run->{forced}) {
+ syslog('warning', "killing unresponsive hook run '$run->{key}' as $pid");
+ kill 'KILL', $pid;
+ $run->{forced} = 1;
+ } elsif (!$run->{stalled} && $self->stalled($run)) {
+ $run->{stalled} = 1;
+ my $age = int(time() - $run->{started});
+ syslog('warning', "hook run '$run->{key}' as $pid still running after ${age}s");
+ }
+ }
+
+ return;
+}
+
+sub idle {
+ my ($self) = @_;
+
+ return !scalar(keys $self->{running}->%*) && !scalar($self->{queue}->@*);
+}
+
+sub schedule {
+ my ($self, $run) = @_;
+
+ my $key = $run->{key};
+
+ if (my $pid = $self->{by_key}->{$key}) {
+ $self->{pending}->{$key} = $run;
+ my $current = $self->{running}->{$pid};
+ if ($self->stalled($current) && !$current->{killed_at}) {
+ syslog('warning', "replacing stalled hook run '$key' as $pid");
+ kill 'TERM', $pid;
+ $current->{killed_at} = time();
+ }
+ } elsif ($self->{queued}->{$key}) {
+ $self->{queued}->{$key} = $run;
+ } else {
+ push $self->{queue}->@*, $key;
+ $self->{queued}->{$key} = $run;
+ }
+
+ return;
+}
+
+# A stalled run gives its slot back so the other runs keep going, and the cap
+# on all runs keeps stalled ones from piling up.
+sub start_queued {
+ my ($self) = @_;
+
+ while (scalar($self->{queue}->@*)) {
+ my $running = scalar(keys $self->{running}->%*);
+ my $active = scalar(grep { !$_->{stalled} } values $self->{running}->%*);
+ last if $active >= $self->{max_workers} || $running >= 2 * $self->{max_workers};
+ my $key = shift $self->{queue}->@*;
+ my $run = delete $self->{queued}->{$key} // next;
+ next if $self->fork_run($run);
+
+ unshift $self->{queue}->@*, $key;
+ $self->{queued}->{$key} = $run;
+ last;
+ }
+
+ return;
+}
+
+sub reap {
+ my ($self) = @_;
+
+ for my $pid (keys $self->{running}->%*) {
+ my $res = waitpid($pid, WNOHANG);
+ next if $res == 0;
+
+ my $run = delete $self->{running}->{$pid};
+ delete $self->{by_key}->{ $run->{key} };
+
+ if ($run->{stalled}) {
+ my $age = int(time() - $run->{started});
+ syslog('info', "hook run '$run->{key}' ended after ${age}s");
+ }
+ # the child logs a hook that dies, so only a signal the runner did not
+ # send is logged here
+ if ($res == $pid && ($? & 127) && !$run->{killed_at}) {
+ syslog('err', "hook run '$run->{key}' killed by signal " . ($? & 127));
+ }
+
+ if (my $again = delete $self->{pending}->{ $run->{key} }) {
+ $self->schedule($again);
+ }
+ }
+
+ return;
+}
+
+# Returns whether the run was started.
+sub fork_run {
+ my ($self, $run) = @_;
+
+ my $pid = fork();
+ if (!defined($pid)) {
+ syslog('err', "fork for hook run '$run->{key}' failed: $!") if !$self->{fork_failed}++;
+ return 0;
+ }
+ $self->{fork_failed} = 0;
+
+ if ($pid == 0) {
+ $self->{watch}->close();
+ $SIG{$_} = 'DEFAULT' for qw(CHLD HUP INT TERM QUIT);
+
+ my $rc = 0;
+ eval {
+ # the hook reads the current state, the cache inherited from the
+ # parent may be older
+ PVE::Cluster::cfs_update();
+ $run->{hook}->{code}->($run->{param}, $run->{event});
+ };
+ if (my $err = $@) {
+ chomp $err;
+ syslog('err', "hook run '$run->{key}' failed: $err");
+ $rc = 1;
+ }
+ POSIX::_exit($rc);
+ }
+
+ $run->{started} = time();
+ $self->{running}->{$pid} = $run;
+ $self->{by_key}->{ $run->{key} } = $pid;
+
+ return 1;
+}
+
+sub reap_within_grace {
+ my ($self) = @_;
+
+ my $deadline = time() + $stop_grace;
+ while (scalar(keys $self->{running}->%*) && time() < $deadline) {
+ sleep(0.1);
+ $self->reap();
+ }
+
+ return;
+}
+
+sub stop_children {
+ my ($self) = @_;
+
+ my @pids = keys $self->{running}->%*;
+ return if !scalar(@pids);
+
+ my $now = time();
+ $_->{killed_at} = $now for values $self->{running}->%*;
+ kill 'TERM', @pids;
+ $self->reap_within_grace();
+
+ if (my @left = keys $self->{running}->%*) {
+ syslog('warning', "killing unresponsive hook runs: " . join(', ', @left));
+ kill 'KILL', @left;
+ # a child blocked on the cluster file system stays in
+ # uninterruptible sleep until pmxcfs answers, so the kill alone
+ # does not bound this wait
+ $self->reap_within_grace();
+
+ if (my @stuck = keys $self->{running}->%*) {
+ my $runs = join(', ', map { "'$self->{running}->{$_}->{key}' as $_" } @stuck);
+ syslog('warning', "hook runs left outstanding by the stop: $runs");
+ }
+ }
+
+ return;
+}
+
+1;
diff --git a/src/PVE/Cluster/Makefile b/src/PVE/Cluster/Makefile
index 0da7838..47666b2 100644
--- a/src/PVE/Cluster/Makefile
+++ b/src/PVE/Cluster/Makefile
@@ -1,6 +1,6 @@
PVEDIR=$(DESTDIR)/usr/share/perl5/PVE
-SOURCES=Hooks.pm IPCConst.pm Setup.pm Watch.pm
+SOURCES=HookRunner.pm Hooks.pm IPCConst.pm Setup.pm Watch.pm
.PHONY: install
install: $(SOURCES)
diff --git a/src/test/Makefile b/src/test/Makefile
index de07608..ab4eba2 100644
--- a/src/test/Makefile
+++ b/src/test/Makefile
@@ -4,7 +4,7 @@ cpgtest: cpgtest.c
gcc -Wall cpgtest.c $(shell pkg-config --cflags --libs libcpg libqb) -o cpgtest
.PHONY: check install clean distclean
-check: corosync-parser-test test-mac-prefix watch-client-test hooks-test
+check: corosync-parser-test test-mac-prefix watch-client-test hooks-test hook-runner-test
.PHONY: corosync-parser-test
corosync-parser-test:
@@ -22,5 +22,9 @@ watch-client-test:
hooks-test:
perl hooks_test.pl
+.PHONY: hook-runner-test
+hook-runner-test:
+ perl hook_runner_test.pl
+
distclean: clean
clean:
diff --git a/src/test/hook_runner_test.pl b/src/test/hook_runner_test.pl
new file mode 100644
index 0000000..571197c
--- /dev/null
+++ b/src/test/hook_runner_test.pl
@@ -0,0 +1,302 @@
+#!/usr/bin/perl
+
+use strict;
+use warnings;
+
+use lib '..';
+
+use File::Temp qw(tempdir);
+use Test::MockModule;
+use Test::More;
+use Time::HiRes qw(sleep time);
+
+use PVE::Cluster::Hooks;
+use PVE::Cluster::HookRunner;
+
+my $dir = tempdir(CLEANUP => 1);
+my $log = "$dir/runs";
+
+# The runner never talks to pmxcfs here. A scripted watch hands out event
+# batches as the daemon would deliver them, parameters included, and
+# cfs_update is stubbed. The forks and the bookkeeping around them are real.
+my $cluster_mock = Test::MockModule->new('PVE::Cluster');
+$cluster_mock->redefine(cfs_update => sub { });
+
+# nothing here belongs in the host's journal
+my @logged;
+my $runner_mock = Test::MockModule->new('PVE::Cluster::HookRunner');
+$runner_mock->redefine(syslog => sub { push @logged, [@_] });
+
+package FakeWatch {
+
+ sub new {
+ my ($class, @batches) = @_;
+ return bless { batches => [@batches] }, $class;
+ }
+
+ sub subscribe {
+ my ($self, %param) = @_;
+ $self->{subscription} = \%param;
+ return;
+ }
+
+ sub wait {
+ my ($self, $timeout) = @_;
+ warn delete($self->{warn}) if $self->{warn};
+ my $batch = shift $self->{batches}->@*;
+ return $batch->@* if $batch;
+ die $self->{fail} if $self->{fail};
+ Time::HiRes::sleep($timeout);
+ return;
+ }
+
+ sub close {
+ my ($self) = @_;
+ return;
+ }
+}
+
+my $record = sub {
+ my ($name, $param, $event) = @_;
+ open(my $fh, '>>', $log) or die "open: $!";
+ my $args = join(',', map { "$_=$param->{$_}" } sort keys $param->%*);
+ print $fh "$name($args) $event->{type}\n";
+ close($fh);
+ return;
+};
+
+PVE::Cluster::Hooks->register_hook({
+ name => 'guest',
+ path => 'nodes/{node}/qemu-server/{vmid}.conf',
+ code => sub {
+ $record->('guest', @_);
+ sleep(0.5);
+ },
+});
+
+PVE::Cluster::Hooks->register_hook({
+ name => 'dc',
+ path => 'datacenter.cfg',
+ code => sub {
+ $record->('dc', @_);
+ sleep(0.5);
+ },
+});
+
+PVE::Cluster::Hooks->register_hook({
+ name => 'broken',
+ path => 'broken.cfg',
+ code => sub { die "on purpose\n" },
+});
+
+# hangs on a rename, records only once it got past that
+PVE::Cluster::Hooks->register_hook({
+ name => 'slow',
+ path => 'slow/{id}.conf',
+ code => sub {
+ my ($param, $event) = @_;
+ sleep(30) if $event->{type} eq 'rename';
+ $record->('slow', @_);
+ },
+});
+
+my $drain = sub {
+ my ($runner) = @_;
+ my $deadline = time() + 10;
+ while (!$runner->idle() && time() < $deadline) {
+ $runner->run_once(0.05);
+ }
+ ok($runner->idle(), 'runner drained');
+ return;
+};
+
+my $lines = sub {
+ open(my $fh, '<', $log) or return [];
+ my @lines = <$fh>;
+ chomp @lines;
+ unlink($log);
+ return \@lines;
+};
+
+my $ev = sub {
+ my ($type, $path, $to, $params) = @_;
+ return {
+ type => $type,
+ path => $path,
+ ($to ? (to => $to) : ()),
+ ($params ? (params => $params) : ()),
+ };
+};
+
+my $guest = sub {
+ my ($vmid) = @_;
+ return { guest => [{ node => 'n1', vmid => $vmid }] };
+};
+
+# saves of one guest that arrive while the run for the first is still in
+# flight collapse into one run plus one rerun
+my $save = sub {
+ return [
+ $ev->(
+ 'rename',
+ 'nodes/n1/qemu-server/100.conf.tmp.1',
+ 'nodes/n1/qemu-server/100.conf',
+ $guest->(100),
+ ),
+ ];
+};
+my $watch = FakeWatch->new($save->(), $save->(), $save->());
+my $runner = PVE::Cluster::HookRunner->new(watch => $watch, max_workers => 4);
+$runner->run_once(0.05);
+is(scalar(keys $runner->{running}->%*), 1, 'one run in flight per key');
+$runner->run_once(0.05);
+$runner->run_once(0.05);
+is(scalar(keys $runner->{running}->%*), 1, 'later saves do not start a second run');
+is(scalar(keys $runner->{pending}->%*), 1, 'they are folded into one pending rerun');
+$drain->($runner);
+is_deeply(
+ $lines->(),
+ ['guest(node=n1,vmid=100) rename', 'guest(node=n1,vmid=100) rename'],
+ 'three saves became one run plus one rerun',
+);
+
+# different keys run concurrently up to the worker cap
+$watch = FakeWatch->new([
+ $ev->('write', 'nodes/n1/qemu-server/100.conf', undef, $guest->(100)),
+ $ev->('write', 'nodes/n1/qemu-server/101.conf', undef, $guest->(101)),
+ $ev->('write', 'nodes/n1/qemu-server/102.conf', undef, $guest->(102)),
+ $ev->('write', 'datacenter.cfg', undef, { dc => [{}] }),
+]);
+$runner = PVE::Cluster::HookRunner->new(watch => $watch, max_workers => 2);
+$runner->run_once(0.05);
+is(scalar(keys $runner->{running}->%*), 2, 'the worker cap holds');
+is(scalar($runner->{queue}->@*), 2, 'the rest waits in the queue');
+$drain->($runner);
+is_deeply(
+ [sort $lines->()->@*],
+ [
+ 'dc() write',
+ 'guest(node=n1,vmid=100) write',
+ 'guest(node=n1,vmid=101) write',
+ 'guest(node=n1,vmid=102) write',
+ ],
+ 'every key ran exactly once',
+);
+
+# a resync runs every hook once without parameters
+$watch = FakeWatch->new([$ev->('resync')]);
+$runner = PVE::Cluster::HookRunner->new(watch => $watch, max_workers => 4);
+$runner->run_once(0.05);
+$drain->($runner);
+is_deeply(
+ [sort $lines->()->@*],
+ ['dc() resync', 'guest() resync', 'slow() resync'],
+ 'resync reaches every hook, a failing hook does not disturb the others',
+);
+
+# the runner subscribes to what the registry derived
+$watch = FakeWatch->new();
+$runner = PVE::Cluster::HookRunner->new(watch => $watch, max_workers => 1);
+$runner->{terminate} = 1;
+$runner->run();
+is_deeply(
+ $watch->{subscription},
+ PVE::Cluster::Hooks->subscription(),
+ 'subscription is derived from the registered hooks',
+);
+
+# a watch that dies ends the runner, and the runner stops its children first
+$watch = FakeWatch->new([$ev->('write', 'datacenter.cfg', undef, { dc => [{}] })]);
+$watch->{fail} = "handshake refused\n";
+$runner = PVE::Cluster::HookRunner->new(watch => $watch, max_workers => 4);
+eval { $runner->run() };
+like($@, qr/handshake refused/, 'a failing watch ends the runner');
+is(scalar(keys $runner->{running}->%*), 0, 'and no child is left behind');
+$lines->();
+
+@logged = ();
+$watch = FakeWatch->new([$ev->('write', 'nodes/n1/qemu-server/100.conf', undef, $guest->(100))]);
+$runner = PVE::Cluster::HookRunner->new(watch => $watch, max_workers => 4);
+$runner->run_once(0.05);
+$runner->stop_children();
+$runner->run_once(0.05);
+is(scalar(keys $runner->{running}->%*), 0, 'the run was stopped');
+ok(!(grep { $_->[0] eq 'err' } @logged), 'a run interrupted by the stop is not an error');
+$lines->();
+
+@logged = ();
+$watch = FakeWatch->new();
+$watch->{warn} = "notification socket closed, reconnecting\n";
+$watch->{fail} = "stop\n";
+$runner = PVE::Cluster::HookRunner->new(watch => $watch, max_workers => 1);
+eval { $runner->run() };
+is_deeply(
+ [grep { $_->[0] eq 'warning' } @logged],
+ [['warning', 'notification socket closed, reconnecting']],
+ 'a warning from the watch is logged',
+);
+
+# a stalled run frees its slot, is logged, and is killed by the next event
+# for its key while the other keys keep running
+@logged = ();
+my $slow = sub {
+ my ($hang) = @_;
+ my $params = { slow => [{ id => 1 }] };
+ return $ev->('rename', 'slow/1.conf.tmp', 'slow/1.conf', $params) if $hang;
+ return $ev->('write', 'slow/1.conf', undef, $params);
+};
+$watch = FakeWatch->new(
+ [$slow->(1)],
+ [$ev->('write', 'datacenter.cfg', undef, { dc => [{}] })],
+ [$slow->()],
+);
+$runner = PVE::Cluster::HookRunner->new(watch => $watch, max_workers => 1, stall_age => 0.5);
+$runner->run_once(0.05);
+is(scalar(keys $runner->{running}->%*), 1, 'the hanging run holds the only slot');
+sleep(0.6);
+$runner->run_once(0.05);
+is(scalar(keys $runner->{running}->%*), 2, 'past the stall age another key runs beside it');
+ok(
+ (grep { $_->[0] eq 'warning' && $_->[1] =~ /^hook run 'slow:.*still running after/ } @logged),
+ 'the stall is logged',
+);
+$runner->run_once(0.05);
+$drain->($runner);
+is_deeply(
+ [sort $lines->()->@*],
+ ['dc() write', 'slow(id=1) write'],
+ 'the next event for the key replaced the stalled run with a rerun',
+);
+ok(
+ (grep { $_->[0] eq 'warning' && $_->[1] =~ /^replacing stalled hook run 'slow:/ } @logged),
+ 'and said so',
+);
+ok((grep { $_->[0] eq 'info' && $_->[1] =~ /ended after/ } @logged), 'the end is logged too');
+ok(!(grep { $_->[0] eq 'err' } @logged), 'a run the runner killed is not an error');
+
+# a child in uninterruptible sleep outlives the kill. A clock that jumps
+# past every deadline leaves the child unreaped the same way, without
+# waiting
+@logged = ();
+$watch = FakeWatch->new([$ev->('write', 'nodes/n1/qemu-server/100.conf', undef, $guest->(100))]);
+$runner = PVE::Cluster::HookRunner->new(watch => $watch, max_workers => 4);
+$runner->run_once(0.05);
+my ($stuck) = keys $runner->{running}->%*;
+my $clock = time();
+$runner_mock->redefine(time => sub () { $clock += 10 });
+$runner->stop_children();
+$runner_mock->unmock('time');
+is_deeply([keys $runner->{running}->%*], [$stuck], 'the stop gives up on a run it cannot reap');
+ok(
+ (
+ grep {
+ $_->[0] eq 'warning'
+ && $_->[1] =~ /^hook runs left outstanding by the stop: 'guest:/
+ } @logged
+ ),
+ 'and says which ones are left',
+);
+waitpid($stuck, 0);
+$lines->();
+
+done_testing();
--
2.47.3
^ permalink raw reply related [flat|nested] 13+ messages in thread