device/telemetry: keep probing when the epoch fetch fails

CHANGELOG.md

@@ -8,6 +8,11 @@ All notable changes to this project will be documented in this file.
 
 ### Changes
 
+- Telemetry
+  - A ledger RPC outage no longer stops TWAMP probing on the device telemetry agent: the pinger caches the last known epoch and refreshes it off the probe path, instead of fetching it inline and skipping the tick on failure. Probing stops only when no epoch has ever been fetched or the cached one exceeds the new `-max-epoch-staleness` (default 12h). (#4143)
+- Device telemetry
+  - A peer discovery refresh that fails after reading the ledger no longer wipes the agent's peer list. It cleared the cache before calling `LocalNet.Interfaces()`, so a transient failure there left the pinger iterating zero peers and probing nothing until a later refresh succeeded. The cache is now replaced only once the new list is built, which also shortens the critical section to the assignment. (#4146)
+
 ## [v0.33.0](https://github.com/malbeclabs/doublezero/compare/client/v0.32.0...client/v0.33.0) - 2026-07-31
 
 ### Breaking

controlplane/telemetry/cmd/telemetry/main.go

@@ -55,6 +55,7 @@ const (
 	defaultBGPStatusInterval          = 60 * time.Second
 	defaultBGPStatusRefreshInterval   = 6 * time.Hour
 	defaultCachingFetcherRPCTimeout   = 30 * time.Second
+	defaultMaxEpochStaleness          = telemetry.DefaultMaxEpochStaleness
 
 	waitForNamespaceTimeout             = 30 * time.Second
 	defaultStateIngestHTTPClientTimeout = 10 * time.Second
@@ -96,6 +97,7 @@ var (
 
 	// caching fetcher flags
 	cachingFetcherRPCTimeout = flag.Duration("caching-fetcher-rpc-timeout", defaultCachingFetcherRPCTimeout, "Timeout for GetProgramData RPC calls inside the caching fetcher.")
+	maxEpochStaleness        = flag.Duration("max-epoch-staleness", defaultMaxEpochStaleness, "How long to keep probing with the last known epoch while the ledger rpc is unreachable, before giving up.")
 
 	// bgp status submitter flags
 	bgpStatusEnable          = flag.Bool("bgp-status-enable", false, "Enable onchain BGP status submission after each collection tick.")
@@ -346,6 +348,7 @@ func main() {
 		SenderTTL:                  *senderTTL,
 		SubmitterMaxConcurrency:    *submitterMaxConcurrency,
 		MaxConsecutiveSenderLosses: *maxConsecutiveSenderLosses,
+		MaxEpochStaleness:          *maxEpochStaleness,
 		GeolocationClient:          geolocationClient,
 		AgentVersion:               version,
 		AgentCommit:                commit,

controlplane/telemetry/internal/metrics/metrics.go

@@ -10,6 +10,7 @@ const (
 	MetricNameBuildInfo                        = "doublezero_device_telemetry_agent_build_info"
 	MetricNameErrors                           = "doublezero_device_telemetry_agent_errors_total"
 	MetricNamePeerDiscoveryLocalTunnelNotFound = "doublezero_device_telemetry_agent_peer_discovery_not_found_tunnels"
+	MetricNameEpochCacheStaleAge               = "doublezero_device_telemetry_agent_epoch_cache_stale_age_seconds"
 
 	// Labels.
 	LabelVersion       = "version"
@@ -27,6 +28,7 @@ const (
 	ErrorTypeSubmitterFailedToInitializeAccount  = "submitter_failed_to_initialize_account"
 	ErrorTypeSubmitterFailedToWriteSamples       = "submitter_failed_to_write_samples"
 	ErrorTypeSubmitterRetriesExhausted           = "submitter_retries_exhausted"
+	ErrorTypePingerEpochUnavailable              = "pinger_epoch_unavailable"
 )
 
 var (
@@ -46,6 +48,15 @@ var (
 		[]string{LabelErrorType},
 	)
 
+	// EpochCacheStaleAge is the age of the cached epoch the probe loop is falling back to while the
+	// epoch fetch is failing, and 0 whenever the cache is fresh.
+	EpochCacheStaleAge = promauto.NewGauge(
+		prometheus.GaugeOpts{
+			Name: MetricNameEpochCacheStaleAge,
+			Help: "Age of the cached ledger epoch served to the probe loop when the epoch fetch is failing (0 when fresh)",
+		},
+	)
+
 	PeerDiscoveryLocalTunnelNotFound = promauto.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Name: MetricNamePeerDiscoveryLocalTunnelNotFound,

controlplane/telemetry/internal/telemetry/collector.go

@@ -80,6 +80,8 @@ func New(log *slog.Logger, cfg Config) (*Collector, error) {
 		GetSender:         c.getOrCreateSender,
 		GetCurrentEpoch:   cfg.GetCurrentEpochFunc,
 		RecordProbeResult: c.recordProbeResult,
+		MaxEpochStaleness: cfg.MaxEpochStaleness,
+		NowFunc:           cfg.NowFunc,
 	})
 
 	// Initialize geoprobe coordinator if onchain discovery is configured.

controlplane/telemetry/internal/telemetry/config.go

@@ -52,6 +52,10 @@ type Config struct {
 	// before a sender is evicted from the cache and recreated.
 	MaxConsecutiveSenderLosses int
 
+	// MaxEpochStaleness is how long the probe loop keeps probing with the last known epoch while
+	// the ledger RPC is unreachable. Defaults to DefaultMaxEpochStaleness.
+	MaxEpochStaleness time.Duration
+
 	// ServiceabilityProgramClient is the client to the serviceability program (for fetching Device/Location).
 	ServiceabilityProgramClient ServiceabilityProgramClient
 
@@ -115,6 +119,9 @@ func (c *Config) Validate() error {
 	if c.MaxConsecutiveSenderLosses <= 0 {
 		c.MaxConsecutiveSenderLosses = 30
 	}
+	if c.MaxEpochStaleness <= 0 {
+		c.MaxEpochStaleness = DefaultMaxEpochStaleness
+	}
 
 	geoprobeEnabled := c.GeolocationClient != nil
 	if geoprobeEnabled {

controlplane/telemetry/internal/telemetry/peers.go

@@ -121,11 +121,9 @@ func (p *ledgerPeerDiscovery) refresh(ctx context.Context) error {
 		return fmt.Errorf("failed to load program from ledger: %w", err)
 	}
 
-	p.peersMu.Lock()
-	defer p.peersMu.Unlock()
-
-	p.peers = make([]*Peer, 0, len(p.peers))
-
+	// The cache is left in place while the new peer list is built, and replaced only once the build
+	// has succeeded. Nothing below this point may clear it: a refresh that fails partway must leave
+	// the agent probing the peers it already knows about rather than none at all.
 	devices := make(map[string]serviceability.Device)
 	for _, device := range data.Devices {
 		pubkey := solana.PublicKeyFromBytes(device.PubKey[:])
@@ -206,7 +204,10 @@ func (p *ledgerPeerDiscovery) refresh(ctx context.Context) error {
 		})
 	}
 
+	p.peersMu.Lock()
 	p.peers = peers
+	p.peersMu.Unlock()
+
 	p.log.Debug("Refreshed peers", "devices", len(devices), "links", len(links), "peers", len(peers), "tunnelsNotFound", tunnelsNotFound)
 
 	// Record the number of tunnels not found.

controlplane/telemetry/internal/telemetry/peers_test.go

@@ -2,9 +2,11 @@ package telemetry_test
 
 import (
 	"context"
+	"errors"
 	"log/slog"
 	"net"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -617,6 +619,83 @@ func TestAgentTelemetry_PeerDiscovery_Ledger(t *testing.T) {
 		cfg.RefreshInterval = 0
 		base(cfg, "zero refresh interval")
 	})
+
+	t.Run("keeps known peers when getting local interfaces fails", func(t *testing.T) {
+		t.Parallel()
+
+		log := log.With("test", t.Name())
+		localDevicePK := stringToPubkey("device1")
+
+		serviceabilityProgram := &mockServiceabilityProgramClient{
+			GetProgramDataFunc: func(ctx context.Context) (*serviceability.ProgramData, error) {
+				return &serviceability.ProgramData{
+					Devices: []serviceability.Device{
+						{PubKey: localDevicePK, PublicIp: [4]uint8{192, 168, 1, 1}},
+						{PubKey: stringToPubkey("device2"), PublicIp: [4]uint8{192, 168, 1, 2}},
+					},
+					Links: []serviceability.Link{
+						{PubKey: stringToPubkey("link_1-2"), Status: serviceability.LinkStatusActivated, SideAPubKey: localDevicePK, SideZPubKey: stringToPubkey("device2"), TunnelNet: [5]uint8{10, 1, 1, 0, 31}},
+					},
+				}, nil
+			},
+		}
+
+		// The first refresh discovers the peer; every refresh after it fails on local interfaces.
+		var interfaceCalls atomic.Int32
+
+		cfg := &telemetry.LedgerPeerDiscoveryConfig{
+			Logger:        log,
+			LocalDevicePK: localDevicePK,
+			ProgramClient: serviceabilityProgram,
+			LocalNet: &netutil.MockLocalNet{
+				InterfacesFunc: func() ([]netutil.Interface, error) {
+					if interfaceCalls.Add(1) > 1 {
+						return nil, errors.New("transient failure getting local interfaces")
+					}
+					return []netutil.Interface{
+						{Name: "tun1-2", Addrs: []net.Addr{&net.IPNet{IP: ipv4([4]uint8{10, 1, 1, 0}), Mask: net.CIDRMask(31, 32)}}},
+					}, nil
+				},
+			},
+			TWAMPPort:       1234,
+			RefreshInterval: 20 * time.Millisecond,
+		}
+
+		peerDiscovery, err := telemetry.NewLedgerPeerDiscovery(cfg)
+		require.NoError(t, err)
+
+		ctx, cancel := context.WithCancel(t.Context())
+		errCh := make(chan error, 1)
+		go func() {
+			errCh <- peerDiscovery.Run(ctx)
+		}()
+
+		expected := []*telemetry.Peer{
+			{
+				LinkPK:   stringToPubkey("link_1-2"),
+				DevicePK: stringToPubkey("device2"),
+				Tunnel: &netutil.LocalTunnel{
+					Interface: "tun1-2",
+					SourceIP:  ipv4([4]uint8{10, 1, 1, 0}),
+					TargetIP:  ipv4([4]uint8{10, 1, 1, 1}),
+				},
+				TWAMPPort: 1234,
+			},
+		}
+
+		require.Eventually(t, func() bool {
+			return len(peerDiscovery.GetPeers()) == 1
+		}, 2*time.Second, 20*time.Millisecond, "first refresh should discover the peer")
+
+		require.Eventually(t, func() bool {
+			return interfaceCalls.Load() >= 4
+		}, 2*time.Second, 20*time.Millisecond, "later refreshes should keep failing on local interfaces")
+
+		assert.Equal(t, expected, peerDiscovery.GetPeers(), "peers should survive a refresh that fails after the ledger read")
+
+		cancel()
+		assert.NoError(t, <-errCh)
+	})
 }
 
 func ipv4(bytes [4]uint8) net.IP {