sentinel: add multicast publisher e2e tests

.github/workflows/e2e.yml

@@ -3,7 +3,6 @@ on:
   push:
     branches: [ main, 'hotfix/**' ]
   pull_request:
-    branches: [ main, 'hotfix/**' ]
 
 concurrency:
   group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.sha }}
@@ -25,6 +24,8 @@ env:
   DZ_CLIENT_IMAGE: ghcr.io/malbeclabs/dz-e2e/client:${{ github.sha }}
   DZ_DEVICE_HEALTH_ORACLE_IMAGE: ghcr.io/malbeclabs/dz-e2e/device-health-oracle:${{ github.sha }}
   DZ_GEOPROBE_IMAGE: ghcr.io/malbeclabs/dz-e2e/geoprobe:${{ github.sha }}
+  DZ_SENTINEL_IMAGE: ghcr.io/malbeclabs/dz-e2e/sentinel:${{ github.sha }}
+  DZ_VALIDATOR_METADATA_SERVICE_MOCK_IMAGE: ghcr.io/malbeclabs/dz-e2e/validator-metadata-service-mock:${{ github.sha }}
 
 jobs:
   setup:
@@ -63,6 +64,8 @@ jobs:
           docker push ${{ env.DZ_IMAGE_REPO }}/client:${{ env.DZ_IMAGE_TAG }}
           docker push ${{ env.DZ_IMAGE_REPO }}/device-health-oracle:${{ env.DZ_IMAGE_TAG }}
           docker push ${{ env.DZ_IMAGE_REPO }}/geoprobe:${{ env.DZ_IMAGE_TAG }}
+          docker push ${{ env.DZ_IMAGE_REPO }}/sentinel:${{ env.DZ_IMAGE_TAG }}
+          docker push ${{ env.DZ_IMAGE_REPO }}/validator-metadata-service-mock:${{ env.DZ_IMAGE_TAG }}
       - name: Discover tests and distribute across shards
         id: shard
         working-directory: e2e/
@@ -176,6 +179,8 @@ jobs:
           pull_with_retry ${{ env.DZ_IMAGE_REPO }}/client:${{ env.DZ_IMAGE_TAG }}
           pull_with_retry ${{ env.DZ_IMAGE_REPO }}/device-health-oracle:${{ env.DZ_IMAGE_TAG }}
           pull_with_retry ${{ env.DZ_IMAGE_REPO }}/geoprobe:${{ env.DZ_IMAGE_TAG }}
+          pull_with_retry ${{ env.DZ_IMAGE_REPO }}/sentinel:${{ env.DZ_IMAGE_TAG }}
+          pull_with_retry ${{ env.DZ_IMAGE_REPO }}/validator-metadata-service-mock:${{ env.DZ_IMAGE_TAG }}
           pull_with_retry quay.io/prometheus/prometheus:v2.54.1
           pull_with_retry public.ecr.aws/influxdb/influxdb:1.8
       - name: test

.github/workflows/flow_schema_migrations.yml

@@ -7,7 +7,6 @@ on:
   pull_request:
     paths:
       - "telemetry/flow-enricher/clickhouse/**"
-    branches: [ main, 'hotfix/**' ]
 
 jobs:
   validate:

.github/workflows/go.yml

@@ -3,7 +3,6 @@ on:
   push:
     branches: [ main, 'hotfix/**' ]
   pull_request:
-    branches: [ main, 'hotfix/**' ]
 
 jobs:
   go-build:

.github/workflows/release.pipeline.validation.yml

@@ -4,7 +4,6 @@ on:
   push:
     branches: [ main, 'hotfix/**' ]
   pull_request:
-    branches: [ main, 'hotfix/**' ]
 
 permissions:
   contents: write

.github/workflows/rust.yml

@@ -3,7 +3,6 @@ on:
   push:
     branches: [ main, 'hotfix/**' ]
   pull_request:
-    branches: [ main, 'hotfix/**' ]
 
 jobs:
   rust-build:

.github/workflows/sdk.yml

