device/telemetry: use ledger epoch (#1004)

## Summary of Changes
- Update the device telemetry agent to use epoch from the DZ ledger
instead of wallclock epoch
- Resolves #1002 
- Resolves #1003
- The telemetry data/dashboard API was not updated to support this yet,
that can happen separately

## Testing Verification
- Existing tests updated to match the change

Author: snormore

controlplane/telemetry/cmd/telemetry/main.go

@@ -261,6 +261,13 @@ func main() {
 		TWAMPReflector:         reflector,
 		PeerDiscovery:          peerDiscovery,
 		TelemetryProgramClient: sdktelemetry.New(log, rpcClient, &keypair, telemetryProgramID),
+		GetCurrentEpochFunc: func(ctx context.Context) (uint64, error) {
+			epochInfo, err := rpcClient.GetEpochInfo(ctx, solanarpc.CommitmentFinalized)
+			if err != nil {
+				return 0, err
+			}
+			return epochInfo.Epoch, nil
+		},
 	})
 	if err != nil {
 		log.Error("failed to create telemetry collector", "error", err)

controlplane/telemetry/pkg/data/epoch.go renamed to controlplane/telemetry/internal/data/epoch.go

@@ -7,7 +7,6 @@ import (
 	"sort"
 	"time"
 
-	"github.com/malbeclabs/doublezero/controlplane/telemetry/pkg/data"
 	"github.com/malbeclabs/doublezero/smartcontract/sdk/go/telemetry"
 )
 
@@ -77,7 +76,7 @@ func (p *provider) GetCircuitLatenciesForEpoch(ctx context.Context, circuitCode
 	samples := enrichSamplesWithTimestamps(account.Samples, account.StartTimestampMicroseconds, account.SamplingIntervalMicroseconds)
 
 	// If the epoch is sufficiently in the past, cache for much longer.
-	currentEpoch := data.DeriveEpoch(time.Now())
+	currentEpoch := DeriveEpoch(time.Now())
 	ttl := p.cfg.CurrentEpochLatenciesCacheTTL
 	if epoch < currentEpoch-1 {
 		ttl = p.cfg.HistoricEpochLatenciesCacheTTL
@@ -88,8 +87,8 @@ func (p *provider) GetCircuitLatenciesForEpoch(ctx context.Context, circuitCode
 }
 
 func (p *provider) GetCircuitLatencies(ctx context.Context, circuitCode string, from, to time.Time) ([]CircuitLatencySample, error) {
-	startEpoch := data.DeriveEpoch(from)
-	endEpoch := data.DeriveEpoch(to)
+	startEpoch := DeriveEpoch(from)
+	endEpoch := DeriveEpoch(to)
 
 	var latencies []CircuitLatencySample
 

controlplane/telemetry/internal/data/latencies.go

@@ -10,7 +10,6 @@ import (
 
 	"github.com/gagliardetto/solana-go"
 	"github.com/malbeclabs/doublezero/controlplane/telemetry/internal/data"
-	datapkg "github.com/malbeclabs/doublezero/controlplane/telemetry/pkg/data"
 	"github.com/malbeclabs/doublezero/smartcontract/sdk/go/serviceability"
 	"github.com/malbeclabs/doublezero/smartcontract/sdk/go/telemetry"
 	"github.com/stretchr/testify/assert"
@@ -39,7 +38,7 @@ func TestTelemetry_Data_Provider_GetCircuitLatencies(t *testing.T) {
 		}, defaultCircuit())
 
 		ctx := context.Background()
-		epoch := datapkg.DeriveEpoch(time.Now())
+		epoch := data.DeriveEpoch(time.Now())
 
 		first, err := provider.GetCircuitLatenciesForEpoch(ctx, "A → B (L1)", epoch)
 		require.NoError(t, err)
@@ -57,7 +56,7 @@ func TestTelemetry_Data_Provider_GetCircuitLatencies(t *testing.T) {
 			return nil, telemetry.ErrAccountNotFound
 		}, defaultCircuit())
 
-		epoch := datapkg.DeriveEpoch(time.Now())
+		epoch := data.DeriveEpoch(time.Now())
 		latencies, err := provider.GetCircuitLatenciesForEpoch(context.Background(), "A → B (L1)", epoch)
 		assert.ErrorIs(t, err, telemetry.ErrAccountNotFound)
 		assert.Empty(t, latencies)
@@ -66,8 +65,7 @@ func TestTelemetry_Data_Provider_GetCircuitLatencies(t *testing.T) {
 	t.Run("GetCircuitLatencies filters by time", func(t *testing.T) {
 		t.Parallel()
 
-		now := time.Now()
-		sampleTime := now.Add(-5 * time.Minute).UTC()
+		sampleTime := time.Date(2023, 1, 1, 12, 0, 0, 0, time.UTC)
 		sampleMicros := uint64(sampleTime.UnixMicro())
 
 		provider := newTestProvider(t, func(epoch uint64) (*telemetry.DeviceLatencySamples, error) {
@@ -80,8 +78,8 @@ func TestTelemetry_Data_Provider_GetCircuitLatencies(t *testing.T) {
 			}, nil
 		}, defaultCircuit())
 
-		from := now.Add(-10 * time.Minute)
-		to := now
+		from := time.Date(2023, 1, 1, 11, 55, 0, 0, time.UTC)
+		to := time.Date(2023, 1, 1, 12, 5, 0, 0, time.UTC)
 		latencies, err := provider.GetCircuitLatencies(context.Background(), "A → B (L1)", from, to)
 		require.NoError(t, err)
 		require.Len(t, latencies, 1)
@@ -116,7 +114,7 @@ func TestTelemetry_Data_Provider_GetCircuitLatencies(t *testing.T) {
 	t.Run("Downsampled returns multiple buckets when maxPoints > 1", func(t *testing.T) {
 		t.Parallel()
 
-		now := time.Now().UTC()
+		now := time.Date(2023, 1, 1, 12, 0, 0, 0, time.UTC)
 		tsMicros := uint64(now.UnixMicro())
 
 		provider := newTestProvider(t, func(epoch uint64) (*telemetry.DeviceLatencySamples, error) {
@@ -149,7 +147,7 @@ func TestTelemetry_Data_Provider_GetCircuitLatencies(t *testing.T) {
 			}, nil
 		}, defaultCircuit())
 
-		now := time.Now()
+		now := time.Date(2023, 1, 1, 12, 0, 0, 0, time.UTC)
 		stats, err := provider.GetCircuitLatenciesDownsampled(context.Background(), "A → B (L1)", now, now.Add(1*time.Minute), 1, data.UnitMicrosecond)
 		require.NoError(t, err)
 		assert.Len(t, stats, 0)
@@ -162,7 +160,7 @@ func TestTelemetry_Data_Provider_GetCircuitLatencies(t *testing.T) {
 			return &telemetry.DeviceLatencySamples{}, nil
 		}, defaultCircuit())
 
-		now := time.Now()
+		now := time.Date(2023, 1, 1, 12, 0, 0, 0, time.UTC)
 		stats, err := provider.GetCircuitLatenciesDownsampled(t.Context(), "A → B (L1)", now, now.Add(1*time.Second), 1, "invalid")
 		require.Error(t, err)
 		assert.Empty(t, stats)

controlplane/telemetry/internal/data/latencies_test.go

@@ -46,21 +46,27 @@ func New(log *slog.Logger, cfg Config) (*Collector, error) {
 		buffer:    buffer,
 	}
 
-	c.submitter = NewSubmitter(log, &SubmitterConfig{
+	var err error
+	c.submitter, err = NewSubmitter(log, &SubmitterConfig{
 		Interval:           cfg.SubmissionInterval,
 		Buffer:             buffer,
 		MetricsPublisherPK: cfg.MetricsPublisherPK,
 		ProbeInterval:      cfg.ProbeInterval,
 		ProgramClient:      cfg.TelemetryProgramClient,
+		GetCurrentEpoch:    cfg.GetCurrentEpochFunc,
 	})
+	if err != nil {
+		return nil, fmt.Errorf("failed to create submitter: %w", err)
+	}
 
 	c.pinger = NewPinger(log, &PingerConfig{
-		LocalDevicePK: cfg.LocalDevicePK,
-		Interval:      cfg.ProbeInterval,
-		ProbeTimeout:  cfg.TWAMPSenderTimeout,
-		Peers:         cfg.PeerDiscovery,
-		Buffer:        buffer,
-		GetSender:     c.getOrCreateSender,
+		LocalDevicePK:   cfg.LocalDevicePK,
+		Interval:        cfg.ProbeInterval,
+		ProbeTimeout:    cfg.TWAMPSenderTimeout,
+		Peers:           cfg.PeerDiscovery,
+		Buffer:          buffer,
+		GetSender:       c.getOrCreateSender,
+		GetCurrentEpoch: cfg.GetCurrentEpochFunc,
 	})
 
 	return c, nil

controlplane/telemetry/internal/telemetry/collector.go

@@ -75,31 +75,31 @@ func TestAgentTelemetry_Collector(t *testing.T) {
 		link2_1 := stringToPubkey("link2-1")
 		link2_3 := stringToPubkey("link2-3")
 
-		ts := time.Now()
+		epoch := uint64(100)
 		originDevice1Link1_2Key := telemetry.AccountKey{
 			OriginDevicePK: device1PK,
 			TargetDevicePK: device2PK,
 			LinkPK:         link1_2,
-			Epoch:          telemetry.DeriveEpoch(ts),
+			Epoch:          epoch,
 		}
 		originDevice1Link1_3Key := telemetry.AccountKey{
 			OriginDevicePK: device1PK,
 			TargetDevicePK: device3PK,
 			LinkPK:         link1_3,
-			Epoch:          telemetry.DeriveEpoch(ts),
+			Epoch:          epoch,
 		}
 
 		originDevice2Link2_1Key := telemetry.AccountKey{
 			OriginDevicePK: device2PK,
 			TargetDevicePK: device1PK,
 			LinkPK:         link2_1,
-			Epoch:          telemetry.DeriveEpoch(ts),
+			Epoch:          epoch,
 		}
 		originDevice2Link2_3Key := telemetry.AccountKey{
 			OriginDevicePK: device2PK,
 			TargetDevicePK: device3PK,
 			LinkPK:         link2_3,
-			Epoch:          telemetry.DeriveEpoch(ts),
+			Epoch:          epoch,
 		}
 
 		telemetryProgram1 := newMemoryTelemetryProgramClient()
@@ -273,6 +273,9 @@ func TestAgentTelemetry_Collector(t *testing.T) {
 			TWAMPReflector:         reflector,
 			PeerDiscovery:          peerDiscovery,
 			TelemetryProgramClient: telemetryProgram,
+			GetCurrentEpochFunc: func(ctx context.Context) (uint64, error) {
+				return 100, nil
+			},
 		})
 		require.NoError(t, err)
 
@@ -283,7 +286,7 @@ func TestAgentTelemetry_Collector(t *testing.T) {
 			require.NoError(t, collector.Run(ctx))
 		}()
 
-		epoch := telemetry.DeriveEpoch(time.Now())
+		epoch := uint64(100)
 
 		accountKey := telemetry.AccountKey{
 			OriginDevicePK: devicePK,
@@ -393,11 +396,13 @@ func TestAgentTelemetry_Collector(t *testing.T) {
 			require.NoError(t, collector.Run(ctx))
 		}()
 
+		epoch := uint64(100)
+
 		key := telemetry.AccountKey{
 			OriginDevicePK: devicePK,
 			TargetDevicePK: peerPK,
 			LinkPK:         linkPK,
-			Epoch:          telemetry.DeriveEpoch(time.Now()),
+			Epoch:          epoch,
 		}
 
 		// Wait for multiple samples
@@ -434,6 +439,9 @@ func newTestCollector(t *testing.T, log *slog.Logger, localDevicePK solana.Publi
 		TWAMPReflector:         reflector,
 		PeerDiscovery:          peerDiscovery,
 		TelemetryProgramClient: telemetryProgramClient,
+		GetCurrentEpochFunc: func(ctx context.Context) (uint64, error) {
+			return 100, nil
+		},
 	})
 	require.NoError(t, err)
 

controlplane/telemetry/internal/telemetry/collector_test.go

@@ -1,6 +1,7 @@
 package telemetry
 
 import (
+	"context"
 	"errors"
 	"time"
 
@@ -15,6 +16,9 @@ type Config struct {
 	// PeerDiscovery is the configured peer discovery implementation.
 	PeerDiscovery PeerDiscovery
 
+	// GetCurrentEpochFunc is the function to get the current epoch.
+	GetCurrentEpochFunc func(ctx context.Context) (uint64, error)
+
 	// TelemetryProgramClient is the client to the telemetry program.
 	TelemetryProgramClient TelemetryProgramClient
 
@@ -41,6 +45,9 @@ func (c *Config) Validate() error {
 	if c.PeerDiscovery == nil {
 		return errors.New("peer discovery is required")
 	}
+	if c.GetCurrentEpochFunc == nil {
+		return errors.New("get current epoch is required")
+	}
 	if c.LocalDevicePK.IsZero() {
 		return errors.New("local device pubkey is required")
 	}

controlplane/telemetry/internal/telemetry/config.go

@@ -111,7 +111,7 @@ func (c *memoryTelemetryProgramClient) InitializeDeviceLatencySamples(ctx contex
 		OriginDevicePK: config.OriginDevicePK,
 		TargetDevicePK: config.TargetDevicePK,
 		LinkPK:         config.LinkPK,
-		Epoch:          config.Epoch,
+		Epoch:          *config.Epoch,
 	}
 
 	c.accounts[accountKey] = make([]telemetry.Sample, 0)
@@ -127,7 +127,7 @@ func (c *memoryTelemetryProgramClient) WriteDeviceLatencySamples(ctx context.Con
 		OriginDevicePK: config.OriginDevicePK,
 		TargetDevicePK: config.TargetDevicePK,
 		LinkPK:         config.LinkPK,
-		Epoch:          config.Epoch,
+		Epoch:          *config.Epoch,
 	}
 
 	if _, ok := c.accounts[accountKey]; !ok {