Skip to content

Commit eb40889

Browse files
snormoresnormore
committed
telemetry: fix timestamp index init failure not clearing PK on retry
When InitializeTimestampIndex failed, both submitters logged "writes will proceed without it" but retried with TimestampIndexPK still set, causing the write to fail again. Now nil out TimestampIndexPK on init failure so the retry actually proceeds without the timestamp index. Also remove the unused TimestampIndexFull error variant and renumber TimestampIndexAccountDoesNotExist to 1018.
1 parent 7cf23bf commit eb40889

6 files changed

Lines changed: 199 additions & 5 deletions

File tree

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

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -154,6 +154,7 @@ func (s *Submitter) SubmitSamples(ctx context.Context, partitionKey PartitionKey
154154
_, _, err = s.cfg.Telemetry.InitializeTimestampIndex(ctx, samplesPDA)
155155
if err != nil {
156156
log.Warn("Failed to initialize timestamp index, writes will proceed without it", "error", err)
157+
writeConfig.TimestampIndexPK = nil
157158
}
158159
_, _, err = s.cfg.Telemetry.WriteInternetLatencySamples(ctx, writeConfig)
159160
if err != nil {
@@ -185,6 +186,7 @@ func (s *Submitter) SubmitSamples(ctx context.Context, partitionKey PartitionKey
185186
_, _, err = s.cfg.Telemetry.InitializeTimestampIndex(ctx, samplesPDA)
186187
if err != nil {
187188
log.Warn("Failed to initialize timestamp index, writes will proceed without it", "error", err)
189+
writeConfig.TimestampIndexPK = nil
188190
}
189191
_, _, err = s.cfg.Telemetry.WriteInternetLatencySamples(ctx, writeConfig)
190192
if err != nil {

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

Lines changed: 98 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -623,6 +623,104 @@ func TestInternetLatency_Submitter(t *testing.T) {
623623
assert.Equal(t, int32(2), atomic.LoadInt32(&writeCalled), "should try write twice (before and after timestamp index init)")
624624
})
625625

626+
t.Run("clears_timestamp_index_pk_when_init_fails", func(t *testing.T) {
627+
t.Parallel()
628+
629+
log := logger.With("test", t.Name())
630+
631+
key := newTestPartitionKey()
632+
sample := newTestSample()
633+
634+
var writeCalled int32
635+
var retryTimestampIndexPK *solana.PublicKey
636+
telemetryProgram := &mockTelemetryProgramClient{
637+
WriteInternetLatencySamplesFunc: func(ctx context.Context, config sdktelemetry.WriteInternetLatencySamplesInstructionConfig) (solana.Signature, *solanarpc.GetTransactionResult, error) {
638+
n := atomic.AddInt32(&writeCalled, 1)
639+
if n == 1 {
640+
return solana.Signature{}, nil, sdktelemetry.ErrTimestampIndexNotFound
641+
}
642+
retryTimestampIndexPK = config.TimestampIndexPK
643+
return solana.Signature{}, nil, nil
644+
},
645+
InitializeTimestampIndexFunc: func(ctx context.Context, samplesAccountPK solana.PublicKey) (solana.Signature, *solanarpc.GetTransactionResult, error) {
646+
return solana.Signature{}, nil, errors.New("init failed")
647+
},
648+
}
649+
650+
buffer := buffer.NewMemoryPartitionedBuffer[exporter.PartitionKey, exporter.Sample](128)
651+
buffer.Add(key, sample)
652+
653+
submitter, err := exporter.NewSubmitter(log, &exporter.SubmitterConfig{
654+
OracleAgentPK: solana.NewWallet().PublicKey(),
655+
Interval: time.Hour,
656+
Buffer: buffer,
657+
Telemetry: telemetryProgram,
658+
MaxAttempts: 2,
659+
BackoffFunc: func(_ int) time.Duration { return 0 },
660+
EpochFinder: &mockEpochFinder{ApproximateAtTimeFunc: func(ctx context.Context, target time.Time) (uint64, error) {
661+
return key.Epoch, nil
662+
}},
663+
})
664+
require.NoError(t, err)
665+
666+
submitter.Tick(t.Context())
667+
668+
assert.Equal(t, int32(2), atomic.LoadInt32(&writeCalled), "should retry write after failed timestamp index init")
669+
assert.Nil(t, retryTimestampIndexPK, "retry write should have nil TimestampIndexPK after failed init")
670+
})
671+
672+
t.Run("clears_timestamp_index_pk_when_init_fails_on_new_account", func(t *testing.T) {
673+
t.Parallel()
674+
675+
log := logger.With("test", t.Name())
676+
677+
key := newTestPartitionKey()
678+
sample := newTestSample()
679+
680+
var writeCalled int32
681+
var retryTimestampIndexPK *solana.PublicKey
682+
telemetryProgram := &mockTelemetryProgramClient{
683+
WriteInternetLatencySamplesFunc: func(ctx context.Context, config sdktelemetry.WriteInternetLatencySamplesInstructionConfig) (solana.Signature, *solanarpc.GetTransactionResult, error) {
684+
n := atomic.AddInt32(&writeCalled, 1)
685+
if n == 1 {
686+
return solana.Signature{}, nil, sdktelemetry.ErrAccountNotFound
687+
}
688+
retryTimestampIndexPK = config.TimestampIndexPK
689+
return solana.Signature{}, nil, nil
690+
},
691+
InitializeInternetLatencySamplesFunc: func(ctx context.Context, config sdktelemetry.InitializeInternetLatencySamplesInstructionConfig) (solana.Signature, *solanarpc.GetTransactionResult, error) {
692+
return solana.Signature{}, nil, nil
693+
},
694+
InitializeTimestampIndexFunc: func(ctx context.Context, samplesAccountPK solana.PublicKey) (solana.Signature, *solanarpc.GetTransactionResult, error) {
695+
return solana.Signature{}, nil, errors.New("init failed")
696+
},
697+
}
698+
699+
buffer := buffer.NewMemoryPartitionedBuffer[exporter.PartitionKey, exporter.Sample](128)
700+
buffer.Add(key, sample)
701+
702+
submitter, err := exporter.NewSubmitter(log, &exporter.SubmitterConfig{
703+
OracleAgentPK: solana.NewWallet().PublicKey(),
704+
Interval: time.Hour,
705+
Buffer: buffer,
706+
Telemetry: telemetryProgram,
707+
MaxAttempts: 2,
708+
BackoffFunc: func(_ int) time.Duration { return 0 },
709+
EpochFinder: &mockEpochFinder{ApproximateAtTimeFunc: func(ctx context.Context, target time.Time) (uint64, error) {
710+
return key.Epoch, nil
711+
}},
712+
DataProviderSamplingIntervals: map[exporter.DataProviderName]time.Duration{
713+
key.DataProvider: time.Second,
714+
},
715+
})
716+
require.NoError(t, err)
717+
718+
submitter.Tick(t.Context())
719+
720+
assert.Equal(t, int32(2), atomic.LoadInt32(&writeCalled), "should retry write after failed timestamp index init")
721+
assert.Nil(t, retryTimestampIndexPK, "retry write should have nil TimestampIndexPK after failed init")
722+
})
723+
626724
t.Run("failed_retries_reinsert_at_front_preserving_order", func(t *testing.T) {
627725
t.Parallel()
628726

‎controlplane/telemetry/internal/telemetry/submitter.go‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -151,6 +151,7 @@ func (s *Submitter) SubmitSamples(ctx context.Context, partitionKey PartitionKey
151151
_, _, err = s.cfg.ProgramClient.InitializeTimestampIndex(ctx, samplesPDA)
152152
if err != nil {
153153
log.Warn("Failed to initialize timestamp index, writes will proceed without it", "error", err)
154+
writeConfig.TimestampIndexPK = nil
154155
}
155156
_, _, err = s.cfg.ProgramClient.WriteDeviceLatencySamples(ctx, writeConfig)
156157
if err != nil {
@@ -180,6 +181,7 @@ func (s *Submitter) SubmitSamples(ctx context.Context, partitionKey PartitionKey
180181
_, _, err = s.cfg.ProgramClient.InitializeTimestampIndex(ctx, samplesPDA)
181182
if err != nil {
182183
log.Warn("Failed to initialize timestamp index, writes will proceed without it", "error", err)
184+
writeConfig.TimestampIndexPK = nil
183185
}
184186
_, _, err = s.cfg.ProgramClient.WriteDeviceLatencySamples(ctx, writeConfig)
185187
if err != nil {

‎controlplane/telemetry/internal/telemetry/submitter_test.go‎

Lines changed: 95 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -749,6 +749,101 @@ func TestAgentTelemetry_Submitter(t *testing.T) {
749749
assert.Equal(t, int32(2), atomic.LoadInt32(&writeCalled), "should try write twice (before and after timestamp index init)")
750750
})
751751

752+
t.Run("clears_timestamp_index_pk_when_init_fails", func(t *testing.T) {
753+
t.Parallel()
754+
755+
log := log.With("test", t.Name())
756+
757+
key := newTestPartitionKey()
758+
sample := newTestSample()
759+
760+
var writeCalled int32
761+
var retryTimestampIndexPK *solana.PublicKey
762+
telemetryProgram := &mockTelemetryProgramClient{
763+
WriteDeviceLatencySamplesFunc: func(ctx context.Context, config sdktelemetry.WriteDeviceLatencySamplesInstructionConfig) (solana.Signature, *solanarpc.GetTransactionResult, error) {
764+
n := atomic.AddInt32(&writeCalled, 1)
765+
if n == 1 {
766+
return solana.Signature{}, nil, sdktelemetry.ErrTimestampIndexNotFound
767+
}
768+
retryTimestampIndexPK = config.TimestampIndexPK
769+
return solana.Signature{}, nil, nil
770+
},
771+
InitializeTimestampIndexFunc: func(ctx context.Context, samplesAccountPK solana.PublicKey) (solana.Signature, *solanarpc.GetTransactionResult, error) {
772+
return solana.Signature{}, nil, errors.New("init failed")
773+
},
774+
}
775+
776+
buffer := buffer.NewMemoryPartitionedBuffer[telemetry.PartitionKey, telemetry.Sample](1024)
777+
buffer.Add(key, sample)
778+
779+
submitter, err := telemetry.NewSubmitter(log, &telemetry.SubmitterConfig{
780+
Interval: time.Hour,
781+
Buffer: buffer,
782+
ProgramClient: telemetryProgram,
783+
MaxAttempts: 2,
784+
MaxConcurrency: 10,
785+
BackoffFunc: func(_ int) time.Duration { return 0 },
786+
GetCurrentEpoch: func(ctx context.Context) (uint64, error) {
787+
return 100, nil
788+
},
789+
})
790+
require.NoError(t, err)
791+
792+
submitter.Tick(context.Background())
793+
794+
assert.Equal(t, int32(2), atomic.LoadInt32(&writeCalled), "should retry write after failed timestamp index init")
795+
assert.Nil(t, retryTimestampIndexPK, "retry write should have nil TimestampIndexPK after failed init")
796+
})
797+
798+
t.Run("clears_timestamp_index_pk_when_init_fails_on_new_account", func(t *testing.T) {
799+
t.Parallel()
800+
801+
log := log.With("test", t.Name())
802+
803+
key := newTestPartitionKey()
804+
sample := newTestSample()
805+
806+
var writeCalled int32
807+
var retryTimestampIndexPK *solana.PublicKey
808+
telemetryProgram := &mockTelemetryProgramClient{
809+
WriteDeviceLatencySamplesFunc: func(ctx context.Context, config sdktelemetry.WriteDeviceLatencySamplesInstructionConfig) (solana.Signature, *solanarpc.GetTransactionResult, error) {
810+
n := atomic.AddInt32(&writeCalled, 1)
811+
if n == 1 {
812+
return solana.Signature{}, nil, sdktelemetry.ErrAccountNotFound
813+
}
814+
retryTimestampIndexPK = config.TimestampIndexPK
815+
return solana.Signature{}, nil, nil
816+
},
817+
InitializeDeviceLatencySamplesFunc: func(ctx context.Context, config sdktelemetry.InitializeDeviceLatencySamplesInstructionConfig) (solana.Signature, *solanarpc.GetTransactionResult, error) {
818+
return solana.Signature{}, nil, nil
819+
},
820+
InitializeTimestampIndexFunc: func(ctx context.Context, samplesAccountPK solana.PublicKey) (solana.Signature, *solanarpc.GetTransactionResult, error) {
821+
return solana.Signature{}, nil, errors.New("init failed")
822+
},
823+
}
824+
825+
buffer := buffer.NewMemoryPartitionedBuffer[telemetry.PartitionKey, telemetry.Sample](1024)
826+
buffer.Add(key, sample)
827+
828+
submitter, err := telemetry.NewSubmitter(log, &telemetry.SubmitterConfig{
829+
Interval: time.Hour,
830+
Buffer: buffer,
831+
ProgramClient: telemetryProgram,
832+
MaxAttempts: 2,
833+
MaxConcurrency: 10,
834+
BackoffFunc: func(_ int) time.Duration { return 0 },
835+
GetCurrentEpoch: func(ctx context.Context) (uint64, error) {
836+
return 100, nil
837+
},
838+
})
839+
require.NoError(t, err)
840+
841+
submitter.Tick(context.Background())
842+
843+
assert.Equal(t, int32(2), atomic.LoadInt32(&writeCalled), "should retry write after failed timestamp index init")
844+
assert.Nil(t, retryTimestampIndexPK, "retry write should have nil TimestampIndexPK after failed init")
845+
})
846+
752847
t.Run("failed_retries_reinsert_at_front_preserving_order", func(t *testing.T) {
753848
t.Parallel()
754849

‎smartcontract/programs/doublezero-telemetry/src/error.rs‎

Lines changed: 1 addition & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -37,10 +37,8 @@ pub enum TelemetryError {
3737
SameTargetAsOrigin = 1016,
3838
/// Write transaction contains no samples
3939
EmptyLatencySamples = 1017,
40-
/// Timestamp index account is full
41-
TimestampIndexFull = 1018,
4240
/// Timestamp index account does not exist
43-
TimestampIndexAccountDoesNotExist = 1019,
41+
TimestampIndexAccountDoesNotExist = 1018,
4442
}
4543

4644
impl From<TelemetryError> for ProgramError {
@@ -82,7 +80,6 @@ impl fmt::Display for TelemetryError {
8280
Self::DataProviderNameTooLong => write!(f, "Data provider name exceeds 32 bytes"),
8381
Self::SameTargetAsOrigin => write!(f, "Origin and target are the same exchange"),
8482
Self::EmptyLatencySamples => write!(f, "Write transaction contains no samples"),
85-
Self::TimestampIndexFull => write!(f, "Timestamp index account is full"),
8683
Self::TimestampIndexAccountDoesNotExist => {
8784
write!(f, "Timestamp index account does not exist")
8885
}

‎smartcontract/sdk/go/telemetry/constants.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@ const (
2525

2626
// InstructionErrorTimestampIndexAccountDoesNotExist is the error code that the telemetry
2727
// program returns when the timestamp index account does not exist.
28-
InstructionErrorTimestampIndexAccountDoesNotExist = 1019
28+
InstructionErrorTimestampIndexAccountDoesNotExist = 1018
2929

3030
// MaxSamplesPerBatch is the maximum number of samples that can be written in a single batch.
3131
//

0 commit comments

Comments
 (0)