@@ -3,7 +3,6 @@ on:
   push:
     branches: [ main, 'hotfix/**' ]
   pull_request:
-    branches: [ main, 'hotfix/**' ]
 
 jobs:
   sdk-version-check:

.github/workflows/shreds-e2e.yml

@@ -3,7 +3,6 @@ on:
   push:
     branches: [main, 'hotfix/**']
   pull_request:
-    branches: [main, 'hotfix/**']
 
 concurrency:
   group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.sha }}

CHANGELOG.md

@@ -38,6 +38,7 @@ All notable changes to this project will be documented in this file.
   - handle non-user owned disconnects gracefully
 - Sentinel
   - Add multicast publisher worker with Solana RPC-based validator discovery
+  - Add e2e tests for multicast publisher worker with validator-metadata-service mock
 - SDK
   - Add Go SDK for shred subscription program with read-only account deserialization (epoch state, seat assignments, pricing, settlement, validator client rewards), PDA derivation helpers, RPC fetchers, compatibility tests, and a fetch example CLI
   - Add `GeoLocationTargetTypeOutboundIcmp` to Go geolocation SDK with deserialization and round-trip test support

Makefile

@@ -223,6 +223,7 @@ generate-fixtures:
 #   make e2e-test-keep-nobuild              # both
 #   make e2e-test-cleanup                   # remove leftover containers
 # -----------------------------------------------------------------------------
+
 .PHONY: e2e-build
 e2e-build:
 	cd e2e && $(MAKE) build

e2e/.env.local

@@ -12,3 +12,5 @@ DZ_DEVICE_IMAGE=${DZ_IMAGE_REPO}/device:${DZ_IMAGE_TAG}
 DZ_CLIENT_IMAGE=${DZ_IMAGE_REPO}/client:${DZ_IMAGE_TAG}
 DZ_DEVICE_HEALTH_ORACLE_IMAGE=${DZ_IMAGE_REPO}/device-health-oracle:${DZ_IMAGE_TAG}
 DZ_GEOPROBE_IMAGE=${DZ_IMAGE_REPO}/geoprobe:${DZ_IMAGE_TAG}
+DZ_SENTINEL_IMAGE=${DZ_IMAGE_REPO}/sentinel:${DZ_IMAGE_TAG}
+DZ_VALIDATOR_METADATA_SERVICE_MOCK_IMAGE=${DZ_IMAGE_REPO}/validator-metadata-service-mock:${DZ_IMAGE_TAG}

e2e/docker/base.dockerfile

@@ -94,6 +94,7 @@ RUN --mount=type=cache,id=cargo-${CARGO_LOCK_HASH},target=/cargo \
     cp /target/release/doublezero-activator ${BIN_DIR}/ && \
     cp /target/release/doublezero-admin ${BIN_DIR}/ && \
     cp /target/release/doublezero-geolocation ${BIN_DIR}/ && \
+    cp /target/release/doublezero-sentinel ${BIN_DIR}/ && \
     cp /target/release/fork-accounts ${BIN_DIR}/
 
 # Force COPY in later stages to always copy the binaries, even if they appear to be the same.

e2e/docker/sentinel/Dockerfile

@@ -0,0 +1,13 @@
+ARG BASE_IMAGE=undefined
+FROM ${BASE_IMAGE} AS base
+
+FROM ubuntu:24.04
+
+ENV DEBIAN_FRONTEND=noninteractive
+
+RUN apt-get update && \
+    apt-get install -y curl bash
+
+COPY --from=base /doublezero/bin/doublezero-sentinel /usr/local/bin/doublezero-sentinel
+
+ENTRYPOINT ["doublezero-sentinel"]

e2e/docker/validator-metadata-service-mock/Dockerfile

