[PATCH proxmox-ebpf v4 2/5] subsystem: add the shared tc program handling

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

A subsystem owns tc classifier programs and their maps, attached per
interface and pinned in bpffs together with an object fingerprint
and a schema version. So state survives between the one-shot calls
of its consumer, and re-running never interrupts traffic.

Changes run under one lock per subsystem, taken before the change
reads its input, so they reach the kernel in the order they read it.

A subsystem describes its programs with a macro, from its name, the
program names, the maps and the schema version. The object is found
by the subsystem's name, and its fingerprint is derived from its
bytes.

Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
 src/lib.rs       |   1 +
 src/subsystem.rs | 829 +++++++++++++++++++++++++++++++++++++++++++++++
 src/tc.rs        |  12 +-
 3 files changed, 836 insertions(+), 6 deletions(-)
 create mode 100644 src/subsystem.rs

diff --git a/src/lib.rs b/src/lib.rs
index 8d4420a..572d861 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -1,4 +1,5 @@
 //! eBPF subsystems, one per concern, each behind a cargo feature of its name. Nothing here runs
 //! on its own, consumers pull in only the subsystem they drive.
 
+pub mod subsystem;
 pub mod tc;
diff --git a/src/subsystem.rs b/src/subsystem.rs
new file mode 100644
index 0000000..c740fa8
--- /dev/null
+++ b/src/subsystem.rs
@@ -0,0 +1,829 @@
+//! The shared subsystem code. A subsystem has one tc program per direction it declares, with
+//! their maps and links. They are pinned under `/sys/fs/bpf/proxmox-ebpf/<name>/<schema>/`. The
+//! programs are pinned by the fingerprint of their object, so the pinned state itself says which
+//! build it belongs to.
+//!
+//! The loaded BPF stays in the kernel between invocations, so [`ApplyLock::ensure_loaded`] loads
+//! only when what is pinned is not this build's, or links still sit under another schema. Every
+//! other step works on what is already pinned. Each subsystem owns one [`TcPrograms`].
+//!
+//! Every change runs under the subsystem's apply lock, a full apply exclusively and a
+//! single-interface step shared. The holder reads the input of its change under the lock and
+//! writes before it lets go, so a full apply and every step reach the kernel in the order they
+//! read their input. Two steps of one interface do not order each other.
+use std::{
+    collections::HashSet,
+    fs::File,
+    io::ErrorKind,
+    marker::PhantomData,
+    path::{Path, PathBuf},
+    time::{Duration, Instant},
+};
+
+use anyhow::Context;
+use aya::{EbpfLoader, programs::SchedClassifier};
+use nix::{
+    errno::Errno,
+    fcntl::{Flock, FlockArg},
+};
+
+use crate::tc::{self, Direction};
+
+const PIN_ROOT: &str = "/sys/fs/bpf/proxmox-ebpf";
+const RUN_ROOT: &str = "/run/proxmox-ebpf";
+
+const fn obj_fingerprint(obj: &[u8]) -> u64 {
+    let mut hash = 0xcbf29ce484222325u64;
+    let mut i = 0;
+    while i < obj.len() {
+        hash ^= obj[i] as u64;
+        hash = hash.wrapping_mul(0x100000001b3);
+        i += 1;
+    }
+    hash
+}
+
+/// A subsystem's [`TcPrograms`] from its name. The object is the build's `<name>.bpf.o`, the
+/// fingerprint follows from the bytes.
+#[macro_export]
+macro_rules! tc_programs {
+    (
+        name: $name:literal,
+        programs: $programs:expr,
+        maps: $maps:expr,
+        schema_version: $schema:expr $(,)?
+    ) => {
+        // built at compile time wherever the macro is used, a let binding included
+        const {
+            const OBJ: &[u8] =
+                ::aya::include_bytes_aligned!(concat!(env!("OUT_DIR"), "/", $name, ".bpf.o"));
+            $crate::subsystem::TcPrograms::new($name, OBJ, $programs, $maps, $schema)
+        }
+    };
+}
+
+const LOCK_WAIT: Duration = Duration::from_secs(10);
+const LOCK_POLL: Duration = Duration::from_millis(10);
+
+// flock has no timeout, so the wait retries a nonblocking flock with a sleep in between.
+//
+// not proxmox-sys lock because: blocks and gives up on a timer signal, and under pve-rs any signal
+// perl handles cuts the blocked wait short the same way, since perl installs its handlers without
+// restart
+fn flock_within(
+    path: &Path,
+    exclusive: bool,
+    what: &str,
+    subsystem: &str,
+) -> anyhow::Result<Flock<File>> {
+    let mut file = File::options()
+        .create(true)
+        .truncate(false)
+        .write(true)
+        .open(path)?;
+    let arg = if exclusive {
+        FlockArg::LockExclusiveNonblock
+    } else {
+        FlockArg::LockSharedNonblock
+    };
+    let deadline = Instant::now() + LOCK_WAIT;
+    loop {
+        match Flock::lock(file, arg) {
+            Ok(lock) => return Ok(lock),
+            Err((returned, Errno::EWOULDBLOCK)) if Instant::now() < deadline => {
+                file = returned;
+                std::thread::sleep(LOCK_POLL);
+            }
+            Err((_, Errno::EWOULDBLOCK)) => anyhow::bail!(
+                "{subsystem}: {what} still held after {}s",
+                LOCK_WAIT.as_secs()
+            ),
+            Err((_, e)) => return Err(e.into()),
+        }
+    }
+}
+
+/// A step that needs the programs pinned found none. Only [`ApplyLock::ensure_loaded`] under the
+/// exclusive lock installs them.
+#[derive(Debug)]
+pub struct NotLoaded;
+
+impl std::fmt::Display for NotLoaded {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str("programs not loaded")
+    }
+}
+
+impl std::error::Error for NotLoaded {}
+
+/// A subsystem's description of its tc programs, written with the `tc_programs!` macro. The
+/// loaded programs live pinned in the kernel and are reached through those pins.
+pub struct TcPrograms {
+    /// The subsystem's name, its directory under the pin root.
+    name: &'static str,
+    /// The embedded BPF object.
+    obj: &'static [u8],
+    /// FNV-1a of `obj`, the programs are pinned under it.
+    fingerprint: u64,
+    /// The directions the object carries a program for, each with that program's name in the
+    /// object. One entry per direction.
+    programs: &'static [(Direction, &'static str)],
+    /// The object's pinned maps. A missing one means the state is not current.
+    maps: &'static [&'static str],
+    /// Bumped with any change to a map's key or value layout. A pinned map is reused by name
+    /// without any check, only the Rust accessor checks its key and value sizes.
+    schema_version: u32,
+}
+
+impl TcPrograms {
+    pub const fn new(
+        name: &'static str,
+        obj: &'static [u8],
+        programs: &'static [(Direction, &'static str)],
+        maps: &'static [&'static str],
+        schema_version: u32,
+    ) -> Self {
+        Self {
+            name,
+            obj,
+            fingerprint: obj_fingerprint(obj),
+            programs,
+            maps,
+            schema_version,
+        }
+    }
+
+    fn pin_root(&self) -> PathBuf {
+        PathBuf::from(PIN_ROOT).join(self.name)
+    }
+    fn schema_root(&self) -> PathBuf {
+        self.pin_root().join(self.schema_version.to_string())
+    }
+    fn links_dir(&self) -> PathBuf {
+        schema_links_dir(&self.schema_root())
+    }
+    fn link_pin_path(&self, pin: &tc::LinkPin) -> PathBuf {
+        self.links_dir().join(pin.filename())
+    }
+    fn progs_dir(&self) -> PathBuf {
+        schema_progs_dir(&self.schema_root())
+    }
+    fn fingerprint_dir_name(&self) -> String {
+        format!("{:016x}", self.fingerprint)
+    }
+    fn prog_dir(&self) -> PathBuf {
+        self.progs_dir().join(self.fingerprint_dir_name())
+    }
+    fn prog_pin_path(&self, dir: Direction) -> PathBuf {
+        self.prog_dir().join(dir.as_str())
+    }
+    fn program_name(&self, dir: Direction) -> Option<&'static str> {
+        self.programs
+            .iter()
+            .find(|&&(d, _)| d == dir)
+            .map(|&(_, name)| name)
+    }
+    fn run_dir(&self) -> PathBuf {
+        PathBuf::from(RUN_ROOT).join(self.name)
+    }
+    fn hash_map<K: aya::Pod, V: aya::Pod>(
+        &self,
+        name: &str,
+    ) -> anyhow::Result<aya::maps::HashMap<aya::maps::MapData, K, V>> {
+        let map = aya::maps::MapData::from_pin(self.schema_root().join(name))
+            .with_context(|| format!("{}: open pinned map {name}", self.name))?;
+        Ok(aya::maps::HashMap::try_from(aya::maps::Map::HashMap(map))?)
+    }
+
+    /// Open one direction's pinned program, [`NotLoaded`] when it is not pinned.
+    fn program(&self, dir: Direction) -> anyhow::Result<SchedClassifier> {
+        let path = self.prog_pin_path(dir);
+        if !path.exists() {
+            return Err(NotLoaded.into());
+        }
+        SchedClassifier::from_pin(path)
+            .with_context(|| format!("{}: open pinned program {}", self.name, dir.as_str()))
+    }
+
+    /// The apply lock under `/run`. A full apply, an install and a teardown take it exclusively,
+    /// a single-interface step shared. A wait times out instead of hanging. Shared holders do not
+    /// yield to a waiting exclusive one, so a node attaching interfaces without pause can keep a
+    /// full apply waiting until it times out. Every write to the pinned state is
+    /// a method of the returned guard, so a step cannot run outside the lock.
+    fn lock<M>(&self, exclusive: bool) -> anyhow::Result<ApplyLock<'_, M>> {
+        let dir = self.run_dir();
+        std::fs::create_dir_all(&dir)?;
+        let lock = flock_within(&dir.join("lock"), exclusive, "apply lock", self.name)?;
+        Ok(ApplyLock {
+            programs: self,
+            _lock: lock,
+            _mode: PhantomData,
+        })
+    }
+
+    /// Whether a link of the interface name is pinned, read without the lock. Most interfaces
+    /// never had a link and their unplug must not wait behind a full apply. A miss is a snapshot,
+    /// a pin arriving after it is another attach's. A failed listing counts as a hit. So do links
+    /// still pinned under another schema, an install may be moving them over.
+    pub fn link_pinned(&self, name: &str) -> bool {
+        let elsewhere = self
+            .other_schemas()
+            .map_or(true, |others| links_pinned_under(&others));
+        elsewhere
+            || self
+                .live_links()
+                .map_or(true, |links| links.iter().any(|pin| pin.name == name))
+    }
+
+    /// Take the apply lock shared, for a single-interface step.
+    pub fn lock_shared(&self) -> anyhow::Result<ApplyLock<'_, Shared>> {
+        self.lock(false)
+    }
+
+    /// Take the apply lock exclusively, for a full apply, an install or a teardown.
+    pub fn lock_exclusive(&self) -> anyhow::Result<ApplyLock<'_, Exclusive>> {
+        self.lock(true)
+    }
+
+    fn ensure_loaded(&self) -> anyhow::Result<()> {
+        let others = self.other_schemas()?;
+        // a link pinned under another schema is a live attachment to a program of that schema.
+        // A run that died between moving the links and pinning its programs leaves some of the
+        // links there. This build is not current until the install below has taken them over
+        let links_elsewhere = links_pinned_under(&others);
+        // programs pinned under a newer schema mean a newer build runs here and this process is
+        // the stale one, so it leaves the state alone. A newer schema directory without programs
+        // is what a rejected install of that build left behind
+        if let Some(newer) = self.newer_schema_installed(&others) {
+            anyhow::bail!(
+                "{}: pinned state is of schema {newer}, newer than this build's {}, clear it to \
+                 downgrade",
+                self.name,
+                self.schema_version
+            );
+        }
+        if self.is_current() && !links_elsewhere {
+            // an older schema here is left over from a run that died between its install and the
+            // removal, its links moved over already
+            for (dir, schema) in others {
+                if (schema.is_some_and(|s| s < self.schema_version)
+                    || !any_program_pinned(&schema_progs_dir(&dir)))
+                    && let Err(e) = std::fs::remove_dir_all(&dir)
+                {
+                    log::warn!(
+                        "{}: remove old pinned state {}: {e:#}",
+                        self.name,
+                        dir.display()
+                    );
+                }
+            }
+            return Ok(());
+        }
+
+        // a lost map pin comes back at load time, ahead of the programs. If the install then
+        // failed, the previous run's program pins would make that half-done state pass for
+        // current. So they go first, the links keep the old programs running meanwhile
+        if !self.maps_pinned() {
+            match std::fs::remove_dir_all(self.prog_dir()) {
+                Ok(()) => {}
+                Err(e) if e.kind() == ErrorKind::NotFound => {}
+                Err(e) => return Err(e).context("remove program pins ahead of the reinstall"),
+            }
+        }
+
+        // any other schema pinned here is old state whose map layout this build cannot bind to.
+        // It stays in place through the install and goes afterward
+        let schema_changed = !others.is_empty();
+        if schema_changed {
+            log::warn!("{}: pinned state of another schema, rebuilding", self.name);
+        }
+
+        // a failed install leaves its directory as it is, a link moved into it already is a live
+        // attachment. Without program pins the directory does not pass for state of that build
+        let old_links: Vec<PathBuf> = others
+            .iter()
+            .map(|(dir, _)| schema_links_dir(dir))
+            .collect();
+        self.install(&old_links).context("install")?;
+        for (dir, _) in &others {
+            // a link whose pin could not be moved is still live there
+            if any_links(&schema_links_dir(dir)) {
+                continue;
+            }
+            if let Err(e) = std::fs::remove_dir_all(dir) {
+                log::warn!(
+                    "{}: remove old pinned state {}: {e:#}",
+                    self.name,
+                    dir.display()
+                );
+            }
+        }
+        Ok(())
+    }
+
+    fn is_current(&self) -> bool {
+        self.programs
+            .iter()
+            .all(|&(dir, _)| self.prog_pin_path(dir).exists())
+            && self.maps_pinned()
+    }
+
+    fn maps_pinned(&self) -> bool {
+        self.maps
+            .iter()
+            .all(|map| self.schema_root().join(map).exists())
+    }
+
+    // the directories under the pin root other than this build's schema. One whose name is not
+    // a version counts as well, so its links are moved over too
+    fn other_schemas(&self) -> anyhow::Result<Vec<(PathBuf, Option<u32>)>> {
+        let entries = match std::fs::read_dir(self.pin_root()) {
+            Ok(entries) => entries,
+            Err(e) if e.kind() == ErrorKind::NotFound => return Ok(Vec::new()),
+            Err(e) => return Err(e.into()),
+        };
+        let mut others = Vec::new();
+        for entry in entries {
+            let entry = entry?;
+            if !entry.file_type()?.is_dir() {
+                continue;
+            }
+            let schema = entry
+                .file_name()
+                .to_str()
+                .and_then(|s| s.parse::<u32>().ok());
+            if schema != Some(self.schema_version) {
+                others.push((entry.path(), schema));
+            }
+        }
+        Ok(others)
+    }
+
+    fn newer_schema_installed(&self, others: &[(PathBuf, Option<u32>)]) -> Option<u32> {
+        others
+            .iter()
+            .filter(|(dir, _)| any_program_pinned(&schema_progs_dir(dir)))
+            .filter_map(|(_, schema)| *schema)
+            .find(|&s| s > self.schema_version)
+    }
+
+    fn tear_down(&self) -> anyhow::Result<()> {
+        match std::fs::remove_dir_all(self.pin_root()) {
+            Ok(()) => Ok(()),
+            Err(e) if e.kind() == ErrorKind::NotFound => Ok(()),
+            Err(e) => Err(e.into()),
+        }
+    }
+
+    /// Load this build's object and move every link onto its programs, this schema's and those
+    /// under `old_links`, then pin the programs and drop those of other builds. The verifier runs
+    /// here and nowhere else, so a rejection leaves what served before untouched. The program
+    /// pins mark this build as current and go last, so after a failure the previous build stays
+    /// current and the next run installs again. A link that cannot be swapped is unpinned, and
+    /// the next full apply attaches it again if it is still wanted.
+    fn install(&self, old_links: &[PathBuf]) -> anyhow::Result<()> {
+        std::fs::create_dir_all(self.links_dir())?;
+        let mut bpf = EbpfLoader::new()
+            .map_pin_path(self.schema_root())
+            .load(self.obj)?;
+        // load everything before touching any link, so a verifier rejection of one direction
+        // leaves the old state whole
+        for &(_, name) in self.programs {
+            let prog: &mut SchedClassifier = bpf
+                .program_mut(name)
+                .with_context(|| format!("{}: program {name} not found in object", self.name))?
+                .try_into()?;
+            prog.load()?;
+        }
+        self.remove_foreign_pins()?;
+
+        let live = self.live_links()?;
+        for &(dir, _) in self.programs {
+            let prog = self.loaded(&mut bpf, dir)?;
+            for pin in live.iter().filter(|pin| pin.dir == dir) {
+                let path = self.link_pin_path(pin);
+                if let Err(e) = tc::swap_pinned_link(prog, &path) {
+                    log::warn!(
+                        "{}: {} swap failed ({e:#}), reclaiming",
+                        self.name,
+                        pin.filename()
+                    );
+                    self.reclaim_link(pin);
+                }
+            }
+        }
+
+        // the links of another schema are rebound the same way and their pins moved over, so
+        // their interfaces never run without a program and keep their place in the tcx order.
+        // One this build has no program for, or that already has a link here, is detached
+        // instead. A rebound link whose pin cannot be moved stays where it is
+        for links_dir in old_links {
+            for pin in tc::read_pinned_links(links_dir)? {
+                let old = links_dir.join(pin.filename());
+                let new = self.link_pin_path(&pin);
+                let migrated = self.program_name(pin.dir).is_some()
+                    && !new.exists()
+                    && match tc::swap_pinned_link(self.loaded(&mut bpf, pin.dir)?, &old) {
+                        Ok(()) => true,
+                        Err(e) => {
+                            log::warn!(
+                                "{}: {} swap failed ({e:#}), reclaiming",
+                                self.name,
+                                pin.filename()
+                            );
+                            false
+                        }
+                    };
+                if migrated {
+                    if let Err(e) = std::fs::rename(&old, &new) {
+                        log::warn!(
+                            "{}: move link pin {} to {}: {e:#}",
+                            self.name,
+                            old.display(),
+                            new.display()
+                        );
+                    }
+                } else if let Err(e) = tc::detach_pinned_link(&old) {
+                    log::warn!("{}: detach old link {}: {e:#}", self.name, old.display());
+                    let _ = std::fs::remove_file(&old);
+                }
+            }
+        }
+
+        // a run that died after pinning some of the directions left a partial set behind
+        match std::fs::remove_dir_all(self.prog_dir()) {
+            Ok(()) => {}
+            Err(e) if e.kind() == ErrorKind::NotFound => {}
+            Err(e) => return Err(e).context("remove partial program pins"),
+        }
+        std::fs::create_dir_all(self.prog_dir())?;
+        for &(dir, _) in self.programs {
+            self.loaded(&mut bpf, dir)?.pin(self.prog_pin_path(dir))?;
+        }
+
+        // the links are on this build now, the other builds' programs go with their pins
+        let mine = self.fingerprint_dir_name();
+        for entry in std::fs::read_dir(self.progs_dir())? {
+            let entry = entry?;
+            if entry.file_name() != mine.as_str()
+                && let Err(e) = std::fs::remove_dir_all(entry.path())
+            {
+                log::warn!(
+                    "{}: remove stale program pins {}: {e:#}",
+                    self.name,
+                    entry.path().display()
+                );
+            }
+        }
+        Ok(())
+    }
+
+    // one direction's program out of the loaded object, resolved once by the load loop
+    fn loaded<'b>(
+        &self,
+        bpf: &'b mut aya::Ebpf,
+        dir: Direction,
+    ) -> anyhow::Result<&'b mut SchedClassifier> {
+        let name = self
+            .program_name(dir)
+            .expect("only directions with a program are loaded");
+        let prog = bpf
+            .program_mut(name)
+            .expect("every program was resolved by the load loop")
+            .try_into()?;
+        Ok(prog)
+    }
+
+    fn live_links(&self) -> anyhow::Result<Vec<tc::LinkPin>> {
+        tc::read_pinned_links(&self.links_dir())
+    }
+
+    // a pin this build cannot read or has no program for is not its state. It would keep a link
+    // running with nothing to update it, so it goes at install time
+    fn remove_foreign_pins(&self) -> anyhow::Result<()> {
+        for entry in std::fs::read_dir(self.links_dir())? {
+            let entry = entry?;
+            let name = entry.file_name();
+            let ours = name
+                .to_str()
+                .and_then(tc::LinkPin::parse)
+                .is_some_and(|pin| self.program_name(pin.dir).is_some());
+            if ours {
+                continue;
+            }
+            let name = name.to_string_lossy();
+            log::warn!("{}: removing pin {name} this build cannot use", self.name);
+            let path = entry.path();
+            if let Err(e) = tc::detach_pinned_link(&path) {
+                log::warn!("{}: detach {name}: {e:#}", self.name);
+                let _ = std::fs::remove_file(&path);
+            }
+        }
+        Ok(())
+    }
+
+    fn detach_named(
+        &self,
+        name: &str,
+        keep: impl Fn(&tc::LinkPin) -> bool,
+    ) -> anyhow::Result<Vec<u32>> {
+        let mut ifindexes = Vec::new();
+        for pin in self.live_links()? {
+            if pin.name != name || keep(&pin) {
+                continue;
+            }
+            self.reclaim_link(&pin);
+            if !ifindexes.contains(&pin.ifindex) {
+                ifindexes.push(pin.ifindex);
+            }
+        }
+        Ok(ifindexes)
+    }
+
+    /// Unpin a link. The pin file goes even when the kernel object behind it cannot be opened
+    /// anymore.
+    fn reclaim_link(&self, pin: &tc::LinkPin) {
+        let path = self.link_pin_path(pin);
+        match tc::detach_pinned_link(&path) {
+            Ok(()) => {}
+            // the other side of a lost race, another attach or unplug took the pin first
+            Err(tc::Error::PinGone) => {
+                log::debug!(
+                    "{}: reclaim {}: pin gone already",
+                    self.name,
+                    pin.filename()
+                );
+            }
+            Err(e) => {
+                log::warn!(
+                    "{}: reclaim {}: unpin stale link: {e:#}",
+                    self.name,
+                    pin.filename()
+                );
+                let _ = std::fs::remove_file(&path);
+            }
+        }
+    }
+
+    fn reconcile(&self, desired: &HashSet<u32>) -> anyhow::Result<()> {
+        // opened before anything is detached, so a missing program fails the full apply before it
+        // changes anything
+        let mut progs = Vec::new();
+        for &(dir, _) in self.programs {
+            progs.push((dir, self.program(dir)?));
+        }
+        let live = self.live_links()?;
+        let mut attached: HashSet<(u32, Direction)> = HashSet::new();
+
+        for pin in &live {
+            // a pin whose index now belongs to another interface links nothing wanted
+            let alive = tc::iface_name(pin.ifindex)?.is_some_and(|name| name == pin.name);
+            if alive && desired.contains(&pin.ifindex) {
+                attached.insert((pin.ifindex, pin.dir));
+                continue;
+            }
+            log::debug!("{}: detach {}", self.name, pin.filename());
+            let path = self.link_pin_path(pin);
+            if let Err(e) = tc::detach_pinned_link(&path) {
+                log::error!("{}: detach {}: {e:#}", self.name, pin.filename());
+                let _ = std::fs::remove_file(&path);
+            } else {
+                log::info!("{}: detached {}", self.name, pin.filename());
+            }
+        }
+
+        let mut failed = 0usize;
+        for (dir, prog) in &mut progs {
+            let dir = *dir;
+            for &ifindex in desired {
+                if attached.contains(&(ifindex, dir)) {
+                    continue;
+                }
+                let Some(name) = tc::iface_name(ifindex)? else {
+                    log::info!("{}: {ifindex} vanished before the attach", self.name);
+                    continue;
+                };
+                let pin = tc::LinkPin { ifindex, name, dir };
+                match tc::attach_and_pin(prog, &pin, &self.link_pin_path(&pin)) {
+                    Ok(()) => {}
+                    Err(tc::Error::IfaceGone) => {
+                        log::info!("{}: {ifindex} vanished before the attach", self.name);
+                    }
+                    Err(e) => {
+                        log::error!("{}: attach {}: {e:#}", self.name, pin.filename());
+                        failed += 1;
+                    }
+                }
+            }
+        }
+        if failed > 0 {
+            anyhow::bail!("{}: {failed} link(s) failed to attach", self.name);
+        }
+        Ok(())
+    }
+
+    fn attach_iface(&self, ifindex: u32) -> anyhow::Result<()> {
+        // opened for every direction ahead of the pin checks, so a missing program fails here,
+        // never counts as a dead link and leaves no direction attached alone
+        let mut progs = Vec::new();
+        for &(dir, _) in self.programs {
+            progs.push((dir, self.program(dir)?));
+        }
+        let Some(name) = tc::iface_name(ifindex)? else {
+            log::info!("{}: {ifindex} vanished before the attach", self.name);
+            return Ok(());
+        };
+        let live = self.live_links()?;
+        for (dir, prog) in &mut progs {
+            let dir = *dir;
+            let pin = tc::LinkPin {
+                ifindex,
+                name: name.clone(),
+                dir,
+            };
+            let path = self.link_pin_path(&pin);
+            for stale in live.iter().filter(|p| p.ifindex == ifindex && p.dir == dir) {
+                if stale.name == name {
+                    match tc::swap_pinned_link(prog, &path) {
+                        Ok(()) => {}
+                        // the pin went away under the swap, another plug of the same interface
+                        // is rebuilding it and the attach below finds its link
+                        Err(tc::Error::PinGone) => {
+                            log::debug!(
+                                "{}: {} pin taken by another attach",
+                                self.name,
+                                path.display()
+                            );
+                            continue;
+                        }
+                        Err(e) => {
+                            log::warn!(
+                                "{}: {} swap failed ({e:#}), rebuilding",
+                                self.name,
+                                pin.filename()
+                            );
+                            self.reclaim_link(stale);
+                            continue;
+                        }
+                    }
+                } else {
+                    // the index was reused since, the old link is not this interface's
+                    self.reclaim_link(stale);
+                }
+            }
+            if path.exists() {
+                continue;
+            }
+            match tc::attach_and_pin(prog, &pin, &path) {
+                Ok(()) => {}
+                Err(tc::Error::IfaceGone) => {
+                    log::info!("{}: {ifindex} vanished before the attach", self.name);
+                }
+                Err(tc::Error::AlreadyAttached) => {
+                    log::debug!("{}: {} attached concurrently", self.name, pin.filename());
+                }
+                Err(e) => {
+                    return Err(e)
+                        .with_context(|| format!("{}: attach {}", self.name, pin.filename()));
+                }
+            }
+        }
+        Ok(())
+    }
+}
+
+fn schema_links_dir(schema_root: &Path) -> PathBuf {
+    schema_root.join("links")
+}
+
+fn schema_progs_dir(schema_root: &Path) -> PathBuf {
+    schema_root.join("prog")
+}
+
+fn links_pinned_under(schemas: &[(PathBuf, Option<u32>)]) -> bool {
+    schemas
+        .iter()
+        .any(|(dir, _)| any_links(&schema_links_dir(dir)))
+}
+
+// whether link pins remain in a links directory. Debris of another kind does not count, it goes
+// with the directory. One that cannot be read may hold anything, a missing one is empty
+fn any_links(links_dir: &Path) -> bool {
+    match tc::read_pinned_links(links_dir) {
+        Ok(links) => !links.is_empty(),
+        Err(_) => true,
+    }
+}
+
+// a program pin is a file under a fingerprint directory. A run killed between creating that
+// directory and pinning leaves it empty. Like the links, a directory that cannot be read may
+// hold anything, a missing one is empty
+fn any_program_pinned(progs_dir: &Path) -> bool {
+    let non_empty = |dir: &Path| match std::fs::read_dir(dir) {
+        Ok(mut entries) => entries.next().is_some(),
+        Err(e) => e.kind() != ErrorKind::NotFound,
+    };
+    match std::fs::read_dir(progs_dir) {
+        Ok(fingerprints) => fingerprints.into_iter().any(|fp| match fp {
+            Ok(fp) => non_empty(&fp.path()),
+            Err(_) => true,
+        }),
+        Err(e) => e.kind() != ErrorKind::NotFound,
+    }
+}
+
+/// Marker for a guard holding the apply lock shared.
+pub struct Shared;
+
+/// Marker for a guard holding the apply lock exclusively.
+pub struct Exclusive;
+
+/// The apply lock of one subsystem, held for as long as the guard lives. Loading and matching
+/// the links to the desired set rewrite pinned state and exist only on the [`Exclusive`]
+/// guard, so nothing runs them next to shared holders. A shared holder adds or drops its own
+/// link and writes map entries that stand on their own, each write atomic in the kernel. A
+/// write that depends on what was read before it takes the exclusive lock, or the kernel's
+/// conditional write, insert only if absent and update only if present. A shared holder cannot
+/// upgrade, one that finds nothing current leaves the install to a full apply under the
+/// exclusive lock.
+pub struct ApplyLock<'a, M> {
+    programs: &'a TcPrograms,
+    _lock: Flock<File>,
+    _mode: PhantomData<M>,
+}
+
+impl<M> ApplyLock<'_, M> {
+    /// Whether the programs and maps of this build are pinned. What a shared holder checks before
+    /// it writes, since it cannot install. Under the lock the program and map pins do not change,
+    /// only links come and go, so the answer holds for the holder's steps.
+    pub fn is_current(&self) -> bool {
+        self.programs.is_current()
+    }
+
+    /// Open a pinned BPF hash map by name. Valid while the programs are pinned, an exclusive
+    /// holder runs [`ensure_loaded`](ApplyLock::ensure_loaded) first.
+    pub fn hash_map<K: aya::Pod, V: aya::Pod>(
+        &self,
+        name: &str,
+    ) -> anyhow::Result<aya::maps::HashMap<aya::maps::MapData, K, V>> {
+        self.programs.hash_map(name)
+    }
+
+    /// Attach the programs to a single interface, leaving the others alone.
+    ///
+    /// If a pin already exists, swap the program in place with no traffic gap. A pin that cannot
+    /// be swapped is reclaimed and the interface attached fresh. Two concurrent attaches of the
+    /// same interface run under the shared lock, the one that loses the race finds the other's
+    /// link in place and is done. A failed attach returns the error, [`NotLoaded`] when nothing
+    /// is pinned yet, and the caller decides what an interface without the program means for it.
+    pub fn attach_iface(&self, ifindex: u32) -> anyhow::Result<()> {
+        self.programs.attach_iface(ifindex)
+    }
+
+    /// Detach the programs from the interface of that name, even when the interface is gone
+    /// already. A pin for which `keep` returns true stays, such as one of a newer interface under
+    /// the same name. Returns the indexes the detached links had.
+    pub fn detach_named(
+        &self,
+        name: &str,
+        keep: impl Fn(&tc::LinkPin) -> bool,
+    ) -> anyhow::Result<Vec<u32>> {
+        self.programs.detach_named(name, keep)
+    }
+
+    /// The links of this subsystem as their pins name them. A shared holder sees a snapshot,
+    /// another attach may add to it meanwhile. The name is not unique. A leftover pin of an
+    /// interface that went away and the pin of its successor under the name share it, only the
+    /// index tells them apart.
+    pub fn attached(&self) -> anyhow::Result<Vec<tc::LinkPin>> {
+        self.programs.live_links()
+    }
+}
+
+impl ApplyLock<'_, Exclusive> {
+    /// Make sure this build's programs and maps are pinned, loading only when they are not. Maps
+    /// keep their entries across a load unless the schema changed, then they start empty, and a
+    /// lost map pin comes back empty. The caller syncs the maps in the same apply.
+    pub fn ensure_loaded(&self) -> anyhow::Result<()> {
+        self.programs.ensure_loaded()
+    }
+
+    /// Make the attached set match `desired`. Refreshing links onto new code is
+    /// [`ensure_loaded`](Self::ensure_loaded)'s job. The live set cannot change under it, every
+    /// other writer waits on the exclusive lock. What may go is decided from the kernel's state.
+    ///
+    /// An interface that failed to attach fails the apply once every other was tried, since the
+    /// subsystem does not work on an interface without its program. One that vanished since it
+    /// was named is no failure, and a failed detach only leaves a program running longer and is
+    /// logged.
+    pub fn reconcile(&self, desired: &HashSet<u32>) -> anyhow::Result<()> {
+        self.programs.reconcile(desired)
+    }
+
+    /// Drop the pinned state of this subsystem, links included. Leaving links attached would keep
+    /// the programs running with nothing left to update them. The lock under `/run` stays,
+    /// unlinking it under a waiter would let that waiter run unserialized.
+    pub fn clear(&self) -> anyhow::Result<()> {
+        self.programs.tear_down()
+    }
+}
diff --git a/src/tc.rs b/src/tc.rs
index 05d823e..63d3fbf 100644
--- a/src/tc.rs
+++ b/src/tc.rs
@@ -17,7 +17,7 @@ use aya::{
 /// Why a link step did not go through, with the kernel's own answer where it gives one, so a
 /// caller can tell a lost race or a vanished interface from a real failure.
 #[derive(Debug)]
-pub enum Error {
+pub(crate) enum Error {
     /// A link of the program is in place already, another attach of the interface won the race.
     AlreadyAttached,
     /// The interface is gone, its index names no netdev anymore.
@@ -176,7 +176,7 @@ impl LinkPin {
 /// verdict ends the chain right there. So a subsystem's program returns `TC_ACT_UNSPEC` for
 /// whatever it leaves alone. Programs of several subsystems run in attach order, each new link
 /// goes last.
-pub fn attach_and_pin(
+pub(crate) fn attach_and_pin(
     prog: &mut SchedClassifier,
     pin: &LinkPin,
     pin_path: &Path,
@@ -201,7 +201,7 @@ pub fn attach_and_pin(
 }
 
 /// Rebind a pinned link to `prog` via LINK_UPDATE. Atomic, traffic sees no detach/reattach gap.
-pub fn swap_pinned_link(prog: &mut SchedClassifier, pin_path: &Path) -> Result<(), Error> {
+pub(crate) fn swap_pinned_link(prog: &mut SchedClassifier, pin_path: &Path) -> Result<(), Error> {
     let pinned = PinnedLink::from_pin(pin_path)?;
     let fd_link: FdLink = pinned.into();
     let link = fd_link.try_into()?;
@@ -211,7 +211,7 @@ pub fn swap_pinned_link(prog: &mut SchedClassifier, pin_path: &Path) -> Result<(
     Ok(())
 }
 
-pub fn detach_pinned_link(pin_path: &Path) -> Result<(), Error> {
+pub(crate) fn detach_pinned_link(pin_path: &Path) -> Result<(), Error> {
     let pinned = PinnedLink::from_pin(pin_path)?;
     let _fd_link = pinned.unpin()?;
     Ok(())
@@ -219,7 +219,7 @@ pub fn detach_pinned_link(pin_path: &Path) -> Result<(), Error> {
 
 /// The name of an interface by index, none when it is gone. Any other failure says nothing
 /// about the interface and is not taken as its absence, so the lookup reads errno itself.
-pub fn iface_name(ifindex: u32) -> anyhow::Result<Option<String>> {
+pub(crate) fn iface_name(ifindex: u32) -> anyhow::Result<Option<String>> {
     let mut buf = [0u8; nix::libc::IF_NAMESIZE];
     // SAFETY: the buffer holds IF_NAMESIZE bytes, the most the call writes
     let name = unsafe { nix::libc::if_indextoname(ifindex, buf.as_mut_ptr().cast()) };
@@ -237,7 +237,7 @@ pub fn iface_name(ifindex: u32) -> anyhow::Result<Option<String>> {
 
 /// The pins in `links_dir`, see [`LinkPin`], none for a missing directory. A file that is not a
 /// pin is skipped with a warning.
-pub fn read_pinned_links(links_dir: &Path) -> anyhow::Result<Vec<LinkPin>> {
+pub(crate) fn read_pinned_links(links_dir: &Path) -> anyhow::Result<Vec<LinkPin>> {
     let mut out = Vec::new();
     let dir = match std::fs::read_dir(links_dir) {
         Ok(d) => d,
-- 
2.47.3