[PATCH proxmox-ebpf v3 1/3] add the shared tc subsystem code

[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 from the
subsystem driving it, and re-running never interrupts traffic. Apply
paths coordinate through a shared flock, loading takes it exclusively. A
link is pinned under its interface's index, name and direction. So an
unplug reported once the interface is gone still finds the pin, and a
reused index does not pass for the old link.

Changes are ordered by a counter the subsystem draws before a change
reads its input. A full apply records the generation it took, and a
later one under an older generation is dropped. A single change carries
its generation in the entry it writes and never overwrites a newer one.
Waits on the locks are bounded, a holder that never returns fails the
waiter instead of hanging it.

Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
 src/lib.rs       |   3 +
 src/subsystem.rs | 901 +++++++++++++++++++++++++++++++++++++++++++++++
 src/tc.rs        | 257 ++++++++++++++
 3 files changed, 1161 insertions(+)
 create mode 100644 src/subsystem.rs
 create mode 100644 src/tc.rs

diff --git a/src/lib.rs b/src/lib.rs
index 244f44c..572d861 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -1,2 +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..efae754
--- /dev/null
+++ b/src/subsystem.rs
@@ -0,0 +1,901 @@
+//! 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
+//! and verifies only when nothing current is pinned, on a schema or object change and after a
+//! lost map pin. Everything else attaches links and syncs maps against what is already there.
+//! Each subsystem owns one [`TcPrograms`].
+//!
+//! A subsystem compiles its desired state outside the lock, so two of its changes can reach the
+//! kernel in the wrong order. A generation from [`TcPrograms::next_generation`] orders them. A
+//! full apply runs through [`ApplyLock::with_generation`] and is dropped when a newer one is in
+//! place. A single-entry change keeps its generation in the entry it writes and compares there,
+//! under [`ApplyLock::change`].
+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";
+
+// a wait for the lock normally lasts one pass. The bound only catches a holder that never
+// returns. flock has no timeout, so the wait asks nonblocking and sleeps in between
+
+const LOCK_WAIT: Duration = Duration::from_secs(10);
+const LOCK_POLL: Duration = Duration::from_millis(10);
+
+fn flock_within(
+    mut file: File,
+    arg: FlockArg,
+    what: &str,
+    subsystem: &str,
+) -> anyhow::Result<Flock<File>> {
+    let nonblocking = match arg {
+        FlockArg::LockShared => FlockArg::LockSharedNonblock,
+        FlockArg::LockExclusive => FlockArg::LockExclusiveNonblock,
+        other => other,
+    };
+    let deadline = Instant::now() + LOCK_WAIT;
+    loop {
+        match Flock::lock(file, nonblocking) {
+            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, so a shared holder hands the work back to its caller.
+#[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 {}
+
+/// The pinned tc programs for one subsystem.
+///
+/// Holds only the immutable description. The loaded BPF lives pinned in the kernel and is reached
+/// through those pins, so normal operation runs no verifier.
+pub struct TcPrograms {
+    /// The subsystem's name, its directory under the pin root.
+    pub name: &'static str,
+    /// The embedded BPF object.
+    pub obj: &'static [u8],
+    /// [`obj_fingerprint`](Self::obj_fingerprint) of `obj`, folded into a `const`.
+    pub fingerprint: u64,
+    /// The name of the object's program for a direction.
+    pub prog_name: fn(Direction) -> &'static str,
+    /// The directions the object carries a program for.
+    pub directions: &'static [Direction],
+    /// The object's pinned maps, they count towards the state being current.
+    pub maps: &'static [&'static str],
+    /// Bumped with any change to a map's key or value layout, a pinned map is reused by name
+    /// with no check beyond its key and value sizes.
+    pub schema_version: u32,
+}
+
+impl TcPrograms {
+    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 {
+        self.schema_root().join("links")
+    }
+    fn link_pin_path(&self, pin: &tc::LinkPin) -> PathBuf {
+        self.links_dir().join(pin.filename())
+    }
+    fn progs_dir(&self) -> PathBuf {
+        self.schema_root().join("prog")
+    }
+    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 run_dir(&self) -> PathBuf {
+        PathBuf::from(RUN_ROOT).join(self.name)
+    }
+    fn generation_path(&self) -> PathBuf {
+        self.run_dir().join("generation")
+    }
+    fn applied_path(&self) -> PathBuf {
+        self.run_dir().join("applied")
+    }
+
+    // the counter and the stamp of the last full apply are plain files next to the apply lock,
+    // read and written under this lock. A single-entry change holds it for its own read, compare
+    // and write too, so changes serialize without giving up the apply lock's sharing
+
+    fn generation_lock(&self) -> anyhow::Result<Flock<File>> {
+        std::fs::create_dir_all(self.run_dir())?;
+        let file = File::options()
+            .create(true)
+            .truncate(false)
+            .write(true)
+            .open(self.run_dir().join("generation.lock"))?;
+        flock_within(file, FlockArg::LockExclusive, "generation lock", self.name)
+    }
+
+    /// Draw the next generation. Draw after the input of a change is complete and before that
+    /// input is read, then two changes are ordered by the input they saw and not by when they
+    /// reached the kernel. A draw before the input was written orders nothing, a newer change may
+    /// then read the older state and win. Restarts with the node like everything it orders.
+    pub fn next_generation(&self) -> anyhow::Result<u64> {
+        let _lock = self.generation_lock()?;
+        let next = read_stamp(&self.generation_path())?.unwrap_or(0) + 1;
+        write_stamp(&self.generation_path(), next)?;
+        Ok(next)
+    }
+
+    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. A plain `BPF_OBJ_GET`, no verifier, since the program
+    /// was checked once when [`ensure_loaded`](ApplyLock::ensure_loaded) installed it.
+    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()))
+    }
+
+    /// FNV-1a over the embedded object. A `const fn`, so each subsystem folds it into a `const` at
+    /// compile time and the per-invocation path never rehashes a constant.
+    pub 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
+    }
+
+    /// The per-subsystem apply lock under `/run`, one file taken in two modes. A full apply and
+    /// any install or teardown take it [exclusively](Self::lock_exclusive), so enumerate, detach
+    /// and attach run as one unit. A single-interface step takes it [shared](Self::lock_shared),
+    /// so guest plugs run concurrently and block only behind a full apply. The kernel drops the
+    /// lock with a dying process. A wait is bounded, a holder that never returns fails the
+    /// waiter instead of hanging it. Shared holders are not held back for a waiting exclusive
+    /// one, a node plugging without pause could keep a full apply waiting until its bound. Every
+    /// write to the pinned state, and the check whether it is current, is a method of the
+    /// returned guard. So a step cannot run outside the lock, and the mode it needs is checked
+    /// at compile time.
+    fn lock<M>(&self, arg: FlockArg) -> anyhow::Result<ApplyLock<'_, M>> {
+        let dir = self.run_dir();
+        std::fs::create_dir_all(&dir)?;
+        let file = File::options()
+            .create(true)
+            .truncate(false)
+            .write(true)
+            .open(dir.join("lock"))?;
+        let lock = flock_within(file, arg, "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 pass. 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| {
+            others.iter().any(|(dir, _)| any_links(&dir.join("links")))
+        });
+        elsewhere
+            || self
+                .live_links()
+                .map_or(true, |links| links.iter().any(|pin| pin.name == name))
+    }
+
+    /// Take the apply lock in shared mode, for an additive single-interface apply.
+    pub fn lock_shared(&self) -> anyhow::Result<ApplyLock<'_, Shared>> {
+        self.lock(FlockArg::LockShared)
+    }
+
+    /// Take the apply lock exclusively, for a full apply or an install/teardown.
+    pub fn lock_exclusive(&self) -> anyhow::Result<ApplyLock<'_, Exclusive>> {
+        self.lock(FlockArg::LockExclusive)
+    }
+
+    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 them there.
+        // This build is not current until the install below has taken them over
+
+        let links_elsewhere = others.iter().any(|(dir, _)| any_links(&dir.join("links")));
+        // a newer schema with programs means a newer build already runs here and this process
+        // is the stale one, it must not touch that. A directory of a newer schema holding no
+        // program is what a rejected install of that build left behind, it goes with the rest
+        if let Some(newer) = others
+            .iter()
+            .filter(|(dir, _)| any_program_pinned(&dir.join("prog")))
+            .filter_map(|(_, schema)| *schema)
+            .find(|&s| s > self.schema_version)
+        {
+            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 is what a run left that died between installing and removing it,
+            // its links moved over already. A newer schema without programs is what a rejected
+            // install left, its maps in the kernel with nothing using them
+
+            for (dir, schema) in others {
+                if (schema.is_some_and(|s| s < self.schema_version)
+                    || !any_program_pinned(&dir.join("prog")))
+                    && 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 pass that half-done state 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.
+        // Its links move onto the new programs while it is still in place, so no interface runs
+        // without one and a verifier rejection leaves it serving. The rest goes afterwards
+        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, the next run swaps it onto its own object again. Without program pins the
+        // directory does not pass for state of that build
+
+        let old_links: Vec<PathBuf> = others.iter().map(|(dir, _)| dir.join("links")).collect();
+        self.install(&old_links).context("install")?;
+        for (dir, _) in &others {
+            // a link whose pin could not be moved is still live there, the next run swaps it
+            // onto its programs again
+            if any_links(&dir.join("links")) {
+                continue;
+            }
+            if let Err(e) = std::fs::remove_dir_all(dir) {
+                log::warn!(
+                    "{}: remove old pinned state {}: {e:#}",
+                    self.name,
+                    dir.display()
+                );
+            }
+        }
+        Ok(())
+    }
+
+    // every direction's program pinned under this build's schema and fingerprint and every map
+    // with them
+    fn is_current(&self) -> bool {
+        self.directions
+            .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 state pinned here under schema versions other than this build's. A directory that is
+    // no version at all is taken along as such state, its links moved over. A plain file is
+    // not state
+    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 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 and verify the object. Move the pinned links onto it, this schema's and those under
+    /// `old_links`. Then pin its programs and drop those of other builds. The verifier runs here
+    /// and nowhere else, a rejection leaves what served before untouched. The programs are pinned
+    /// last since their pins mark this build as current. A failure before that leaves the
+    /// previous build current, and the next run repeats the install. Links moved over already are
+    /// swapped onto its object again. A link the update cannot reach has lost its netdev and is
+    /// unpinned, the next reconcile attaches the interface fresh if still wanted.
+    fn install(&self, old_links: &[PathBuf]) -> anyhow::Result<()> {
+        std::fs::create_dir_all(self.links_dir())?;
+        self.remove_foreign_pins()?;
+        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 &dir in self.directions {
+            let name = (self.prog_name)(dir);
+            let prog: &mut SchedClassifier = bpf
+                .program_mut(name)
+                .with_context(|| format!("{}: program {name} not found in object", self.name))?
+                .try_into()?;
+            prog.load()?;
+        }
+
+        let live = self.live_links()?;
+        for &dir in self.directions {
+            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, the
+        // interface never runs without a program and keeps its 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, the next run finds it again
+        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.directions.contains(&pin.dir)
+                    && !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.directions {
+            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().to_string_lossy() != 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 prog = bpf
+            .program_mut((self.prog_name)(dir))
+            .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().to_string_lossy().into_owned();
+            let ours =
+                tc::LinkPin::parse(&name).is_some_and(|pin| self.directions.contains(&pin.dir));
+            if ours {
+                continue;
+            }
+            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(())
+    }
+
+    // every link pinned under the interface name goes, whatever index it had, unless `keep`
+    // answers for it. An interface that was never attached is done. Returns the indexes the
+    // detached links had
+
+    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 defunct link. The pin file goes even when the kernel object behind it cannot be
+    /// opened any more.
+    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(e) if tc::pin_missing(&e) => {
+                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, a missing program fails the pass whole
+        let mut progs = Vec::new();
+        for &dir in self.directions {
+            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 is not this interface's anymore is not a link of anything
+            // wanted, whatever its index says
+
+            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(e) if e.downcast_ref::<tc::IfaceGone>().is_some() => {
+                        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.directions {
+            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(e) if tc::pin_missing(&e) => {
+                            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(e) if e.downcast_ref::<tc::IfaceGone>().is_some() => {
+                    log::info!("{}: {ifindex} vanished before the attach", self.name);
+                }
+                Err(e) if e.downcast_ref::<tc::PinExists>().is_some() => {
+                    log::debug!("{}: {} attached concurrently", self.name, pin.filename());
+                }
+                Err(e) => {
+                    return Err(e)
+                        .with_context(|| format!("{}: attach {}", self.name, pin.filename()));
+                }
+            }
+        }
+        Ok(())
+    }
+
+    // the generation lock is held for the stamp check and write only, the pass itself runs
+    // under the exclusive apply lock, which already keeps every other writer out
+    fn with_generation<R>(
+        &self,
+        generation: u64,
+        f: impl FnOnce() -> anyhow::Result<R>,
+    ) -> anyhow::Result<Option<R>> {
+        {
+            let _lock = self.generation_lock()?;
+            if let Some(last) = read_stamp(&self.applied_path())?
+                && last >= generation
+            {
+                log::info!(
+                    "{}: full apply of generation {generation} is older than the applied \
+                     {last}, dropping it",
+                    self.name
+                );
+                return Ok(None);
+            }
+            write_stamp(&self.applied_path(), generation)?;
+        }
+        Ok(Some(f()?))
+    }
+
+    fn change<R>(&self, f: impl FnOnce(Option<u64>) -> anyhow::Result<R>) -> anyhow::Result<R> {
+        let _lock = self.generation_lock()?;
+        f(read_stamp(&self.applied_path())?)
+    }
+}
+
+// whether link pins remain in a links directory, only those count, debris of another kind 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, an empty fingerprint directory is
+// what a run killed between creating it and pinning left behind. 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,
+    }
+}
+
+fn read_stamp(path: &Path) -> anyhow::Result<Option<u64>> {
+    match std::fs::read_to_string(path) {
+        Ok(s) => s
+            .trim()
+            .parse()
+            .map(Some)
+            .with_context(|| format!("stamp {} holds no number", path.display())),
+        Err(e) if e.kind() == ErrorKind::NotFound => Ok(None),
+        Err(e) => Err(e.into()),
+    }
+}
+
+// written beside and renamed over, so a stamp is either the old one or the new one
+fn write_stamp(path: &Path, stamp: u64) -> anyhow::Result<()> {
+    let tmp = path.with_extension("tmp");
+    std::fs::write(&tmp, stamp.to_string())?;
+    Ok(std::fs::rename(&tmp, path)?)
+}
+
+/// 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 the
+/// reconcile of links 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 its
+/// own map entries. There is no upgrade. A shared holder that finds nothing current hands off,
+/// the full pass takes 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, for the owning subsystem to sync. Valid while the
+    /// programs are pinned. An exclusive holder runs [`ensure_loaded`](ApplyLock::ensure_loaded)
+    /// first, a shared one checks [`is_current`](ApplyLock::is_current) and hands off when it is
+    /// not.
+    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, additively, without touching others.
+    ///
+    /// If a pin already exists, swap the program in place with no traffic gap. The swap only
+    /// succeeds while the link is still on a live netdev, a defunct pin is reclaimed and the
+    /// interface attached fresh. Two concurrent attaches of the same interface run under the
+    /// shared lock, the one that loses the pin race finds the other's link in place and is done.
+    /// A failed attach propagates so the caller can decide what an interface without the program
+    /// means for it, [`NotLoaded`] among them when nothing is pinned yet.
+    pub fn attach_iface(&self, ifindex: u32) -> anyhow::Result<()> {
+        self.programs.attach_iface(ifindex)
+    }
+
+    /// Detach the programs from the interface of that name. The pin goes even when the interface
+    /// is gone already, which it is for an unplug reported after the fact. A pin `keep` answers
+    /// true for stays, for a later incarnation of the name that is not the one the unplug is
+    /// about. Returns the indexes the detached links had, for state keyed by them.
+    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 plug may add to it meanwhile. The name is not unique. The leftover pin of a
+    /// crashed guest and its restarted interface share it, only the index tells them apart.
+    pub fn attached(&self) -> anyhow::Result<Vec<tc::LinkPin>> {
+        self.programs.live_links()
+    }
+
+    /// Run `f` as a single-entry change under the lock every such change takes. Its read of the
+    /// entry, the compare of generations and its write are then one step against another change
+    /// of the same entry. `f` gets the generation of the last full apply. A change older than
+    /// that was in the input the apply took, whether the apply went through or not. The
+    /// subsystem decides what such a change may still write. Keep `f` to the map accesses of one
+    /// change.
+    pub fn change<R>(&self, f: impl FnOnce(Option<u64>) -> anyhow::Result<R>) -> anyhow::Result<R> {
+        self.programs.change(f)
+    }
+}
+
+impl ApplyLock<'_, Exclusive> {
+    /// Run `f` as a full apply if `generation` is newer than the last one. The generation is
+    /// recorded before `f` runs, so a state compiled from older input than one already in place
+    /// is dropped and `None` returned. What single-entry changes wrote since `f` read its input is
+    /// `f`'s to leave alone, by the generation each entry carries. The record marks the input as
+    /// taken, not the state as complete. A change older than it is dropped even when `f` failed,
+    /// its input was this apply's and the next apply carries it. A failed apply is reported and
+    /// healed by the next one under a new generation. A retry under the same one is dropped as
+    /// applied.
+    pub fn with_generation<R>(
+        &self,
+        generation: u64,
+        f: impl FnOnce() -> anyhow::Result<R>,
+    ) -> anyhow::Result<Option<R>> {
+        self.programs.with_generation(generation, f)
+    }
+
+    /// Make sure the programs are loaded and pinned. The load runs when nothing current is
+    /// pinned, on a schema change (a rebuild), on a lost map pin, or on an object change (a
+    /// refresh). Otherwise what an earlier run pinned is reused untouched. A rebuild starts with
+    /// empty maps. A refresh keeps every map with its entries. A lost map pin brings only that
+    /// map back empty. The caller syncs all of them in the same pass.
+    pub fn ensure_loaded(&self) -> anyhow::Result<()> {
+        self.programs.ensure_loaded()
+    }
+
+    /// Make the attached set match `desired`. Detach what is no longer wanted, attach what is
+    /// missing. 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. Links
+    /// carry no generation, what may go is decided from the kernel's state.
+    ///
+    /// An interface that failed to attach fails the apply once every other was tried, one left
+    /// without its program is the subsystem not doing its job there. 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, so retiring the subsystem on a node
+    /// ends here. The lock and the generations under `/run` stay. Unlinking the lock under a waiter
+    /// would let that waiter run unserialized, and the generations keep ordering the changes that
+    /// follow.
+    pub fn clear(&self) -> anyhow::Result<()> {
+        self.programs.tear_down()
+    }
+}
diff --git a/src/tc.rs b/src/tc.rs
new file mode 100644
index 0000000..71c0ac7
--- /dev/null
+++ b/src/tc.rs
@@ -0,0 +1,257 @@
+//! TC link plumbing shared by all subsystems. A `Direction` enum, attach/swap/detach free
+//! functions over tcx links (kernel 6.6 and newer), and a uniform pin-filename layout, see
+//! [`LinkPin`].
+
+use std::{io::ErrorKind, path::Path, str::FromStr};
+
+use anyhow::Context;
+use aya::{
+    pin::PinError,
+    programs::{
+        SchedClassifier, TcAttachType,
+        links::{FdLink, LinkError, LinkOrder, PinnedLink},
+        tc::TcAttachOptions,
+    },
+};
+
+/// An attach that found its pin already present, so a caller racing another attach of the same
+/// interface can tell that outcome from a real failure.
+#[derive(Debug)]
+pub struct PinExists;
+
+impl std::fmt::Display for PinExists {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str("pin path already exists")
+    }
+}
+
+impl std::error::Error for PinExists {}
+
+/// An attach whose interface is gone, the index names no netdev any more.
+#[derive(Debug)]
+pub struct IfaceGone;
+
+impl std::fmt::Display for IfaceGone {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str("interface is gone")
+    }
+}
+
+impl std::error::Error for IfaceGone {}
+
+#[derive(Copy, Clone, Hash, PartialEq, Eq, Debug)]
+pub enum Direction {
+    Ingress,
+    Egress,
+}
+
+impl Direction {
+    pub fn as_str(self) -> &'static str {
+        match self {
+            Self::Ingress => "ingress",
+            Self::Egress => "egress",
+        }
+    }
+    pub fn aya_type(self) -> TcAttachType {
+        match self {
+            Self::Ingress => TcAttachType::Ingress,
+            Self::Egress => TcAttachType::Egress,
+        }
+    }
+}
+
+impl FromStr for Direction {
+    type Err = ();
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        match s {
+            "ingress" => Ok(Self::Ingress),
+            "egress" => Ok(Self::Egress),
+            _ => Err(()),
+        }
+    }
+}
+
+/// One pinned link as its pin file names it. The interface name is part of the file name, so an
+/// unplug reported after the interface is gone still finds the pin and a reused index does not
+/// pass for the old link. The colon separates, the kernel forbids it in interface names. It also
+/// stands in for the dot inside the name, which bpffs refuses in a file name.
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub struct LinkPin {
+    pub ifindex: u32,
+    pub name: String,
+    pub dir: Direction,
+}
+
+impl LinkPin {
+    pub fn filename(&self) -> String {
+        format!(
+            "{}:{}:{}",
+            self.ifindex,
+            self.dir.as_str(),
+            self.name.replace('.', ":")
+        )
+    }
+
+    /// The pin a file name stands for, none when it is not one of ours.
+    pub fn parse(filename: &str) -> Option<Self> {
+        let mut parts = filename.splitn(3, ':');
+        let ifindex = parts.next()?.parse().ok()?;
+        let dir = parts.next()?.parse().ok()?;
+        let name = parts.next()?.replace(':', ".");
+        let pin = LinkPin { ifindex, name, dir };
+        // a name that parses but is not the one written for that link is not ours, and no
+        // interface carries what the kernel refuses in a name
+        let valid_name = !pin.name.is_empty()
+            && pin.name.len() < 16
+            && pin.name != "."
+            && pin.name != ".."
+            && !pin.name.chars().any(|c| c.is_whitespace() || c == '/');
+        (pin.filename() == filename && valid_name).then_some(pin)
+    }
+}
+
+/// Attach `prog` in `dir` as a tcx link and pin it. tcx (kernel 6.6 and newer) runs its
+/// programs on the netdev hook ahead of the interface's qdisc filters, among them the `ingress`
+/// qdisc the rate limiter installs. `TC_ACT_UNSPEC` hands a packet on to the next program and
+/// then to the filters. Any other 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(
+    prog: &mut SchedClassifier,
+    pin: &LinkPin,
+    pin_path: &Path,
+) -> anyhow::Result<()> {
+    let link_id = prog
+        .attach_with_options(
+            &pin.name,
+            pin.dir.aya_type(),
+            TcAttachOptions::TcxOrder(LinkOrder::last()),
+        )
+        .map_err(|e| {
+            // the interface can go away between the name lookup and the attach
+            if matches!(iface_name(pin.ifindex), Ok(None)) {
+                anyhow::Error::from(IfaceGone)
+            } else {
+                e.into()
+            }
+        })?;
+    let link = prog.take_link(link_id)?;
+    let fd_link: FdLink = link.try_into()?;
+    // a lost race against another attach of the same interface shows up here, the unpinned link
+    // detaches again when its fd drops
+    match fd_link.pin(pin_path) {
+        Ok(_) => Ok(()),
+        Err(PinError::SyscallError(e)) if e.io_error.kind() == ErrorKind::AlreadyExists => {
+            Err(PinExists.into())
+        }
+        Err(e) => Err(e.into()),
+    }
+}
+
+/// Whether a swap or an unpin failed because the pin was gone by the time it was opened or
+/// removed.
+pub fn pin_missing(e: &anyhow::Error) -> bool {
+    matches!(
+        e.downcast_ref::<LinkError>(),
+        Some(LinkError::SyscallError(s)) if s.io_error.kind() == ErrorKind::NotFound
+    ) || e
+        .downcast_ref::<std::io::Error>()
+        .is_some_and(|e| e.kind() == ErrorKind::NotFound)
+}
+
+/// 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) -> anyhow::Result<()> {
+    let pinned = PinnedLink::from_pin(pin_path)?;
+    let fd_link: FdLink = pinned.into();
+    let link = fd_link.try_into()?;
+    let new_id = prog.attach_to_link(link)?;
+    // take the handle out of aya's internal tracking, we have the pin
+    let _ = prog.take_link(new_id)?;
+    Ok(())
+}
+
+pub fn detach_pinned_link(pin_path: &Path) -> anyhow::Result<()> {
+    let pinned = PinnedLink::from_pin(pin_path)?;
+    let _fd_link = pinned.unpin()?;
+    Ok(())
+}
+
+/// 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>> {
+    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()) };
+    if name.is_null() {
+        return match nix::errno::Errno::last() {
+            nix::errno::Errno::ENXIO => Ok(None),
+            e => Err(e).with_context(|| format!("name of interface {ifindex}")),
+        };
+    }
+    let name = std::ffi::CStr::from_bytes_until_nul(&buf)?;
+    Ok(Some(name.to_string_lossy().into_owned()))
+}
+
+/// Read and parse every pin file in `links_dir`, see [`LinkPin`]. Unrecognized names are skipped
+/// with a warning.
+pub 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,
+        Err(e) if matches!(e.kind(), ErrorKind::NotFound | ErrorKind::NotADirectory) => {
+            return Ok(out);
+        }
+        Err(e) => return Err(e.into()),
+    };
+    for entry in dir {
+        let entry = entry?;
+        let name = entry.file_name();
+        let name = name.to_string_lossy();
+        match LinkPin::parse(&name) {
+            Some(pin) => out.push(pin),
+            None => log::warn!("unrecognized pin file {name} in {}", links_dir.display()),
+        }
+    }
+    Ok(out)
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+
+    #[test]
+    fn pin_names_round_trip() {
+        for name in ["tap100i0", "veth100i31", "vmbr0.100", "bond0.5.7"] {
+            let pin = LinkPin {
+                ifindex: 42,
+                name: name.into(),
+                dir: Direction::Ingress,
+            };
+            assert!(!pin.filename().contains('.'), "{}", pin.filename());
+            assert_eq!(LinkPin::parse(&pin.filename()), Some(pin));
+        }
+        assert_eq!(
+            LinkPin::parse("7:egress:eno1").map(|p| p.name),
+            Some("eno1".into())
+        );
+    }
+
+    #[test]
+    fn pin_names_are_canonical() {
+        for bad in [
+            "042:ingress:tap0",
+            "+42:ingress:tap0",
+            "42:ingress:",
+            "42:ingress",
+            "42:Ingress:tap0",
+            "42-ingress",
+            "4294967296:ingress:tap0",
+            "42:ingress:tap 0",
+            "42:ingress::",
+            "42:ingress:::",
+            "42:ingress:sixteencharsname",
+        ] {
+            assert_eq!(LinkPin::parse(bad), None, "{bad}");
+        }
+    }
+}
-- 
2.47.3