AF_XDP implementation per ADR-002 (3 commits)

.docs/operations.md

@@ -26,7 +26,7 @@ The LB node uses two configuration files:
 | `data_iface` | string | -- | NIC used for packet forwarding |
 | `num_threads` | integer | `7` | Number of packet rewriter threads |
 | `metrics_addr` | string | `127.0.0.1:9100` | Bind address for `/healthz`, `/readyz`, `/metrics`. Use `0.0.0.0:9100` for cross-host scrape |
-| `io_backend` | string | `mock` | `"mock"` (in-memory queues, dev/test) or `"af_xdp"` (production direction; scaffold today — returns `Unsupported` at startup) |
+| `io_backend` | string | `mock` | `"mock"` (in-memory queues, dev/test) or `"af_xdp"` (production). AF_XDP requires building with `--features af-xdp` and running `deploy/xdp/load.sh <iface>` once on the host before startup; see [Loading the XDP redirect program](#loading-the-xdp-redirect-program-af_xdp-only). |
 | `cpu_affinity` | table | unset | Optional per-role CPU pinning (`steering`, `rewriters`, `muxer`). See [CPU pinning](#cpu-pinning). |
 
 #### `[bgp]`
@@ -357,6 +357,38 @@ Releases are cut by pushing a `v*` tag to the repo; see
 aarch64 tarballs containing `lb-node` + `lb-trace` + the unit file, plus
 a multi-arch container image).
 
+### Loading the XDP redirect program (AF_XDP only)
+
+`lb-node` on the `af_xdp` backend depends on a kernel-side XDP program
+that redirects packets into the userspace XSK ring. Build and attach it
+once per host before starting `lb-node`:
+
+```bash
+# One-time per host — installs clang + libbpf headers
+sudo apt install clang libbpf-dev bpftool
+
+# Build the redirect program (deploy/xdp/redirect.bpf.o is gitignored)
+make -C deploy/xdp
+
+# Attach to the data interface in native XDP mode
+sudo deploy/xdp/load.sh eth0 native
+```
+
+The script attaches the program, pins the XSKMAP at
+`/sys/fs/bpf/lb/xsks_map`, and prints next steps. `lb-node` opens the
+pinned map at startup and inserts its own XSK fd at index `queue_id`
+(per `[node].cpu_affinity.rewriters`); no manual `bpftool map update`
+is needed.
+
+To detach (e.g. before swapping to a different program):
+
+```bash
+sudo deploy/xdp/unload.sh eth0
+```
+
+`mock` backend deployments don't need any of this — skip straight to
+the systemd setup below.
+
 ### Systemd
 
 A ready-to-use unit file is at `deploy/lb-node.service`. The service runs as a dedicated unprivileged `lb-node` user with capabilities granted explicitly — create the user first, then install the unit:
@@ -379,7 +411,23 @@ The unit file:
 
 ### Thread layout
 
-The LB node spawns the following threads:
+The LB node uses one of two threading models depending on the I/O backend:
+
+**Per-queue (AF_XDP)**: each rewriter owns its own AF_XDP socket bound
+to one kernel queue. The kernel's RSS hardware hash partitions traffic
+across queues, so userspace steering would just add latency. No
+inter-thread queues, no `PacketPool`; each thread runs `recv_batch →
+process_batch → send_batch` inline.
+
+| Thread | Name | Role |
+|--------|------|------|
+| Rewriter 0..N | `lb-rewriter-{i}` | Bound to queue *i*. RX, conn-table lookup, Maglev, GRE encap, TX. |
+| Config watcher | `lb-config-watcher` | Watch config file, trigger reload |
+| Main | -- | Monitors thread health, signal handling |
+
+**Steering+muxer (mock IO, tests)**: a single shared NIC handle for RX
+and TX, with userspace steering distributing 4-byte frame indices to
+per-rewriter SPSC queues.
 
 | Thread | Name | Role |
 |--------|------|------|
@@ -389,7 +437,7 @@ The LB node spawns the following threads:
 | Config watcher | `lb-config-watcher` | Watch config file, trigger reload |
 | Main | -- | Monitors thread health, signal handling |
 
-SPSC queues between threads carry 4-byte frame indices (`FrameIndex = u32`), not 2KB packet buffers. Packet data lives in a shared `PacketPool` arena and is mutated in-place by whichever thread holds the index. This eliminates the ~166ns/packet memcpy overhead that dominated inter-thread handoff with full `PacketBuf` queues.
+SPSC queues in the steering+muxer model carry 4-byte frame indices, not 2KB packet buffers, eliminating the ~166ns/packet memcpy overhead.
 
 #### CPU pinning
 

.gitignore

@@ -20,3 +20,6 @@ Thumbs.db
 .env
 .env.*
 .claude/
+
+# XDP build artifact (regenerated via `make -C deploy/xdp`)
+deploy/xdp/*.o

Cargo.toml

@@ -87,8 +87,12 @@ bytes = "1"
 # Hashing
 xxhash-rust = { version = "0.8", features = ["xxh3"] }
 
-# AF_XDP (Linux kernel bypass)
-libbpf-sys = "1"
+# AF_XDP (Linux kernel bypass). `xdpilone` is the Rust-native XSK-ring
+# wrapper chosen in ADR-002 — it does *not* depend on libbpf/libxdp at
+# runtime, which keeps the build dependency footprint to "kernel headers
+# present at compile time" only. `libc` is for the few socket/syscall
+# bits xdpilone doesn't expose itself.
+xdpilone = "1.2"
 libc = "0.2"
 
 # TLS

crates/lb-forwarder/src/threading.rs

@@ -208,6 +208,88 @@ impl MultiThreadedForwarder {
         let handles = self.handles.lock();
         !handles.is_empty() && handles.iter().all(|h| !h.is_finished())
     }
+
+    /// Start in **per-queue** mode: each rewriter thread owns its own
+    /// `PacketIo` handle and runs the full `recv → process → send` loop
+    /// inline. Used by the AF_XDP backend, where the NIC's RSS hardware
+    /// hash already partitions traffic across kernel-side queues — there's
+    /// no value in adding userspace steering on top of that, and binding a
+    /// single (iface, queue) AF_XDP socket from two threads (the
+    /// steering/muxer model below) is a kernel-side error (`EBUSY`).
+    ///
+    /// `ios.len()` determines `num_rewriters`. Each `PacketIo` handle is
+    /// expected to be bound to a distinct kernel queue (e.g. queue 0 for
+    /// rewriter 0, queue 1 for rewriter 1, …). The mock backend can also
+    /// run this way, but tests that need shared in-memory queues should
+    /// stick with [`Self::start`].
+    pub fn start_per_queue<T: PacketIo>(
+        ios: Vec<T>,
+        config: ForwarderConfig,
+        shared: ForwarderSharedState,
+    ) -> Self {
+        assert!(
+            !ios.is_empty(),
+            "start_per_queue requires at least one IO handle"
+        );
+        let ForwarderSharedState {
+            lookup_tables,
+            vip_matcher,
+            health_status,
+            metrics,
+        } = shared;
+        let shutdown = Arc::new(AtomicBool::new(false));
+        let num_rewriters = ios.len();
+
+        let affinity = config.cpu_affinity.clone().unwrap_or_default();
+        let resolved_affinity = resolve_cpu_affinity(&affinity, num_rewriters);
+
+        let mut handles = Vec::with_capacity(num_rewriters);
+        for (i, io) in ios.into_iter().enumerate() {
+            let shutdown = Arc::clone(&shutdown);
+            let lookup_tables = lookup_tables.clone();
+            let vip_matcher = Arc::clone(&vip_matcher);
+            let health_status = Arc::clone(&health_status);
+            let metrics = metrics.clone();
+            let conn_ttls = config.conn_ttls;
+            let conn_table_size = config.connection_table_size;
+            let fragment_table_size = config.fragment_table_size;
+            let fragment_ttl = config.fragment_ttl;
+            let mtu_config = config.mtu_config;
+            let icmp_rate_limit = config.icmp_rate_limit;
+            let src_ip = config.src_ip;
+            let batch_size = config.batch_size;
+            let name = format!("lb-rewriter-{i}");
+            let pin = resolved_affinity.rewriters.get(i).copied().flatten();
+
+            handles.push(
+                thread::Builder::new()
+                    .name(name.clone())
+                    .spawn(move || {
+                        pin_current_thread(&name, pin);
+                        let rewriter = crate::rewriter::RewriterThread::new(
+                            src_ip,
+                            conn_table_size,
+                            conn_ttls,
+                            fragment_table_size,
+                            fragment_ttl,
+                            lookup_tables,
+                            vip_matcher,
+                            health_status,
+                            metrics,
+                            mtu_config,
+                            icmp_rate_limit,
+                        );
+                        run_per_queue(io, rewriter, batch_size, &shutdown);
+                    })
+                    .expect("failed to spawn per-queue rewriter thread"),
+            );
+        }
+
+        Self {
+            handles: parking_lot::Mutex::new(handles),
+            shutdown,
+        }
+    }
 }
 
 /// Steering loop: receive packets into pool frames, distribute indices to rewriter queues.
@@ -534,6 +616,46 @@ fn apply_tcp_transitions(
     }
 }
 
+/// Per-queue rewriter loop: this thread owns one `PacketIo` handle and
+/// runs the full pipeline inline — `recv_batch` → `RewriterThread::process_batch`
+/// → `send_batch`. Used by `start_per_queue`. No inter-thread queues, no
+/// PacketPool: the working buffer is a single `Vec<PacketBuf>` reused
+/// across iterations.
+fn run_per_queue<T: PacketIo>(
+    mut io: T,
+    mut rewriter: crate::rewriter::RewriterThread,
+    batch_size: usize,
+    shutdown: &AtomicBool,
+) {
+    let mut buf: Vec<PacketBuf> = vec![PacketBuf::new(); batch_size];
+    let mut spin = 0u32;
+    while !shutdown.load(Ordering::Relaxed) {
+        let n = io.recv_batch(&mut buf).unwrap_or_default();
+        if n == 0 {
+            spin += 1;
+            if spin < SPIN_BEFORE_PARK {
+                std::hint::spin_loop();
+            } else {
+                std::thread::park_timeout(PARK_TIMEOUT);
+                spin = 0;
+            }
+            continue;
+        }
+        spin = 0;
+
+        let processed = rewriter.process_batch(&mut buf[..n]);
+        if processed > 0 {
+            // `send_batch` may return less than processed under
+            // backpressure (AF_XDP COMPLETION drain pace). Just re-try
+            // on the next iteration — packets that didn't go this round
+            // sit in `buf` but will be overwritten by the next recv_batch.
+            // This is correct because under backpressure dropping is
+            // preferable to head-of-line blocking the receive loop.
+            let _ = io.send_batch(&buf[..processed]);
+        }
+    }
+}
+
 /// Muxer loop: drain frame indices from TX queues, send via PacketIo, return frames to pool.
 fn run_muxer<T: PacketIo>(
     io: &mut T,

crates/lb-io/Cargo.toml

@@ -9,12 +9,20 @@ publish.workspace = true
 [features]
 default = ["mock"]
 mock = ["dep:crossbeam"]
-af-xdp = ["dep:libbpf-sys", "dep:libc"]
+# `af-xdp` pulls in the actual XSK ring plumbing. Linux-only; the
+# `xdpilone` dep is gated under `target.'cfg(target_os = "linux")'`
+# below, and our `af_xdp.rs` re-gates with `cfg(target_os = "linux")` so
+# non-Linux dev builds stay clean even with the feature flag forced on
+# in a workspace `Cargo.toml`.
+af-xdp = ["dep:libc", "dep:xdpilone", "dep:smallvec"]
 
 [dependencies]
-libbpf-sys = { workspace = true, optional = true }
 libc = { workspace = true, optional = true }
 crossbeam = { workspace = true, optional = true }
+smallvec = { version = "1", optional = true }
 tracing = { workspace = true }
 
+[target.'cfg(target_os = "linux")'.dependencies]
+xdpilone = { workspace = true, optional = true }
+
 [dev-dependencies]