device/telemetry: keep probing when the epoch fetch fails

CHANGELOG.md

@@ -8,6 +8,9 @@ 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). ([#4125](https://github.com/malbeclabs/doublezero/issues/4125))
+
 ## [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/pinger.go

@@ -9,10 +9,22 @@ import (
 
 	"github.com/cenkalti/backoff/v5"
 	"github.com/gagliardetto/solana-go"
+	"github.com/malbeclabs/doublezero/controlplane/telemetry/internal/metrics"
 	"github.com/malbeclabs/doublezero/controlplane/telemetry/pkg/buffer"
 	twamplight "github.com/malbeclabs/doublezero/tools/twamp/pkg/light"
 )
 
+const (
+	// DefaultMaxEpochStaleness bounds how long the pinger keeps probing with a cached epoch after
+	// the ledger RPC stops answering. Samples recorded with a stale epoch are written to that
+	// epoch's account, so once we are far enough behind that a rollover is likely they would be
+	// misattributed to the previous epoch.
+	DefaultMaxEpochStaleness = 12 * time.Hour
+
+	// defaultEpochRefreshInterval is used when neither EpochRefreshInterval nor Interval is set.
+	defaultEpochRefreshInterval = 10 * time.Second
+)
+
 type PingerConfig struct {
 	LocalDevicePK     solana.PublicKey
 	Interval          time.Duration
@@ -22,6 +34,18 @@ type PingerConfig struct {
 	GetSender         func(ctx context.Context, peer *Peer) twamplight.Sender
 	GetCurrentEpoch   func(ctx context.Context) (uint64, error)
 	RecordProbeResult func(peer *Peer, success bool)
+
+	// EpochRefreshInterval is how often the cached epoch is refreshed in the background.
+	// Defaults to Interval, which keeps the epoch RPC rate the same as when the fetch was inline.
+	EpochRefreshInterval time.Duration
+
+	// MaxEpochStaleness is how long a cached epoch is trusted after the last successful fetch.
+	// Defaults to DefaultMaxEpochStaleness.
+	MaxEpochStaleness time.Duration
+
+	// NowFunc is the function used to measure the age of the cached epoch.
+	// Defaults to time.Now().UTC.
+	NowFunc func() time.Time
 }
 
 // Pinger is responsible for periodically probing remote peers using TWAMP.
@@ -30,15 +54,44 @@ type PingerConfig struct {
 type Pinger struct {
 	log *slog.Logger
 	cfg *PingerConfig
+
+	// The epoch is only used to build the sample buffer's partition key, so probing itself needs
+	// no ledger access. It is refreshed on its own loop and cached here, and the probe path reads
+	// the cache rather than the ledger: a total RPC outage costs us epoch precision, not
+	// measurements.
+	mu             sync.Mutex
+	epoch          uint64
+	haveEpoch      bool
+	epochAt        time.Time
+	servingStale   bool
+	warnedNoEpoch  bool
+	warnedTooStale bool
 }
 
 func NewPinger(log *slog.Logger, cfg *PingerConfig) *Pinger {
+	if cfg.EpochRefreshInterval <= 0 {
+		cfg.EpochRefreshInterval = cfg.Interval
+	}
+	if cfg.EpochRefreshInterval <= 0 {
+		cfg.EpochRefreshInterval = defaultEpochRefreshInterval
+	}
+	if cfg.MaxEpochStaleness <= 0 {
+		cfg.MaxEpochStaleness = DefaultMaxEpochStaleness
+	}
+	if cfg.NowFunc == nil {
+		cfg.NowFunc = func() time.Time { return time.Now().UTC() }
+	}
 	return &Pinger{log: log, cfg: cfg}
 }
 
 func (p *Pinger) Run(ctx context.Context) error {
 	p.log.Info("Starting probe loop")
 
+	// Refresh the epoch on its own loop so an unreachable ledger RPC cannot stall probing. A
+	// failing fetch burns ~130s across its retries and the probe ticker only buffers one tick, so
+	// fetching inline dropped roughly a dozen probe opportunities per failure.
+	go p.refreshEpochLoop(ctx)
+
 	ticker := time.NewTicker(p.cfg.Interval)
 	defer ticker.Stop()
 
@@ -54,9 +107,8 @@ func (p *Pinger) Run(ctx context.Context) error {
 }
 
 func (p *Pinger) Tick(ctx context.Context) {
-	epoch, err := p.getCurrentEpoch(ctx)
-	if err != nil {
-		p.log.Error("failed to get current epoch", "error", err)
+	epoch, ok := p.epochForTick(ctx)
+	if !ok {
 		return
 	}
 
@@ -143,14 +195,155 @@ func (p *Pinger) Tick(ctx context.Context) {
 	wg.Wait()
 }
 
+// epochForTick returns the epoch to stamp this tick's samples with, and whether to probe at all.
+// It reads the cached epoch and does not contact the ledger, except on the first call before the
+// refresh loop has landed a value.
+//
+// Probing is refused only when no epoch has ever been fetched, or when the cached one is older
+// than MaxEpochStaleness. Both cases log once rather than per tick.
+func (p *Pinger) epochForTick(ctx context.Context) (uint64, bool) {
+	p.mu.Lock()
+	epoch, have, at := p.epoch, p.haveEpoch, p.epochAt
+	p.mu.Unlock()
+
+	if !have {
+		// Nothing cached: either the agent just started and the refresh loop has not produced a
+		// value yet, or Tick is being driven directly. Fetch inline so the tick is not wasted.
+		epoch, err := p.getCurrentEpoch(ctx)
+		if err != nil {
+			metrics.Errors.WithLabelValues(metrics.ErrorTypePingerEpochUnavailable).Inc()
+
+			p.mu.Lock()
+			first := !p.warnedNoEpoch
+			p.warnedNoEpoch = true
+			p.mu.Unlock()
+
+			if first {
+				p.log.Error("No epoch available and none cached, skipping probes until the ledger answers", "error", err)
+			} else {
+				p.log.Debug("No epoch available and none cached, skipping probe tick", "error", err)
+			}
+			return 0, false
+		}
+		p.storeEpoch(epoch)
+		return epoch, true
+	}
+
+	age := p.cfg.NowFunc().Sub(at)
+	if age > p.cfg.MaxEpochStaleness {
+		metrics.Errors.WithLabelValues(metrics.ErrorTypePingerEpochUnavailable).Inc()
+
+		p.mu.Lock()
+		first := !p.warnedTooStale
+		p.warnedTooStale = true
+		p.mu.Unlock()
+
+		if first {
+			p.log.Warn("Cached epoch is too stale to probe with, skipping probes until the ledger answers", "epoch", epoch, "age", age, "maxStaleness", p.cfg.MaxEpochStaleness)
+		} else {
+			p.log.Debug("Cached epoch is too stale to probe with, skipping probe tick", "epoch", epoch, "age", age)
+		}
+		return 0, false
+	}
+
+	return epoch, true
+}
+
+// refreshEpochLoop keeps the cached epoch warm, independently of the probe loop.
+func (p *Pinger) refreshEpochLoop(ctx context.Context) {
+	p.refreshEpoch(ctx)
+
+	ticker := time.NewTicker(p.cfg.EpochRefreshInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			p.log.Debug("Epoch refresh loop done")
+			return
+		case <-ticker.C:
+			p.refreshEpoch(ctx)
+		}
+	}
+}
+
+func (p *Pinger) refreshEpoch(ctx context.Context) {
+	epoch, err := p.getCurrentEpoch(ctx)
+	if err != nil {
+		if ctx.Err() != nil {
+			return
+		}
+		p.markEpochStale(err)
+		return
+	}
+	p.storeEpoch(epoch)
+}
+
+// storeEpoch caches a freshly fetched epoch, and reports the recovery if we had been falling back
+// to a cached value.
+func (p *Pinger) storeEpoch(epoch uint64) {
+	now := p.cfg.NowFunc()
+
+	p.mu.Lock()
+	wasStale, hadEpoch := p.servingStale, p.haveEpoch
+	cached, staleSince := p.epoch, p.epochAt
+	p.epoch = epoch
+	p.haveEpoch = true
+	p.epochAt = now
+	p.servingStale = false
+	p.warnedNoEpoch = false
+	p.warnedTooStale = false
+	p.mu.Unlock()
+
+	metrics.EpochCacheStaleAge.Set(0)
+
+	if wasStale {
+		if hadEpoch {
+			p.log.Info("Epoch fetch recovered", "epoch", epoch, "cachedEpoch", cached, "staleFor", now.Sub(staleSince))
+		} else {
+			// Started while the ledger was unreachable, so there was nothing to fall back to.
+			p.log.Info("Epoch fetch recovered, starting to probe", "epoch", epoch)
+		}
+	}
+}
+
+// markEpochStale records that the epoch fetch is failing. Repeated failures are collapsed into the
+// fresh->stale transition so that a multi-hour outage produces a handful of log lines rather than
+// one per attempt.
+func (p *Pinger) markEpochStale(err error) {
+	p.mu.Lock()
+	epoch, have, at := p.epoch, p.haveEpoch, p.epochAt
+	first := !p.servingStale
+	p.servingStale = true
+	p.mu.Unlock()
+
+	if !have {
+		// epochForTick reports this case; there is no cached epoch to fall back to.
+		p.log.Debug("Failed to get current epoch, none cached", "error", err)
+		return
+	}
+
+	age := p.cfg.NowFunc().Sub(at)
+	metrics.EpochCacheStaleAge.Set(age.Seconds())
+
+	if first {
+		p.log.Warn("Failed to get current epoch, probing with the last known epoch", "epoch", epoch, "age", age, "maxStaleness", p.cfg.MaxEpochStaleness, "error", err)
+	} else {
+		p.log.Debug("Failed to get current epoch, still probing with the last known epoch", "epoch", epoch, "age", age, "error", err)
+	}
+}
+
 // getCurrentEpoch gets the current epoch, with a few retries to mitigate any transient network
-// issues. The pinger does not rely on this to succeed, and will just try again on the next tick
-// if it fails all retries.
+// issues. Callers do not rely on this to succeed: the refresh loop tries again on its next tick,
+// and probing continues against the cached epoch in the meantime.
 func (p *Pinger) getCurrentEpoch(ctx context.Context) (uint64, error) {
 	attempt := 0
 	epoch, err := backoff.Retry(ctx, func() (uint64, error) {
 		if attempt > 1 {
-			p.log.Warn("Failed to get current epoch, retrying", "attempt", attempt)
+			// Debug, not Warn: markEpochStale carries the operator-facing signal, collapsed into
+			// the fresh->stale transition. A per-attempt warning here means thousands of lines
+			// across a multi-hour outage.
+			p.log.Debug("Failed to get current epoch, retrying", "attempt", attempt)
 		}
 		attempt++
 		epoch, err := p.cfg.GetCurrentEpoch(ctx)