feat(init): Add Horizon

.env

@@ -1 +1,7 @@
-CHVER=25.5.10
+CHVER=25.5.10
+
+# Port overrides to avoid conflicts
+GRAFANA_PORT=3001
+XATU_SERVER_PORT=8082
+POSTGRES_PORT=5433
+POSTGRES_ADDRESS=0.0.0.0

cmd/horizon.go

@@ -0,0 +1,175 @@
+//nolint:dupl // disable duplicate code warning for cmds
+package cmd
+
+import (
+	"os"
+
+	"github.com/creasty/defaults"
+	"github.com/ethpandaops/xatu/pkg/horizon"
+	"github.com/spf13/cobra"
+	yaml "gopkg.in/yaml.v3"
+)
+
+var (
+	horizonCfgFile string
+)
+
+type HorizonOverride struct {
+	FlagHelper func(cmd *cobra.Command)
+	Setter     func(cmd *cobra.Command, overrides *horizon.Override) error
+}
+
+type HorizonOverrideConfig struct {
+	FlagName     string
+	EnvName      string
+	Description  string
+	OverrideFunc func(val string, overrides *horizon.Override)
+}
+
+func createHorizonOverride(config HorizonOverrideConfig) HorizonOverride {
+	return HorizonOverride{
+		FlagHelper: func(cmd *cobra.Command) {
+			cmd.Flags().String(config.FlagName, "", config.Description+` (env: `+config.EnvName+`)`)
+		},
+		Setter: func(cmd *cobra.Command, overrides *horizon.Override) error {
+			val := ""
+
+			if cmd.Flags().Changed(config.FlagName) {
+				val = cmd.Flags().Lookup(config.FlagName).Value.String()
+			}
+
+			if os.Getenv(config.EnvName) != "" {
+				val = os.Getenv(config.EnvName)
+			}
+
+			if val == "" {
+				return nil
+			}
+
+			config.OverrideFunc(val, overrides)
+
+			return nil
+		},
+	}
+}
+
+var HorizonOverrides = []HorizonOverride{
+	createHorizonOverride(HorizonOverrideConfig{
+		FlagName:    "horizon-xatu-output-authorization",
+		EnvName:     "HORIZON_XATU_OUTPUT_AUTHORIZATION",
+		Description: "sets the authorization secret for all xatu outputs",
+		OverrideFunc: func(val string, overrides *horizon.Override) {
+			overrides.XatuOutputAuth.Enabled = true
+			overrides.XatuOutputAuth.Value = val
+		},
+	}),
+	createHorizonOverride(HorizonOverrideConfig{
+		FlagName:    "horizon-xatu-coordinator-authorization",
+		EnvName:     "HORIZON_XATU_COORDINATOR_AUTHORIZATION",
+		Description: "sets the authorization secret for coordinator requests",
+		OverrideFunc: func(val string, overrides *horizon.Override) {
+			overrides.CoordinatorAuth.Enabled = true
+			overrides.CoordinatorAuth.Value = val
+		},
+	}),
+	createHorizonOverride(HorizonOverrideConfig{
+		FlagName:    "metrics-addr",
+		EnvName:     "METRICS_ADDR",
+		Description: "sets the metrics address",
+		OverrideFunc: func(val string, overrides *horizon.Override) {
+			overrides.MetricsAddr.Enabled = true
+			overrides.MetricsAddr.Value = val
+		},
+	}),
+	createHorizonOverride(HorizonOverrideConfig{
+		FlagName:    "horizon-beacon-node-url",
+		EnvName:     "HORIZON_BEACON_NODE_URL",
+		Description: "sets a single beacon node URL (overrides configured list)",
+		OverrideFunc: func(val string, overrides *horizon.Override) {
+			overrides.BeaconNodeURLs.Enabled = true
+			overrides.BeaconNodeURLs.Value = val
+		},
+	}),
+	createHorizonOverride(HorizonOverrideConfig{
+		FlagName:    "horizon-network-name",
+		EnvName:     "HORIZON_NETWORK_NAME",
+		Description: "overrides the network name detected from the beacon node",
+		OverrideFunc: func(val string, overrides *horizon.Override) {
+			overrides.NetworkName.Enabled = true
+			overrides.NetworkName.Value = val
+		},
+	}),
+}
+
+// horizonCmd represents the horizon command
+var horizonCmd = &cobra.Command{
+	Use:   "horizon",
+	Short: "Runs Xatu in horizon mode.",
+	Long: `Runs Xatu in horizon mode, which provides real-time head tracking
+	with multi-beacon node support and dual-iterator coordination.`,
+	Run: func(cmd *cobra.Command, args []string) {
+		initCommon()
+
+		config, err := loadHorizonConfigFromFile(horizonCfgFile)
+		if err != nil {
+			log.Fatal(err)
+		}
+
+		log = getLogger(config.LoggingLevel, "")
+
+		log.WithField("location", horizonCfgFile).Info("Loaded config")
+
+		overrides := &horizon.Override{}
+		for _, override := range HorizonOverrides {
+			if errr := override.Setter(cmd, overrides); errr != nil {
+				log.Fatal(errr)
+			}
+		}
+
+		h, err := horizon.New(cmd.Context(), log, config, overrides)
+		if err != nil {
+			log.Fatal(err)
+		}
+
+		if err := h.Start(cmd.Context()); err != nil {
+			log.Fatal(err)
+		}
+
+		log.Info("Xatu horizon exited - cya!")
+	},
+}
+
+func init() {
+	rootCmd.AddCommand(horizonCmd)
+
+	horizonCmd.Flags().StringVar(&horizonCfgFile, "config", "horizon.yaml", "config file (default is horizon.yaml)")
+
+	for _, override := range HorizonOverrides {
+		override.FlagHelper(horizonCmd)
+	}
+}
+
+func loadHorizonConfigFromFile(file string) (*horizon.Config, error) {
+	if file == "" {
+		file = "horizon.yaml"
+	}
+
+	config := &horizon.Config{}
+
+	if err := defaults.Set(config); err != nil {
+		return nil, err
+	}
+
+	yamlFile, err := os.ReadFile(file)
+	if err != nil {
+		return nil, err
+	}
+
+	type plain horizon.Config
+
+	if err := yaml.Unmarshal(yamlFile, (*plain)(config)); err != nil {
+		return nil, err
+	}
+
+	return config, nil
+}

