network: detect a wedged discv5 socket (operator health check + boot-node /healthz)

network/discovery/dv5_service.go

@@ -78,6 +78,9 @@ type DiscV5Service struct {
 
 	conn       *net.UDPConn
 	sharedConn *SharedUDPConn
+	// socketConn wraps conn to time socket reads; the post-fork listener drains
+	// through it, so DiscoveryStale can detect a wedged socket.
+	socketConn *TimedConn
 
 	ssvConfig *networkconfig.SSV
 	subnets   commons.Subnets
@@ -159,6 +162,21 @@ func (dvs *DiscV5Service) close() error {
 	return nil
 }
 
+// DiscoveryStale reports whether the discv5 socket has gone unread for longer
+// than grace — the sign discovery has wedged. False before the listener is built
+// (or after a failed init), when there's nothing to judge.
+//
+// The operator health check polls this periodically, so it doubles as the
+// sampling point for the read-staleness gauge.
+func (dvs *DiscV5Service) DiscoveryStale(grace time.Duration) bool {
+	if dvs.socketConn == nil {
+		return false
+	}
+	age, ok := dvs.socketConn.ReadStaleness()
+	recordDiscoveryReadStaleness(dvs.ctx, int64(age.Seconds()))
+	return ok && age > grace
+}
+
 // Self returns self node
 func (dvs *DiscV5Service) Self() *enode.LocalNode {
 	return dvs.dv5Listener.LocalNode()
@@ -402,13 +420,20 @@ func (dvs *DiscV5Service) initDiscV5Listener(discOpts *Options) (err error) {
 	}
 	dvs.conn = udpConn
 
+	// Wrap the socket for liveness (see DiscoveryStale): only the post-fork
+	// listener drains it, so only it gets the wrapped conn — the pre-fork
+	// listener reads sharedConn's buffer, not the socket.
+	socketConn := NewTimedConn(udpConn)
+	dvs.socketConn = socketConn
+
 	// Registered before anything else can fail, so every error path below
 	// releases the socket. Runs last of the deferred cleanups, by which point a
 	// listener may already have closed it.
 	defer func() {
 		if err != nil {
 			_ = udpConn.Close()
 			dvs.conn = nil
+			dvs.socketConn = nil
 		}
 	}()
 
@@ -453,7 +478,7 @@ func (dvs *DiscV5Service) initDiscV5Listener(discOpts *Options) (err error) {
 		return err
 	}
 
-	dv5PostForkListener, err := listenV5(udpConn, localNode, *dv5PostForkCfg)
+	dv5PostForkListener, err := listenV5(socketConn, localNode, *dv5PostForkCfg)
 	if err != nil {
 		return fmt.Errorf("could not create discV5 listener: %w", err)
 	}

network/discovery/dv5_service_stale_test.go

@@ -0,0 +1,35 @@
+package discovery
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+// TestDiscV5Service_DiscoveryStale covers the operator-side liveness accessor:
+// nil before the listener is built, never stale until the socket has actually
+// been read, and driven by the wrapped socket's last-read time thereafter.
+func TestDiscV5Service_DiscoveryStale(t *testing.T) {
+	// No socket wrapped yet (not-yet / failed init) → never stale.
+	require.False(t, (&DiscV5Service{}).DiscoveryStale(3*time.Minute))
+
+	sc := NewTimedConn(nil)
+	t0 := sc.LastRead()
+	dvs := &DiscV5Service{ctx: t.Context(), socketConn: sc}
+
+	// Never read → never stale, however long it sits idle: an operator that has
+	// simply had no inbound discv5 traffic yet must not be flagged as wedged.
+	sc.now = func() time.Time { return t0.Add(time.Hour) }
+	require.False(t, dvs.DiscoveryStale(3*time.Minute), "never-read socket is not a wedge")
+
+	// A read arms the signal, stamped at t0.
+	sc.read.Store(true)
+	sc.lastReadUnixNano.Store(t0.UnixNano())
+
+	sc.now = func() time.Time { return t0.Add(2 * time.Minute) }
+	require.False(t, dvs.DiscoveryStale(3*time.Minute), "read within grace → not stale")
+
+	sc.now = func() time.Time { return t0.Add(4 * time.Minute) }
+	require.True(t, dvs.DiscoveryStale(3*time.Minute), "read then unread past grace → stale")
+}

network/discovery/local_service.go

@@ -114,6 +114,12 @@ func (md *localDiscovery) PublishENR() {
 	// TODO
 }
 
+// DiscoveryStale implements Service. Local discovery has no discv5 socket to
+// wedge, so it's never stale.
+func (md *localDiscovery) DiscoveryStale(time.Duration) bool {
+	return false
+}
+
 // discoveryNotifee gets notified when we find a new peer via mDNS discovery
 type discoveryNotifee struct {
 	handler HandleNewPeer

network/discovery/observability.go

@@ -65,12 +65,22 @@ var (
 			observability.InstrumentName(observabilityNamespace, "unhandled_packets.dropped"),
 			metric.WithUnit("{packet}"),
 			metric.WithDescription("total number of packets forwarded to the pre-fork listener that were dropped because its buffer was full")))
+
+	discoveryReadStalenessGauge = metrics.New(
+		meter.Int64Gauge(
+			observability.InstrumentName(observabilityNamespace, "socket.read_staleness"),
+			metric.WithUnit("s"),
+			metric.WithDescription("seconds since the discv5 socket was last read; 0 until the first read")))
 )
 
 func recordUnhandledPacketDropped(ctx context.Context) {
 	unhandledPacketsDroppedCounter.Add(ctx, 1)
 }
 
+func recordDiscoveryReadStaleness(ctx context.Context, seconds int64) {
+	discoveryReadStalenessGauge.Record(ctx, seconds)
+}
+
 func recordPeerSkipped(ctx context.Context, reason skipReason) {
 	peerRejectionsCounter.Add(ctx, 1, metric.WithAttributes(peerSkipReasonAttribute(reason)))
 }

network/discovery/service.go

@@ -62,6 +62,9 @@ type Service interface {
 	DeregisterSubnets(subnets ...uint64) (updated bool, err error)
 	Bootstrap(handler HandleNewPeer) error
 	PublishENR()
+	// DiscoveryStale reports whether discovery has stopped draining its socket
+	// for longer than grace (the "wedge").
+	DiscoveryStale(grace time.Duration) bool
 }
 
 type DiscoveredPeer struct {

network/discovery/service_test.go

@@ -37,6 +37,7 @@ func TestNewDiscV5Service(t *testing.T) {
 
 	assert.NotNil(t, dvs.dv5Listener)
 	assert.NotNil(t, dvs.conns)
+	assert.NotNil(t, dvs.socketConn)
 	assert.NotNil(t, dvs.subnetsIdx)
 	assert.NotNil(t, dvs.ssvConfig)
 

network/discovery/shared_conn_test.go

@@ -312,6 +312,33 @@ func TestInitDiscV5Listener_CleansUpOnError(t *testing.T) {
 	}
 }
 
+// TestInitDiscV5Listener_WrapsPostForkConn pins the wiring the wedge detector
+// depends on: the post-fork listener — the only one that drains the real socket —
+// must read through the TimedConn, and the pre-fork listener must read the
+// SharedUDPConn buffer. Every other test builds these conns by hand, so without
+// this a refactor that passed the wrong conn to ListenV5 would leave the detector
+// silently reporting "healthy" forever while the whole suite still passed.
+//
+// Swaps the package-level listenV5, so the test must stay non-parallel.
+func TestInitDiscV5Listener_WrapsPostForkConn(t *testing.T) {
+	orig := listenV5
+	t.Cleanup(func() { listenV5 = orig })
+
+	var conns []discover.UDPConn
+	listenV5 = func(conn discover.UDPConn, ln *enode.LocalNode, cfg discover.Config) (Listener, error) {
+		conns = append(conns, conn)
+		return orig(conn, ln, cfg)
+	}
+
+	dvs := testingDiscovery(t)
+	t.Cleanup(func() { require.NoError(t, dvs.Close()) })
+
+	require.Len(t, conns, 2, "init builds the post-fork listener, then the pre-fork one")
+	require.IsType(t, &TimedConn{}, conns[0], "post-fork listener must read through the TimedConn")
+	require.Same(t, dvs.socketConn, conns[0].(*TimedConn), "and it must be the service's own socketConn")
+	require.IsType(t, &SharedUDPConn{}, conns[1], "pre-fork listener must read the SharedUDPConn buffer")
+}
+
 // TestSharedUDPConn_DrainsWhileClosing guards the shutdown ordering: the
 // producer keeps forwarding while the reader shuts down, and must not block or
 // panic. Previously Close() closed Unhandled, so an in-flight send panicked

network/discovery/timed_conn.go

@@ -0,0 +1,84 @@
+package discovery
+
+import (
+	"net"
+	"net/netip"
+	"sync/atomic"
+	"time"
+
+	"github.com/ethereum/go-ethereum/p2p/discover"
+)
+
+// TimedConn wraps a discv5 UDP socket and records when it was last read from.
+//
+// A discv5 listener drains its socket only through ReadFromUDPAddrPort, so a
+// wedged socket — bound but no longer read, leaving discovery silently dead —
+// shows up as a last-read timestamp that stops advancing. StaleFor turns that
+// into a liveness signal, used by both the operator and boot nodes.
+//
+// Only ReadFromUDPAddrPort is overridden; writes, Close and LocalAddr fall
+// through to the embedded conn.
+type TimedConn struct {
+	discover.UDPConn
+
+	lastReadUnixNano atomic.Int64
+	read             atomic.Bool      // set once the socket has been read at least once
+	now              func() time.Time // overridable in tests; nil means time.Now
+}
+
+var _ discover.UDPConn = (*TimedConn)(nil)
+
+// NewTimedConn wraps conn, seeding the last-read time so LastRead reports a
+// sensible value before the first read. Staleness only arms once the socket has
+// actually been read (see StaleFor).
+func NewTimedConn(conn *net.UDPConn) *TimedConn {
+	c := &TimedConn{UDPConn: conn}
+	c.lastReadUnixNano.Store(c.nowFn().UnixNano())
+	return c
+}
+
+func (c *TimedConn) nowFn() time.Time {
+	if c.now != nil {
+		return c.now()
+	}
+	return time.Now()
+}
+
+// ReadFromUDPAddrPort delegates and, on success, records the read time and marks
+// the socket as having been read. Errors leave both untouched, so a closed or
+// failing socket can't look freshly drained.
+func (c *TimedConn) ReadFromUDPAddrPort(b []byte) (n int, addr netip.AddrPort, err error) {
+	n, addr, err = c.UDPConn.ReadFromUDPAddrPort(b)
+	if err == nil {
+		c.lastReadUnixNano.Store(c.nowFn().UnixNano())
+		c.read.Store(true)
+	}
+	return n, addr, err
+}
+
+// LastRead reports when a packet was last read from the socket.
+func (c *TimedConn) LastRead() time.Time {
+	return time.Unix(0, c.lastReadUnixNano.Load())
+}
+
+// ReadStaleness reports how long the socket has gone unread, and whether it has
+// ever been read. Before the first successful read there is nothing to measure,
+// so ok is false: a socket that has never delivered a packet is not wedged, it
+// has simply had no inbound discv5 traffic yet (unreachable bootnodes, blocked
+// UDP), which a restart cannot fix.
+func (c *TimedConn) ReadStaleness() (age time.Duration, ok bool) {
+	if !c.read.Load() {
+		return 0, false
+	}
+	return c.nowFn().Sub(c.LastRead()), true
+}
+
+// StaleFor reports whether the socket has wedged: it was being drained, then
+// went unread for longer than d. A socket that has never been read is never
+// stale — a wedge can only arise after draining has started, so requiring a
+// prior read rules out the false positive (a healthy node that has simply had
+// no inbound traffic yet) without missing any real wedge.
+func (c *TimedConn) StaleFor(d time.Duration) bool {
+	age, ok := c.ReadStaleness()
+	return ok && age > d
+}

network/discovery/timed_conn_test.go

@@ -0,0 +1,110 @@
+package discovery
+
+import (
+	"errors"
+	"net"
+	"net/netip"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+// fakeUDPConn is a minimal discover.UDPConn so the read path can be exercised
+// without a real socket.
+type fakeUDPConn struct {
+	readErr error
+}
+
+func (f *fakeUDPConn) ReadFromUDPAddrPort(b []byte) (int, netip.AddrPort, error) {
+	if f.readErr != nil {
+		return 0, netip.AddrPort{}, f.readErr
+	}
+	n := copy(b, []byte{0x1})
+	return n, netip.MustParseAddrPort("1.2.3.4:30303"), nil
+}
+
+func (f *fakeUDPConn) WriteToUDPAddrPort(b []byte, _ netip.AddrPort) (int, error) {
+	return len(b), nil
+}
+
+func (f *fakeUDPConn) Close() error        { return nil }
+func (f *fakeUDPConn) LocalAddr() net.Addr { return nil }
+
+// TestTimedConn_SeededNotStale: a freshly constructed conn is not stale — it has
+// not been read yet, so there is nothing to flag at startup.
+func TestTimedConn_SeededNotStale(t *testing.T) {
+	c := NewTimedConn(nil)
+	require.False(t, c.StaleFor(3*time.Minute))
+}
+
+// TestTimedConn_NeverReadNeverStale: a socket that has never delivered a packet
+// is not a wedge — it's a node with no inbound discv5 traffic (unreachable
+// bootnodes, blocked UDP), which a restart cannot fix — so it must never report
+// stale, however much time passes. Once it has been read, staleness arms as
+// usual.
+func TestTimedConn_NeverReadNeverStale(t *testing.T) {
+	c := NewTimedConn(nil)
+	t0 := c.LastRead()
+
+	c.now = func() time.Time { return t0.Add(time.Hour) }
+	require.False(t, c.StaleFor(3*time.Minute), "never-read socket must not be stale")
+	_, ok := c.ReadStaleness()
+	require.False(t, ok, "never-read socket has no staleness to report")
+
+	// A successful read arms the signal; from the read time on, it goes stale.
+	c.UDPConn = &fakeUDPConn{}
+	_, _, err := c.ReadFromUDPAddrPort(make([]byte, 16))
+	require.NoError(t, err)
+	age, ok := c.ReadStaleness()
+	require.True(t, ok, "a read arms the signal")
+	require.Zero(t, age, "staleness resets to the read time")
+
+	c.now = func() time.Time { return t0.Add(time.Hour + 4*time.Minute) }
+	require.True(t, c.StaleFor(3*time.Minute), "past grace after a read")
+}
+
+// TestTimedConn_StaleForTransitions pins the staleness boundary using an
+// injected clock, so no real time passes.
+func TestTimedConn_StaleForTransitions(t *testing.T) {
+	c := NewTimedConn(nil)
+	c.read.Store(true) // staleness only arms once the socket has been read
+	t0 := c.LastRead()
+
+	c.now = func() time.Time { return t0.Add(2 * time.Minute) }
+	require.False(t, c.StaleFor(3*time.Minute), "within grace")
+
+	c.now = func() time.Time { return t0.Add(3 * time.Minute) }
+	require.False(t, c.StaleFor(3*time.Minute), "exactly at grace is not yet stale")
+
+	c.now = func() time.Time { return t0.Add(4 * time.Minute) }
+	require.True(t, c.StaleFor(3*time.Minute), "past grace")
+}
+
+// TestTimedConn_ReadStampsLastRead: a successful read advances the timestamp,
+// so an actively drained socket stays fresh.
+func TestTimedConn_ReadStampsLastRead(t *testing.T) {
+	c := &TimedConn{UDPConn: &fakeUDPConn{}}
+	t0 := time.Now()
+	c.lastReadUnixNano.Store(t0.UnixNano())
+	c.now = func() time.Time { return t0.Add(time.Hour) }
+
+	n, addr, err := c.ReadFromUDPAddrPort(make([]byte, 16))
+	require.NoError(t, err)
+	require.Equal(t, 1, n)
+	require.Equal(t, netip.MustParseAddrPort("1.2.3.4:30303"), addr)
+	require.Equal(t, t0.Add(time.Hour).UnixNano(), c.LastRead().UnixNano())
+}
+
+// TestTimedConn_FailedReadDoesNotStamp: a read error must leave the timestamp
+// untouched, so a wedged/closed socket cannot look freshly drained.
+func TestTimedConn_FailedReadDoesNotStamp(t *testing.T) {
+	c := &TimedConn{UDPConn: &fakeUDPConn{readErr: errors.New("boom")}}
+	t0 := time.Now()
+	c.lastReadUnixNano.Store(t0.UnixNano())
+	c.now = func() time.Time { return t0.Add(time.Hour) }
+
+	_, _, err := c.ReadFromUDPAddrPort(make([]byte, 16))
+	require.Error(t, err)
+	require.Equal(t, t0.UnixNano(), c.LastRead().UnixNano(), "errored read must not stamp")
+}