device-health-oracle: add interface_counters activation criterion

CHANGELOG.md

@@ -9,6 +9,7 @@ All notable changes to this project will be documented in this file.
 ### Changes
 
 - Device Health Oracle
+  - Add `interface_counters` activation criterion to device-health-oracle to verify devices have recent interface counter data in ClickHouse before activation
   - Add `controller_success` activation criterion to device-health-oracle to verify devices have consistent controller call coverage over a configurable burn-in period by querying ClickHouse
 - Telemetry
   - Add `GET /device-link/agent-versions` endpoint to data-api and `agent-versions` subcommand to data-cli, exposing per-device telemetry agent version and commit from onchain `DeviceLatencySamplesHeader`

controlplane/device-health-oracle/cmd/device-health-oracle/main.go

@@ -145,15 +145,16 @@ func main() {
 
 		chClient, err := worker.NewClickHouseClient(chAddr, chDB, chUser, chPass, chTLSDisabled)
 		if err != nil {
-			log.Warn("ClickHouse connection failed, continuing without controller_success criterion", "addr", chAddr, "error", err)
+			log.Warn("ClickHouse connection failed, continuing without ClickHouse-based criteria", "addr", chAddr, "error", err)
 		} else {
 			defer chClient.Close()
 			log.Info("ClickHouse enabled", "addr", chAddr, "db", chDB, "user", chUser, "tls", !chTLSDisabled)
 			controllerSuccess := worker.NewControllerSuccessCriterion(chClient, log)
-			deviceCriteria = append(deviceCriteria, controllerSuccess)
+			interfaceCounters := worker.NewInterfaceCountersCriterion(chClient, log)
+			deviceCriteria = append(deviceCriteria, controllerSuccess, interfaceCounters)
 		}
 	} else {
-		log.Error("ClickHouse disabled (CLICKHOUSE_ADDR not set), no controller_success criterion")
+		log.Error("ClickHouse disabled (CLICKHOUSE_ADDR not set), no ClickHouse-based criteria")
 	}
 
 	deviceEvaluator := &worker.DeviceHealthEvaluator{

controlplane/device-health-oracle/internal/worker/clickhouse.go

@@ -19,7 +19,7 @@ type ControllerCallChecker interface {
 	Close() error
 }
 
-// ClickHouseClient wraps a ClickHouse connection for reading controller call data.
+// ClickHouseClient wraps a connection for reading device health data from ClickHouse.
 type ClickHouseClient struct {
 	conn clickhouse.Conn
 	db   string
@@ -80,6 +80,32 @@ func (c *ClickHouseClient) ControllerCallCoverage(ctx context.Context, devicePub
 	return int64(minutesWithCalls), nil
 }
 
+// InterfaceCountersChecker queries ClickHouse for device interface counter records.
+type InterfaceCountersChecker interface {
+	InterfaceCountersCoverage(ctx context.Context, devicePubkey string, start, end time.Time) (minutesWithRecords int64, err error)
+}
+
+// InterfaceCountersCoverage returns the number of distinct minutes in [start, end] that have
+// at least one fact_dz_device_interface_counters record for the given device.
+func (c *ClickHouseClient) InterfaceCountersCoverage(ctx context.Context, devicePubkey string, start, end time.Time) (int64, error) {
+	query := fmt.Sprintf(
+		`SELECT count(DISTINCT toStartOfMinute(event_ts)) AS minutes_with_records
+		 FROM "%s".fact_dz_device_interface_counters
+		 WHERE device_pk = ?
+		   AND event_ts >= ?
+		   AND event_ts <= ?`,
+		c.db,
+	)
+
+	var minutesWithRecords uint64
+	err := c.conn.QueryRow(ctx, query, devicePubkey, start, end).Scan(&minutesWithRecords)
+	if err != nil {
+		return 0, fmt.Errorf("clickhouse query: %w", err)
+	}
+
+	return int64(minutesWithRecords), nil
+}
+
 func (c *ClickHouseClient) Close() error {
 	return c.conn.Close()
 }

controlplane/device-health-oracle/internal/worker/clickhouse_test.go

@@ -98,6 +98,75 @@ func TestControllerCallCoverage_QuotesDatabaseName(t *testing.T) {
 	assert.Equal(t, int64(0), minutes)
 }
 
+func TestInterfaceCountersCoverage_ReturnsCount(t *testing.T) {
+	conn := &mockConn{
+		queryRowFunc: func(_ context.Context, query string, args ...any) driver.Row {
+			assert.Contains(t, query, `"testdb".fact_dz_device_interface_counters`)
+			assert.Contains(t, query, "event_ts")
+			assert.Contains(t, query, "device_pk")
+			assert.Len(t, args, 3)
+			assert.Equal(t, "device123", args[0])
+			return &mockRow{
+				scanFunc: func(dest ...any) error {
+					p := dest[0].(*uint64)
+					*p = 55
+					return nil
+				},
+			}
+		},
+	}
+
+	client := &ClickHouseClient{conn: conn, db: "testdb"}
+	start := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
+	end := start.Add(1 * time.Hour)
+
+	minutes, err := client.InterfaceCountersCoverage(context.Background(), "device123", start, end)
+	require.NoError(t, err)
+	assert.Equal(t, int64(55), minutes)
+}
+
+func TestInterfaceCountersCoverage_QueryError(t *testing.T) {
+	conn := &mockConn{
+		queryRowFunc: func(_ context.Context, _ string, _ ...any) driver.Row {
+			return &mockRow{
+				scanFunc: func(_ ...any) error {
+					return errors.New("connection reset")
+				},
+			}
+		},
+	}
+
+	client := &ClickHouseClient{conn: conn, db: "testdb"}
+	start := time.Now().Add(-1 * time.Hour)
+	end := time.Now()
+
+	_, err := client.InterfaceCountersCoverage(context.Background(), "device123", start, end)
+	assert.ErrorContains(t, err, "connection reset")
+}
+
+func TestInterfaceCountersCoverage_QuotesDatabaseName(t *testing.T) {
+	conn := &mockConn{
+		queryRowFunc: func(_ context.Context, query string, _ ...any) driver.Row {
+			assert.Contains(t, query, `"mainnet-beta".fact_dz_device_interface_counters`)
+			return &mockRow{
+				scanFunc: func(dest ...any) error {
+					p := dest[0].(*uint64)
+					*p = 0
+					return nil
+				},
+			}
+		},
+	}
+
+	client := &ClickHouseClient{conn: conn, db: "mainnet-beta"}
+	start := time.Now().Add(-1 * time.Hour)
+	end := time.Now()
+
+	minutes, err := client.InterfaceCountersCoverage(context.Background(), "device123", start, end)
+	require.NoError(t, err)
+	assert.Equal(t, int64(0), minutes)
+}
+
 func TestNewClickHouseClient_StripsScheme(t *testing.T) {
 	tests := []struct {
 		name string

controlplane/device-health-oracle/internal/worker/interface_counters.go

@@ -0,0 +1,62 @@
+package worker
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+
+	"github.com/gagliardetto/solana-go"
+	"github.com/malbeclabs/doublezero/smartcontract/sdk/go/serviceability"
+)
+
+// InterfaceCountersCriterion checks that a device has at least one interface
+// counter record per minute over the burn-in period by querying the ClickHouse
+// fact_dz_device_interface_counters table.
+//
+// The burn-in start times are resolved from ledger slot numbers via GetBlockTime
+// and passed through the context (see BurnInTimes / ContextWithBurnInTimes).
+type InterfaceCountersCriterion struct {
+	checker InterfaceCountersChecker
+	log     *slog.Logger
+}
+
+func NewInterfaceCountersCriterion(checker InterfaceCountersChecker, log *slog.Logger) *InterfaceCountersCriterion {
+	return &InterfaceCountersCriterion{
+		checker: checker,
+		log:     log,
+	}
+}
+
+func (c *InterfaceCountersCriterion) Name() string {
+	return "interface_counters"
+}
+
+func (c *InterfaceCountersCriterion) Check(ctx context.Context, device serviceability.Device) (bool, string) {
+	start, now, expectedMinutes, ok := DeviceBurnIn(ctx, device.Status)
+	if !ok {
+		return false, "burn-in times not available in context"
+	}
+	if expectedMinutes == 0 {
+		return true, ""
+	}
+
+	pubkey := solana.PublicKeyFromBytes(device.PubKey[:]).String()
+	minutesWithRecords, err := c.checker.InterfaceCountersCoverage(ctx, pubkey, start, now)
+	if err != nil {
+		c.log.Error("Failed to query interface counters coverage",
+			"device", pubkey, "code", device.Code, "error", err)
+		return false, fmt.Sprintf("clickhouse query failed: %v", err)
+	}
+
+	c.log.Debug("Interface counters coverage",
+		"device", pubkey, "code", device.Code,
+		"minutesWithRecords", minutesWithRecords,
+		"expectedMinutes", expectedMinutes,
+		"start", start)
+
+	if minutesWithRecords < expectedMinutes {
+		return false, fmt.Sprintf("interface counters cover %d/%d minutes", minutesWithRecords, expectedMinutes)
+	}
+
+	return true, ""
+}

controlplane/device-health-oracle/internal/worker/interface_counters_test.go

@@ -0,0 +1,134 @@
+package worker
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/malbeclabs/doublezero/smartcontract/sdk/go/serviceability"
+	"github.com/stretchr/testify/assert"
+)
+
+type mockInterfaceCountersChecker struct {
+	minutesWithRecords int64
+	err                error
+}
+
+func (m *mockInterfaceCountersChecker) InterfaceCountersCoverage(_ context.Context, _ string, _, _ time.Time) (int64, error) {
+	return m.minutesWithRecords, m.err
+}
+
+func TestInterfaceCountersCriterion_Name(t *testing.T) {
+	c := NewInterfaceCountersCriterion(nil, testLoggerSlog())
+	assert.Equal(t, "interface_counters", c.Name())
+}
+
+func TestInterfaceCountersCriterion_Passes(t *testing.T) {
+	now := time.Now()
+	start := now.Add(-33 * time.Minute)
+	ctx := ContextWithBurnInTimes(context.Background(), BurnInTimes{
+		DrainedStart: start,
+		Now:          now,
+	})
+
+	checker := &mockInterfaceCountersChecker{minutesWithRecords: 33}
+	c := NewInterfaceCountersCriterion(checker, testLoggerSlog())
+
+	device := serviceability.Device{Status: serviceability.DeviceStatusDrained}
+	passed, reason := c.Check(ctx, device)
+	assert.True(t, passed)
+	assert.Empty(t, reason)
+}
+
+func TestInterfaceCountersCriterion_Fails_InsufficientCoverage(t *testing.T) {
+	now := time.Now()
+	start := now.Add(-33 * time.Minute)
+	ctx := ContextWithBurnInTimes(context.Background(), BurnInTimes{
+		DrainedStart: start,
+		Now:          now,
+	})
+
+	checker := &mockInterfaceCountersChecker{minutesWithRecords: 20}
+	c := NewInterfaceCountersCriterion(checker, testLoggerSlog())
+
+	device := serviceability.Device{Status: serviceability.DeviceStatusDrained}
+	passed, reason := c.Check(ctx, device)
+	assert.False(t, passed)
+	assert.Contains(t, reason, "interface counters cover 20/33 minutes")
+}
+
+func TestInterfaceCountersCriterion_Fails_ClickHouseError(t *testing.T) {
+	now := time.Now()
+	start := now.Add(-1 * time.Hour)
+	ctx := ContextWithBurnInTimes(context.Background(), BurnInTimes{
+		ProvisioningStart: start,
+		Now:               now,
+	})
+
+	checker := &mockInterfaceCountersChecker{err: errors.New("connection refused")}
+	c := NewInterfaceCountersCriterion(checker, testLoggerSlog())
+
+	device := serviceability.Device{Status: serviceability.DeviceStatusDeviceProvisioning}
+	passed, reason := c.Check(ctx, device)
+	assert.False(t, passed)
+	assert.Contains(t, reason, "clickhouse query failed")
+}
+
+func TestInterfaceCountersCriterion_Fails_NoBurnInTimes(t *testing.T) {
+	checker := &mockInterfaceCountersChecker{minutesWithRecords: 100}
+	c := NewInterfaceCountersCriterion(checker, testLoggerSlog())
+
+	device := serviceability.Device{Status: serviceability.DeviceStatusDeviceProvisioning}
+	passed, reason := c.Check(context.Background(), device)
+	assert.False(t, passed)
+	assert.Contains(t, reason, "burn-in times not available")
+}
+
+func TestInterfaceCountersCriterion_UsesProvisioningStart(t *testing.T) {
+	now := time.Now()
+	ctx := ContextWithBurnInTimes(context.Background(), BurnInTimes{
+		ProvisioningStart: now.Add(-60 * time.Minute),
+		DrainedStart:      now.Add(-10 * time.Minute),
+		Now:               now,
+	})
+
+	checker := &mockInterfaceCountersChecker{minutesWithRecords: 60}
+	c := NewInterfaceCountersCriterion(checker, testLoggerSlog())
+
+	device := serviceability.Device{Status: serviceability.DeviceStatusDeviceProvisioning}
+	passed, _ := c.Check(ctx, device)
+	assert.True(t, passed)
+}
+
+func TestInterfaceCountersCriterion_UsesDrainedStart(t *testing.T) {
+	now := time.Now()
+	ctx := ContextWithBurnInTimes(context.Background(), BurnInTimes{
+		ProvisioningStart: now.Add(-60 * time.Minute),
+		DrainedStart:      now.Add(-10 * time.Minute),
+		Now:               now,
+	})
+
+	checker := &mockInterfaceCountersChecker{minutesWithRecords: 10}
+	c := NewInterfaceCountersCriterion(checker, testLoggerSlog())
+
+	device := serviceability.Device{Status: serviceability.DeviceStatusDrained}
+	passed, _ := c.Check(ctx, device)
+	assert.True(t, passed)
+}
+
+func TestInterfaceCountersCriterion_ZeroBurnIn_Passes(t *testing.T) {
+	now := time.Now()
+	ctx := ContextWithBurnInTimes(context.Background(), BurnInTimes{
+		ProvisioningStart: now,
+		Now:               now,
+	})
+
+	checker := &mockInterfaceCountersChecker{err: errors.New("should not be called")}
+	c := NewInterfaceCountersCriterion(checker, testLoggerSlog())
+
+	device := serviceability.Device{Status: serviceability.DeviceStatusDeviceProvisioning}
+	passed, reason := c.Check(ctx, device)
+	assert.True(t, passed)
+	assert.Empty(t, reason)
+}

rfcs/rfc12-network-provisioning.md

@@ -190,8 +190,8 @@ Link onboarding has two stages:
             ▼                          │
 ┌───────────────────────-┐   ┌─────────│────────────────────────────┐
 │ Link health info from  │   │         │                            │
-│ DZ_Ledger, Influx,     │   │         ▼                            │
-│ Prometheus             │   │  ┌──────────────┐                    │
+│ DZ_Ledger, ClickHouse  │   │         ▼                            │
+│                        │   │  ┌──────────────┐                    │
 └──────────────────────-─┘   │  │ link.health  │                    │
                              │  └──────┬───────┘                    │
                              │         │                            │
@@ -277,8 +277,8 @@ Device activation has three stages:
             ▼                          │ are met, update device.health to ReadyForUsers
 ┌───────────────────────-┐          ┌─────────│────────────────────────────────────────────────────────────────────┐
 │ Device health info from│          │         ▼                                                                    │
-│ (DZ_Ledger / Influx /  │          │  ┌──────────────┐                                                            │
-│ Prometheus)            │          │  │ device.health│                                                            │
+│ (DZ_Ledger /           │          │  ┌──────────────┐                                                            │
+│ ClickHouse)            │          │  │ device.health│                                                            │
 └──────────────────────-─┘          │  └──────┬───────┘                                                            │
                                     │         │ if desired_status is Activated and status is DeviceProvisioning    │
                                     │         │ and health is ReadForLinks, set device.status to LinkProvisioning  │
@@ -323,7 +323,7 @@ These criteria must be met before links connected to the device can be activated
 1. DIA
     1. At least 1 DIA interface defined on chain with status = activated
     1. At least 1 DIA interface up for `<burn-in slots>` with zero errors and non-zero utilization
-1. Device is reporting to InfluxDB for `<burn-in slots>` (already established by link RFS criteria)
+1. Device is reporting interface counter data to ClickHouse for `<burn-in slots>`. Verified by querying the ClickHouse `fact_dz_device_interface_counters` table and checking that the device has at least one record per minute over the burn-in window.
 1. Config agent installed and calling the controller consistently for `<burn-in slots>`. Verified by querying the ClickHouse `controller_grpc_getconfig_success` table (database per environment: `devnet`, `testnet`, `mainnet-beta`) and checking that the device has at least one record per minute over the burn-in window. The burn-in window start time is determined by calling `GetBlockTime` on the boundary slot (`current_slot - burn_in_slots`).
 1. Telemetry agent installed and running for `<burn-in slots>` (already established by link RFS criteria)
 
@@ -483,15 +483,15 @@ Changes are needed to the following components:
 
 This RFC should improve the operational controls to manage DZDs and links in the network.  It introduces an intent based methodology that uses explict fields to achieve the desired state.
 
-This RFC adds a new device-health-oracle component that collects data from Solana and the DZ ledger (serviceability and telemetry), reads data from Grafana and InfluxDB, and writes data to serviceability.
+This RFC adds a new device-health-oracle component that collects data from Solana and the DZ ledger (serviceability and telemetry), reads data from ClickHouse, and writes data to serviceability.
 
 Contributors should not be rewarded for devices and links that are not in activated status. This check is already present in contributor-rewards (doublezero-offchain/crates/contributor-rewards/src/calculator/shapley_handler.rs).
 
 This RFC improves network operations, but it also works against DoubleZero's long-term goal of being fully decentralized by adding a centralized component, `device-health-oracle`. Two possible approaches to decentralizing this in the future are: 1) move all health data and related logic to the DZ Ledger, and 2) have contributors run instances of `device-health-oracle` that operate only on that contributor's links and devices.
 
 ## Security Considerations
 
-- The device-health-oracle component will API keys for reading from Grafana and InfluxDB, as well as a key with write access to link.health and device.health. If this key is leaked, an attacker could move a device into activated status even though it's not healthy.
+- The device-health-oracle component will need credentials for reading from ClickHouse, as well as a key with write access to link.health and device.health. If this key is leaked, an attacker could move a device into activated status even though it's not healthy.
 - An attacker could shut down user BGP sessions on DZDs if they gains the ability to update device.status or device.desired_status to Drained.
 
 ## Backward Compatibility