deploy/kurtosis/horizon-test.yaml

@@ -0,0 +1,74 @@
+# Kurtosis ethereum-package configuration for Horizon E2E testing
+#
+# This configuration creates a local Ethereum testnet with all consensus clients
+# for testing Horizon's multi-beacon node support.
+#
+# Usage:
+#   kurtosis run github.com/ethpandaops/ethereum-package --args-file horizon-test.yaml --enclave horizon
+#
+# After starting, beacon nodes will be available at:
+#   - lighthouse: http://cl-lighthouse-geth:4000
+#   - prysm: http://cl-prysm-nethermind:3500
+#   - teku: http://cl-teku-erigon:4000
+#   - lodestar: http://cl-lodestar-reth:4000
+#   - nimbus: http://cl-nimbus-besu:4000
+#   - grandine: http://cl-grandine-geth:4000
+#
+# Note: Actual hostnames will vary based on Kurtosis enclave. Use:
+#   kurtosis enclave inspect horizon
+# to get the actual service names and ports.
+
+participants:
+  # Lighthouse CL + Geth EL
+  - el_type: geth
+    cl_type: lighthouse
+    count: 1
+
+  # Prysm CL + Nethermind EL
+  - el_type: nethermind
+    cl_type: prysm
+    count: 1
+
+  # Teku CL + Erigon EL
+  - el_type: erigon
+    cl_type: teku
+    count: 1
+
+  # Lodestar CL + Reth EL
+  - el_type: reth
+    cl_type: lodestar
+    count: 1
+
+  # Nimbus CL + Besu EL
+  - el_type: besu
+    cl_type: nimbus
+    count: 1
+    # Nimbus needs subscribe-all-subnets for full attestation coverage
+    cl_extra_params:
+      - --subscribe-all-subnets
+
+  # Grandine CL + Geth EL (different Geth instance)
+  - el_type: geth
+    cl_type: grandine
+    count: 1
+
+# Network configuration for faster testing
+network_params:
+  # Shorter genesis delay for faster startup
+  genesis_delay: 120
+  # Standard slot time
+  seconds_per_slot: 12
+  # Deneb fork for blob testing
+  deneb_fork_epoch: 0
+  # Electra fork for testing Electra attestations
+  electra_fork_epoch: 1
+
+# Disable additional services - we'll run xatu separately
+additional_services: []
+
+# Global settings
+global_log_level: info
+
+# Port publishing disabled - we'll use docker network for connectivity
+port_publisher:
+  nat_exit_ip: KURTOSIS_IP_ADDR_PLACEHOLDER