@@ -0,0 +1,19 @@
+FROM golang:1.26 AS builder
+
+WORKDIR /app
+ARG DOCKERFILE_DIR=e2e/docker/validator-metadata-service-mock
+COPY ${DOCKERFILE_DIR}/main.go .
+RUN go build -o /validator-metadata-service-mock main.go
+
+FROM ubuntu:24.04
+
+RUN apt-get update && apt-get install -y curl && rm -rf /var/lib/apt/lists/*
+
+COPY --from=builder /validator-metadata-service-mock /usr/local/bin/
+
+EXPOSE 8080
+
+HEALTHCHECK --interval=2s --timeout=2s --retries=10 \
+    CMD curl -sf http://localhost:8080/health || exit 1
+
+CMD ["validator-metadata-service-mock"]

e2e/docker/validator-metadata-service-mock/main.go

@@ -0,0 +1,215 @@
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	"io"
+	"log"
+	"net/http"
+	"os"
+	"sync"
+)
+
+type ValidatorMetadataItem struct {
+	IP              string `json:"ip"`
+	ActiveStake     int64  `json:"active_stake"`
+	VoteAccount     string `json:"vote_account"`
+	SoftwareClient  string `json:"software_client"`
+	SoftwareVersion string `json:"software_version"`
+}
+
+type server struct {
+	mu         sync.RWMutex
+	validators []ValidatorMetadataItem
+}
+
+func main() {
+	s := &server{}
+
+	// Load initial response from config file if provided.
+	configPath := os.Getenv("CONFIG_PATH")
+	if configPath == "" {
+		configPath = "/etc/mock/validators.json"
+	}
+	if data, err := os.ReadFile(configPath); err == nil {
+		if err := json.Unmarshal(data, &s.validators); err != nil {
+			log.Fatalf("Failed to parse config file %s: %v", configPath, err)
+		}
+		log.Printf("Loaded %d validators from %s", len(s.validators), configPath)
+	} else {
+		log.Printf("No config file at %s, starting with empty response", configPath)
+	}
+
+	http.HandleFunc("/api/v1/validators-metadata", s.handleValidatorsMetadata)
+	http.HandleFunc("/solana-rpc", s.handleSolanaRPC)
+	http.HandleFunc("/config", s.handleConfig)
+	http.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusOK)
+		fmt.Fprint(w, "ok")
+	})
+
+	addr := ":8080"
+	log.Printf("Validator metadata service mock listening on %s", addr)
+	log.Fatal(http.ListenAndServe(addr, nil))
+}
+
+func (s *server) handleValidatorsMetadata(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodGet {
+		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
+		return
+	}
+
+	// Log for debugging.
+	log.Printf("GET /api/v1/validators-metadata")
+
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	w.Header().Set("Content-Type", "application/json")
+	_ = json.NewEncoder(w).Encode(s.validators)
+}
+
+// handleSolanaRPC handles Solana JSON-RPC requests (getVoteAccounts, getClusterNodes).
+// It synthesizes responses from the configured validator metadata items.
+func (s *server) handleSolanaRPC(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodPost {
+		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
+		return
+	}
+
+	body, err := io.ReadAll(r.Body)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+
+	var req struct {
+		JSONRPC string `json:"jsonrpc"`
+		ID      int    `json:"id"`
+		Method  string `json:"method"`
+	}
+	if err := json.Unmarshal(body, &req); err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+
+	log.Printf("POST /solana-rpc method=%s", req.Method)
+
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	w.Header().Set("Content-Type", "application/json")
+
+	switch req.Method {
+	case "getVoteAccounts":
+		s.handleGetVoteAccounts(w, req.ID)
+	case "getClusterNodes":
+		s.handleGetClusterNodes(w, req.ID)
+	default:
+		resp := map[string]any{
+			"jsonrpc": "2.0",
+			"id":      req.ID,
+			"error":   map[string]any{"code": -32601, "message": "Method not found"},
+		}
+		_ = json.NewEncoder(w).Encode(resp)
+	}
+}
+
+func (s *server) handleGetVoteAccounts(w http.ResponseWriter, id int) {
+	type VoteAccount struct {
+		VotePubkey       string  `json:"votePubkey"`
+		NodePubkey       string  `json:"nodePubkey"`
+		ActivatedStake   int64   `json:"activatedStake"`
+		Commission       int     `json:"commission"`
+		EpochVoteAccount bool    `json:"epochVoteAccount"`
+		LastVote         int     `json:"lastVote"`
+		RootSlot         int     `json:"rootSlot"`
+		EpochCredits     [][]int `json:"epochCredits"`
+	}
+
+	var current []VoteAccount
+	for _, v := range s.validators {
+		nodePubkey := fmt.Sprintf("node-%s", v.IP)
+		current = append(current, VoteAccount{
+			VotePubkey:       v.VoteAccount,
+			NodePubkey:       nodePubkey,
+			ActivatedStake:   v.ActiveStake,
+			Commission:       0,
+			EpochVoteAccount: true,
+			LastVote:         100,
+			RootSlot:         99,
+			EpochCredits:     [][]int{{1, 64, 0}},
+		})
+	}
+	if current == nil {
+		current = []VoteAccount{}
+	}
+
+	resp := map[string]any{
+		"jsonrpc": "2.0",
+		"id":      id,
+		"result": map[string]any{
+			"current":    current,
+			"delinquent": []VoteAccount{},
+		},
+	}
+	_ = json.NewEncoder(w).Encode(resp)
+}
+
+func (s *server) handleGetClusterNodes(w http.ResponseWriter, id int) {
+	type ClusterNode struct {
+		Pubkey       string  `json:"pubkey"`
+		Gossip       *string `json:"gossip"`
+		TPU          *string `json:"tpu"`
+		RPC          *string `json:"rpc"`
+		Version      *string `json:"version"`
+		FeatureSet   *int    `json:"featureSet"`
+		ShredVersion *int    `json:"shredVersion"`
+	}
+
+	var nodes []ClusterNode
+	for _, v := range s.validators {
+		nodePubkey := fmt.Sprintf("node-%s", v.IP)
+		gossip := fmt.Sprintf("%s:8001", v.IP)
+		nodes = append(nodes, ClusterNode{
+			Pubkey: nodePubkey,
+			Gossip: &gossip,
+		})
+	}
+	if nodes == nil {
+		nodes = []ClusterNode{}
+	}
+
+	resp := map[string]any{
+		"jsonrpc": "2.0",
+		"id":      id,
+		"result":  nodes,
+	}
+	_ = json.NewEncoder(w).Encode(resp)
+}
+
+func (s *server) handleConfig(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodPut {
+		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
+		return
+	}
+
+	body, err := io.ReadAll(r.Body)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+
+	var validators []ValidatorMetadataItem
+	if err := json.Unmarshal(body, &validators); err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+
+	s.mu.Lock()
+	s.validators = validators
+	s.mu.Unlock()
+
+	log.Printf("Config updated: %d validators", len(validators))
+	w.WriteHeader(http.StatusOK)
+}

e2e/internal/devnet/builder.go

@@ -118,6 +118,18 @@ func BuildContainerImages(ctx context.Context, log *slog.Logger, workspaceDir st
 			dockerfile: filepath.Join(dockerfilesDir, "geoprobe", "Dockerfile"),
 			args:       append([]string{"--build-arg", baseImageArg, "--build-arg", "DOCKERFILE_DIR=" + filepath.Join(dockerfilesDirRelativeToWorkspace, "geoprobe")}, extraArgs...),
 		},
+		{
+			name:       "sentinel",
+			image:      os.Getenv("DZ_SENTINEL_IMAGE"),
+			dockerfile: filepath.Join(dockerfilesDir, "sentinel", "Dockerfile"),
+			args:       append([]string{"--build-arg", baseImageArg}, extraArgs...),
+		},
+		{
+			name:       "validator-metadata-service-mock",
+			image:      os.Getenv("DZ_VALIDATOR_METADATA_SERVICE_MOCK_IMAGE"),
+			dockerfile: filepath.Join(dockerfilesDir, "validator-metadata-service-mock", "Dockerfile"),
+			args:       append([]string{"--build-arg", "DOCKERFILE_DIR=" + filepath.Join(dockerfilesDirRelativeToWorkspace, "validator-metadata-service-mock")}, extraArgs...),
+		},
 	}
 
 	if verbose {