fix(daemon/shaper): wake the statusz poll loop on endpoint discovery

docs/dev/daemon/traffic-shaper-statusz.md

@@ -154,11 +154,22 @@ until a ready BN pod is observed the poll loop idles quietly and starts polling
 as soon as one appears.
 
 **Using pod discovery without `base_url`:** after a daemon restart or host
-reboot, the entry reconcile fires immediately but skips if no URL has been
-discovered yet. Convergence then waits until the pod watcher observes a ready
-BN pod, after which the next ticker tick (up to one poll interval) triggers
-the reconcile. For the lowest post-reboot convergence latency, set `base_url`
-to a stable statusz endpoint (e.g. a node-local port-forward).
+reboot, the entry reconcile fires immediately and skips if no URL has been
+discovered yet — the pod watcher runs concurrently, so on a cold start it has
+usually not recorded an endpoint yet. Convergence is **not** deferred to the next
+tick in that case: the watcher signals the poll loop as soon as it records an
+endpoint, and the loop wakes on that signal as well as on the ticker. Both paths
+therefore converge as fast as the BN statusz endpoint responds — bounded by BN
+startup time, not by `poll_interval`.
+
+The same signal fires when a rescheduled pod changes the endpoint, and when the
+owning pod is deleted (the loop wakes, observes an empty URL, and logs the loss
+promptly instead of up to one interval later).
+
+> Before #1000 convergence on this path was bounded by *pod discovery plus up to
+> one poll interval*, because the ticker was the loop's only wake-up source.
+> Setting `base_url` was the documented workaround; it is no longer needed to get
+> prompt convergence.
 
 ### Enablement at install time
 

docs/dev/traffic-shaper.md

@@ -279,7 +279,7 @@ fill in the parts that are deliberately **not** persisted (see below).
 | Artifact | Persisted at boot? | Rebuilt by |
 |---|---|---|
 | nft tables, chains, rules (both tables) | Yes — replayed from the `.nft` files via `nft -f` | — |
-| nft **set elements** (the CIDR membership of `bn-*` sets) | **No** | daemon statusz poll loop — entry reconcile fires immediately on daemon start; bounded by BN startup time when `base_url` is set, or by pod readiness + up to one poll interval with pod discovery |
+| nft **set elements** (the CIDR membership of `bn-*` sets) | **No** | daemon statusz poll loop — entry reconcile fires immediately on daemon start, and the pod watcher wakes the loop the moment it discovers the endpoint, so convergence is bounded by BN startup time on both the `base_url` and pod-discovery paths |
 | `$EGRESS` HTB hierarchy | Yes — the `solo-provisioner-bandwidth-shaper.sh` script | — |
 | `$VETH` (per-pod) HTB hierarchy | **No** | daemon pod-lifecycle watcher, on the next pod-create event |
 

internal/daemon/blocknode/pod_watcher.go