deploy/kurtosis/xatu-horizon.yaml

@@ -0,0 +1,136 @@
+# Horizon configuration for Kurtosis E2E testing
+#
+# This configuration connects Horizon to all consensus clients in the Kurtosis network.
+# Beacon node URLs must be updated based on the Kurtosis enclave inspection output.
+#
+# Usage:
+#   1. Start Kurtosis network: kurtosis run github.com/ethpandaops/ethereum-package --args-file horizon-test.yaml --enclave horizon
+#   2. Get beacon node URLs: kurtosis enclave inspect horizon | grep -E "cl-.+-http"
+#   3. Update beaconNodes section below with actual URLs
+#   4. Run Horizon: xatu horizon --config xatu-horizon.yaml
+#
+# Or use environment variables:
+#   export HORIZON_BEACON_NODES="lighthouse=http://...,prysm=http://...,..."
+#   xatu horizon --config xatu-horizon.yaml
+
+logging: "info" # panic,fatal,warn,info,debug,trace
+metricsAddr: ":9098"
+pprofAddr: ":6062"
+
+name: xatu-horizon-e2e
+
+# Labels for E2E test identification
+labels:
+  environment: e2e-test
+  network: kurtosis
+
+# NTP server
+ntpServer: time.google.com
+
+# Coordinator for tracking processing locations
+coordinator:
+  address: xatu-server:8080
+  tls: false
+
+# Multi-beacon node pool - all 6 consensus clients
+# Update these URLs after starting the Kurtosis network
+ethereum:
+  # Override network name for Kurtosis devnet
+  overrideNetworkName: kurtosis
+
+  # Allow extra time for clients to become healthy after genesis
+  startupTimeout: 5m
+
+  beaconNodes:
+    # Lighthouse
+    - name: lighthouse
+      address: http://cl-lighthouse-geth:4000
+    # Prysm (uses port 3500 by default)
+    - name: prysm
+      address: http://cl-prysm-nethermind:3500
+    # Teku
+    - name: teku
+      address: http://cl-teku-erigon:4000
+    # Lodestar
+    - name: lodestar
+      address: http://cl-lodestar-reth:4000
+    # Nimbus
+    - name: nimbus
+      address: http://cl-nimbus-besu:4000
+    # Grandine
+    - name: grandine
+      address: http://cl-grandine-geth:4000
+
+  # Health check interval
+  healthCheckInterval: 3s
+
+  # Block cache settings
+  blockCacheSize: 1000
+  blockCacheTtl: 1h
+  blockPreloadWorkers: 5
+  blockPreloadQueueSize: 5000
+
+# Deduplication cache - 13 minutes covers ~1 epoch plus delays
+dedupCache:
+  ttl: 13m
+
+# SSE subscription settings
+subscription:
+  bufferSize: 1000
+
+# Reorg handling
+reorg:
+  enabled: true
+  maxDepth: 64
+  bufferSize: 100
+
+# Epoch iterator - trigger at 50% through epoch
+epochIterator:
+  enabled: true
+  triggerPercent: 0.5
+
+# Enable all derivers for comprehensive E2E testing
+derivers:
+  # Block-based derivers
+  beaconBlock:
+    enabled: true
+  attesterSlashing:
+    enabled: true
+  proposerSlashing:
+    enabled: true
+  deposit:
+    enabled: true
+  withdrawal:
+    enabled: true
+  voluntaryExit:
+    enabled: true
+  blsToExecutionChange:
+    enabled: true
+  executionTransaction:
+    enabled: true
+  elaboratedAttestation:
+    enabled: true
+
+  # Epoch-based derivers
+  proposerDuty:
+    enabled: true
+  beaconBlob:
+    enabled: true
+  beaconValidators:
+    enabled: true
+    chunkSize: 100
+  beaconCommittee:
+    enabled: true
+
+# Output to local xatu server
+outputs:
+  - name: xatu
+    type: xatu
+    config:
+      address: xatu-server:8080
+      tls: false
+      maxQueueSize: 51200
+      batchTimeout: 0.5s
+      exportTimeout: 30s
+      maxExportBatchSize: 32
+      workers: 50