[PATCH proxmox-ebpf v5 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 writes its maps entry by entry through the live maps,
or hands over the whole content. For the latter, and whenever its
own programs and maps are not pinned yet, the object is loaded with
fresh maps in a staging directory, the content is written there, and
only then do the links move onto the new programs, one link update
each, with the maps coming along since a program holds its maps. A
write that stops halfway, an insert fails when the kernel cannot
allocate for the map, leaves the live maps whole instead of a mix of
old and new entries. The pins of maps and programs switch in one
rename, an exchange of the staging directory with the live one, so a
plug always finds the maps its program reads.

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 | 850 +++++++++++++++++++++++++++++++++++++++++++++++
 src/tc.rs        |  12 +-
 3 files changed, 857 insertions(+), 6 deletions(-)
 create mode 100644 src/subsystem.rs

diff --git a/src/lib.rs b/src/lib.rs
index 8d4420a8245d..572d8618661d 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 000000000000..8233faa31567
--- /dev/null
+++ b/src/subsystem.rs
@@ -0,0 +1,850 @@
+//! 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>/`, the links in
+//! `links/` and the maps and programs under `<schema>/live/`. 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. [`ApplyLock::replace_maps`] loads the
+//! object with fresh maps and moves the links onto it, a full apply calls it when what is pinned
+//! is not this build's, when the last replacement stopped halfway, or whenever it wants the
+//! content of its maps switched as a whole. Every other step works on what is 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,
+    os::unix::fs::{DirBuilderExt, OpenOptionsExt},
+    path::{Path, PathBuf},
+    time::{Duration, Instant},
+};
+
+use anyhow::Context;
+use aya::{EbpfLoader, programs::SchedClassifier};
+use nix::{
+    errno::Errno,
+    fcntl::{Flock, FlockArg, RenameFlags, renameat2},
+};
+
+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 takes a read-only descriptor, so nobody but root gets to open the lock files
+fn private_dir(dir: &Path) -> std::io::Result<()> {
+    std::fs::DirBuilder::new()
+        .recursive(true)
+        .mode(0o700)
+        .create(dir)
+}
+
+// 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)
+        .mode(0o600)
+        .open(path)
+        .with_context(|| format!("{subsystem}: open {what} {}", path.display()))?;
+    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)
+                    .with_context(|| format!("{subsystem}: take {what} {}", path.display()));
+            }
+        }
+    }
+}
+
+/// A step that needs the programs pinned found none. Only [`ApplyLock::replace_maps`] under the
+/// exclusive lock brings them up.
+#[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. Each schema has its own live set,
+    /// so a step never writes into maps of another layout, and a build leaves the state of a
+    /// newer schema alone.
+    schema_version: u32,
+}
+
+impl TcPrograms {
+    /// What [`tc_programs!`](crate::tc_programs) expands to, with the fingerprint derived at
+    /// compile time.
+    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 live_dir(&self) -> PathBuf {
+        schema_live_dir(&self.schema_root())
+    }
+    fn staging_dir(&self) -> PathBuf {
+        self.schema_root().join("staging")
+    }
+    fn links_dir(&self) -> PathBuf {
+        links_dir_of(self.name)
+    }
+    fn link_pin_path(&self, pin: &tc::LinkPin) -> PathBuf {
+        self.links_dir().join(pin.filename())
+    }
+    fn fingerprint_dir_name(&self) -> String {
+        format!("{:016x}", self.fingerprint)
+    }
+    // the fingerprint directory of a set of maps and programs, the live one or a staged one
+    fn prog_dir_in(&self, set: &Path) -> PathBuf {
+        set.join("prog").join(self.fingerprint_dir_name())
+    }
+    fn prog_dir(&self) -> PathBuf {
+        self.prog_dir_in(&self.live_dir())
+    }
+    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>> {
+        self.hash_map_in(&self.live_dir(), name)
+    }
+
+    fn hash_map_in<K: aya::Pod, V: aya::Pod>(
+        &self,
+        set: &Path,
+        name: &str,
+    ) -> anyhow::Result<aya::maps::HashMap<aya::maps::MapData, K, V>> {
+        let path = set.join(name);
+        if !path.exists() {
+            // a set comes up with every map the object pins, so an existing set lacks only a map
+            // the object never had under that name
+            if set.exists() {
+                anyhow::bail!("{}: no map {name} pinned in {}", self.name, set.display());
+            }
+            return Err(NotLoaded.into());
+        }
+        let map = aya::maps::MapData::from_pin(path)
+            .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 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();
+        private_dir(&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.
+    pub fn link_pinned(&self, name: &str) -> bool {
+        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 or a teardown.
+    pub fn lock_exclusive(&self) -> anyhow::Result<ApplyLock<'_, Exclusive>> {
+        self.lock(true)
+    }
+
+    fn is_current(&self) -> bool {
+        self.programs
+            .iter()
+            .all(|&(dir, _)| self.prog_pin_path(dir).exists())
+            && self.maps_pinned()
+            // a failed listing cannot rule a newer schema out
+            && matches!(self.newer_schema_installed(), Ok(None))
+    }
+
+    fn maps_pinned(&self) -> bool {
+        self.maps
+            .iter()
+            .all(|map| self.live_dir().join(map).exists())
+    }
+
+    // the directories under the pin root other than this build's schema and the links. One whose
+    // name is not a version counts as well and goes with the old schemas
+    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() || entry.file_name() == LINKS_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) -> anyhow::Result<Option<u32>> {
+        let newer = self
+            .other_schemas()?
+            .iter()
+            .filter(|(dir, _)| any_program_pinned(&schema_progs_dir(dir)))
+            .filter_map(|(_, schema)| *schema)
+            .find(|&s| s > self.schema_version);
+        Ok(newer)
+    }
+
+    fn interrupted(&self) -> bool {
+        let other_pinned = match self.other_schemas() {
+            Ok(others) => others
+                .iter()
+                .any(|(dir, _)| any_program_pinned(&schema_progs_dir(dir))),
+            // a failed listing cannot rule a leftover out
+            Err(_) => true,
+        };
+        self.staging_dir().exists() || other_pinned
+    }
+
+    fn tear_down(&self) -> anyhow::Result<()> {
+        Ok(remove_dir_if_present(&self.pin_root())?)
+    }
+
+    // this build's object with all of its programs loaded and its maps created and pinned under
+    // `set`. Everything is loaded before any link is touched, so a verifier rejection of one
+    // direction leaves the old state whole
+    fn load(&self, set: &Path) -> anyhow::Result<aya::Ebpf> {
+        let mut bpf = EbpfLoader::new()
+            .map_pin_path(set)
+            .load(self.obj)
+            .with_context(|| format!("{}: load the object", self.name))?;
+        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()
+                .with_context(|| format!("{}: load program {name}", self.name))?;
+        }
+        Ok(bpf)
+    }
+
+    // move every pinned link onto the loaded programs. One that cannot be swapped is unpinned,
+    // the next full apply attaches it again if it is still wanted
+    fn swap_links(&self, bpf: &mut aya::Ebpf) -> anyhow::Result<()> {
+        let live = self.live_links()?;
+        for &(dir, _) in self.programs {
+            let prog = self.loaded(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);
+                }
+            }
+        }
+        Ok(())
+    }
+
+    fn pin_programs(&self, bpf: &mut aya::Ebpf, set: &Path) -> anyhow::Result<()> {
+        let dir = self.prog_dir_in(set);
+        std::fs::create_dir_all(&dir)?;
+        for &(direction, _) in self.programs {
+            self.loaded(bpf, direction)?
+                .pin(dir.join(direction.as_str()))
+                .with_context(|| format!("{}: pin program {}", self.name, direction.as_str()))?;
+        }
+        Ok(())
+    }
+
+    // load the object with fresh maps under `staging`, have `fill` write them, then put the
+    // staged set in the live one's place and move the links over. That is one rename, an
+    // exchange when a live set is there, so a plug never finds the maps of one set next to the
+    // programs of the other, and a failure before it leaves the live set untouched. The retired
+    // set and the directories of other schemas go last, their programs stay alive through the
+    // links until each is swapped
+    fn replace_maps(
+        &self,
+        fill: impl FnOnce(&Staged<'_>) -> anyhow::Result<()>,
+    ) -> anyhow::Result<()> {
+        let others = self.other_schemas()?;
+        // 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
+        if let Some(newer) = self.newer_schema_installed()? {
+            anyhow::bail!(
+                "{}: pinned state is of schema {newer}, newer than this build's {}, clear it to \
+                 downgrade",
+                self.name,
+                self.schema_version
+            );
+        }
+        std::fs::create_dir_all(self.links_dir())?;
+        let staging = self.staging_dir();
+        // what a run that died left behind, an unfinished fill or the set it retired
+        remove_dir_if_present(&staging)
+            .with_context(|| format!("{}: remove the previous staging directory", self.name))?;
+        let live = self.live_dir();
+        let staged = self.stage(&staging, fill).and_then(|bpf| {
+            if live.exists() {
+                renameat2(None, &staging, None, &live, RenameFlags::RENAME_EXCHANGE).with_context(
+                    || format!("{}: exchange the staged set with the live one", self.name),
+                )?;
+            } else {
+                std::fs::rename(&staging, &live)
+                    .with_context(|| format!("{}: move the staged set into place", self.name))?;
+            }
+            Ok(bpf)
+        });
+        let mut bpf = match staged {
+            Ok(bpf) => bpf,
+            Err(e) => {
+                // the pins would keep the fresh maps alive
+                if let Err(err) = std::fs::remove_dir_all(&staging) {
+                    log::warn!("{}: remove the staged set: {err:#}", self.name);
+                }
+                return Err(e);
+            }
+        };
+        self.remove_foreign_pins()?;
+        self.swap_links(&mut bpf)?;
+        if let Err(e) = remove_dir_if_present(&staging) {
+            log::warn!("{}: remove the retired set: {e:#}", self.name);
+        }
+        for (dir, _) in others {
+            if let Err(e) = std::fs::remove_dir_all(&dir) {
+                log::warn!(
+                    "{}: remove old pinned state {}: {e:#}",
+                    self.name,
+                    dir.display()
+                );
+            }
+        }
+        Ok(())
+    }
+
+    // the fresh set under `staging`, its maps written by `fill` and its programs pinned
+    fn stage(
+        &self,
+        staging: &Path,
+        fill: impl FnOnce(&Staged<'_>) -> anyhow::Result<()>,
+    ) -> anyhow::Result<aya::Ebpf> {
+        std::fs::create_dir_all(staging)?;
+        let mut bpf = self.load(staging)?;
+        fill(&Staged { programs: self })?;
+        self.pin_programs(&mut bpf, staging)?;
+        Ok(bpf)
+    }
+
+    // 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 with the next replacement
+    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::info!("{}: detach {}", self.name, pin.filename());
+            self.reclaim_link(pin);
+        }
+
+        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(())
+    }
+}
+
+const LINKS_DIR: &str = "links";
+
+fn links_dir_of(subsystem: &str) -> PathBuf {
+    PathBuf::from(PIN_ROOT).join(subsystem).join(LINKS_DIR)
+}
+
+fn schema_live_dir(schema_root: &Path) -> PathBuf {
+    schema_root.join("live")
+}
+
+fn schema_progs_dir(schema_root: &Path) -> PathBuf {
+    schema_live_dir(schema_root).join("prog")
+}
+
+fn remove_dir_if_present(dir: &Path) -> std::io::Result<()> {
+    match std::fs::remove_dir_all(dir) {
+        Err(e) if e.kind() != ErrorKind::NotFound => Err(e),
+        _ => Ok(()),
+    }
+}
+
+// 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. Replacing the maps 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. A shared holder cannot upgrade, one that finds nothing current
+/// leaves bringing the set up 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 this build's programs and maps are pinned and no newer schema is. A shared holder
+    /// checks this before it writes, since only a full apply brings the pins up and an older build
+    /// stays out of a newer schema. 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()
+    }
+
+    /// Whether a replacement stopped halfway. Its staged set or the directories of other schemas
+    /// are still there, so the links may still run the programs it retired.
+    pub fn interrupted(&self) -> bool {
+        self.programs.interrupted()
+    }
+
+    /// Open one of this build's live maps by name, [`NotLoaded`] when it is not pinned.
+    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> {
+    /// Bring this build's programs and maps up, or replace the content of the maps when they are
+    /// up already, without a program ever reading a mix of old and new entries. A first install,
+    /// a new build and a schema change are all this one call. `fill` writes the whole desired
+    /// content into fresh maps, and only once it returned do the programs holding them take over
+    /// the links, one link at a time. A failure inside `fill`, a refused entry included, leaves
+    /// what is pinned untouched. Nothing written into the live maps carries over, a fill that
+    /// wants some of it reads it through [`hash_map`](Self::hash_map), which answers
+    /// [`NotLoaded`] when nothing of this build is pinned. Pinned state of a newer schema is
+    /// refused, a downgrade starts with [`clear`](Self::clear).
+    pub fn replace_maps(
+        &self,
+        fill: impl FnOnce(&Staged<'_>) -> anyhow::Result<()>,
+    ) -> anyhow::Result<()> {
+        self.programs.replace_maps(fill)
+    }
+
+    /// Make the attached set match `desired`. Moving links onto new programs is
+    /// [`replace_maps`](Self::replace_maps)'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()
+    }
+}
+
+/// The fresh maps of a [`replace_maps`](ApplyLock::replace_maps), seen only by the fill that
+/// writes them.
+pub struct Staged<'a> {
+    programs: &'a TcPrograms,
+}
+
+impl Staged<'_> {
+    /// Open one of the staged maps by name, empty until the fill writes it.
+    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_in(&self.programs.staging_dir(), name)
+    }
+}
diff --git a/src/tc.rs b/src/tc.rs
index 1217703e8906..b66d5ab56269 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.
@@ -182,7 +182,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,
@@ -211,7 +211,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()?;
@@ -221,7 +221,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(())
@@ -229,7 +229,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()) };
@@ -247,7 +247,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.
-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