@@ -222,13 +222,23 @@ func (m *TrafficShaperMonitor) handlePodDelete(ctx context.Context, pod *corev1.
 	// Drop the discovered statusz endpoint only if this is the pod that set it, so
 	// a stale delete for some other pod cannot blank a still-valid endpoint. The
 	// poll loop then idles until another ready BN pod is observed.
-	if m.discoveredStatuszPod == pod.UID {
+	endpointCleared := m.discoveredStatuszPod == pod.UID
+	if endpointCleared {
 		m.discoveredStatuszURL = ""
 		m.discoveredStatuszPod = ""
 	}
 	veth, ok := m.attached[pod.UID]
 	delete(m.attached, pod.UID)
 	m.mu.Unlock()
+
+	// Wake the poll loop so it logs the endpoint loss promptly instead of up to a
+	// full interval later. The woken reconcile is a no-op while the URL is empty.
+	// Signalled after the unlock, and before the early return below so a pod with
+	// no recorded veth still reports the loss.
+	if endpointCleared {
+		m.signalURLChanged()
+	}
+
 	if !ok {
 		return
 	}
@@ -282,6 +292,9 @@ func (m *TrafficShaperMonitor) recordDiscoveredStatusz(pod *corev1.Pod) {
 			Str("pod", pod.Namespace+"/"+pod.Name).
 			Str("statusz_url", url).
 			Msg("discovered BN statusz endpoint from pod")
+		// Wake the poll loop so membership converges now instead of at the next
+		// tick. Sent after the unlock above — never while holding m.mu.
+		m.signalURLChanged()
 	}
 }
 

internal/daemon/blocknode/pod_watcher_test.go

@@ -65,10 +65,11 @@ func (f *fakeDelegator) ReconcileShaperCheck(context.Context, string) (string, e
 
 func newTestMonitor(r vethResolver, d *fakeDelegator) *TrafficShaperMonitor {
 	return &TrafficShaperMonitor{
-		resolver:  r,
-		delegator: d,
-		attached:  make(map[types.UID]string),
-		inflight:  make(map[types.UID]bool),
+		resolver:   r,
+		delegator:  d,
+		attached:   make(map[types.UID]string),
+		inflight:   make(map[types.UID]bool),
+		urlChanged: make(chan struct{}, 1),
 	}
 }
 
@@ -305,6 +306,56 @@ func TestHandlePodDelete_ClearsDiscoveredStatuszForOwningPod(t *testing.T) {
 	require.Equal(t, types.UID(""), m.discoveredStatuszPod)
 }
 
+// signalled reports whether a wake-up is pending, consuming it so successive
+// assertions in one test observe only new signals.
+func signalled(m *TrafficShaperMonitor) bool {
+	select {
+	case <-m.urlChanged:
+		return true
+	default:
+		return false
+	}
+}
+
+// TestRecordDiscoveredStatusz_SignalsOnlyOnChange verifies the poll loop is woken
+// when the endpoint actually changes, and is left alone when a repeat event
+// records the same URL — otherwise every watch event would cost a probe (#1000).
+func TestRecordDiscoveredStatusz_SignalsOnlyOnChange(t *testing.T) {
+	m := newTestMonitor(&fakeResolver{results: []resolveResult{{veth: "lxc1"}}}, &fakeDelegator{})
+	pod := readyPodWithNet("u1", "bn-0", "10.1.2.3",
+		corev1.ContainerPort{Name: bnHealthPortName, ContainerPort: 40983})
+
+	m.recordDiscoveredStatusz(pod)
+	require.True(t, signalled(m), "first discovery wakes the poll loop")
+
+	m.recordDiscoveredStatusz(pod)
+	require.False(t, signalled(m), "an unchanged endpoint must not wake the loop")
+
+	// A rescheduled pod on a new IP is a change and must wake the loop so the
+	// roster is re-read against the new endpoint.
+	moved := readyPodWithNet("u2", "bn-0", "10.1.2.9",
+		corev1.ContainerPort{Name: bnHealthPortName, ContainerPort: 40983})
+	m.recordDiscoveredStatusz(moved)
+	require.True(t, signalled(m), "a changed endpoint wakes the poll loop")
+}
+
+// TestHandlePodDelete_SignalsWhenEndpointCleared verifies losing the endpoint also
+// wakes the loop, so the endpoint-lost transition is logged promptly rather than
+// up to a full poll interval later. An unrelated pod delete must stay silent.
+func TestHandlePodDelete_SignalsWhenEndpointCleared(t *testing.T) {
+	m := newTestMonitor(&fakeResolver{results: []resolveResult{{veth: "lxc1"}}}, &fakeDelegator{})
+	pod := readyPodWithNet("u1", "bn-0", "10.1.2.3",
+		corev1.ContainerPort{Name: bnHealthPortName, ContainerPort: 40983})
+	m.recordDiscoveredStatusz(pod)
+	require.True(t, signalled(m), "drain the discovery signal")
+
+	m.handlePodDelete(context.Background(), readyPod("u2", "bn-1"))
+	require.False(t, signalled(m), "an unrelated pod delete must not wake the loop")
+
+	m.handlePodDelete(context.Background(), pod)
+	require.True(t, signalled(m), "clearing the endpoint wakes the loop")
+}
+
 func TestHandlePodDelete_KeepsDiscoveredStatuszForOtherPod(t *testing.T) {
 	m := newTestMonitor(&fakeResolver{results: []resolveResult{{veth: "lxc1"}}}, &fakeDelegator{})
 	pod := readyPodWithNet("u1", "bn-0", "10.1.2.3",

internal/daemon/blocknode/traffic_shaper_monitor.go

@@ -89,6 +89,21 @@ type TrafficShaperMonitor struct {
 	// a delete only clears the endpoint when it is that pod (not some other pod in
 	// a multi-pod window) that went away.
 	discoveredStatuszPod types.UID
+
+	// urlChanged wakes the statusz poll loop when the pod watcher discovers,
+	// changes, or loses the statusz endpoint, so convergence is not deferred to
+	// the next ticker tick. The two responsibilities start concurrently, so the
+	// entry reconcile usually runs before the watcher's initial list has recorded
+	// an endpoint; without this signal the loop would sleep a full poll interval
+	// with a ready pod and a reachable statusz (#1000).
+	//
+	// Capacity 1 with a non-blocking send is load-bearing: it makes the signal
+	// order-independent. A send that lands before the loop reaches its select is
+	// retained in the buffer and returns immediately on arrival, whereas an
+	// unbuffered send would find no receiver, drop, and reintroduce the race it
+	// exists to close. Extra wake-ups are harmless — the digest gate skips the
+	// privileged apply when the desired state has not changed.
+	urlChanged chan struct{}
 }
 
 // NewTrafficShaperMonitor constructs a TrafficShaperMonitor. resolver and client
@@ -107,6 +122,27 @@ func NewTrafficShaperMonitor(resolver *VethResolver, client kubernetes.Interface
 		pollInterval: pollInterval,
 		attached:     make(map[types.UID]string),
 		inflight:     make(map[types.UID]bool),
+		urlChanged:   make(chan struct{}, 1),
+	}
+}
+
+// signalURLChanged wakes the statusz poll loop after the discovered endpoint
+// changed. The send is non-blocking: when a signal is already buffered the loop
+// has not consumed the previous one yet and will observe the latest URL when it
+// wakes, so coalescing loses nothing. Safe to call with no loop running (e.g.
+// while runStatuszPoll is restarting after a fault) — the buffered signal is
+// consumed by the next loop, costing one extra reconcile, which is idempotent.
+//
+// Never call this while holding m.mu: the send must not be able to interleave
+// with a reader of the discovered-statusz fields.
+func (m *TrafficShaperMonitor) signalURLChanged() {
+	if m.urlChanged == nil {
+		// Zero-value monitor (unit-test scaffolding); nothing to wake.
+		return
+	}
+	select {
+	case m.urlChanged <- struct{}{}:
+	default:
 	}
 }
 
@@ -335,11 +371,12 @@ func (m *TrafficShaperMonitor) runStatuszPoll(ctx context.Context) error {
 	// error and superviseResponsibility retries with back-off — convergence is
 	// bounded by BN startup time, not the poll interval.
 	//
-	// Pod-discovery caveat: when base_url is not configured, effectiveStatuszURL
-	// returns "" until the pod watcher observes a ready BN pod. In that case this
-	// entry reconcile is a silent no-op (not an error), and convergence after a
-	// restart waits for pod discovery plus up to one poll interval for the next
-	// tick. Set base_url for guaranteed immediate convergence after a reboot.
+	// Pod-discovery path: when base_url is not configured, effectiveStatuszURL
+	// returns "" until the pod watcher observes a ready BN pod, so this entry
+	// reconcile is a silent no-op (not an error) on a cold start. Convergence is
+	// not deferred to the next tick in that case — the watcher signals urlChanged
+	// as soon as it records an endpoint and the select below wakes on it, so both
+	// the discovery and base_url paths converge as fast as statusz responds.
 	if err := runReconcile(); err != nil {
 		return err
 	}
@@ -352,6 +389,13 @@ func (m *TrafficShaperMonitor) runStatuszPoll(ctx context.Context) error {
 			if err := runReconcile(); err != nil {
 				return err
 			}
+		case <-m.urlChanged:
+			// The pod watcher discovered, changed, or lost the endpoint. Reconcile
+			// now rather than waiting out the interval. reconcile() no-ops when the
+			// URL is empty (endpoint lost), so a teardown signal costs nothing.
+			if err := runReconcile(); err != nil {
+				return err
+			}
 		}
 	}
 }

internal/daemon/blocknode/traffic_shaper_monitor_test.go

@@ -78,6 +78,7 @@ func newPollMonitor(d *pollFakeDelegator, statuszURL string, interval time.Durat
 		delegator:    d,
 		statuszURL:   statuszURL,
 		pollInterval: interval,
+		urlChanged:   make(chan struct{}, 1),
 	}
 }
 
@@ -142,6 +143,90 @@ func TestRunStatuszPoll_ReconcilesOnceDiscovered(t *testing.T) {
 	d.mu.Unlock()
 }
 
+// TestSignalURLChanged_RetainsSignalForLateReceiver pins the buffering guarantee
+// that closes #1000. The pod watcher and the poll loop start concurrently, so
+// discovery routinely signals before the loop reaches its select. With capacity 1
+// the signal waits in the buffer; an unbuffered channel would drop it for want of
+// a receiver and the loop would sleep a full poll interval with a ready pod.
+func TestSignalURLChanged_RetainsSignalForLateReceiver(t *testing.T) {
+	m := &TrafficShaperMonitor{urlChanged: make(chan struct{}, 1)}
+
+	m.signalURLChanged() // no receiver is waiting yet
+
+	select {
+	case <-m.urlChanged:
+	default:
+		t.Fatal("signal dropped with no receiver waiting — the poll loop would sleep a full interval (#1000)")
+	}
+}
+
+// TestSignalURLChanged_CoalescesAndNeverBlocks verifies repeated signals with no
+// reader neither block nor queue: extra wake-ups are redundant because the loop
+// re-reads the current URL when it wakes. Also covers the zero-value monitor,
+// where the channel is nil (unit-test scaffolding constructs monitors directly).
+func TestSignalURLChanged_CoalescesAndNeverBlocks(t *testing.T) {
+	m := &TrafficShaperMonitor{urlChanged: make(chan struct{}, 1)}
+	for range 100 {
+		m.signalURLChanged() // must not block once the buffer is full
+	}
+	require.Len(t, m.urlChanged, 1, "signals coalesce into a single pending wake-up")
+
+	require.NotPanics(t, (&TrafficShaperMonitor{}).signalURLChanged,
+		"a monitor with no channel wired must be safe to signal")
+}
+
+// TestRunStatuszPoll_DiscoverySignalWakesLoopBeforeNextTick is the regression test
+// for #1000. The poll interval is an hour, so the ticker cannot account for any
+// reconcile: the only way the loop can converge is by waking on the discovery
+// signal. Before the fix this test would hang until the deadline.
+func TestRunStatuszPoll_DiscoverySignalWakesLoopBeforeNextTick(t *testing.T) {
+	d := &pollFakeDelegator{digests: []string{"D1"}}
+	m := newPollMonitor(d, "", time.Hour) // no base_url override; tick is unreachable
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	done := make(chan error, 1)
+	go func() { done <- m.runStatuszPoll(ctx) }()
+
+	// The entry reconcile runs with no endpoint and must stay quiet.
+	time.Sleep(20 * time.Millisecond)
+	require.Zero(t, d.checkCalls.Load(), "no exec while the endpoint is undiscovered")
+
+	// The pod watcher records an endpoint and signals, as recordDiscoveredStatusz does.
+	m.mu.Lock()
+	m.discoveredStatuszURL = "http://10.1.2.3:40983"
+	m.mu.Unlock()
+	m.signalURLChanged()
+
+	waitForCount(t, d.applyCalls.Load, 1)
+	cancel()
+	require.NoError(t, <-done)
+
+	d.mu.Lock()
+	require.Equal(t, "http://10.1.2.3:40983", d.lastURL, "reconciled against the newly discovered URL")
+	d.mu.Unlock()
+}
+
+// TestRunStatuszPoll_EndpointLossSignalIsHarmless verifies a teardown signal (the
+// owning pod went away, so the URL is now empty) wakes the loop without exec'ing
+// anything — it only lets the endpoint-lost transition be logged promptly.
+func TestRunStatuszPoll_EndpointLossSignalIsHarmless(t *testing.T) {
+	d := &pollFakeDelegator{digests: []string{"D1"}}
+	m := newPollMonitor(d, "", time.Hour)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	done := make(chan error, 1)
+	go func() { done <- m.runStatuszPoll(ctx) }()
+
+	m.signalURLChanged() // endpoint still empty, as after a pod delete
+	time.Sleep(20 * time.Millisecond)
+	require.Zero(t, d.checkCalls.Load(), "a wake-up with no endpoint must not exec")
+
+	cancel()
+	require.NoError(t, <-done)
+}
+
 // TestRunStatuszPoll_InertWhenUnset verifies that with no statusz base_url the
 // poll loop touches no delegator path and returns nil on ctx cancel.
 func TestRunStatuszPoll_InertWhenUnset(t *testing.T) {