Re: [PATCH pve-cluster 03/10] rust: ffi: add C ABI staticlib for pmxcfs

[Wolfgang Bumiller <w.bumiller@proxmox.com>]

On Fri, Sep 18, 2026 at 04:41:45PM +0200, Hannes Laimer wrote:
> 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;

^ we could make EventKind `#[repr(C)]` and `#[non_exhaustive]`, that way
`pmxcfs_notify_emit` should be able to use it directly as an ffi type
instead of `kind: c_int` → `kind: EventKind`.

We'd still need to check it for validity on the rust side, though, so
we'd keep the match, but we'd drop one additional list of the same types
we already have twice.

> +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 {

Add `+ UnwindSafe` to the `impl` type and…

> +    if DISABLED.load(Ordering::SeqCst) {
> +        return fallback;
> +    }
> +    match catch_unwind(AssertUnwindSafe(f)) {

…and drop the `AssertUnwindSafe` here. The wrapper doesn't *make* `f`
unwind safe, it says "I don't care if it is unwind-safe".


> +        Ok(result) => result,
> +        Err(payload) => {
> +            DISABLED.store(true, Ordering::SeqCst);

`DISABLED` is a simple atomic, so use Ordering::Relaxed
It is a flag independent of other values (neither setting nor loading it
requires any other values to be visible or synced).

> +            let reason = payload
> +                .downcast_ref::<&str>()
> +                .map(|s| s.to_string())
> +                .or_else(|| payload.downcast_ref::<String>().cloned())

^ as Robert noted, the map & or_else can actually simply be dropped
here, we don't need these.

> +                .unwrap_or_default();

^ And maybe replace this with `.unwrap_or("<unexpected panic value>")`.

> +            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.

They won't be able to reconnect, though.
Not sure if having pmxcfs restart is better…
At least after this was tested successfully for a while.

> +            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())

A lossy-converted string does not represent the path.

We have 2 options: Either we ignore such paths here and if users create
them they just won't be able to watch them. (*WE* don't create paths
which fail here).

Or we encode them both here and in the client protocol where we add the
watches. (Iow. use encode an `OsString` (yes, that means
the one inside the json would be double-encoded).

I'm leaning towards dropping invalid paths. We don't have any. Users
don't need to create-and-poll them.

> +}
> +
> +/// # 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(

If you use `Path::new` here you produce fewer clones.

> +            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(),

Just the one here, via path.to_owned().

> +            gid,
> +            ring: RING_SIZE,
> +            seq,
> +        }) {
> +            Ok(started) => {
> +                *server = Some(started);
> +                log::info!("listening on {}", path.display());

While this one uses the original reference.

> +                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) => {}

Here we should probably also warn if `to` is `Some`, as that's quite
unexpected?

If we go with dropping invalid path, FYI, I'd emit a warning for it here, FYI.

> +        }
> +    })
> +}
> +
> +#[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
> 
> 
> 
> 
> 

--