perf(traffic): optimize production reconciliation

README.md

@@ -277,7 +277,7 @@ Swap:
   drift: no
 ```
 
-Если peers ещё нет, `apply-limits` корректно напишет `no peers found, skip tc apply`. Это не ошибка: traffic topology строится, когда есть хотя бы один peer allowed IP в целевой подсети.
+Если peers ещё нет, это здоровое desired state. `apply-limits` удалит оставшуюся AWG2-owned traffic topology, если раньше peers были, и завершится успешно. В штатном цикле daemon это не создаёт повторяющийся `info`-шум.
 
 ### 8. Включить systemd services
 
@@ -345,7 +345,7 @@ awg2-harness --config /etc/awg2-harness/config.toml <command>
 
 `apply-firewall` делает два reconcile: host INPUT firewall в таблице `inet awg2_host_firewall` и anti-torrent guard в выбранном namespace. Команда не управляет Docker NAT и host `FORWARD`.
 
-`apply-limits` находит Amnezia container, заходит в его network namespace, читает peers через container-side `awg`/`wg`, строит per-peer HTB/fq_codel классы на AWG/WG интерфейсе и IFB ingress path. Если drift не найден, пишет `traffic rules up-to-date, skip apply`.
+`apply-limits` находит Amnezia container, заходит в его network namespace, читает peers через container-side `awg`/`wg`, строит per-peer HTB/fq_codel классы на AWG/WG интерфейсе и IFB ingress path. Peer filters создаются classifier-ом `flower` с единым `pref 10`; legacy peer-`фильтры `u32` считаются drift и мигрируют через controlled rebuild. Healthy state не печатает сообщение на каждом цикле daemon.
 
 `status` печатает фактическую диагностику: сервис, контейнер, netns, интерфейс, peers, WireGuard runtime, traffic desired/actual, host firewall drift, Docker/Amnezia port path, anti-torrent counters, swap state, conntrack, recent events и planned reboot.
 
@@ -355,7 +355,7 @@ awg2-harness --config /etc/awg2-harness/config.toml <command>
 
 ## Что происходит под капотом
 
-Traffic shaping реализован через `tc` в network namespace контейнера Amnezia. На egress используется root qdisc `htb 1:` на AWG/WG интерфейсе, root class `1:1`, per-peer classes `1:<id>` и `fq_codel` под каждым peer class. На ingress ставится `ingress ffff:` на AWG/WG интерфейс, входящие пакеты перенаправляются через `mirred` в `ifb0`, а там строится отдельная HTB topology `2:` с такими же per-peer лимитами.
+Traffic shaping реализован через `tc` в network namespace контейнера Amnezia. На egress используется root qdisc `htb 1:` на AWG/WG интерфейсе, root class `1:1`, per-peer classes `1:<id>` и `fq_codel` под каждым peer class. Egress peer filter сопоставляет `dst_ip <peer>/32`, ingress peer filter — `src_ip <peer>/32`; оба используют `flower pref 10`. На ingress ставится `ingress ffff:` на AWG/WG интерфейс, входящие пакеты перенаправляются через `mirred` в `ifb0`, а там строится отдельная HTB topology `2:` с такими же per-peer лимитами. Только catch-all redirect в IFB остаётся `u32`. Вся `tc` topology после teardown передаётся одним fail-fast batch через `tc -batch -`, затем обязательно проверяется.
 
 Peer discovery берётся из WireGuard/AmneziaWG allowed IPs, а не из handshake state. Поэтому offline peer с `allowed ips: 10.8.1.1/32` тоже получает лимит заранее.
 

config.example.toml

@@ -48,8 +48,10 @@ default_qdisc = "fq"
 rmem_max = 67108864
 wmem_max = 67108864
 
-netdev_backlog_enabled = true
-netdev_max_backlog = 50000
+# По умолчанию harness не меняет receive backlog хоста. 10000 — только
+# стартовое opt-in значение для хоста, где рост drops подтверждён метриками.
+netdev_backlog_enabled = false
+netdev_max_backlog = 10000
 
 udp_rmem_min_enabled = true
 udp_rmem_min = 8192

docs/ACCEPTANCE_CHECKLIST.md

@@ -38,7 +38,7 @@
 - [ ] На AWG/WG-интерфейсе есть root class `1:1` с `traffic.root_rate`.
 - [ ] Для каждого peer есть class `1:<id>` с `traffic.peer_limit`.
 - [ ] Для каждого peer есть `fq_codel` под `1:<id>`.
-- [ ] Для каждого peer есть egress filter `dst peer_ip -> 1:<id>`.
+- [ ] Для каждо peer есть egress filter `flower pref 10 dst_ip peer_ip/32 -> 1:<id>`.
 - [ ] Есть ingress qdisc `ffff:` на AWG/WG-интерфейсе.
 - [ ] Есть ingress redirect через `mirred ... redirect dev ifb0`.
 - [ ] `ifb0` существует и находится в UP state.
@@ -46,7 +46,8 @@
 - [ ] На `ifb0` есть root class `2:1` с `traffic.root_rate`.
 - [ ] Для каждого peer есть class `2:<id>` с `traffic.peer_limit`.
 - [ ] Для каждого peer есть `fq_codel` под `2:<id>`.
-- [ ] Для каждого peer есть ingress filter `src peer_ip -> 2:<id>`.
+- [ ] Для каждого peer есть ingress filter `flower pref 10 src_ip peer_ip/32 -> 2:<id>`.
+- [ ] Все peer-фильтры на каждом parent используют один `pref 10`; `u32` остаётся только у catch-all ingress redirect.
 - [ ] Class ids начинаются с `traffic.class_start_id`, default `10`.
 - [ ] Peer IP → class id mapping использует стабильный лексикографический контракт: IP сортируются как строки, не numeric IPv4.
 
@@ -63,7 +64,10 @@
 - [ ] Если изменить `rate/ceil` peer class, drift детектируется и topology восстанавливается.
 - [ ] Если добавить stale peer class/filter, drift детектируется и topology восстанавливается.
 - [ ] После удаления peer устаревшие class/filter исчезают через rebuild.
+- [ ] После удаления последнего peer вся AWG2-owned traffic topology удаляется; повторный reconcile не выполняет teardown снова.
 - [ ] После добавления peer появляются новые class/filter.
+- [ ] Legacy peer-фильтры `u32` детектируются как migration drift и заменяются на `flower` через rebuild.
+- [ ] Ошибка в любой строке `tc -batch -` останавливает apply; после успеха проходит verify-after-apply.
 
 ## E. Самовосстановление в жизненном цикле
 
@@ -80,6 +84,7 @@
 - [ ] На старте daemon сбой `sysctl_tuning` или `swap` завершает процесс, а сбой traffic/firewall только логируется — сервис остаётся живым.
 - [ ] После обрыва потока `docker events` он перезапускается автоматически (проверяется, например, `systemctl restart docker`).
 - [ ] После восстановления потока `docker events` ошибка подсистемы снимается: в `events.log` появляется `docker_events=recovered`, а `last_action.state` не остаётся с этой ошибкой навсегда.
+- [ ] Один startup/periodic/event-цикл использует один discovery `RuntimeTarget` для traffic и anti-torrent, а не повторяет `docker inspect` и interface discovery.
 - [ ] Успешный reconcile одной подсистемы не гасит незакрытую ошибку другой: при постоянно падающем `traffic` успешный `firewall` не очищает `last_error` и не пишет `firewall=recovered`.
 - [ ] Параллельный запуск `apply-limits` во время работы daemon не приводит к гонке: второй процесс ждёт блокировку до 10 секунд и затем сообщает `lock busy timeout`.
 - [ ] После `kill -9` блокировка освобождается ядром автоматически; оставшийся lock-файл не мешает следующему захвату `/run/awg2-harness/traffic.lock`.
@@ -93,7 +98,9 @@
 - [ ] `status` показывает недостающие и устаревшие правила peers: в `Traffic actual` — строка `reason` (например, `missing egress class`, `egress per-peer class set mismatch`, `duplicate egress class flowid mapping detected`), в `peer_table` — колонка `drift` по каждому peer.
 - [ ] `status` явно показывает частичное состояние и drift, а не общий признак «всё хорошо».
 - [ ] `status` показывает tc filter counters по peer или `filter statistics: unavailable`.
+- [ ] Flower counters и legacy u32 counters в `status` разбираются одним parser-ом и привязываются к правильному peer IP.
 - [ ] Логи доступны через `journalctl -u awg2-harness`.
+- [ ] Повтор одной и той же reconcile-ошибки не создаёт `error`-запись на каждом цикле; первая ошибка, изменение её текста и recovery видны как отдельные переходы.
 - [ ] Отсутствие обязательной host-утилиты (`tc`, `ip`, `docker`, `nsenter`) логируется явно с именем утилиты.
 - [ ] Последний reconcile result виден в status или journal.
 - [ ] `status` показывает последние события reconcile/firewall из `/run/awg2-harness/events.log`.
@@ -143,6 +150,9 @@
 - [ ] В обоих managed-файлах первая строка содержит marker `Managed by awg2-harness sysctl_tuning`.
 - [ ] Файл без marker по managed-пути не перезаписывается и не удаляется: команда завершается ошибкой.
 - [ ] В managed-файле отсутствует `udp_wmem_min`.
+- [ ] В shipped/default config `netdev_backlog_enabled = false`, `netdev_max_backlog = 10000`, а `net.core.netdev_max_backlog` отсутствует в managed-файле.
+- [ ] При смене `netdev_backlog_enabled = true` на `false` key удаляется из persistence; оператор учитывает, что текущее runtime-значение до reboot автоматически не сбрасывается.
+- [ ] До/после ручного включения backlog сняты `/proc/net/softnet_stat`, CPU и latency под реальной нагрузкой.
 - [ ] `net.ipv4.tcp_congestion_control` и `net.core.default_qdisc` присутствуют в managed-файле только при активном BBR.
 - [ ] `swap` пишет `vm.swappiness` в `/etc/sysctl.d/90-awg2-harness-swap.conf`, а не в `99-awg2-harness.conf`.
 - [ ] При обновлении с прежней версии `/etc/sysctl.d/99-awg2-harness.conf` удаляется, если несёт маркер `# Managed by awg2-harness`.
@@ -184,7 +194,7 @@
 - [ ] При превышении порога новые flows дропаются, established flows ниже порога не ломаются.
 - [ ] После restart контейнера nftables rule пересоздаётся в новом netns.
 - [ ] При `anti_torrent.enabled = false` удаляется/отключается только AWG2-owned nftables state; Docker/Amnezia rules не flush-ятся.
-- [ ] `status` показывает `nf_conntrack_count` / `nf_conntrack_max`, если sysctl доступны.
+- [ ] `status` показывает `nf_conntrack_count`, `nf_conntrack_max` и `nf_conntrack_utilization`, если sysctl доступны.
 
 ## J. Откат
 

docs/ANTI_TORRENT.md

@@ -97,7 +97,7 @@ sysctl net.netfilter.nf_conntrack_max
 nft list table inet awg2_harness
 ```
 
-Если `nf_conntrack_count` приближается к `nf_conntrack_max`, проблема уже системная: надо разбираться с timeout, abuse или capacity. Не надо автоматически увеличивать `nf_conntrack_max` из orchestrator без отдельного решения.
+Если `nf_conntrack_count` приближается к `nf_conntrack_max`, проблема уже системная: надо разбираться с timeout, abuse или capacity. `awg2-harness status` показывает оба значения и вычисленный `nf_conntrack_utilization`, чтобы давление было видно без ручного деления. Не надо автоматически увеличивать `nf_conntrack_max` из orchestrator без отдельного решения.
 
 ## 7. Критерии приёмки
 
@@ -106,6 +106,7 @@ nft list table inet awg2_harness
 - Правило ограничивает `ip saddr <peer_ipv4_cidr>`.
 - Лимит применяется per source VPN IP, а не глобально на всю подсеть.
 - Drop counter растёт при превышении threshold.
+- `status` показывает count/max/utilization conntrack, если эти sysctl доступны.
 - Нормальные сценарии с числом flows ниже threshold не затрагиваются.
 - При restart контейнера правило пересоздаётся в новом netns.
 - При `anti_torrent.enabled = false` удаляется или отключается только AWG2-owned nftables state.

docs/CONFIGURATION.md

@@ -58,8 +58,8 @@ tcp_bbr = "auto"          # off | auto | require
 default_qdisc = "fq"      # none | fq | fq_codel
 rmem_max = 67108864
 wmem_max = 67108864
-netdev_backlog_enabled = true
-netdev_max_backlog = 50000
+netdev_backlog_enabled = false
+netdev_max_backlog = 10000
 udp_rmem_min_enabled = true
 udp_rmem_min = 8192
 
@@ -129,6 +129,10 @@ swappiness = 1
 - `tcp_bbr = auto` не валит запуск при недоступном BBR (warning), `require` — валит;
 - для `sysctl_tuning`: `modprobe` обязателен только для `tcp_bbr = "require"`.
 
+`netdev_backlog_enabled = false` — безопасный production default: harness не управляет `net.core.netdev_max_backlog`, пока оператор не включит это явно после измерений. Значение `netdev_max_backlog = 10000` используется только при `netdev_backlog_enabled = true`.
+
+При обновлении с конфига, где backlog раньше был включён, переключение флага в `false` удаляет key из managed-файла, но не перезаписывает уже изменённое runtime-значение. До следующей перезагрузки его можно вернуть вручную к baseline дистрибутива. Перед осознанным включением backlog снимите pre/post-метрики `/proc/net/softnet_stat` (в том числе dropped/time_squeeze) и CPU/latency под реальной нагрузкой.
+
 Отдельно для всего пакета:
 
 - `traffic`-модуль требует `modprobe` для IFB при `traffic.enabled = true`.

docs/IMPLEMENTATION_SPEC.md

@@ -243,8 +243,8 @@ struct PeerTrafficStatus {
 `discover_target()`:
 
 1. Прочитать упорядоченный `amnezia.container_names`.
-2. Найти первый запущенный контейнер с точным совпадением имени.
-3. Получить container ID и PID.
+2. Для каждого имени по порядку выполнить один `docker inspect`, который сразу возвращает running state, container ID и PID.
+3. Выбрать первый запущенный контейнер с точным совпадением имени; отсутствующий или stopped candidate не меняет порядок fallback.
 4. Построить `netns_path`.
 5. Определить интерфейс в этом namespace.
 6. Вернуть `RuntimeTarget`.
@@ -313,7 +313,7 @@ tc class add dev <iface> parent 1: classid 1:1 htb rate <root_rate> ceil <root_r
 for each peer:
   tc class add dev <iface> parent 1:1 classid 1:<id> htb rate <peer_limit> ceil <peer_limit>
   tc qdisc add dev <iface> parent 1:<id> fq_codel
-  tc filter add dev <iface> parent 1: protocol ip u32 match ip dst <peer_ip> flowid 1:<id>
+  tc filter add dev <iface> parent 1: protocol ip pref 10 flower dst_ip <peer_ip>/32 classid 1:<id>
 ```
 
 Ingress через IFB:
