device-health-oracle: add controller_success activation criterion

Add a criteria-based evaluation pattern to the device-health-oracle and
implement the first criterion: devices must have called the controller
at least once per minute over the burn-in period (verified via ClickHouse
controller_grpc_getconfig_success table).

- Introduce DeviceCriterion/LinkCriterion interfaces and stage-aware
  evaluators that enforce Pending → ReadyForLinks → ReadyForUsers
  progression for devices (minimum two ticks to reach ReadyForUsers)
- Add ControllerSuccessCriterion querying ClickHouse for controller call
  coverage over the burn-in window, with start times resolved via
  GetBlockTime
- Optimize update logic to skip onchain health writes when the value is
  already at the desired state
- ClickHouse connection is optional via CLICKHOUSE_ADDR env var; when
  not set, the oracle falls back to no-criteria behavior
- Validate ClickHouse database name to prevent SQL injection

Author: nikw9944

CHANGELOG.md

@@ -3,11 +3,13 @@
 All notable changes to this project will be documented in this file.
 
 ## Unreleased
-  
+
 ### Breaking
 
 ### Changes
 
+- Controlplane
+  - Add `controller_success` activation criterion to device-health-oracle that verifies devices have consistent controller call coverage over a configurable burn-in period by querying ClickHouse
 - Smartcontract
   - Allow `SubscribeMulticastGroup` for users in `Pending` status so that `CreateSubscribeUser` can be followed by additional subscribe calls before the activator runs ([#3521](https://github.com/malbeclabs/doublezero/pull/3521))
 

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

@@ -129,6 +129,43 @@ func main() {
 	serviceabilityExecutor := serviceability.NewExecutor(log, rpcClient, &signer, networkConfig.ServiceabilityProgramID)
 	telemetryClient := telemetry.New(log, rpcClient, nil, networkConfig.TelemetryProgramID)
 
+	// Initialize ClickHouse-dependent criteria.
+	var deviceCriteria []worker.DeviceCriterion
+	if chAddr := os.Getenv("CLICKHOUSE_ADDR"); chAddr != "" {
+		chDB := os.Getenv("CLICKHOUSE_DB")
+		if chDB == "" {
+			chDB = *env
+		}
+		chUser := os.Getenv("CLICKHOUSE_USER")
+		if chUser == "" {
+			chUser = "default"
+		}
+		chPass := os.Getenv("CLICKHOUSE_PASS")
+		chTLSDisabled := os.Getenv("CLICKHOUSE_TLS_DISABLED") == "true"
+
+		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)
+		} 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)
+		}
+	} else {
+		log.Error("ClickHouse disabled (CLICKHOUSE_ADDR not set), no controller_success criterion")
+	}
+
+	deviceEvaluator := &worker.DeviceHealthEvaluator{
+		ReadyForLinksCriteria: deviceCriteria,
+		ReadyForUsersCriteria: nil,
+		Log:                   log,
+	}
+	linkEvaluator := &worker.LinkHealthEvaluator{
+		ReadyForServiceCriteria: nil,
+		Log:                     log,
+	}
+
 	worker.MetricBuildInfo.WithLabelValues(version, commit, date).Set(1)
 	go func() {
 		listener, err := net.Listen("tcp", *metricsAddr)
@@ -155,6 +192,8 @@ func main() {
 		Env:                     *env,
 		ProvisioningSlotCount:   *provisioningSlotCount,
 		DrainedSlotCount:        *drainedSlotCount,
+		DeviceEvaluator:         deviceEvaluator,
+		LinkEvaluator:           linkEvaluator,
 	})
 	if err != nil {
 		log.Error("Failed to create worker", "error", err)

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

@@ -0,0 +1,85 @@
+package worker
+
+import (
+	"context"
+	"crypto/tls"
+	"fmt"
+	"regexp"
+	"strings"
+	"time"
+
+	"github.com/ClickHouse/clickhouse-go/v2"
+)
+
+var validDBName = regexp.MustCompile(`^[a-zA-Z0-9_-]+$`)
+
+// ControllerCallChecker queries ClickHouse for controller call records.
+type ControllerCallChecker interface {
+	ControllerCallCoverage(ctx context.Context, devicePubkey string, start, end time.Time) (minutesWithCalls int64, err error)
+	Close() error
+}
+
+// ClickHouseClient wraps a ClickHouse connection for reading controller call data.
+type ClickHouseClient struct {
+	conn clickhouse.Conn
+	db   string
+}
+
+func NewClickHouseClient(addr, db, user, pass string, disableTLS bool) (*ClickHouseClient, error) {
+	if !validDBName.MatchString(db) {
+		return nil, fmt.Errorf("invalid clickhouse database name: %q", db)
+	}
+
+	addr = strings.TrimPrefix(addr, "https://")
+	addr = strings.TrimPrefix(addr, "http://")
+
+	opts := &clickhouse.Options{
+		Protocol: clickhouse.HTTP,
+		Addr:     []string{addr},
+		Auth: clickhouse.Auth{
+			Database: db,
+			Username: user,
+			Password: pass,
+		},
+		MaxOpenConns: 5,
+		DialTimeout:  30 * time.Second,
+	}
+	if !disableTLS {
+		opts.TLS = &tls.Config{}
+	}
+
+	conn, err := clickhouse.Open(opts)
+	if err != nil {
+		return nil, fmt.Errorf("clickhouse open: %w", err)
+	}
+	if err := conn.Ping(context.Background()); err != nil {
+		return nil, fmt.Errorf("clickhouse ping: %w", err)
+	}
+
+	return &ClickHouseClient{conn: conn, db: db}, nil
+}
+
+// ControllerCallCoverage returns the number of distinct minutes in [start, end] that have
+// at least one controller_grpc_getconfig_success record for the given device.
+func (c *ClickHouseClient) ControllerCallCoverage(ctx context.Context, devicePubkey string, start, end time.Time) (int64, error) {
+	query := fmt.Sprintf(
+		`SELECT count(DISTINCT toStartOfMinute(timestamp)) AS minutes_with_calls
+		 FROM "%s".controller_grpc_getconfig_success
+		 WHERE device_pubkey = ?
+		   AND timestamp >= ?
+		   AND timestamp <= ?`,
+		c.db,
+	)
+
+	var minutesWithCalls uint64
+	err := c.conn.QueryRow(ctx, query, devicePubkey, start, end).Scan(&minutesWithCalls)
+	if err != nil {
+		return 0, fmt.Errorf("clickhouse query: %w", err)
+	}
+
+	return int64(minutesWithCalls), nil
+}
+
+func (c *ClickHouseClient) Close() error {
+	return c.conn.Close()
+}

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

@@ -0,0 +1,118 @@
+package worker
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// mockRow implements driver.Row for testing.
+type mockRow struct {
+	scanFunc func(dest ...any) error
+}
+
+func (r *mockRow) Err() error             { return nil }
+func (r *mockRow) Scan(dest ...any) error { return r.scanFunc(dest...) }
+func (r *mockRow) ScanStruct(_ any) error { return nil }
+
+// mockConn implements the subset of driver.Conn used by ClickHouseClient.
+type mockConn struct {
+	driver.Conn
+	queryRowFunc func(ctx context.Context, query string, args ...any) driver.Row
+}
+
+func (c *mockConn) QueryRow(ctx context.Context, query string, args ...any) driver.Row {
+	return c.queryRowFunc(ctx, query, args...)
+}
+
+func TestControllerCallCoverage_ReturnsCount(t *testing.T) {
+	conn := &mockConn{
+		queryRowFunc: func(_ context.Context, query string, args ...any) driver.Row {
+			assert.Contains(t, query, `"testdb".controller_grpc_getconfig_success`)
+			assert.Len(t, args, 3)
+			assert.Equal(t, "device123", args[0])
+			return &mockRow{
+				scanFunc: func(dest ...any) error {
+					p := dest[0].(*uint64)
+					*p = 42
+					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.ControllerCallCoverage(context.Background(), "device123", start, end)
+	require.NoError(t, err)
+	assert.Equal(t, int64(42), minutes)
+}
+
+func TestControllerCallCoverage_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.ControllerCallCoverage(context.Background(), "device123", start, end)
+	assert.ErrorContains(t, err, "connection reset")
+}
+
+func TestControllerCallCoverage_QuotesDatabaseName(t *testing.T) {
+	// Verify that database names with hyphens (mainnet-beta) are quoted.
+	conn := &mockConn{
+		queryRowFunc: func(_ context.Context, query string, _ ...any) driver.Row {
+			assert.Contains(t, query, `"mainnet-beta".controller_grpc_getconfig_success`)
+			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.ControllerCallCoverage(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
+		addr string
+	}{
+		{"plain host:port", "localhost:8123"},
+		{"https prefix", "https://clickhouse.example.com:8443"},
+		{"http prefix", "http://localhost:8123"},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			_, err := NewClickHouseClient(tt.addr, "default", "default", "", true)
+			// Connection will fail (no server) — verify no panic and an error is returned.
+			assert.Error(t, err)
+		})
+	}
+}

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

@@ -47,6 +47,10 @@ type Config struct {
 	// DrainedSlotCount is used for reactivated devices/links (status = Drained, HardDrained, SoftDrained).
 	ProvisioningSlotCount uint64
 	DrainedSlotCount      uint64
+
+	// Health evaluators determine target health based on criteria.
+	DeviceEvaluator *DeviceHealthEvaluator
+	LinkEvaluator   *LinkHealthEvaluator
 }
 
 func (c *Config) Validate() error {
@@ -71,5 +75,11 @@ func (c *Config) Validate() error {
 	if c.Interval <= 0 {
 		return errors.New("interval must be greater than 0")
 	}
+	if c.DeviceEvaluator == nil {
+		return errors.New("device evaluator is required")
+	}
+	if c.LinkEvaluator == nil {
+		return errors.New("link evaluator is required")
+	}
 	return nil
 }

controlplane/device-health-oracle/internal/worker/controller_success.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"
+)
+
+// ControllerSuccessCriterion checks that a device has called the controller
+// at least once per minute over the burn-in period by querying the ClickHouse
+// controller_grpc_getconfig_success table.
+//
+// The burn-in start times are resolved from ledger slot numbers via GetBlockTime
+// and passed through the context (see BurnInTimes / ContextWithBurnInTimes).
+type ControllerSuccessCriterion struct {
+	checker ControllerCallChecker
+	log     *slog.Logger
+}
+
+func NewControllerSuccessCriterion(checker ControllerCallChecker, log *slog.Logger) *ControllerSuccessCriterion {
+	return &ControllerSuccessCriterion{
+		checker: checker,
+		log:     log,
+	}
+}
+
+func (c *ControllerSuccessCriterion) Name() string {
+	return "controller_success"
+}
+
+func (c *ControllerSuccessCriterion) 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()
+	minutesWithCalls, err := c.checker.ControllerCallCoverage(ctx, pubkey, start, now)
+	if err != nil {
+		c.log.Error("Failed to query controller call coverage",
+			"device", pubkey, "code", device.Code, "error", err)
+		return false, fmt.Sprintf("clickhouse query failed: %v", err)
+	}
+
+	c.log.Debug("Controller call coverage",
+		"device", pubkey, "code", device.Code,
+		"minutesWithCalls", minutesWithCalls,
+		"expectedMinutes", expectedMinutes,
+		"start", start)
+
+	if minutesWithCalls < expectedMinutes {
+		return false, fmt.Sprintf("controller calls cover %d/%d minutes", minutesWithCalls, expectedMinutes)
+	}
+
+	return true, ""
+}