[PATCH pve-cluster 06/12] rust: ffi: add C ABI staticlib for pmxcfs

[Hannes Laimer <h.laimer@proxmox.com>]

pmxcfs is a C daemon, so the notification socket gets a C interface
in a static library the daemon can link. Three functions start the
socket, emit an event and shut the socket down. Log messages go with
a syslog priority to a callback from the caller, so they can end up
in the daemon's own log. The event type crosses the boundary as the
Rust enum itself, laid out as a C enum, so nothing translates between
the two sides.

Rust aborts the process when a panic reaches the C caller. In pmxcfs
that would take /etc/pve down because of a bug in the notification
socket. So every entry point catches a panic. A panic while starting
or emitting disables notifications for the rest of the daemon's life
and shuts the socket down. Clients then see EOF and the socket file
gone, instead of waiting on a connection nobody serves.

Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
 src/rust/Cargo.toml                         |   1 +
 src/rust/pmxcfs-ffi/Cargo.toml              |  17 ++
 src/rust/pmxcfs-ffi/include/pmxcfs-notify.h |  24 ++
 src/rust/pmxcfs-ffi/src/lib.rs              | 306 ++++++++++++++++++++
 src/rust/pmxcfs-notify/src/protocol.rs      |   6 +-
 5 files changed, 353 insertions(+), 1 deletion(-)
 create mode 100644 src/rust/pmxcfs-ffi/Cargo.toml
 create mode 100644 src/rust/pmxcfs-ffi/include/pmxcfs-notify.h
 create mode 100644 src/rust/pmxcfs-ffi/src/lib.rs

diff --git a/src/rust/Cargo.toml b/src/rust/Cargo.toml
index 8cf928e..55e0565 100644
--- a/src/rust/Cargo.toml
+++ b/src/rust/Cargo.toml
@@ -1,5 +1,6 @@
 [workspace]
 members = [
+    "pmxcfs-ffi",
     "pmxcfs-notify",
 ]
 resolver = "3"