@@ -329,9 +329,11 @@ tc class add dev <ifb_name> parent 2: classid 2:1 htb rate <root_rate> ceil <roo
 for each peer:
   tc class add dev <ifb_name> parent 2:1 classid 2:<id> htb rate <peer_limit> ceil <peer_limit>
   tc qdisc add dev <ifb_name> parent 2:<id> fq_codel
-  tc filter add dev <ifb_name> parent 2: protocol ip u32 match ip src <peer_ip> flowid 2:<id>
+  tc filter add dev <ifb_name> parent 2: protocol ip pref 10 flower src_ip <peer_ip>/32 classid 2:<id>
 ```
 
+Все peer-фильтры на каждом parent используют один `pref 10`: для `flower` это даёт один hash table на priority вместо множества priority structures. Маска всегда `/32`, поэтому один priority не смешивает несовместимые masks. Catch-all redirect в IFB остаётся `u32`, поскольку это одно правило, а не per-peer lookup.
+
 ## 9. Алгоритм reconcile
 
 Traffic и firewall — два независимых reconcile с отдельными блокировками. Они не объединены в одну транзакцию: сбой одного не должен блокировать другой.
@@ -343,10 +345,12 @@ reconcile_traffic():
   target = discover_target()
   peers  = discover_peers(config, target)
   если peers пуст:
-      залогировать "no peers found, skip tc apply" и выйти
+      проверить наличие AWG2-owned topology
+      если она осталась от прежних peers: teardown_owned_topology(target)
+      иначе выйти с debug-диагностикой
   drift = inspect_and_diff(target, peers, config)
   если drift отсутствует:
-      залогировать "traffic rules up-to-date, skip apply" и выйти
+      выйти с debug-диагностикой
   teardown_owned_topology(target)
   apply_full_topology(target, peers, config)   # внутри — verify-after-apply
 ```
@@ -365,12 +369,15 @@ apply_firewall():
 Требования к устойчивости парсеров:
 
 - `tc filter show` разбирается блоками (конечный автомат), а не наивным сканированием токенов по несвязанным строкам;
+- единый parser понимает production `flower` и legacy `u32`, включая counters; точный healthy-контракт требует `flower`, `pref 10`, правильное direction-field и `/32`, а legacy `u32` peer rules считаются migration drift;
+- нераспознанный или malformed detailed filter block не игнорируется, а делает инспекцию drifted;
 - отображение `peer_ip -> class_id` проверяется для обеих сторон: egress (`1:`) и ingress (`2:`);
 - проверка обязана падать, если классы и фильтры существуют, но отображение отличается от детерминированного присвоения.
 
 Политика apply/verify:
 
-- все шаги apply — явные вызовы команд;
+- `modprobe ifb`, создание IFB и его перевод в UP выполняются строго и до `tc` apply;
+- после teardown вся `tc` topology подаётся одним fail-fast сценарием в stdin команды `tc -batch -`; `-force` не используется, поэтому первая ошибка останавливает batch и сохраняет номер строки в stderr;
 - ожидаемые конфликты идемпотентности допускаются выборочно, через отдельную обёртку, и по закрытому списку признаков: `file exists`, `eexist`, `object already exists`, `exclusive flag on`;
 - ошибки «объект не найден» (`no such`, `not found`, `cannot find`, `does not exist`, `handle of zero`) допускаются только на шагах удаления — в teardown и при отключении владеемых таблиц;
 - любая другая ошибка apply фатальна;
