From mboxrd@z Thu Jan 1 00:00:00 1970 Return-Path: Received: from gate001.proxmox.com (gate001.proxmox.com [IPv6:2a0f:8001:1:32::40]) by lore.proxmox.com (Postfix) with ESMTPS id 441201FF0C1 for ; Fri, 18 Sep 2026 16:42:33 +0200 (CEST) Received: from gate001.proxmox.com (localhost.localdomain [127.0.0.1]) by gate001.proxmox.com (Proxmox) with ESMTP id DC6F2215DC; Fri, 18 Sep 2026 16:42:11 +0200 (CEST) From: Hannes Laimer To: pve-devel@lists.proxmox.com Subject: [PATCH pve-cluster 03/10] rust: ffi: add C ABI staticlib for pmxcfs Date: Fri, 18 Sep 2026 16:41:45 +0200 Message-ID: <20260918144152.575163-4-h.laimer@proxmox.com> X-Mailer: git-send-email 2.47.3 In-Reply-To: <20260918144152.575163-1-h.laimer@proxmox.com> References: <20260918144152.575163-1-h.laimer@proxmox.com> MIME-Version: 1.0 Content-Transfer-Encoding: 8bit X-Bm-Milter-Handled: 55990f41-d878-4baa-be0a-ee34c49e34d2 X-Bm-Transport-Timestamp: 1789742526122 X-SPAM-LEVEL: Spam detection results: 0 AWL 0.494 Adjusted score from AWL reputation of From: address DMARC_MISSING 0.1 Missing DMARC policy KAM_DMARC_STATUS 0.01 Test Rule for DKIM or SPF Failure with Strict Alignment (newer systems) RCVD_IN_DNSWL_MED -2.3 Sender listed at https://www.dnswl.org/, medium trust RCVD_IN_MSPIKE_H2 0.001 Average reputation (+2) SPF_HELO_NONE 0.001 SPF: HELO does not publish an SPF Record SPF_PASS -0.001 SPF: sender matches SPF record Message-ID-Hash: 67BNOITC37LTZPLRSRDS6LQ3FEN2WTDM X-Message-ID-Hash: 67BNOITC37LTZPLRSRDS6LQ3FEN2WTDM X-MailFrom: h.laimer@proxmox.com X-Mailman-Rule-Misses: dmarc-mitigation; no-senders; approved; loop; banned-address; emergency; member-moderation; nonmember-moderation; administrivia; implicit-dest; max-recipients; max-size; news-moderation; no-subject; digests; suspicious-header X-Mailman-Version: 3.3.10 Precedence: list List-Id: Proxmox VE development discussion List-Help: List-Owner: List-Post: List-Subscribe: List-Unsubscribe: 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 --- 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 +#include + +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> = Mutex::new(None); +static DISABLED: AtomicBool = AtomicBool::new(false); +static LOG_CB: Mutex> = 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(mutex: &Mutex) -> MutexGuard<'_, T> { + mutex + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) +} + +fn guard(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::().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 { + 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, + 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> = 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