telemetry: surface program errors from finalized transactions

CHANGELOG.md

@@ -10,7 +10,13 @@ All notable changes to this project will be documented in this file.
 
 - CLI
   - `doublezero feed list` gains a `group_codes` column naming the multicast groups the feed holds, alongside the existing `groups` count. A group the ledger no longer carries renders as its raw pubkey. The JSON output gains the field as well. (malbeclabs/infra#2172, #4150)
+- SDK
+  - The Go telemetry SDK no longer reports a transaction the program rejected as a success. Finalization only means the cluster agreed on the transaction: a rejected instruction finalizes too, carrying the rejection in `err`, which the executor never read. It now returns a `*telemetry.ProgramError` holding the ledger's error and the program's log output, and leads the message with the program's own explanation so a caller that just prints the error still gets the reason. This is the check the serviceability executor already made. (malbeclabs/infra#1703, #4152)
+  - A samples-account-full or missing-account rejection that reaches execution now returns the same `ErrSamplesAccountFull` / `ErrAccountNotFound` the equivalent preflight rejection does, via the new `ProgramError.CustomErrorCode()`. Preflight catches nearly all of these, but a write that simulated cleanly and then failed against the bank it landed on reported its code only through the finalized transaction, so a caller's account-full handling worked on one side of preflight and not the other. (malbeclabs/infra#1703, #4152)
+- Collector
+  - A failed internet-latency submission now retries from the first unwritten sample rather than restarting at the beginning of the flushed partition, and only the unwritten remainder is requeued — the same fix the device telemetry submitter got in #4145. A partition larger than one transaction is written in batches, so any failure part-way through left the earlier batches onchain while the retry and the next tick re-sent them, appending those samples a second time and skewing the latency they feed. Reachable today from an RPC timeout mid-partition; surfacing program rejections adds another way in. (malbeclabs/infra#1703, #4152)
 - Device Telemetry
+  - A submission the telemetry program rejects onchain no longer burns the tick's remaining attempts: the agent logs the rejection with the program's explanation at Error and moves on, counting `submitter_program_error` on the errors counter. Before this, the init half of the init→write path could not be seen to fail — it skips preflight, so the rejection only showed up on the finalized transaction, which the SDK read as success — and the agent looped init→write→`account not found` every few seconds with nothing in the log naming the cause. Observed on chi-dn-dzd4, where the device's `metrics_publisher` had been set to a key the agent did not hold. Samples are requeued as with any other failure, so the next tick retries once the cause is fixed. An init the program rejects because the account already exists is excepted: that leaves the write with what it needed, so the write now runs either way and only a write that still finds nothing there reports the init failure as the reason. (malbeclabs/infra#1703, #4152)
   - 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 when no epoch has ever been fetched, when the cached one exceeds the new `-max-epoch-staleness` (default 10h, clamped to what the sample buffer holds at `-probe-interval`), or when the cached epoch's projected end has passed. Samples taken against a cached epoch are written to that epoch's account, so a query scoped to a later epoch will not return them — the projected-end bound is what keeps that from spanning a rollover. The refresh cadence follows `-probe-interval` and can be set with the new `-epoch-refresh-interval`. (#4143)
   - 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)
   - The telemetry agent now logs and counts samples it discards when a submission fails and the partition buffer is already at capacity; that path previously recycled the batch with no signal at all. New counter `doublezero_device_telemetry_agent_samples_dropped_total` with a `reason` label (`buffer_full`), plus `submitter_buffer_full` on the existing errors counter. Requeue behavior below capacity is unchanged, and neither signal fires in steady state. (#4144)

controlplane/internet-latency-collector/internal/exporter/submitter.go

@@ -104,12 +104,16 @@ func (s *Submitter) Run(ctx context.Context) error {
 	}
 }
 
-func (s *Submitter) SubmitSamples(ctx context.Context, partitionKey PartitionKey, samples []Sample) error {
+// SubmitSamples writes samples to the partition's onchain account in batches, and returns how many
+// of them were written. That count is the caller's resume point: the batches before it are already
+// onchain, so a retry must pass samples[written:] rather than re-sending the whole slice, or those
+// samples are appended a second time.
+func (s *Submitter) SubmitSamples(ctx context.Context, partitionKey PartitionKey, samples []Sample) (int, error) {
 	log := s.log.With("partition", partitionKey)
 
 	if len(samples) == 0 {
 		log.Debug("No samples to submit, skipping")
-		return nil
+		return 0, nil
 	}
 
 	for i := 0; i < len(samples); i += telemetry.MaxInternetLatencySamplesPerBatch {
@@ -140,7 +144,7 @@ func (s *Submitter) SubmitSamples(ctx context.Context, partitionKey PartitionKey
 				log.Info("Account not found, initializing new account")
 				samplingInterval, ok := s.cfg.DataProviderSamplingIntervals[partitionKey.DataProvider]
 				if !ok {
-					return fmt.Errorf("no sampling interval found for data provider: %s", partitionKey.DataProvider)
+					return i, fmt.Errorf("no sampling interval found for data provider: %s", partitionKey.DataProvider)
 				}
 				_, _, err = s.cfg.Telemetry.InitializeInternetLatencySamples(ctx, telemetry.InitializeInternetLatencySamplesInstructionConfig{
 					DataProviderName:             string(partitionKey.DataProvider),
@@ -150,33 +154,33 @@ func (s *Submitter) SubmitSamples(ctx context.Context, partitionKey PartitionKey
 					SamplingIntervalMicroseconds: uint64(samplingInterval.Microseconds()),
 				})
 				if err != nil {
-					return fmt.Errorf("failed to initialize internet latency samples: %w", err)
+					return i, fmt.Errorf("failed to initialize internet latency samples: %w", err)
 				}
 				_, _, err = s.cfg.Telemetry.WriteInternetLatencySamples(ctx, writeConfig)
 				if err != nil {
 					if errors.Is(err, telemetry.ErrSamplesAccountFull) {
-						log.Warn("Partition account is full, dropping samples from buffer and moving on", "droppedSamples", len(samples))
+						log.Warn("Partition account is full, dropping samples from buffer and moving on", "droppedSamples", len(samples)-i)
 						metrics.ExporterSubmitterAccountFull.WithLabelValues(string(partitionKey.DataProvider), partitionKey.SourceExchangePK.String(), partitionKey.TargetExchangePK.String(), strconv.FormatUint(partitionKey.Epoch, 10)).Inc()
 						s.cfg.Buffer.Remove(partitionKey)
-						return nil
+						return i, nil
 					}
-					return fmt.Errorf("failed to write internet latency samples after init: %w", err)
+					return i, fmt.Errorf("failed to write internet latency samples after init: %w", err)
 				}
 			} else if errors.Is(err, telemetry.ErrSamplesAccountFull) {
-				log.Warn("Partition account is full, dropping samples from buffer and moving on", "droppedSamples", len(samples))
+				log.Warn("Partition account is full, dropping samples from buffer and moving on", "droppedSamples", len(samples)-i)
 				metrics.ExporterSubmitterAccountFull.WithLabelValues(string(partitionKey.DataProvider), partitionKey.SourceExchangePK.String(), partitionKey.TargetExchangePK.String(), strconv.FormatUint(partitionKey.Epoch, 10)).Inc()
 				s.cfg.Buffer.Remove(partitionKey)
-				return nil
+				return i, nil
 			} else {
-				return fmt.Errorf("failed to write internet latency samples: %w", err)
+				return i, fmt.Errorf("failed to write internet latency samples: %w", err)
 			}
 		}
 
 		metrics.ExporterPartitionedBufferSize.WithLabelValues(string(partitionKey.DataProvider), partitionKey.SourceExchangePK.String(), partitionKey.TargetExchangePK.String()).Set(float64(len(samples)))
-		log.Debug("Submitted partition samples batch", "count", len(samples), "samples", rtts)
+		log.Debug("Submitted partition samples batch", "count", len(batch), "samples", rtts)
 	}
 
-	return nil
+	return len(samples), nil
 }
 
 func (s *Submitter) Tick(ctx context.Context) {
@@ -232,14 +236,20 @@ func (s *Submitter) Tick(ctx context.Context) {
 				return
 			}
 
+			// Samples written so far across attempts. Each retry resumes here rather than at the
+			// start of tmp, so batches an earlier attempt put onchain are neither re-sent nor
+			// counted as lost.
+			written := 0
+
 			success := false
 			for attempt := 1; attempt <= maxAttempts; attempt++ {
 				// Bound each attempt so a slow/degraded ledger RPC can't leave the submission
 				// blocked past its blockhash's validity window; the next attempt re-fetches a
 				// fresh blockhash.
 				attemptCtx, cancel := context.WithTimeout(ctx, attemptTimeout)
-				err := s.SubmitSamples(attemptCtx, partitionKey, tmp)
+				n, err := s.SubmitSamples(attemptCtx, partitionKey, tmp[written:])
 				cancel()
+				written += n
 				if err == nil {
 					log.Debug("Submitted samples", "count", len(tmp), "attempt", attempt)
 					success = true
@@ -272,7 +282,7 @@ func (s *Submitter) Tick(ctx context.Context) {
 			}
 
 			if !success {
-				s.cfg.Buffer.PriorityPrepend(partitionKey, tmp)
+				s.cfg.Buffer.PriorityPrepend(partitionKey, tmp[written:])
 			}
 
 			// Always recycle the slice for reuse

controlplane/internet-latency-collector/internal/exporter/submitter_test.go

@@ -741,6 +741,58 @@ func TestInternetLatency_Submitter(t *testing.T) {
 		}
 	})
 
+	// A partition too large for one transaction is written in batches, so a failure part-way through
+	// leaves some of them onchain. Resuming at the first unwritten sample is what keeps the retry
+	// from appending the earlier batches a second time and skewing the latency data they feed.
+	t.Run("retries_resume_at_the_first_unwritten_sample", func(t *testing.T) {
+		t.Parallel()
+
+		log := logger.With("test", t.Name())
+
+		key := newTestPartitionKey()
+		total := sdktelemetry.MaxInternetLatencySamplesPerBatch + 17
+
+		var writes int32
+		var submitted []uint32
+		var mu sync.Mutex
+		telemetryProgram := &mockTelemetryProgramClient{
+			WriteInternetLatencySamplesFunc: func(_ context.Context, config sdktelemetry.WriteInternetLatencySamplesInstructionConfig) (solana.Signature, *solanarpc.GetTransactionResult, error) {
+				// The first batch lands, the second fails, and the retry picks up from there.
+				if atomic.AddInt32(&writes, 1) == 1 {
+					mu.Lock()
+					submitted = append(submitted, config.Samples...)
+					mu.Unlock()
+					return solana.Signature{}, nil, nil
+				}
+				return solana.Signature{}, nil, errors.New("ledger rpc unreachable")
+			},
+		}
+
+		buf := buffer.NewMemoryPartitionedBuffer[exporter.PartitionKey, exporter.Sample](4096)
+		for range total {
+			buf.Add(key, newTestSample())
+		}
+
+		submitter, err := exporter.NewSubmitter(log, &exporter.SubmitterConfig{
+			OracleAgentPK: solana.NewWallet().PublicKey(),
+			Interval:      time.Hour,
+			Buffer:        buf,
+			Telemetry:     telemetryProgram,
+			MaxAttempts:   2,
+			BackoffFunc:   func(_ int) time.Duration { return 0 },
+			EpochFinder:   &mockEpochFinder{ApproximateAtTimeFunc: func(context.Context, time.Time) (uint64, error) { return key.Epoch, nil }},
+		})
+		require.NoError(t, err)
+
+		submitter.Tick(t.Context())
+
+		mu.Lock()
+		defer mu.Unlock()
+		assert.Len(t, submitted, sdktelemetry.MaxInternetLatencySamplesPerBatch,
+			"the batch that landed should be written exactly once, not re-sent by the retry")
+		assert.Len(t, buf.CopyAndReset(key), 17,
+			"only the samples never written should be requeued, or the next tick appends them twice")
+	})
 }
 
 func waitTimeout(wg *sync.WaitGroup, timeout time.Duration) bool {

controlplane/telemetry/internal/metrics/metrics.go

@@ -43,6 +43,12 @@ const (
 	ErrorTypePingerEpochFetch     = "pinger_epoch_fetch"
 	ErrorTypeSubmitterBufferFull  = "submitter_buffer_full"
 	ErrorTypeSubmitterAccountFull = "submitter_account_full"
+	// ErrorTypeSubmitterProgramError counts submissions the telemetry program rejected onchain.
+	// It overlaps the write/init failure types rather than replacing them: those name the operation
+	// that failed, this one narrows why to a rejection that will recur until something changes
+	// onchain or in config, as opposed to the transient RPC trouble they also cover. A rejected init
+	// increments both.
+	ErrorTypeSubmitterProgramError = "submitter_program_error"
 
 	// Sample drop reasons.
 	DropReasonBufferFull  = "buffer_full"

controlplane/telemetry/internal/telemetry/submitter.go

@@ -136,7 +136,7 @@ func (s *Submitter) SubmitSamples(ctx context.Context, partitionKey PartitionKey
 		if err != nil {
 			if errors.Is(err, telemetry.ErrAccountNotFound) {
 				log.Info("Account not found, initializing new account")
-				_, _, err = s.cfg.ProgramClient.InitializeDeviceLatencySamples(ctx, telemetry.InitializeDeviceLatencySamplesInstructionConfig{
+				_, _, initErr := s.cfg.ProgramClient.InitializeDeviceLatencySamples(ctx, telemetry.InitializeDeviceLatencySamplesInstructionConfig{
 					AgentPK:                      s.cfg.MetricsPublisherPK,
 					OriginDevicePK:               partitionKey.OriginDevicePK,
 					TargetDevicePK:               partitionKey.TargetDevicePK,
@@ -146,16 +146,28 @@ func (s *Submitter) SubmitSamples(ctx context.Context, partitionKey PartitionKey
 					AgentVersion:                 s.cfg.AgentVersion,
 					AgentCommit:                  s.cfg.AgentCommit,
 				})
-				if err != nil {
-					metrics.Errors.WithLabelValues(metrics.ErrorTypeSubmitterFailedToInitializeAccount).Inc()
-					return i, fmt.Errorf("failed to initialize device latency samples: %w", err)
+				if initErr != nil {
+					// Not fatal on its own. An init the program rejects because the account already
+					// exists has left us with exactly what the write needs, which happens when a
+					// previous init landed onchain without the agent seeing it succeed. The write
+					// below is what decides whether the rejection mattered, so it runs either way.
+					// The error counter waits for that verdict rather than firing on a failure the
+					// write goes on to absorb.
+					log.Warn("Failed to initialize account, attempting the write anyway", "error", initErr)
 				}
 				_, _, err = s.cfg.ProgramClient.WriteDeviceLatencySamples(ctx, writeConfig)
 				if err != nil {
 					if errors.Is(err, telemetry.ErrSamplesAccountFull) {
 						s.handleAccountFull(log, partitionKey, len(samples)-i)
 						return i, nil
 					}
+					if initErr != nil {
+						// The account is still not there, so the init failure is the reason the
+						// write had nothing to write to. Report that rather than the missing
+						// account it caused.
+						metrics.Errors.WithLabelValues(metrics.ErrorTypeSubmitterFailedToInitializeAccount).Inc()
+						return i, fmt.Errorf("failed to initialize device latency samples: %w", initErr)
+					}
 					metrics.Errors.WithLabelValues(metrics.ErrorTypeSubmitterFailedToWriteSamples).Inc()
 					return i, fmt.Errorf("failed to write device latency samples after init: %w", err)
 				}
@@ -247,6 +259,19 @@ func (s *Submitter) Tick(ctx context.Context) {
 					break
 				}
 
+				// A rejection by the program is not transient: the ledger executed the instruction
+				// and refused it, so every attempt this tick would be refused the same way. Report
+				// it once at Error and leave the rest of the attempts unspent, rather than burying
+				// the reason under a backoff loop. The samples are requeued as with any other
+				// failure, so the next tick retries once the operator fixes what was wrong.
+				var programErr *telemetry.ProgramError
+				if errors.As(err, &programErr) {
+					metrics.Errors.WithLabelValues(metrics.ErrorTypeSubmitterProgramError).Inc()
+					log.Error("Submission rejected by the telemetry program, not retrying this tick",
+						"attempt", attempt, "samplesCount", len(tmp), "error", err)
+					break
+				}
+
 				var backoff time.Duration
 				if s.cfg.BackoffFunc != nil {
 					backoff = s.cfg.BackoffFunc(attempt)