From c9e4489c9db060ed2aa59c48c8f34fda72731ebd Mon Sep 17 00:00:00 2001
From: Greg Mitchell <greg@malbeclabs.com>
Date: Wed, 5 Aug 2026 16:31:59 +0000
Subject: [PATCH 1/4] telemetry: surface program errors from finalized
 transactions

The Go telemetry SDK executor treated any finalized transaction as a
success. Finalization only means the cluster agreed on the transaction:
an instruction the program rejected finalizes too, carrying the
rejection in err, which the executor never read.

The executor now returns a *ProgramError holding the ledger's error and
the program's log output, and the device telemetry submitter treats it
as a permanent failure for the tick: it logs the rejection with the
program's explanation and leaves the remaining attempts unspent instead
of burying the reason under a backoff loop.

Seen on chi-dn-dzd4, whose metrics_publisher had been set to a key the
agent did not hold: the init half of the init->write path skips
preflight, so the rejection only arrived on the finalized transaction,
and the agent looped init -> write -> "account not found" for over an
hour with nothing naming the cause.
---
 CHANGELOG.md                                  |   3 +
 .../telemetry/internal/metrics/metrics.go     |   4 +
 .../telemetry/internal/telemetry/submitter.go |  13 ++
 .../internal/telemetry/submitter_test.go      |  62 ++++++++++
 smartcontract/sdk/go/telemetry/executor.go    |  73 +++++++++++
 .../sdk/go/telemetry/executor_test.go         | 116 ++++++++++++++++++
 6 files changed, 271 insertions(+)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 95306cf3b1..59f540830b 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -10,7 +10,10 @@ 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)
 - 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. (malbeclabs/infra#1703)
   - 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)
diff --git a/controlplane/telemetry/internal/metrics/metrics.go b/controlplane/telemetry/internal/metrics/metrics.go
index e7197d844a..c02a59ec6d 100644
--- a/controlplane/telemetry/internal/metrics/metrics.go
+++ b/controlplane/telemetry/internal/metrics/metrics.go
@@ -43,6 +43,10 @@ const (
 	ErrorTypePingerEpochFetch     = "pinger_epoch_fetch"
 	ErrorTypeSubmitterBufferFull  = "submitter_buffer_full"
 	ErrorTypeSubmitterAccountFull = "submitter_account_full"
+	// ErrorTypeSubmitterProgramError counts submissions the telemetry program rejected onchain.
+	// Distinct from the write/init failure types, which also cover transient RPC trouble: this one
+	// only fires on a rejection that will recur until something changes onchain or in config.
+	ErrorTypeSubmitterProgramError = "submitter_program_error"
 
 	// Sample drop reasons.
 	DropReasonBufferFull  = "buffer_full"
diff --git a/controlplane/telemetry/internal/telemetry/submitter.go b/controlplane/telemetry/internal/telemetry/submitter.go
index de9ff2a43b..cb6c5ed378 100644
--- a/controlplane/telemetry/internal/telemetry/submitter.go
+++ b/controlplane/telemetry/internal/telemetry/submitter.go
@@ -247,6 +247,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)
diff --git a/controlplane/telemetry/internal/telemetry/submitter_test.go b/controlplane/telemetry/internal/telemetry/submitter_test.go
index d8b268b582..bf0dba761e 100644
--- a/controlplane/telemetry/internal/telemetry/submitter_test.go
+++ b/controlplane/telemetry/internal/telemetry/submitter_test.go
@@ -1211,4 +1211,66 @@ func TestAgentTelemetry_Submitter(t *testing.T) {
 		assert.Equal(t, float64(1), errs, "account-full error counter should increment once")
 		assert.Contains(t, logs.String(), "Partition account is full, dropping partition")
 	})
+
+	// The chi-dn-dzd4 case (malbeclabs/infra#1703): the agent key is not the device's
+	// metrics_publisher, so the program rejects the init. Deliberately not parallel: the assertions
+	// are exact deltas on package-level prometheus counters.
+	t.Run("does_not_retry_a_submission_the_program_rejected", func(t *testing.T) {
+		var logs bytes.Buffer
+		log := slog.New(slog.NewTextHandler(&logs, &slog.HandlerOptions{Level: slog.LevelWarn}))
+
+		key := newTestPartitionKey()
+
+		var writes, inits int32
+		prog := &mockTelemetryProgramClient{
+			WriteDeviceLatencySamplesFunc: func(context.Context, sdktelemetry.WriteDeviceLatencySamplesInstructionConfig) (solana.Signature, *solanarpc.GetTransactionResult, error) {
+				atomic.AddInt32(&writes, 1)
+				return solana.Signature{}, nil, sdktelemetry.ErrAccountNotFound
+			},
+			InitializeDeviceLatencySamplesFunc: func(context.Context, sdktelemetry.InitializeDeviceLatencySamplesInstructionConfig) (solana.Signature, *solanarpc.GetTransactionResult, error) {
+				atomic.AddInt32(&inits, 1)
+				return solana.Signature{}, nil, &sdktelemetry.ProgramError{
+					Err: map[string]any{"InstructionError": []any{0, map[string]any{"Custom": 1001}}},
+					Logs: []string{
+						"Program log: Instruction: InitializeDeviceLatencySamples",
+						"Program log: Agent BA14eqpRNmkcQhjsH5abfvaUxRi7RcGGuQVeQuJdwPZc is not authorized for origin device FYkmttUmox6kZVjVNCATXEdGt3bfLicn5fJ8fnGfF4fZ",
+					},
+				}
+			},
+		}
+
+		buf := buffer.NewMemoryPartitionedBuffer[telemetry.PartitionKey, telemetry.Sample](1024)
+		buf.Add(key, newTestSample())
+
+		s, err := telemetry.NewSubmitter(log, &telemetry.SubmitterConfig{
+			Interval:        time.Hour,
+			Buffer:          buf,
+			ProgramClient:   prog,
+			MaxAttempts:     5,
+			MaxConcurrency:  10,
+			BackoffFunc:     func(int) time.Duration { return 0 },
+			GetCurrentEpoch: func(context.Context) (uint64, error) { return 100, nil },
+		})
+		require.NoError(t, err)
+
+		programErrsBefore := testutil.ToFloat64(metrics.Errors.WithLabelValues(metrics.ErrorTypeSubmitterProgramError))
+		exhaustedBefore := testutil.ToFloat64(metrics.Errors.WithLabelValues(metrics.ErrorTypeSubmitterRetriesExhausted))
+
+		s.Tick(context.Background())
+
+		assert.Equal(t, int32(1), atomic.LoadInt32(&writes), "the rejection should end the tick, not spend the remaining attempts")
+		assert.Equal(t, int32(1), atomic.LoadInt32(&inits), "an init the program rejected should not be re-sent unchanged")
+
+		programErrs := testutil.ToFloat64(metrics.Errors.WithLabelValues(metrics.ErrorTypeSubmitterProgramError)) - programErrsBefore
+		assert.Equal(t, float64(1), programErrs, "program-error counter should increment once")
+
+		exhausted := testutil.ToFloat64(metrics.Errors.WithLabelValues(metrics.ErrorTypeSubmitterRetriesExhausted)) - exhaustedBefore
+		assert.Equal(t, float64(0), exhausted, "retries were not exhausted, they were skipped")
+
+		out := logs.String()
+		assert.Contains(t, out, "Submission rejected by the telemetry program")
+		assert.Contains(t, out, "is not authorized for origin device", "the reason the program gave has to reach the log")
+
+		assert.Len(t, buf.CopyAndReset(key), 1, "samples should be requeued for the next tick")
+	})
 }
