Milestone 5: metered awareness, DSCP, systemd, man pages, packaging

Metered detection polls NetworkManager's Metered property over D-Bus
(feature "metered", default on; builds without D-Bus via
no-default-features). While metered the daemon closes incoming blob
connections and defers every fetch; VARDE_FORCE_METERED=true forces the
state as a kill switch and test hook. Endpoint UDP sockets get DSCP CS1
best-effort by matching bound ports to /proc/net/udp inodes (iroh hides
its fds). systemd socket activation adopts LISTEN_FDS fd 3, readiness
is a hand-rolled sd_notify READY=1 (abstract + path sockets), and
standalone binding still works unchanged. dist/ ships hardened system
and user units (DynamicUser, ProtectSystem=strict, StateDirectory,
RestrictAddressFamilies), a commented config example, scdoc man pages
validated with scdoc, and an untested PKGBUILD skeleton.

Tests: activation-socket round trip via a real fd-3 handoff, READY=1
received on a NOTIFY_SOCKET, metered daemons neither serve nor fetch.

Dependencies: zbus (optional, feature-gated D-Bus client for the
NetworkManager metered flag).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

varde-daemon/src/daemon.rs

@@ -58,6 +58,7 @@ impl Daemon {
         let open_filter: crate::shaped::ServeFilter =
             Arc::new(move |hash| filter_set.read().expect("open set lock").contains(hash));
 
+        let metered = crate::metered::start();
         let transfer = Transfer::start(
             &config,
             &store,
@@ -65,6 +66,7 @@ impl Daemon {
             open_filter,
             store.pool_handle(),
             events.clone(),
+            metered,
         )
         .await?;
 

varde-daemon/src/dscp.rs

@@ -0,0 +1,102 @@
+//! Best-effort DSCP marking of the endpoint's UDP sockets.
+//!
+//! Background transfer traffic is marked CS1 (low priority) so routers
+//! that honor DSCP deprioritize it. iroh does not expose its socket fds,
+//! so this walks `/proc/self/fd` for the UDP sockets bound to the
+//! endpoint's ports — Linux-only by construction, like the daemon.
+//! Failures are logged at debug and ignored: some environments strip or
+//! forbid DSCP (may require CAP_NET_ADMIN), and that's fine.
+
+use std::net::SocketAddr;
+use std::os::fd::RawFd;
+
+use tracing::debug;
+
+/// DSCP CS1 (001000) in the TOS byte.
+const TOS_CS1: u32 = 0x20;
+
+extern "C" {
+    fn setsockopt(fd: i32, level: i32, name: i32, value: *const u32, len: u32) -> i32;
+}
+
+const IPPROTO_IP: i32 = 0;
+const IP_TOS: i32 = 1;
+const IPPROTO_IPV6: i32 = 41;
+const IPV6_TCLASS: i32 = 67;
+
+/// Mark every UDP socket bound to one of `addrs` with DSCP CS1.
+pub fn mark_endpoint_sockets(addrs: &[SocketAddr]) {
+    let ports: Vec<u16> = addrs.iter().map(|a| a.port()).collect();
+    let inodes = udp_socket_inodes(&ports);
+    if inodes.is_empty() {
+        debug!("dscp: no matching udp sockets found");
+        return;
+    }
+    let mut marked = 0;
+    for fd in socket_fds(&inodes) {
+        // Set both levels; one of them will apply depending on family.
+        let v4 = unsafe { setsockopt(fd, IPPROTO_IP, IP_TOS, &TOS_CS1, 4) };
+        let v6 = unsafe { setsockopt(fd, IPPROTO_IPV6, IPV6_TCLASS, &TOS_CS1, 4) };
+        if v4 == 0 || v6 == 0 {
+            marked += 1;
+        }
+    }
+    debug!(marked, "dscp: marked endpoint sockets CS1");
+}
+
+/// Inodes of UDP sockets locally bound to one of `ports`, from
+/// /proc/net/udp{,6}. Format: whitespace columns, local_address is
+/// `HEXIP:HEXPORT` in column 1, inode in column 9.
+fn udp_socket_inodes(ports: &[u16]) -> Vec<u64> {
+    let mut inodes = Vec::new();
+    for table in ["/proc/net/udp", "/proc/net/udp6"] {
+        let Ok(text) = std::fs::read_to_string(table) else {
+            continue;
+        };
+        for line in text.lines().skip(1) {
+            let fields: Vec<&str> = line.split_whitespace().collect();
+            let (Some(local), Some(inode)) = (fields.get(1), fields.get(9)) else {
+                continue;
+            };
+            let Some((_, port_hex)) = local.rsplit_once(':') else {
+                continue;
+            };
+            let Ok(port) = u16::from_str_radix(port_hex, 16) else {
+                continue;
+            };
+            if ports.contains(&port) {
+                if let Ok(inode) = inode.parse() {
+                    inodes.push(inode);
+                }
+            }
+        }
+    }
+    inodes
+}
+
+/// Our process's fds that are sockets with one of the given inodes.
+fn socket_fds(inodes: &[u64]) -> Vec<RawFd> {
+    let mut fds = Vec::new();
+    let Ok(entries) = std::fs::read_dir("/proc/self/fd") else {
+        return fds;
+    };
+    for entry in entries.flatten() {
+        let Ok(target) = std::fs::read_link(entry.path()) else {
+            continue;
+        };
+        let target = target.to_string_lossy();
+        let Some(inode) = target
+            .strip_prefix("socket:[")
+            .and_then(|s| s.strip_suffix(']'))
+            .and_then(|s| s.parse::<u64>().ok())
+        else {
+            continue;
+        };
+        if inodes.contains(&inode) {
+            if let Ok(fd) = entry.file_name().to_string_lossy().parse::<RawFd>() {
+                fds.push(fd);
+            }
+        }
+    }
+    fds
+}

varde-daemon/src/lib.rs

@@ -5,7 +5,9 @@
 
 pub mod config;
 pub mod daemon;
+pub mod dscp;
 pub mod meta;
+pub mod metered;
 pub mod server;
 pub mod shaped;
 pub mod store;

varde-daemon/src/main.rs

@@ -82,16 +82,25 @@ fn main() -> Result<()> {
 async fn run(config: Config) -> Result<()> {
     let socket_path = config.socket_path.clone();
     let daemon = daemon::Daemon::open(config).await?;
-    let listener = server::bind_socket(&socket_path)?;
+    // Prefer a systemd activation socket; bind ourselves otherwise so
+    // non-systemd distros work identically.
+    let (listener, activated) = match server::activation_listener()? {
+        Some(listener) => (listener, true),
+        None => (server::bind_socket(&socket_path)?, false),
+    };
+    server::notify_ready();
 
     let result = tokio::select! {
         r = server::serve(listener, daemon.clone()) => r,
         r = shutdown_signal() => r.map(|signal| info!(signal, "shutting down")),
     };
 
-    // Close the endpoint, flush the store, remove the socket.
+    // Close the endpoint and flush the store. The socket file is ours to
+    // remove only when we bound it (systemd owns activation sockets).
     daemon.shutdown().await;
-    let _ = std::fs::remove_file(&socket_path);
+    if !activated {
+        let _ = std::fs::remove_file(&socket_path);
+    }
     result
 }
 

varde-daemon/src/metered.rs

@@ -0,0 +1,99 @@
+//! Metered-connection awareness.
+//!
+//! When the active connection is metered, varde stops moving bytes:
+//! no background fetching, no serving. Detection queries
+//! NetworkManager's `Metered` property over D-Bus and is feature-gated
+//! (`metered`, on by default) so the daemon still builds and runs on
+//! systems without D-Bus — those simply always count as unmetered.
+//!
+//! `VARDE_FORCE_METERED=true` forces the metered state regardless of
+//! NetworkManager; useful for testing and as a manual kill switch.
+
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::Arc;
+
+/// Shared metered state, cheap to read on every transfer decision.
+#[derive(Debug, Clone, Default)]
+pub struct MeteredState(Arc<AtomicBool>);
+
+impl MeteredState {
+    pub fn is_metered(&self) -> bool {
+        self.0.load(Ordering::Relaxed)
+    }
+
+    fn set(&self, metered: bool) {
+        let was = self.0.swap(metered, Ordering::Relaxed);
+        if was != metered {
+            tracing::info!(metered, "metered state changed");
+        }
+    }
+}
+
+/// Start metered detection and return the shared state.
+pub fn start() -> MeteredState {
+    let state = MeteredState::default();
+    if std::env::var("VARDE_FORCE_METERED").as_deref() == Ok("true") {
+        tracing::warn!("VARDE_FORCE_METERED=true: all transfers disabled");
+        state.set(true);
+        return state;
+    }
+    #[cfg(feature = "metered")]
+    {
+        let state = state.clone();
+        tokio::spawn(async move { poll_network_manager(state).await });
+    }
+    state
+}
+
+/// NetworkManager's NM_METERED enum: 1 = yes, 3 = guess-yes.
+#[cfg(feature = "metered")]
+async fn poll_network_manager(state: MeteredState) {
+    use std::time::Duration;
+
+    let connection = match zbus::Connection::system().await {
+        Ok(c) => c,
+        Err(e) => {
+            tracing::debug!(error = %e, "no D-Bus system bus; assuming unmetered");
+            return;
+        }
+    };
+    let proxy = match zbus::Proxy::new(
+        &connection,
+        "org.freedesktop.NetworkManager",
+        "/org/freedesktop/NetworkManager",
+        "org.freedesktop.NetworkManager",
+    )
+    .await
+    {
+        Ok(p) => p,
+        Err(e) => {
+            tracing::debug!(error = %e, "NetworkManager proxy failed; assuming unmetered");
+            return;
+        }
+    };
+
+    loop {
+        match proxy.get_property::<u32>("Metered").await {
+            Ok(value) => state.set(value == 1 || value == 3),
+            Err(e) => {
+                tracing::debug!(error = %e, "reading Metered property");
+            }
+        }
+        tokio::time::sleep(Duration::from_secs(30)).await;
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn defaults_unmetered_and_flips() {
+        let state = MeteredState::default();
+        assert!(!state.is_metered());
+        state.set(true);
+        assert!(state.is_metered());
+        state.set(false);
+        assert!(!state.is_metered());
+    }
+}

varde-daemon/src/server.rs

@@ -11,6 +11,57 @@ use varde_proto::{Envelope, ErrorCode, Request};
 
 use crate::daemon::Daemon;
 
+/// A listener passed in by systemd socket activation (`LISTEN_FDS`),
+/// if any. Protocol: fds start at 3; `LISTEN_PID`, when set, must be us.
+pub fn activation_listener() -> Result<Option<UnixListener>> {
+    let Ok(listen_fds) = std::env::var("LISTEN_FDS") else {
+        return Ok(None);
+    };
+    if let Ok(pid) = std::env::var("LISTEN_PID") {
+        if pid != std::process::id().to_string() {
+            return Ok(None);
+        }
+    }
+    let n: u32 = listen_fds.parse().context("parsing LISTEN_FDS")?;
+    anyhow::ensure!(n == 1, "expected exactly one activation fd, got {n}");
+    // SAFETY: fd 3 is the first activation fd per the LISTEN_FDS
+    // protocol; we take sole ownership of it.
+    let std_listener = unsafe {
+        use std::os::fd::FromRawFd;
+        std::os::unix::net::UnixListener::from_raw_fd(3)
+    };
+    std_listener
+        .set_nonblocking(true)
+        .context("setting activation socket nonblocking")?;
+    let listener = UnixListener::from_std(std_listener).context("adopting activation socket")?;
+    info!("using systemd activation socket");
+    Ok(Some(listener))
+}
+
+/// Tell the service manager we are ready (`sd_notify(READY=1)`),
+/// hand-rolled to avoid a libsystemd dependency. No-op without
+/// `NOTIFY_SOCKET`.
+pub fn notify_ready() {
+    let Some(path) = std::env::var_os("NOTIFY_SOCKET") else {
+        return;
+    };
+    let result = (|| -> std::io::Result<()> {
+        let socket = std::os::unix::net::UnixDatagram::unbound()?;
+        let bytes = path.as_encoded_bytes();
+        if let Some(name) = bytes.strip_prefix(b"@") {
+            use std::os::linux::net::SocketAddrExt;
+            let addr = std::os::unix::net::SocketAddr::from_abstract_name(name)?;
+            socket.send_to_addr(b"READY=1", &addr)?;
+        } else {
+            socket.send_to(b"READY=1", &path)?;
+        }
+        Ok(())
+    })();
+    if let Err(e) = result {
+        debug!(error = %e, "sd_notify failed");
+    }
+}
+
 /// Bind the API socket, replacing a stale socket file if the previous
 /// daemon did not shut down cleanly.
 pub fn bind_socket(path: &Path) -> Result<UnixListener> {

varde-daemon/src/transfer.rs

@@ -27,6 +27,7 @@ use varde_proto::{Direction, Event};
 
 use crate::config::Config;
 use crate::meta::Meta;
+use crate::metered::MeteredState;
 use crate::shaped::{ServeFilter, ShapedStore, TokenBucket};
 use crate::store::BlobStore;
 
@@ -39,6 +40,7 @@ pub struct Transfer {
     router: Router,
     downloader: Downloader,
     events: broadcast::Sender<Event>,
+    metered: MeteredState,
     discovery_rx: std::sync::Mutex<Option<mpsc::Receiver<NodeId>>>,
 }
 
@@ -54,6 +56,7 @@ struct GatedProvider {
     events: EventSender,
     pool: LocalPoolHandle,
     serving_enabled: bool,
+    metered: MeteredState,
 }
 
 impl ProtocolHandler for GatedProvider {
@@ -61,8 +64,8 @@ impl ProtocolHandler for GatedProvider {
         let this = self.clone();
         Box::pin(async move {
             let remote = connection.remote_node_id()?;
-            if !this.serving_enabled {
-                debug!(%remote, "serving disabled, closing connection");
+            if !this.serving_enabled || this.metered.is_metered() {
+                debug!(%remote, "serving disabled or metered, closing connection");
                 connection.close(0u32.into(), b"serving disabled");
                 return Ok(());
             }
@@ -116,6 +119,7 @@ impl Transfer {
         open_filter: ServeFilter,
         pool: LocalPoolHandle,
         events: broadcast::Sender<Event>,
+        metered: MeteredState,
     ) -> Result<Transfer> {
         let secret = load_or_create_secret(&config.store_dir.join("secret.key"))?;
         let mut builder = Endpoint::builder().secret_key(secret.clone());
@@ -146,6 +150,11 @@ impl Transfer {
         let endpoint = builder.bind().await.context("binding iroh endpoint")?;
         info!(node_id = %endpoint.node_id(), discovery = config.discovery, "endpoint up");
 
+        // Best-effort low-priority marking of our UDP traffic.
+        let (v4, v6) = endpoint.bound_sockets();
+        let bound: Vec<std::net::SocketAddr> = std::iter::once(v4).chain(v6).collect();
+        crate::dscp::mark_endpoint_sockets(&bound);
+
         let upload_bucket = TokenBucket::new(config.max_upload_bytes_per_sec);
         let download_bucket = TokenBucket::new(config.max_download_bytes_per_sec);
         let shaped = ShapedStore::new(store.inner().clone(), upload_bucket, download_bucket);
@@ -158,6 +167,7 @@ impl Transfer {
             events: provider_events,
             pool: pool.clone(),
             serving_enabled: config.max_upload_bytes_per_sec > 0,
+            metered: metered.clone(),
         };
 
         let downloader = Downloader::new(shaped, endpoint.clone(), pool);
@@ -170,6 +180,7 @@ impl Transfer {
             router,
             downloader,
             events,
+            metered,
             discovery_rx: std::sync::Mutex::new(Some(discovery_rx)),
         })
     }
@@ -200,6 +211,11 @@ impl Transfer {
     /// Queue a background fetch of `content` from `providers`. Emits
     /// progress events and [`Event::PinComplete`] when fully verified.
     pub fn spawn_fetch(&self, content: HashAndFormat, providers: Vec<NodeAddr>) {
+        if self.metered.is_metered() {
+            // The pin stays recorded; auto-sync retries once unmetered.
+            info!(hash = %content.hash.to_hex(), "metered connection: fetch deferred");
+            return;
+        }
         let downloader = self.downloader.clone();
         let events = self.events.clone();
         let endpoint = self.endpoint.clone();