* [RFC cluster/manager 00/10] pmxcfs: add a change notification socket
@ 2026-09-18 14:41 Hannes Laimer
2026-09-18 14:41 ` [PATCH pve-cluster 01/10] buildsys: add rust workspace under src/rust Hannes Laimer
` (9 more replies)
0 siblings, 10 replies; 11+ messages in thread
From: Hannes Laimer @ 2026-09-18 14:41 UTC (permalink / raw)
To: pve-devel
Every daemon that cares about /etc/pve learns about changes by calling
cfs_update on each loop iteration and comparing the version vector it
gets over the libqb IPC. inotify cannot replace that, since remote
changes arrive through corosync and never touch the local VFS, so
pvestatd, pve-firewall, the HA daemons and pvescheduler all wake up on
timers and re-read what usually has not changed.
This series adds a push path. pmxcfs gets a Unix stream socket at
/run/pve-cluster/pmxcfs.sock, served by a Rust thread linked into the C
daemon as a static library with a small C ABI. A client subscribes with
named path templates such as nodes/{node}/qemu-server/{vmid}.conf and
receives one JSON line per matching mutation, carrying the memdb version
as sequence number, the event type, the path and the values the
placeholders captured. Connections authorized by group membership never
see private paths, as on the IPC and FUSE side.
The daemon keeps the last mutations in a fixed size ring and each
connection a cursor into it, so the mutating thread only appends and
never waits for a client. A slow client is caught up from the ring, one
that fell off it gets a single resync event, and a reconnecting client
resumes at the last sequence number it saw, also across a restart of
pmxcfs when nothing changed meanwhile. A node that takes the whole
state from the cluster after a membership change hands its clients a
resync event, since no sequence of mutations describes that. A panic in
the Rust code only disables the notifier for the rest of the daemon's
life. A sequence number identifies a state within one line of history
only. A client that missed a resync while disconnected and returns at
the same version after a pmxcfs restart resumes as if nothing changed.
Carrying the root entry's mtime next to the version would close that.
On top of the socket, pve-cluster ships a Perl client with reconnect
and resync handling and a hook registry, where a hook is registered
like an API method with a path template whose placeholders become the
parameters of a run. pve-manager gets the runner that executes those
runs in children of a listener hosted by pvescheduler, one run per hook
and parameter set in flight and later events for the same set collapsed
into a single rerun, so a config update through the API costs one extra
run rather than one per save. The listener records the position it has
processed up to under /run and resumes there after a reload. No hook
ships in this series. A consumer registers one as the example below
shows.
pve-manager depends on the pve-cluster packages of this series for the
new modules and the socket, at build time as well, since its tests run
through the check target.
Example usage:
package PVE::Network::Hooks;
use base qw(PVE::Cluster::Hooks);
__PACKAGE__->register_hook({
name => 'guest-firewall',
path => 'firewall/{vmid}.fw',
code => sub {
my ($param, $event) = @_;
my $vmids = $event->{type} eq 'resync'
? [] # every guest with a firewall config
: [ $param->{vmid} ];
for my $vmid ($vmids->@*) {
# reload and apply the firewall for guest $vmid
}
},
});
1;
pve-cluster:
Hannes Laimer (8):
buildsys: add rust workspace under src/rust
rust: notify: add change notification socket server
rust: ffi: add C ABI staticlib for pmxcfs
pmxcfs: memdb: add change notification hook
buildsys: link pmxcfs against the rust notify staticlib
pmxcfs: notify: emit change events over the notification socket
cfs: add perl client for the change notification socket
cfs: add hook registry for change notification consumers
.gitignore | 2 +
Makefile | 1 +
debian/control | 16 +-
debian/pve-cluster.install | 2 +
debian/rules | 16 +
src/Makefile | 2 +-
src/PVE/Cluster/Hooks.pm | 134 ++++
src/PVE/Cluster/Makefile | 2 +-
src/PVE/Cluster/Watch.pm | 371 ++++++++++
src/pmxcfs/Makefile | 15 +-
src/pmxcfs/cfs-utils.h | 4 +
src/pmxcfs/database.c | 2 +
src/pmxcfs/memdb.c | 24 +
src/pmxcfs/memdb.h | 11 +
src/pmxcfs/pmxcfs.c | 25 +
src/rust/.cargo/config.toml | 8 +
src/rust/Cargo.toml | 23 +
src/rust/Makefile | 18 +
src/rust/pmxcfs-ffi/Cargo.toml | 17 +
src/rust/pmxcfs-ffi/include/pmxcfs-notify.h | 29 +
src/rust/pmxcfs-ffi/src/lib.rs | 309 ++++++++
src/rust/pmxcfs-notify/Cargo.toml | 16 +
src/rust/pmxcfs-notify/src/conn.rs | 317 ++++++++
src/rust/pmxcfs-notify/src/lib.rs | 43 ++
src/rust/pmxcfs-notify/src/protocol.rs | 237 ++++++
src/rust/pmxcfs-notify/src/registry.rs | 405 ++++++++++
src/rust/pmxcfs-notify/src/ring.rs | 171 +++++
src/rust/pmxcfs-notify/src/server.rs | 780 ++++++++++++++++++++
src/rust/pmxcfs-notify/src/template.rs | 254 +++++++
src/rust/rustfmt.toml | 2 +
src/test/Makefile | 10 +-
src/test/hooks_test.pl | 219 ++++++
src/test/watch_client_test.pl | 294 ++++++++
33 files changed, 3773 insertions(+), 6 deletions(-)
create mode 100644 src/PVE/Cluster/Hooks.pm
create mode 100644 src/PVE/Cluster/Watch.pm
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-ffi/Cargo.toml
create mode 100644 src/rust/pmxcfs-ffi/include/pmxcfs-notify.h
create mode 100644 src/rust/pmxcfs-ffi/src/lib.rs
create mode 100644 src/rust/pmxcfs-notify/Cargo.toml
create mode 100644 src/rust/pmxcfs-notify/src/conn.rs
create mode 100644 src/rust/pmxcfs-notify/src/lib.rs
create mode 100644 src/rust/pmxcfs-notify/src/protocol.rs
create mode 100644 src/rust/pmxcfs-notify/src/registry.rs
create mode 100644 src/rust/pmxcfs-notify/src/ring.rs
create mode 100644 src/rust/pmxcfs-notify/src/server.rs
create mode 100644 src/rust/pmxcfs-notify/src/template.rs
create mode 100644 src/rust/rustfmt.toml
create mode 100644 src/test/hooks_test.pl
create mode 100644 src/test/watch_client_test.pl
pve-manager:
Hannes Laimer (2):
hooks: add runner executing cluster change hooks in children
pvescheduler: run cluster change hooks from a listener child
PVE/HookRunner.pm | 368 ++++++++++++++++++++++++++++++++
PVE/Makefile | 1 +
PVE/Service/pvescheduler.pm | 26 ++-
test/Makefile | 6 +-
test/hook_runner_test.pl | 405 ++++++++++++++++++++++++++++++++++++
5 files changed, 801 insertions(+), 5 deletions(-)
create mode 100644 PVE/HookRunner.pm
create mode 100755 test/hook_runner_test.pl
Summary over all repositories:
38 files changed, 4574 insertions(+), 11 deletions(-)
--
Generated by murpp 0.12.0
^ permalink raw reply [flat|nested] 11+ messages in thread
* [PATCH pve-cluster 01/10] buildsys: add rust workspace under src/rust
2026-09-18 14:41 [RFC cluster/manager 00/10] pmxcfs: add a change notification socket Hannes Laimer
@ 2026-09-18 14:41 ` Hannes Laimer
2026-09-18 14:41 ` [PATCH pve-cluster 02/10] rust: notify: add change notification socket server Hannes Laimer
` (8 subsequent siblings)
9 siblings, 0 replies; 11+ messages in thread
From: Hannes Laimer @ 2026-09-18 14:41 UTC (permalink / raw)
To: pve-devel
Add a Cargo workspace under src/rust, next to the C and Perl trees, for
the Rust code that goes into pmxcfs, a server crate and a small C ABI
crate. Their dependencies resolve through the Debian cargo registry as
pve-rs does, and cargo hooks into the src/Makefile recursion ahead of
pmxcfs so a static library built here is linked into the daemon.
Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
.gitignore | 2 ++
Makefile | 1 +
debian/control | 16 +++++++++++++++-
debian/rules | 16 ++++++++++++++++
src/Makefile | 2 +-
src/rust/.cargo/config.toml | 8 ++++++++
src/rust/Cargo.toml | 22 ++++++++++++++++++++++
src/rust/Makefile | 18 ++++++++++++++++++
src/rust/pmxcfs-notify/Cargo.toml | 11 +++++++++++
src/rust/pmxcfs-notify/src/lib.rs | 1 +
src/rust/rustfmt.toml | 2 ++
11 files changed, 97 insertions(+), 2 deletions(-)
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..c3cb27a 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,23 @@ 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+event-dev,
+ librust-nix-0.29+poll-dev,
+ librust-nix-0.29+socket-dev,
+ librust-serde-1+default-dev,
+ librust-serde-1+derive-dev,
+ librust-serde-json-1+default-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..0717d86 100755
--- a/debian/rules
+++ b/debian/rules
@@ -1,12 +1,28 @@
#!/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
+CARGO=/usr/share/cargo/bin/cargo
+
+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/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..6de0ab8
--- /dev/null
+++ b/src/rust/Cargo.toml
@@ -0,0 +1,22 @@
+[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 = ["event", "poll", "socket"] }
+serde = { version = "1.0", features = ["derive"] }
+serde_json = "1.0"
+
+pmxcfs-notify = { path = "pmxcfs-notify" }
diff --git a/src/rust/Makefile b/src/rust/Makefile
new file mode 100644
index 0000000..aba5cb9
--- /dev/null
+++ b/src/rust/Makefile
@@ -0,0 +1,18 @@
+all:
+ cargo build --release
+
+.PHONY: check
+check:
+ cargo test --release
+
+.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] 11+ messages in thread
* [PATCH pve-cluster 02/10] rust: notify: add change notification socket server
2026-09-18 14:41 [RFC cluster/manager 00/10] pmxcfs: add a change notification socket Hannes Laimer
2026-09-18 14:41 ` [PATCH pve-cluster 01/10] buildsys: add rust workspace under src/rust Hannes Laimer
@ 2026-09-18 14:41 ` Hannes Laimer
2026-09-18 14:41 ` [PATCH pve-cluster 03/10] rust: ffi: add C ABI staticlib for pmxcfs Hannes Laimer
` (7 subsequent siblings)
9 siblings, 0 replies; 11+ messages in thread
From: Hannes Laimer @ 2026-09-18 14:41 UTC (permalink / raw)
To: pve-devel
Daemons currently learn about /etc/pve changes by polling the version
vector over IPC on every loop iteration, and inotify only sees writes
made on the local node. Add a small server that clients reach through a
Unix stream socket. A client subscribes with named path templates such
as nodes/{node}/qemu-server/{vmid}.conf. Each matching mutation then
arrives as one pushed line, carrying the values the placeholders
captured. 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.
Matching has to stay bounded, since it runs for every mutation and
every connection. A template allows one placeholder per path component
and a spanning one only as the last component, which makes a match a
single pass over the path with no regex engine behind it, and the
captures come out of that same pass for free.
The daemon must never block on a client. The last few thousand
mutations live in a fixed size ring and each connection keeps a cursor
into it. The mutating thread only appends and wakes the delivery
thread, which does every send, matches on the way out, and finishes a
partially written line before the next one. A connection whose socket
is full is caught up once it drains, and a cursor that fell off the
ring turns into a single resync event. A reconnecting client resumes
from the last number it saw. The hello reply hands out that number too,
so a client that has seen no event yet can resume as well. The sequence
number is the memdb version, so it means the same on every node. Should
the delivery thread fail, it closes every connection on its way out, so
a client always learns that it has to reconnect instead of staying
attached to a socket that went silent.
Connections authorized by group membership never see events about
private paths, the same rule the IPC and FUSE side apply to content.
Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
src/rust/pmxcfs-notify/Cargo.toml | 5 +
src/rust/pmxcfs-notify/src/conn.rs | 317 ++++++++++
src/rust/pmxcfs-notify/src/lib.rs | 42 ++
src/rust/pmxcfs-notify/src/protocol.rs | 237 ++++++++
src/rust/pmxcfs-notify/src/registry.rs | 405 +++++++++++++
src/rust/pmxcfs-notify/src/ring.rs | 171 ++++++
src/rust/pmxcfs-notify/src/server.rs | 780 +++++++++++++++++++++++++
src/rust/pmxcfs-notify/src/template.rs | 254 ++++++++
8 files changed, 2211 insertions(+)
create mode 100644 src/rust/pmxcfs-notify/src/conn.rs
create mode 100644 src/rust/pmxcfs-notify/src/protocol.rs
create mode 100644 src/rust/pmxcfs-notify/src/registry.rs
create mode 100644 src/rust/pmxcfs-notify/src/ring.rs
create mode 100644 src/rust/pmxcfs-notify/src/server.rs
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..5a67507 100644
--- a/src/rust/pmxcfs-notify/Cargo.toml
+++ b/src/rust/pmxcfs-notify/Cargo.toml
@@ -9,3 +9,8 @@ homepage.workspace = true
rust-version.workspace = true
[dependencies]
+anyhow.workspace = true
+log.workspace = true
+nix.workspace = true
+serde.workspace = true
+serde_json.workspace = true
diff --git a/src/rust/pmxcfs-notify/src/conn.rs b/src/rust/pmxcfs-notify/src/conn.rs
new file mode 100644
index 0000000..e5df526
--- /dev/null
+++ b/src/rust/pmxcfs-notify/src/conn.rs
@@ -0,0 +1,317 @@
+use std::io::{self, Read};
+use std::os::fd::{AsRawFd, RawFd};
+use std::os::unix::net::UnixStream;
+
+use nix::errno::Errno;
+use nix::sys::socket::{MsgFlags, Shutdown, send, shutdown};
+
+use crate::protocol::Params;
+use crate::registry::is_private;
+use crate::template::Template;
+
+const MAX_LINE: usize = 64 * 1024;
+const MAX_OUTBUF: usize = 256 * 1024;
+
+pub(crate) struct Conn {
+ stream: UnixStream,
+ privileged: bool,
+ pub(crate) templates: Vec<(String, Template)>,
+ pub(crate) subscribed: bool,
+ pub(crate) dead: bool,
+ pub(crate) cursor: u64,
+ pub(crate) resync_pending: bool,
+ outbuf: Vec<u8>,
+ inbuf: Vec<u8>,
+}
+
+#[derive(Debug)]
+pub(crate) enum ReadError {
+ Eof,
+ LineTooLong,
+ Io(io::Error),
+}
+
+impl Conn {
+ pub(crate) fn new(stream: UnixStream, privileged: bool) -> io::Result<Self> {
+ stream.set_nonblocking(true)?;
+ Ok(Self {
+ stream,
+ privileged,
+ templates: Vec::new(),
+ subscribed: false,
+ dead: false,
+ cursor: 0,
+ resync_pending: false,
+ outbuf: Vec::new(),
+ inbuf: Vec::new(),
+ })
+ }
+
+ pub(crate) fn fd(&self) -> RawFd {
+ self.stream.as_raw_fd()
+ }
+
+ pub(crate) fn lagging(&self) -> bool {
+ !self.outbuf.is_empty()
+ }
+
+ pub(crate) fn matches<'a>(
+ &'a self,
+ path: Option<&'a str>,
+ to: Option<&'a str>,
+ ) -> Option<Params<'a>> {
+ if !self.subscribed {
+ return None;
+ }
+ if !self.privileged && (path.is_some_and(is_private) || to.is_some_and(is_private)) {
+ return None;
+ }
+ let mut params = Params::new();
+ for (name, template) in &self.templates {
+ let mut sets = Vec::new();
+ for candidate in [path, to].into_iter().flatten() {
+ if let Some(captures) = template.matches(candidate)
+ && !sets.contains(&captures)
+ {
+ sets.push(captures);
+ }
+ }
+ if !sets.is_empty() {
+ params.insert(name.as_str(), sets);
+ }
+ }
+ (!params.is_empty()).then_some(params)
+ }
+
+ // Bytes the kernel did not take stay here and go out first once the
+ // socket has room again, so a line is never torn. A client that keeps
+ // sending commands without reading the replies is cut at the cap.
+ pub(crate) fn queue(&mut self, line: &str) {
+ if self.dead {
+ return;
+ }
+ self.outbuf.extend_from_slice(line.as_bytes());
+ if self.outbuf.len() > MAX_OUTBUF {
+ self.mark_dead();
+ }
+ }
+
+ /// Returns whether everything queued went out.
+ pub(crate) fn flush(&mut self) -> bool {
+ while !self.outbuf.is_empty() && !self.dead {
+ let flags = MsgFlags::MSG_NOSIGNAL | MsgFlags::MSG_DONTWAIT;
+ match send(self.fd(), &self.outbuf, flags) {
+ Ok(0) | Err(Errno::EAGAIN) => return false,
+ Ok(n) => {
+ self.outbuf.drain(..n);
+ }
+ Err(Errno::EINTR) => continue,
+ Err(_) => self.mark_dead(),
+ }
+ }
+ !self.dead
+ }
+
+ pub(crate) fn mark_dead(&mut self) {
+ self.dead = true;
+ self.outbuf.clear();
+ let _ = shutdown(self.fd(), Shutdown::Both);
+ }
+
+ pub(crate) fn read_lines(&mut self) -> Result<Vec<String>, ReadError> {
+ let mut buf = [0u8; 4096];
+ loop {
+ match self.stream.read(&mut buf) {
+ Ok(0) => return Err(ReadError::Eof),
+ Ok(n) => {
+ self.inbuf.extend_from_slice(&buf[..n]);
+ break;
+ }
+ Err(err) if err.kind() == io::ErrorKind::WouldBlock => return Ok(Vec::new()),
+ Err(err) if err.kind() == io::ErrorKind::Interrupted => continue,
+ Err(err) => return Err(ReadError::Io(err)),
+ }
+ }
+ let mut lines = Vec::new();
+ while let Some(pos) = self.inbuf.iter().position(|&b| b == b'\n') {
+ let line = String::from_utf8_lossy(&self.inbuf[..pos]).into_owned();
+ self.inbuf.drain(..=pos);
+ lines.push(line);
+ }
+ if self.inbuf.len() > MAX_LINE {
+ return Err(ReadError::LineTooLong);
+ }
+ Ok(lines)
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use std::collections::BTreeMap;
+ use std::io::Write;
+
+ use super::*;
+
+ fn pair(privileged: bool) -> (Conn, UnixStream) {
+ let (server, client) = UnixStream::pair().unwrap();
+ (Conn::new(server, privileged).unwrap(), client)
+ }
+
+ fn templates(specs: &[(&str, &str)]) -> Vec<(String, Template)> {
+ specs
+ .iter()
+ .map(|(name, text)| ((*name).to_owned(), Template::parse(text).unwrap()))
+ .collect()
+ }
+
+ #[test]
+ fn reassembles_partial_lines() {
+ let (mut conn, mut client) = pair(true);
+ client.write_all(b"{\"command\":").unwrap();
+ assert!(conn.read_lines().unwrap().is_empty());
+ client
+ .write_all(b"\"hello\"}\n{\"command\":\"unsubscribe\"}\n")
+ .unwrap();
+ let lines = conn.read_lines().unwrap();
+ assert_eq!(
+ lines,
+ vec!["{\"command\":\"hello\"}", "{\"command\":\"unsubscribe\"}"]
+ );
+ assert!(conn.read_lines().unwrap().is_empty());
+ }
+
+ #[test]
+ fn reports_eof_and_overlong_lines() {
+ let (mut conn, mut client) = pair(true);
+ client.write_all(&vec![b'x'; MAX_LINE + 1]).unwrap();
+ let mut result = conn.read_lines();
+ while let Ok(lines) = &result {
+ assert!(lines.is_empty());
+ result = conn.read_lines();
+ }
+ assert!(matches!(result, Err(ReadError::LineTooLong)));
+
+ let (mut conn, client) = pair(true);
+ drop(client);
+ assert!(matches!(conn.read_lines(), Err(ReadError::Eof)));
+ }
+
+ #[test]
+ fn short_writes_are_finished_before_the_next_line() {
+ let (mut conn, mut client) = pair(true);
+ client.set_nonblocking(true).unwrap();
+ let line = format!("{}\n", "x".repeat(3000));
+
+ let mut queued = 0;
+ while !conn.lagging() {
+ conn.queue(&line);
+ conn.flush();
+ queued += 1;
+ assert!(queued < 100_000, "socket never filled");
+ }
+
+ let mut received = Vec::new();
+ let mut buf = [0u8; 65536];
+ let mut read_available = |client: &mut UnixStream, received: &mut Vec<u8>| {
+ loop {
+ match client.read(&mut buf) {
+ Ok(0) => break,
+ Ok(n) => received.extend_from_slice(&buf[..n]),
+ Err(err) if err.kind() == io::ErrorKind::WouldBlock => break,
+ Err(err) => panic!("read failed: {err}"),
+ }
+ }
+ };
+ loop {
+ read_available(&mut client, &mut received);
+ if !conn.lagging() {
+ break;
+ }
+ conn.flush();
+ }
+ read_available(&mut client, &mut received);
+
+ assert_eq!(received.len(), queued * line.len());
+ assert!(
+ received
+ .chunks(line.len())
+ .all(|chunk| chunk == line.as_bytes())
+ );
+ assert!(!conn.dead);
+ }
+
+ #[test]
+ fn matching_needs_a_subscription() {
+ let (mut conn, _client) = pair(true);
+ conn.templates = templates(&[
+ ("guest", "nodes/{node}/qemu-server/{vmid}.conf"),
+ ("dc", "datacenter.cfg"),
+ ]);
+ let guest = "nodes/n1/qemu-server/100.conf";
+ assert!(conn.matches(Some(guest), None).is_none());
+
+ conn.subscribed = true;
+ let params = conn.matches(Some(guest), None).unwrap();
+ assert_eq!(params.len(), 1);
+ assert_eq!(
+ params["guest"],
+ vec![BTreeMap::from([("node", "n1"), ("vmid", "100")])]
+ );
+ assert_eq!(
+ conn.matches(Some("datacenter.cfg"), None).unwrap()["dc"],
+ vec![BTreeMap::new()]
+ );
+ assert!(conn.matches(Some("datacenter.cfg.tmp.1"), None).is_none());
+ assert!(conn.matches(Some("other"), None).is_none());
+ assert!(conn.matches(None, None).is_none());
+ }
+
+ #[test]
+ fn rename_matches_on_either_side() {
+ let (mut conn, _client) = pair(true);
+ conn.templates = templates(&[("guest", "nodes/{node}/qemu-server/{vmid}.conf")]);
+ conn.subscribed = true;
+ let n1 = "nodes/n1/qemu-server/100.conf";
+ let n2 = "nodes/n2/qemu-server/100.conf";
+
+ let params = conn.matches(Some(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(Some("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(Some(n1), Some(n1)).unwrap()["guest"].len(), 1);
+ }
+
+ #[test]
+ fn private_paths_need_privileges() {
+ let (mut conn, _client) = pair(false);
+ conn.templates = templates(&[("all", "{path...}")]);
+ conn.subscribed = true;
+ assert!(conn.matches(Some("datacenter.cfg"), None).is_some());
+ assert!(conn.matches(Some("priv/lock/x"), None).is_none());
+ assert!(conn.matches(Some("nodes/n1/priv/ssl.key"), None).is_none());
+ assert!(
+ conn.matches(Some("datacenter.cfg"), Some("priv/moved"))
+ .is_none()
+ );
+ assert!(
+ conn.matches(Some("priv/old"), Some("datacenter.cfg"))
+ .is_none()
+ );
+
+ let (mut conn, _client) = pair(true);
+ conn.templates = templates(&[("all", "{path...}")]);
+ conn.subscribed = true;
+ assert!(conn.matches(Some("priv/lock/x"), None).is_some());
+ assert!(
+ conn.matches(Some("priv/old"), Some("datacenter.cfg"))
+ .is_some()
+ );
+ }
+}
diff --git a/src/rust/pmxcfs-notify/src/lib.rs b/src/rust/pmxcfs-notify/src/lib.rs
index 8b13789..d363efb 100644
--- a/src/rust/pmxcfs-notify/src/lib.rs
+++ b/src/rust/pmxcfs-notify/src/lib.rs
@@ -1 +1,43 @@
+//! 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
+//! memdb version as sequence number, the event type, the path, the
+//! rename target, and per subscription name the parameter sets captured
+//! from the path and the rename target. A client only ever needs to look
+//! at which key an object has.
+//!
+//! A template is a memdb path whose components may each hold one
+//! `{name}` placeholder standing for part of that component, and whose
+//! last component may be `{name...}` for the rest of the path, so
+//! `nodes/{node}/qemu-server/{vmid}.conf` captures node and vmid and
+//! `{path...}` matches everything. Connections authorized by group
+//! membership rather than as root never receive events that mention a
+//! private path, matching what the IPC and FUSE side hide from them.
+//!
+//! The daemon keeps the last mutations in a fixed size ring and each
+//! connection a cursor into it, so a client whose socket is full is
+//! caught up once it reads again, and one that fell off the ring gets a
+//! single resync event instead. A reconnecting client passes the last
+//! sequence number it saw as `since` and receives what it missed, or a
+//! resync when that is no longer available. The reply to `hello` carries
+//! the protocol version and the newest number, so a client that has not
+//! seen an event yet can still resume later.
+mod conn;
+mod protocol;
+mod registry;
+mod ring;
+mod server;
+mod template;
+
+pub use protocol::EventKind;
+pub use server::{Config, Server};
+
+/// memdb paths carry no leading slash.
+pub(crate) fn normalize_path(path: &str) -> &str {
+ path.trim_start_matches('/')
+}
diff --git a/src/rust/pmxcfs-notify/src/protocol.rs b/src/rust/pmxcfs-notify/src/protocol.rs
new file mode 100644
index 0000000..917cec6
--- /dev/null
+++ b/src/rust/pmxcfs-notify/src/protocol.rs
@@ -0,0 +1,237 @@
+use std::collections::BTreeMap;
+
+use anyhow::{Error, bail, format_err};
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+
+use crate::template::{Captures, Template};
+
+pub(crate) const PROTOCOL_VERSION: u32 = 1;
+const MAX_PATTERNS: usize = 256;
+
+/// One entry per matching side of the event, keyed by subscription name.
+pub(crate) type Params<'a> = BTreeMap<&'a str, Vec<Captures<'a>>>;
+
+#[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> {
+ pub seq: u64,
+ #[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, so a client dispatches on that key.
+#[derive(Serialize)]
+#[serde(rename_all = "lowercase")]
+pub(crate) enum Message<'a> {
+ Ok(Value),
+ Error(String),
+ Event(Event<'a>),
+}
+
+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(Deserialize)]
+struct Request {
+ command: String,
+ #[serde(default)]
+ args: Value,
+}
+
+#[derive(Debug, PartialEq, Eq)]
+pub(crate) enum Command {
+ Hello,
+ Subscribe {
+ patterns: Vec<(String, Template)>,
+ since: Option<u64>,
+ },
+ Unsubscribe,
+}
+
+fn patterns(args: &Value) -> Result<Vec<(String, Template)>, Error> {
+ let entries = match args.get("patterns") {
+ None | Some(Value::Null) => return Ok(Vec::new()),
+ Some(Value::Object(entries)) => entries,
+ Some(_) => bail!("'patterns' must map names to path templates"),
+ };
+ if entries.len() > MAX_PATTERNS {
+ bail!("at most {MAX_PATTERNS} patterns per subscription");
+ }
+ entries
+ .iter()
+ .map(|(name, text)| {
+ let text = text
+ .as_str()
+ .ok_or_else(|| format_err!("pattern '{name}' must be a string"))?;
+ let template =
+ Template::parse(text).map_err(|err| format_err!("pattern '{name}': {err}"))?;
+ Ok((name.clone(), template))
+ })
+ .collect()
+}
+
+fn since(args: &Value) -> Result<Option<u64>, Error> {
+ match args.get("since") {
+ None | Some(Value::Null) => Ok(None),
+ Some(value) => value
+ .as_u64()
+ .map(Some)
+ .ok_or_else(|| format_err!("'since' must be an unsigned integer")),
+ }
+}
+
+pub(crate) fn parse_command(line: &str) -> Result<Command, Error> {
+ let request: Request = serde_json::from_str(line)?;
+ match request.command.as_str() {
+ "hello" => Ok(Command::Hello),
+ "subscribe" => Ok(Command::Subscribe {
+ patterns: patterns(&request.args)?,
+ since: since(&request.args)?,
+ }),
+ "unsubscribe" => Ok(Command::Unsubscribe),
+ other => bail!("unknown command '{other}'"),
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn parses_commands() {
+ assert_eq!(
+ parse_command(r#"{"command":"hello"}"#).unwrap(),
+ Command::Hello
+ );
+ assert_eq!(
+ parse_command(
+ r#"{"command":"subscribe","args":{"patterns":{"guest":"nodes/{node}/qemu-server/{vmid}.conf","dc":"/datacenter.cfg"}}}"#
+ )
+ .unwrap(),
+ Command::Subscribe {
+ patterns: vec![
+ ("dc".to_owned(), Template::parse("datacenter.cfg").unwrap()),
+ (
+ "guest".to_owned(),
+ Template::parse("nodes/{node}/qemu-server/{vmid}.conf").unwrap()
+ ),
+ ],
+ since: None,
+ }
+ );
+ assert_eq!(
+ parse_command(r#"{"command":"subscribe"}"#).unwrap(),
+ Command::Subscribe {
+ patterns: Vec::new(),
+ since: None,
+ }
+ );
+ assert_eq!(
+ parse_command(r#"{"command":"subscribe","args":{"since":42}}"#).unwrap(),
+ Command::Subscribe {
+ patterns: Vec::new(),
+ since: Some(42),
+ }
+ );
+ assert_eq!(
+ parse_command(r#"{"command":"unsubscribe"}"#).unwrap(),
+ Command::Unsubscribe
+ );
+ }
+
+ #[test]
+ fn rejects_bad_requests() {
+ assert!(parse_command("not json").is_err());
+ assert!(parse_command(r#"{"command":"subscribe","args":{"patterns":["a.cfg"]}}"#).is_err());
+ assert!(parse_command(r#"{"command":"subscribe","args":{"patterns":{"x":1}}}"#).is_err());
+ assert!(parse_command(r#"{"command":"subscribe","args":{"since":"x"}}"#).is_err());
+ assert!(parse_command(r#"{"command":"subscribe","args":{"since":-1}}"#).is_err());
+ let err = parse_command(r#"{"command":"subscribe","args":{"patterns":{"x":"{a}{b}"}}}"#)
+ .unwrap_err();
+ assert_eq!(
+ err.to_string(),
+ "pattern 'x': only one placeholder per component in '{a}{b}'"
+ );
+ let err = parse_command(r#"{"command":"bogus"}"#).unwrap_err();
+ assert_eq!(err.to_string(), "unknown command 'bogus'");
+ }
+
+ #[test]
+ fn formats_events() {
+ let params = Params::from([(
+ "guest",
+ vec![Captures::from([("node", "n1"), ("vmid", "100")])],
+ )]);
+ let write = Message::Event(Event {
+ seq: 7,
+ kind: EventKind::Write,
+ path: Some("nodes/n1/qemu-server/100.conf"),
+ to: None,
+ params: Some(params),
+ });
+ assert_eq!(
+ write.line(),
+ "{\"event\":{\"seq\":7,\"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 {
+ seq: 8,
+ kind: EventKind::Rename,
+ path: Some("a"),
+ to: Some("datacenter.cfg"),
+ params: Some(params),
+ });
+ assert_eq!(
+ rename.line(),
+ "{\"event\":{\"seq\":8,\"type\":\"rename\",\"path\":\"a\",\"to\":\"datacenter.cfg\",\"params\":{\"dc\":[{}]}}}\n"
+ );
+ let resync = Message::Event(Event {
+ seq: 9,
+ kind: EventKind::Resync,
+ path: None,
+ to: None,
+ params: None,
+ });
+ assert_eq!(
+ resync.line(),
+ "{\"event\":{\"seq\":9,\"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"
+ );
+ }
+}
diff --git a/src/rust/pmxcfs-notify/src/registry.rs b/src/rust/pmxcfs-notify/src/registry.rs
new file mode 100644
index 0000000..29fe4dd
--- /dev/null
+++ b/src/rust/pmxcfs-notify/src/registry.rs
@@ -0,0 +1,405 @@
+use std::collections::{HashMap, HashSet};
+use std::os::fd::RawFd;
+use std::sync::{Mutex, MutexGuard};
+
+use crate::conn::Conn;
+use crate::protocol::{Event, EventKind, Message};
+use crate::ring::{Entry, Ring};
+use crate::template::Template;
+
+// Entries copied out of the ring per lock hold. Bounds how long the
+// mutating thread can wait behind a delivery walk.
+const BATCH: usize = 256;
+
+// Batches one connection may take per delivery round before the others get a
+// turn, so a client catching up a large backlog cannot starve them.
+const WALK_ROUNDS: usize = 8;
+
+pub(crate) fn lock_ring(ring: &Mutex<Ring>) -> MutexGuard<'_, Ring> {
+ ring.lock().unwrap_or_else(|poisoned| poisoned.into_inner())
+}
+
+#[derive(Default)]
+pub(crate) struct Registry {
+ conns: HashMap<u64, Conn>,
+ next_id: u64,
+ #[cfg(test)]
+ pub(crate) panic_on_delivery: bool,
+}
+
+impl Registry {
+ pub(crate) fn add(&mut self, mut conn: Conn, head: u64) -> u64 {
+ conn.cursor = head;
+ let id = self.next_id;
+ self.next_id += 1;
+ self.conns.insert(id, conn);
+ id
+ }
+
+ pub(crate) fn remove(&mut self, id: u64) -> Option<Conn> {
+ self.conns.remove(&id)
+ }
+
+ pub(crate) fn get_mut(&mut self, id: u64) -> Option<&mut Conn> {
+ self.conns.get_mut(&id)
+ }
+
+ pub(crate) fn drain_dead(&mut self) -> Vec<Conn> {
+ self.conns
+ .extract_if(|_, conn| conn.dead)
+ .map(|(_, conn)| conn)
+ .collect()
+ }
+
+ pub(crate) fn fds(&self) -> Vec<(u64, RawFd, bool)> {
+ self.conns
+ .iter()
+ .map(|(id, conn)| (*id, conn.fd(), conn.lagging()))
+ .collect()
+ }
+
+ pub(crate) fn len(&self) -> usize {
+ self.conns.len()
+ }
+
+ pub(crate) fn clear(&mut self) {
+ self.conns.clear();
+ }
+
+ // A fresh subscription starts at the tail, or right after `since` when
+ // the ring still holds it, and with a resync when it does not. On a live
+ // subscription the cursor stays put, what the client asks for may still
+ // be in flight to it.
+ pub(crate) fn subscribe(
+ &mut self,
+ id: u64,
+ patterns: Vec<(String, Template)>,
+ since: Option<u64>,
+ ring: &Mutex<Ring>,
+ ) {
+ let Some(conn) = self.conns.get_mut(&id) else {
+ return;
+ };
+ let fresh = !conn.subscribed;
+ conn.templates = patterns;
+ conn.subscribed = true;
+ if !fresh {
+ return;
+ }
+ let ring = lock_ring(ring);
+ conn.cursor = ring.head();
+ conn.resync_pending = false;
+ let Some(seq) = since else {
+ return;
+ };
+ match ring.resume(seq) {
+ Some(pos) => conn.cursor = pos,
+ None => conn.resync_pending = true,
+ }
+ }
+
+ pub(crate) fn unsubscribe(&mut self, id: u64, head: u64) {
+ let Some(conn) = self.conns.get_mut(&id) else {
+ return;
+ };
+ conn.templates.clear();
+ conn.subscribed = false;
+ conn.resync_pending = false;
+ conn.cursor = head;
+ }
+
+ /// Sends to every connection that can take data, a lagging one only
+ /// when its socket was reported writable. Returns whether a connection
+ /// stopped at its batch cap with more still to send.
+ pub(crate) fn deliver(&mut self, ring: &Mutex<Ring>, writable: &HashSet<u64>) -> bool {
+ let mut more = false;
+ for (id, conn) in &mut self.conns {
+ if conn.dead || (conn.lagging() && !writable.contains(id)) {
+ continue;
+ }
+ more |= walk(ring, conn);
+ }
+ more
+ }
+}
+
+fn resync_line(seq: u64) -> String {
+ Message::Event(Event {
+ seq,
+ kind: EventKind::Resync,
+ path: None,
+ to: None,
+ params: None,
+ })
+ .line()
+}
+
+// Moves a connection's cursor toward the head, queueing the lines it
+// subscribed to, until the socket is full, nothing is left, or it has taken
+// its batch cap for this round. A cursor that fell off the ring turns into
+// one resync at the head. The ring is only locked to copy a batch, never
+// while a line is built or sent. Returns whether the cap stopped it with more
+// still to send.
+fn walk(ring: &Mutex<Ring>, conn: &mut Conn) -> bool {
+ for _ in 0..WALK_ROUNDS {
+ if !conn.flush() {
+ return false;
+ }
+ let (batch, last_seq) = {
+ let ring = lock_ring(ring);
+ if !conn.subscribed {
+ conn.cursor = ring.head();
+ return false;
+ }
+ if conn.cursor < ring.oldest() {
+ conn.resync_pending = true;
+ }
+ // the resync stands for everything up to the head as of now, so
+ // nothing before it may follow it
+ if conn.resync_pending {
+ conn.cursor = ring.head();
+ }
+ let batch: Vec<Entry> = (conn.cursor..ring.head())
+ .take(BATCH)
+ .filter_map(|pos| ring.get(pos).cloned())
+ .collect();
+ (batch, ring.last_seq())
+ };
+ if conn.resync_pending {
+ conn.resync_pending = false;
+ conn.queue(&resync_line(last_seq.unwrap_or(0)));
+ continue;
+ }
+ if batch.is_empty() {
+ return false;
+ }
+ for entry in &batch {
+ conn.cursor += 1;
+ let line = if entry.kind == EventKind::Resync {
+ resync_line(entry.seq)
+ } else {
+ match conn.matches(entry.path.as_deref(), entry.to.as_deref()) {
+ Some(params) => Message::Event(Event {
+ seq: entry.seq,
+ kind: entry.kind,
+ path: entry.path.as_deref(),
+ to: entry.to.as_deref(),
+ params: Some(params),
+ })
+ .line(),
+ None => continue,
+ }
+ };
+ conn.queue(&line);
+ if !conn.flush() {
+ return false;
+ }
+ }
+ }
+ true
+}
+
+// Same rule as path_is_private in the daemon, which also hides these paths
+// from group authorized IPC and FUSE clients.
+pub(crate) 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 std::io::{BufRead, BufReader};
+ use std::os::unix::net::UnixStream;
+ use std::time::Duration;
+
+ use super::*;
+
+ fn entry(seq: u64) -> Entry {
+ Entry {
+ seq,
+ kind: EventKind::Write,
+ path: Some(format!("f{seq}").into()),
+ to: None,
+ }
+ }
+
+ fn all() -> Vec<(String, Template)> {
+ vec![("all".to_owned(), Template::parse("{path...}").unwrap())]
+ }
+
+ fn client(registry: &mut Registry, ring: &Mutex<Ring>) -> (u64, BufReader<UnixStream>) {
+ let (server, client) = UnixStream::pair().unwrap();
+ client
+ .set_read_timeout(Some(Duration::from_millis(300)))
+ .unwrap();
+ let head = lock_ring(ring).head();
+ (
+ registry.add(Conn::new(server, true).unwrap(), head),
+ BufReader::new(client),
+ )
+ }
+
+ fn lines(reader: &mut BufReader<UnixStream>) -> Vec<String> {
+ let mut lines = Vec::new();
+ let mut line = String::new();
+ while reader.read_line(&mut line).is_ok_and(|n| n > 0) {
+ lines.push(line.trim_end().to_owned());
+ line.clear();
+ }
+ lines
+ }
+
+ #[test]
+ fn resume_positions_the_cursor() {
+ let ring = Mutex::new(Ring::new(4));
+ let mut registry = Registry::default();
+ for seq in 1..=3 {
+ lock_ring(&ring).push(entry(seq));
+ }
+ let (id, _reader) = client(&mut registry, &ring);
+
+ registry.subscribe(id, all(), None, &ring);
+ assert_eq!(registry.get_mut(id).unwrap().cursor, 3);
+
+ registry.unsubscribe(id, 3);
+ registry.subscribe(id, all(), Some(1), &ring);
+ assert_eq!(registry.get_mut(id).unwrap().cursor, 1);
+ assert!(!registry.get_mut(id).unwrap().resync_pending);
+
+ registry.unsubscribe(id, 3);
+ registry.subscribe(id, all(), Some(3), &ring);
+ assert_eq!(registry.get_mut(id).unwrap().cursor, 3);
+
+ registry.unsubscribe(id, 3);
+ registry.subscribe(id, all(), Some(9), &ring);
+ assert!(registry.get_mut(id).unwrap().resync_pending);
+
+ // a live subscription keeps its cursor whatever it asks for
+ registry.get_mut(id).unwrap().resync_pending = false;
+ registry.subscribe(id, all(), Some(1), &ring);
+ assert_eq!(registry.get_mut(id).unwrap().cursor, 3);
+ assert!(!registry.get_mut(id).unwrap().resync_pending);
+ }
+
+ #[test]
+ fn walk_delivers_or_resyncs() {
+ let ring = Mutex::new(Ring::new(3));
+ let mut registry = Registry::default();
+ let (id, mut reader) = client(&mut registry, &ring);
+ registry.subscribe(id, all(), None, &ring);
+
+ lock_ring(&ring).push(entry(1));
+ registry.deliver(&ring, &HashSet::new());
+ assert_eq!(
+ lines(&mut reader),
+ vec![
+ "{\"event\":{\"seq\":1,\"type\":\"write\",\"path\":\"f1\",\"params\":{\"all\":[{\"path\":\"f1\"}]}}}"
+ ]
+ );
+
+ for seq in 2..=6 {
+ lock_ring(&ring).push(entry(seq));
+ }
+ registry.deliver(&ring, &HashSet::new());
+ assert_eq!(
+ lines(&mut reader),
+ vec!["{\"event\":{\"seq\":6,\"type\":\"resync\"}}"],
+ "a cursor that fell off gets one resync"
+ );
+
+ lock_ring(&ring).push(entry(7));
+ registry.deliver(&ring, &HashSet::new());
+ assert_eq!(lines(&mut reader).len(), 1);
+ }
+
+ #[test]
+ fn a_resync_at_the_resume_number_is_delivered() {
+ let ring = Mutex::new(Ring::new(4));
+ let mut registry = Registry::default();
+ lock_ring(&ring).push(Entry::marker(5));
+ lock_ring(&ring).push(Entry {
+ seq: 5,
+ kind: EventKind::Resync,
+ path: None,
+ to: None,
+ });
+ let (id, mut reader) = client(&mut registry, &ring);
+ registry.subscribe(id, all(), Some(5), &ring);
+ registry.deliver(&ring, &HashSet::new());
+ assert_eq!(
+ lines(&mut reader),
+ vec!["{\"event\":{\"seq\":5,\"type\":\"resync\"}}"],
+ "the state was replaced at the number the client resumes from"
+ );
+ }
+
+ #[test]
+ fn a_pending_resync_swallows_what_arrived_before_the_walk() {
+ let ring = Mutex::new(Ring::new(3));
+ let mut registry = Registry::default();
+ for seq in 1..=3 {
+ lock_ring(&ring).push(entry(seq));
+ }
+ let (id, mut reader) = client(&mut registry, &ring);
+ registry.subscribe(id, all(), Some(0), &ring);
+ assert!(registry.get_mut(id).unwrap().resync_pending);
+
+ lock_ring(&ring).push(entry(4));
+ registry.deliver(&ring, &HashSet::new());
+ assert_eq!(
+ lines(&mut reader),
+ vec!["{\"event\":{\"seq\":4,\"type\":\"resync\"}}"],
+ "the entry pushed between subscribe and walk is covered by the resync"
+ );
+
+ lock_ring(&ring).push(entry(5));
+ registry.deliver(&ring, &HashSet::new());
+ assert_eq!(lines(&mut reader).len(), 1);
+ }
+
+ #[test]
+ fn walks_hand_the_ring_back_between_batches() {
+ let ring = Mutex::new(Ring::new(2000));
+ let mut registry = Registry::default();
+ let (id, mut reader) = client(&mut registry, &ring);
+ registry.subscribe(id, all(), None, &ring);
+ for seq in 1..=1000 {
+ lock_ring(&ring).push(entry(seq));
+ }
+ // the pair's buffer holds a few hundred lines, so the walk stops
+ // lagging and is resumed as the poll loop would on writability
+ let mut received = 0;
+ for _ in 0..20 {
+ registry.deliver(&ring, &HashSet::from([id]));
+ received += lines(&mut reader).len();
+ if received == 1000 {
+ break;
+ }
+ }
+ assert_eq!(received, 1000, "every entry arrives across batches");
+ assert!(
+ !lock_ring(&ring).entries_in_use(),
+ "no batch outlives its walk"
+ );
+ }
+
+ #[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/ring.rs b/src/rust/pmxcfs-notify/src/ring.rs
new file mode 100644
index 0000000..f6f4594
--- /dev/null
+++ b/src/rust/pmxcfs-notify/src/ring.rs
@@ -0,0 +1,171 @@
+use std::collections::VecDeque;
+use std::sync::Arc;
+
+use crate::protocol::EventKind;
+
+// Reference-counted strings, so copying an entry out of the ring for
+// delivery clones only a pointer.
+#[derive(Clone)]
+pub(crate) struct Entry {
+ pub seq: u64,
+ pub kind: EventKind,
+ pub path: Option<Arc<str>>,
+ pub to: Option<Arc<str>>,
+}
+
+impl Entry {
+ /// Stands for the last mutation before the server started. It carries
+ /// that mutation's number and no path, so a resuming client at that
+ /// number is current, while nothing ever matches it.
+ pub(crate) fn marker(seq: u64) -> Self {
+ Self {
+ seq,
+ kind: EventKind::Write,
+ path: None,
+ to: None,
+ }
+ }
+}
+
+// Positions are absolute append counts, so a cursor stays valid while the
+// ring wraps, and a position below the oldest one means its reader fell
+// off and has to resync.
+pub(crate) struct Ring {
+ entries: VecDeque<Entry>,
+ capacity: usize,
+ head: u64,
+}
+
+impl Ring {
+ pub(crate) fn new(capacity: usize) -> Self {
+ Self {
+ entries: VecDeque::new(),
+ capacity: capacity.max(1),
+ head: 0,
+ }
+ }
+
+ pub(crate) fn push(&mut self, entry: Entry) {
+ if self.entries.len() == self.capacity {
+ self.entries.pop_front();
+ }
+ self.entries.push_back(entry);
+ self.head += 1;
+ }
+
+ pub(crate) fn head(&self) -> u64 {
+ self.head
+ }
+
+ pub(crate) fn oldest(&self) -> u64 {
+ self.head - self.entries.len() as u64
+ }
+
+ pub(crate) fn get(&self, pos: u64) -> Option<&Entry> {
+ if pos < self.oldest() || pos >= self.head {
+ return None;
+ }
+ self.entries.get((pos - self.oldest()) as usize)
+ }
+
+ pub(crate) fn last_seq(&self) -> Option<u64> {
+ self.entries.back().map(|entry| entry.seq)
+ }
+
+ /// Whether any entry is shared with a copy outside the ring.
+ #[cfg(test)]
+ pub(crate) fn entries_in_use(&self) -> bool {
+ self.entries.iter().any(|entry| {
+ entry
+ .path
+ .as_ref()
+ .is_some_and(|p| Arc::strong_count(p) > 1)
+ || entry.to.as_ref().is_some_and(|t| Arc::strong_count(t) > 1)
+ })
+ }
+
+ /// Where a client that saw `seq` last continues. Right after the newest
+ /// entry carrying it, unless that entry is a resync, which the client
+ /// then has to see since the state was replaced at that number.
+ pub(crate) fn resume(&self, seq: u64) -> Option<u64> {
+ self.entries
+ .iter()
+ .rposition(|entry| entry.seq == seq)
+ .map(|index| {
+ let pos = self.oldest() + index as u64;
+ if self.entries[index].kind == EventKind::Resync {
+ pos
+ } else {
+ pos + 1
+ }
+ })
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ fn entry(seq: u64) -> Entry {
+ Entry {
+ seq,
+ kind: EventKind::Write,
+ path: Some(format!("f{seq}").into()),
+ to: None,
+ }
+ }
+
+ #[test]
+ fn positions_survive_wrapping() {
+ let mut ring = Ring::new(3);
+ assert_eq!(ring.head(), 0);
+ assert_eq!(ring.oldest(), 0);
+ assert_eq!(ring.last_seq(), None);
+ assert!(ring.get(0).is_none());
+
+ for seq in 10..15 {
+ ring.push(entry(seq));
+ }
+ assert_eq!(ring.head(), 5);
+ assert_eq!(ring.oldest(), 2);
+ assert_eq!(ring.last_seq(), Some(14));
+ assert!(ring.get(1).is_none());
+ assert_eq!(ring.get(2).unwrap().seq, 12);
+ assert_eq!(ring.get(4).unwrap().seq, 14);
+ assert!(ring.get(5).is_none());
+ }
+
+ #[test]
+ fn resume_positions() {
+ let mut ring = Ring::new(4);
+ for seq in [1, 2, 2, 3] {
+ ring.push(entry(seq));
+ }
+ assert_eq!(ring.resume(1), Some(1));
+ assert_eq!(ring.resume(2), Some(3), "the newest duplicate wins");
+ assert_eq!(ring.resume(3), Some(4));
+ assert_eq!(ring.resume(0), None);
+ ring.push(entry(4));
+ assert_eq!(ring.resume(1), None, "fell off");
+ assert_eq!(ring.resume(4), Some(5));
+ }
+
+ #[test]
+ fn a_resync_at_the_resume_number_is_not_skipped() {
+ let mut ring = Ring::new(4);
+ ring.push(Entry::marker(5));
+ ring.push(Entry {
+ seq: 5,
+ kind: EventKind::Resync,
+ path: None,
+ to: None,
+ });
+ assert_eq!(
+ ring.resume(5),
+ Some(1),
+ "the state was replaced at that number"
+ );
+ ring.push(entry(6));
+ assert_eq!(ring.resume(6), Some(3));
+ }
+}
diff --git a/src/rust/pmxcfs-notify/src/server.rs b/src/rust/pmxcfs-notify/src/server.rs
new file mode 100644
index 0000000..65d5744
--- /dev/null
+++ b/src/rust/pmxcfs-notify/src/server.rs
@@ -0,0 +1,780 @@
+use std::any::Any;
+use std::collections::HashSet;
+use std::fs;
+use std::io;
+use std::os::fd::{AsFd, BorrowedFd, RawFd};
+use std::os::unix::fs::PermissionsExt;
+use std::os::unix::net::{UnixListener, UnixStream};
+use std::panic::{AssertUnwindSafe, catch_unwind};
+use std::path::PathBuf;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Arc, Mutex, MutexGuard};
+use std::thread::JoinHandle;
+use std::time::Duration;
+
+use anyhow::{Context, Error};
+use nix::errno::Errno;
+use nix::poll::{PollFd, PollFlags, PollTimeout, poll};
+use nix::sys::eventfd::{EfdFlags, EventFd};
+use nix::sys::socket::{getsockopt, setsockopt, sockopt};
+use serde_json::{Value, json};
+
+use crate::conn::{Conn, ReadError};
+use crate::normalize_path;
+use crate::protocol::{Command, EventKind, Message, PROTOCOL_VERSION, parse_command};
+use crate::registry::{Registry, lock_ring};
+use crate::ring::{Entry, Ring};
+
+const MAX_CONNECTIONS: usize = 128;
+const SNDBUF: usize = 2 * 1024 * 1024;
+
+pub struct Config {
+ pub path: PathBuf,
+ pub gid: u32,
+ /// Mutations kept for catching up lagging and reconnecting clients.
+ pub ring: usize,
+ /// The number of the last mutation before the server starts, so a
+ /// client resuming at it after a restart is current.
+ pub seq: u64,
+}
+
+pub struct Server {
+ #[cfg_attr(not(test), allow(dead_code))]
+ registry: Arc<Mutex<Registry>>,
+ ring: Arc<Mutex<Ring>>,
+ stop: Arc<AtomicBool>,
+ wake: Arc<EventFd>,
+ thread: Option<JoinHandle<()>>,
+ path: PathBuf,
+}
+
+fn lock(registry: &Mutex<Registry>) -> MutexGuard<'_, Registry> {
+ registry
+ .lock()
+ .unwrap_or_else(|poisoned| poisoned.into_inner())
+}
+
+impl Server {
+ pub fn start(config: Config) -> Result<Server, Error> {
+ match fs::remove_file(&config.path) {
+ Ok(()) => {}
+ Err(err) if err.kind() == io::ErrorKind::NotFound => {}
+ Err(err) => {
+ return Err(err)
+ .with_context(|| format!("failed to remove stale {}", config.path.display()));
+ }
+ }
+ let listener = UnixListener::bind(&config.path)
+ .with_context(|| format!("failed to bind {}", config.path.display()))?;
+ listener.set_nonblocking(true)?;
+ std::os::unix::fs::chown(&config.path, None, Some(config.gid))?;
+ fs::set_permissions(&config.path, fs::Permissions::from_mode(0o660))?;
+
+ let wake = Arc::new(EventFd::from_value_and_flags(
+ 0,
+ EfdFlags::EFD_NONBLOCK | EfdFlags::EFD_CLOEXEC,
+ )?);
+ let registry = Arc::new(Mutex::new(Registry::default()));
+ let mut seeded = Ring::new(config.ring);
+ seeded.push(Entry::marker(config.seq));
+ let ring = Arc::new(Mutex::new(seeded));
+ let stop = Arc::new(AtomicBool::new(false));
+ let worker = Worker {
+ listener,
+ path: config.path.clone(),
+ wake: Arc::clone(&wake),
+ stop: Arc::clone(&stop),
+ registry: Arc::clone(®istry),
+ ring: Arc::clone(&ring),
+ gid: config.gid,
+ accept_failing: false,
+ };
+ let thread = std::thread::Builder::new()
+ .name("pmxcfs-notify".to_owned())
+ .spawn(move || worker.run())?;
+
+ Ok(Server {
+ registry,
+ ring,
+ stop,
+ wake,
+ thread: Some(thread),
+ path: config.path,
+ })
+ }
+
+ /// Records a mutation and wakes the delivery thread. It never touches
+ /// a client socket and only takes the ring lock, which the delivery
+ /// thread holds for one batch copy at a time, so it is safe on the
+ /// mutating thread.
+ pub fn emit(&self, seq: u64, kind: EventKind, path: &str, to: Option<&str>) {
+ if self.stop.load(Ordering::SeqCst) {
+ return;
+ }
+ lock_ring(&self.ring).push(Entry {
+ seq,
+ kind,
+ path: Some(normalize_path(path).into()),
+ to: to.map(|to| normalize_path(to).into()),
+ });
+ let _ = self.wake.write(1);
+ }
+
+ pub fn resync(&self, seq: u64) {
+ if self.stop.load(Ordering::SeqCst) {
+ return;
+ }
+ lock_ring(&self.ring).push(Entry {
+ seq,
+ kind: EventKind::Resync,
+ path: None,
+ to: None,
+ });
+ let _ = self.wake.write(1);
+ }
+
+ pub fn shutdown(mut self) {
+ self.stop_worker();
+ }
+
+ fn stop_worker(&mut self) {
+ let Some(thread) = self.thread.take() else {
+ return;
+ };
+ self.stop.store(true, Ordering::SeqCst);
+ let _ = self.wake.write(1);
+ let _ = thread.join();
+ let _ = fs::remove_file(&self.path);
+ }
+
+ #[cfg(test)]
+ pub(crate) fn connection_count(&self) -> usize {
+ lock(&self.registry).len()
+ }
+}
+
+impl Drop for Server {
+ fn drop(&mut self) {
+ self.stop_worker();
+ }
+}
+
+struct Worker {
+ listener: UnixListener,
+ path: PathBuf,
+ wake: Arc<EventFd>,
+ stop: Arc<AtomicBool>,
+ registry: Arc<Mutex<Registry>>,
+ ring: Arc<Mutex<Ring>>,
+ gid: u32,
+ accept_failing: bool,
+}
+
+fn panic_message(payload: &(dyn Any + Send)) -> String {
+ payload
+ .downcast_ref::<&str>()
+ .map(|s| s.to_string())
+ .or_else(|| payload.downcast_ref::<String>().cloned())
+ .unwrap_or_default()
+}
+
+impl Worker {
+ // Whatever ends the loop, the connections are closed and the socket
+ // path removed on the way out, so clients notice, reconnect and log
+ // rather than staying attached to a thread that no longer serves them.
+ fn run(mut self) {
+ while !self.stop.load(Ordering::SeqCst) {
+ match catch_unwind(AssertUnwindSafe(|| self.iteration())) {
+ Ok(Ok(())) => {}
+ Ok(Err(err)) => {
+ log::error!("poll loop failed, stopping: {err}");
+ break;
+ }
+ Err(payload) => {
+ log::error!("poll loop panicked, stopping: {}", panic_message(&*payload));
+ break;
+ }
+ }
+ }
+ self.stop.store(true, Ordering::SeqCst);
+ lock(&self.registry).clear();
+ let _ = fs::remove_file(&self.path);
+ }
+
+ fn iteration(&mut self) -> Result<(), Error> {
+ let (dead, clients): (Vec<Conn>, Vec<(u64, RawFd, bool)>) = {
+ let mut registry = lock(&self.registry);
+ (registry.drain_dead(), registry.fds())
+ };
+ drop(dead);
+
+ let mut fds: Vec<PollFd> = Vec::with_capacity(clients.len() + 2);
+ fds.push(PollFd::new(self.listener.as_fd(), PollFlags::POLLIN));
+ fds.push(PollFd::new(self.wake.as_fd(), PollFlags::POLLIN));
+ for (_, fd, lagging) in &clients {
+ // Client fds are closed by this thread only, so the raw fds
+ // collected under the lock stay valid for the whole poll.
+ let fd = unsafe { BorrowedFd::borrow_raw(*fd) };
+ let mut flags = PollFlags::POLLIN;
+ if *lagging {
+ flags |= PollFlags::POLLOUT;
+ }
+ fds.push(PollFd::new(fd, flags));
+ }
+
+ match poll(&mut fds, PollTimeout::NONE) {
+ Ok(_) => {}
+ Err(Errno::EINTR) => return Ok(()),
+ // out of memory is transient, so back off and re-poll rather than
+ // tear the notifier down for the rest of the daemon's life
+ Err(Errno::ENOMEM) => {
+ log::warn!("poll transiently out of memory, retrying");
+ std::thread::sleep(Duration::from_millis(100));
+ return Ok(());
+ }
+ Err(err) => return Err(err.into()),
+ }
+ let revents: Vec<PollFlags> = fds
+ .iter()
+ .map(|fd| fd.revents().unwrap_or(PollFlags::empty()))
+ .collect();
+ drop(fds);
+
+ if revents[1].intersects(PollFlags::POLLIN) {
+ let _ = self.wake.read();
+ }
+ if self.stop.load(Ordering::SeqCst) {
+ return Ok(());
+ }
+ if revents[0].intersects(PollFlags::POLLIN) {
+ self.accept_all();
+ }
+ let readable =
+ PollFlags::POLLIN | PollFlags::POLLHUP | PollFlags::POLLERR | PollFlags::POLLNVAL;
+ let mut writable = HashSet::new();
+ for (index, (id, _, _)) in clients.iter().enumerate() {
+ let flags = revents[index + 2];
+ if flags.intersects(readable) {
+ self.service(*id);
+ }
+ if flags.intersects(PollFlags::POLLOUT) {
+ writable.insert(*id);
+ }
+ }
+ let mut registry = lock(&self.registry);
+ #[cfg(test)]
+ if registry.panic_on_delivery {
+ panic!("injected");
+ }
+ let more = registry.deliver(&self.ring, &writable);
+ drop(registry);
+ // a connection that stopped at its batch cap still has data queued, so
+ // wake the loop to resume it once the other fds have had a turn
+ if more {
+ let _ = self.wake.write(1);
+ }
+ Ok(())
+ }
+
+ fn accept_all(&mut self) {
+ loop {
+ match self.listener.accept() {
+ Ok((stream, _)) => {
+ self.accept_failing = false;
+ self.admit(stream);
+ }
+ Err(err) if err.kind() == io::ErrorKind::WouldBlock => break,
+ Err(err) if err.kind() == io::ErrorKind::Interrupted => continue,
+ Err(err) => {
+ // the pending connection stays in the backlog and keeps
+ // the listener readable, so do not spin on it. a cause
+ // that lasts, like exhausted descriptors, would flood
+ // the log with one warning per round
+ if !self.accept_failing {
+ log::warn!("accept failed: {err}");
+ self.accept_failing = true;
+ }
+ std::thread::sleep(Duration::from_millis(100));
+ break;
+ }
+ }
+ }
+ }
+
+ fn admit(&self, stream: UnixStream) {
+ let cred = match getsockopt(&stream, sockopt::PeerCredentials) {
+ 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() != self.gid {
+ log::warn!(
+ "connection from bad user {}/{} rejected",
+ cred.uid(),
+ cred.gid()
+ );
+ return;
+ }
+ if lock(&self.registry).len() >= MAX_CONNECTIONS {
+ log::warn!("connection limit of {MAX_CONNECTIONS} reached, refusing");
+ return;
+ }
+ // root may raise the send buffer past wmem_max, which gives a
+ // lagging client room for thousands of events instead of hundreds
+ 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");
+ }
+ match Conn::new(stream, privileged) {
+ Ok(conn) => {
+ let head = lock_ring(&self.ring).head();
+ lock(&self.registry).add(conn, head);
+ }
+ Err(err) => log::warn!("failed to set up connection: {err}"),
+ }
+ }
+
+ fn service(&self, id: u64) {
+ let mut registry = lock(&self.registry);
+ let lines = match registry.get_mut(id).map(Conn::read_lines) {
+ None => return,
+ Some(Ok(lines)) => lines,
+ Some(Err(err)) => {
+ match err {
+ ReadError::Eof => {}
+ ReadError::LineTooLong => log::warn!("closing connection, line too long"),
+ ReadError::Io(err) => log::debug!("closing connection: {err}"),
+ }
+ registry.remove(id);
+ return;
+ }
+ };
+ for line in lines {
+ let reply = match parse_command(&line) {
+ Ok(Command::Hello) => {
+ // the newest number lets a client that saw no event yet
+ // resume after a restart instead of resyncing
+ let seq = lock_ring(&self.ring).last_seq().unwrap_or(0);
+ Message::Ok(json!({ "protocol": PROTOCOL_VERSION, "seq": seq })).line()
+ }
+ Ok(Command::Subscribe { patterns, since }) => {
+ registry.subscribe(id, patterns, since, &self.ring);
+ Message::Ok(Value::Null).line()
+ }
+ Ok(Command::Unsubscribe) => {
+ let head = lock_ring(&self.ring).head();
+ registry.unsubscribe(id, head);
+ Message::Ok(Value::Null).line()
+ }
+ Err(err) => Message::Error(err.to_string()).line(),
+ };
+ let Some(conn) = registry.get_mut(id) else {
+ return;
+ };
+ conn.queue(&reply);
+ conn.flush();
+ if conn.dead {
+ return;
+ }
+ }
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use std::io::{BufRead, BufReader, Write};
+ use std::os::unix::fs::MetadataExt;
+ use std::sync::atomic::AtomicUsize;
+ use std::time::Instant;
+
+ 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(ring: usize) -> (Server, PathBuf, TempDir) {
+ start_at(ring, 0)
+ }
+
+ fn start_at(ring: usize, seq: u64) -> (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,
+ ring,
+ seq,
+ })
+ .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,
+ since: Option<u64>,
+ ) {
+ let since = since.map(|s| format!(",\"since\":{s}")).unwrap_or_default();
+ let line =
+ format!("{{\"command\":\"subscribe\",\"args\":{{\"patterns\":{patterns}{since}}}}}");
+ assert_eq!(request(stream, reader, &line), "{\"ok\":null}\n");
+ }
+
+ 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 shortly, None when the line stays quiet.
+ 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,\"seq\":0}}\n"
+ );
+ assert_eq!(server.connection_count(), 1);
+ assert_eq!(
+ request(&mut stream, &mut reader, r#"{"command":"bogus"}"#),
+ "{\"error\":\"unknown command 'bogus'\"}\n"
+ );
+ server.emit(1, EventKind::Write, "datacenter.cfg", None);
+ subscribe(
+ &mut stream,
+ &mut reader,
+ r#"{"guest":"/nodes/{node}/qemu-server/{vmid}.conf","dc":"datacenter.cfg"}"#,
+ None,
+ );
+ // a rejected subscription leaves the previous one in place
+ assert_eq!(
+ request(
+ &mut stream,
+ &mut reader,
+ r#"{"command":"subscribe","args":{"patterns":{"bad":"{a}{b}"}}}"#
+ ),
+ "{\"error\":\"pattern 'bad': only one placeholder per component in '{a}{b}'\"}\n"
+ );
+
+ server.emit(2, EventKind::Write, "storage.cfg", None);
+ server.emit(3, EventKind::Create, "/nodes/n1/qemu-server/100.conf", None);
+ server.emit(
+ 4,
+ EventKind::Write,
+ "nodes/n1/qemu-server/100.conf.tmp.1",
+ None,
+ );
+ server.emit(
+ 5,
+ EventKind::Rename,
+ "nodes/n1/qemu-server/100.conf.tmp.1",
+ Some("datacenter.cfg"),
+ );
+ server.resync(6);
+
+ let create = event(&mut reader);
+ assert_eq!(create["seq"], 3);
+ assert_eq!(create["type"], "create");
+ assert_eq!(create["path"], "nodes/n1/qemu-server/100.conf");
+ assert!(create.get("to").is_none());
+ assert_eq!(
+ create["params"],
+ json!({ "guest": [{ "node": "n1", "vmid": "100" }] })
+ );
+
+ let rename = event(&mut reader);
+ assert_eq!(rename["seq"], 5);
+ assert_eq!(rename["type"], "rename");
+ assert_eq!(rename["to"], "datacenter.cfg");
+ assert_eq!(rename["params"], json!({ "dc": [{}] }));
+
+ assert_eq!(event(&mut reader), json!({ "seq": 6, "type": "resync" }));
+
+ assert_eq!(
+ request(&mut stream, &mut reader, r#"{"command":"unsubscribe"}"#),
+ "{\"ok\":null}\n"
+ );
+ server.emit(7, EventKind::Write, "datacenter.cfg", None);
+ server.resync(8);
+ // a fresh subscription without a resume point starts at the tail
+ subscribe(&mut stream, &mut reader, r#"{"dc":"datacenter.cfg"}"#, None);
+ server.emit(9, EventKind::Write, "datacenter.cfg", None);
+ assert_eq!(event(&mut reader)["seq"], 9);
+ 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...}"}"#, None);
+ server.emit(1, EventKind::Write, "nodes/n1/priv/ssl.key", None);
+ server.emit(2, EventKind::Write, "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 panic_in_the_loop_closes_the_connections() {
+ let (server, path, _dir) = start(64);
+ let (mut stream, mut reader) = connect(&path);
+ subscribe(&mut stream, &mut reader, r#"{"all":"{path...}"}"#, None);
+ lock(&server.registry).panic_on_delivery = true;
+ server.emit(1, EventKind::Write, "a", None);
+
+ let mut line = String::new();
+ assert_eq!(reader.read_line(&mut line).unwrap(), 0, "client sees EOF");
+ wait_for(|| server.thread.as_ref().is_some_and(|t| t.is_finished()));
+ assert!(!path.exists());
+ assert!(UnixStream::connect(&path).is_err());
+ server.shutdown();
+ }
+
+ #[test]
+ fn client_close_is_noticed() {
+ let (server, path, _dir) = start(64);
+ let (stream, mut reader) = connect(&path);
+ let mut stream = stream;
+ 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 lagging_client_is_caught_up_or_resynced() {
+ let (server, path, _dir) = start(100);
+ let (mut stream, mut reader) = connect(&path);
+ subscribe(&mut stream, &mut reader, r#"{"all":"{path...}"}"#, None);
+ server.emit(1, EventKind::Write, "warm", None);
+ assert_eq!(event(&mut reader)["seq"], 1);
+
+ // delivery interleaves with the burst, so where the cursor falls
+ // off the ring and how much still goes out depends on scheduling.
+ // Fixed are at least one resync, no event twice, and after the last
+ // resync a gapless run of events up to the newest one.
+ let big = "x".repeat(3000);
+ for seq in 2..=1201 {
+ server.emit(seq, EventKind::Write, &big, None);
+ }
+
+ let mut events = vec![event(&mut reader)];
+ while let Some(ev) = pending(&mut reader) {
+ events.push(ev);
+ }
+ let writes: Vec<u64> = events
+ .iter()
+ .filter(|ev| ev["type"] == "write")
+ .map(|ev| ev["seq"].as_u64().unwrap())
+ .collect();
+ let resyncs: Vec<u64> = events
+ .iter()
+ .filter(|ev| ev["type"] == "resync")
+ .map(|ev| ev["seq"].as_u64().unwrap())
+ .collect();
+ assert!(!resyncs.is_empty(), "the cursor never fell off");
+ assert!(writes.len() < 1200, "delivered {}", writes.len());
+ assert!(
+ writes.windows(2).all(|w| w[0] < w[1]),
+ "an event arrived twice"
+ );
+ let last_resync = events
+ .iter()
+ .rposition(|ev| ev["type"] == "resync")
+ .unwrap();
+ let tail: Vec<u64> = events[last_resync + 1..]
+ .iter()
+ .map(|ev| ev["seq"].as_u64().unwrap())
+ .collect();
+ assert!(
+ tail.windows(2).all(|w| w[1] == w[0] + 1),
+ "gap after the last resync"
+ );
+ let newest = tail.last().or(resyncs.last()).copied();
+ assert_eq!(newest, Some(1201), "the newest event never arrived");
+
+ server.emit(2000, EventKind::Write, "after", None);
+ let after = event(&mut reader);
+ assert_eq!(after["seq"], 2000);
+ assert_eq!(after["path"], "after");
+ assert_eq!(server.connection_count(), 1);
+ }
+
+ #[test]
+ fn a_client_at_the_persisted_version_is_current() {
+ let (server, path, _dir) = start_at(100, 41);
+ let all = r#"{"all":"{path...}"}"#;
+
+ let (mut stream, mut reader) = connect(&path);
+ assert_eq!(
+ request(&mut stream, &mut reader, r#"{"command":"hello"}"#),
+ "{\"ok\":{\"protocol\":1,\"seq\":41}}\n",
+ "hello tells a client where the daemon stands"
+ );
+ subscribe(&mut stream, &mut reader, all, Some(41));
+ assert!(pending(&mut reader).is_none(), "nothing to catch up on");
+ server.emit(42, EventKind::Write, "a", None);
+ assert_eq!(event(&mut reader)["seq"], 42);
+ assert!(
+ pending(&mut reader).is_none(),
+ "the marker is never delivered"
+ );
+ drop(stream);
+ drop(reader);
+
+ for since in [40, 43] {
+ let (mut stream, mut reader) = connect(&path);
+ subscribe(&mut stream, &mut reader, all, Some(since));
+ assert_eq!(
+ event(&mut reader),
+ json!({ "seq": 42, "type": "resync" }),
+ "a number the ring cannot vouch for gets a resync"
+ );
+ drop(stream);
+ drop(reader);
+ }
+
+ let (mut stream, mut reader) = connect(&path);
+ subscribe(&mut stream, &mut reader, all, None);
+ assert!(pending(&mut reader).is_none());
+ server.emit(43, EventKind::Write, "b", None);
+ assert_eq!(event(&mut reader)["seq"], 43);
+ }
+
+ #[test]
+ fn since_resumes_where_the_client_left_off() {
+ let (server, path, _dir) = start(100);
+ let all = r#"{"all":"{path...}"}"#;
+
+ // nothing recorded yet, so a resume point cannot be honored
+ let (mut stream, mut reader) = connect(&path);
+ subscribe(&mut stream, &mut reader, all, Some(1));
+ assert_eq!(event(&mut reader), json!({ "seq": 0, "type": "resync" }));
+ drop(stream);
+ drop(reader);
+
+ let (mut stream, mut reader) = connect(&path);
+ subscribe(&mut stream, &mut reader, all, None);
+ for (seq, name) in [(1, "a"), (2, "b"), (3, "c")] {
+ server.emit(seq, EventKind::Write, name, None);
+ assert_eq!(event(&mut reader)["seq"], seq);
+ }
+ drop(stream);
+ drop(reader);
+ wait_for(|| server.connection_count() == 0);
+
+ server.emit(4, EventKind::Write, "d", None);
+ server.emit(5, EventKind::Write, "e", None);
+
+ let (mut stream, mut reader) = connect(&path);
+ subscribe(&mut stream, &mut reader, all, Some(3));
+ assert_eq!(event(&mut reader)["path"], "d");
+ assert_eq!(event(&mut reader)["path"], "e");
+ assert!(pending(&mut reader).is_none());
+ server.emit(6, EventKind::Write, "f", None);
+ assert_eq!(event(&mut reader)["seq"], 6);
+ drop(stream);
+ drop(reader);
+
+ let (mut stream, mut reader) = connect(&path);
+ subscribe(&mut stream, &mut reader, all, Some(6));
+ assert!(pending(&mut reader).is_none());
+ server.emit(7, EventKind::Write, "g", None);
+ assert_eq!(event(&mut reader)["seq"], 7);
+ drop(stream);
+ drop(reader);
+
+ let (mut stream, mut reader) = connect(&path);
+ subscribe(&mut stream, &mut reader, all, Some(99));
+ assert_eq!(event(&mut reader), json!({ "seq": 7, "type": "resync" }));
+ assert!(pending(&mut reader).is_none());
+ drop(stream);
+ drop(reader);
+
+ for seq in 8..=200 {
+ server.emit(seq, EventKind::Write, "h", None);
+ }
+ let (mut stream, mut reader) = connect(&path);
+ subscribe(&mut stream, &mut reader, all, Some(3));
+ assert_eq!(event(&mut reader), json!({ "seq": 200, "type": "resync" }));
+ assert!(pending(&mut reader).is_none());
+ }
+}
diff --git a/src/rust/pmxcfs-notify/src/template.rs b/src/rust/pmxcfs-notify/src/template.rs
new file mode 100644
index 0000000..a7367eb
--- /dev/null
+++ b/src/rust/pmxcfs-notify/src/template.rs
@@ -0,0 +1,254 @@
+use std::collections::BTreeMap;
+
+use anyhow::{Error, bail};
+
+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 entry, template and connection on the delivery thread.
+#[derive(Debug, PartialEq, Eq)]
+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 Template {
+ pub(crate) fn parse(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 })
+ }
+
+ 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,
+ }
+ }
+ None
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ 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!(
+ Template::parse("/datacenter.cfg").unwrap(),
+ Template {
+ components: vec![literal("datacenter.cfg")]
+ }
+ );
+ assert_eq!(
+ Template::parse("nodes/{node}/qemu-server/{vmid}.conf").unwrap(),
+ Template {
+ components: vec![
+ literal("nodes"),
+ param("", "node", ""),
+ literal("qemu-server"),
+ param("", "vmid", ".conf"),
+ ]
+ }
+ );
+ assert_eq!(
+ Template::parse("{path...}").unwrap(),
+ Template {
+ components: vec![Component::Rest("path".to_owned())]
+ }
+ );
+ assert_eq!(
+ Template::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 = Template::parse(text).unwrap_err().to_string();
+ assert!(err.contains(message), "{text:?}: {err}");
+ }
+ }
+
+ #[test]
+ fn matches_paths() {
+ let guest = Template::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 = Template::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 = Template::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 = Template::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 = Template::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] 11+ messages in thread
* [PATCH pve-cluster 03/10] rust: ffi: add C ABI staticlib for pmxcfs
2026-09-18 14:41 [RFC cluster/manager 00/10] pmxcfs: add a change notification socket Hannes Laimer
2026-09-18 14:41 ` [PATCH pve-cluster 01/10] buildsys: add rust workspace under src/rust Hannes Laimer
2026-09-18 14:41 ` [PATCH pve-cluster 02/10] rust: notify: add change notification socket server Hannes Laimer
@ 2026-09-18 14:41 ` Hannes Laimer
2026-09-18 14:41 ` [PATCH pve-cluster 04/10] pmxcfs: memdb: add change notification hook Hannes Laimer
` (6 subsequent siblings)
9 siblings, 0 replies; 11+ messages in thread
From: Hannes Laimer @ 2026-09-18 14:41 UTC (permalink / raw)
To: pve-devel
pmxcfs is a C daemon, so the notification server needs a small C
facing surface it can link statically. Three functions cover start,
event emission and shutdown, and a log callback taking a syslog
priority routes messages into the daemon's own logging. The emit call
takes the memdb version of the mutation along, which becomes the
sequence number clients see and resume from.
Rust aborts the process when a panic reaches the C caller. That would
take down /etc/pve because of a bug in an optional feature, so every
entry point catches a panic at that boundary, and one only disables
the notifier for the rest of the daemon's life. It also drops every
connection, so clients notice, reconnect and log the outage rather
than waiting on a socket that will never speak again.
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 | 29 ++
src/rust/pmxcfs-ffi/src/lib.rs | 309 ++++++++++++++++++++
4 files changed, 356 insertions(+)
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 6de0ab8..9324685 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..811213d
--- /dev/null
+++ b/src/rust/pmxcfs-ffi/include/pmxcfs-notify.h
@@ -0,0 +1,29 @@
+#ifndef PMXCFS_NOTIFY_H
+#define PMXCFS_NOTIFY_H
+
+#include <stdint.h>
+#include <sys/types.h>
+
+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);
+
+/* seq is the version of the last mutation so far, a client resuming at it is current */
+int pmxcfs_notify_init(
+ const char *socket_path, gid_t gid, pmxcfs_notify_log_fn log_cb, uint64_t seq
+);
+void pmxcfs_notify_emit(
+ enum pmxcfs_notify_type type, uint64_t seq, 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..f8a9629
--- /dev/null
+++ b/src/rust/pmxcfs-ffi/src/lib.rs
@@ -0,0 +1,309 @@
+//! 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 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::{AssertUnwindSafe, catch_unwind};
+use std::path::PathBuf;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Mutex, MutexGuard, OnceLock};
+
+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 TYPE_CREATE: c_int = 0;
+const TYPE_WRITE: c_int = 1;
+const TYPE_MTIME: c_int = 2;
+const TYPE_RENAME: c_int = 3;
+const TYPE_DELETE: c_int = 4;
+const TYPE_MKDIR: c_int = 5;
+const TYPE_RESYNC: c_int = 6;
+const RING_SIZE: 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: OnceLock<()> = OnceLock::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 = record.args().to_string().replace('\0', " ");
+ let Ok(msg) = CString::new(msg) else {
+ return;
+ };
+ 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) -> R {
+ if DISABLED.load(Ordering::SeqCst) {
+ return fallback;
+ }
+ match catch_unwind(AssertUnwindSafe(f)) {
+ Ok(result) => result,
+ Err(payload) => {
+ DISABLED.store(true, Ordering::SeqCst);
+ let reason = payload
+ .downcast_ref::<&str>()
+ .map(|s| s.to_string())
+ .or_else(|| payload.downcast_ref::<String>().cloned())
+ .unwrap_or_default();
+ let _ = catch_unwind(|| {
+ log::error!("{what} panicked, notifications disabled: {reason}");
+ });
+ // Drop every connection so clients notice, reconnect and log,
+ // instead of sitting on a socket that stays silent for good.
+ let _ = catch_unwind(|| {
+ let server = lock(&SERVER).take();
+ if let Some(server) = server {
+ server.shutdown();
+ }
+ });
+ fallback
+ }
+ }
+}
+
+unsafe fn c_str(ptr: *const c_char) -> Option<String> {
+ if ptr.is_null() {
+ return None;
+ }
+ let bytes = unsafe { CStr::from_ptr(ptr) }.to_bytes();
+ Some(String::from_utf8_lossy(bytes).into_owned())
+}
+
+/// # 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>,
+ seq: u64,
+) -> c_int {
+ guard("init", -1, || {
+ if socket_path.is_null() {
+ return -1;
+ }
+ let path = PathBuf::from(OsStr::from_bytes(
+ unsafe { CStr::from_ptr(socket_path) }.to_bytes(),
+ ));
+ *lock(&LOG_CB) = log_cb;
+ LOGGER_INSTALLED.get_or_init(|| {
+ 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.clone(),
+ gid,
+ ring: RING_SIZE,
+ seq,
+ }) {
+ Ok(started) => {
+ *server = Some(started);
+ log::info!("listening on {}", path.display());
+ 0
+ }
+ Err(err) => {
+ log::error!("failed to start notification socket: {err:#}");
+ -1
+ }
+ }
+ })
+}
+
+/// # Safety
+///
+/// `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: c_int,
+ seq: u64,
+ path: *const c_char,
+ to: *const c_char,
+) {
+ guard("emit", (), || {
+ let kind = match kind {
+ TYPE_CREATE => EventKind::Create,
+ TYPE_WRITE => EventKind::Write,
+ TYPE_MTIME => EventKind::Mtime,
+ TYPE_RENAME => EventKind::Rename,
+ TYPE_DELETE => EventKind::Delete,
+ TYPE_MKDIR => EventKind::Mkdir,
+ TYPE_RESYNC => EventKind::Resync,
+ other => {
+ log::warn!("ignoring event with unknown type {other}");
+ return;
+ }
+ };
+ let server = lock(&SERVER);
+ let Some(server) = server.as_ref() else {
+ return;
+ };
+ let path = unsafe { c_str(path) };
+ let to = unsafe { c_str(to) };
+ match (kind, path) {
+ (EventKind::Resync, _) => {
+ server.resync(seq);
+ }
+ (kind, Some(path)) => {
+ server.emit(seq, kind, &path, to.as_deref());
+ }
+ (_, None) => {}
+ }
+ })
+}
+
+#[unsafe(no_mangle)]
+pub extern "C" fn pmxcfs_notify_shutdown() {
+ let _ = catch_unwind(|| {
+ let server = lock(&SERVER).take();
+ if let Some(server) = server {
+ server.shutdown();
+ }
+ *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));
+ }
+
+ #[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), 7) };
+ assert_eq!(rc, 0);
+ let path = CString::new("nodes/n1/qemu-server/100.conf").unwrap();
+ unsafe { pmxcfs_notify_emit(TYPE_WRITE, 1, path.as_ptr(), std::ptr::null()) };
+ unsafe { pmxcfs_notify_emit(TYPE_RESYNC, 2, std::ptr::null(), std::ptr::null()) };
+ unsafe { pmxcfs_notify_emit(99, 3, path.as_ptr(), std::ptr::null()) };
+ pmxcfs_notify_shutdown();
+ }
+
+ let rc = unsafe { pmxcfs_notify_init(std::ptr::null(), gid, Some(collect), 7) };
+ assert_eq!(rc, -1);
+ let _ = std::fs::remove_dir_all(&dir);
+
+ let lines = lock(&LINES);
+ let listening = lines
+ .iter()
+ .filter(|(prio, msg)| *prio == libc::LOG_INFO && msg.starts_with("listening on "))
+ .count();
+ assert_eq!(listening, 2);
+ assert!(
+ lines
+ .iter()
+ .any(|(prio, msg)| *prio == libc::LOG_WARNING && msg.contains("unknown type 99"))
+ );
+ }
+
+ #[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), 7) },
+ 0
+ );
+ let mut client = UnixStream::connect(&path).unwrap();
+ client
+ .set_read_timeout(Some(Duration::from_secs(5)))
+ .unwrap();
+ // a reply proves the server side accepted the connection, so the
+ // shutdown below closes an established socket rather than a
+ // pending one, which 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::SeqCst));
+ 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), 7) },
+ -1
+ );
+ DISABLED.store(false, Ordering::SeqCst);
+ let _ = std::fs::remove_dir_all(&dir);
+ }
+}
--
2.47.3
^ permalink raw reply related [flat|nested] 11+ messages in thread
* [PATCH pve-cluster 04/10] pmxcfs: memdb: add change notification hook
2026-09-18 14:41 [RFC cluster/manager 00/10] pmxcfs: add a change notification socket Hannes Laimer
` (2 preceding siblings ...)
2026-09-18 14:41 ` [PATCH pve-cluster 03/10] rust: ffi: add C ABI staticlib for pmxcfs Hannes Laimer
@ 2026-09-18 14:41 ` Hannes Laimer
2026-09-18 14:41 ` [PATCH pve-cluster 05/10] buildsys: link pmxcfs against the rust notify staticlib Hannes Laimer
` (5 subsequent siblings)
9 siblings, 0 replies; 11+ messages in thread
From: Hannes Laimer @ 2026-09-18 14:41 UTC (permalink / raw)
To: pve-devel
Every mutation of the cluster file system, local or delivered from
another node, passes through memdb, which makes it the one place where
a change notification can be raised for all of them. Add a hook the
daemon installs at startup and call it as the last step of a mutation,
after the change is persisted and the guest list updated. An observer
is then told only about changes that survived, and whatever it reads
next, the guest list included, already reflects them. A node that
takes the whole state from the cluster during a synchronization raises
a resync in place of the mutations it never saw, since its state is
replaced as a whole then.
The hook stays a plain function pointer rather than a direct call
into the notifier, since memdb is part of the static C library the
database tools and the unit tests also link, which must not drag the
socket server along. The event types come straight from the notifier's
header, so the values cross the boundary without translation.
Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
src/pmxcfs/Makefile | 1 +
src/pmxcfs/database.c | 2 ++
src/pmxcfs/memdb.c | 24 ++++++++++++++++++++++++
src/pmxcfs/memdb.h | 11 +++++++++++
4 files changed, 38 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..486fe84 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, memdb->root->version, NULL, NULL);
+
result = TRUE;
ret:
diff --git a/src/pmxcfs/memdb.c b/src/pmxcfs/memdb.c
index 77d7652..a6a2a62 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, uint64_t seq, const char *path, const char *to) {
+ if (notify_hook) {
+ notify_hook(type, seq, 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, memdb->root->version, path, NULL);
+
ret = 0;
ret:
@@ -824,6 +838,10 @@ static int memdb_pwrite(
vmlist_register_vm(vmtype, vmid, nodename);
}
+ memdb_notify(
+ old ? PMXCFS_NOTIFY_WRITE : PMXCFS_NOTIFY_CREATE, memdb->root->version, path, NULL
+ );
+
ret = count;
ret:
@@ -929,6 +947,8 @@ int memdb_mtime(memdb_t *memdb, const char *path, guint32 writer, guint32 mtime)
}
}
+ memdb_notify(PMXCFS_NOTIFY_MTIME, memdb->root->version, path, NULL);
+
ret = 0;
ret:
@@ -1191,6 +1211,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, memdb->root->version, from, to);
+
ret = 0;
ret:
@@ -1264,6 +1286,8 @@ int memdb_delete(memdb_t *memdb, const char *path, guint32 writer, guint32 mtime
vmlist_delete_vm(vmid);
}
+ memdb_notify(PMXCFS_NOTIFY_DELETE, memdb->root->version, path, NULL);
+
ret = 0;
ret:
diff --git a/src/pmxcfs/memdb.h b/src/pmxcfs/memdb.h
index bca8967..f7dff5b 100644
--- a/src/pmxcfs/memdb.h
+++ b/src/pmxcfs/memdb.h
@@ -22,11 +22,14 @@
#define _PVE_MEMDB_H_
#include <stdio.h>
+#include <stdint.h>
#include <stdlib.h>
#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 +86,10 @@ typedef struct {
db_backend_t *bdb;
} memdb_t;
+typedef void (*memdb_notify_fn)(
+ enum pmxcfs_notify_type type, uint64_t seq, const char *path, const char *to
+);
+
memdb_t *memdb_open(const char *dbfilename);
void memdb_close(memdb_t *memdb);
@@ -138,6 +145,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, uint64_t seq, const char *path, const char *to);
+
void memdb_dump(memdb_t *memdb);
gboolean
--
2.47.3
^ permalink raw reply related [flat|nested] 11+ messages in thread
* [PATCH pve-cluster 05/10] buildsys: link pmxcfs against the rust notify staticlib
2026-09-18 14:41 [RFC cluster/manager 00/10] pmxcfs: add a change notification socket Hannes Laimer
` (3 preceding siblings ...)
2026-09-18 14:41 ` [PATCH pve-cluster 04/10] pmxcfs: memdb: add change notification hook Hannes Laimer
@ 2026-09-18 14:41 ` Hannes Laimer
2026-09-18 14:41 ` [PATCH pve-cluster 06/10] pmxcfs: notify: emit change events over the notification socket Hannes Laimer
` (4 subsequent siblings)
9 siblings, 0 replies; 11+ messages in thread
From: Hannes Laimer @ 2026-09-18 14:41 UTC (permalink / raw)
To: pve-devel
Only the daemon binary links the Rust staticlib, while the database
tools and the unit tests keep linking the plain C convenience library.
Building it from here as well keeps a standalone make in src/pmxcfs
working and picks up Rust changes without a manual step.
Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
src/pmxcfs/Makefile | 14 ++++++++++++--
1 file changed, 12 insertions(+), 2 deletions(-)
diff --git a/src/pmxcfs/Makefile b/src/pmxcfs/Makefile
index 90bc6a5..981171e 100644
--- a/src/pmxcfs/Makefile
+++ b/src/pmxcfs/Makefile
@@ -12,6 +12,10 @@ CFLAGS += -I../rust/pmxcfs-ffi/include
LDFLAGS += -Wl,-z,relro $(shell pkg-config --libs $(DEPENDENCIES))
+NOTIFY_LIB := ../rust/target/release/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
@@ -26,8 +30,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): FORCE
+ $(MAKE) -C ../rust all
+
+.PHONY: FORCE
+FORCE:
create_pmxcfs_db: create_pmxcfs_db.o libpmxcfs.a
--
2.47.3
^ permalink raw reply related [flat|nested] 11+ messages in thread
* [PATCH pve-cluster 06/10] pmxcfs: notify: emit change events over the notification socket
2026-09-18 14:41 [RFC cluster/manager 00/10] pmxcfs: add a change notification socket Hannes Laimer
` (4 preceding siblings ...)
2026-09-18 14:41 ` [PATCH pve-cluster 05/10] buildsys: link pmxcfs against the rust notify staticlib Hannes Laimer
@ 2026-09-18 14:41 ` Hannes Laimer
2026-09-18 14:41 ` [PATCH pve-cluster 07/10] cfs: add perl client for the change " Hannes Laimer
` (3 subsequent siblings)
9 siblings, 0 replies; 11+ messages in thread
From: Hannes Laimer @ 2026-09-18 14:41 UTC (permalink / raw)
To: pve-devel
Start the notification socket after the daemonize fork and before any
thread can mutate the tree, with the memdb hook installed and the
version of the last mutation so far handed over. That way no change
slips past the socket while it comes up, and a client resuming at that
version after a restart is current and needs no resync. A failure to
bring the socket up is logged and the daemon carries on.
Messages from the notifier are routed through the daemon's own logging
under a dedicated domain so they end up in the same journal stream as
everything else.
Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
src/pmxcfs/cfs-utils.h | 4 ++++
src/pmxcfs/pmxcfs.c | 25 +++++++++++++++++++++++++
2 files changed, 29 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..575ea3a 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,14 @@ int main(int argc, char *argv[]) {
cfs_loop_add_service(corosync_loop, service_status, QB_LOOP_LOW);
}
+ memdb_set_notify_hook(pmxcfs_notify_emit);
+
+ // brought up before any thread can mutate the tree, so the version it is
+ // handed is exact and no change slips past it while it comes up
+ if (pmxcfs_notify_init(NOTIFY_SOCKET, cfs.gid, notify_log_cb, memdb->root->version) != 0) {
+ cfs_critical("running without change notification socket");
+ }
+
cfs_loop_start_worker(corosync_loop);
server_start(memdb);
@@ -1053,6 +1076,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] 11+ messages in thread
* [PATCH pve-cluster 07/10] cfs: add perl client for the change notification socket
2026-09-18 14:41 [RFC cluster/manager 00/10] pmxcfs: add a change notification socket Hannes Laimer
` (5 preceding siblings ...)
2026-09-18 14:41 ` [PATCH pve-cluster 06/10] pmxcfs: notify: emit change events over the notification socket Hannes Laimer
@ 2026-09-18 14:41 ` Hannes Laimer
2026-09-18 14:41 ` [PATCH pve-cluster 08/10] cfs: add hook registry for change notification consumers Hannes Laimer
` (2 subsequent siblings)
9 siblings, 0 replies; 11+ messages in thread
From: Hannes Laimer @ 2026-09-18 14:41 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 with a deadline, and recover when pmxcfs
restarts. A shared client module handles that, so every consumer gets
the reconnect and resync handling right instead of reimplementing it.
The client resumes from the last sequence number it has, and only does
a full resync when it has none or the daemon can no longer replay from
it.
A refused handshake, an error reply or a protocol mismatch is reported
to the caller instead of retried, since a retry would not change any of
them, while a lost connection is retried with backoff. A forked child
opens its own connection instead of using the one it inherited, as the
IPC client does, since two readers on one stream would corrupt the line
framing.
Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
debian/pve-cluster.install | 1 +
src/PVE/Cluster/Makefile | 2 +-
src/PVE/Cluster/Watch.pm | 371 ++++++++++++++++++++++++++++++++++
src/test/Makefile | 6 +-
src/test/watch_client_test.pl | 294 +++++++++++++++++++++++++++
5 files changed, 672 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..a41860b
--- /dev/null
+++ b/src/PVE/Cluster/Watch.pm
@@ -0,0 +1,371 @@
+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);
+
+use PVE::Tools;
+
+my $default_socket = '/run/pve-cluster/pmxcfs.sock';
+my $protocol_version = 1;
+my $reply_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 => {},
+ sock => undef,
+ buf => '',
+ queue => [],
+ backoff => $min_backoff,
+ next_connect => 0,
+ warned => 0,
+ last_seq => undef,
+ state => $param{state},
+ key => $param{key},
+ pid => $$,
+ }, $class;
+
+ # resume point from the caller, or a previous process's checkpoint,
+ # numified since the daemon takes a JSON number and refuses a string
+ if (defined($param{since})) {
+ $self->{last_seq} = 0 + $param{since};
+ } elsif (defined($self->{state}) && -e $self->{state}) {
+ my $line = PVE::Tools::file_read_firstline($self->{state}) // '';
+ # a recorded point stands only for the key it was recorded under,
+ # anything else is as good as no state file
+ if (my ($seq, $key) = $line =~ m/^(\d+)(?:\s+(\S+))?$/) {
+ $self->{last_seq} = 0 + $seq if ($key // '') eq ($self->{key} // '');
+ }
+ }
+
+ return $self;
+}
+
+# A connection inherited across a fork belongs to the parent, so the child
+# drops its copy without a shutdown and connects on its own, as IPCC does.
+sub check_fork {
+ my ($self) = @_;
+
+ return if $self->{pid} == $$;
+
+ CORE::close($self->{sock}) if $self->{sock};
+ $self->{sock} = undef;
+ $self->{buf} = '';
+ $self->{queue} = [];
+ $self->{pid} = $$;
+ $self->{next_connect} = time();
+ $self->{backoff} = $min_backoff;
+ $self->{warned} = 0;
+ $self->{last_seq} = undef;
+ $self->{state} = undef;
+
+ return;
+}
+
+# Takes effect when connected and is replayed on every reconnect. A refusal
+# is raised, a dead connection is dropped and comes back through the next wait.
+sub subscribe {
+ my ($self, %param) = @_;
+
+ $self->check_fork();
+
+ my $previous = $self->{patterns};
+ $self->{patterns} = { ($param{patterns} // {})->%* };
+
+ return if !$self->{sock};
+
+ my $refusal = eval { $self->send_subscription() };
+ if (my $err = $@) {
+ chomp $err;
+ $self->disconnect($err) if $self->{sock};
+ return;
+ }
+ # the daemon keeps serving the previous subscription, so keep
+ # describing that one
+ if (defined($refusal)) {
+ $self->{patterns} = $previous;
+ die "subscription refused: $refusal\n";
+ }
+
+ return;
+}
+
+# The number the caller has processed up to, so a restart resumes there.
+sub checkpoint {
+ my ($self, $seq) = @_;
+
+ $self->check_fork();
+
+ return if !defined($self->{state});
+
+ if (!defined($seq)) {
+ unlink($self->{state});
+ return;
+ }
+
+ my $key = defined($self->{key}) ? " $self->{key}" : '';
+ PVE::Tools::file_set_contents($self->{state}, "$seq$key\n");
+
+ return;
+}
+
+sub connected {
+ my ($self) = @_;
+
+ $self->check_fork();
+
+ return defined($self->{sock});
+}
+
+# The descriptor changes on every reconnect and is undef while disconnected,
+# so a select-loop caller refreshes it after each wait, 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 events that arrived within $timeout seconds, or none. Reconnect
+# with handshake and reply timeout happens in here too, the one way a zero
+# timeout can block. A first connect yields a resync, and a refused
+# handshake dies, since it will not change on retry.
+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 has to get back control 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) = @_;
+
+ # a daemon that cannot accept for a while leaves a blocking connect in
+ # its backlog, so the attempt carries a timeout and a failure goes to retry
+ my $sock = IO::Socket::UNIX->new(
+ Type => SOCK_STREAM,
+ Peer => $self->{socket_path},
+ Timeout => $reply_timeout,
+ );
+ if (!$sock) {
+ $self->connect_failed("connect to $self->{socket_path} failed: $!");
+ return;
+ }
+
+ $self->{sock} = $sock;
+ $self->{buf} = '';
+ $self->{established} = 0;
+
+ my $resume = defined($self->{last_seq});
+ my ($refusal, $hello_seq) = eval { $self->handshake() };
+ my $err = $@;
+ # a number learned in a handshake that did not go through is no resume
+ # point, the first connection that does still owes the caller a resync
+ $self->{last_seq} = undef if !$resume && ($err || defined($refusal));
+ if ($err) {
+ $self->close();
+ $self->connect_failed("handshake with $self->{socket_path} failed: $err");
+ return;
+ }
+ if (defined($refusal)) {
+ $self->close();
+ $self->delay_retry();
+ die "handshake with $self->{socket_path} refused: $refusal\n";
+ }
+
+ $self->{established} = 1;
+ $self->{backoff} = $min_backoff;
+ $self->{warned} = 0;
+ # the resync carries the number the daemon handed out, so a
+ # consumer can record it once the resync is processed
+ unshift $self->{queue}->@*,
+ { type => 'resync', (defined($hello_seq) ? (seq => 0 + $hello_seq) : ()) }
+ if !$resume;
+
+ 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 ($version // -1) != $protocol_version;
+
+ my $refusal = $self->send_subscription();
+
+ # a client that saw no event yet still learns where the daemon stands, so
+ # a later reconnect resumes there instead of resyncing. The subscription
+ # goes out first, so a first connect still owes and gets its resync
+ $self->{last_seq} //= 0 + $hello->{seq} if defined($hello->{seq});
+
+ return ($refusal, $hello->{seq});
+}
+
+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 the connect attempt's
+# failure to report, so a daemon that accepts and closes does not warn twice.
+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 $args = { patterns => $self->{patterns} };
+ $args->{since} = $self->{last_seq} if defined($self->{last_seq});
+ my (undef, $error) = $self->request('subscribe', $args);
+
+ return $error;
+}
+
+# Sends a request and returns its reply as (data, error), queueing any events
+# that arrive meanwhile. Dies when no reply comes at all, meaning the
+# connection is gone.
+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() + $reply_timeout;
+ while (1) {
+ my $remaining = $deadline - time();
+ die "timeout waiting for reply to '$command'\n" if $remaining <= 0;
+ 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 {
+ $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;
+ $self->{last_seq} = $event->{seq} if defined($event->{seq});
+ } 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..3ad3aa5
--- /dev/null
+++ b/src/test/watch_client_test.pl
@@ -0,0 +1,294 @@
+#!/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 fourteen connections in a row. The
+# fourth refuses the subscription, the ninth hangs up after answering the
+# hello, the eleventh refuses a second subscription on a live connection
+# and the last three hang up before answering anything. The others answer
+# the handshake, mirror the subscription back inside two events and hang
+# up, which forces the client through its reconnect path. Like the daemon
+# it takes a resume point only as a number.
+my $server = fork() // die "fork failed: $!\n";
+if (!$server) {
+ for my $round (1 .. 14) {
+ my $conn = $listener->accept() or die "accept failed: $!\n";
+ if ($round >= 12) {
+ 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, seq => 3 } }), "\n";
+ } elsif ($round == 9) {
+ last;
+ } elsif ($req->{command} eq 'subscribe') {
+ if ($round == 4) {
+ print $conn encode_json({ error => "pattern 'guest': no good" }), "\n";
+ last;
+ }
+ if ($round == 11) {
+ if (!$subscribed++) {
+ print $conn encode_json({ ok => undef }), "\n";
+ next;
+ }
+ print $conn encode_json({ error => "pattern 'bad': no good" }), "\n";
+ last;
+ }
+ my $patterns = $req->{args}->{patterns};
+ my $since = $req->{args}->{since};
+ if (defined($since) && encode_json([$since]) !~ /^\[\d+\]$/) {
+ print $conn encode_json({ error => "'since' must be an unsigned integer" }),
+ "\n";
+ last;
+ }
+ print $conn encode_json({ ok => undef }), "\n";
+ print $conn encode_json({
+ event => {
+ seq => 1,
+ type => 'write',
+ path => 'x',
+ params =>
+ { map { $_ => [{ template => $patterns->{$_} }] } keys %$patterns },
+ },
+ }),
+ "\n";
+ print $conn encode_json({
+ event => {
+ seq => 2,
+ type => 'rename',
+ path => 'old',
+ to => join(',', sort keys %$patterns),
+ (defined($since) ? (since => $since) : ()),
+ },
+ }),
+ "\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 = {
+ seq => 1,
+ type => 'write',
+ path => 'x',
+ params => { guest => [{ template => $template }] },
+};
+my $rename = { seq => 2, type => 'rename', path => 'old', to => 'guest' };
+# a fresh client has nothing to resume from, so it subscribes without one
+my $expected = [{ type => 'resync', seq => 3 }, $write, $rename];
+my $resumed = [$write, { $rename->%*, since => 2 }];
+
+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, 2, 1),
+ $resumed,
+ 'reconnect after hangup resumes after the last event and replays the subscription',
+);
+
+# the server has hung up on that connection by now, writing to it must
+# neither raise SIGPIPE nor die, the next wait reconnects
+sleep(0.2);
+$watch->subscribe(patterns => { guest => $template });
+ok(!$watch->connected(), 'a subscription on a dead connection drops it');
+
+eval { $watch->wait(1) };
+like($@, qr/refused: pattern 'guest': no good/, 'a refused subscription is raised, not retried');
+ok(!$watch->connected(), 'and its connection is closed');
+
+# a client resuming from a state file, served by rounds five and six
+my $state = "$dir/state";
+open(my $fh, '>', $state) or die "open: $!";
+print $fh "5\n";
+close($fh);
+my $resuming = PVE::Cluster::Watch->new(socket => $path, state => $state);
+$resuming->subscribe(patterns => { guest => $template });
+is_deeply(
+ $collect->($resuming, 2, 1),
+ [$write, { $rename->%*, since => 5 }],
+ 'a state file provides the resume point and replaces the first resync',
+);
+is_deeply(
+ $collect->($resuming, 2, 1),
+ [$write, { $rename->%*, since => 2 }],
+ 'from then on the last event received is the resume point',
+);
+$resuming->checkpoint(2);
+open($fh, '<', $state) or die "open: $!";
+my $stored = <$fh>;
+close($fh);
+is($stored, "2\n", 'checkpoint writes the state file');
+is(PVE::Cluster::Watch->new(since => 7)->{last_seq}, 7, 'an explicit resume point is taken as is');
+ok(eval { $watch->checkpoint(1); 1 }, 'checkpoint without a state file does nothing');
+
+# a resume point bound to a key, served by rounds seven and eight
+my $key = 'a1b2c3';
+open($fh, '>', $state) or die "open: $!";
+print $fh "5 $key\n";
+close($fh);
+my $keyed = PVE::Cluster::Watch->new(socket => $path, state => $state, key => $key);
+$keyed->subscribe(patterns => { guest => $template });
+is_deeply(
+ $collect->($keyed, 2, 1),
+ [$write, { $rename->%*, since => 5 }],
+ 'a state file written under the same key provides the resume point',
+);
+$keyed->checkpoint(2);
+open($fh, '<', $state) or die "open: $!";
+$stored = <$fh>;
+close($fh);
+is($stored, "2 $key\n", 'which a checkpoint records next to the number');
+
+open($fh, '>', $state) or die "open: $!";
+print $fh "5 d4e5f6\n";
+close($fh);
+my $rekeyed = PVE::Cluster::Watch->new(socket => $path, state => $state, key => $key);
+$rekeyed->subscribe(patterns => { guest => $template });
+is_deeply(
+ $collect->($rekeyed, 3, 1),
+ $expected,
+ 'a state file written under another key is no resume point',
+);
+
+# a handshake cut after the hello, rounds nine and ten. The number the
+# first attempt learned must not pass as a resume point on the second.
+{
+ 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 first connect that is cut after the hello still gives a resync',
+ );
+ is(scalar(@warnings), 1, 'and the cut handshake is reported');
+}
+
+# a refused subscription on a live connection, round eleven. The daemon keeps
+# serving the previous one, so the client keeps describing that one.
+{
+ my $live = PVE::Cluster::Watch->new(socket => $path);
+ $live->subscribe(patterns => { guest => $template });
+ is_deeply(
+ $collect->($live, 1, 1),
+ [{ type => 'resync', seq => 3 }],
+ 'connected with the first subscription',
+ );
+ eval { $live->subscribe(patterns => { bad => 'x/{y}' }) };
+ like($@, qr/subscription refused: pattern 'bad': no good/, 'a refusal is raised');
+ 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 twelve to fourteen
+{
+ 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, here while the client is waiting for its
+# next connect attempt. A caller whose handler only raises a flag has to get
+# back control to act on it, rather than at the end of the wait.
+{
+ 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 while the socket is gone');
+ok(!$watch->connected(), 'disconnected after the server went away');
+
+done_testing();
--
2.47.3
^ permalink raw reply related [flat|nested] 11+ messages in thread
* [PATCH pve-cluster 08/10] cfs: add hook registry for change notification consumers
2026-09-18 14:41 [RFC cluster/manager 00/10] pmxcfs: add a change notification socket Hannes Laimer
` (6 preceding siblings ...)
2026-09-18 14:41 ` [PATCH pve-cluster 07/10] cfs: add perl client for the change " Hannes Laimer
@ 2026-09-18 14:41 ` Hannes Laimer
2026-09-18 14:41 ` [PATCH pve-manager 09/10] hooks: add runner executing cluster change hooks in children Hannes Laimer
2026-09-18 14:41 ` [PATCH pve-manager 10/10] pvescheduler: run cluster change hooks from a listener child Hannes Laimer
9 siblings, 0 replies; 11+ messages in thread
From: Hannes Laimer @ 2026-09-18 14:41 UTC (permalink / raw)
To: pve-devel
Reacting to a cluster config change should look like handling an API
request, a path template with placeholders that turn into parameters
and a code reference that receives them.
The templates go to pmxcfs as they are, which matches them and
delivers the captured values, so the registry only turns those into
runs after checking the event type. It refuses at registration what
the daemon would refuse at subscribe time, so a bad template fails
where the hook is written rather than in the listener. Each run
carries a key made of the hook name and its parameters, so an executor
can keep one run per key in flight and collapse the burst that
repeated saves of the same file produce, a config update through the
API being several of them. A resync becomes a parameterless run of
every hook.
Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
debian/pve-cluster.install | 1 +
src/PVE/Cluster/Hooks.pm | 134 +++++++++++++++++++++++
src/PVE/Cluster/Makefile | 2 +-
src/test/Makefile | 6 +-
src/test/hooks_test.pl | 219 +++++++++++++++++++++++++++++++++++++
5 files changed, 360 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..e6c6001
--- /dev/null
+++ b/src/PVE/Cluster/Hooks.pm
@@ -0,0 +1,134 @@
+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, so a burst collapses into one rerun. An event only says the
+# state changed, so a run reads it as it is then, which is what keeps
+# collapsing safe. Runs are not serialized, so a hook writing shared
+# state takes the same locks an API handler would.
+
+my $hooks = {};
+my $order = [];
+
+my $default_types = [qw(create write rename delete mkdir)];
+my $known_types = { map { $_ => 1 } qw(create write mtime rename delete mkdir) };
+
+# what pmxcfs accepts in one subscription
+my $max_hooks = 256;
+
+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';
+ die "hook '$name': at most $max_hooks hooks can be registered\n"
+ if scalar($order->@*) >= $max_hooks;
+
+ my $template = compile_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;
+}
+
+# Fails a bad path at registration instead of leaving it for the daemon.
+sub compile_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:" . JSON->new->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 @runs;
+
+ if ($event->{type} eq 'resync') {
+ for my $name ($order->@*) {
+ push @runs, { hook => $hooks->{$name}, param => {}, key => $name, event => $event };
+ }
+ 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 => $event };
+ }
+ }
+
+ 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..8adceda
--- /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...}',
+ },
+ },
+ 'subscription sends 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] 11+ messages in thread
* [PATCH pve-manager 09/10] hooks: add runner executing cluster change hooks in children
2026-09-18 14:41 [RFC cluster/manager 00/10] pmxcfs: add a change notification socket Hannes Laimer
` (7 preceding siblings ...)
2026-09-18 14:41 ` [PATCH pve-cluster 08/10] cfs: add hook registry for change notification consumers Hannes Laimer
@ 2026-09-18 14:41 ` Hannes Laimer
2026-09-18 14:41 ` [PATCH pve-manager 10/10] pvescheduler: run cluster change hooks from a listener child Hannes Laimer
9 siblings, 0 replies; 11+ messages in thread
From: Hannes Laimer @ 2026-09-18 14:41 UTC (permalink / raw)
To: pve-devel
The hook registry in pve-cluster only turns events into runs. Add the
piece that executes them on a node, holding one notification socket
connection and forking a child per run so a slow or failing hook never
touches the connection or the other hooks.
One run per hook and parameter set is in flight at a time, events for
the same set that arrive meanwhile collapse into a single rerun after it
finishes, and the number of children is capped. A failed fork puts the
run back for the next round instead of retrying immediately, which
would spin without ever returning to the socket.
The runner records how far it has got, the highest sequence number with
no run still outstanding below it, and which hooks were active when it
did. A runner that takes over after a reload resumes from there and
reruns only what was interrupted. It resyncs instead when the set of
hooks changed, so a newly added hook runs on the current state, not
just on the next change. A run only counts towards that position once
it has been reaped while the runner was still up, whatever its exit
code, since failures are not retried, and while a resync is in flight
the position is dropped, so the next runner resyncs as well.
A run that runs too long stops counting against the worker limit, so the
other keys keep running, and the next event for its key kills it and
starts over. The total number of runs is capped even so, and a shutdown
that cannot kill a run gives up after a grace period rather than hang,
leaving that run behind.
Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
PVE/HookRunner.pm | 368 +++++++++++++++++++++++++++++++++++
PVE/Makefile | 1 +
test/Makefile | 6 +-
test/hook_runner_test.pl | 405 +++++++++++++++++++++++++++++++++++++++
4 files changed, 779 insertions(+), 1 deletion(-)
create mode 100644 PVE/HookRunner.pm
create mode 100755 test/hook_runner_test.pl
diff --git a/PVE/HookRunner.pm b/PVE/HookRunner.pm
new file mode 100644
index 00000000..7f7b8a58
--- /dev/null
+++ b/PVE/HookRunner.pm
@@ -0,0 +1,368 @@
+package PVE::HookRunner;
+
+use strict;
+use warnings;
+
+use Digest::SHA qw(sha1_hex);
+use File::Basename qw(dirname);
+use File::Path qw(make_path);
+use JSON;
+use List::Util qw(min);
+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, recording how far it got under /run so
+# a replacement runner resumes there instead of resyncing.
+
+my $stop_grace = 5;
+
+sub subscription_key {
+ my ($hooks) = @_;
+
+ my $digest = [
+ map { {
+ name => $_->{name},
+ template => $_->{template},
+ types => [sort keys $_->{types}->%*],
+ } } sort { $a->{name} cmp $b->{name} } $hooks->@*
+ ];
+
+ return sha1_hex(JSON->new->canonical->encode($digest));
+}
+
+sub new {
+ my ($class, %param) = @_;
+
+ my $self = bless {
+ watch => $param{watch} // PVE::Cluster::Watch->new(
+ state => $param{state},
+ key => subscription_key(PVE::Cluster::Hooks->hooks()),
+ ),
+ state => $param{state},
+ max_workers => $param{max_workers} // 4,
+ stall_age => $param{stall_age} // 120,
+ running => {},
+ by_key => {},
+ pending => {},
+ queue => [],
+ queued => {},
+ outstanding => {},
+ last_seq => undef,
+ mark => undef,
+ withdrawn => 0,
+ stopping => 0,
+ 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 it would sleep through to the next tick
+ local $SIG{CHLD} = sub { };
+
+ # the daemon hosting this has no stderr, so what the client warns about
+ # its connection has to reach the journal
+ local $SIG{__WARN__} = sub {
+ my ($msg) = @_;
+ chomp $msg;
+ syslog('warning', $msg);
+ };
+
+ make_path(dirname($self->{state})) if defined($self->{state});
+
+ $self->{watch}->subscribe(PVE::Cluster::Hooks->subscription()->%*);
+
+ # a refused handshake comes back from the wait, the children are stopped
+ # before it propagates so a restart does not leave them behind
+ eval {
+ # a long idle wait is safe since a signal or a child's exit cuts it
+ # short, runs in flight get the short one so stalls are 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)) {
+ my $runs = PVE::Cluster::Hooks->dispatch($event);
+ $self->track($event, $runs);
+ $self->schedule($_) for $runs->@*;
+ }
+
+ $self->reap();
+ $self->check_stalls();
+ # the recorded position is all that survives an unclean death, so it
+ # is settled before the runs start rather than after
+ $self->advance_mark();
+ $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 track {
+ my ($self, $event, $runs) = @_;
+
+ my $seq = $event->{seq};
+ $self->{last_seq} = $seq if defined($seq);
+
+ for my $run ($runs->@*) {
+ $run->{seqs} = defined($seq) ? [$seq] : [];
+ $self->{outstanding}->{$seq}++ if defined($seq);
+ $run->{resync} = 1 if $event->{type} eq 'resync';
+ }
+
+ return;
+}
+
+sub reconciling {
+ my ($self) = @_;
+
+ return scalar(grep { $_->{resync} } (
+ values $self->{running}->%*, values $self->{pending}->%*,
+ values $self->{queued}->%*,
+ ));
+}
+
+sub schedule {
+ my ($self, $run) = @_;
+
+ my $key = $run->{key};
+ $run->{seqs} //= [];
+
+ # a run that replaces a waiting one stands for that one's events too
+ if (my $pid = $self->{by_key}->{$key}) {
+ if (my $waiting = $self->{pending}->{$key}) {
+ push $run->{seqs}->@*, $waiting->{seqs}->@*;
+ $run->{resync} ||= $waiting->{resync};
+ }
+ $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 (my $waiting = $self->{queued}->{$key}) {
+ push $run->{seqs}->@*, $waiting->{seqs}->@*;
+ $run->{resync} ||= $waiting->{resync};
+ $self->{queued}->{$key} = $run;
+ } else {
+ push $self->{queue}->@*, $key;
+ $self->{queued}->{$key} = $run;
+ }
+
+ return;
+}
+
+# Stalled runs do not count against the cap, but everything running
+# together is bounded at twice the cap.
+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;
+ last if !$self->fork_run($run);
+ }
+
+ 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");
+ }
+ # a non-zero exit was already logged by the child, so only an
+ # unexpected signal death, which the child could not report, is logged
+ if ($res == $pid && ($? & 127) && !$run->{killed_at}) {
+ syslog('err', "hook run '$run->{key}' killed by signal " . ($? & 127));
+ }
+
+ $self->complete($run) if !$self->{stopping};
+
+ if (my $again = delete $self->{pending}->{ $run->{key} }) {
+ $self->schedule($again);
+ }
+ }
+
+ return;
+}
+
+sub complete {
+ my ($self, $run) = @_;
+
+ for my $seq ($run->{seqs}->@*) {
+ delete $self->{outstanding}->{$seq} if --$self->{outstanding}->{$seq} <= 0;
+ }
+
+ return;
+}
+
+sub advance_mark {
+ my ($self) = @_;
+
+ if ($self->reconciling()) {
+ return if $self->{withdrawn};
+ $self->{watch}->checkpoint(undef);
+ $self->{withdrawn} = 1;
+ $self->{mark} = undef;
+ return;
+ }
+
+ return if !defined($self->{last_seq});
+
+ my $oldest = min(keys $self->{outstanding}->%*);
+ my $mark = defined($oldest) ? $oldest - 1 : $self->{last_seq};
+ return if defined($self->{mark}) && $mark <= $self->{mark};
+
+ $self->{mark} = $mark;
+ $self->{withdrawn} = 0;
+ $self->{watch}->checkpoint($mark);
+
+ return;
+}
+
+# Returns whether the run was started. A failed fork puts it back at the
+# head of the queue for the next round rather than retrying immediately.
+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}++;
+ unshift $self->{queue}->@*, $run->{key};
+ $self->{queued}->{ $run->{key} } = $run;
+ 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 {
+ 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 stop_children {
+ my ($self) = @_;
+
+ $self->{stopping} = 1;
+
+ my @pids = keys $self->{running}->%*;
+ return if !scalar(@pids);
+
+ my $now = time();
+ $_->{killed_at} = $now for values $self->{running}->%*;
+ kill 'TERM', @pids;
+
+ my $deadline = time() + $stop_grace;
+ while (scalar(keys $self->{running}->%*) && time() < $deadline) {
+ sleep(0.1);
+ $self->reap();
+ }
+
+ 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
+ $deadline = time() + $stop_grace;
+ while (scalar(keys $self->{running}->%*) && time() < $deadline) {
+ sleep(0.1);
+ $self->reap();
+ }
+
+ 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/PVE/Makefile b/PVE/Makefile
index efcb250d..05c211ce 100644
--- a/PVE/Makefile
+++ b/PVE/Makefile
@@ -11,6 +11,7 @@ PERLSOURCE = \
CertHelpers.pm \
ExtMetric.pm \
HTTPServer.pm \
+ HookRunner.pm \
Jobs.pm \
NodeConfig.pm \
PullMetric.pm \
diff --git a/test/Makefile b/test/Makefile
index 026af9cc..f30f6abb 100644
--- a/test/Makefile
+++ b/test/Makefile
@@ -5,7 +5,7 @@ all:
export PERLLIB=..
.PHONY: check
-check: test-replication test-balloon test-vzdump test-osd test-ceph-health test-ceph-auth test-ceph-key-migration test-ceph-lockbox-migration test-custom-cpu-models test-pvesh
+check: test-replication test-balloon test-vzdump test-osd test-ceph-health test-ceph-auth test-ceph-key-migration test-ceph-lockbox-migration test-custom-cpu-models test-pvesh test-hook-runner
.PHONY: test-balloon
test-balloon:
@@ -57,6 +57,10 @@ test-custom-cpu-models:
test-pvesh:
./pvesh_test.pl
+.PHONY: test-hook-runner
+test-hook-runner:
+ ./hook_runner_test.pl
+
.PHONY: install
install:
diff --git a/test/hook_runner_test.pl b/test/hook_runner_test.pl
new file mode 100755
index 00000000..79a81390
--- /dev/null
+++ b/test/hook_runner_test.pl
@@ -0,0 +1,405 @@
+#!/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::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, cfs_update
+# is stubbed, the forks and the bookkeeping around them are real.
+my $cluster = Test::MockModule->new('PVE::Cluster');
+$cluster->redefine(cfs_update => sub { });
+
+# nothing here belongs in the host's journal
+my @logged;
+my $journal = Test::MockModule->new('PVE::HookRunner');
+$journal->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;
+ }
+
+ sub checkpoint {
+ my ($self, $seq) = @_;
+ push $self->{checkpoints}->@*, $seq;
+ 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 its first event, 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->{seq} // 0) == 50;
+ $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, $seq) = @_;
+ return {
+ type => $type,
+ path => $path,
+ ($to ? (to => $to) : ()),
+ ($params ? (params => $params) : ()),
+ (defined($seq) ? (seq => $seq) : ()),
+ };
+};
+
+my $guest = sub {
+ my ($vmid) = @_;
+ return { guest => [{ node => 'n1', vmid => $vmid }] };
+};
+
+# two saves of one guest in a row, the second arriving 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::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::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::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::HookRunner->new(watch => $watch, max_workers => 1);
+$runner->{terminate} = 1;
+$runner->run();
+is_deeply(
+ $watch->{subscription},
+ {
+ patterns => {
+ broken => 'broken.cfg',
+ dc => 'datacenter.cfg',
+ guest => 'nodes/{node}/qemu-server/{vmid}.conf',
+ slow => 'slow/{id}.conf',
+ },
+ },
+ 'subscription is derived from the registered hooks',
+);
+
+# the client the runner builds for itself gets the position it keeps and
+# a key for the hooks that position stands for
+{
+ my $params;
+ my $fake = FakeWatch->new();
+ my $client = Test::MockModule->new('PVE::Cluster::Watch');
+ $client->redefine(
+ new => sub {
+ my ($class, %param) = @_;
+ $params = \%param;
+ return $fake;
+ },
+ );
+ my $own = PVE::HookRunner->new(state => "$dir/hooks.seq", max_workers => 1);
+ $own->{terminate} = 1;
+ $own->run();
+ is($params->{state}, "$dir/hooks.seq", 'the state path reaches the client');
+ is(
+ $params->{key},
+ PVE::HookRunner::subscription_key(PVE::Cluster::Hooks->hooks()),
+ 'and a key derived from the hooks that position stands for',
+ );
+}
+
+# a hook that keeps its path but reacts to other events is a different
+# set of hooks and must not resume
+{
+ my $registry = sub {
+ my (@types) = @_;
+ return [{
+ name => 'dc',
+ template => 'datacenter.cfg',
+ types => { map { $_ => 1 } @types },
+ }];
+ };
+ isnt(
+ PVE::HookRunner::subscription_key($registry->('write')),
+ PVE::HookRunner::subscription_key($registry->('write', 'delete')),
+ 'a changed set of event types changes the key',
+ );
+}
+
+# a watch that gives up takes the children down with it
+$watch = FakeWatch->new([$ev->('write', 'datacenter.cfg', undef, { dc => [{}] })]);
+$watch->{fail} = "handshake refused\n";
+$runner = PVE::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->();
+
+# the checkpoint trails the runs still in flight and skips events that
+# produced none
+$watch = FakeWatch->new([
+ $ev->('write', 'nodes/n1/qemu-server/100.conf', undef, $guest->(100), 10),
+ $ev->('write', 'nodes/n1/qemu-server/101.conf', undef, $guest->(101), 11),
+ $ev->('write', 'unrelated.cfg', undef, undef, 12),
+]);
+$runner = PVE::HookRunner->new(watch => $watch, max_workers => 4);
+$runner->run_once(0.05);
+is_deeply($watch->{checkpoints}, [9], 'the checkpoint stops short of the runs in flight');
+$drain->($runner);
+# the two runs may be reaped in one round or in two, so the mark may or
+# may not pass through the first of them on its way
+my $marks = $watch->{checkpoints};
+is($marks->[-1], 12, 'and moves past them once they completed');
+ok(!(grep { $marks->[$_] < $marks->[$_ - 1] } 1 .. $#$marks), 'without ever moving back');
+$lines->();
+
+# a rerun stands for the event it absorbed
+$watch = FakeWatch->new(
+ [$ev->('write', 'nodes/n1/qemu-server/100.conf', undef, $guest->(100), 20)],
+ [$ev->('write', 'nodes/n1/qemu-server/100.conf', undef, $guest->(100), 21)],
+);
+$runner = PVE::HookRunner->new(watch => $watch, max_workers => 4);
+$runner->run_once(0.05);
+$runner->run_once(0.05);
+is_deeply($watch->{checkpoints}, [19], 'a pending rerun keeps its event outstanding');
+$drain->($runner);
+is_deeply($watch->{checkpoints}, [19, 20, 21], 'the rerun completes the event it absorbed');
+$lines->();
+
+# a stop leaves the interrupted run outstanding
+@logged = ();
+$watch =
+ FakeWatch->new([$ev->('write', 'nodes/n1/qemu-server/100.conf', undef, $guest->(100), 30)]);
+$runner = PVE::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');
+is_deeply($watch->{checkpoints}, [29], 'and stays outstanding for the next runner');
+ok(!(grep { $_->[0] eq 'err' } @logged), 'a run interrupted by the stop is not an error');
+$lines->();
+
+# what the client warns about reaches the journal, the daemon has no stderr
+@logged = ();
+$watch = FakeWatch->new();
+$watch->{warn} = "notification socket closed, reconnecting\n";
+$watch->{fail} = "stop\n";
+$runner = PVE::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 client is logged',
+);
+
+# a resync in flight withdraws the checkpoint, the next runner must start
+# with a resync of its own
+$watch = FakeWatch->new([$ev->('write', 'datacenter.cfg', undef, { dc => [{}] }, 40)]);
+$runner = PVE::HookRunner->new(watch => $watch, max_workers => 4);
+$runner->run_once(0.05);
+$drain->($runner);
+is_deeply($watch->{checkpoints}, [39, 40], 'a plain run moves the mark as before');
+push $watch->{batches}->@*, [$ev->('resync', undef, undef, undef, 41)];
+$runner->run_once(0.05);
+is_deeply($watch->{checkpoints}, [39, 40, undef], 'a resync in flight withdraws it');
+$drain->($runner);
+is_deeply($watch->{checkpoints}, [39, 40, undef, 41], 'and it returns once the resync completed');
+$lines->();
+
+# a stalled run frees its slot, is logged, and dies to the next event for
+# its key while the other keys keep running
+@logged = ();
+my $slow = sub {
+ my ($seq) = @_;
+ return $ev->('write', 'slow/1.conf', undef, { slow => [{ id => 1 }] }, $seq);
+};
+$watch = FakeWatch->new(
+ [$slow->(50)],
+ [$ev->('write', 'datacenter.cfg', undef, { dc => [{}] }, 51)],
+ [$slow->(52)],
+);
+$runner = PVE::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 we killed is not an error');
+
+# a child in uninterruptible sleep outlives the kill, a clock past every
+# deadline puts the stop in the same spot without the wait
+@logged = ();
+$watch =
+ FakeWatch->new([$ev->('write', 'nodes/n1/qemu-server/100.conf', undef, $guest->(100), 60)]);
+$runner = PVE::HookRunner->new(watch => $watch, max_workers => 4);
+$runner->run_once(0.05);
+my ($stuck) = keys $runner->{running}->%*;
+my $clock = time();
+$journal->redefine(time => sub () { $clock += 10 });
+$runner->stop_children();
+$journal->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] 11+ messages in thread
* [PATCH pve-manager 10/10] pvescheduler: run cluster change hooks from a listener child
2026-09-18 14:41 [RFC cluster/manager 00/10] pmxcfs: add a change notification socket Hannes Laimer
` (8 preceding siblings ...)
2026-09-18 14:41 ` [PATCH pve-manager 09/10] hooks: add runner executing cluster change hooks in children Hannes Laimer
@ 2026-09-18 14:41 ` Hannes Laimer
9 siblings, 0 replies; 11+ messages in thread
From: Hannes Laimer @ 2026-09-18 14:41 UTC (permalink / raw)
To: pve-devel
On HUP the listener is stopped rather than handed over, so the reloaded
daemon runs the newly loaded hook modules. It is recorded under a
separate marker, which frees the hook slot for its replacement to start
immediately and still has the new daemon reap it, since reaping now
covers every recorded job type rather than a fixed list.
The listener keeps its position under /run, so the one started after a
reload resumes where the stopped one left off. Around a reload the
stopped listener may still be ending its runs while the new one starts,
and both may hold a connection and record a position for that moment,
which is safe since a recorded position only ever stands for completed
runs.
Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
PVE/Service/pvescheduler.pm | 26 ++++++++++++++++++++++----
1 file changed, 22 insertions(+), 4 deletions(-)
diff --git a/PVE/Service/pvescheduler.pm b/PVE/Service/pvescheduler.pm
index f7271fc0..7c773829 100755
--- a/PVE/Service/pvescheduler.pm
+++ b/PVE/Service/pvescheduler.pm
@@ -5,6 +5,8 @@ use warnings;
use POSIX qw(WNOHANG);
+use PVE::Cluster::Hooks;
+use PVE::HookRunner;
use PVE::Jobs;
use PVE::SafeSyslog;
@@ -17,8 +19,6 @@ my $cmdline = [$0, @ARGV];
my %daemon_options = (stop_wait_time => 180, max_workers => 0);
my $daemon = __PACKAGE__->new('pvescheduler', $cmdline, %daemon_options);
-my @JOB_TYPES = qw(replication jobs);
-
my sub running_job_pids : prototype($) {
my ($self) = @_;
my $pids = [map { keys $_->%* } values $self->{jobs}->%*];
@@ -27,7 +27,7 @@ my sub running_job_pids : prototype($) {
my sub finish_jobs : prototype($) {
my ($self) = @_;
- for my $type (@JOB_TYPES) {
+ for my $type (keys $self->{jobs}->%*) {
for my $cpid (keys $self->{jobs}->{$type}->%*) {
if (my $waitpid = waitpid($cpid, WNOHANG)) {
delete $self->{jobs}->{$type}->{$cpid} if $waitpid == $cpid || $waitpid == -1;
@@ -40,8 +40,14 @@ sub hup {
my ($self) = @_;
my $old_workers = "";
- for my $type (@JOB_TYPES) {
+ for my $type (sort keys $self->{jobs}->%*) {
my $worker = $self->{jobs}->{$type} // next;
+ # stopped rather than handed over, so the reload runs the new hook code
+ if ($type eq 'hooks') {
+ kill 'TERM', keys $worker->%*;
+ $old_workers .= "stopped:$_;" for keys $worker->%*;
+ next;
+ }
$old_workers .= "$type:$_;" for keys $worker->%*;
}
$ENV{"PVE_DAEMON_WORKER_PIDS"} = $old_workers;
@@ -118,6 +124,18 @@ sub run {
},
);
+ # only run while a hook is registered, unlike the other job types
+ if (scalar(PVE::Cluster::Hooks->hooks()->@*)) {
+ $fork->(
+ 'hooks',
+ sub {
+ PVE::HookRunner->new(
+ state => '/run/pvescheduler/hooks.seq',
+ )->run();
+ },
+ );
+ }
+
$first_run = 0;
};
--
2.47.3
^ permalink raw reply related [flat|nested] 11+ messages in thread
end of thread, other threads:[~2026-09-18 15:08 UTC | newest]
Thread overview: 11+ messages (download: mbox.gz follow: Atom feed
-- links below jump to the message on this page --
2026-09-18 14:41 [RFC cluster/manager 00/10] pmxcfs: add a change notification socket Hannes Laimer
2026-09-18 14:41 ` [PATCH pve-cluster 01/10] buildsys: add rust workspace under src/rust Hannes Laimer
2026-09-18 14:41 ` [PATCH pve-cluster 02/10] rust: notify: add change notification socket server Hannes Laimer
2026-09-18 14:41 ` [PATCH pve-cluster 03/10] rust: ffi: add C ABI staticlib for pmxcfs Hannes Laimer
2026-09-18 14:41 ` [PATCH pve-cluster 04/10] pmxcfs: memdb: add change notification hook Hannes Laimer
2026-09-18 14:41 ` [PATCH pve-cluster 05/10] buildsys: link pmxcfs against the rust notify staticlib Hannes Laimer
2026-09-18 14:41 ` [PATCH pve-cluster 06/10] pmxcfs: notify: emit change events over the notification socket Hannes Laimer
2026-09-18 14:41 ` [PATCH pve-cluster 07/10] cfs: add perl client for the change " Hannes Laimer
2026-09-18 14:41 ` [PATCH pve-cluster 08/10] cfs: add hook registry for change notification consumers Hannes Laimer
2026-09-18 14:41 ` [PATCH pve-manager 09/10] hooks: add runner executing cluster change hooks in children Hannes Laimer
2026-09-18 14:41 ` [PATCH pve-manager 10/10] pvescheduler: run cluster change hooks from a listener child Hannes Laimer
This is a public inbox, see mirroring instructions
for how to clone and mirror all data and code used for this inbox