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

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

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