diff --git a/src/rust/pmxcfs-ffi/Cargo.toml b/src/rust/pmxcfs-ffi/Cargo.toml
new file mode 100644
index 0000000..8264cde
--- /dev/null
+++ b/src/rust/pmxcfs-ffi/Cargo.toml
@@ -0,0 +1,17 @@
+[package]
+name = "pmxcfs-ffi"
+version = "0.1.0"
+description = "C ABI bridge exposing pmxcfs-notify to the pmxcfs daemon"
+authors.workspace = true
+edition.workspace = true
+license.workspace = true
+homepage.workspace = true
+rust-version.workspace = true
+
+[lib]
+crate-type = ["staticlib"]
+
+[dependencies]
+libc.workspace = true
+log.workspace = true
+pmxcfs-notify.workspace = true
diff --git a/src/rust/pmxcfs-ffi/include/pmxcfs-notify.h b/src/rust/pmxcfs-ffi/include/pmxcfs-notify.h
new file mode 100644
index 0000000..8e0f0ce
--- /dev/null
+++ b/src/rust/pmxcfs-ffi/include/pmxcfs-notify.h
@@ -0,0 +1,24 @@
+#ifndef PMXCFS_NOTIFY_H
+#define PMXCFS_NOTIFY_H
+
+#include <sys/types.h>
+
+/* mirrored by EventKind on the Rust side, the values must stay the same */
+enum pmxcfs_notify_type {
+    PMXCFS_NOTIFY_CREATE = 0,
+    PMXCFS_NOTIFY_WRITE = 1,
+    PMXCFS_NOTIFY_MTIME = 2,
+    PMXCFS_NOTIFY_RENAME = 3,
+    PMXCFS_NOTIFY_DELETE = 4,
+    PMXCFS_NOTIFY_MKDIR = 5,
+    PMXCFS_NOTIFY_RESYNC = 6,
+};
+
+/* priority is a syslog(3) level */
+typedef void (*pmxcfs_notify_log_fn)(int priority, const char *msg);
+
+int pmxcfs_notify_init(const char *socket_path, gid_t gid, pmxcfs_notify_log_fn log_cb);
+void pmxcfs_notify_emit(enum pmxcfs_notify_type type, const char *path, const char *to);
+void pmxcfs_notify_shutdown(void);
+
+#endif /* PMXCFS_NOTIFY_H */
diff --git a/src/rust/pmxcfs-ffi/src/lib.rs b/src/rust/pmxcfs-ffi/src/lib.rs
new file mode 100644
index 0000000..7291d7d
--- /dev/null
+++ b/src/rust/pmxcfs-ffi/src/lib.rs
@@ -0,0 +1,306 @@
+//! C ABI for the pmxcfs daemon, declared in include/pmxcfs-notify.h.
+//!
+//! Rust aborts the process when a panic reaches an `extern "C"` frame, which must never happen to
+//! pmxcfs. Every entry point therefore runs under `catch_unwind`, and a panic in one switches the
+//! notifier off for the rest of the daemon's life instead of taking it down.
+
+use std::ffi::{CStr, CString, OsStr, c_char, c_int};
+use std::os::unix::ffi::OsStrExt;
+use std::panic::{UnwindSafe, catch_unwind};
+use std::path::Path;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Mutex, MutexGuard, Once};
+
+use log::{Level, LevelFilter, Log, Metadata, Record};
+use pmxcfs_notify::{Config, EventKind, Server};
+
+pub type LogFn = unsafe extern "C" fn(c_int, *const c_char);
+
+const CAPACITY: usize = 8192;
+
+static SERVER: Mutex<Option<Server>> = Mutex::new(None);
+static DISABLED: AtomicBool = AtomicBool::new(false);
+static LOG_CB: Mutex<Option<LogFn>> = Mutex::new(None);
+static LOGGER: CLogger = CLogger;
+static LOGGER_INSTALLED: Once = Once::new();
+
+struct CLogger;
+
+impl Log for CLogger {
+    fn enabled(&self, _: &Metadata) -> bool {
+        true
+    }
+
+    fn log(&self, record: &Record) {
+        let Some(callback) = *lock(&LOG_CB) else {
+            return;
+        };
+        let priority = match record.level() {
+            Level::Error => libc::LOG_ERR,
+            Level::Warn => libc::LOG_WARNING,
+            Level::Info => libc::LOG_INFO,
+            Level::Debug | Level::Trace => libc::LOG_DEBUG,
+        };
+        let msg = CString::new(record.args().to_string().replace('\0', " "))
+            .expect("no NUL left after the replace");
+        unsafe { callback(priority, msg.as_ptr()) };
+    }
+
+    fn flush(&self) {}
+}
+
+fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
+    mutex
+        .lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner())
+}
+
+fn guard<R>(what: &str, fallback: R, f: impl FnOnce() -> R + UnwindSafe) -> R {
+    if DISABLED.load(Ordering::Relaxed) {
+        return fallback;
+    }
+    match catch_unwind(f) {
+        Ok(result) => result,
+        Err(payload) => {
+            DISABLED.store(true, Ordering::Relaxed);
+            let reason = payload
+                .downcast_ref::<&str>()
+                .copied()
+                .or_else(|| payload.downcast_ref::<String>().map(String::as_str))
+                .unwrap_or("<unexpected panic value>");
+            let _ = catch_unwind(|| {
+                log::error!("{what} panicked, notifications disabled: {reason}");
+            });
+            let _ = catch_unwind(stop_server);
+            fallback
+        }
+    }
+}
+
+fn stop_server() {
+    let server = lock(&SERVER).take();
+    if let Some(server) = server {
+        server.shutdown();
+    }
+}
+
+unsafe fn c_str<'a>(ptr: *const c_char) -> Option<&'a CStr> {
+    if ptr.is_null() {
+        return None;
+    }
+    Some(unsafe { CStr::from_ptr(ptr) })
+}
+
+/// # Safety
+///
+/// `socket_path` must be NULL or point to a NUL terminated string.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn pmxcfs_notify_init(
+    socket_path: *const c_char,
+    gid: libc::gid_t,
+    log_cb: Option<LogFn>,
+) -> c_int {
+    guard("init", -1, || {
+        let Some(path) = (unsafe { c_str(socket_path) }) else {
+            return -1;
+        };
+        let path = Path::new(OsStr::from_bytes(path.to_bytes()));
+        *lock(&LOG_CB) = log_cb;
+        LOGGER_INSTALLED.call_once(|| {
+            let _ = log::set_logger(&LOGGER);
+            log::set_max_level(LevelFilter::Debug);
+        });
+
+        let mut server = lock(&SERVER);
+        if server.is_some() {
+            log::warn!("notification socket is already running");
+            return 0;
+        }
+        match Server::start(Config {
+            path: path.to_owned(),
+            gid,
+            capacity: CAPACITY,
+        }) {
+            Ok(started) => {
+                *server = Some(started);
+                log::info!("listening on {}", path.display());
+                0
+            }
+            Err(err) => {
+                log::error!("failed to start notification socket: {err:#}");
+                -1
+            }
+        }
+    })
+}
+
+/// # Safety
+///
+/// `kind` must be a value of enum pmxcfs_notify_type, and `path` and `to` must each be NULL or
+/// point to a NUL terminated string.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn pmxcfs_notify_emit(
+    kind: EventKind,
+    path: *const c_char,
+    to: *const c_char,
+) {
+    guard("emit", (), || {
+        let server = lock(&SERVER);
+        let Some(server) = server.as_ref() else {
+            return;
+        };
+        match kind {
+            EventKind::Create
+            | EventKind::Write
+            | EventKind::Mtime
+            | EventKind::Rename
+            | EventKind::Delete
+            | EventKind::Mkdir
+            | EventKind::Resync => {}
+            // #[non_exhaustive] requires this arm. It cannot catch a bad value from C, which is
+            // undefined behavior before this point
+            _ => {
+                log::warn!("ignoring event with unknown type");
+                return;
+            }
+        }
+        let path = unsafe { c_str(path) }.map(CStr::to_str).transpose();
+        let to = unsafe { c_str(to) }.map(CStr::to_str).transpose();
+        match (path, to) {
+            (Ok(path), Ok(to)) => server.emit(kind, path, to),
+            // a JSON string holds only Unicode, so a path that is not UTF-8 is dropped rather than
+            // sent mangled
+            _ => log::warn!("ignoring {kind:?} event for a path that is not UTF-8"),
+        }
+    })
+}
+
+#[unsafe(no_mangle)]
+pub extern "C" fn pmxcfs_notify_shutdown() {
+    let _ = catch_unwind(|| {
+        stop_server();
+        *lock(&LOG_CB) = None;
+    });
+}
+
+#[cfg(test)]
+mod tests {
+    use std::io::{BufRead, BufReader, Read, Write};
+    use std::os::unix::fs::MetadataExt;
+    use std::os::unix::net::UnixStream;
+    use std::time::Duration;
+
+    use super::*;
+
+    static LINES: Mutex<Vec<(c_int, String)>> = Mutex::new(Vec::new());
+    // the tests share the global notifier state, so they run one at a time
+    static SERIAL: Mutex<()> = Mutex::new(());
+
+    unsafe extern "C" fn collect(priority: c_int, msg: *const c_char) {
+        let msg = unsafe { CStr::from_ptr(msg) }
+            .to_string_lossy()
+            .into_owned();
+        lock(&LINES).push((priority, msg));
+    }
+
+    fn warnings_containing(text: &str) -> usize {
+        lock(&LINES)
+            .iter()
+            .filter(|(prio, msg)| *prio == libc::LOG_WARNING && msg.contains(text))
+            .count()
+    }
+
+    #[test]
+    fn init_emit_shutdown_cycle() {
+        let _serial = lock(&SERIAL);
+        lock(&LINES).clear();
+        let dir = std::env::temp_dir().join(format!("pmxcfs-ffi-{}", std::process::id()));
+        std::fs::create_dir_all(&dir).unwrap();
+        let gid = std::fs::metadata(&dir).unwrap().gid();
+        let socket = CString::new(dir.join("sock").to_str().unwrap()).unwrap();
+
+        for _ in 0..2 {
+            let rc = unsafe { pmxcfs_notify_init(socket.as_ptr(), gid, Some(collect)) };
+            assert_eq!(rc, 0);
+            let path = CString::new("nodes/n1/qemu-server/100.conf").unwrap();
+            let odd = CString::new(b"nodes/n1/\xff.conf".to_vec()).unwrap();
+            unsafe {
+                pmxcfs_notify_emit(EventKind::Write, path.as_ptr(), std::ptr::null());
+                pmxcfs_notify_emit(EventKind::Resync, std::ptr::null(), std::ptr::null());
+                pmxcfs_notify_emit(EventKind::Delete, std::ptr::null(), std::ptr::null());
+                pmxcfs_notify_emit(EventKind::Rename, std::ptr::null(), path.as_ptr());
+                pmxcfs_notify_emit(EventKind::Write, odd.as_ptr(), std::ptr::null());
+                pmxcfs_notify_emit(EventKind::Rename, path.as_ptr(), odd.as_ptr());
+            }
+            pmxcfs_notify_shutdown();
+        }
+
+        let rc = unsafe { pmxcfs_notify_init(std::ptr::null(), gid, Some(collect)) };
+        assert_eq!(rc, -1);
+        let _ = std::fs::remove_dir_all(&dir);
+
+        let listening = lock(&LINES)
+            .iter()
+            .filter(|(prio, msg)| *prio == libc::LOG_INFO && msg.starts_with("listening on "))
+            .count();
+        assert_eq!(listening, 2);
+        assert_eq!(warnings_containing("Delete event without a path"), 2);
+        assert_eq!(warnings_containing("Rename event without a path"), 2);
+        assert_eq!(
+            warnings_containing("Write event for a path that is not UTF-8"),
+            2
+        );
+        assert_eq!(
+            warnings_containing("Rename event for a path that is not UTF-8"),
+            2
+        );
+    }
+
+    #[test]
+    fn panic_closes_the_connections() {
+        let _serial = lock(&SERIAL);
+        lock(&LINES).clear();
+        let dir = std::env::temp_dir().join(format!("pmxcfs-ffi-panic-{}", std::process::id()));
+        std::fs::create_dir_all(&dir).unwrap();
+        let gid = std::fs::metadata(&dir).unwrap().gid();
+        let path = dir.join("sock");
+        let socket = CString::new(path.to_str().unwrap()).unwrap();
+        assert_eq!(
+            unsafe { pmxcfs_notify_init(socket.as_ptr(), gid, Some(collect)) },
+            0
+        );
+        let mut client = UnixStream::connect(&path).unwrap();
+        client
+            .set_read_timeout(Some(Duration::from_secs(5)))
+            .unwrap();
+        // a reply proves the server accepted the connection. Stopping the server after the panic
+        // below then closes an established socket, not a pending one the peer would see as a reset
+        client.write_all(b"{\"command\":\"hello\"}\n").unwrap();
+        let mut reply = String::new();
+        BufReader::new(client.try_clone().unwrap())
+            .read_line(&mut reply)
+            .unwrap();
+        assert!(reply.starts_with("{\"ok\""), "{reply}");
+
+        assert_eq!(guard("test", -1, || -> c_int { panic!("injected") }), -1);
+
+        assert!(DISABLED.load(Ordering::Relaxed));
+        assert!(lock(&SERVER).is_none());
+        assert!(!path.exists());
+        let mut sink = Vec::new();
+        assert_eq!(client.read_to_end(&mut sink).unwrap(), 0, "client sees EOF");
+        assert!(
+            lock(&LINES)
+                .iter()
+                .any(|(prio, msg)| *prio == libc::LOG_ERR && msg.contains("test panicked"))
+        );
+
+        // a disabled notifier ignores everything, including a new init
+        assert_eq!(
+            unsafe { pmxcfs_notify_init(socket.as_ptr(), gid, Some(collect)) },
+            -1
+        );
+        DISABLED.store(false, Ordering::Relaxed);
+        let _ = std::fs::remove_dir_all(&dir);
+    }
+}
diff --git a/src/rust/pmxcfs-notify/src/protocol.rs b/src/rust/pmxcfs-notify/src/protocol.rs
index d28c9dc..e706cda 100644
--- a/src/rust/pmxcfs-notify/src/protocol.rs
+++ b/src/rust/pmxcfs-notify/src/protocol.rs
@@ -10,6 +10,9 @@ pub(crate) const PROTOCOL_VERSION: u32 = 1;
 /// Keyed by subscription name, with one capture set per side of the event that matched.
 pub(crate) type Params<'a> = BTreeMap<&'a str, Vec<Captures<'a>>>;
 
+// Crosses the C ABI by value, so the order has to match enum pmxcfs_notify_type in the pmxcfs-ffi
+// header.
+#[repr(C)]
 #[non_exhaustive]
 #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
 #[serde(rename_all = "lowercase")]
@@ -35,7 +38,8 @@ pub(crate) struct Event<'a> {
     pub params: Option<Params<'a>>,
 }
 
-/// Everything the server sends. The variant name is the single key of the JSON object on the wire.
+/// Everything the server sends. The variant name is the single key of
+/// the JSON object on the wire.
 #[derive(Serialize)]
 #[serde(rename_all = "lowercase")]
 pub(crate) enum Message<'a> {
-- 
2.47.3