@@ -437,9 +444,9 @@ table inet awg2_harness {
 - counter `antitorrent_dropped_v4`;
 - meter/set state, создаваемый правилом.
 
-Status обязан показывать наличие таблицы/chain/правила, настроенный порог и drop counter. Также он показывает давление на conntrack хоста через `net.netfilter.nf_conntrack_count` и `net.netfilter.nf_conntrack_max`, когда они доступны.
+Status обязан показывать наличие таблицы/chain/правила, настроенный порог и drop counter. Также он один раз читает `net.netfilter.nf_conntrack_count` и `net.netfilter.nf_conntrack_max` и, когда оба значения доступны, печатает `nf_conntrack_utilization` в процентах.
 
-При `anti_torrent.enabled = false` демон удаляет только владеемую anti-torrent таблицу. Глобальный ruleset nftables не сбрасывается никогда. Если контейнер в этот момент отсутствует, удаление в его namespace просто пропускается — это не ошибка.
+При `anti_torrent.enabled = false` демон удаляет только владеемую anti-torrent таблицу. Глобальный ruleset nftables не сбрасывается никогда. Если контейнер в этот момент именно не найден, удаление в его namespace пропускается — это не ошибка. Ошибка Docker API/daemon/permissions не маскируется под «контейнер не найден» и возвращается оператору.
 
 ## 12. Инспекция фактического состояния
 
@@ -456,6 +463,8 @@ Status обязан показывать наличие таблицы/chain/п
 - классы на `<ifb_name>`;
 - фильтры под parent `2:` на `<ifb_name>`.
 
+Один вывод `tc qdisc show dev <iface>` используется одновременно для root, ingress и per-peer `fq_codel`; один вывод `tc class show dev <device>` — для root и per-peer classes. Инспекция не делает отдельный child process на каждый parent.
+
 Дополнительно детектируются дубли: один и тот же peer IP в нескольких фильтрах и один и тот же class id у нескольких IP. Оба случая считаются drift.
 
 `status` показывает компактную сводку и таблицу по peers.
@@ -469,6 +478,8 @@ Status обязан показывать наличие таблицы/chain/п
 
 Таймер обязателен. События могут быть пропущены при рестарте демона, рестарте Docker или транзиентных ошибках.
 
+Внутри одного startup/periodic/event-цикла traffic и anti-torrent получают один и тот же результат discovery `RuntimeTarget`; повторные `docker inspect`, PID/interface discovery и transport construction в этом же цикле не выполняются. Наличие host `nsenter` кэшируется на время жизни процесса.
+
 Проверка готовности не должна хардкодить `awg0`. Она вызывает общий детектор интерфейса с интервалом 2 секунды до истечения `traffic.container_ready_timeout_secs`.
 
 Порядок на старте демона и разница в критичности ошибок:
@@ -487,7 +498,7 @@ Status обязан показывать наличие таблицы/chain/п
 
 Ошибки отслеживаются по подсистемам (`traffic`, `firewall`, `swap`, `sysctl_tuning`, `docker_events`), а не одним общим полем. Общее поле в `last_action.state` показывает ту ошибку, которая ещё не устранена: успешный reconcile одной подсистемы не гасит незакрытую ошибку другой и не объявляет восстановление того, что не падало. Поток `docker events` при успешном переподключении тоже отмечается как восстановленный — иначе один рестарт Docker оставлял бы ошибку в `status` навсегда при исправно работающем новом потоке.
 
-В `events.log` пишутся только **переходы состояния**: первое появление ошибки, изменение текста ошибки и восстановление после неё. Рутинный успех не пишется. Раньше каждый успешный цикл добавлял строку — при интервале 15 секунд это порядка 17 тысяч записей в сутки, каждая с полным чтением, дополнением и перезаписью файла, и на их фоне терялись записи, ради которых журнал и читают. Полная поцикловая картина остаётся в journald через tracing.
+В `events.log` и journald на уровне `info`/`error` пишутся только **переходы состояния**: первое появление ошибки, изменение её текста и восстановление. Рутинный healthy-цикл и повтор той же ошибки остаются на `debug`. Это исключае повторяющийся journal-шум на каждом 15-секундном reconcile, но сохраняет детали для отладки при включённом debug-уровне.
 
 ## 14. Блокировки
 
@@ -552,7 +563,7 @@ Swap:
   desired_active, exclusive, fstab_exclusive, fstab_swap_entries,
   swappiness_current, swappiness_persisted, drift[, drift_reason]
 Conntrack:
-  nf_conntrack_count, nf_conntrack_max
+  nf_conntrack_count, nf_conntrack_max, nf_conntrack_utilization
 Last actions:
   содержимое /run/awg2-harness/last_action.state
 Recent events:

docs/INSTALLATION.md

@@ -292,7 +292,7 @@ Swap:
   drift: no
 ```
 
-Если peers ещё нет, `apply-limits` напечатает `no peers found, skip tc apply`. Для пустого сервера это здоровое состояние.
+Если peers ещё нет, это здоровое desired state. `apply-limits` успешно завершится и удалит оставшуюся AWG2-owned traffic topology, если до этого peers были.
 
 При `swap.enabled = true` AWG2 Harness намеренно делает swap эксклюзивным: один управляемый swapfile, одна swap-запись в `/etc/fstab`, заданный `swappiness`.
 

traffic_core/src/config.rs

@@ -77,8 +77,8 @@ impl Default for SysctlTuningConfig {
             default_qdisc: "fq".to_string(),
             rmem_max: 67_108_864,
             wmem_max: 67_108_864,
-            netdev_backlog_enabled: true,
-            netdev_max_backlog: 50_000,
+            netdev_backlog_enabled: false,
+            netdev_max_backlog: 10_000,
             udp_rmem_min_enabled: true,
             udp_rmem_min: 8192,
         }
@@ -830,6 +830,8 @@ mod tests {
     fn sysctl_tuning_default_is_valid() {
         let cfg = base_config();
         assert!(cfg.validate_config().is_ok());
+        assert!(!cfg.sysctl_tuning.netdev_backlog_enabled);
+        assert_eq!(cfg.sysctl_tuning.netdev_max_backlog, 10_000);
     }
 
     #[test]

traffic_core/src/daemon.rs

@@ -7,9 +7,7 @@ use std::{
     time::{Duration, Instant},
 };
 
-use crate::{
-    config::Config, docker, firewall, iface, netns::build_transport, swap, sysctl_tuning, traffic,
-};
+use crate::{config::Config, docker, firewall, swap, sysctl_tuning, traffic};
 use tracing::{error, info};
 
 #[derive(Debug, Clone, Default)]
@@ -67,14 +65,15 @@ pub fn run(cfg: Config) -> anyhow::Result<()> {
         append_event("startup_swap ok");
     }
 
-    if let Err(e) = traffic::reconcile_traffic(&cfg) {
+    let startup_target = discover_runtime_target_if_needed(&cfg);
+    if let Err(e) = reconcile_traffic_for_cycle(&cfg, startup_target.as_ref()) {
         set_err(&state, "traffic", format!("startup traffic reconcile: {e}"));
         append_event(&format!("startup_reconcile traffic=error error=\"{e}\""));
     } else {
         set_ok(&state, "traffic", "startup traffic reconcile ok");
         append_event("startup_reconcile traffic=ok");
     }
-    if let Err(e) = firewall::apply_firewall(&cfg) {
+    if let Err(e) = reconcile_firewall_for_cycle(&cfg, startup_target.as_ref()) {
         set_err(
             &state,
             "firewall",
@@ -119,17 +118,18 @@ pub fn run(cfg: Config) -> anyhow::Result<()> {
             );
         }
 
+        let target = discover_runtime_target_if_needed(&cfg);
         report(
             &state,
             "traffic",
             "periodic traffic reconcile",
-            traffic::reconcile_traffic(&cfg),
+            reconcile_traffic_for_cycle(&cfg, target.as_ref()),
         );
         report(
             &state,
             "firewall",
             "periodic firewall reconcile",
-            firewall::apply_firewall(&cfg),
+            reconcile_firewall_for_cycle(&cfg, target.as_ref()),
         );
 
         thread::sleep(Duration::from_secs(
@@ -138,6 +138,44 @@ pub fn run(cfg: Config) -> anyhow::Result<()> {
     }
 }
 
+fn discover_runtime_target_if_needed(
+    cfg: &Config,
+) -> Option<anyhow::Result<docker::RuntimeTarget>> {
+    (cfg.traffic.enabled || cfg.anti_torrent.apply_in_amnezia_netns)
+        .then(|| docker::discover_target(cfg))
+}
+
+fn reconcile_traffic_for_cycle(
+    cfg: &Config,
+    target: Option<&anyhow::Result<docker::RuntimeTarget>>,
+) -> anyhow::Result<()> {
+    if !cfg.traffic.enabled {
+        return Ok(());
+    }
+    match target {
+        Some(Ok(target)) => traffic::reconcile_traffic_with_target(cfg, target),
+        Some(Err(e)) => Err(anyhow::anyhow!("runtime target discovery failed: {e:#}")),
+        None => Err(anyhow::anyhow!(
+            "runtime target was not discovered for enabled traffic"
+        )),
+    }
+}
+
+fn reconcile_firewall_for_cycle(
+    cfg: &Config,
+    target: Option<&anyhow::Result<docker::RuntimeTarget>>,
+) -> anyhow::Result<()> {
+    if !cfg.anti_torrent.apply_in_amnezia_netns {
+        return firewall::apply_firewall(cfg);
+    }
+    match target {
+        Some(target) => firewall::apply_firewall_with_target_result(cfg, target.as_ref()),
+        None => Err(anyhow::anyhow!(
+            "runtime target was not discovered for namespace firewall"
+        )),
+    }
+}
+
 /// Records the outcome of one reconcile, writing an event only when something changed.
 ///
 /// A routine success used to append a line every cycle: at a 15 second interval that is
@@ -155,14 +193,17 @@ fn report(
         Ok(()) => {
             let recovered = set_ok(state, subsystem, &format!("{what} ok"));
             if recovered {
+                info!("{subsystem} reconcile recovered");
                 append_event(&format!("{subsystem}=recovered"));
             }
         }
         Err(e) => {
-            error!(error = %e, "{subsystem} reconcile error");
             let changed = set_err(state, subsystem, format!("{what}: {e}"));
             if changed {
+                error!(error = %e, "{subsystem} reconcile error");
                 append_event(&format!("{subsystem}=error error=\"{e}\""));
+            } else {
+                tracing::debug!(error = %e, "{subsystem} reconcile error unchanged");
             }
         }
     }
@@ -217,44 +258,58 @@ fn consume_docker_events(
     for line in reader.lines() {
         let name = line?.trim().to_string();
         if cfg.amnezia.container_names.iter().any(|n| n == &name)
-            && wait_for_interface(cfg, &name, cfg.traffic.container_ready_timeout_secs)
+            && let Some(target) =
+                wait_for_target(cfg, &name, cfg.traffic.container_ready_timeout_secs)
         {
             append_event(&format!("docker_event container={name} action=start"));
-            if let Err(e) = traffic::reconcile_traffic(cfg) {
-                set_err(state, "traffic", format!("event traffic reconcile: {e}"));
+            if let Err(e) = traffic::reconcile_traffic_with_target(cfg, &target) {
+                if set_err(state, "traffic", format!("event traffic reconcile: {e}")) {
+                    error!(error = %e, "traffic event reconcile error");
                     append_event(&format!("event_reconcile traffic=error error=\"{e}\""));
+                }
             } else {
-                set_ok(state, "traffic", "event traffic reconcile ok");
-                append_event("event_reconcile traffic=ok");
+                if set_ok(state, "traffic", "event traffic reconcile ok") {
+                    info!("traffic event reconcile recovered");
+                    append_event("event_reconcile traffic=recovered");
+                }
             }
 
-            if let Err(e) = firewall::apply_firewall(cfg) {
-                set_err(state, "firewall", format!("event firewall reconcile: {e}"));
+            if let Err(e) = firewall::apply_firewall_with_target_result(cfg, Ok(&target)) {
+                if set_err(state, "firewall", format!("event firewall reconcile: {e}")) {
+                    error!(error = %e, "firewall event reconcile error");
                     append_event(&format!("event_reconcile firewall=error error=\"{e}\""));
+                }
             } else {
-                set_ok(state, "firewall", "event firewall reconcile ok");
-                append_event("event_reconcile firewall=ok");
+                if set_ok(state, "firewall", "event firewall reconcile ok") {
+                    info!("firewall event reconcile recovered");
+                    append_event("event_reconcile firewall=recovered");
+                }
             }
         }
     }
     Ok(())
 }
 
-fn wait_for_interface(cfg: &Config, container_name: &str, timeout_secs: u64) -> bool {
+fn wait_for_target(
+    cfg: &Config,
+    container_name: &str,
+    timeout_secs: u64,
+) -> Option<docker::RuntimeTarget> {
     let start = Instant::now();
     while start.elapsed() < Duration::from_secs(timeout_secs.max(1)) {
-        if let Ok(target) = docker::discover_target(cfg)
-            && target.container_name == container_name
-        {
-            let transport = build_transport(target.container_pid, &target.container_name);
-            if iface::detect_interface_with_transport(cfg, transport.as_ref()).is_ok() {
-                return true;
+        if let Ok(target) = docker::discover_target(cfg) {
+            if target.container_name == container_name {
+                return Some(target);
             }
+            // A higher-priority configured container is already the active target. A start
+            // event for a lower-priority candidate cannot change desired state, so waiting
+            // for the full readiness timeout would only stall the event consumer.
+            return None;
         }
 
         thread::sleep(Duration::from_secs(2));
     }
-    false
+    None
 }
 
 impl LastActionState {

traffic_core/src/docker.rs

@@ -60,18 +60,10 @@ impl PortPathDiagnostics {
 }
 
 pub fn discover_target(cfg: &Config) -> anyhow::Result<RuntimeTarget> {
-    let container_name = find_running_container(&cfg.amnezia.container_names)?;
-    let container_id = run_capture(&["docker", "inspect", "--format", "{{.Id}}", &container_name])?
-        .trim()
-        .to_string();
-    let pid_str = run_capture(&[
-        "docker",
-        "inspect",
-        "--format",
-        "{{.State.Pid}}",
-        &container_name,
-    ])?;
-    let container_pid = pid_str.trim().parse::<u32>()?;
+    let snapshot = discover_container_snapshot_with(&cfg.amnezia.container_names, run_capture)?;
+    let container_name = snapshot.name;
+    let container_id = snapshot.id;
+    let container_pid = snapshot.pid;
     let netns_path = format!("/proc/{container_pid}/ns/net");
     let interface_name = crate::iface::detect_interface(cfg, container_pid, &container_name)?;
 
@@ -84,16 +76,69 @@ pub fn discover_target(cfg: &Config) -> anyhow::Result<RuntimeTarget> {
     })
 }
 
-fn find_running_container(names: &[String]) -> anyhow::Result<String> {
-    let ps = run_capture(&["docker", "ps", "--format", "{{.Names}}"])?;
-    for name in names {
-        if ps.lines().any(|x| x.trim() == name) {
-            return Ok(name.clone());
+#[derive(Debug, Clone, PartialEq, Eq)]
+struct ContainerSnapshot {
+    name: String,
+    id: String,
+    pid: u32,
 }
+
+fn discover_container_snapshot_with<F>(
+    names: &[String],
+    mut run: F,
+) -> anyhow::Result<ContainerSnapshot>
+where
+    F: FnMut(&[&str]) -> anyhow::Result<String>,
+{
+    for name in names {
+        let raw = match run(&[
+            "docker",
+            "inspect",
+            "--format",
+            "{{.State.Running}}\t{{.Id}}\t{{.State.Pid}}",
+            name,
+        ]) {
+            Ok(raw) => raw,
+            Err(e) if is_missing_container_error(&e.to_string()) => continue,
+            Err(e) => return Err(e.context(format!("cannot inspect container candidate {name}"))),
+        };
+        let fields: Vec<&str> = raw.trim().split('\t').collect();
+        if fields.len() != 3 {
+            anyhow::bail!(
+                "docker inspect returned malformed target snapshot for {name}: expected 3 fields"
+            );
+        }
+        let running = fields[0]
+            .parse::<bool>()
+            .map_err(|_| anyhow::anyhow!("invalid running state for container {name}"))?;
+        if !running {
+            continue;
+        }
+        let id = fields[1].trim();
+        if id.is_empty() {
+            anyhow::bail!("docker inspect returned an empty container ID for {name}");
+        }
+        let pid = fields[2]
+            .trim()
+            .parse::<u32>()
+            .map_err(|_| anyhow::anyhow!("invalid PID for running container {name}"))?;
+        if pid == 0 {
+            anyhow::bail!("docker inspect returned PID 0 for running container {name}");
+        }
+        return Ok(ContainerSnapshot {
+            name: name.clone(),
+            id: id.to_string(),
+            pid,
+        });
     }
     Err(AppError::NotFound("amnezia container not running".to_string()).into())
 }
 
+fn is_missing_container_error(message: &str) -> bool {
+    let lower = message.to_ascii_lowercase();
+    lower.contains("no such object") || lower.contains("no such container")
+}
+
 pub fn run_capture(args: &[&str]) -> anyhow::Result<String> {
     let (program, rest) = args
         .split_first()
@@ -346,6 +391,61 @@ fn normalize_nft_forward_text(raw: &str) -> String {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use std::cell::RefCell;
+
+    #[test]
+    fn target_discovery_uses_one_inspect_per_candidate_in_config_order() {
+        let calls = RefCell::new(Vec::<String>::new());
+        let names = vec!["primary".to_string(), "legacy".to_string()];
+        let got = discover_container_snapshot_with(&names, |args| {
+            let name = args.last().expect("candidate").to_string();
+            calls.borrow_mut().push(name.clone());
+            match name.as_str() {
+                "primary" => Ok("false\tstopped-id\t0\n".into()),
+                "legacy" => Ok("true\trunning-id\t4242\n".into()),
+                _ => unreachable!(),
+            }
+        })
+        .expect("running fallback");
+
+        assert_eq!(calls.into_inner(), ["primary", "legacy"]);
+        assert_eq!(
+            got,
+            ContainerSnapshot {
+                name: "legacy".into(),
+                id: "running-id".into(),
+                pid: 4242,
+            }
+        );
+    }
+
+    #[test]
+    fn target_discovery_skips_only_explicitly_missing_candidates() {
+        let names = vec!["missing".to_string(), "running".to_string()];
+        let got = discover_container_snapshot_with(&names, |args| match args.last().copied() {
+            Some("missing") => anyhow::bail!("Error: No such object: missing"),
+            Some("running") => Ok("true\tid\t77\n".into()),
+            _ => unreachable!(),
+        })
+        .expect("running candidate");
+        assert_eq!(got.name, "running");
+    }
+
+    #[test]
+    fn target_discovery_does_not_mask_docker_daemon_errors() {
+        let names = vec!["primary".to_string(), "legacy".to_string()];
+        let calls = RefCell::new(0usize);
+        let err = discover_container_snapshot_with(&names, |_args| {
+            *calls.borrow_mut() += 1;
+            anyhow::bail!("permission denied connecting to docker socket")
+        })
+        .expect_err("daemon failure must propagate");
+        assert_eq!(*calls.borrow(), 1);
+        assert!(
+            err.to_string()
+                .contains("cannot inspect container candidate")
+        );
+    }
 
     #[test]
     fn parse_userland_proxy_from_docker_info_works() {

traffic_core/src/firewall/antitorrent.rs

@@ -13,7 +13,7 @@ use serde_json::Value;
 use crate::{config::Config, docker, error::AppError, netns::NsTransport};
 
 use super::{
-    ANTITORRENT_TABLE, firewall_transport,
+    ANTITORRENT_TABLE, firewall_transport, firewall_transport_for_target,
     nft::{is_nft_not_found, normalize_nft_text, run_allow_notfound, run_nft_batch, table_exists},
 };
 
@@ -68,6 +68,64 @@ pub(super) fn apply_antitorrent_firewall(cfg: &Config) -> anyhow::Result<()> {
     Ok(())
 }
 
+pub(super) fn apply_antitorrent_firewall_with_target_result(
+    cfg: &Config,
+    target: Result<&docker::RuntimeTarget, &anyhow::Error>,
+) -> anyhow::Result<()> {
+    if !cfg.anti_torrent.apply_in_amnezia_netns {
+        return apply_antitorrent_firewall(cfg);
+    }
+
+    match target {
+        Ok(target) => apply_antitorrent_with_transport(
+            cfg,
+            firewall_transport_for_target(cfg, target).as_ref(),
+        ),
+        Err(e) if !cfg.anti_torrent.enabled && is_container_absent(e) => Ok(()),
+        Err(e) if !cfg.anti_torrent.enabled => Err(anyhow::anyhow!(
+            "cannot reach the Amnezia container to remove the anti-torrent table: {e:#}"
+        )),
+        Err(e) => Err(anyhow::anyhow!(
+            "cannot reach the Amnezia container to reconcile anti-torrent: {e:#}"
+        )),
+    }
+}
+
+fn apply_antitorrent_with_transport(
+    cfg: &Config,
+    transport: &dyn NsTransport,
+) -> anyhow::Result<()> {
+    if !cfg.anti_torrent.enabled {
+        run_allow_notfound(
+            transport,
+            &["nft", "delete", "table", "inet", ANTITORRENT_TABLE],
+        )?;
+        return Ok(());
+    }
+
+    let actual = inspect_firewall(transport, cfg)?;
+    if !firewall_has_drift(&actual) {
+        return Ok(());
+    }
+    let present = table_exists(transport, ANTITORRENT_TABLE)?;
+    run_nft_batch(
+        transport,
+        &build_antitorrent_batch_script(
+            &cfg.anti_torrent.peer_ipv4_cidr,
+            cfg.anti_torrent.max_connections_per_peer,
+            present,
+        ),
+    )?;
+    let verify = inspect_firewall(transport, cfg)?;
+    if firewall_has_drift(&verify) {
+        anyhow::bail!(
+            "anti-torrent firewall verify-after-apply failed: {}",
+            firewall_drift_reason(&verify)
+        );
+    }
+    Ok(())
+}
+
 fn apply_antitorrent_disabled(cfg: &Config) -> anyhow::Result<()> {
     if cfg.anti_torrent.apply_in_amnezia_netns {
         match docker::discover_target(cfg) {
@@ -162,7 +220,17 @@ pub fn antitorrent_drift_reason(cfg: &Config) -> anyhow::Result<Option<String>>
         return Ok(None);
     }
     let transport = firewall_transport(cfg)?;
-    let actual = inspect_firewall(transport.as_ref(), cfg)?;
+    antitorrent_drift_reason_with_transport(cfg, transport.as_ref())
+}
+
+pub(super) fn antitorrent_drift_reason_with_transport(
+    cfg: &Config,
+    transport: &dyn NsTransport,
+) -> anyhow::Result<Option<String>> {
+    if !cfg.anti_torrent.enabled {
+        return Ok(None);
+    }
+    let actual = inspect_firewall(transport, cfg)?;
     if firewall_has_drift(&actual) {
         return Ok(Some(firewall_drift_reason(&actual)));
     }
@@ -175,6 +243,16 @@ pub fn antitorrent_drop_counter(cfg: &Config) -> anyhow::Result<Option<(u64, u64
     }
 
     let transport = firewall_transport(cfg)?;
+    antitorrent_drop_counter_with_transport(cfg, transport.as_ref())
+}
+
+pub(super) fn antitorrent_drop_counter_with_transport(
+    cfg: &Config,
+    transport: &dyn NsTransport,
+) -> anyhow::Result<Option<(u64, u64)>> {
+    if !cfg.anti_torrent.enabled {
+        return Ok(None);
+    }
     let json = transport.run(&[
         "nft",
         "-j",
@@ -449,4 +527,31 @@ table inet awg2_harness {
 "#;
         assert!(firewall_rule_contract_matches(table, "10.8.1.0/24", 100));
     }
+
+    #[test]
+    fn disabled_namespace_cleanup_allows_only_an_absent_container() {
+        let mut cfg = crate::firewall::tests::test_config();
+        cfg.anti_torrent.enabled = false;
+        cfg.anti_torrent.apply_in_amnezia_netns = true;
+        let missing: anyhow::Error =
+            AppError::NotFound("amnezia container not running".into()).into();
+
+        apply_antitorrent_firewall_with_target_result(&cfg, Err(&missing))
+            .expect("there is no namespace to clean when the container is absent");
+    }
+
+    #[test]
+    fn disabled_namespace_cleanup_does_not_mask_docker_failures() {
+        let mut cfg = crate::firewall::tests::test_config();
+        cfg.anti_torrent.enabled = false;
+        cfg.anti_torrent.apply_in_amnezia_netns = true;
+        let unavailable = anyhow::anyhow!("permission denied on Docker socket");
+
+        let err = apply_antitorrent_firewall_with_target_result(&cfg, Err(&unavailable))
+            .expect_err("Docker failures must remain visible");
+        assert!(
+            err.to_string()
+                .contains("cannot reach the Amnezia container")
+        );
+    }
 }

traffic_core/src/firewall/mod.rs

@@ -23,6 +23,20 @@ use crate::{
 pub use antitorrent::{antitorrent_drift_reason, antitorrent_drop_counter};
 pub use host::host_firewall_drift_reason;
 
+pub(crate) fn antitorrent_drift_reason_with_transport(
+    cfg: &Config,
+    transport: &dyn NsTransport,
+) -> anyhow::Result<Option<String>> {
+    antitorrent::antitorrent_drift_reason_with_transport(cfg, transport)
+}
+
+pub(crate) fn antitorrent_drop_counter_with_transport(
+    cfg: &Config,
+    transport: &dyn NsTransport,
+) -> anyhow::Result<Option<(u64, u64)>> {
+    antitorrent::antitorrent_drop_counter_with_transport(cfg, transport)
+}
+
 const ANTITORRENT_TABLE: &str = "awg2_harness";
 const HOST_FW_TABLE: &str = "awg2_host_firewall";
 
@@ -35,6 +49,17 @@ pub fn apply_firewall(cfg: &Config) -> anyhow::Result<()> {
     Ok(())
 }
 
+pub(crate) fn apply_firewall_with_target_result(
+    cfg: &Config,
+    target: Result<&docker::RuntimeTarget, &anyhow::Error>,
+) -> anyhow::Result<()> {
+    let _lock =
+        FileLock::acquire_timeout("/run/awg2-harness/firewall.lock", Duration::from_secs(10))?;
+
+    host::apply_host_firewall(cfg)?;
+    antitorrent::apply_antitorrent_firewall_with_target_result(cfg, target)
+}
+
 /// Anti-torrent rules go wherever the peer source addresses are actually visible.
 fn firewall_transport(cfg: &Config) -> anyhow::Result<Box<dyn NsTransport>> {
     if cfg.anti_torrent.apply_in_amnezia_netns {
@@ -48,6 +73,17 @@ fn firewall_transport(cfg: &Config) -> anyhow::Result<Box<dyn NsTransport>> {
     }
 }
 
+fn firewall_transport_for_target(
+    cfg: &Config,
+    target: &docker::RuntimeTarget,
+) -> Box<dyn NsTransport> {
+    if cfg.anti_torrent.apply_in_amnezia_netns {
+        build_transport(target.container_pid, &target.container_name)
+    } else {
+        build_host_transport()
+    }
+}
+
 #[cfg(test)]
 pub(crate) mod tests {
     use crate::config::{

traffic_core/src/iface.rs

@@ -1,17 +1,10 @@
 use crate::{
     config::Config,
-    netns::{DockerExecTransport, NsTransport, NsenterTransport, nsenter_available},
+    netns::{NsTransport, build_transport},
 };
 
 pub fn detect_interface(cfg: &Config, pid: u32, container: &str) -> anyhow::Result<String> {
-    let boxed: Box<dyn NsTransport> = if nsenter_available() {
-        Box::new(NsenterTransport { pid })
-    } else {
-        Box::new(DockerExecTransport {
-            container: container.to_string(),
-        })
-    };
-
+    let boxed = build_transport(pid, container);
     detect_interface_with_transport(cfg, boxed.as_ref())
 }
 

traffic_core/src/netns.rs

@@ -1,6 +1,7 @@
 use std::{
     io::Write,
     process::{Command, Stdio},
+    sync::OnceLock,
 };
 
 pub trait NsTransport: Send + Sync {
@@ -129,7 +130,8 @@ impl NsTransport for HostTransport {
 }
 
 pub fn nsenter_available() -> bool {
-    Command::new("nsenter").arg("--help").output().is_ok()
+    static AVAILABLE: OnceLock<bool> = OnceLock::new();
+    *AVAILABLE.get_or_init(|| Command::new("nsenter").arg("--help").output().is_ok())
 }
 
 pub fn build_transport(pid: u32, container: &str) -> Box<dyn NsTransport> {

traffic_core/src/status.rs

@@ -56,6 +56,7 @@ pub fn print_status(cfg: &Config) -> anyhow::Result<()> {
         print_degraded_sections(cfg);
         return Ok(());
     };
+    let transport = build_transport(target.container_pid, &target.container_name);
 
     let docker_path_diag = docker::collect_port_path_diagnostics(
         &target.container_name,
@@ -118,7 +119,17 @@ pub fn print_status(cfg: &Config) -> anyhow::Result<()> {
     print_traffic_desired_section(cfg);
 
     println!("Traffic actual:");
-    match traffic::inspect_drift_summary(cfg) {
+    let traffic_drift = if cfg.traffic.enabled {
+        match &peers {
+            Ok(peers) => {
+                traffic::inspect_drift_summary_with_state(cfg, &target, transport.as_ref(), peers)
+            }
+            Err(e) => Err(anyhow::anyhow!("peer discovery failed: {e}")),
+        }
+    } else {
+        Ok(None)
+    };
+    match traffic_drift {
         Ok(None) => println!("  drift: no"),
         Ok(Some(reason)) => {
             println!("  drift: yes");
@@ -128,7 +139,6 @@ pub fn print_status(cfg: &Config) -> anyhow::Result<()> {
     }
 
     println!("  topology:");
-    let transport = build_transport(target.container_pid, &target.container_name);
     let egress_qdisc = transport
         .run(&["tc", "qdisc", "show", "dev", &target.interface_name])
         .unwrap_or_default();
@@ -141,7 +151,17 @@ pub fn print_status(cfg: &Config) -> anyhow::Result<()> {
     println!("    ifb_htb_present: {}", ifb_qdisc.contains("htb"));
 
     println!("  peer_table:");
-    match traffic::peer_status_table(cfg) {
+    let peer_table = if cfg.traffic.enabled {
+        match &peers {
+            Ok(peers) => {
+                traffic::peer_status_table_with_state(cfg, &target, transport.as_ref(), peers)
+            }
+            Err(e) => Err(anyhow::anyhow!("peer discovery failed: {e}")),
+        }
+    } else {
+        Ok(None)
+    };
+    match peer_table {
         Ok(Some(rows)) if rows.is_empty() => println!("    no peers discovered"),
         Ok(Some(rows)) => {
             for r in rows {
@@ -168,8 +188,7 @@ pub fn print_status(cfg: &Config) -> anyhow::Result<()> {
     match collect_peer_filter_hits(
         &target.interface_name,
         &cfg.traffic.ifb_name,
-        target.container_pid,
-        &target.container_name,
+        transport.as_ref(),
     ) {
         Ok(map) if map.is_empty() => println!("  filter statistics: unavailable"),
         Ok(map) => {
@@ -182,7 +201,7 @@ pub fn print_status(cfg: &Config) -> anyhow::Result<()> {
 
     print_host_firewall_section(cfg);
     print_docker_amnezia_port_path(&docker_path_diag);
-    print_antitorrent_section(cfg);
+    print_antitorrent_section_with_transport(cfg, Some(transport.as_ref()));
     print_swap_status(cfg);
     print_conntrack_section();
     print_last_actions_section();
@@ -267,6 +286,13 @@ fn print_host_firewall_section(cfg: &Config) {
 }
 
 fn print_antitorrent_section(cfg: &Config) {
+    print_antitorrent_section_with_transport(cfg, None);
+}
+
+fn print_antitorrent_section_with_transport(
+    cfg: &Config,
+    container_transport: Option<&dyn crate::netns::NsTransport>,
+) {
     println!("Anti-torrent:");
     println!("  enabled: {}", cfg.anti_torrent.enabled);
     println!(
@@ -276,7 +302,15 @@ fn print_antitorrent_section(cfg: &Config) {
     if !cfg.anti_torrent.enabled {
         return;
     }
-    match firewall::antitorrent_drift_reason(cfg) {
+    let drift = if cfg.anti_torrent.apply_in_amnezia_netns {
+        match container_transport {
+            Some(transport) => firewall::antitorrent_drift_reason_with_transport(cfg, transport),
+            None => firewall::antitorrent_drift_reason(cfg),
+        }
+    } else {
+        firewall::antitorrent_drift_reason(cfg)
+    };
+    match drift {
         Ok(None) => println!("  drift: no"),
         Ok(Some(reason)) => {
             println!("  drift: yes");
@@ -284,7 +318,15 @@ fn print_antitorrent_section(cfg: &Config) {
         }
         Err(e) => println!("  drift: unknown ({e})"),
     }
-    match firewall::antitorrent_drop_counter(cfg) {
+    let counter = if cfg.anti_torrent.apply_in_amnezia_netns {
+        match container_transport {
+            Some(transport) => firewall::antitorrent_drop_counter_with_transport(cfg, transport),
+            None => firewall::antitorrent_drop_counter(cfg),
+        }
+    } else {
+        firewall::antitorrent_drop_counter(cfg)
+    };
+    match counter {
         Ok(Some((packets, bytes))) => {
             println!("  dropped_packets: {}", packets);
             println!("  dropped_bytes: {}", bytes);
@@ -296,16 +338,25 @@ fn print_antitorrent_section(cfg: &Config) {
 
 fn print_conntrack_section() {
     println!("Conntrack:");
+    let count = read_sysctl_or_na("net.netfilter.nf_conntrack_count");
+    let max = read_sysctl_or_na("net.netfilter.nf_conntrack_max");
+    println!("  nf_conntrack_count: {count}");
+    println!("  nf_conntrack_max: {max}");
     println!(
-        "  nf_conntrack_count: {}",
-        read_sysctl_or_na("net.netfilter.nf_conntrack_count")
-    );
-    println!(
-        "  nf_conntrack_max: {}",
-        read_sysctl_or_na("net.netfilter.nf_conntrack_max")
+        "  nf_conntrack_utilization: {}",
+        conntrack_utilization(&count, &max).unwrap_or_else(|| "n/a".to_string())
     );
 }
 
+fn conntrack_utilization(count: &str, max: &str) -> Option<String> {
+    let count = count.trim().parse::<u64>().ok()?;
+    let max = max.trim().parse::<u64>().ok()?;
+    if max == 0 {
+        return None;
+    }
+    Some(format!("{:.1}%", count as f64 * 100.0 / max as f64))
+}
+
 fn read_sysctl_or_na(key: &str) -> String {
     Command::new("sysctl")
         .args(["-n", key])
@@ -494,14 +545,12 @@ fn format_bytes(v: u64) -> String {
 fn collect_peer_filter_hits(
     iface: &str,
     ifb: &str,
-    pid: u32,
-    container_name: &str,
+    transport: &dyn crate::netns::NsTransport,
 ) -> anyhow::Result<BTreeMap<String, (u64, u64)>> {
-    let transport = build_transport(pid, container_name);
     let egress = transport.run(&["tc", "-s", "filter", "show", "dev", iface, "parent", "1:"])?;
     let ingress = transport.run(&["tc", "-s", "filter", "show", "dev", ifb, "parent", "2:"])?;
-    let egress_hits = parse_tc_filter_hits(&egress);
-    let ingress_hits = parse_tc_filter_hits(&ingress);
+    let egress_hits = traffic::parse_peer_filter_hits(&egress);
+    let ingress_hits = traffic::parse_peer_filter_hits(&ingress);
 
     let mut out: BTreeMap<String, (u64, u64)> = BTreeMap::new();
     for (ip, c) in egress_hits {
@@ -513,74 +562,19 @@ fn collect_peer_filter_hits(
     Ok(out)
 }
 
-fn parse_tc_filter_hits(raw: &str) -> BTreeMap<String, u64> {
-    let mut out = BTreeMap::new();
-    let mut current_ip: Option<String> = None;
-    for line in raw.lines() {
-        let t = line.trim();
-
-        if t.contains("match ") {
-            current_ip = extract_tc_match_ip(t);
-        }
-
-        if t.starts_with("Sent ") {
-            if let Some(ip) = current_ip.clone() {
-                let hits = parse_sent_packets(t).unwrap_or(0);
-                out.insert(ip, hits);
-            }
-        } else if (t.contains(" success ") || t.starts_with("success "))
-            && let Some(ip) = current_ip.clone()
-        {
-            let hits = parse_success_packets(t).unwrap_or(0);
-            out.insert(ip, hits);
-        }
-    }
-    out
-}
-
-fn parse_sent_packets(line: &str) -> Option<u64> {
-    let toks: Vec<&str> = line.split_whitespace().collect();
-    if toks.len() >= 5 && toks[0] == "Sent" && toks[2] == "bytes" && toks[4] == "pkt" {
-        return toks[3].parse::<u64>().ok();
-    }
-    None
-}
-
-fn parse_success_packets(line: &str) -> Option<u64> {
-    let toks: Vec<&str> = line.split_whitespace().collect();
-    let pos = toks.iter().position(|v| *v == "success")?;
-    toks.get(pos + 1)?.parse::<u64>().ok()
-}
-
-fn extract_tc_match_ip(line: &str) -> Option<String> {
-    let toks: Vec<&str> = line.split_whitespace().collect();
-    let pos = toks.iter().position(|v| *v == "match")?;
-    let match_val = *toks.get(pos + 1)?;
-    let (hex, mask) = match_val.split_once('/')?;
-    if !mask.eq_ignore_ascii_case("ffffffff") || hex.len() != 8 {
-        return None;
-    }
-    let b1 = u8::from_str_radix(&hex[0..2], 16).ok()?;
-    let b2 = u8::from_str_radix(&hex[2..4], 16).ok()?;
-    let b3 = u8::from_str_radix(&hex[4..6], 16).ok()?;
-    let b4 = u8::from_str_radix(&hex[6..8], 16).ok()?;
-    Some(format!("{b1}.{b2}.{b3}.{b4}"))
-}
-
 #[cfg(test)]
 mod tests {
-    use super::{parse_sent_packets, parse_success_packets, parse_tc_filter_hits};
+    use super::conntrack_utilization;
+    use crate::traffic;
 
     #[test]
-    fn parse_sent_pkt_counter() {
-        let line = "Sent 12345 bytes 67 pkt (dropped 0, overlimits 0 requeues 0)";
-        assert_eq!(parse_sent_packets(line), Some(67));
-    }
-
-    #[test]
-    fn parse_success_counter() {
-        let line = "match 0a080102/ffffffff at 16 success 42";
-        assert_eq!(parse_success_packets(line), Some(42));
+    fn conntrack_utilization_handles_valid_and_unavailable_values() {
+        assert_eq!(
+            conntrack_utilization("184", "1000").as_deref(),
+            Some("18.4%")
+        );
+        assert_eq!(conntrack_utilization("n/a", "1000"), None);
+        assert_eq!(conntrack_utilization("10", "0"), None);
     }
 
     #[test]
@@ -595,8 +589,20 @@ filter parent 1: protocol ip pref 1 u32 chain 0 fh 800::801 order 2049 key ht 80
   match 0a080103/ffffffff at 16 success 9
 "#;
 
-        let map = parse_tc_filter_hits(raw);
+        let map = traffic::parse_peer_filter_hits(raw);
         assert_eq!(map.get("10.8.1.2"), Some(&7));
         assert_eq!(map.get("10.8.1.3"), Some(&9));
     }
+
+    #[test]
+    fn parse_tc_filter_hits_supports_flower() {
+        let raw = r#"
+filter protocol ip pref 10 flower chain 0
+filter protocol ip pref 10 flower chain 0 handle 0x1 classid 1:10
+  dst_ip 10.8.1.2/32
+  Sent 2048 bytes 17 pkt (dropped 0, overlimits 0 requeues 0)
+"#;
+        let map = traffic::parse_peer_filter_hits(raw);
+        assert_eq!(map.get("10.8.1.2"), Some(&17));
+    }
 }

traffic_core/src/sysctl_tuning.rs

@@ -635,4 +635,16 @@ mod tests {
         assert_eq!(plan.bbr_available, None);
         assert_eq!(plan.bbr_active, None);
     }
+
+    #[test]
+    fn default_render_plan_does_not_manage_netdev_backlog() {
+        let mut c = cfg();
+        assert!(!c.sysctl_tuning.netdev_backlog_enabled);
+        assert_eq!(c.sysctl_tuning.netdev_max_backlog, 10_000);
+        // Keep this assertion independent of host kernel/module availability.
+        c.sysctl_tuning.tcp_bbr = "off".into();
+        let plan = build_render_plan(&c, false).expect("render plan");
+        assert!(!plan.desired.contains_key("net.core.netdev_max_backlog"));
+        assert!(!plan.persisted.contains("net.core.netdev_max_backlog"));
+    }
 }

traffic_core/src/traffic/apply.rs

@@ -6,7 +6,11 @@
 
 use crate::{config::Config, netns::NsTransport, wg};
 
-use super::inspect::{expected_peer_class_map, inspect_and_diff};
+use super::{
+    PEER_FILTER_PREF,
+    inspect::{expected_peer_class_map, inspect_and_diff},
+    parse_tc::{is_not_found_error, normalize_tc_rate},
+};
 
 pub(super) fn teardown_owned_topology(
     transport: &dyn NsTransport,
@@ -26,169 +30,14 @@ pub(super) fn apply_full_topology(
     ifb: &str,
     peers: &[wg::Peer],
 ) -> anyhow::Result<()> {
-    let root = cfg.traffic.root_rate.as_str();
-    run_allow_exists(transport, &["modprobe", "ifb"])?;
-    run_allow_exists(
+    run_step(transport, &["modprobe", "ifb"])?;
+    run_step(
         transport,
         &["ip", "link", "add", "name", ifb, "type", "ifb"],
     )?;
     run_step(transport, &["ip", "link", "set", "dev", ifb, "up"])?;
-    run_allow_exists(
-        transport,
-        &[
-            "tc", "qdisc", "add", "dev", iface, "root", "handle", "1:", "htb",
-        ],
-    )?;
-    run_allow_exists(
-        transport,
-        &[
-            "tc", "class", "add", "dev", iface, "parent", "1:", "classid", "1:1", "htb", "rate",
-            root, "ceil", root,
-        ],
-    )?;
-    run_allow_exists(
-        transport,
-        &[
-            "tc", "qdisc", "add", "dev", iface, "handle", "ffff:", "ingress",
-        ],
-    )?;
-    run_allow_exists(
-        transport,
-        &[
-            "tc", "filter", "add", "dev", iface, "parent", "ffff:", "protocol", "all", "u32",
-            "match", "u32", "0", "0", "action", "mirred", "egress", "redirect", "dev", ifb,
-        ],
-    )?;
-    run_allow_exists(
-        transport,
-        &[
-            "tc", "qdisc", "add", "dev", ifb, "root", "handle", "2:", "htb",
-        ],
-    )?;
-    run_allow_exists(
-        transport,
-        &[
-            "tc", "class", "add", "dev", ifb, "parent", "2:", "classid", "2:1", "htb", "rate",
-            root, "ceil", root,
-        ],
-    )?;
-
-    // The same map the inspector will verify against, so apply and verify cannot disagree
-    // about which class a peer belongs to.
-    let class_map = expected_peer_class_map(peers, cfg.traffic.class_start_id)?;
-
-    for (ip, class_id) in &class_map {
-        let class_egress = format!("1:{class_id}");
-        let class_ingress = format!("2:{class_id}");
-        let limit = cfg.traffic.peer_limit.as_str();
-
-        run_allow_exists(
-            transport,
-            &[
-                "tc",
-                "class",
-                "add",
-                "dev",
-                iface,
-                "parent",
-                "1:1",
-                "classid",
-                class_egress.as_str(),
-                "htb",
-                "rate",
-                limit,
-                "ceil",
-                limit,
-            ],
-        )?;
-        run_allow_exists(
-            transport,
-            &[
-                "tc",
-                "qdisc",
-                "add",
-                "dev",
-                iface,
-                "parent",
-                class_egress.as_str(),
-                "fq_codel",
-            ],
-        )?;
-        run_allow_exists(
-            transport,
-            &[
-                "tc",
-                "filter",
-                "add",
-                "dev",
-                iface,
-                "parent",
-                "1:",
-                "protocol",
-                "ip",
-                "u32",
-                "match",
-                "ip",
-                "dst",
-                ip.as_str(),
-                "flowid",
-                class_egress.as_str(),
-            ],
-        )?;
-        run_allow_exists(
-            transport,
-            &[
-                "tc",
-                "class",
-                "add",
-                "dev",
-                ifb,
-                "parent",
-                "2:1",
-                "classid",
-                class_ingress.as_str(),
-                "htb",
-                "rate",
-                limit,
-                "ceil",
-                limit,
-            ],
-        )?;
-        run_allow_exists(
-            transport,
-            &[
-                "tc",
-                "qdisc",
-                "add",
-                "dev",
-                ifb,
-                "parent",
-                class_ingress.as_str(),
-                "fq_codel",
-            ],
-        )?;
-        run_allow_exists(
-            transport,
-            &[
-                "tc",
-                "filter",
-                "add",
-                "dev",
-                ifb,
-                "parent",
-                "2:",
-                "protocol",
-                "ip",
-                "u32",
-                "match",
-                "ip",
-                "src",
-                ip.as_str(),
-                "flowid",
-                class_ingress.as_str(),
-            ],
-        )?;
-    }
+    let batch = build_tc_batch_script(cfg, iface, ifb, peers)?;
+    run_tc_batch(transport, &batch)?;
 
     let verify = inspect_and_diff(
         transport,
@@ -206,28 +55,125 @@ pub(super) fn apply_full_topology(
     Ok(())
 }
 
-fn run_step(transport: &dyn NsTransport, args: &[&str]) -> anyhow::Result<()> {
-    let _ = transport.run(args)?;
+fn run_tc_batch(transport: &dyn NsTransport, batch: &str) -> anyhow::Result<()> {
+    transport
+        .run_with_stdin(&["tc", "-batch", "-"], batch)
+        .map_err(|e| anyhow::anyhow!("tc batch apply failed: {e}"))?;
     Ok(())
 }
 
-/// Tolerates only the idempotency conflicts that a rebuild legitimately produces.
-/// Anything else is a real apply failure and must propagate.
-fn run_allow_exists(transport: &dyn NsTransport, args: &[&str]) -> anyhow::Result<()> {
-    match transport.run(args) {
-        Ok(_) => Ok(()),
-        Err(e) => {
-            let msg = e.to_string().to_ascii_lowercase();
-            if msg.contains("file exists")
-                || msg.contains("eexist")
-                || msg.contains("object already exists")
-                || msg.contains("exclusive flag on")
-            {
-                return Ok(());
+pub(super) fn empty_topology_drift_reason(
+    transport: &dyn NsTransport,
+    iface: &str,
+    ifb: &str,
+) -> anyhow::Result<Option<String>> {
+    let qdiscs = transport
+        .run(&["tc", "qdisc", "show", "dev", iface])
+        .map_err(|e| anyhow::anyhow!("empty topology qdisc inspect failed: {e}"))?;
+    if qdiscs.lines().any(|line| {
+        let line = line.trim();
+        (line.contains("qdisc htb 1:") && line.contains(" root"))
+            || line.contains("qdisc ingress ffff:")
+    }) {
+        return Ok(Some("owned tc topology exists without peers".to_string()));
     }
-            Err(e)
+
+    match transport.run(&["ip", "link", "show", ifb]) {
+        Ok(out) if !out.trim().is_empty() => {
+            Ok(Some(format!("owned IFB {ifb} exists without peers")))
+        }
+        Ok(_) => Ok(None),
+        Err(e) if is_not_found_error(&e.to_string()) => Ok(None),
+        Err(e) => Err(anyhow::anyhow!("empty topology IFB inspect failed: {e}")),
     }
 }
+
+fn build_tc_batch_script(
+    cfg: &Config,
+    iface: &str,
+    ifb: &str,
+    peers: &[wg::Peer],
+) -> anyhow::Result<String> {
+    let root = normalize_tc_rate(&cfg.traffic.root_rate);
+    let limit = normalize_tc_rate(&cfg.traffic.peer_limit);
+    let class_map = expected_peer_class_map(peers, cfg.traffic.class_start_id)?;
+    let mut batch = String::new();
+
+    push_command(
+        &mut batch,
+        format!("qdisc add dev {iface} root handle 1: htb"),
+    );
+    push_command(
+        &mut batch,
+        format!("class add dev {iface} parent 1: classid 1:1 htb rate {root} ceil {root}"),
+    );
+    push_command(
+        &mut batch,
+        format!("qdisc add dev {iface} handle ffff: ingress"),
+    );
+    push_command(
+        &mut batch,
+        format!(
+            "filter add dev {iface} parent ffff: protocol all u32 match u32 0 0 action mirred egress redirect dev {ifb}"
+        ),
+    );
+    push_command(
+        &mut batch,
+        format!("qdisc add dev {ifb} root handle 2: htb"),
+    );
+    push_command(
+        &mut batch,
+        format!("class add dev {ifb} parent 2: classid 2:1 htb rate {root} ceil {root}"),
+    );
+
+    for (ip, class_id) in class_map {
+        let egress = format!("1:{class_id}");
+        let ingress = format!("2:{class_id}");
+        push_command(
+            &mut batch,
+            format!(
+                "class add dev {iface} parent 1:1 classid {egress} htb rate {limit} ceil {limit}"
+            ),
+        );
+        push_command(
+            &mut batch,
+            format!("qdisc add dev {iface} parent {egress} fq_codel"),
+        );
+        push_command(
+            &mut batch,
+            format!(
+                "filter add dev {iface} parent 1: protocol ip pref {PEER_FILTER_PREF} flower dst_ip {ip}/32 classid {egress}"
+            ),
+        );
+        push_command(
+            &mut batch,
+            format!(
+                "class add dev {ifb} parent 2:1 classid {ingress} htb rate {limit} ceil {limit}"
+            ),
+        );
+        push_command(
+            &mut batch,
+            format!("qdisc add dev {ifb} parent {ingress} fq_codel"),
+        );
+        push_command(
+            &mut batch,
+            format!(
+                "filter add dev {ifb} parent 2: protocol ip pref {PEER_FILTER_PREF} flower src_ip {ip}/32 classid {ingress}"
+            ),
+        );
+    }
+
+    Ok(batch)
+}
+
+fn push_command(batch: &mut String, command: String) {
+    batch.push_str(&command);
+    batch.push('\n');
+}
+
+fn run_step(transport: &dyn NsTransport, args: &[&str]) -> anyhow::Result<()> {
+    let _ = transport.run(args)?;
+    Ok(())
 }
 
 /// Deleting an object that is already gone is the expected outcome of teardown, not an
@@ -252,3 +198,155 @@ fn run_allow_notfound(transport: &dyn NsTransport, args: &[&str]) -> anyhow::Res
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::{net::Ipv4Addr, sync::Mutex};
+
+    #[derive(Default)]
+    struct BatchTransport {
+        call: Mutex<Option<(Vec<String>, String)>>,
+        fail: bool,
+    }
+
+    impl NsTransport for BatchTransport {
+        fn run(&self, _args: &[&str]) -> anyhow::Result<String> {
+            anyhow::bail!("unexpected run")
+        }
+
+        fn run_with_stdin(&self, args: &[&str], input: &str) -> anyhow::Result<String> {
+            *self.call.lock().expect("call lock") = Some((
+                args.iter().map(|v| (*v).to_string()).collect(),
+                input.to_string(),
+            ));
+            if self.fail {
+                anyhow::bail!("Command failed -:4")
+            }
+            Ok(String::new())
+        }
+    }
+
+    struct EmptyTopologyTransport {
+        qdiscs: &'static str,
+        ifb: anyhow::Result<&'static str>,
+    }
+
+    impl NsTransport for EmptyTopologyTransport {
+        fn run(&self, args: &[&str]) -> anyhow::Result<String> {
+            match args {
+                ["tc", "qdisc", "show", "dev", "awg0"] => Ok(self.qdiscs.to_string()),
+                ["ip", "link", "show", "ifb0"] => self
+                    .ifb
+                    .as_ref()
+                    .map(|out| (*out).to_string())
+                    .map_err(|e| anyhow::anyhow!(e.to_string())),
+                _ => anyhow::bail!("unexpected command: {args:?}"),
+            }
+        }
+
+        fn run_with_stdin(&self, args: &[&str], _input: &str) -> anyhow::Result<String> {
+            anyhow::bail!("unexpected stdin command: {args:?}")
+        }
+    }
+
+    fn peers(ips: &[&str]) -> Vec<wg::Peer> {
+        ips.iter()
+            .map(|ip| wg::Peer {
+                ipv4: ip.parse::<Ipv4Addr>().expect("peer IP"),
+            })
+            .collect()
+    }
+
+    #[test]
+    fn batch_uses_one_shared_flower_pref_and_keeps_redirect_u32() {
+        let mut cfg = crate::firewall::tests::test_config();
+        cfg.traffic.root_rate = " 1 Gbit ".into();
+        cfg.traffic.peer_limit = " 50 Mbit ".into();
+        let script = build_tc_batch_script(&cfg, "awg0", "ifb0", &peers(&["10.8.1.2", "10.8.1.3"]))
+            .expect("render batch");
+
+        assert_eq!(script.lines().count(), 18, "six base and six per peer");
+        assert_eq!(script.matches(" pref 10 flower ").count(), 4);
+        assert_eq!(
+            script.lines().filter(|line| line.contains(" u32 ")).count(),
+            1,
+            "redirect only"
+        );
+        assert!(script.contains("flower dst_ip 10.8.1.2/32 classid 1:10"));
+        assert!(script.contains("flower src_ip 10.8.1.2/32 classid 2:10"));
+        assert!(script.contains("rate 1gbit ceil 1gbit"));
+        assert!(script.contains("rate 50mbit ceil 50mbit"));
+        assert!(!script.lines().any(|line| line.starts_with("tc ")));
+    }
+
+    #[test]
+    fn batch_runner_uses_stdin_filename_and_never_force() {
+        let transport = BatchTransport::default();
+        run_tc_batch(&transport, "qdisc add dev awg0 root handle 1: htb\n")
+            .expect("batch succeeds");
+        let (args, input) = transport
+            .call
+            .lock()
+            .expect("call lock")
+            .clone()
+            .expect("recorded call");
+        assert_eq!(args, ["tc", "-batch", "-"]);
+        assert!(!args.iter().any(|arg| arg == "-force"));
+        assert!(input.starts_with("qdisc add"));
+    }
+
+    #[test]
+    fn batch_failure_preserves_tc_line_diagnostic() {
+        let transport = BatchTransport {
+            fail: true,
+            ..BatchTransport::default()
+        };
+        let err = run_tc_batch(&transport, "bad command\n").expect_err("must fail");
+        assert!(err.to_string().contains("Command failed -:4"));
+    }
+
+    #[test]
+    fn empty_peer_state_detects_owned_tc_topology_for_cleanup() {
+        let transport = EmptyTopologyTransport {
+            qdiscs: "qdisc htb 1: root refcnt 2 r2q 10 default 0x30\nqdisc ingress ffff: parent ffff:fff1",
+            ifb: Err(anyhow::anyhow!("must not inspect IFB after owned qdisc")),
+        };
+
+        let reason = empty_topology_drift_reason(&transport, "awg0", "ifb0")
+            .expect("inspect empty desired state");
+        assert_eq!(
+            reason.as_deref(),
+            Some("owned tc topology exists without peers")
+        );
+    }
+
+    #[test]
+    fn empty_peer_state_is_idempotent_after_owned_topology_is_gone() {
+        let transport = EmptyTopologyTransport {
+            qdiscs: "qdisc noqueue 0: root refcnt 2",
+            ifb: Err(anyhow::anyhow!("Cannot find device \"ifb0\"")),
+        };
+
+        assert_eq!(
+            empty_topology_drift_reason(&transport, "awg0", "ifb0")
+                .expect("inspect clean empty desired state"),
+            None
+        );
+    }
+
+    #[test]
+    fn empty_peer_state_detects_leftover_ifb_for_cleanup() {
+        let transport = EmptyTopologyTransport {
+            qdiscs: "qdisc noqueue 0: root refcnt 2",
+            ifb: Ok("7: ifb0: <BROADCAST,NOARP,UP,LOWER_UP> mtu 1500"),
+        };
+
+        let reason =
+            empty_topology_drift_reason(&transport, "awg0", "ifb0").expect("inspect leftover IFB");
+        assert_eq!(
+            reason.as_deref(),
+            Some("owned IFB ifb0 exists without peers")
+        );
+    }
+}

traffic_core/src/traffic/inspect.rs

@@ -9,7 +9,7 @@ use std::collections::{BTreeMap, BTreeSet};
 use crate::{netns::NsTransport, wg};
 
 use super::{
-    TrafficDrift,
+    FilterClassifier, FilterDirection, FilterInspect, PEER_FILTER_PREF, TrafficDrift,
     parse_tc::{
         collect_leaf_class_ids, extract_filter_ip_class_map, has_fq_codel_for_parent,
         has_ingress_redirect_to_ifb, is_not_found_error, link_is_up, normalize_tc_rate,
@@ -27,24 +27,19 @@ pub(super) fn inspect_and_diff(
     peer_limit: &str,
     peers: &[wg::Peer],
 ) -> anyhow::Result<TrafficDrift> {
-    let root_qdisc = inspect_cmd(
+    let iface_qdiscs = inspect_cmd(
         transport,
         &["tc", "qdisc", "show", "dev", iface],
-        "egress qdisc",
+        "interface qdiscs",
     )?;
-    if !root_qdisc.contains("htb") {
+    if !iface_qdiscs.contains("htb") {
         return Ok(TrafficDrift {
             has_drift: true,
             reason: "missing egress root htb qdisc".to_string(),
         });
     }
 
-    let ingress_qdisc = inspect_cmd(
-        transport,
-        &["tc", "qdisc", "show", "dev", iface],
-        "ingress qdisc",
-    )?;
-    if !ingress_qdisc.contains("ingress") {
+    if !iface_qdiscs.contains("ingress") {
         return Ok(TrafficDrift {
             has_drift: true,
             reason: "missing ingress qdisc".to_string(),
@@ -99,6 +94,22 @@ pub(super) fn inspect_and_diff(
 
     let inspect_egress = extract_filter_ip_class_map(&filters_egress);
     let inspect_ingress = extract_filter_ip_class_map(&filters_ingress);
+    if let Some(reason) =
+        filter_contract_drift_reason(&inspect_egress, FilterDirection::Destination, "egress")
+    {
+        return Ok(TrafficDrift {
+            has_drift: true,
+            reason,
+        });
+    }
+    if let Some(reason) =
+        filter_contract_drift_reason(&inspect_ingress, FilterDirection::Source, "ingress")
+    {
+        return Ok(TrafficDrift {
+            has_drift: true,
+            reason,
+        });
+    }
     if !inspect_egress.duplicate_ips.is_empty() {
         return Ok(TrafficDrift {
             has_drift: true,
@@ -240,36 +251,21 @@ pub(super) fn inspect_and_diff(
         }
     }
 
-    let classes_egress = inspect_cmd(
-        transport,
-        &["tc", "class", "show", "dev", iface, "parent", "1:1"],
-        "egress leaf classes",
-    )?;
-    let classes_ingress = inspect_cmd(
-        transport,
-        &["tc", "class", "show", "dev", ifb, "parent", "2:1"],
-        "ingress leaf classes",
-    )?;
     let expected_set: BTreeSet<u16> = expected_map.values().copied().collect();
-    if collect_leaf_class_ids(&classes_egress) != expected_set {
+    if collect_leaf_class_ids(&classes_egress_full) != expected_set {
         return Ok(TrafficDrift {
             has_drift: true,
             reason: "egress per-peer class set mismatch".to_string(),
         });
     }
-    if collect_leaf_class_ids(&classes_ingress) != expected_set {
+    if collect_leaf_class_ids(&classes_ingress_full) != expected_set {
         return Ok(TrafficDrift {
             has_drift: true,
             reason: "ingress per-peer class set mismatch".to_string(),
         });
     }
 
-    let qdisc_egress = inspect_cmd(
-        transport,
-        &["tc", "qdisc", "show", "dev", iface],
-        "egress leaf qdisc",
-    )?;
-    let qdisc_ingress = inspect_cmd(
+    let ifb_qdiscs = inspect_cmd(
         transport,
         &["tc", "qdisc", "show", "dev", ifb],
         "ingress leaf qdisc",
@@ -277,13 +273,13 @@ pub(super) fn inspect_and_diff(
     for class_id in expected_map.values() {
         let class_egress = format!("1:{class_id}");
         let class_ingress = format!("2:{class_id}");
-        if !has_fq_codel_for_parent(&qdisc_egress, &class_egress) {
+        if !has_fq_codel_for_parent(&iface_qdiscs, &class_egress) {
             return Ok(TrafficDrift {
                 has_drift: true,
                 reason: format!("missing fq_codel for egress class {class_egress}"),
             });
         }
-        if !has_fq_codel_for_parent(&qdisc_ingress, &class_ingress) {
+        if !has_fq_codel_for_parent(&ifb_qdiscs, &class_ingress) {
             return Ok(TrafficDrift {
                 has_drift: true,
                 reason: format!("missing fq_codel for ingress class {class_ingress}"),
@@ -297,6 +293,46 @@ pub(super) fn inspect_and_diff(
     })
 }
 
+fn filter_contract_drift_reason(
+    inspect: &FilterInspect,
+    direction: FilterDirection,
+    label: &str,
+) -> Option<String> {
+    if inspect.unrecognized_rules > 0 {
+        return Some(format!(
+            "{label} contains {} unrecognized peer filter rule(s)",
+            inspect.unrecognized_rules
+        ));
+    }
+    for filter in &inspect.filters {
+        if filter.classifier != FilterClassifier::Flower {
+            return Some(format!(
+                "{label} peer filter for {} uses legacy u32 instead of flower",
+                filter.ip
+            ));
+        }
+        if filter.pref != Some(PEER_FILTER_PREF) {
+            return Some(format!(
+                "{label} flower filter for {} has pref {:?}, expected {}",
+                filter.ip, filter.pref, PEER_FILTER_PREF
+            ));
+        }
+        if filter.direction != direction {
+            return Some(format!(
+                "{label} flower filter for {} matches the wrong IP direction",
+                filter.ip
+            ));
+        }
+        if filter.prefix != 32 {
+            return Some(format!(
+                "{label} flower filter for {} uses /{}, expected /32",
+                filter.ip, filter.prefix
+            ));
+        }
+    }
+    None
+}
+
 /// The class-id contract: peer IP strings sorted lexicographically, then
 /// `class_start_id + index`.
 ///
@@ -327,7 +363,47 @@ fn inspect_cmd(transport: &dyn NsTransport, args: &[&str], what: &str) -> anyhow
 #[cfg(test)]
 mod tests {
     use super::*;
-    use std::net::Ipv4Addr;
+    use std::{net::Ipv4Addr, sync::Mutex};
+
+    #[derive(Default)]
+    struct HealthyTransport {
+        calls: Mutex<Vec<String>>,
+    }
+
+    impl NsTransport for HealthyTransport {
+        fn run(&self, args: &[&str]) -> anyhow::Result<String> {
+            let key = args.join(" ");
+            self.calls.lock().expect("calls lock").push(key.clone());
+            let out = match key.as_str() {
+                "tc qdisc show dev awg0" => {
+                    "qdisc htb 1: root\nqdisc ingress ffff: parent ffff:fff1\nqdisc fq_codel 8001: parent 1:10\n"
+                }
+                "tc filter show dev awg0 parent ffff:" => {
+                    "filter parent ffff: protocol all pref 49152 u32\n action order 1: mirred egress redirect dev ifb0\n"
+                }
+                "ip link show ifb0" => "2: ifb0: <BROADCAST,NOARP,UP> mtu 1500 state UP\n",
+                "tc filter show dev awg0 parent 1:" => {
+                    "filter protocol ip pref 10 flower chain 0\nfilter protocol ip pref 10 flower chain 0 handle 0x1 classid 1:10\n  dst_ip 10.8.1.2/32\n"
+                }
+                "tc filter show dev ifb0 parent 2:" => {
+                    "filter protocol ip pref 10 flower chain 0\nfilter protocol ip pref 10 flower chain 0 handle 0x1 classid 2:10\n  src_ip 10.8.1.2/32\n"
+                }
+                "tc class show dev awg0" => {
+                    "class htb 1:1 root rate 1000Mbit ceil 1000Mbit\nclass htb 1:10 parent 1:1 rate 70Mbit ceil 70Mbit\n"
+                }
+                "tc class show dev ifb0" => {
+                    "class htb 2:1 root rate 1000Mbit ceil 1000Mbit\nclass htb 2:10 parent 2:1 rate 70Mbit ceil 70Mbit\n"
+                }
+                "tc qdisc show dev ifb0" => "qdisc htb 2: root\nqdisc fq_codel 8002: parent 2:10\n",
+                _ => anyhow::bail!("unexpected inspect command: {key}"),
+            };
+            Ok(out.to_string())
+        }
+
+        fn run_with_stdin(&self, _args: &[&str], _input: &str) -> anyhow::Result<String> {
+            anyhow::bail!("unexpected stdin command")
+        }
+    }
 
     fn peers(ips: &[&str]) -> Vec<wg::Peer> {
         ips.iter()
@@ -364,4 +440,49 @@ mod tests {
     fn empty_peer_list_yields_empty_map() {
         assert!(expected_peer_class_map(&[], 10).expect("map").is_empty());
     }
+
+    #[test]
+    fn healthy_inspection_reuses_full_qdisc_and_class_outputs() {
+        let transport = HealthyTransport::default();
+        let drift = inspect_and_diff(
+            &transport,
+            "awg0",
+            "ifb0",
+            10,
+            "1000mbit",
+            "70mbit",
+            &peers(&["10.8.1.2"]),
+        )
+        .expect("inspect");
+        assert!(!drift.has_drift, "{}", drift.reason);
+
+        let calls = transport.calls.lock().expect("calls lock");
+        assert_eq!(calls.len(), 8);
+        assert_eq!(
+            calls
+                .iter()
+                .filter(|call| *call == "tc qdisc show dev awg0")
+                .count(),
+            1
+        );
+        assert_eq!(
+            calls
+                .iter()
+                .filter(|call| call.starts_with("tc class show dev"))
+                .count(),
+            2
+        );
+        assert!(!calls.iter().any(|call| call.contains(" parent 1:1")));
+        assert!(!calls.iter().any(|call| call.contains(" parent 2:1")));
+    }
+
+    #[test]
+    fn legacy_u32_peer_filter_requires_migration() {
+        let inspect = extract_filter_ip_class_map(
+            "filter parent 1: protocol ip pref 10 u32\n  match 0a080102/ffffffff at 16\n  flowid 1:10",
+        );
+        let reason = filter_contract_drift_reason(&inspect, FilterDirection::Destination, "egress")
+            .expect("legacy drift");
+        assert!(reason.contains("legacy u32 instead of flower"));
+    }
 }

traffic_core/src/traffic/mod.rs

@@ -16,13 +16,18 @@ use std::{
     time::Duration,
 };
 
-use apply::{apply_full_topology, teardown_owned_topology};
+use apply::{apply_full_topology, empty_topology_drift_reason, teardown_owned_topology};
 use inspect::{expected_peer_class_map, inspect_and_diff};
 use parse_tc::{
     extract_filter_ip_class_map, normalize_tc_rate, parse_class_rate_ceil_map,
     rate_ceil_matches_limit,
 };
 
+/// All peer filters of one direction deliberately share one flower priority and mask.
+/// A different priority would create another classifier lookup before the packet reaches
+/// the matching filter and would defeat the purpose of replacing the linear u32 list.
+pub(super) const PEER_FILTER_PREF: u16 = 10;
+
 #[derive(Debug, Clone)]
 struct TrafficDrift {
     has_drift: bool,
@@ -31,9 +36,34 @@ struct TrafficDrift {
 
 #[derive(Debug, Clone)]
 struct FilterInspect {
+    filters: Vec<ParsedPeerFilter>,
     map: BTreeMap<String, u16>,
     duplicate_ips: BTreeSet<String>,
     duplicate_classes: BTreeSet<u16>,
+    unrecognized_rules: usize,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum FilterClassifier {
+    Flower,
+    U32,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum FilterDirection {
+    Source,
+    Destination,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+struct ParsedPeerFilter {
+    ip: String,
+    prefix: u8,
+    class_id: u16,
+    classifier: FilterClassifier,
+    direction: FilterDirection,
+    pref: Option<u16>,
+    packets: Option<u64>,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -55,27 +85,32 @@ pub struct PeerTrafficStatus {
     pub drift_reason: Option<String>,
 }
 
-pub fn inspect_drift_summary(cfg: &Config) -> anyhow::Result<Option<String>> {
+pub(crate) fn inspect_drift_summary_with_state(
+    cfg: &Config,
+    target: &docker::RuntimeTarget,
+    transport: &dyn crate::netns::NsTransport,
+    peers: &[wg::Peer],
+) -> anyhow::Result<Option<String>> {
     if !cfg.traffic.enabled {
         return Ok(None);
     }
 
-    let target = docker::discover_target(cfg)?;
-    let transport = build_transport(target.container_pid, &target.container_name);
-    let peers = wg::discover_peers(cfg, &target)?;
-
     if peers.is_empty() {
-        return Ok(Some("no peers discovered".to_string()));
+        return empty_topology_drift_reason(
+            transport,
+            &target.interface_name,
+            &cfg.traffic.ifb_name,
+        );
     }
 
     let drift = inspect_and_diff(
-        transport.as_ref(),
+        transport,
         &target.interface_name,
         &cfg.traffic.ifb_name,
         cfg.traffic.class_start_id,
         &cfg.traffic.root_rate,
         &cfg.traffic.peer_limit,
-        &peers,
+        peers,
     )?;
 
     if drift.has_drift {
@@ -93,11 +128,40 @@ pub fn reconcile_traffic(cfg: &Config) -> anyhow::Result<()> {
     let _lock =
         FileLock::acquire_timeout("/run/awg2-harness/traffic.lock", Duration::from_secs(10))?;
     let target = docker::discover_target(cfg)?;
+    reconcile_traffic_locked(cfg, &target)
+}
+
+pub(crate) fn reconcile_traffic_with_target(
+    cfg: &Config,
+    target: &docker::RuntimeTarget,
+) -> anyhow::Result<()> {
+    if !cfg.traffic.enabled {
+        return Ok(());
+    }
+    let _lock =
+        FileLock::acquire_timeout("/run/awg2-harness/traffic.lock", Duration::from_secs(10))?;
+    reconcile_traffic_locked(cfg, target)
+}
+
+fn reconcile_traffic_locked(cfg: &Config, target: &docker::RuntimeTarget) -> anyhow::Result<()> {
     let transport = build_transport(target.container_pid, &target.container_name);
-    let peers = wg::discover_peers(cfg, &target)?;
+    let peers = wg::discover_peers(cfg, target)?;
 
     if peers.is_empty() {
-        println!("no peers found, skip tc apply");
+        if let Some(reason) = empty_topology_drift_reason(
+            transport.as_ref(),
+            &target.interface_name,
+            &cfg.traffic.ifb_name,
+        )? {
+            tracing::info!(target: "traffic", "{reason}; removing owned topology");
+            teardown_owned_topology(
+                transport.as_ref(),
+                &target.interface_name,
+                &cfg.traffic.ifb_name,
+            )?;
+        } else {
+            tracing::debug!(target: "traffic", "no peers and no owned topology");
+        }
         return Ok(());
     }
 
@@ -112,11 +176,11 @@ pub fn reconcile_traffic(cfg: &Config) -> anyhow::Result<()> {
     )?;
 
     if !drift.has_drift {
-        println!("traffic rules up-to-date, skip apply");
+        tracing::debug!(target: "traffic", "traffic rules are up-to-date");
         return Ok(());
     }
 
-    println!("traffic drift detected: {}", drift.reason);
+    tracing::info!(target: "traffic", reason = %drift.reason, "traffic drift detected");
 
     teardown_owned_topology(
         transport.as_ref(),
@@ -134,14 +198,15 @@ pub fn reconcile_traffic(cfg: &Config) -> anyhow::Result<()> {
     Ok(())
 }
 
-pub fn peer_status_table(cfg: &Config) -> anyhow::Result<Option<Vec<PeerTrafficStatus>>> {
+pub(crate) fn peer_status_table_with_state(
+    cfg: &Config,
+    target: &docker::RuntimeTarget,
+    transport: &dyn crate::netns::NsTransport,
+    peers: &[wg::Peer],
+) -> anyhow::Result<Option<Vec<PeerTrafficStatus>>> {
     if !cfg.traffic.enabled {
         return Ok(None);
     }
-
-    let target = docker::discover_target(cfg)?;
-    let transport = build_transport(target.container_pid, &target.container_name);
-    let peers = wg::discover_peers(cfg, &target)?;
     if peers.is_empty() {
         return Ok(Some(Vec::new()));
     }
@@ -181,7 +246,7 @@ pub fn peer_status_table(cfg: &Config) -> anyhow::Result<Option<Vec<PeerTrafficS
     let actual_map_ingress = inspect_ingress.map;
     let egress_rates = parse_class_rate_ceil_map(&classes_egress_full, "1:");
     let ingress_rates = parse_class_rate_ceil_map(&classes_ingress_full, "2:");
-    let expected_map = expected_peer_class_map(&peers, cfg.traffic.class_start_id)?;
+    let expected_map = expected_peer_class_map(peers, cfg.traffic.class_start_id)?;
     let expected_limit = normalize_tc_rate(&cfg.traffic.peer_limit);
 
     let mut out = Vec::with_capacity(peers.len());
@@ -225,6 +290,22 @@ pub fn peer_status_table(cfg: &Config) -> anyhow::Result<Option<Vec<PeerTrafficS
         if inspect_ingress.duplicate_ips.contains(&ip) {
             reasons.push("duplicate ingress filters".to_string());
         }
+        if let Some(filter) = inspect_egress
+            .filters
+            .iter()
+            .find(|filter| filter.ip == ip && filter.class_id == class_id)
+            && !peer_filter_contract_matches(filter, FilterDirection::Destination)
+        {
+            reasons.push("egress filter is not flower/pref 10/dst /32".to_string());
+        }
+        if let Some(filter) = inspect_ingress
+            .filters
+            .iter()
+            .find(|filter| filter.ip == ip && filter.class_id == class_id)
+            && !peer_filter_contract_matches(filter, FilterDirection::Source)
+        {
+            reasons.push("ingress filter is not flower/pref 10/src /32".to_string());
+        }
 
         out.push(PeerTrafficStatus {
             ip,
@@ -245,3 +326,14 @@ pub fn peer_status_table(cfg: &Config) -> anyhow::Result<Option<Vec<PeerTrafficS
 
     Ok(Some(out))
 }
+
+fn peer_filter_contract_matches(filter: &ParsedPeerFilter, direction: FilterDirection) -> bool {
+    filter.classifier == FilterClassifier::Flower
+        && filter.pref == Some(PEER_FILTER_PREF)
+        && filter.direction == direction
+        && filter.prefix == 32
+}
+
+pub(crate) fn parse_peer_filter_hits(raw: &str) -> BTreeMap<String, u64> {
+    parse_tc::parse_peer_filter_hits(raw)
+}

traffic_core/src/traffic/parse_tc.rs

@@ -8,7 +8,7 @@ use std::{
     net::Ipv4Addr,
 };
 
-use super::{ClassRateCeil, FilterInspect};
+use super::{ClassRateCeil, FilterClassifier, FilterDirection, FilterInspect, ParsedPeerFilter};
 
 /// Maps peer IP to class id from `tc filter show`, and reports both kinds of duplicate.
 ///
@@ -16,43 +16,196 @@ use super::{ClassRateCeil, FilterInspect};
 /// the IP and flowid are taken from within one block. Scanning tokens across unrelated
 /// lines would happily pair an IP from one filter with a flowid from the next.
 pub(super) fn extract_filter_ip_class_map(filters: &str) -> FilterInspect {
+    let mut parsed = Vec::new();
     let mut out = BTreeMap::new();
     let mut duplicate_ips = BTreeSet::new();
     let mut class_to_ip = BTreeMap::new();
     let mut duplicate_classes = BTreeSet::new();
+    let mut unrecognized_rules = 0;
 
     for block in split_filter_blocks(filters) {
-        let mut ip: Option<String> = None;
-        let mut class_id: Option<u16> = None;
-
-        for line in block.lines() {
-            let t = line.trim();
-            if t.contains("match") && ip.is_none() {
-                ip = extract_u32_match_hex(t).and_then(|hex| hex_to_ipv4_string(&hex));
+        if let Some(filter) = parse_peer_filter_block(&block) {
+            if out.contains_key(&filter.ip) {
+                duplicate_ips.insert(filter.ip.clone());
             }
-            if t.contains("flowid") && class_id.is_none() {
-                class_id = extract_flowid_class_id(t);
+            if class_to_ip
+                .insert(filter.class_id, filter.ip.clone())
+                .is_some()
+            {
+                duplicate_classes.insert(filter.class_id);
             }
-        }
-
-        if let (Some(ip), Some(class_id)) = (ip, class_id) {
-            if out.contains_key(&ip) {
-                duplicate_ips.insert(ip.clone());
-            }
-            if class_to_ip.insert(class_id, ip.clone()).is_some() {
-                duplicate_classes.insert(class_id);
-            }
-            out.insert(ip, class_id);
+            out.insert(filter.ip.clone(), filter.class_id);
+            parsed.push(filter);
+        } else if block_looks_like_rule(&block) {
+            unrecognized_rules += 1;
         }
     }
 
     FilterInspect {
+        filters: parsed,
         map: out,
         duplicate_ips,
         duplicate_classes,
+        unrecognized_rules,
     }
 }
 
+pub(super) fn parse_peer_filter_hits(filters: &str) -> BTreeMap<String, u64> {
+    extract_filter_ip_class_map(filters)
+        .filters
+        .into_iter()
+        .filter_map(|filter| filter.packets.map(|packets| (filter.ip, packets)))
+        .collect()
+}
+
+fn parse_peer_filter_block(block: &str) -> Option<ParsedPeerFilter> {
+    let lower = block.to_ascii_lowercase();
+    let classifier = if lower.split_whitespace().any(|token| token == "flower") {
+        FilterClassifier::Flower
+    } else if lower.split_whitespace().any(|token| token == "u32") || lower.contains("match ") {
+        FilterClassifier::U32
+    } else {
+        return None;
+    };
+    let pref = extract_token_u16(block, "pref");
+    let class_id = extract_class_id(block)?;
+    let packets = extract_packet_count(block);
+
+    let (ip, prefix, direction) = match classifier {
+        FilterClassifier::Flower => extract_flower_ip(block)?,
+        FilterClassifier::U32 => extract_u32_ip(block)?,
+    };
+
+    Some(ParsedPeerFilter {
+        ip,
+        prefix,
+        class_id,
+        classifier,
+        direction,
+        pref,
+        packets,
+    })
+}
+
+fn block_looks_like_rule(block: &str) -> bool {
+    let lower = block.to_ascii_lowercase();
+    lower.contains("classid ")
+        || lower.contains("flowid ")
+        || lower.contains("dst_ip ")
+        || lower.contains("src_ip ")
+        || lower.contains("match ")
+        || lower.contains(" action ")
+        || lower.contains("\nsent ")
+        || lower.lines().next().is_some_and(|line| {
+            line.split_whitespace().any(|token| token == "handle")
+                && line
+                    .split_whitespace()
+                    .any(|token| token == "flower" || token == "u32")
+        })
+}
+
+fn extract_flower_ip(block: &str) -> Option<(String, u8, FilterDirection)> {
+    for line in block.lines() {
+        let tokens: Vec<&str> = line.split_whitespace().collect();
+        for (key, direction) in [
+            ("src_ip", FilterDirection::Source),
+            ("dst_ip", FilterDirection::Destination),
+        ] {
+            let Some(pos) = tokens.iter().position(|token| *token == key) else {
+                continue;
+            };
+            let raw = *tokens.get(pos + 1)?;
+            let (addr, prefix) = match raw.split_once('/') {
+                Some((addr, prefix)) => (addr, prefix.parse::<u8>().ok()?),
+                None => (raw, 32),
+            };
+            let ip = addr.parse::<Ipv4Addr>().ok()?;
+            if prefix > 32 {
+                return None;
+            }
+            return Some((ip.to_string(), prefix, direction));
+        }
+    }
+    None
+}
+
+fn extract_u32_ip(block: &str) -> Option<(String, u8, FilterDirection)> {
+    for line in block.lines() {
+        let tokens: Vec<&str> = line.split_whitespace().collect();
+        let Some(pos) = tokens.iter().position(|token| *token == "match") else {
+            continue;
+        };
+        let (hex, mask) = tokens.get(pos + 1)?.split_once('/')?;
+        if hex.len() != 8 || mask.len() != 8 {
+            continue;
+        }
+        let prefix = mask_hex_to_prefix(mask)?;
+        let at = tokens
+            .iter()
+            .position(|token| *token == "at")
+            .and_then(|idx| tokens.get(idx + 1))
+            .and_then(|raw| raw.parse::<u16>().ok())?;
+        let direction = match at {
+            12 => FilterDirection::Source,
+            16 => FilterDirection::Destination,
+            _ => continue,
+        };
+        return Some((hex_to_ipv4_string(hex)?, prefix, direction));
+    }
+    None
+}
+
+fn mask_hex_to_prefix(mask: &str) -> Option<u8> {
+    let mask = u32::from_str_radix(mask, 16).ok()?;
+    let prefix = mask.leading_ones() as u8;
+    let expected = if prefix == 0 {
+        0
+    } else {
+        u32::MAX << (32 - prefix)
+    };
+    (mask == expected).then_some(prefix)
+}
+
+fn extract_class_id(block: &str) -> Option<u16> {
+    for line in block.lines() {
+        let tokens: Vec<&str> = line.split_whitespace().collect();
+        for key in ["classid", "flowid"] {
+            let Some(pos) = tokens.iter().position(|token| *token == key) else {
+                continue;
+            };
+            let class = tokens.get(pos + 1)?;
+            return class.split(':').nth(1)?.parse::<u16>().ok();
+        }
+    }
+    None
+}
+
+fn extract_token_u16(block: &str, key: &str) -> Option<u16> {
+    for line in block.lines() {
+        let tokens: Vec<&str> = line.split_whitespace().collect();
+        if let Some(pos) = tokens.iter().position(|token| *token == key) {
+            return tokens.get(pos + 1)?.parse::<u16>().ok();
+        }
+    }
+    None
+}
+
+fn extract_packet_count(block: &str) -> Option<u64> {
+    for line in block.lines() {
+        let tokens: Vec<&str> = line.split_whitespace().collect();
+        if tokens.first() == Some(&"Sent")
+            && tokens.get(2) == Some(&"bytes")
+            && tokens.get(4) == Some(&"pkt")
+        {
+            return tokens.get(3)?.parse::<u64>().ok();
+        }
+        if let Some(pos) = tokens.iter().position(|token| *token == "success") {
+            return tokens.get(pos + 1)?.parse::<u64>().ok();
+        }
+    }
+    None
+}
+
 pub(super) fn has_fq_codel_for_parent(raw_qdisc: &str, parent_class: &str) -> bool {
     raw_qdisc.lines().map(str::trim).any(|line| {
         line.contains("fq_codel") && line.contains("parent") && line.contains(parent_class)
@@ -210,27 +363,6 @@ fn split_filter_blocks(raw: &str) -> Vec<String> {
     blocks
 }
 
-fn extract_flowid_class_id(line: &str) -> Option<u16> {
-    let toks: Vec<&str> = line.split_whitespace().collect();
-    let flowid = toks
-        .iter()
-        .position(|v| *v == "flowid")
-        .and_then(|i| toks.get(i + 1))?;
-    let class_tail = flowid.split(':').nth(1)?;
-    class_tail.parse::<u16>().ok()
-}
-
-fn extract_u32_match_hex(line: &str) -> Option<String> {
-    let toks: Vec<&str> = line.split_whitespace().collect();
-    let pos = toks.iter().position(|v| *v == "match")?;
-    let match_val = *toks.get(pos + 1)?;
-    let (hex, mask) = match_val.split_once('/')?;
-    if mask.eq_ignore_ascii_case("ffffffff") && hex.len() == 8 {
-        return Some(hex.to_ascii_lowercase());
-    }
-    None
-}
-
 fn hex_to_ipv4_string(hex: &str) -> Option<String> {
     if hex.len() != 8 {
         return None;
@@ -299,6 +431,56 @@ mod tests {
         assert!(inspect.duplicate_classes.contains(&10));
     }
 
+    #[test]
+    fn parse_debian13_flower_filters_and_statistics() {
+        let raw = r#"
+filter protocol ip pref 10 flower chain 0
+filter protocol ip pref 10 flower chain 0 handle 0x1 classid 1:10
+  dst_ip 10.8.1.2/32
+  not_in_hw
+  Sent 1200 bytes 12 pkt (dropped 0, overlimits 0 requeues 0)
+filter protocol ip pref 10 flower chain 0 handle 0x2 classid 1:11
+  dst_ip 10.8.1.3
+  not_in_hw
+  Sent 900 bytes 9 pkt (dropped 0, overlimits 0 requeues 0)
+"#;
+        let inspect = extract_filter_ip_class_map(raw);
+        assert_eq!(inspect.unrecognized_rules, 0);
+        assert_eq!(inspect.map.get("10.8.1.2"), Some(&10));
+        assert_eq!(inspect.map.get("10.8.1.3"), Some(&11));
+        assert!(inspect.filters.iter().all(|filter| {
+            filter.classifier == FilterClassifier::Flower
+                && filter.direction == FilterDirection::Destination
+                && filter.pref == Some(10)
+                && filter.prefix == 32
+        }));
+        assert_eq!(parse_peer_filter_hits(raw).get("10.8.1.2"), Some(&12));
+    }
+
+    #[test]
+    fn parse_flower_source_prefix_and_classid_on_separate_lines() {
+        let raw = r#"
+filter protocol ip pref 10 flower chain 0 handle 0x7
+  src_ip 10.8.1.5/24
+  classid 2:42
+"#;
+        let inspect = extract_filter_ip_class_map(raw);
+        let filter = inspect.filters.first().expect("flower filter");
+        assert_eq!(filter.ip, "10.8.1.5");
+        assert_eq!(filter.prefix, 24);
+        assert_eq!(filter.direction, FilterDirection::Source);
+        assert_eq!(filter.class_id, 42);
+    }
+
+    #[test]
+    fn malformed_peer_rule_is_not_silently_ignored() {
+        let raw =
+            "filter protocol ip pref 10 flower chain 0 handle 0x1 classid 1:10\n  action drop";
+        let inspect = extract_filter_ip_class_map(raw);
+        assert_eq!(inspect.unrecognized_rules, 1);
+        assert!(inspect.map.is_empty());
+    }
+
     #[test]
     fn detect_fq_codel_parent() {
         let raw = "qdisc fq_codel 8001: dev wg0 parent 1:10 limit 10240p";