telemetry: surface program errors from finalized transactions (#4152)

Resolves: malbeclabs/infra#1703

## Summary of Changes

- The Go telemetry SDK executor no longer reports a transaction the
program rejected as a success. Finalization only means the cluster
agreed on the transaction: an instruction the program refused 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.
- The device telemetry submitter treats that rejection as permanent for
the tick — it logs the program's own explanation at `Error` and leaves
the tick's remaining attempts unspent, rather than spending them on an
instruction the ledger has already refused. New
`submitter_program_error` type on the existing errors counter.
- `ProgramError.Error()` leads with the program's explanation (`Program
log:` lines, minus the runtime's invoke/consumed boilerplate and the
instruction-name echo), so a caller that only prints the error still
gets the reason. This is the check
`smartcontract/sdk/go/serviceability/executor.go` already made;
telemetry was the outlier.

This is what left chi-dn-dzd4 silent for over an hour. Its
`metrics_publisher` had been set to a key the agent did not hold, and
the init half of the init→write path skips preflight — so
`UnauthorizedAgent` (0x3e9) only arrived on the finalized transaction,
which the SDK read as success. The agent looped init → write → `account
not found` every few seconds, and the only error it printed named the
missing account, not the authorization failure that caused it.

## Diff Breakdown

| Category     | Files | Lines (+/-) | Net  |
|--------------|-------|-------------|------|
| Tests        |     2 | +178 / -0   | +178 |
| Core logic   |     2 | +86 / -0    |  +86 |
| Docs         |     1 | +3 / -0     |   +3 |
| Scaffolding  |     1 | +4 / -0     |   +4 |
| **Total**    |     6 | +271 / -0   | +271 |

Two thirds tests: the fix itself is 86 lines across the SDK executor and
one branch in the submitter's retry loop.

<details>
<summary>Key files (click to expand)</summary>

- `smartcontract/sdk/go/telemetry/executor.go` — new `ProgramError` type
with the program-log filter; `waitForTransactionFinalized` returns it
when the finalized signature status or the transaction meta carries an
error, fetching the logs best effort so a node that cannot return the
transaction costs the logs rather than replacing the rejection with an
RPC error
- `controlplane/telemetry/internal/telemetry/submitter.go` — `errors.As`
branch in `Tick`'s retry loop: count, log at `Error`, stop the tick's
attempts; samples requeue as with any other failure
- `controlplane/telemetry/internal/metrics/metrics.go` —
`ErrorTypeSubmitterProgramError`, kept distinct from the write/init
failure types because those also cover transient RPC trouble

</details>

## Testing Verification

- `TestSDK_Telemetry_Executor_FinalizedWithProgramError` reproduces the
chi-dn-dzd4 transaction — finalized, `Custom: 1001`, with the "not
authorized for origin device" program log — across three shapes: the
rejection on the signature status, on the transaction meta only (a node
that returns a clean status), and with the logs unfetchable. All three
must return an error rather than a signature; the error message carries
the custom code in every case and the program's explanation whenever the
logs were available, with the boilerplate stripped.
- `does_not_retry_a_submission_the_program_rejected` drives a full
submitter tick with `MaxAttempts: 5` and asserts exactly one write and
one init reach the program, `submitter_program_error` increments once,
`submitter_retries_exhausted` does not, the reason reaches the log, and
the samples are requeued for the next tick.
- The two pre-existing finalization edge cases (`Meta == nil`,
`GetTransaction` returning nil) keep their original error messages — a
clean signature status still falls through to them unchanged.
- Full suites pass for `smartcontract/sdk/go/telemetry`,
`controlplane/telemetry/...`, and
`controlplane/internet-latency-collector/...`. One unrelated
pre-existing failure in `controlplane/telemetry/internal/netns`
(`TestRunInNamespace_EmptyNameErrors` needs namespace privileges)
reproduces identically on a clean `main` checkout.

### Not in scope

-
`controlplane/internet-latency-collector/internal/exporter/submitter.go`
has the same init→write shape and picks up the SDK's real error for
free, but keeps retrying a rejection through its attempts. Left for a
follow-up rather than widened here.
- A write that passes preflight and then finalizes against a full
account now surfaces as a generic `ProgramError` instead of the
`ErrSamplesAccountFull` sentinel, so it requeues and takes the drop path
on the next tick instead of immediately. Previously that race reported
success and lost the samples with no signal at all, so this is strictly
better; mapping `meta.err` custom codes back onto the sentinels would
need a parser in `client.go` and is a separate change.

Author: elitegreg

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)