Recover from interface recreation

src/reflector/address_monitor.h

@@ -2,6 +2,8 @@
 
 #include "util/delegate.h"
 
+#include <span>
+
 namespace reflector {
 
 // Watches the kernel for interface address changes and reports the affected interface index, so a
@@ -11,17 +13,20 @@ namespace reflector {
 // watching begins there, so the owner can bind a callback into itself first.
 class AddressMonitor {
 public:
-    // Invoked with the index of an interface whose addresses changed, or 0 ("all interfaces")
-    // when notifications may have been dropped (kernel buffer overflow) and everything should be
-    // re-resolved. Kernel interface indices are >= 1, so 0 is an unambiguous sentinel.
-    using OnInterfaceChanged = Delegate<void(unsigned interface_index)>;
+    // Invoked once per drain of the kernel's notification socket, with the indexes whose addresses
+    // changed. `refresh_all` means the drain could not deliver a list — notifications were dropped
+    // (kernel buffer overflow), or more interfaces changed than one drain can carry — so every
+    // interface must be re-resolved and `indexes` is empty. One call per drain rather than one per
+    // index: a burst commonly names several, and the work a subscriber does per call is not
+    // proportional to how many it names.
+    using OnInterfacesChanged = Delegate<void(std::span<const unsigned> indexes, bool refresh_all)>;
 
     virtual ~AddressMonitor() noexcept = default;
 
-    // Begin watching for interface address changes, delivering each changed interface index to
+    // Begin watching for interface address changes, delivering each drain's changed indexes to
     // `on_change`. Call exactly once, after construction. Returns false (after logging the cause)
     // if the monitor could not start watching; whether to proceed without it is the caller's call.
-    [[nodiscard]] virtual bool Start(const OnInterfaceChanged& on_change) noexcept = 0;
+    [[nodiscard]] virtual bool Start(const OnInterfacesChanged& on_change) noexcept = 0;
 };
 
 } // namespace reflector

src/reflector/application.cpp

@@ -12,6 +12,7 @@
 #include "util/fd_util.h"
 #include "wol_reflector.h"
 
+#include <algorithm>
 #include <array>
 #include <cassert>
 #include <chrono>
@@ -31,6 +32,14 @@ Logger& GetLogger() noexcept {
 }
 // How often the memory diagnostic logs once debug_memory is enabled.
 constexpr std::chrono::seconds MEMORY_REPORT_INTERVAL{60};
+// The reconcile backstop. Every other trigger depends on the kernel telling us something, and a
+// notification can be dropped without the kernel saying so, so this is the one channel that always
+// arrives. It rides the wakeup the poll already makes, costing only one attachment probe per
+// capture when nothing is wrong.
+constexpr std::chrono::seconds RECONCILE_BACKSTOP_INTERVAL{30};
+// The retry cadence while a repair is outstanding. A rebind that failed on a transient error has
+// no announcement coming, so nothing but this would ever finish it.
+constexpr std::chrono::seconds REPAIR_RETRY_INTERVAL{1};
 } // namespace
 
 namespace reflector {
@@ -42,6 +51,7 @@ Application::Application()
           }}
         , dispatcher_{std::make_unique<EventLoopDispatcher>()}
         , address_monitor_{std::make_unique<DefaultAddressMonitor>(*dispatcher_)} {
+    packet_dispatcher_.OnCaptureFailure(CreateDelegate<&Application::OnCaptureFailure>(this));
     StartMonitor();
 }
 