diff --git a/smartcontract/sdk/go/telemetry/executor.go b/smartcontract/sdk/go/telemetry/executor.go
index 1b09362154..7e5f429571 100644
--- a/smartcontract/sdk/go/telemetry/executor.go
+++ b/smartcontract/sdk/go/telemetry/executor.go
@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"log/slog"
+	"strings"
 	"time"
 
 	"github.com/gagliardetto/solana-go"
@@ -19,6 +20,42 @@ var (
 	ErrNoProgramID = errors.New("no program ID configured")
 )
 
+// ProgramError reports a transaction that finalized onchain while the program rejected the
+// instruction it carried. The instruction did not take effect, and re-sending it unchanged will be
+// rejected the same way, so callers should treat it as a permanent failure for that input rather
+// than retry it.
+type ProgramError struct {
+	// Err is the transaction error the ledger reported, e.g.
+	// map[InstructionError:[0 map[Custom:1001]]] for TelemetryError::UnauthorizedAgent.
+	Err any
+
+	// Logs is the program's log output for the transaction. It is empty when the RPC returned the
+	// failure on the signature status but could not return the transaction itself.
+	Logs []string
+}
+
+func (e *ProgramError) Error() string {
+	if msgs := e.ProgramLogMessages(); len(msgs) > 0 {
+		return fmt.Sprintf("transaction finalized with program error: %v (program logs: %v)", e.Err, msgs)
+	}
+	return fmt.Sprintf("transaction finalized with program error: %v", e.Err)
+}
+
+// ProgramLogMessages returns the program's own log lines with the runtime's invoke/consumed/success
+// boilerplate and the instruction-name echo removed. These are the lines that say why the program
+// rejected the instruction, e.g. "Agent <pubkey> is not authorized for origin device <pubkey>".
+func (e *ProgramError) ProgramLogMessages() []string {
+	var msgs []string
+	for _, line := range e.Logs {
+		msg, ok := strings.CutPrefix(line, "Program log: ")
+		if !ok || strings.HasPrefix(msg, "Instruction: ") {
+			continue
+		}
+		msgs = append(msgs, msg)
+	}
+	return msgs
+}
+
 type executor struct {
 	log                   *slog.Logger
 	rpc                   RPCClient
@@ -122,6 +159,12 @@ func (e *executor) ExecuteTransactions(ctx context.Context, instructions []solan
 	// Wait for the transaction to be finalized
 	res, err := e.waitForTransactionFinalized(ctx, sig)
 	if err != nil {
+		// A program rejection is not a failure to read the transaction; pass it through so the
+		// reason stays at the front of the message.
+		var programErr *ProgramError
+		if errors.As(err, &programErr) {
+			return solana.Signature{}, nil, err
+		}
 		return solana.Signature{}, nil, fmt.Errorf("failed to get transaction: %w", err)
 	}
 
@@ -147,6 +190,7 @@ func (e *executor) waitForSignatureVisible(ctx context.Context, sig solana.Signa
 func (e *executor) waitForTransactionFinalized(ctx context.Context, sig solana.Signature) (*solanarpc.GetTransactionResult, error) {
 	e.log.Debug("--> Waiting for transaction to be finalized", "sig", sig)
 	start := time.Now()
+	var finalStatus *solanarpc.SignatureStatusesResult
 	for {
 		statusResp, err := e.rpc.GetSignatureStatuses(ctx, true, sig)
 		if err != nil {
@@ -158,6 +202,7 @@ func (e *executor) waitForTransactionFinalized(ctx context.Context, sig solana.S
 		status := statusResp.Value[0]
 		if status != nil && status.ConfirmationStatus == solanarpc.ConfirmationStatusFinalized {
 			e.log.Debug("--> Transaction finalized", "sig", sig, "duration", time.Since(start))
+			finalStatus = status
 			break
 		}
 		select {
@@ -170,6 +215,14 @@ func (e *executor) waitForTransactionFinalized(ctx context.Context, sig solana.S
 		}
 	}
 
+	// Finalization only says the cluster agreed on the transaction, not that the program accepted
+	// it: a rejected instruction finalizes and carries the rejection in Err. Reporting that as
+	// success leaves the caller believing an account it never got was written, and the program
+	// error never reaches the log.
+	if finalStatus.Err != nil {
+		return nil, &ProgramError{Err: finalStatus.Err, Logs: e.transactionLogs(ctx, sig)}
+	}
+
 	tx, err := e.rpc.GetTransaction(ctx, sig, &solanarpc.GetTransactionOpts{
 		Encoding:   solana.EncodingBase64,
 		Commitment: solanarpc.CommitmentFinalized,
@@ -180,5 +233,25 @@ func (e *executor) waitForTransactionFinalized(ctx context.Context, sig solana.S
 	if tx == nil || tx.Meta == nil {
 		return nil, errors.New("transaction not found or missing metadata after finalization")
 	}
+	// The same rejection is carried on the transaction metadata. Checked here as well because the
+	// two come from separate RPC calls, and a node that omits it on the status still reports it here.
+	if tx.Meta.Err != nil {
+		return nil, &ProgramError{Err: tx.Meta.Err, Logs: tx.Meta.LogMessages}
+	}
 	return tx, nil
 }
+
+// transactionLogs fetches the program logs for a finalized transaction, best effort. They are
+// context for a failure that is already known, so a node that cannot return the transaction costs
+// the logs rather than replacing the program error with an RPC error.
+func (e *executor) transactionLogs(ctx context.Context, sig solana.Signature) []string {
+	tx, err := e.rpc.GetTransaction(ctx, sig, &solanarpc.GetTransactionOpts{
+		Encoding:   solana.EncodingBase64,
+		Commitment: solanarpc.CommitmentFinalized,
+	})
+	if err != nil || tx == nil || tx.Meta == nil {
+		e.log.Debug("--> Could not fetch program logs for failed transaction", "sig", sig, "error", err)
+		return nil
+	}
+	return tx.Meta.LogMessages
+}
diff --git a/smartcontract/sdk/go/telemetry/executor_test.go b/smartcontract/sdk/go/telemetry/executor_test.go
index ca99b4329b..8787bdeba9 100644
--- a/smartcontract/sdk/go/telemetry/executor_test.go
+++ b/smartcontract/sdk/go/telemetry/executor_test.go
@@ -285,6 +285,122 @@ func TestSDK_Telemetry_Executor_TransactionNeverFinalized(t *testing.T) {
 	require.Nil(t, res)
 }
 
+// TestSDK_Telemetry_Executor_FinalizedWithProgramError covers the chi-dn-dzd4 case
+// (malbeclabs/infra#1703): an InitializeDeviceLatencySamples the program rejected with
+// UnauthorizedAgent (0x3e9) still finalizes, and reporting it as success left the submitter
+// re-initializing an account that never existed with nothing in the log naming the cause.
+func TestSDK_Telemetry_Executor_FinalizedWithProgramError(t *testing.T) {
+	t.Parallel()
+
+	signer := solana.NewWallet().PrivateKey
+	signerPub := signer.PublicKey()
+	programID := solana.NewWallet().PublicKey()
+	blockhash := solana.MustHashFromBase58("5NzX7jrPWeTkGsDnVnszdEa7T3Yyr3nSgyc78z3CwjWQ")
+
+	// UnauthorizedAgent = 1001 = 0x3e9, as the RPC renders it.
+	txErr := map[string]any{"InstructionError": []any{0, map[string]any{"Custom": 1001}}}
+	logMessages := []string{
+		"Program " + programID.String() + " invoke [1]",
+		"Program log: Instruction: InitializeDeviceLatencySamples",
+		"Program log: Agent BA14eqpRNmkcQhjsH5abfvaUxRi7RcGGuQVeQuJdwPZc is not authorized for origin device FYkmttUmox6kZVjVNCATXEdGt3bfLicn5fJ8fnGfF4fZ",
+		"Program " + programID.String() + " failed: custom program error: 0x3e9",
+	}
+
+	tests := []struct {
+		name       string
+		statusErr  any
+		metaErr    any
+		getTxFails bool
+		wantLogs   []string
+	}{
+		{
+			name:      "reported on the signature status",
+			statusErr: txErr,
+			metaErr:   txErr,
+			wantLogs:  logMessages,
+		},
+		{
+			// A node that returns a clean status still reports the rejection on the transaction.
+			name:     "reported only on the transaction meta",
+			metaErr:  txErr,
+			wantLogs: logMessages,
+		},
+		{
+			// The rejection still surfaces when the logs cannot be fetched to explain it.
+			name:       "logs unavailable",
+			statusErr:  txErr,
+			getTxFails: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			mockRPC := &mockRPCClient{
+				GetLatestBlockhashFunc: func(_ context.Context, _ solanarpc.CommitmentType) (*solanarpc.GetLatestBlockhashResult, error) {
+					return &solanarpc.GetLatestBlockhashResult{
+						Value: &solanarpc.LatestBlockhashResult{Blockhash: blockhash},
+					}, nil
+				},
+				SendTransactionWithOptsFunc: func(_ context.Context, tx *solana.Transaction, _ solanarpc.TransactionOpts) (solana.Signature, error) {
+					return tx.Signatures[0], nil
+				},
+				GetSignatureStatusesFunc: func(_ context.Context, _ bool, _ ...solana.Signature) (*solanarpc.GetSignatureStatusesResult, error) {
+					return &solanarpc.GetSignatureStatusesResult{
+						Value: []*solanarpc.SignatureStatusesResult{
+							{
+								ConfirmationStatus: solanarpc.ConfirmationStatusFinalized,
+								Err:                tt.statusErr,
+							},
+						},
+					}, nil
+				},
+				GetTransactionFunc: func(_ context.Context, _ solana.Signature, _ *solanarpc.GetTransactionOpts) (*solanarpc.GetTransactionResult, error) {
+					if tt.getTxFails {
+						return nil, errors.New("rpc unavailable")
+					}
+					return &solanarpc.GetTransactionResult{
+						Meta: &solanarpc.TransactionMeta{
+							Err:         tt.metaErr,
+							LogMessages: logMessages,
+						},
+					}, nil
+				},
+			}
+
+			exec := telemetry.NewExecutor(log, mockRPC, &signer, programID)
+
+			instruction := solana.NewInstruction(
+				programID,
+				solana.AccountMetaSlice{
+					{PublicKey: signerPub, IsSigner: true, IsWritable: true},
+				},
+				[]byte{1, 2, 3},
+			)
+
+			sig, res, err := exec.ExecuteTransaction(t.Context(), instruction, &telemetry.ExecuteTransactionOptions{SkipPreflight: true})
+
+			var programErr *telemetry.ProgramError
+			require.ErrorAs(t, err, &programErr, "a finalized rejection must not be reported as success")
+			require.Equal(t, txErr, programErr.Err)
+			require.Equal(t, tt.wantLogs, programErr.Logs)
+			require.Equal(t, solana.Signature{}, sig)
+			require.Nil(t, res)
+
+			// The custom error code identifies the rejection even when the logs are missing.
+			require.ErrorContains(t, err, "Custom:1001")
+			if len(tt.wantLogs) > 0 {
+				// The program's own explanation reaches the message, without the runtime's
+				// invoke/failed boilerplate or the instruction-name echo.
+				require.ErrorContains(t, err, "is not authorized for origin device")
+				require.NotContains(t, err.Error(), "Instruction: InitializeDeviceLatencySamples")
+				require.NotContains(t, err.Error(), "invoke [1]")
+			}
+		})
+	}
+}
+
 func TestSDK_Telemetry_Executor_FinalizedButMissingTransactionMeta(t *testing.T) {
 	t.Parallel()
 

From 0a6f772ffd95d13ffd4acf476434501e5aabbec0 Mon Sep 17 00:00:00 2001
From: Greg Mitchell <greg@malbeclabs.com>
Date: Wed, 5 Aug 2026 17:06:08 +0000
Subject: [PATCH 2/4] telemetry: reference the pr in the changelog entries

---
 CHANGELOG.md | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 59f540830b..fd678bf022 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -11,9 +11,9 @@ 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)
+  - 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)
 - 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. (malbeclabs/infra#1703)
+  - 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. (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)

From 3393f70d1b1981448c79638b68b10ab2c0b94555 Mon Sep 17 00:00:00 2001
From: Greg Mitchell <greg@malbeclabs.com>
Date: Wed, 5 Aug 2026 18:09:42 +0000
Subject: [PATCH 3/4] telemetry: keep writing when an init the program refused
 left the account there

The e2e SDK telemetry tests re-initialize an existing account and, before
the executor surfaced program errors, read the rejection off
res.Meta.Err themselves. They now assert the error the SDK returns.

That exposed a path the submitter got right only by accident. An init
rejected with AccountAlreadyExists (1010) still leaves the write with
what it needed, and the old silent-success behavior meant the write ran
and succeeded. Surfacing the rejection would have turned that into a
skipped tick and a spurious error, so the write now runs whether or not
the init was accepted, and only a write that still finds nothing there
reports the init failure as the reason.
---
 CHANGELOG.md                                  |  2 +-
 .../telemetry/internal/telemetry/submitter.go | 20 +++++--
 .../internal/telemetry/submitter_test.go      | 54 ++++++++++++++++++-
 e2e/sdk_device_telemetry_test.go              | 12 +++--
 e2e/sdk_internet_telemetry_test.go            | 12 +++--
 5 files changed, 86 insertions(+), 14 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index fd678bf022..add50e358f 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -13,7 +13,7 @@ All notable changes to this project will be documented in this file.
 - 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)
 - 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. (malbeclabs/infra#1703, #4152)
+  - 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)
diff --git a/controlplane/telemetry/internal/telemetry/submitter.go b/controlplane/telemetry/internal/telemetry/submitter.go
index cb6c5ed378..2ccdceabf2 100644
--- a/controlplane/telemetry/internal/telemetry/submitter.go
+++ b/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,9 +146,14 @@ 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 {
@@ -156,6 +161,13 @@ func (s *Submitter) SubmitSamples(ctx context.Context, partitionKey PartitionKey
 						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)
 				}
diff --git a/controlplane/telemetry/internal/telemetry/submitter_test.go b/controlplane/telemetry/internal/telemetry/submitter_test.go
index bf0dba761e..7f93de3123 100644
--- a/controlplane/telemetry/internal/telemetry/submitter_test.go
+++ b/controlplane/telemetry/internal/telemetry/submitter_test.go
@@ -1212,6 +1212,57 @@ func TestAgentTelemetry_Submitter(t *testing.T) {
 		assert.Contains(t, logs.String(), "Partition account is full, dropping partition")
 	})
 
+	// An init the program rejects because the account already exists still leaves the write able to
+	// proceed, so it must not end the submission. Reachable when a previous init landed onchain
+	// without the agent seeing it succeed.
+	t.Run("writes_anyway_when_the_account_already_exists", func(t *testing.T) {
+		t.Parallel()
+
+		var logs bytes.Buffer
+		log := slog.New(slog.NewTextHandler(&logs, &slog.HandlerOptions{Level: slog.LevelWarn}))
+
+		key := newTestPartitionKey()
+
+		var writes int32
+		prog := &mockTelemetryProgramClient{
+			WriteDeviceLatencySamplesFunc: func(context.Context, sdktelemetry.WriteDeviceLatencySamplesInstructionConfig) (solana.Signature, *solanarpc.GetTransactionResult, error) {
+				// Preflight reports the account missing, then the post-init write finds it there.
+				if atomic.AddInt32(&writes, 1) == 1 {
+					return solana.Signature{}, nil, sdktelemetry.ErrAccountNotFound
+				}
+				return solana.Signature{}, nil, nil
+			},
+			InitializeDeviceLatencySamplesFunc: func(context.Context, sdktelemetry.InitializeDeviceLatencySamplesInstructionConfig) (solana.Signature, *solanarpc.GetTransactionResult, error) {
+				return solana.Signature{}, nil, &sdktelemetry.ProgramError{
+					Err:  map[string]any{"InstructionError": []any{0, map[string]any{"Custom": 1010}}},
+					Logs: []string{"Program log: Latency samples account already exists"},
+				}
+			},
+		}
+
+		buf := buffer.NewMemoryPartitionedBuffer[telemetry.PartitionKey, telemetry.Sample](1024)
+		buf.Add(key, newTestSample())
+
+		s, err := telemetry.NewSubmitter(log, &telemetry.SubmitterConfig{
+			Interval:        time.Hour,
+			Buffer:          buf,
+			ProgramClient:   prog,
+			MaxAttempts:     3,
+			MaxConcurrency:  10,
+			BackoffFunc:     func(int) time.Duration { return 0 },
+			GetCurrentEpoch: func(context.Context) (uint64, error) { return 100, nil },
+		})
+		require.NoError(t, err)
+
+		s.Tick(context.Background())
+
+		assert.Equal(t, int32(2), atomic.LoadInt32(&writes), "the write should be attempted after the rejected init")
+		assert.Len(t, buf.CopyAndReset(key), 0, "the samples were written, so nothing should be requeued")
+		assert.Contains(t, logs.String(), "attempting the write anyway")
+		assert.NotContains(t, logs.String(), "Submission rejected by the telemetry program",
+			"an account that already exists is not a submission failure")
+	})
+
 	// The chi-dn-dzd4 case (malbeclabs/infra#1703): the agent key is not the device's
 	// metrics_publisher, so the program rejects the init. Deliberately not parallel: the assertions
 	// are exact deltas on package-level prometheus counters.
@@ -1258,7 +1309,8 @@ func TestAgentTelemetry_Submitter(t *testing.T) {
 
 		s.Tick(context.Background())
 
-		assert.Equal(t, int32(1), atomic.LoadInt32(&writes), "the rejection should end the tick, not spend the remaining attempts")
+		// One write to find the account missing, one to confirm the rejected init left it missing.
+		assert.Equal(t, int32(2), atomic.LoadInt32(&writes), "the rejection should end the tick, not spend the remaining attempts")
 		assert.Equal(t, int32(1), atomic.LoadInt32(&inits), "an init the program rejected should not be re-sent unchanged")
 
 		programErrs := testutil.ToFloat64(metrics.Errors.WithLabelValues(metrics.ErrorTypeSubmitterProgramError)) - programErrsBefore
diff --git a/e2e/sdk_device_telemetry_test.go b/e2e/sdk_device_telemetry_test.go
index bce1344c50..d0eab29079 100644
--- a/e2e/sdk_device_telemetry_test.go
+++ b/e2e/sdk_device_telemetry_test.go
@@ -287,12 +287,16 @@ func TestE2E_SDK_Telemetry_DeviceLatencySamples(t *testing.T) {
 			Epoch:                        &epoch,
 			SamplingIntervalMicroseconds: 1000000,
 		})
-		require.NoError(t, err)
-		for _, msg := range res.Meta.LogMessages {
+		// The rejection comes back as an error rather than in the transaction metadata: the
+		// instruction finalized, but the program refused it. Reading it off res.Meta.Err used to be
+		// the only way to see that (malbeclabs/infra#1703).
+		var programErr *telemetry.ProgramError
+		require.ErrorAs(t, err, &programErr, "re-initializing an existing account should fail")
+		require.Nil(t, res)
+		for _, msg := range programErr.Logs {
 			log.Debug("solana log message", "msg", msg)
 		}
-		log.Debug("transaction error", "error", res.Meta.Err)
-		require.NotNil(t, res.Meta.Err, "transaction should fail")
+		require.ErrorContains(t, err, "Latency samples account already exists")
 	})
 
 	// Write more device latency samples.
diff --git a/e2e/sdk_internet_telemetry_test.go b/e2e/sdk_internet_telemetry_test.go
index ea23f4353c..6cc99d8fd2 100644
--- a/e2e/sdk_internet_telemetry_test.go
+++ b/e2e/sdk_internet_telemetry_test.go
@@ -226,12 +226,16 @@ func TestE2E_SDK_Telemetry_InternetLatencySamples(t *testing.T) {
 			Epoch:                        epoch,
 			SamplingIntervalMicroseconds: samplingIntervalMicroseconds,
 		})
-		require.NoError(t, err)
-		for _, msg := range res.Meta.LogMessages {
+		// The rejection comes back as an error rather than in the transaction metadata: the
+		// instruction finalized, but the program refused it. Reading it off res.Meta.Err used to be
+		// the only way to see that (malbeclabs/infra#1703).
+		var programErr *telemetry.ProgramError
+		require.ErrorAs(t, err, &programErr, "re-initializing an existing account should fail")
+		require.Nil(t, res)
+		for _, msg := range programErr.Logs {
 			log.Debug("solana log message", "msg", msg)
 		}
-		log.Debug("transaction error", "error", res.Meta.Err)
-		require.NotNil(t, res.Meta.Err, "transaction should fail")
+		require.ErrorContains(t, err, "Latency samples account already exists")
 	})
 
 	// Write more internet latency samples.

From 911d56de8d00eb29adc580e1aee3f43bc78ed89b Mon Sep 17 00:00:00 2001
From: Greg Mitchell <greg@malbeclabs.com>
Date: Wed, 5 Aug 2026 19:02:20 +0000
Subject: [PATCH 4/4] telemetry: address review on program error surfacing

Five findings from review on 3393f70d.

The SDK's log filter was an allowlist described as a denylist: keeping
only "Program log:" lines dropped the reason a native program logs
through CPI, which for the system program is the line that says an agent
could not fund the account it was creating. It now drops the runtime's
own bookkeeping and keeps everything else.

Custom error codes arriving on the finalized path skipped the sentinel
mapping, so a caller's account-full and missing-account handling worked
on the preflight side of a condition and not the other. ProgramError
gains CustomErrorCode(), and the write methods map 1006 and 1011 onto
the same sentinels preflight returns.

The internet-latency collector re-sent a partition from index 0 on every
attempt and requeued all of it on failure, so a partial write appended
its earlier batches again. It now threads a written count the way the
device submitter has since #4145. Its per-batch debug log also reports
the batch size rather than the whole partition, which slicing would
otherwise have made the remainder.

The retries-exhausted assertion is now a log assertion:
TestSubmitter_RetainsEverySampleAcrossTheStalenessBound drives that
package-level counter from a sibling parallel test, so a zero delta was
racy. And the new metric's comment claimed it was distinct from the
write/init failure types when a rejected init increments both.
---
 CHANGELOG.md                                  |  3 +
 .../internal/exporter/submitter.go            | 38 ++++++----
 .../internal/exporter/submitter_test.go       | 52 +++++++++++++
 .../telemetry/internal/metrics/metrics.go     |  6 +-
 .../internal/telemetry/submitter_test.go      | 12 +--
 smartcontract/sdk/go/telemetry/client.go      | 29 ++++++++
 .../sdk/go/telemetry/client_device_test.go    | 72 ++++++++++++++++++
 smartcontract/sdk/go/telemetry/executor.go    | 73 ++++++++++++++++++-
 .../sdk/go/telemetry/executor_test.go         | 69 ++++++++++++++++++
 9 files changed, 329 insertions(+), 25 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index add50e358f..87af8fb96b 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -12,6 +12,9 @@ All notable changes to this project will be documented in this file.
   - `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)
diff --git a/controlplane/internet-latency-collector/internal/exporter/submitter.go b/controlplane/internet-latency-collector/internal/exporter/submitter.go
index 02126034a6..2fa133903c 100644
--- a/controlplane/internet-latency-collector/internal/exporter/submitter.go
+++ b/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
diff --git a/controlplane/internet-latency-collector/internal/exporter/submitter_test.go b/controlplane/internet-latency-collector/internal/exporter/submitter_test.go
index 5987dfc84b..c6d947126d 100644
--- a/controlplane/internet-latency-collector/internal/exporter/submitter_test.go
+++ b/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 {
diff --git a/controlplane/telemetry/internal/metrics/metrics.go b/controlplane/telemetry/internal/metrics/metrics.go
index c02a59ec6d..deab84c8ad 100644
--- a/controlplane/telemetry/internal/metrics/metrics.go
+++ b/controlplane/telemetry/internal/metrics/metrics.go
@@ -44,8 +44,10 @@ const (
 	ErrorTypeSubmitterBufferFull  = "submitter_buffer_full"
 	ErrorTypeSubmitterAccountFull = "submitter_account_full"
 	// ErrorTypeSubmitterProgramError counts submissions the telemetry program rejected onchain.
-	// Distinct from the write/init failure types, which also cover transient RPC trouble: this one
-	// only fires on a rejection that will recur until something changes onchain or in config.
+	// 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.
diff --git a/controlplane/telemetry/internal/telemetry/submitter_test.go b/controlplane/telemetry/internal/telemetry/submitter_test.go
index 7f93de3123..86e5fcd451 100644
--- a/controlplane/telemetry/internal/telemetry/submitter_test.go
+++ b/controlplane/telemetry/internal/telemetry/submitter_test.go
@@ -1264,8 +1264,8 @@ func TestAgentTelemetry_Submitter(t *testing.T) {
 	})
 
 	// The chi-dn-dzd4 case (malbeclabs/infra#1703): the agent key is not the device's
-	// metrics_publisher, so the program rejects the init. Deliberately not parallel: the assertions
-	// are exact deltas on package-level prometheus counters.
+	// metrics_publisher, so the program rejects the init. Deliberately not parallel: the
+	// program-error delta is on a package-level prometheus counter.
 	t.Run("does_not_retry_a_submission_the_program_rejected", func(t *testing.T) {
 		var logs bytes.Buffer
 		log := slog.New(slog.NewTextHandler(&logs, &slog.HandlerOptions{Level: slog.LevelWarn}))
@@ -1305,7 +1305,6 @@ func TestAgentTelemetry_Submitter(t *testing.T) {
 		require.NoError(t, err)
 
 		programErrsBefore := testutil.ToFloat64(metrics.Errors.WithLabelValues(metrics.ErrorTypeSubmitterProgramError))
-		exhaustedBefore := testutil.ToFloat64(metrics.Errors.WithLabelValues(metrics.ErrorTypeSubmitterRetriesExhausted))
 
 		s.Tick(context.Background())
 
@@ -1316,10 +1315,11 @@ func TestAgentTelemetry_Submitter(t *testing.T) {
 		programErrs := testutil.ToFloat64(metrics.Errors.WithLabelValues(metrics.ErrorTypeSubmitterProgramError)) - programErrsBefore
 		assert.Equal(t, float64(1), programErrs, "program-error counter should increment once")
 
-		exhausted := testutil.ToFloat64(metrics.Errors.WithLabelValues(metrics.ErrorTypeSubmitterRetriesExhausted)) - exhaustedBefore
-		assert.Equal(t, float64(0), exhausted, "retries were not exhausted, they were skipped")
-
 		out := logs.String()
+		// Asserted on the log rather than a delta of submitter_retries_exhausted: that counter is
+		// package-level and TestSubmitter_RetainsEverySampleAcrossTheStalenessBound drives it from a
+		// sibling parallel test, so a zero delta there would be racy.
+		assert.NotContains(t, out, "Submission failed after all retries", "the attempts should be skipped, not exhausted")
 		assert.Contains(t, out, "Submission rejected by the telemetry program")
 		assert.Contains(t, out, "is not authorized for origin device", "the reason the program gave has to reach the log")
 
diff --git a/smartcontract/sdk/go/telemetry/client.go b/smartcontract/sdk/go/telemetry/client.go
index b4bcd4e1fe..71028abbd2 100644
--- a/smartcontract/sdk/go/telemetry/client.go
+++ b/smartcontract/sdk/go/telemetry/client.go
@@ -286,6 +286,29 @@ func (c *Client) InitializeDeviceLatencySamples(
 	return sig, res, nil
 }
 
+// sentinelForProgramError maps a rejection that reached execution onto the sentinel its preflight
+// equivalent returns. Preflight catches most of these, but a transaction that simulated cleanly and
+// then failed against the bank it landed on reports the same code through the finalized transaction
+// instead, and callers should not have to know which side of preflight a rejection came from.
+// Returns nil when the error is not a program rejection, or carries a code with no sentinel.
+func sentinelForProgramError(err error) error {
+	var programErr *ProgramError
+	if !errors.As(err, &programErr) {
+		return nil
+	}
+	code, ok := programErr.CustomErrorCode()
+	if !ok {
+		return nil
+	}
+	switch code {
+	case InstructionErrorAccountDoesNotExist:
+		return ErrAccountNotFound
+	case InstructionErrorAccountSamplesAccountFull:
+		return ErrSamplesAccountFull
+	}
+	return nil
+}
+
 func (c *Client) WriteDeviceLatencySamples(
 	ctx context.Context,
 	config WriteDeviceLatencySamplesInstructionConfig,
@@ -326,6 +349,9 @@ func (c *Client) WriteDeviceLatencySamples(
 				}
 			}
 		}
+		if sentinel := sentinelForProgramError(err); sentinel != nil {
+			return solana.Signature{}, nil, sentinel
+		}
 		return solana.Signature{}, nil, fmt.Errorf("failed to execute instruction: %w", err)
 	}
 
@@ -432,6 +458,9 @@ func (c *Client) WriteInternetLatencySamples(
 				}
 			}
 		}
+		if sentinel := sentinelForProgramError(err); sentinel != nil {
+			return solana.Signature{}, nil, sentinel
+		}
 		return solana.Signature{}, nil, fmt.Errorf("failed to execute instruction: %w", err)
 	}
 
diff --git a/smartcontract/sdk/go/telemetry/client_device_test.go b/smartcontract/sdk/go/telemetry/client_device_test.go
index 14734e7433..5ba430cf85 100644
--- a/smartcontract/sdk/go/telemetry/client_device_test.go
+++ b/smartcontract/sdk/go/telemetry/client_device_test.go
@@ -982,6 +982,78 @@ func TestSDK_Telemetry_Client_WriteDeviceLatencySamples_CustomInstructionErrorSa
 	require.Nil(t, tx)
 }
 
+// A write that simulates cleanly and then fails against the bank it lands on reports the same codes
+// through the finalized transaction rather than a preflight RPCError. Those have to reach the same
+// sentinels, or the caller's account-full and missing-account handling only works on the preflight
+// side of the same condition (malbeclabs/infra#1703).
+func TestSDK_Telemetry_Client_WriteDeviceLatencySamples_FinalizedCustomErrorsMapToSentinels(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name string
+		code int
+		want error
+	}{
+		{"samples account full", telemetry.InstructionErrorAccountSamplesAccountFull, telemetry.ErrSamplesAccountFull},
+		{"account does not exist", telemetry.InstructionErrorAccountDoesNotExist, telemetry.ErrAccountNotFound},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			signer := solana.NewWallet().PrivateKey
+			programID := solana.NewWallet().PublicKey()
+
+			mockRPC := &mockRPCClient{
+				GetLatestBlockhashFunc: func(_ context.Context, _ solanarpc.CommitmentType) (*solanarpc.GetLatestBlockhashResult, error) {
+					return &solanarpc.GetLatestBlockhashResult{
+						Value: &solanarpc.LatestBlockhashResult{
+							Blockhash: solana.MustHashFromBase58("5NzX7jrPWeTkGsDnVnszdEa7T3Yyr3nSgyc78z3CwjWQ"),
+						},
+					}, nil
+				},
+				SendTransactionWithOptsFunc: func(_ context.Context, tx *solana.Transaction, _ solanarpc.TransactionOpts) (solana.Signature, error) {
+					return tx.Signatures[0], nil
+				},
+				GetSignatureStatusesFunc: func(context.Context, bool, ...solana.Signature) (*solanarpc.GetSignatureStatusesResult, error) {
+					return &solanarpc.GetSignatureStatusesResult{
+						Value: []*solanarpc.SignatureStatusesResult{
+							{
+								ConfirmationStatus: solanarpc.ConfirmationStatusFinalized,
+								Err: map[string]any{
+									"InstructionError": []any{
+										0,
+										map[string]any{"Custom": json.Number(strconv.Itoa(tt.code))},
+									},
+								},
+							},
+						},
+					}, nil
+				},
+				GetTransactionFunc: func(context.Context, solana.Signature, *solanarpc.GetTransactionOpts) (*solanarpc.GetTransactionResult, error) {
+					return &solanarpc.GetTransactionResult{Meta: &solanarpc.TransactionMeta{}}, nil
+				},
+			}
+
+			client := telemetry.New(slog.Default(), mockRPC, &signer, programID)
+
+			sig, tx, err := client.WriteDeviceLatencySamples(context.Background(), telemetry.WriteDeviceLatencySamplesInstructionConfig{
+				AgentPK:                    signer.PublicKey(),
+				OriginDevicePK:             solana.NewWallet().PublicKey(),
+				TargetDevicePK:             solana.NewWallet().PublicKey(),
+				LinkPK:                     solana.NewWallet().PublicKey(),
+				Epoch:                      ptr(uint64(42)),
+				StartTimestampMicroseconds: 1_600_000_000,
+				Samples:                    []uint32{10},
+			})
+			require.ErrorIs(t, err, tt.want)
+			require.Equal(t, solana.Signature{}, sig)
+			require.Nil(t, tx)
+		})
+	}
+}
+
 func TestSDK_Telemetry_Client_GetDeviceLatencySamplesHeader_HappyPath(t *testing.T) {
 	t.Parallel()
 
diff --git a/smartcontract/sdk/go/telemetry/executor.go b/smartcontract/sdk/go/telemetry/executor.go
index 7e5f429571..fbe424308e 100644
--- a/smartcontract/sdk/go/telemetry/executor.go
+++ b/smartcontract/sdk/go/telemetry/executor.go
@@ -2,9 +2,11 @@ package telemetry
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"log/slog"
+	"math"
 	"strings"
 	"time"
 
@@ -44,18 +46,83 @@ func (e *ProgramError) Error() string {
 // ProgramLogMessages returns the program's own log lines with the runtime's invoke/consumed/success
 // boilerplate and the instruction-name echo removed. These are the lines that say why the program
 // rejected the instruction, e.g. "Agent <pubkey> is not authorized for origin device <pubkey>".
+//
+// Everything the runtime did not write is kept, not just the "Program log:" lines: a native program
+// reached through CPI logs its reason unprefixed (the system program's "Transfer: insufficient
+// lamports 0, need 890880" when an agent cannot fund the account it is creating), and that line is
+// often the only one that explains the failure.
 func (e *ProgramError) ProgramLogMessages() []string {
 	var msgs []string
 	for _, line := range e.Logs {
-		msg, ok := strings.CutPrefix(line, "Program log: ")
-		if !ok || strings.HasPrefix(msg, "Instruction: ") {
+		if msg, ok := strings.CutPrefix(line, "Program log: "); ok {
+			// The instruction echo names what was attempted, which the caller already knows.
+			if strings.HasPrefix(msg, "Instruction: ") {
+				continue
+			}
+			msgs = append(msgs, msg)
+			continue
+		}
+		// Runtime bookkeeping: "Program <id> invoke [1]", "... success", "... failed: <err>",
+		// "... consumed N of M compute units", "Program data:", "Program return:".
+		if strings.HasPrefix(line, "Program ") {
 			continue
 		}
-		msgs = append(msgs, msg)
+		msgs = append(msgs, line)
 	}
 	return msgs
 }
 
+// CustomErrorCode returns the program's error code from a transaction error shaped as
+// {"InstructionError": [index, {"Custom": code}]}, which is how the ledger reports a TelemetryError.
+// The second return is false for any other shape, including a runtime error that carries no program
+// code at all.
+func (e *ProgramError) CustomErrorCode() (uint32, bool) {
+	errMap, ok := e.Err.(map[string]any)
+	if !ok {
+		return 0, false
+	}
+	instructionErr, ok := errMap["InstructionError"].([]any)
+	if !ok || len(instructionErr) != 2 {
+		return 0, false
+	}
+	custom, ok := instructionErr[1].(map[string]any)
+	if !ok {
+		return 0, false
+	}
+	raw, ok := custom["Custom"]
+	if !ok {
+		return 0, false
+	}
+	// Which numeric type the code arrives as depends on the JSON decoder the RPC client was built
+	// with, so accept the ones that reach us rather than pinning one.
+	var code int64
+	switch v := raw.(type) {
+	case json.Number:
+		parsed, err := v.Int64()
+		if err != nil {
+			return 0, false
+		}
+		code = parsed
+	case float64:
+		code = int64(v)
+	case int64:
+		code = v
+	case int:
+		code = int64(v)
+	case uint64:
+		if v > math.MaxUint32 {
+			return 0, false
+		}
+		return uint32(v), true
+	default:
+		return 0, false
+	}
+	if code < 0 || code > math.MaxUint32 {
+		return 0, false
+	}
+	return uint32(code), true
+}
+
 type executor struct {
 	log                   *slog.Logger
 	rpc                   RPCClient
diff --git a/smartcontract/sdk/go/telemetry/executor_test.go b/smartcontract/sdk/go/telemetry/executor_test.go
index 8787bdeba9..b73281e5fb 100644
--- a/smartcontract/sdk/go/telemetry/executor_test.go
+++ b/smartcontract/sdk/go/telemetry/executor_test.go
@@ -2,6 +2,7 @@ package telemetry_test
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"testing"
 	"time"
@@ -401,6 +402,74 @@ func TestSDK_Telemetry_Executor_FinalizedWithProgramError(t *testing.T) {
 	}
 }
 
+func TestSDK_Telemetry_ProgramError_CustomErrorCode(t *testing.T) {
+	t.Parallel()
+
+	customErr := func(code any) map[string]any {
+		return map[string]any{"InstructionError": []any{0, map[string]any{"Custom": code}}}
+	}
+
+	tests := []struct {
+		name string
+		err  any
+		want uint32
+		ok   bool
+	}{
+		// Which numeric type the code arrives as depends on the decoder behind the RPC client.
+		{name: "json.Number", err: customErr(json.Number("1001")), want: 1001, ok: true},
+		{name: "float64", err: customErr(float64(1006)), want: 1006, ok: true},
+		{name: "int", err: customErr(1011), want: 1011, ok: true},
+		{name: "uint64", err: customErr(uint64(1010)), want: 1010, ok: true},
+		{name: "not a custom error", err: map[string]any{"InstructionError": []any{0, "InvalidAccountData"}}},
+		{name: "runtime error with no instruction error", err: "BlockhashNotFound"},
+		{name: "nil", err: nil},
+		{name: "negative code", err: customErr(float64(-1))},
+		{name: "code beyond uint32", err: customErr(json.Number("4294967296"))},
+		{name: "unparseable code", err: customErr(json.Number("not-a-number"))},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			code, ok := (&telemetry.ProgramError{Err: tt.err}).CustomErrorCode()
+			require.Equal(t, tt.ok, ok)
+			require.Equal(t, tt.want, code)
+		})
+	}
+}
+
+// The reason a failure happened is not always a "Program log:" line. A native program reached
+// through CPI — the system program, when an agent cannot fund the account it is creating — logs it
+// unprefixed, and dropping those lines would leave only an opaque error code.
+func TestSDK_Telemetry_ProgramError_ProgramLogMessages(t *testing.T) {
+	t.Parallel()
+
+	programErr := &telemetry.ProgramError{
+		Err: map[string]any{"InstructionError": []any{0, map[string]any{"Custom": 1}}},
+		Logs: []string{
+			"Program TeLeMetRy1111111111111111111111111111111111 invoke [1]",
+			"Program log: Instruction: InitializeDeviceLatencySamples",
+			"Program log: Processing InitializeDeviceLatencySamples",
+			"Program 11111111111111111111111111111111 invoke [2]",
+			"Transfer: insufficient lamports 0, need 890880",
+			"Program 11111111111111111111111111111111 failed: custom program error: 0x1",
+			"Program TeLeMetRy1111111111111111111111111111111111 consumed 4242 of 200000 compute units",
+			"Program TeLeMetRy1111111111111111111111111111111111 failed: custom program error: 0x1",
+		},
+	}
+
+	require.Equal(t, []string{
+		"Processing InitializeDeviceLatencySamples",
+		"Transfer: insufficient lamports 0, need 890880",
+	}, programErr.ProgramLogMessages())
+
+	// And the same lines reach anyone who only prints the error.
+	require.Contains(t, programErr.Error(), "insufficient lamports")
+	require.NotContains(t, programErr.Error(), "compute units")
+	require.NotContains(t, programErr.Error(), "Instruction: InitializeDeviceLatencySamples")
+}
+
 func TestSDK_Telemetry_Executor_FinalizedButMissingTransactionMeta(t *testing.T) {
 	t.Parallel()