From 8d6212abf047c60d7e78950af7bf490e9dc56ff6 Mon Sep 17 00:00:00 2001 From: Sergii Bogomolov Date: Sun, 9 Aug 2026 13:23:21 +0200 Subject: [PATCH 1/5] feat: let an interface re-resolve its kernel index The name is an interface's stable identity; the index is not, since a recreated interface keeps the name and gets a new number. Reidentify re-resolves it and reports whether the interface was repointed, parked, unchanged, or unresolvable, so a caller can tell recreation from absence. if_nametoindex is not a pure lookup -- glibc opens a socket inside it -- so under fd pressure it returns 0 for a live interface. Reading that as absence would clear the addresses and shut the egress gate on something healthy, so the resource errnos report Unresolved and leave the identity untouched. An interface that is absent and stays absent reports Unchanged, which keeps a retry loop from logging or working on every pass. Joining a group now refuses index 0 rather than letting the kernel read it as "any interface". --- src/reflector/interface.cpp | 59 ++++++++++++++++++++++++++++++++---- src/reflector/interface.h | 27 ++++++++++++++++- src/reflector/raw_socket.cpp | 7 +++++ tests/interface_test.cpp | 37 ++++++++++++++++++++++ tests/mocks/fake_interface.h | 24 +++++++++++++++ 5 files changed, 147 insertions(+), 7 deletions(-) diff --git a/src/reflector/interface.cpp b/src/reflector/interface.cpp index 5187b90..11804e1 100644 --- a/src/reflector/interface.cpp +++ b/src/reflector/interface.cpp @@ -3,22 +3,30 @@ #include "error.h" #include "platform.h" +#include #include +#include +#include #include +namespace { +// The errnos that mean the lookup could not run, as opposed to naming no interface. Anything else +// still reads as absent, so an errno missing from this list cannot mask a genuinely removed one. +bool LookupCouldNotRun(int error) noexcept { + return error == EMFILE || error == ENFILE || error == ENOMEM || error == ENOBUFS; +} +} // namespace + namespace reflector { Interface::Interface(std::string_view name) : logger_{std::format("Interface:{}", name)} , name_{name} { - // Guard before any name lookup: if_nametoindex (and BPF's BIOCSETIF) copy into a fixed - // IFNAMSIZ buffer, so an over-long name would be silently truncated and could match the - // wrong interface. if (name_.size() >= IFNAMSIZ) { logger_.Error("Interface name \"{}\" is too long (max {} characters)", name_, IFNAMSIZ - 1); return; } - index_ = if_nametoindex(name_.c_str()); + index_ = ResolveIndex().value_or(0); if (index_ == 0) { logger_.Error("Cannot resolve interface index: {}", Error::FromErrno()); return; @@ -28,11 +36,50 @@ Interface::Interface(std::string_view name) Refresh(); } +std::optional Interface::ResolveIndex() const noexcept { + // Guard before any name lookup: if_nametoindex (and BPF's BIOCSETIF) copy into a fixed + // IFNAMSIZ buffer, so an over-long name would be silently truncated and could match the + // wrong interface. Silent here — the constructor reports it once, and this runs on a retry + // loop where the name cannot have changed. + if (name_.size() >= IFNAMSIZ) { + return 0U; + } + const unsigned index = if_nametoindex(name_.c_str()); + if (index != 0) { + return index; + } + // if_nametoindex reaches a 0 return only through a failed socket() or ioctl(), so errno is + // this call's own rather than something left over. + return LookupCouldNotRun(errno) ? std::nullopt : std::optional{0U}; +} + +Interface::IdentityChange Interface::Reidentify() noexcept { + const auto resolved = ResolveIndex(); + if (!resolved) { + logger_.Error("Cannot resolve interface index: {}; keeping index {}", Error::FromErrno(), index_); + return IdentityChange::Unresolved; + } + if (*resolved == index_) { + return IdentityChange::Unchanged; // covers still-absent, so a retry loop stays quiet + } + + const unsigned previous = std::exchange(index_, *resolved); + if (index_ == 0) { + addresses_ = {}; // nothing may send or join against an identity that is gone + logger_.Info("Interface is gone (was index {}); parked until it returns", previous); + return IdentityChange::Parked; + } + + Refresh(); + logger_.Info("Interface reappeared as index {} (was {})", index_, previous); + return IdentityChange::Repointed; +} + Interface::Interface(std::string_view name, unsigned index, const InterfaceAddresses& addresses) noexcept : addresses_{addresses} + , index_{index} , logger_{std::format("Interface:{}", name)} - , name_{name} - , index_{index} {} + , name_{name} {} std::optional Interface::SourceAddress(IpAddress::Family family) const noexcept { return family == IpAddress::Family::V4 ? addresses_.v4 : addresses_.v6; diff --git a/src/reflector/interface.h b/src/reflector/interface.h index 52c06f9..ed49c06 100644 --- a/src/reflector/interface.h +++ b/src/reflector/interface.h @@ -48,16 +48,41 @@ class Interface : NoMove { // a change on this interface. Virtual so tests substitute a no-syscall fake. virtual void Refresh() noexcept; + // What re-resolving the identity changed. Anything but Unchanged is a transition the caller + // acts on; an interface that was already parked and still is reports Unchanged, so a retry + // loop neither logs nor works on every pass. + enum class IdentityChange : uint8_t { + Unchanged, // same index as before + Repointed, // a different index: the interface was recreated, so captures bound to the old + // one are attached to a dead kernel object and must re-attach + Parked, // no longer resolvable; index and addresses cleared so nothing sends or joins + // against a dead identity until it comes back + Unresolved, // the lookup could not run, which says nothing about the interface: identity + // and addresses are kept as they were, and the caller retries + }; + + // Re-resolves the kernel index from the name, and the addresses when it resolves. The name is + // the stable identity — the index is not, since a recreated interface keeps the name and gets + // a new number. Virtual so tests substitute a no-syscall fake. + virtual IdentityChange Reidentify() noexcept; + protected: // Test seam: fixed identity, no kernel lookups (see FakeInterface). Interface(std::string_view name, unsigned index, const InterfaceAddresses& addresses) noexcept; InterfaceAddresses addresses_; + // Protected so a fake can stage the identity a real Reidentify reads from the kernel. + unsigned index_ = 0; private: + // name_ -> kernel index: 0 when the name is over-long or the kernel does not know it, nullopt + // when the lookup could not run. if_nametoindex is not a pure lookup — glibc opens a socket + // inside it — so under fd or memory pressure it reports a live interface as absent, and acting + // on that would park a healthy interface. + [[nodiscard]] std::optional ResolveIndex() const noexcept; + Logger logger_; std::string name_; - unsigned index_ = 0; }; } // namespace reflector diff --git a/src/reflector/raw_socket.cpp b/src/reflector/raw_socket.cpp index 780b7ec..e1a76c1 100644 --- a/src/reflector/raw_socket.cpp +++ b/src/reflector/raw_socket.cpp @@ -391,6 +391,13 @@ bool RawSocket::SendFrame(MacAddress dst_mac, const IpEndpoint& dst, uint16_t sr } LinkSocket::MulticastMembership RawSocket::JoinMulticastGroup(const IpAddress& group) noexcept { + // Index 0 is the kernel's "any interface" wildcard, so a parked interface would silently join + // on whichever one the routing table picks. + if (interface_->Index() == 0) { + logger_.Error("Cannot join multicast group {}: the interface is not resolvable", group); + return {}; + } + const auto family = group.AddressFamily(); auto& memberships = group_memberships_.Get(family); diff --git a/tests/interface_test.cpp b/tests/interface_test.cpp index a3e60b1..dd8a390 100644 --- a/tests/interface_test.cpp +++ b/tests/interface_test.cpp @@ -44,6 +44,43 @@ TEST(InterfaceTest, RefreshKeepsLoopbackAddresses) { EXPECT_TRUE(iface.SourceAddress(IpAddress::Family::V4).has_value()); } +TEST(InterfaceTest, ReidentifyReportsNoChangeForALiveInterface) { + Interface iface{LoopbackInterface()}; + ASSERT_TRUE(iface.IsValid()); + const unsigned index = iface.Index(); + + EXPECT_EQ(iface.Reidentify(), Interface::IdentityChange::Unchanged); + EXPECT_EQ(iface.Index(), index); + EXPECT_TRUE(iface.SourceAddress(IpAddress::Family::V4).has_value()); +} + +// A name the kernel does not know reports Unchanged, not Parked: the reconcile pass runs on a +// timer, so an absent interface that stays absent must not log or do work on every pass. +TEST(InterfaceTest, ReidentifyIsQuietWhileAnInterfaceStaysAbsent) { + Interface iface{"nonex0"}; + ASSERT_FALSE(iface.IsValid()); + + const std::string output = CaptureStdout([&] { + EXPECT_EQ(iface.Reidentify(), Interface::IdentityChange::Unchanged); + EXPECT_EQ(iface.Reidentify(), Interface::IdentityChange::Unchanged); + }); + + EXPECT_TRUE(output.empty()) << output; + EXPECT_FALSE(iface.IsValid()); +} + +// The name is checked once at construction; a retry cannot change it, so it stays silent too. +TEST(InterfaceTest, ReidentifyIsQuietForAnOverlongName) { + Interface iface{std::string(IFNAMSIZ, 'x')}; + ASSERT_FALSE(iface.IsValid()); + + const std::string output = CaptureStdout([&] { + EXPECT_EQ(iface.Reidentify(), Interface::IdentityChange::Unchanged); + }); + + EXPECT_TRUE(output.empty()) << output; +} + TEST(InterfaceTest, SourceAddressForMatchesTheDestinationScope) { const auto link_local = *IpAddress::FromString("fe80::1"); const auto unique_local = *IpAddress::FromString("fd00::1"); diff --git a/tests/mocks/fake_interface.h b/tests/mocks/fake_interface.h index b9f49a2..a33f287 100644 --- a/tests/mocks/fake_interface.h +++ b/tests/mocks/fake_interface.h @@ -34,7 +34,31 @@ class FakeInterface : public Interface { void Refresh() noexcept override { ++refresh_count; } + // Stages what the next lookup would report: a different index models a recreation, 0 the + // interface going away, nullopt a lookup that could not run. Unset, the fake keeps the + // identity it was built with. + void StageIndex(std::optional index) noexcept { staged_index_ = index; } + + IdentityChange Reidentify() noexcept override { + if (!staged_index_) { + return IdentityChange::Unresolved; + } + if (*staged_index_ == index_) { + return IdentityChange::Unchanged; + } + index_ = *staged_index_; + if (index_ == 0) { + addresses_ = {}; + return IdentityChange::Parked; + } + Refresh(); + return IdentityChange::Repointed; + } + unsigned refresh_count = 0; + +private: + std::optional staged_index_ = index_; }; } // namespace reflector From 254967e428469d2e0430adfabb78f8c11ea4eb69 Mon Sep 17 00:00:00 2001 From: Sergii Bogomolov Date: Sun, 9 Aug 2026 14:13:43 +0200 Subject: [PATCH 2/5] feat: re-attach a capture to a recreated interface A capture holds an fd, so nothing in userland notices its interface leaving; only the kernel knows. Attached asks it -- the bound ifindex via getsockname on Linux, BIOCGDLT on the BSDs, which fails outright once a descriptor detaches. The probe is not a corner case. Linux gives a recreated interface a fresh index, but macOS hands back the one it had, so there the index is unchanged and the probe is the only thing that reports the capture is detached. Rebind re-runs the attach path on the same fd, so registrations keyed by it stay valid. Linux consumes the ENETDOWN the kernel parked when the old interface died, and the BSDs reset the batch cursor the re-attach invalidated, re-reading the framing in case the interface came back as a different link type. --- src/reflector/link_socket.h | 9 +++ src/reflector/raw_socket.cpp | 129 ++++++++++++++++++++++++--------- src/reflector/raw_socket.h | 8 ++ tests/mocks/fake_link_socket.h | 21 +++++- tests/raw_socket_test.cpp | 95 ++++++++++++++++++++++++ 5 files changed, 224 insertions(+), 38 deletions(-) diff --git a/src/reflector/link_socket.h b/src/reflector/link_socket.h index 70241b0..f7b6a61 100644 --- a/src/reflector/link_socket.h +++ b/src/reflector/link_socket.h @@ -81,6 +81,15 @@ class LinkSocket { // where a frame-source MAC filter can never match. [[nodiscard]] virtual bool LinkCarriesMacs() const noexcept = 0; + // Whether the capture is still attached to the live interface. A recreated interface can be + // handed back the index it had, so comparing indexes alone can miss the swap; this asks the + // kernel about the fd itself. + [[nodiscard]] virtual bool Attached() const noexcept = 0; + + // Re-attaches the capture to its interface's current kernel object, keeping the same fd so + // registrations keyed by it stay valid. False (having logged) if the kernel refuses. + [[nodiscard]] virtual bool Rebind() noexcept = 0; + protected: // Mints a membership for `group` owned by this socket; the join must already have happened. [[nodiscard]] MulticastMembership MakeMembership(const IpAddress& group) noexcept { diff --git a/src/reflector/raw_socket.cpp b/src/reflector/raw_socket.cpp index e1a76c1..24e52f1 100644 --- a/src/reflector/raw_socket.cpp +++ b/src/reflector/raw_socket.cpp @@ -197,12 +197,7 @@ RawSocket::RawSocket(const Interface& interface) // Start capturing: bind to the interface with protocol ETH_P_ALL. The filter is already in place, // so every delivered frame is filtered — there is no unfiltered-capture window. - sockaddr_ll addr{}; - addr.sll_family = AF_PACKET; - addr.sll_protocol = htons(ETH_P_ALL); - addr.sll_ifindex = static_cast(interface_->Index()); - if (bind(fd_.Get(), reinterpret_cast(&addr), sizeof(addr)) != 0) { - logger_.Error("Cannot bind AF_PACKET socket to interface: {}", Error::FromErrno()); + if (!AttachToInterface()) { Close(); return; } @@ -225,20 +220,57 @@ RawSocket::RawSocket(const Interface& interface) return; } + if (!AttachToInterface()) { + Close(); + return; + } + + u_int blen = 0; + if (ioctl(fd_.Get(), BIOCGBLEN, &blen) != 0) { + logger_.Error("Cannot query BPF buffer length: {}", Error::FromErrno()); + Close(); + return; + } + receive_buffer_.resize(blen); + + if (!SetNonBlocking(fd_.Get())) { + logger_.Error("Cannot set BPF socket non-blocking: {}", Error::FromErrno()); + Close(); + return; + } + + logger_.Debug("Opened BPF fd {} on interface (buffer {} bytes)", fd_.Get(), blen); +#endif +} + +bool RawSocket::AttachToInterface() noexcept { +#if defined(__linux__) + // Bind with protocol ETH_P_ALL. The filter and PACKET_IGNORE_OUTGOING are socket options set + // before the first bind and survive a re-bind, so there is no unfiltered-capture window here. + sockaddr_ll addr{}; + addr.sll_family = AF_PACKET; + addr.sll_protocol = htons(ETH_P_ALL); + addr.sll_ifindex = static_cast(interface_->Index()); + if (bind(fd_.Get(), reinterpret_cast(&addr), sizeof(addr)) != 0) { + logger_.Error("Cannot bind AF_PACKET socket to interface: {}", Error::FromErrno()); + return false; + } + return true; +#else ifreq ifr{}; // ifr is zero-initialized and Interface guarantees Name().size() < IFNAMSIZ. std::memcpy(ifr.ifr_name, interface_->Name().data(), interface_->Name().size()); if (ioctl(fd_.Get(), BIOCSETIF, &ifr) != 0) { logger_.Error("Cannot bind BPF to interface: {}", Error::FromErrno()); - Close(); - return; + return false; } + // Re-read the framing every time: a recreated interface can come back as a different link + // type, and the see-sent mode and filter below are both chosen from it. u_int dlt = 0; if (ioctl(fd_.Get(), BIOCGDLT, &dlt) != 0) { logger_.Error("Cannot query BPF link type: {}", Error::FromErrno()); - Close(); - return; + return false; } if (dlt == DLT_EN10MB) { link_type_ = LinkType::Ethernet; @@ -246,15 +278,13 @@ RawSocket::RawSocket(const Interface& interface) link_type_ = LinkType::Loopback; } else { logger_.Error("BPF link type {} is not supported (need DLT_EN10MB or DLT_NULL)", dlt); - Close(); - return; + return false; } u_int immediate = 1; if (ioctl(fd_.Get(), BIOCIMMEDIATE, &immediate) != 0) { logger_.Error("Cannot set BIOCIMMEDIATE: {}", Error::FromErrno()); - Close(); - return; + return false; } // Suppress locally-generated frames on Ethernet links: stops two mirrored reflector @@ -269,14 +299,11 @@ RawSocket::RawSocket(const Interface& interface) // isn't an escape hatch — same ioctl, same kernel handler, just a wider value // set. Linux doesn't need this gate: PACKET_IGNORE_OUTGOING on its AF_PACKET socket // drops the egress copy, collapsing lo's egress+ingress duplication to the ingress - // copy. - if (link_type_ == LinkType::Ethernet) { - u_int see_sent = 0; - if (ioctl(fd_.Get(), BIOCSSEESENT, &see_sent) != 0) { - logger_.Error("Cannot clear BIOCSSEESENT: {}", Error::FromErrno()); - Close(); - return; - } + // copy. Set both ways, so a re-attach onto different framing restores the right mode. + u_int see_sent = link_type_ == LinkType::Ethernet ? 0 : 1; + if (ioctl(fd_.Get(), BIOCSSEESENT, &see_sent) != 0) { + logger_.Error("Cannot set BIOCSSEESENT: {}", Error::FromErrno()); + return false; } // Different link types need different filter programs because the byte offsets to the @@ -289,26 +316,56 @@ RawSocket::RawSocket(const Interface& interface) }; if (ioctl(fd_.Get(), BIOCSETF, &program) != 0) { logger_.Error("Cannot attach BPF UDP filter: {}", Error::FromErrno()); - Close(); - return; + return false; } + return true; +#endif +} - u_int blen = 0; - if (ioctl(fd_.Get(), BIOCGBLEN, &blen) != 0) { - logger_.Error("Cannot query BPF buffer length: {}", Error::FromErrno()); - Close(); - return; +bool RawSocket::Attached() const noexcept { + if (!fd_) { + return false; } - receive_buffer_.resize(blen); - - if (!SetNonBlocking(fd_.Get())) { - logger_.Error("Cannot set BPF socket non-blocking: {}", Error::FromErrno()); - Close(); - return; +#if defined(__linux__) + // The kernel clears the binding when the interface unregisters, so the bound index no longer + // matches — including the case where a recreated interface was handed back its old number. + sockaddr_ll addr{}; + socklen_t length = sizeof(addr); + if (getsockname(fd_.Get(), reinterpret_cast(&addr), &length) != 0) { + return false; } + return addr.sll_ifindex > 0 + && static_cast(addr.sll_ifindex) == interface_->Index(); +#else + // BPF detaches the descriptor with its interface, after which even this query fails. + u_int dlt = 0; + return ioctl(fd_.Get(), BIOCGDLT, &dlt) == 0; +#endif +} - logger_.Debug("Opened BPF fd {} on interface (buffer {} bytes)", fd_.Get(), blen); +bool RawSocket::Rebind() noexcept { + if (!fd_ || interface_->Index() == 0) { + return false; + } + if (!AttachToInterface()) { + return false; + } +#if defined(__linux__) + // The kernel parked ENETDOWN on the socket when the old interface died; consume it so the + // first recv after this surfaces frames rather than the stale failure. + int pending = 0; + socklen_t length = sizeof(pending); + if (getsockopt(fd_.Get(), SOL_SOCKET, SO_ERROR, &pending, &length) == 0 && pending != 0) { + logger_.Debug("Cleared a pending error on the re-bound capture: {}", Error::FromErrno(pending)); + } +#else + // BPF reset its buffer at the re-attach, so drop the drained-batch cursor to match rather + // than walking frames the old interface left behind. + receive_buffer_filled_ = 0; + receive_buffer_offset_ = 0; #endif + logger_.Debug("Re-bound capture to interface index {}", interface_->Index()); + return true; } RawSocket::RawSocket(TestingTag, const Interface& interface, int owned_fd, diff --git a/src/reflector/raw_socket.h b/src/reflector/raw_socket.h index 3a18607..3a5333d 100644 --- a/src/reflector/raw_socket.h +++ b/src/reflector/raw_socket.h @@ -62,6 +62,9 @@ class RawSocket : public LinkSocket, NoMove { [[nodiscard]] const Interface& GetInterface() const noexcept override { return *interface_; } + [[nodiscard]] bool Attached() const noexcept override; + [[nodiscard]] bool Rebind() noexcept override; + [[nodiscard]] bool LinkCarriesMacs() const noexcept override { #if defined(__linux__) return true; // AF_PACKET delivers Ethernet framing everywhere, even on lo @@ -114,6 +117,11 @@ class RawSocket : public LinkSocket, NoMove { enum class TestingTag {}; RawSocket(TestingTag, const Interface& interface, int owned_fd, size_t receive_buffer_size) noexcept; + // Points the capture fd at interface_'s current kernel object: bind on Linux, the + // BIOCSETIF/BIOCGDLT/see-sent/filter sequence on the BSDs. Used at construction and by Rebind, + // so a recreated interface re-attaches through exactly the path that first attached it. + [[nodiscard]] bool AttachToInterface() noexcept; + void Close() noexcept; // Drops one membership of `group`: leaves the group in the kernel when its last membership diff --git a/tests/mocks/fake_link_socket.h b/tests/mocks/fake_link_socket.h index 985a175..8325c45 100644 --- a/tests/mocks/fake_link_socket.h +++ b/tests/mocks/fake_link_socket.h @@ -74,12 +74,29 @@ struct FakeLinkSocket : LinkSocket { return borrowed != nullptr ? *borrowed : iface; } - FakeInterface iface; // the owned identity; ignored when `borrowed` is set - const Interface* borrowed = nullptr; [[nodiscard]] bool LinkCarriesMacs() const noexcept override { return carries_macs; } + [[nodiscard]] bool Attached() const noexcept override { return attached; } + + [[nodiscard]] bool Rebind() noexcept override { + ++rebinds; + if (fail_rebind) { + return false; + } + attached = true; + return true; + } + + FakeInterface iface; // the owned identity; ignored when `borrowed` is set + const Interface* borrowed = nullptr; bool valid = true; bool carries_macs = true; + // Capture attachment, as the reconcile pass sees it: clear `attached` to model an interface + // that went away under the socket; `rebinds` counts recoveries so a test can assert the + // capture was actually re-attached rather than merely re-resolved. + bool attached = true; + bool fail_rebind = false; + size_t rebinds = 0; int fd = -1; bool fail_send = false; bool fail_join = false; diff --git a/tests/raw_socket_test.cpp b/tests/raw_socket_test.cpp index 4e74318..7a1c374 100644 --- a/tests/raw_socket_test.cpp +++ b/tests/raw_socket_test.cpp @@ -124,6 +124,56 @@ class InterfacePair { [[nodiscard]] const std::string& InjectInterface() const noexcept { return inject_; } [[nodiscard]] const std::string& ReceiveInterface() const noexcept { return receive_; } + // Tears the pair down early, so a test can watch what a capture does once its interface is + // gone. The destructor then finds nothing left to remove. + void DestroyNow() { Destroy(); } + + // Deletes the pair and creates it again under the same names, which the kernel gives fresh + // indexes — the recreation a reflector has to survive. Naming a unit explicitly re-creates + // that unit on every platform (`ifconfig` picks an arbitrary one only when the unit is + // omitted), so the names carry over. + [[nodiscard]] bool Recreate() { + Destroy(); +#if defined(__linux__) + if (!Run("ip link add " + inject_ + " type veth peer name " + receive_)) { + return false; + } + created_ = true; + valid_ = Run("ip addr add 10.99.0.1/24 dev " + inject_) + && Run("ip -6 addr add fe80::1/64 dev " + inject_ + " nodad") + && Run("ip link set " + inject_ + " up") + && Run("ip link set " + receive_ + " up"); +#elif defined(__APPLE__) + if (!Run("ifconfig " + inject_ + " create")) { + return false; + } + created_inject_ = true; + if (!Run("ifconfig " + receive_ + " create")) { + return false; + } + created_receive_ = true; + valid_ = Run("ifconfig " + inject_ + " peer " + receive_) + && Run("ifconfig " + inject_ + " inet 10.99.0.1/24 up") + && Run("ifconfig " + inject_ + " inet6 fe80::1 prefixlen 64") + && Run("ifconfig " + receive_ + " up"); +#elif defined(__FreeBSD__) + // The cloner names the pair, not each end: creating unit "epair0" yields epair0a/epair0b. + std::string unit = inject_; + unit.pop_back(); + if (!Run("ifconfig " + unit + " create")) { + return false; + } + created_ = true; + valid_ = Run("ifconfig " + inject_ + " inet 10.99.0.1/24 up") + && Run("ifconfig " + inject_ + " inet6 fe80::1 prefixlen 64") + && Run("ifconfig " + receive_ + " up"); +#endif + if (valid_) { + WaitUntilRunning(); + } + return valid_; + } + private: static bool Run(const std::string& command) { // POSIX std::system returns the wait status; a command that exits 0 yields 0. @@ -1014,6 +1064,51 @@ class RawSocketInterfacePairRequiresRootTest : public ::testing::Test { static void CloseSocket(RawSocket& socket) { socket.Close(); } }; +// The capture holds an fd, not a name, so nothing in userland notices the interface leaving. Only +// the kernel knows, which is what Attached() asks. +TEST_F(RawSocketInterfacePairRequiresRootTest, AttachedTurnsFalseWhenTheInterfaceGoesAway) { + const Interface iface{pair.InjectInterface()}; + ASSERT_TRUE(iface.IsValid()); + const RawSocket socket{iface}; + ASSERT_TRUE(socket.IsValid()); + ASSERT_TRUE(socket.Attached()); + + pair.DestroyNow(); + + EXPECT_FALSE(socket.Attached()); +} + +// The recovery this milestone exists for: an interface destroyed and recreated under the same name +// comes back with a new kernel index, and the capture -- still holding the same fd, so every +// registration keyed by it stays valid -- re-attaches to the new one. Worth running on the BSDs +// especially, where re-attaching redoes the whole BIOCSETIF/DLT/see-sent/filter sequence. +TEST_F(RawSocketInterfacePairRequiresRootTest, RebindRestoresTheCaptureAfterRecreation) { + Interface iface{pair.InjectInterface()}; + ASSERT_TRUE(iface.IsValid()); + RawSocket socket{iface}; + ASSERT_TRUE(socket.IsValid()); + const int fd = socket.Fd(); + const unsigned original_index = iface.Index(); + + ASSERT_TRUE(pair.Recreate()); + + // Detached whatever the index did — this is the check that has to carry the decision, because + // the index alone cannot: Linux allocates a fresh one, while macOS hands the recreated unit + // the number it had, so there comparing indexes sees nothing at all. + EXPECT_FALSE(socket.Attached()); + const auto change = iface.Reidentify(); + EXPECT_NE(change, Interface::IdentityChange::Parked) << "the interface is back, so not parked"; + if (iface.Index() == original_index) { + EXPECT_EQ(change, Interface::IdentityChange::Unchanged); // index reused + } else { + EXPECT_EQ(change, Interface::IdentityChange::Repointed); + } + + ASSERT_TRUE(socket.Rebind()); + EXPECT_TRUE(socket.Attached()); + EXPECT_EQ(socket.Fd(), fd); // same fd throughout, so dispatcher registrations survive +} + TEST_F(RawSocketInterfacePairRequiresRootTest, InjectsIpv4BroadcastCapturedOnPeer) { Interface inject_iface{pair.InjectInterface()}; RawSocket injector{inject_iface}; From 8d62b3b131f0bd2de95b755240e37b75888a7f82 Mon Sep 17 00:00:00 2001 From: Sergii Bogomolov Date: Mon, 10 Aug 2026 17:25:24 +0200 Subject: [PATCH 3/5] refactor: let a capture read report why it yielded no packet Receive returned nullopt for three outcomes: an empty queue, a frame we do not forward, and a read that failed. The drain has to tell the third from the others, since a failed read is often the first sign of an interface destroyed under the capture, so Receive now returns the reason. The reason is an enum rather than the codebase's Error: the drain switches on it instead of printing it, and building an Error for the would-block that ends every drain would allocate a string per readable event. The socket still logs the errno where it knows it. --- src/reflector/default_packet_dispatcher.cpp | 18 +++++++--- src/reflector/default_packet_dispatcher.h | 15 +++++++- src/reflector/link_socket.h | 14 ++++++-- src/reflector/raw_socket.cpp | 40 +++++++++++++-------- src/reflector/raw_socket.h | 3 +- tests/mocks/fake_link_socket.h | 7 +++- tests/raw_socket_test.cpp | 27 +++++++++----- 7 files changed, 91 insertions(+), 33 deletions(-) diff --git a/src/reflector/default_packet_dispatcher.cpp b/src/reflector/default_packet_dispatcher.cpp index 924349b..3f69fb4 100644 --- a/src/reflector/default_packet_dispatcher.cpp +++ b/src/reflector/default_packet_dispatcher.cpp @@ -86,15 +86,20 @@ void DefaultPacketDispatcher::OnReadable(int fd) noexcept { GetLogger().Warning("Readable callback for unknown capture fd {}", fd); return; } - DrainReadableFd(*it->second.socket); + // Reported after the drain, not from inside it: the sweep may have dropped this capture + // source, and the owner's repair is not work to do underneath a drain. + if (!DrainReadableFd(*it->second.socket) && on_capture_failure_.IsValid()) { + on_capture_failure_(); + } } -void DefaultPacketDispatcher::DrainReadableFd(LinkSocket& socket) noexcept { +bool DefaultPacketDispatcher::DrainReadableFd(LinkSocket& socket) noexcept { // Bracket the whole drain: a callback's Unregister only marks its entry disabled (DispatchPacket // skips it from here on), and the single sweep below erases the marked entries plus any now-orphaned // capture source. A socket whose last registration is dropped mid-drain just dispatches to nothing // for the rest of the drain, then loses its capture source in the sweep. dispatching_ = true; + bool failed = false; #if defined(__linux__) for (size_t packet_count = 0; packet_count < MAX_PACKETS_PER_READ_EVENT; ++packet_count) { @@ -103,11 +108,15 @@ void DefaultPacketDispatcher::DrainReadableFd(LinkSocket& socket) noexcept { #endif const auto packet = socket.Receive(); if (!packet) { + if (packet.error() == LinkSocket::ReceiveError::Failed) { + failed = true; // the kernel refusing the capture, not just this frame + } #if !defined(__linux__) if (socket.HasBufferedData()) { // Drain all userland-buffered frames. kqueue/epoll only fire on kernel-side - // activity, so if we leave frames buffered they'll stall. Only macOS BPF - // buffers in userland; HasBufferedData is not defined on Linux. + // activity, so if we leave frames buffered they'll stall. Only macOS BPF buffers + // in userland; HasBufferedData is not defined on Linux. Deliberately not gated on + // `failed`: frames captured before a read failed are still worth dispatching. continue; } #endif @@ -119,6 +128,7 @@ void DefaultPacketDispatcher::DrainReadableFd(LinkSocket& socket) noexcept { dispatching_ = false; Sweep(); + return !failed; } void DefaultPacketDispatcher::DispatchPacket(const LinkSocket& socket, const Packet& packet) const { diff --git a/src/reflector/default_packet_dispatcher.h b/src/reflector/default_packet_dispatcher.h index afd902c..b356634 100644 --- a/src/reflector/default_packet_dispatcher.h +++ b/src/reflector/default_packet_dispatcher.h @@ -29,6 +29,15 @@ class DefaultPacketDispatcher : public PacketDispatcher, NoMove { [[nodiscard]] Dispatcher& UnderlyingDispatcher() noexcept override { return *dispatcher_; } + // Invoked after a drain whose capture reported a read failure — the kernel saying the capture + // is broken, which the owner repairs. Carries no argument: the owner has to re-examine every + // interface anyway, and one failing capture does not tell it which others also died. + using CaptureFailureCallback = Delegate; + + void OnCaptureFailure(const CaptureFailureCallback& callback) noexcept { + on_capture_failure_ = callback; + } + private: friend class DefaultPacketDispatcherTest; @@ -93,7 +102,9 @@ class DefaultPacketDispatcher : public PacketDispatcher, NoMove { bool Unregister(RegistrationId id) noexcept override; void OnReadable(int fd) noexcept; - void DrainReadableFd(LinkSocket& socket) noexcept; + // False when the capture read failed — the kernel reporting the capture rather than a frame, + // which is the owner's cue to re-examine the interface; the drain draws no conclusion itself. + [[nodiscard]] bool DrainReadableFd(LinkSocket& socket) noexcept; void DispatchPacket(const LinkSocket& socket, const Packet& packet) const; // Erases the registrations Unregister marked disabled (their dtors release the capture-source count), // then drops every capture source left at 0. Runs after DrainReadableFd's drain -- never mid-walk, @@ -101,6 +112,8 @@ class DefaultPacketDispatcher : public PacketDispatcher, NoMove { void Sweep() noexcept; Dispatcher* dispatcher_; + // Unbound unless the owner asked to hear about broken captures; OnReadable checks before use. + CaptureFailureCallback on_capture_failure_; // capture_sources_ is declared before registrations_ so registrations_ is destroyed FIRST: each // entry's dtor releases its CaptureSource's count, so the sources must still be alive. std::unordered_map capture_sources_; diff --git a/src/reflector/link_socket.h b/src/reflector/link_socket.h index f7b6a61..373c383 100644 --- a/src/reflector/link_socket.h +++ b/src/reflector/link_socket.h @@ -9,6 +9,7 @@ #include #include +#include #include #include @@ -33,10 +34,17 @@ class LinkSocket { [[nodiscard]] virtual bool IsValid() const noexcept = 0; [[nodiscard]] virtual int Fd() const noexcept = 0; - // The next parsed datagram, or nullopt when none is currently available (EAGAIN) or the - // next frame is unparseable. The payload may span the socket's buffer and stays valid only + // Why Receive() yielded no packet. Only Failed says anything about the interface; the others + // are ordinary outcomes of draining a queue. + enum class ReceiveError : uint8_t { + WouldBlock, // nothing queued right now + Dropped, // a frame arrived but is not one we forward: unparseable, or too big to hold + Failed, // the read itself failed; the socket has already logged the errno + }; + + // The next parsed datagram. The payload may span the socket's buffer and stays valid only // until the next Receive() on the same socket. - [[nodiscard]] virtual std::optional Receive() noexcept = 0; + [[nodiscard]] virtual std::expected Receive() noexcept = 0; #if !defined(__linux__) // macOS BPF batches several frames into one read(); this lets the dispatcher keep draining the diff --git a/src/reflector/raw_socket.cpp b/src/reflector/raw_socket.cpp index 24e52f1..488d389 100644 --- a/src/reflector/raw_socket.cpp +++ b/src/reflector/raw_socket.cpp @@ -566,9 +566,11 @@ void RawSocket::Close() noexcept { #endif } -std::optional RawSocket::Receive() noexcept { +std::expected RawSocket::Receive() noexcept { if (!fd_) { - return std::nullopt; + // Defensive: an invalid socket is never watched. Reporting Failed here would ask for a + // repair that re-attaching cannot deliver. + return std::unexpected(ReceiveError::WouldBlock); } #if defined(__linux__) @@ -577,30 +579,36 @@ std::optional RawSocket::Receive() noexcept { const ssize_t bytes = recv(fd_.Get(), receive_buffer_.data(), receive_buffer_.size(), MSG_TRUNC); if (bytes < 0) { if (IsWouldBlockErrno(errno)) { - return std::nullopt; + return std::unexpected(ReceiveError::WouldBlock); } + // Reported for any non-would-block errno rather than a guessed list: this only asks the + // owner to re-examine the interface, and Attached() is what decides. logger_.Error("Cannot receive frame: {}", Error::FromErrno()); - return std::nullopt; + return std::unexpected(ReceiveError::Failed); } if (static_cast(bytes) > receive_buffer_.size()) { logger_.Warning("Dropping oversized frame: {} bytes exceeds {}-byte receive buffer", bytes, receive_buffer_.size()); - return std::nullopt; + return std::unexpected(ReceiveError::Dropped); } - return ParseFrame({receive_buffer_.data(), static_cast(bytes)}); + auto packet = ParseFrame({receive_buffer_.data(), static_cast(bytes)}); + if (!packet) { + return std::unexpected(ReceiveError::Dropped); // ParseFrame logged the reason + } + return *packet; #else if (receive_buffer_offset_ >= receive_buffer_filled_) { const ssize_t bytes = read(fd_.Get(), receive_buffer_.data(), receive_buffer_.size()); if (bytes < 0) { if (IsWouldBlockErrno(errno)) { - return std::nullopt; + return std::unexpected(ReceiveError::WouldBlock); } logger_.Error("Cannot receive frame: {}", Error::FromErrno()); - return std::nullopt; + return std::unexpected(ReceiveError::Failed); } if (bytes == 0) { - return std::nullopt; + return std::unexpected(ReceiveError::WouldBlock); } receive_buffer_filled_ = static_cast(bytes); receive_buffer_offset_ = 0; @@ -610,7 +618,7 @@ std::optional RawSocket::Receive() noexcept { logger_.Error("BPF batch truncated: {} bytes remaining, need at least {} for header", receive_buffer_filled_ - receive_buffer_offset_, sizeof(bpf_hdr)); receive_buffer_offset_ = receive_buffer_filled_; - return std::nullopt; + return std::unexpected(ReceiveError::Dropped); } bpf_hdr header{}; std::memcpy(&header, receive_buffer_.data() + receive_buffer_offset_, sizeof(header)); @@ -621,7 +629,7 @@ std::optional RawSocket::Receive() noexcept { logger_.Error("BPF frame extends past batch end (frame_end {} > filled {})", frame_end, receive_buffer_filled_); receive_buffer_offset_ = receive_buffer_filled_; - return std::nullopt; + return std::unexpected(ReceiveError::Dropped); } receive_buffer_offset_ = BPF_WORDALIGN(frame_offset + header.bh_caplen); @@ -630,7 +638,7 @@ std::optional RawSocket::Receive() noexcept { if (header.bh_datalen > header.bh_caplen) { logger_.Warning("Dropping oversized frame: {} bytes exceeds {}-byte receive buffer", header.bh_datalen, receive_buffer_.size()); - return std::nullopt; + return std::unexpected(ReceiveError::Dropped); } // Fully captured, but bigger than anything the send path can re-emit — the BIOCGBLEN-sized @@ -639,10 +647,14 @@ std::optional RawSocket::Receive() noexcept { if (header.bh_caplen > MAX_FRAME_SIZE) { logger_.Warning("Dropping oversized frame: {} bytes exceeds the {}-byte frame ceiling", header.bh_caplen, MAX_FRAME_SIZE); - return std::nullopt; + return std::unexpected(ReceiveError::Dropped); } - return ParseFrame({receive_buffer_.data() + frame_offset, header.bh_caplen}); + auto packet = ParseFrame({receive_buffer_.data() + frame_offset, header.bh_caplen}); + if (!packet) { + return std::unexpected(ReceiveError::Dropped); // ParseFrame logged the reason + } + return *packet; #endif } diff --git a/src/reflector/raw_socket.h b/src/reflector/raw_socket.h index 3a5333d..b31beff 100644 --- a/src/reflector/raw_socket.h +++ b/src/reflector/raw_socket.h @@ -12,6 +12,7 @@ #include #include +#include #include #include #include @@ -98,7 +99,7 @@ class RawSocket : public LinkSocket, NoMove { // Returns nullopt when no datagram is currently available (EAGAIN), or when the next // frame is unparseable / fragmented — caller treats both the same way and tries again // on the next read event. - [[nodiscard]] std::optional Receive() noexcept override; + [[nodiscard]] std::expected Receive() noexcept override; #if !defined(__linux__) // True if there are unparsed bytes in the socket's userland buffer. macOS BPF batches diff --git a/tests/mocks/fake_link_socket.h b/tests/mocks/fake_link_socket.h index 8325c45..73ca35a 100644 --- a/tests/mocks/fake_link_socket.h +++ b/tests/mocks/fake_link_socket.h @@ -8,6 +8,7 @@ #include #include +#include #include #include #include @@ -42,7 +43,9 @@ struct FakeLinkSocket : LinkSocket { [[nodiscard]] bool IsValid() const noexcept override { return valid; } [[nodiscard]] int Fd() const noexcept override { return fd; } - [[nodiscard]] std::optional Receive() noexcept override { return std::nullopt; } + [[nodiscard]] std::expected Receive() noexcept override { + return std::unexpected(receive_error); + } #if !defined(__linux__) [[nodiscard]] bool HasBufferedData() const noexcept override { return false; } #endif @@ -95,6 +98,8 @@ struct FakeLinkSocket : LinkSocket { // that went away under the socket; `rebinds` counts recoveries so a test can assert the // capture was actually re-attached rather than merely re-resolved. bool attached = true; + // What Receive() reports; ReceiveError::Failed models a read the kernel refused. + ReceiveError receive_error = ReceiveError::WouldBlock; bool fail_rebind = false; size_t rebinds = 0; int fd = -1; diff --git a/tests/raw_socket_test.cpp b/tests/raw_socket_test.cpp index 7a1c374..ae318d2 100644 --- a/tests/raw_socket_test.cpp +++ b/tests/raw_socket_test.cpp @@ -722,8 +722,9 @@ TEST(RawSocketBatchTest, ReceiveWalksMultiFrameBpfBatch) { EXPECT_EQ(received, frame_count); EXPECT_TRUE(observed_buffered) << "Expected the buffer walker to leave bytes between frames — got single-frame reads"; - EXPECT_FALSE(capture.socket.Receive().has_value()) - << "Expected no more frames after draining the batch"; + const auto drained = capture.socket.Receive(); + ASSERT_FALSE(drained.has_value()) << "Expected no more frames after draining the batch"; + EXPECT_EQ(drained.error(), LinkSocket::ReceiveError::WouldBlock); } TEST(RawSocketBatchTest, ReceiveAdvancesPastUnparseableFrameInBatch) { @@ -758,7 +759,9 @@ TEST(RawSocketBatchTest, ReceiveAdvancesPastUnparseableFrameInBatch) { EXPECT_EQ(first->header.source.port, 11111); CaptureStdout([&] { - EXPECT_FALSE(capture.socket.Receive().has_value()); + const auto skipped = capture.socket.Receive(); + ASSERT_FALSE(skipped.has_value()); + EXPECT_EQ(skipped.error(), LinkSocket::ReceiveError::Dropped); }); const auto last = capture.socket.Receive(); @@ -781,7 +784,9 @@ TEST(RawSocketBatchTest, ReceiveDropsBpfTruncatedFrame) { ASSERT_TRUE(capture.WriteTruncatedFrame(f.bytes, 70000)); const std::string output = CaptureStdout([&] { - EXPECT_FALSE(capture.socket.Receive().has_value()); + const auto dropped = capture.socket.Receive(); + ASSERT_FALSE(dropped.has_value()); + EXPECT_EQ(dropped.error(), LinkSocket::ReceiveError::Dropped); }); EXPECT_NE(output.find("oversized frame"), std::string::npos) << output; } @@ -795,7 +800,9 @@ TEST(RawSocketReceiveTest, DropsFullyCapturedFrameLargerThanTheFrameCeiling) { ASSERT_TRUE(capture.WriteFrame(frame)); const std::string output = CaptureStdout([&] { - EXPECT_FALSE(capture.socket.Receive().has_value()); + const auto dropped = capture.socket.Receive(); + ASSERT_FALSE(dropped.has_value()); + EXPECT_EQ(dropped.error(), LinkSocket::ReceiveError::Dropped); }); EXPECT_NE(output.find("oversized frame"), std::string::npos) << output; } @@ -810,7 +817,9 @@ TEST(RawSocketReceiveTest, DropsFrameLargerThanReceiveBuffer) { ASSERT_TRUE(capture.WriteFrame(frame)); const std::string output = CaptureStdout([&] { - EXPECT_FALSE(capture.socket.Receive().has_value()); + const auto dropped = capture.socket.Receive(); + ASSERT_FALSE(dropped.has_value()); + EXPECT_EQ(dropped.error(), LinkSocket::ReceiveError::Dropped); }); EXPECT_NE(output.find("oversized frame"), std::string::npos) << output; } @@ -844,7 +853,7 @@ class RawSocketRequiresRootTest : public ::testing::Test { if (packet) { if (packet->header.dest.port == listener_port && packet->header.dest.addr == IpAddress::LoopbackV4()) { - return packet; + return *packet; } continue; } @@ -945,7 +954,7 @@ TEST_F(RawSocketRequiresRootTest, CapturesInjectedDatagramOnLoopback) { auto packet = socket->Receive(); if (packet && packet->header.source.port == INJECT_SRC_PORT && packet->header.dest.addr == IpAddress::BroadcastV4()) { - captured = std::move(packet); + captured = std::move(*packet); break; } if (!packet) { @@ -1016,7 +1025,7 @@ class RawSocketInterfacePairRequiresRootTest : public ::testing::Test { auto frame = peer.Receive(); if (frame && frame->header.source.port == INJECT_SRC_PORT && frame->header.dest.addr == dest_ip) { - return frame; + return *frame; } if (!frame) { pollfd pfd{.fd = peer.Fd(), .events = POLLIN, .revents = 0}; From bea00d0f0ad1d6060a24190b9489d2854c98117c Mon Sep 17 00:00:00 2001 From: Sergii Bogomolov Date: Mon, 10 Aug 2026 17:25:24 +0200 Subject: [PATCH 4/5] feat: reconcile interfaces and captures against the kernel A destroyed and recreated interface kept its old index, and its capture stayed bound to a dead kernel object, so the link went quiet until a restart. The reconcile re-resolves every interface, not just the one an event named: a returning interface reports a new index, which matches nothing we hold. It probes each capture first -- one getsockname against three syscalls for a name lookup -- and resolves when the probe fails or the drain asked for that interface. The second case is what catches a rename, which keeps both the index and the capture. Two timers back it up. The backstop is the only channel that does not wait on the kernel to say something; at 30s it rides a wakeup the poll already makes, costing one probe per capture. The retry runs each second and only while a repair is outstanding, since a rebind that failed on a transient error has nothing coming to finish it. The address monitor now reports a whole drain at once instead of once per index, into a fixed-size list that degrades to refresh-all when a drain names more interfaces than it holds. --- src/reflector/address_monitor.h | 17 ++- src/reflector/application.cpp | 98 ++++++++++++-- src/reflector/application.h | 37 ++++- src/reflector/default_address_monitor.cpp | 54 ++++---- src/reflector/default_address_monitor.h | 28 +++- tests/application_test.cpp | 158 +++++++++++++++++++++- tests/default_address_monitor_test.cpp | 49 ++++++- tests/mocks/fake_address_monitor.h | 22 ++- 8 files changed, 402 insertions(+), 61 deletions(-) diff --git a/src/reflector/address_monitor.h b/src/reflector/address_monitor.h index acb7fdd..bc43d68 100644 --- a/src/reflector/address_monitor.h +++ b/src/reflector/address_monitor.h @@ -2,6 +2,8 @@ #include "util/delegate.h" +#include + 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; + // 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 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 diff --git a/src/reflector/application.cpp b/src/reflector/application.cpp index 7893409..16d7a07 100644 --- a/src/reflector/application.cpp +++ b/src/reflector/application.cpp @@ -12,6 +12,7 @@ #include "util/fd_util.h" #include "wol_reflector.h" +#include #include #include #include @@ -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()} , address_monitor_{std::make_unique(*dispatcher_)} { + packet_dispatcher_.OnCaptureFailure(CreateDelegate<&Application::OnCaptureFailure>(this)); StartMonitor(); } @@ -51,6 +61,7 @@ Application::Application(std::unique_ptr 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, 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(config.WolConfigs(), "wol") && ConfigureReflectors(config.MdnsConfigs(), "mdns") && ConfigureReflectors(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 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 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(); } diff --git a/src/reflector/application.h b/src/reflector/application.h index 6970140..428d6f5 100644 --- a/src/reflector/application.h +++ b/src/reflector/application.h @@ -17,6 +17,7 @@ #include #include #include +#include #include #include #include @@ -74,7 +75,7 @@ class Application : NoMove { Application(std::unique_ptr dispatcher, std::unique_ptr 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 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 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 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. diff --git a/src/reflector/default_address_monitor.cpp b/src/reflector/default_address_monitor.cpp index 753bf42..57e9221 100644 --- a/src/reflector/default_address_monitor.cpp +++ b/src/reflector/default_address_monitor.cpp @@ -13,7 +13,6 @@ #include #include #include -#include #include #include @@ -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& 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 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 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{}, 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 messages, - std::vector& changed) const noexcept { + ChangedInterfaces& changed) const noexcept { auto* header = start_lifetime_as(messages.data()); for (int length = static_cast(messages.size()); NLMSG_OK(header, length); header = start_lifetime_as(NLMSG_NEXT(header, length))) { if (header->nlmsg_type == RTM_NEWADDR || header->nlmsg_type == RTM_DELADDR) { const auto* address = start_lifetime_as(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(NLMSG_DATA(header)); - AddUnique(changed, static_cast(link->ifi_index)); + changed.Add(static_cast(link->ifi_index)); } } } @@ -266,7 +274,7 @@ void DefaultAddressMonitor::CollectChangedInterfaces(std::span messag #else void DefaultAddressMonitor::CollectChangedInterfaces(std::span messages, - std::vector& 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 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; } diff --git a/src/reflector/default_address_monitor.h b/src/reflector/default_address_monitor.h index cc6e7fe..2d4c50f 100644 --- a/src/reflector/default_address_monitor.h +++ b/src/reflector/default_address_monitor.h @@ -5,9 +5,9 @@ #include "util/no_move.h" #include "util/unique_fd.h" +#include #include #include -#include #include namespace reflector { @@ -43,10 +43,14 @@ class DefaultAddressMonitor : public AddressMonitor, NoMove { [[nodiscard]] static DefaultAddressMonitor ForTesting(Dispatcher& dispatcher, int fd, bool verify_sender = false); - [[nodiscard]] bool Start(const OnInterfaceChanged& on_change) noexcept override; + [[nodiscard]] bool Start(const OnInterfacesChanged& on_change) noexcept override; [[nodiscard]] bool IsValid() const noexcept { return fd_.IsValid(); } + // How many distinct interfaces one drain can list before it degrades to refresh-all. Public so + // a test can drive that degrade path. + static constexpr size_t MAX_CHANGED_INTERFACES = 32; + private: // Used by ForTesting: adopts an already-open `fd` instead of opening the kernel socket. // Watching begins at Start(). @@ -66,18 +70,34 @@ class DefaultAddressMonitor : public AddressMonitor, NoMove { // the Dispatcher fd callback by Watch(); the int argument (the ready fd) is unused. void OnReadable(int fd) noexcept; + // The interfaces one drain named, deduplicated. Fixed capacity keeps the notification path + // free of allocation; a drain naming more than it holds sets `overflowed`, which reports as + // refresh-all — the correct answer for "more changed than this can list". + struct ChangedInterfaces { + // Appends `index` unless it is already listed; sets `overflowed` instead once full. + void Add(unsigned index) noexcept; + + [[nodiscard]] std::span Indexes() const noexcept { + return std::span{indexes}.first(count); + } + + std::array indexes{}; + size_t count = 0; + bool overflowed = false; + }; + // Parses a buffer of kernel notification messages and appends each changed interface index to // `changed`, skipping any already present. Split out from OnReadable so tests can drive it with // synthesized messages. The span is mutable because the Linux walk start_lifetime_as's the // netlink structs over the received bytes, which reuses that storage. void CollectChangedInterfaces(std::span messages, - std::vector& changed) const noexcept; + ChangedInterfaces& changed) const noexcept; Dispatcher* dispatcher_; // Invalid (default-constructed) until Start() binds it (the testing constructor leaves it // unbound, so tests must call Start() too). Always valid by the time the fd is watched — // Watch() runs only inside Start(), after the bind — so OnReadable can call it. - OnInterfaceChanged on_change_; + OnInterfacesChanged on_change_; Dispatcher::Registration registration_; UniqueFd fd_; // Whether OnReadable rejects datagrams whose source isn't the kernel (production always does; diff --git a/tests/application_test.cpp b/tests/application_test.cpp index d2e938c..81190ac 100644 --- a/tests/application_test.cpp +++ b/tests/application_test.cpp @@ -187,8 +187,8 @@ TEST_F(ApplicationTest, StartsMemoryReportTimerWhenDebugMemoryEnabled) { ASSERT_TRUE(app.Configure(config)); - // A WoL-only config starts no timers of its own, so the periodic memory report is the only one. - EXPECT_EQ(dispatcher_->TimerCount(), 1u); + // The reconcile backstop always runs; debug_memory adds the periodic report on top of it. + EXPECT_EQ(dispatcher_->TimerCount(), 2u); // Firing it exercises the ReportMemory callback (reads /proc + mallinfo2 on glibc); must not crash. dispatcher_->FireTimers(std::chrono::steady_clock::now()); } @@ -201,7 +201,7 @@ TEST_F(ApplicationTest, NoMemoryReportTimerByDefault) { ASSERT_TRUE(app.Configure(config)); - EXPECT_EQ(dispatcher_->TimerCount(), 0u); + EXPECT_EQ(dispatcher_->TimerCount(), 1u); // the reconcile backstop, and nothing else } TEST_F(ApplicationTest, CreatesDistinctSocketsForDistinctInterfaces) { @@ -503,7 +503,7 @@ TEST_F(ApplicationTest, RefreshesAllInterfacesOnOverflowSignal) { auto app = MakeApp(); ASSERT_TRUE(app.Configure(TestConfigBuilder{}.Add(MakeWolConfig("tv", "src", "dst", {9})).Build())); - monitor_->FireChange(0); // 0 is the "refresh everything" overflow signal + monitor_->FireOverflow(); // the drain that lost its list and can only say "re-resolve all" EXPECT_EQ(Iface("src")->refresh_count, 1u); EXPECT_EQ(Iface("dst")->refresh_count, 1u); @@ -521,6 +521,156 @@ TEST_F(ApplicationTest, RefreshesNothingForUnknownInterface) { EXPECT_EQ(Iface("dst")->refresh_count, 0u); } +// A recreated interface keeps its name but gets a new kernel index; the capture was bound to the +// old object, so it has to re-attach. +TEST_F(ApplicationTest, RebindsTheCaptureWhenTheInterfaceIsRecreated) { + ConfigureSocket("src", {.interface_index = 5}); + ConfigureSocket("dst", {.interface_index = 9}); + auto app = MakeApp(); + ASSERT_TRUE(app.Configure(TestConfigBuilder{}.Add(MakeWolConfig("tv", "src", "dst", {9})).Build())); + + Iface("src")->StageIndex(12); + monitor_->FireChange(5); + + EXPECT_EQ(Iface("src")->Index(), 12u); + EXPECT_EQ(Socket("src")->rebinds, 1u); + EXPECT_EQ(Socket("dst")->rebinds, 0u); // untouched interface, untouched capture +} + +// The other recreation mode, and the reason Attached() exists: a kernel that hands the recreated +// interface the number it had leaves the index looking untouched while the capture points at a +// dead object. Measured on macOS; Linux takes the branch above. +TEST_F(ApplicationTest, RebindsTheCaptureWhenTheIndexIsReusedOnRecreation) { + ConfigureSocket("src", {.interface_index = 5}); + ConfigureSocket("dst", {.interface_index = 9}); + auto app = MakeApp(); + ASSERT_TRUE(app.Configure(TestConfigBuilder{}.Add(MakeWolConfig("tv", "src", "dst", {9})).Build())); + + Socket("src")->attached = false; // same index, capture detached under it + monitor_->FireChange(5); + + EXPECT_EQ(Socket("src")->rebinds, 1u); + EXPECT_TRUE(Socket("src")->attached); +} + +// An interface that is gone has no kernel object to bind to, so retrying the capture would only +// fail. It waits parked until an index resolves again. +TEST_F(ApplicationTest, DoesNotRebindWhileTheInterfaceIsGone) { + ConfigureSocket("src", {.interface_index = 5}); + ConfigureSocket("dst", {.interface_index = 9}); + auto app = MakeApp(); + ASSERT_TRUE(app.Configure(TestConfigBuilder{}.Add(MakeWolConfig("tv", "src", "dst", {9})).Build())); + + Iface("src")->StageIndex(0); + Socket("src")->attached = false; + monitor_->FireChange(5); + + EXPECT_FALSE(Iface("src")->IsValid()); + EXPECT_EQ(Socket("src")->rebinds, 0u); +} + +// Monitor traffic is constant on a busy link; a plain address change must not tear the capture +// down and rebuild it. +TEST_F(ApplicationTest, LeavesTheCaptureAloneOnAnAddressChange) { + ConfigureSocket("src", {.interface_index = 5}); + ConfigureSocket("dst", {.interface_index = 9}); + auto app = MakeApp(); + ASSERT_TRUE(app.Configure(TestConfigBuilder{}.Add(MakeWolConfig("tv", "src", "dst", {9})).Build())); + + monitor_->FireChange(5); + + EXPECT_EQ(Socket("src")->rebinds, 0u); + EXPECT_EQ(Iface("src")->refresh_count, 1u); +} + +// The backstop is the only channel that does not wait on the kernel: no notification here at all. +TEST_F(ApplicationTest, TheBackstopRebindsADetachedCapture) { + ConfigureSocket("src", {.interface_index = 5}); + ConfigureSocket("dst", {.interface_index = 9}); + auto app = MakeApp(); + ASSERT_TRUE(app.Configure(TestConfigBuilder{}.Add(MakeWolConfig("tv", "src", "dst", {9})).Build())); + + Socket("src")->attached = false; + dispatcher_->FireTimers(std::chrono::steady_clock::now()); + + EXPECT_EQ(Socket("src")->rebinds, 1u); + EXPECT_EQ(Socket("dst")->rebinds, 0u); +} + +// A rebind that failed has no announcement coming, so the pass has to schedule its own retry — +// and stop it again once the repair lands, so a healthy daemon runs only the backstop. +TEST_F(ApplicationTest, RetriesUntilAFailedRepairCompletes) { + ConfigureSocket("src", {.interface_index = 5}); + ConfigureSocket("dst", {.interface_index = 9}); + auto app = MakeApp(); + ASSERT_TRUE(app.Configure(TestConfigBuilder{}.Add(MakeWolConfig("tv", "src", "dst", {9})).Build())); + + Socket("src")->attached = false; + Socket("src")->fail_rebind = true; + monitor_->FireChange(5); + EXPECT_EQ(dispatcher_->TimerCount(), 2u); // backstop plus the retry + + Socket("src")->fail_rebind = false; + dispatcher_->FireTimers(std::chrono::steady_clock::now()); + + EXPECT_TRUE(Socket("src")->attached); + EXPECT_EQ(dispatcher_->TimerCount(), 1u); // repaired, so the retry stands down +} + +// A lookup that cannot run says nothing about the interface. Reading it as absence would clear a +// live interface's addresses and shut its egress gate over a transient fd shortage. +TEST_F(ApplicationTest, KeepsTheIdentityWhenTheLookupCannotRun) { + ConfigureSocket("src", {.interface_index = 5}); + ConfigureSocket("dst", {.interface_index = 9}); + auto app = MakeApp(); + ASSERT_TRUE(app.Configure(TestConfigBuilder{}.Add(MakeWolConfig("tv", "src", "dst", {9})).Build())); + + Iface("src")->StageIndex(std::nullopt); + monitor_->FireChange(5); + + EXPECT_EQ(Iface("src")->Index(), 5u); + EXPECT_TRUE(Iface("src")->CanSend(IpAddress::Family::V4)); + EXPECT_EQ(Socket("src")->rebinds, 0u); + EXPECT_EQ(dispatcher_->TimerCount(), 2u); // retried rather than acted on +} + +// A failed read is a reason to re-examine the interface, and on a link whose notification never +// arrives it is the only one that shows up. +TEST_F(ApplicationTest, ACaptureReadFailureSchedulesTheRepair) { + ConfigureSocket("src", {.interface_index = 5}); + ConfigureSocket("dst", {.interface_index = 9}); + auto app = MakeApp(); + ASSERT_TRUE(app.Configure(TestConfigBuilder{}.Add(MakeWolConfig("tv", "src", "dst", {9})).Build())); + ASSERT_EQ(dispatcher_->TimerCount(), 1u); // just the backstop + + Socket("src")->receive_error = LinkSocket::ReceiveError::Failed; + Socket("src")->attached = false; // the interface really did go away under it + dispatcher_->FireReadable(Socket("src")->fd); + EXPECT_EQ(dispatcher_->TimerCount(), 2u); // repaired on the retry, not inside the drain + + dispatcher_->FireTimers(std::chrono::steady_clock::now()); + + EXPECT_EQ(Socket("src")->rebinds, 1u); + EXPECT_EQ(dispatcher_->TimerCount(), 1u); +} + +// The read error that is not about the interface: the kernel parks ENETDOWN on a capture whose +// interface merely went down, and re-attaching a capture that is still attached fixes nothing. +TEST_F(ApplicationTest, ACaptureReadFailureLeavesAnAttachedCaptureAlone) { + ConfigureSocket("src", {.interface_index = 5}); + ConfigureSocket("dst", {.interface_index = 9}); + auto app = MakeApp(); + ASSERT_TRUE(app.Configure(TestConfigBuilder{}.Add(MakeWolConfig("tv", "src", "dst", {9})).Build())); + + // attached stays true: the interface is still there + Socket("src")->receive_error = LinkSocket::ReceiveError::Failed; + dispatcher_->FireReadable(Socket("src")->fd); + dispatcher_->FireTimers(std::chrono::steady_clock::now()); + + EXPECT_EQ(Socket("src")->rebinds, 0u); + EXPECT_EQ(dispatcher_->TimerCount(), 1u); // nothing outstanding, so the retry stands down +} + TEST_F(ApplicationTest, FailsConfigureWhenTheInterfaceIsInvalid) { ConfigureSocket("src", {.interface_valid = false}); auto app = MakeApp(); diff --git a/tests/default_address_monitor_test.cpp b/tests/default_address_monitor_test.cpp index d46e82b..5f35b24 100644 --- a/tests/default_address_monitor_test.cpp +++ b/tests/default_address_monitor_test.cpp @@ -27,11 +27,18 @@ namespace { -// Records the interface indices the monitor reports, so tests assert on its actual on_change output. +// Records what the monitor reports, so tests assert on its actual on_change output. Indexes +// accumulate across drains; `drains` counts the calls, which is what pins one-call-per-drain. struct RecordingChangeSink { - void OnChange(unsigned interface_index) noexcept { changed.push_back(interface_index); } + void OnChange(std::span indexes, bool all) noexcept { + changed.insert(changed.end(), indexes.begin(), indexes.end()); + refresh_all += all ? 1 : 0; + ++drains; + } std::vector changed; + int refresh_all = 0; + int drains = 0; }; #if defined(__linux__) @@ -281,8 +288,8 @@ TEST_F(DefaultAddressMonitorTest, IgnoresUnrelatedMessages) { } TEST_F(DefaultAddressMonitorTest, NeverForwardsIndexZero) { - // 0 names no interface (kernel indices are >= 1) and is the overflow path's refresh-all - // sentinel — a stray 0 in a kernel message must not masquerade as it. + // Kernel indices are >= 1, so a 0 in a message names no interface and would only make a + // subscriber resolve something that cannot exist. auto monitor = MakeMonitor(); ASSERT_TRUE(StartWatching(monitor)); std::vector messages; @@ -296,6 +303,40 @@ TEST_F(DefaultAddressMonitorTest, NeverForwardsIndexZero) { EXPECT_EQ(sink.changed, (std::vector{6})); } +TEST_F(DefaultAddressMonitorTest, ReportsAWholeDrainAsOneCall) { + auto monitor = MakeMonitor(); + ASSERT_TRUE(StartWatching(monitor)); + std::vector messages; + AppendAddrMessage(messages, ADDR_MESSAGE, 3); + AppendAddrMessage(messages, ADDR_MESSAGE, 5); + AppendAddrMessage(messages, ADDR_MESSAGE, 3); // a repeat, coalesced away + + Write(messages); + FireReadable(); + + EXPECT_EQ(sink.drains, 1); + EXPECT_EQ(sink.changed, (std::vector{3, 5})); + EXPECT_EQ(sink.refresh_all, 0); +} + +TEST_F(DefaultAddressMonitorTest, ReportsRefreshAllWhenOneDrainNamesMoreThanItLists) { + auto monitor = MakeMonitor(); + ASSERT_TRUE(StartWatching(monitor)); + std::vector messages; + for (size_t index = 1; index <= DefaultAddressMonitor::MAX_CHANGED_INTERFACES + 1; ++index) { + // The two platform helpers take different index widths; uint16_t converts cleanly to both. + AppendAddrMessage(messages, ADDR_MESSAGE, narrow_cast(index)); + } + + Write(messages); + FireReadable(); + + // A partial list would silently skip whichever interfaces fell off the end, so the drain says + // "re-resolve everything" instead. + EXPECT_EQ(sink.refresh_all, 1); + EXPECT_TRUE(sink.changed.empty()); +} + #if defined(__linux__) TEST_F(DefaultAddressMonitorTest, DropsNotificationsFromANonKernelSender) { // With sender verification on, a datagram whose source isn't the kernel is dropped. The diff --git a/tests/mocks/fake_address_monitor.h b/tests/mocks/fake_address_monitor.h index 39309eb..d3d915d 100644 --- a/tests/mocks/fake_address_monitor.h +++ b/tests/mocks/fake_address_monitor.h @@ -2,6 +2,9 @@ #include "reflector/address_monitor.h" +#include +#include + namespace reflector { // Fake AddressMonitor: stands in for the real netlink/route monitor so an owner can be wired @@ -10,25 +13,34 @@ namespace reflector { // can be exercised. class FakeAddressMonitor : public AddressMonitor { public: - [[nodiscard]] bool Start(const OnInterfaceChanged& on_change) noexcept override { + [[nodiscard]] bool Start(const OnInterfacesChanged& on_change) noexcept override { on_change_ = on_change; return start_succeeds; } - // Invokes the subscribed callback as a kernel address-change notification would. - void FireChange(unsigned interface_index) { + // Invokes the subscribed callback as one drain of kernel notifications would. + void FireChanges(std::span indexes, bool refresh_all = false) { if (on_change_.IsValid()) { - on_change_(interface_index); + on_change_(indexes, refresh_all); } } + // The single-interface drain, which is what most tests want. + void FireChange(unsigned interface_index) { + const std::array indexes{interface_index}; + FireChanges(indexes); + } + + // The drain that lost notifications and so can only say "re-resolve everything". + void FireOverflow() { FireChanges({}, true); } + // True once Start() has been given a bound callback. [[nodiscard]] bool Started() const noexcept { return on_change_.IsValid(); } bool start_succeeds = true; private: - OnInterfaceChanged on_change_; + OnInterfacesChanged on_change_; }; } // namespace reflector From 710fc54c856e69432bd05f13db4a05a259e64f72 Mon Sep 17 00:00:00 2001 From: Sergii Bogomolov Date: Mon, 10 Aug 2026 17:26:38 +0200 Subject: [PATCH 5/5] fix: keep an interface's addresses when the enumeration cannot run Both resolvers bailed out on a resource failure and returned an empty address set, which Refresh assigned -- so a transient EMFILE blanked a live interface and shut its egress gate. They now report whether the enumeration ran at all, and Refresh keeps what it had when it did not. An interface that genuinely has no addresses still resolves to an empty set. --- src/reflector/interface.cpp | 11 ++++++++-- src/reflector/interface_address.cpp | 32 ++++++++++++++++++----------- src/reflector/interface_address.h | 9 +++++--- tests/interface_address_test.cpp | 15 +++++++++----- tests/raw_socket_test.cpp | 2 +- 5 files changed, 46 insertions(+), 23 deletions(-) diff --git a/src/reflector/interface.cpp b/src/reflector/interface.cpp index 11804e1..3817beb 100644 --- a/src/reflector/interface.cpp +++ b/src/reflector/interface.cpp @@ -101,10 +101,17 @@ bool Interface::CanSend(IpAddress::Family family) const noexcept { void Interface::Refresh() noexcept { #if defined(__linux__) - addresses_ = ResolveInterfaceAddresses(index_); + auto resolved = ResolveInterfaceAddresses(index_); #else - addresses_ = ResolveInterfaceAddresses(name_); + auto resolved = ResolveInterfaceAddresses(name_); #endif + if (!resolved) { + // The enumeration could not run (it logged why), which says nothing about the interface. + // Keeping the last known addresses beats closing a live interface's egress gate over a + // transient shortage. + return; + } + addresses_ = *resolved; logger_.Debug("Resolved addresses (index {}): MAC {}, IPv4 {}, IPv6 {}, IPv6 routable {}", index_, addresses_.mac, addresses_.v4 ? addresses_.v4->ToString() : "none", addresses_.v6 ? addresses_.v6->ToString() : "none", diff --git a/src/reflector/interface_address.cpp b/src/reflector/interface_address.cpp index 9d35e85..af00ba8 100644 --- a/src/reflector/interface_address.cpp +++ b/src/reflector/interface_address.cpp @@ -103,7 +103,7 @@ bool IsUsable(uint32_t ifa_flags) noexcept { // `handle` until NLMSG_DONE. Dumps everything and lets the handler filter by interface index — // simpler and uniformly terminated than a single-interface request, and the set is tiny. template -bool NetlinkDump(int fd, uint16_t request_type, uint32_t seq, Handler&& handle) noexcept { +[[nodiscard]] bool NetlinkDump(int fd, uint16_t request_type, uint32_t seq, Handler&& handle) noexcept { struct { nlmsghdr header; union { @@ -179,14 +179,14 @@ bool NetlinkDump(int fd, uint16_t request_type, uint32_t seq, Handler&& handle) } } -void ResolveViaNetlink(unsigned index, InterfaceAddresses& result) noexcept { +bool ResolveViaNetlink(unsigned index, InterfaceAddresses& result) noexcept { const int fd = socket(AF_NETLINK, SOCK_RAW, NETLINK_ROUTE); if (fd < 0) { GetLogger().Error("Cannot open netlink socket: {}", Error::FromErrno()); - return; + return false; } - NetlinkDump(fd, RTM_GETLINK, 1, [&](nlmsghdr* header) { + const bool link_dumped = NetlinkDump(fd, RTM_GETLINK, 1, [&](nlmsghdr* header) { const auto* link = start_lifetime_as(NLMSG_DATA(header)); if (static_cast(link->ifi_index) != index) { return; @@ -203,7 +203,7 @@ void ResolveViaNetlink(unsigned index, InterfaceAddresses& result) noexcept { }); std::vector candidates; - NetlinkDump(fd, RTM_GETADDR, 2, [&](nlmsghdr* header) { + const bool addresses_dumped = NetlinkDump(fd, RTM_GETADDR, 2, [&](nlmsghdr* header) { const auto* addr = start_lifetime_as(NLMSG_DATA(header)); if (addr->ifa_index != index) { return; @@ -252,6 +252,9 @@ void ResolveViaNetlink(unsigned index, InterfaceAddresses& result) noexcept { close(fd); detail::SelectSourceAddresses(candidates, result); + // Either dump failing means we did not see the interface's whole address set, so the caller + // must not read the result as "it has none". + return link_dumped && addresses_dumped; } #pragma GCC diagnostic pop @@ -274,21 +277,21 @@ bool IsUsableIpv6(int inet6_fd, const char* interface, const sockaddr_in6& sin6) return detail::Ipv6SourceFlagsUsable(request.ifr_ifru.ifru_flags6); } -void ResolveViaGetifaddrs(std::string_view interface, InterfaceAddresses& result) noexcept { +bool ResolveViaGetifaddrs(std::string_view interface, InterfaceAddresses& result) noexcept { // Unbound datagram socket for the per-address SIOCGIFAFLAG_IN6 flag queries. On a host with // IPv6 addresses to enumerate this effectively always succeeds; if it can't be opened we // can't verify any IPv6 source, so fail early rather than guess. const int inet6_fd = socket(AF_INET6, SOCK_DGRAM, 0); if (inet6_fd < 0) { GetLogger().Error("Cannot open IPv6 socket to query address flags: {}", Error::FromErrno()); - return; + return false; } ifaddrs* head = nullptr; if (getifaddrs(&head) != 0) { GetLogger().Error("Cannot enumerate interface addresses: {}", Error::FromErrno()); close(inet6_fd); - return; + return false; } std::vector candidates; @@ -331,6 +334,7 @@ void ResolveViaGetifaddrs(std::string_view interface, InterfaceAddresses& result freeifaddrs(head); detail::SelectSourceAddresses(candidates, result); + return true; } #endif @@ -384,17 +388,21 @@ IpAddress CanonicalizeLinkLocalV6(const IpAddress& address) noexcept { #if defined(__linux__) -InterfaceAddresses ResolveInterfaceAddresses(unsigned interface_index) noexcept { +std::optional ResolveInterfaceAddresses(unsigned interface_index) noexcept { InterfaceAddresses result; - ResolveViaNetlink(interface_index, result); + if (!ResolveViaNetlink(interface_index, result)) { + return std::nullopt; + } return result; } #else -InterfaceAddresses ResolveInterfaceAddresses(std::string_view interface) noexcept { +std::optional ResolveInterfaceAddresses(std::string_view interface) noexcept { InterfaceAddresses result; - ResolveViaGetifaddrs(interface, result); + if (!ResolveViaGetifaddrs(interface, result)) { + return std::nullopt; + } return result; } diff --git a/src/reflector/interface_address.h b/src/reflector/interface_address.h index aac3170..6f754f7 100644 --- a/src/reflector/interface_address.h +++ b/src/reflector/interface_address.h @@ -53,16 +53,19 @@ void SelectSourceAddresses(std::span candidates, InterfaceAddre } // namespace detail // Resolves an interface's MAC and per-family source addresses; fields are left empty for -// anything it lacks (or if it's unknown). The IPv6 result prefers a link-local address (the +// anything it lacks (or if it's unknown). nullopt means the enumeration could not run at all — +// distinct from an interface that genuinely has no addresses, which resolves to an empty set, so +// a caller can keep what it had rather than blank a live interface over a transient failure. +// The IPv6 result prefers a link-local address (the // correct source for the link-local multicast we send), falling back to ULA then GUA, and skips // tentative/deprecated/duplicated addresses. Needs no special privilege. // // Keyed by the identifier each platform resolves natively — and that RawSocket already holds: // the kernel interface index on Linux (netlink), the interface name on macOS (getifaddrs). #if defined(__linux__) -[[nodiscard]] InterfaceAddresses ResolveInterfaceAddresses(unsigned interface_index) noexcept; +[[nodiscard]] std::optional ResolveInterfaceAddresses(unsigned interface_index) noexcept; #else -[[nodiscard]] InterfaceAddresses ResolveInterfaceAddresses(std::string_view interface) noexcept; +[[nodiscard]] std::optional ResolveInterfaceAddresses(std::string_view interface) noexcept; #endif } // namespace reflector diff --git a/tests/interface_address_test.cpp b/tests/interface_address_test.cpp index ed49257..8859502 100644 --- a/tests/interface_address_test.cpp +++ b/tests/interface_address_test.cpp @@ -29,10 +29,12 @@ using namespace reflector; // netlink resolver and production refresh use), name on macOS. InterfaceAddresses ResolveLoopback() { #if defined(__linux__) - return ResolveInterfaceAddresses(if_nametoindex(std::string{LoopbackInterface()}.c_str())); + auto addresses = ResolveInterfaceAddresses(if_nametoindex(std::string{LoopbackInterface()}.c_str())); #else - return ResolveInterfaceAddresses(LoopbackInterface()); + auto addresses = ResolveInterfaceAddresses(LoopbackInterface()); #endif + EXPECT_TRUE(addresses.has_value()) << "loopback address enumeration did not run"; + return addresses.value_or(InterfaceAddresses{}); } } // namespace @@ -82,9 +84,12 @@ TEST(InterfaceAddressTest, UnknownInterfaceResolvesNothing) { #else const auto addresses = ResolveInterfaceAddresses("nonexistent-reflector-iface"); #endif - EXPECT_FALSE(addresses.v4.has_value()); - EXPECT_FALSE(addresses.v6.has_value()); - EXPECT_EQ(addresses.mac, MacAddress{}); + // The enumeration ran and found nothing, which is not the same as failing to run — the + // caller keeps its addresses in the second case and clears them in this one. + ASSERT_TRUE(addresses.has_value()); + EXPECT_FALSE(addresses->v4.has_value()); + EXPECT_FALSE(addresses->v6.has_value()); + EXPECT_EQ(addresses->mac, MacAddress{}); } // The source-selection policy, driven directly with synthetic candidate lists — the part the diff --git a/tests/raw_socket_test.cpp b/tests/raw_socket_test.cpp index ae318d2..41a03f1 100644 --- a/tests/raw_socket_test.cpp +++ b/tests/raw_socket_test.cpp @@ -1047,7 +1047,7 @@ class RawSocketInterfacePairRequiresRootTest : public ::testing::Test { #else const auto addresses = ResolveInterfaceAddresses(pair.ReceiveInterface()); #endif - if (addresses.v6 && addresses.v6->IsLinkLocal()) { + if (addresses && addresses->v6 && addresses->v6->IsLinkLocal()) { return true; } ::poll(nullptr, 0, POLL_SLICE_MS);