@@ -51,6 +61,7 @@ Application::Application(std::unique_ptr<Dispatcher> dispatcher, std::unique_ptr
         , socket_factory_{std::move(socket_factory)}
         , dispatcher_{std::move(dispatcher)}
         , address_monitor_{std::move(monitor)} {
+    packet_dispatcher_.OnCaptureFailure(CreateDelegate<&Application::OnCaptureFailure>(this));
     StartMonitor();
 }
 
@@ -64,7 +75,7 @@ Application Application::ForTesting(std::unique_ptr<Dispatcher> dispatcher,
 void Application::StartMonitor() {
     // Address-change refresh is best-effort: if the monitor can't start (it logs the cause),
     // carry on without it rather than failing the daemon.
-    if (!address_monitor_->Start(CreateDelegate<&Application::OnInterfaceChanged>(this))) {
+    if (!address_monitor_->Start(CreateDelegate<&Application::OnInterfacesChanged>(this))) {
         GetLogger().Warning("Address monitor unavailable; source addresses will not refresh on interface changes");
     }
 }
@@ -125,6 +136,8 @@ bool Application::Configure(const Config& config) {
     if (ConfigureReflectors<WolReflector>(config.WolConfigs(), "wol")
         && ConfigureReflectors<MdnsReflector>(config.MdnsConfigs(), "mdns")
         && ConfigureReflectors<SsdpReflector>(config.SsdpConfigs(), "ssdp")) {
+        reconcile_timer_.Start(
+            RECONCILE_BACKSTOP_INTERVAL, CreateDelegate<&Application::OnReconcileTick>(this));
         if (config.DebugMemory()) {
             GetLogger().Info("Memory diagnostics enabled; reporting RSS/heap every {}s",
                 MEMORY_REPORT_INTERVAL.count());
@@ -140,19 +153,82 @@ bool Application::Configure(const Config& config) {
     return false;
 }
 
-void Application::OnInterfaceChanged(unsigned interface_index) noexcept {
-    // index 0 is the monitor's "refresh everything" signal (notification overflow); otherwise
-    // refresh only the changed interface.
-    for (const auto& entry : interfaces_) {
-        const auto& iface = entry.second;
-        if (iface && (interface_index == 0 || iface->Index() == interface_index)) {
-            iface->Refresh();
+void Application::OnInterfacesChanged(std::span<const unsigned> indexes, bool refresh_all) noexcept {
+    ArmRepairRetry(ReconcileInterfaces(indexes, refresh_all));
+    NotifyReflectors();
+}
+
+void Application::OnReconcileTick(std::chrono::steady_clock::time_point) noexcept {
+    ArmRepairRetry(ReconcileInterfaces({}, false));
+    NotifyReflectors();
+}
+
+bool Application::ReconcileInterfaces(std::span<const unsigned> indexes, bool refresh_all) noexcept {
+    bool outstanding = false;
+    for (const auto& [name, iface] : interfaces_) {
+        // Configure fails unless every configured interface resolved and opened a valid socket, so
+        // a daemon that reached the event loop holds one for each interface.
+        const auto entry = sockets_.find(name);
+        assert(entry != sockets_.end());
+        LinkSocket& socket = *entry->second;
+
+        // Both read before Reidentify, which is what moves the index out from under them.
+        const bool refresh_requested =
+            refresh_all || std::ranges::find(indexes, iface->Index()) != indexes.end();
+        // Probe before resolving: getsockname is one syscall where a name lookup is three (glibc
+        // opens a socket inside if_nametoindex), and a capture the kernel still has attached
+        // proves its interface has not gone anywhere. A rename is the exception — it keeps both
+        // the index and the capture, so only the name lookup sees it — but it also announces
+        // itself, which is why a requested interface resolves regardless.
+        const bool attached = socket.Attached();
+        if (attached && !refresh_requested) {
+            continue;
+        }
+
+        const auto change = iface->Reidentify();
+        if (change == Interface::IdentityChange::Unresolved) {
+            outstanding = true;  // says nothing about the interface, so act on nothing
+            continue;
+        }
+        if (change == Interface::IdentityChange::Unchanged && refresh_requested) {
+            iface->Refresh();  // Reidentify already refreshed the other cases
+        }
+
+        // A capture is bound to a kernel object, not to a name, so it does not follow the
+        // interface across a recreation. Two ways it shows: the index moved, or it did not and the
+        // capture is detached anyway, which is what happens where the kernel hands a recreated
+        // interface the number it had.
+        if (!iface->IsValid()) {
+            continue;  // parked, so there is nothing to bind to until it comes back
+        }
+        if ((!attached || change == Interface::IdentityChange::Repointed) && !socket.Rebind()) {
+            outstanding = true;  // Rebind logs its own failure
         }
     }
+    return outstanding;
+}
+
+void Application::OnCaptureFailure() noexcept {
+    // Repair on the retry cadence rather than inline: this runs inside a drain, and the reconcile
+    // rebinds sockets and broadcasts to reflectors, which is not work to do underneath one.
+    ArmRepairRetry(true);
+}
+
+void Application::ArmRepairRetry(bool outstanding) noexcept {
+    if (!outstanding) {
+        repair_timer_.Stop();
+        return;
+    }
+    // Re-registering re-anchors the deadline to now, so restarting an already-running retry on
+    // every failed pass would let a stream of address events push it back indefinitely.
+    if (!repair_timer_.IsRunning()) {
+        repair_timer_.Start(REPAIR_RETRY_INTERVAL, CreateDelegate<&Application::OnReconcileTick>(this));
+    }
+}
 
-    // The fresh addresses are now visible. Let every reflector react (re-gate families, join/leave
-    // groups, log transitions) — each reads live interface state and no-ops if nothing relevant to
-    // it changed, so a single broadcast after the refresh is enough.
+void Application::NotifyReflectors() noexcept {
+    // Each reads live interface state and no-ops if nothing relevant to it changed, so one
+    // broadcast after the reconcile is enough.
     for (const auto& reflector : reflectors_) {
         reflector->OnInterfaceChanged();
     }

src/reflector/application.h

@@ -17,6 +17,7 @@
 #include <functional>
 #include <memory>
 #include <optional>
+#include <span>
 #include <string>
 #include <string_view>
 #include <unordered_map>
@@ -74,7 +75,7 @@ class Application : NoMove {
     Application(std::unique_ptr<Dispatcher> dispatcher, std::unique_ptr<AddressMonitor> monitor,
         InterfaceFactory interface_factory, SocketFactory socket_factory);
 
-    // Starts the address monitor, routing changes to OnInterfaceChanged. Logs a warning and
+    // Starts the address monitor, routing changes to OnInterfacesChanged. Logs a warning and
     // continues if it can't start — address refresh is best-effort, not required to run.
     void StartMonitor();
 
@@ -99,9 +100,30 @@ class Application : NoMove {
     // the caller logs with the interface's role (source vs target).
     [[nodiscard]] LinkSocket* GetOrCreateSocket(const std::string& interface);
 
-    // Address-monitor callback: re-resolve the source addresses of the changed interface, or of
-    // every interface when index == 0 (the monitor's overflow signal).
-    void OnInterfaceChanged(unsigned interface_index) noexcept;
+    // Address-monitor callback: reconcile against the kernel, then let every reflector react.
+    void OnInterfacesChanged(std::span<const unsigned> indexes, bool refresh_all) noexcept;
+
+    // Both reconcile timers fire this: the backstop, and the retry while a repair is outstanding.
+    // The fire-time argument is unused.
+    void OnReconcileTick(std::chrono::steady_clock::time_point) noexcept;
+
+    // Brings every interface and its capture back in line with the kernel, refreshing the
+    // addresses of the interfaces in `indexes` (all of them when `refresh_all`, which is what the
+    // monitor reports when it lost notifications). Returns whether a repair did not complete, so
+    // the pass must run again.
+    [[nodiscard]] bool ReconcileInterfaces(
+        std::span<const unsigned> indexes, bool refresh_all) noexcept;
+
+    // Packet-dispatcher callback: a capture read failed, so the kernel has already reported what a
+    // probe would. Schedules the repair instead of running it inside the drain.
+    void OnCaptureFailure() noexcept;
+
+    // Runs the retry timer while `outstanding`, stops it otherwise.
+    void ArmRepairRetry(bool outstanding) noexcept;
+
+    // Lets every reflector re-gate families, join/leave groups and log transitions off the
+    // interface state the reconcile just refreshed.
+    void NotifyReflectors() noexcept;
 
     // Drains the signal-wakeup self-pipe set up by PrepareSignalWakeup. The byte exists only to wake the
     // poll; the loop's stop_requested check ends the run. Level-triggered, so drain fully.
@@ -134,6 +156,13 @@ class Application : NoMove {
     // tear down before dispatcher_ — it does, being declared after it.
     std::optional<Timer> memory_timer_;
 
+    // The reconcile backstop, armed for the daemon's life: the only channel that does not depend
+    // on the kernel telling us something, so it is what covers a notification we never got.
+    Timer reconcile_timer_{*dispatcher_};
+    // Armed only while a repair is outstanding. A rebind that failed on a transient error has no
+    // external announcement coming, so nothing else would ever finish it.
+    Timer repair_timer_{*dispatcher_};
+
     // Signal-wakeup self-pipe (best-effort; populated by PrepareSignalWakeup, otherwise empty). Declared
     // last so it tears down first: wakeup_reg_ unregisters from dispatcher_ while it is still alive, then
     // the pipe fds close. wakeup_reg_ holds a callback bound to `this`, so it must outlive nothing here.

src/reflector/default_address_monitor.cpp

@@ -13,7 +13,6 @@
 #include <cstddef>
 #include <cstring>
 #include <span>
-#include <vector>
 #include <net/if.h>
 #include <sys/socket.h>
 
@@ -45,15 +44,6 @@ constexpr size_t NOTIFICATION_BUFFER_SIZE = 8 * 1024;
 constexpr int ROUTE_RECEIVE_BUFFER_BYTES = 256 * 1024;
 #endif
 
-void AddUnique(std::vector<unsigned>& indices, unsigned index) {
-    if (index == 0) {
-        return;  // names no interface (kernel indices are >= 1) — and 0 is the refresh-all sentinel
-    }
-    if (std::ranges::find(indices, index) == indices.end()) {
-        indices.push_back(index);
-    }
-}
-
 } // namespace
 
 namespace reflector {
@@ -91,7 +81,7 @@ DefaultAddressMonitor DefaultAddressMonitor::ForTesting(Dispatcher& dispatcher,
     return DefaultAddressMonitor{dispatcher, fd, verify_sender};
 }
 
-bool DefaultAddressMonitor::Start(const OnInterfaceChanged& on_change) noexcept {
+bool DefaultAddressMonitor::Start(const OnInterfacesChanged& on_change) noexcept {
     if (!on_change.IsValid()) {
         GetLogger().Error("Cannot start address monitor: the change callback is not bound");
         Close();
@@ -181,9 +171,9 @@ void DefaultAddressMonitor::OnReadable(int /*fd*/) noexcept {
 #endif
     std::array<std::byte, NOTIFICATION_BUFFER_SIZE> buffer;
 
-    // Coalesce a whole drain into one callback per interface: a burst commonly repeats the same
-    // index, and on overflow we drop the partial list and emit a single refresh-all instead.
-    std::vector<unsigned> changed;
+    // Coalesce a whole drain into one callback: a burst commonly repeats the same index, and on
+    // overflow we drop the partial list and report refresh-all instead.
+    ChangedInterfaces changed;
     bool overflowed = false;
     while (true) {
         sockaddr_storage src{};
@@ -224,12 +214,30 @@ void DefaultAddressMonitor::OnReadable(int /*fd*/) noexcept {
 
     if (overflowed) {
         GetLogger().Warning("Address notifications overflowed; refreshing all interfaces");
-        on_change_(0u);
+    } else if (changed.overflowed) {
+        GetLogger().Debug("More than {} interfaces changed in one drain; refreshing all",
+            MAX_CHANGED_INTERFACES);
+    } else if (changed.count == 0) {
+        return;  // the drain carried nothing we track, so there is nothing to tell anyone
     } else {
-        for (const unsigned index : changed) {
-            on_change_(index);
-        }
+        on_change_(changed.Indexes(), false);
+        return;
+    }
+    on_change_(std::span<const unsigned>{}, true);
+}
+
+void DefaultAddressMonitor::ChangedInterfaces::Add(unsigned index) noexcept {
+    if (index == 0) {
+        return;  // names no interface; kernel indices are >= 1
+    }
+    if (std::ranges::find(Indexes(), index) != Indexes().end()) {
+        return;
+    }
+    if (count == MAX_CHANGED_INTERFACES) {
+        overflowed = true;
+        return;
     }
+    indexes[count++] = index;
 }
 
 #if defined(__linux__)
@@ -247,16 +255,16 @@ void DefaultAddressMonitor::OnReadable(int /*fd*/) noexcept {
 #pragma GCC diagnostic ignored "-Wconversion"
 
 void DefaultAddressMonitor::CollectChangedInterfaces(std::span<std::byte> messages,
-        std::vector<unsigned>& changed) const noexcept {
+        ChangedInterfaces& changed) const noexcept {
     auto* header = start_lifetime_as<nlmsghdr>(messages.data());
     for (int length = static_cast<int>(messages.size()); NLMSG_OK(header, length);
             header = start_lifetime_as<nlmsghdr>(NLMSG_NEXT(header, length))) {
         if (header->nlmsg_type == RTM_NEWADDR || header->nlmsg_type == RTM_DELADDR) {
             const auto* address = start_lifetime_as<ifaddrmsg>(NLMSG_DATA(header));
-            AddUnique(changed, address->ifa_index);
+            changed.Add(address->ifa_index);
         } else if (header->nlmsg_type == RTM_NEWLINK || header->nlmsg_type == RTM_DELLINK) {
             const auto* link = start_lifetime_as<ifinfomsg>(NLMSG_DATA(header));
-            AddUnique(changed, static_cast<unsigned>(link->ifi_index));
+            changed.Add(static_cast<unsigned>(link->ifi_index));
         }
     }
 }
@@ -266,7 +274,7 @@ void DefaultAddressMonitor::CollectChangedInterfaces(std::span<std::byte> messag
 #else
 
 void DefaultAddressMonitor::CollectChangedInterfaces(std::span<std::byte> messages,
-        std::vector<unsigned>& changed) const noexcept {
+        ChangedInterfaces& changed) const noexcept {
     // PF_ROUTE messages pack back-to-back; each begins with rt_msghdr's prefix (u_short msglen;
     // u_char version; u_char type). Read the fields with memcpy — the buffer carries no
     // alignment guarantee and the messages aren't a single struct type.
@@ -288,7 +296,7 @@ void DefaultAddressMonitor::CollectChangedInterfaces(std::span<std::byte> messag
                 && message_length >= index_end) {
             u_short index = 0;
             std::memcpy(&index, messages.data() + offset + offsetof(ifa_msghdr, ifam_index), sizeof(index));
-            AddUnique(changed, index);
+            changed.Add(index);
         }
         offset += message_length;
     }