node.zig

const std = @import("std");
const enr_lib = @import("enr");
const ENR = enr_lib.ENR;
const utils_lib = @import("@zeam/utils");
const Yaml = @import("yaml").Yaml;
const configs = @import("@zeam/configs");
const api = @import("@zeam/api");
const api_server = @import("api_server.zig");
const metrics_server = @import("metrics_server.zig");
const event_broadcaster = api.event_broadcaster;
const ChainConfig = configs.ChainConfig;
const Chain = configs.Chain;
const ChainOptions = configs.ChainOptions;
const sft = @import("@zeam/state-transition");
const xev = @import("xev").Dynamic;
const networks = @import("@zeam/network");
const Multiaddr = @import("multiaddr").Multiaddr;
const node_lib = @import("@zeam/node");
const key_manager_lib = @import("@zeam/key-manager");
const Clock = node_lib.Clock;
const BeamNode = node_lib.BeamNode;
const SlotDriverWatchdog = node_lib.SlotDriverWatchdog;
const ThreadPool = @import("@zeam/thread-pool").ThreadPool;
const xmss = @import("@zeam/xmss");
const types = @import("@zeam/types");
const LoggerConfig = utils_lib.ZeamLoggerConfig;
const NodeCommand = @import("main.zig").NodeCommand;
const zeam_utils = @import("@zeam/utils");
const database = @import("@zeam/database");
const json = std.json;
const utils = @import("@zeam/utils");
const ssz = @import("ssz");
const zeam_metrics = @import("@zeam/metrics");
const build_options = @import("build_options");

// Structure to hold parsed ENR fields from validator-config.yaml
const EnrFields = struct {
    ip: ?[]const u8 = null,
    ip6: ?[]const u8 = null,
    tcp: ?u16 = null,
    udp: ?u16 = null,
    quic: ?u16 = null,
    seq: ?u64 = null,
    // Allow for custom fields
    custom_fields: std.StringHashMap([]const u8),

    pub fn deinit(self: *EnrFields, allocator: std.mem.Allocator) void {
        if (self.ip) |ip_str| allocator.free(ip_str);
        if (self.ip6) |ip6_str| allocator.free(ip6_str);
        var iterator = self.custom_fields.iterator();
        while (iterator.next()) |entry| {
            allocator.free(entry.key_ptr.*);
            allocator.free(entry.value_ptr.*);
        }
        self.custom_fields.deinit();
    }
};

/// Represents a validator assignment from annotated_validators.yaml
pub const ValidatorAssignment = struct {
    index: usize,
    pubkey_hex: []const u8,
    privkey_file: []const u8,

    pub fn deinit(self: *ValidatorAssignment, allocator: std.mem.Allocator) void {
        allocator.free(self.pubkey_hex);
        allocator.free(self.privkey_file);
    }
};

pub const NodeOptions = struct {
    network_id: u32,
    node_key: []const u8,
    node_key_index: usize,
    /// Path to the lean-quickstart `<node>.key` file holding a 64-hex-char
    /// ASCII representation of the 32-byte ECDSA-P-256 host-identity seed.
    /// Threaded through to `EthLibp2pParams.host_identity_key_path` so the
    /// libp2p PeerId derives from the SAME seed `eth-beacon-genesis` used to
    /// populate `nodes.yaml`. Without this, every outbound dial fails
    /// `PeerIdMismatch` against the dial-multiaddr's expected peer id.
    /// `null` only when the legacy code path (no real networking) is in use.
    host_identity_key_path: ?[]const u8 = null,
    // 1. a special value of "genesis_bootnode" for validator config means its a genesis bootnode and so
    //   the configuration is to be picked from genesis
    // 2. otherwise validator_config is dir path to this nodes's validator_config.yaml and annotated_validators.yaml
    //   and one must use all the nodes in genesis nodes.yaml as peers
    validator_config: []const u8,
    bootnodes: []const []const u8,
    validator_assignments: []ValidatorAssignment,
    genesis_spec: types.GenesisSpec,
    metrics_enable: bool,
    is_aggregator: bool,
    /// If aggregator, additional subnet ids to import and aggregate
    aggregation_subnet_ids: ?[]u32 = null,
    api_port: u16,
    metrics_port: u16,
    local_priv_key: []const u8,
    logger_config: *LoggerConfig,
    database_path: []const u8,
    hash_sig_key_dir: []const u8,
    node_registry: *node_lib.NodeNameRegistry,
    checkpoint_sync_url: ?[]const u8 = null,
    attestation_committee_count: ?u64 = null,
    max_attestations_data: ?u8 = null,
    db_backend: database.Backend = .rocksdb,
    chain_spec: ?[]const u8 = null,
    /// Route producer-side gossip handlers through the chain-worker
    /// queue. Defaults to `true`: the worker path is the supported prod
    /// path; surfaced as `--chain-worker` on the `zeam node` CLI, with
    /// `--chain-worker false` as the kill-switch for the legacy
    /// synchronous path.
    chain_worker_enabled: bool = true,
    /// Override the rayon worker count for the multisig aggregate prover.
    /// `null` keeps the existing automatic split (half of the post-system-
    /// thread budget to Zig workers, half to rayon). Aggregators in
    /// CPU-rich environments can set this higher to give the prover more
    /// parallelism without rebuilding. Surfaced as `--rayon-threads`
    /// on the `zeam node` CLI.
    rayon_threads: ?u32 = null,
    /// Minimum (children + gossip-sig) inputs required before the
    /// aggregator pre-filter lets an `AttestationData` reach the FFI.
    /// Threaded through to `ForkChoice.min_aggregation_inputs` and
    /// applied by `pruneTrivialFromAggregateSnapshot` before
    /// `computeAggregatedSignatures` runs (the FFI itself stays
    /// free of this filtering). Surfaced as `--min-aggregation-inputs` on the
    /// `zeam node` CLI; default is `default_min_aggregation_inputs`.
    min_aggregation_inputs: u32 = types.default_min_aggregation_inputs,
    /// Percentage of the proposal interval allocated to interval-4 Type-1
    /// warm-up. Interval-0 proposal packaging does not run Type-1 aggregation.
    type1_aggregation_deadline_pct: u32 = node_lib.default_type1_aggregation_deadline_pct,
    /// Cap on the number of child STARK proofs merged with raw signatures
    /// by the aggregator-worker path. Threaded through to
    /// `ForkChoice.max_aggregation_children` and applied by
    /// `prepareAggregateAttData` after greedy + subset-prune. Surfaced as
    /// `--max-aggregation-children` on the `zeam node` CLI; default is
    /// `pkgs/types/src/block.zig:default_max_aggregation_children` (0 —
    /// flat-only worker path).
    max_aggregation_children: u32 = types.default_max_aggregation_children,
    /// Cap on the number of `AttestationData` the aggregator proves+publishes
    /// per tick (greedy justification path). Threaded through to
    /// `ForkChoice.max_aggregations_per_tick` and consumed by
    /// `aggregateUnlocked`. Surfaced as `--max-aggregations-per-tick` on the
    /// `zeam node` CLI; default is
    /// `pkgs/types/src/block.zig:default_max_aggregations_per_tick` (1 —
    /// single-aggregation behaviour).
    max_aggregations_per_tick: u32 = types.default_max_aggregations_per_tick,

    pub fn deinit(self: *NodeOptions, allocator: std.mem.Allocator) void {
        for (self.bootnodes) |b| allocator.free(b);
        allocator.free(self.bootnodes);
        for (self.validator_assignments) |*assignment| {
            @constCast(assignment).deinit(allocator);
        }
        allocator.free(self.validator_assignments);
        allocator.free(self.local_priv_key);
        allocator.free(self.hash_sig_key_dir);
        if (self.aggregation_subnet_ids) |ids| allocator.free(ids);
        self.node_registry.deinit();
        allocator.destroy(self.node_registry);
    }

    pub fn getValidatorIndices(self: *const NodeOptions, allocator: std.mem.Allocator) ![]usize {
        // Deduplicate: each validator index may appear multiple times in
        // assignments (e.g. once for the attester key, once for the proposer
        // key). The validator only needs to attest/propose once per slot.
        var seen = std.AutoHashMap(usize, void).init(allocator);
        defer seen.deinit();
        var unique: std.ArrayList(usize) = .empty;
        errdefer unique.deinit(allocator);
        for (self.validator_assignments) |assignment| {
            const result = try seen.getOrPut(assignment.index);
            if (!result.found_existing) {
                try unique.append(allocator, assignment.index);
            }
        }
        return try unique.toOwnedSlice(allocator);
    }
};

/// A Node that encapsulates the networking, blockchain, and validator functionalities.
/// It manages the event loop, network interface, clock, and beam node.
pub const Node = struct {
    loop: xev.Loop,
    /// `EthLibp2p.init` returns a heap-allocated `*Self`, so this is a
    /// pointer (was a value field under the legacy `EthLibp2p` flow).
    network: *networks.EthLibp2p,
    /// String-form listen + connect multiaddrs handed to v2. Owned by
    /// `Node`; freed at `deinit` so v2 (which stores slice refs, not
    /// dupes) is safe across its lifetime.
    listen_addresses_str: []u8,
    connect_peers_str: []u8,
    beam_node: BeamNode,
    clock: Clock,
    enr: ENR,
    options: *const NodeOptions,
    allocator: std.mem.Allocator,
    logger: zeam_utils.ModuleLogger,
    db: database.Db,
    key_manager: key_manager_lib.KeyManager,
    api_server_handle: ?*api_server.ApiServer,
    metrics_server_handle: ?*metrics_server.MetricsServer,
    anchor_state: *types.BeamState,
    /// Shared worker pool for CPU-bound chain work (attestation signature verification).
    thread_pool: *ThreadPool,
    /// Background watchdog that monitors libxev slot-driver liveness.
    /// `null` until `run()` spawns it; `stop()` joins it.
    slot_driver_watchdog: ?SlotDriverWatchdog = null,

    const Self = @This();

    /// Thread roles configured at startup. Counts are logged once init
    /// completes so operators can verify `--rayon-threads` / aggregator
    /// auto-tune and compare against `cpu_count`.
    const StartupThreadBudget = struct {
        cpu_count: usize,
        zig_worker_pool: usize,
        rayon: usize,
        aggregate_max_inflight: u32,
        chain_worker: usize,
        metrics_server: usize,
        api_server: usize,
        rayon_source: []const u8,

        fn estimatedTotal(self: @This()) usize {
            // main (xev slot-driver) + Io.Threaded + libp2p + zig pool + rayon
            // + chain-worker + metrics + api + slot watchdog (run()).
            return 1 + 1 + 1 + self.zig_worker_pool + self.rayon + self.chain_worker +
                self.metrics_server + self.api_server + 1;
        }
    };

    fn logStartupThreadBudget(self: *Self, budget: StartupThreadBudget) void {
        self.logger.info(
            "startup thread budget: cpu_count={d} estimated_os_threads≈{d}",
            .{ budget.cpu_count, budget.estimatedTotal() },
        );
        self.logger.info(
            "  main/xev slot-driver: 1 (this thread runs xev.Loop via clock.run)",
            .{},
        );
        self.logger.info(
            "  std.Io.Threaded:      1 (blocking I/O for Zig pool + disk)",
            .{},
        );
        self.logger.info(
            "  libp2p rust bridge:   1 (spawned in network.run)",
            .{},
        );
        self.logger.info(
            "  zig worker pool:      {d} (aggregate_max_inflight={d})",
            .{ budget.zig_worker_pool, budget.aggregate_max_inflight },
        );
        self.logger.info(
            "  rayon (XMSS prover):  {d}{s}",
            .{ budget.rayon, budget.rayon_source },
        );
        self.logger.info(
            "  chain-worker:         {d}{s}",
            .{ budget.chain_worker, if (budget.chain_worker == 0) " (disabled)" else "" },
        );
        self.logger.info(
            "  metrics HTTP server:  {d}",
            .{budget.metrics_server},
        );
        self.logger.info(
            "  api HTTP server:      {d}",
            .{budget.api_server},
        );
        self.logger.info(
            "  slot-driver watchdog: 1 (spawned at run() start)",
            .{},
        );
    }

    /// Closes the current database, wipes the on-disk rocksdb directory, and
    /// reopens a fresh database at the same path.
    ///
    /// If `ignore_not_found` is true, `error.FileNotFound` from the directory
    /// deletion is silently swallowed (used for first-run installs where the
    /// db directory has never been created). Set it to false when wiping a db
    /// that is known to exist (genesis time mismatch case).
    fn wipeAndReopenDb(
        db: *database.Db,
        allocator: std.mem.Allocator,
        database_path: []const u8,
        logger_config: *LoggerConfig,
        logger: zeam_utils.ModuleLogger,
        backend: database.Backend,
        ignore_not_found: bool,
    ) !void {
        db.deinit();
        const io = std.Io.Threaded.global_single_threaded.io();
        // Both backends store their working set under the same base
        // directory; deleting it yields a clean slate for either engine.
        const backend_dir = switch (backend) {
            .rocksdb => "rocksdb",
            .lmdb => "lmdb",
        };
        const db_path = try std.fmt.allocPrint(allocator, "{s}/{s}", .{ database_path, backend_dir });
        defer allocator.free(db_path);
        std.Io.Dir.cwd().deleteTree(io, db_path) catch |wipe_err| {
            if (!ignore_not_found or wipe_err != error.FileNotFound) {
                logger.err("failed to delete database directory '{s}': {any}", .{ db_path, wipe_err });
                return wipe_err;
            }
        };
        db.* = try database.Db.openBackend(allocator, logger_config.logger(.database), database_path, backend);
    }

    pub fn init(
        self: *Self,
        allocator: std.mem.Allocator,
        options: *const NodeOptions,
    ) !void {
        self.allocator = allocator;
        self.options = options;
        self.api_server_handle = null;
        self.metrics_server_handle = null;
        self.logger = options.logger_config.logger(.node);
        // If path is specified load from it, otherwise use default settings
        const chain_spec_owned = self.options.chain_spec != null;
        const chain_spec = if (self.options.chain_spec) |path|
            std.Io.Dir.cwd().readFileAlloc(std.Io.Threaded.global_single_threaded.io(), path, allocator, .limited(1024 * 1024)) catch |err| {
                self.logger.err("failed to load chain spec at '{s}': {any}", .{ path, err });
                return err;
            }
        else
            \\{"preset": "mainnet", "name": "devnet0", "fork_digest": "12345678"}
        ;

        defer if (chain_spec_owned) allocator.free(chain_spec);

        const json_options = json.ParseOptions{
            .ignore_unknown_fields = true,
            .allocate = .alloc_if_needed,
        };
        // `parseFromSlice` allocates string fields inside the `Parsed` arena.
        // The slice headers it returns alias arena memory; `chain_config` later
        // owns these fields and `ChainSpec.deinit(allocator)` calls
        // `allocator.free(self.name)` / `allocator.free(self.fork_digest)`. We
        // must move both fields out of the arena onto the top-level allocator
        // before dropping the arena, otherwise shutdown panics with
        // "Invalid free" once `chain.deinit -> config.deinit` runs.
        const parsed = try json.parseFromSlice(ChainOptions, allocator, chain_spec, json_options);
        defer parsed.deinit();
        var chain_options = parsed.value;
        chain_options.name = try allocator.dupe(u8, chain_options.name.?);
        errdefer if (chain_options.name) |n| allocator.free(n);
        chain_options.fork_digest = try allocator.dupe(u8, chain_options.fork_digest.?);
        errdefer if (chain_options.fork_digest) |d| allocator.free(d);
        chain_options.genesis_time = options.genesis_spec.genesis_time;

        if (chain_spec_owned) {
            if (chain_options.preset == null) {
                self.logger.err("chain spec: 'preset' field is required", .{});
                return error.InvalidChainSpec;
            }

            if (chain_options.name == null or chain_options.name.?.len == 0) {
                self.logger.err("chain spec: 'name' field is required", .{});
                return error.InvalidChainSpec;
            }

            if (chain_options.fork_digest == null or chain_options.fork_digest.?.len != 8) {
                self.logger.err("chain spec: 'fork_digest' field must be 4 bytes (8 hex characters)", .{});
                return error.InvalidChainSpec;
            }
        }

        // Set validator pubkeys from genesis_spec (read from config.yaml via genesisConfigFromYAML)
        chain_options.validator_attestation_pubkeys = options.genesis_spec.validator_attestation_pubkeys;
        chain_options.validator_proposal_pubkeys = options.genesis_spec.validator_proposal_pubkeys;

        // Apply attestation_committee_count if provided via CLI flag or config.yaml.
        // ChainConfig.init falls back to 1 when this field is null, so we only override when set.
        if (options.attestation_committee_count) |count| {
            chain_options.attestation_committee_count = @intCast(count);
        }

        // Apply max_attestations_data if provided via config.yaml.
        // ChainConfig.init falls back to 16 (spec default) when this field is null.
        if (options.max_attestations_data) |max| {
            chain_options.max_attestations_data = max;
        }

        // transfer ownership of the chain_options to ChainConfig
        const chain_config = try ChainConfig.init(Chain.custom, chain_options);

        // TODO we seem to be needing one loop because then the events added to loop are not being fired
        // in the order to which they have been added even with the an appropriate delay added
        // behavior of this further needs to be investigated but for now we will share the same loop
        self.loop = try xev.Loop.init(.{});

        // v2 wants strings (one listen multiaddr; comma-separated connect
        // multiaddrs). The legacy ENR plumbing returns `[]Multiaddr`; we
        // convert here and own the resulting strings on `Node` so v2
        // (which stores slice references, not dupes) is safe across its
        // lifetime.
        const addresses = try self.constructMultiaddrs();
        defer {
            for (addresses.listen_addresses) |addr| addr.deinit();
            allocator.free(addresses.listen_addresses);
            for (addresses.connect_peers) |addr| addr.deinit();
            allocator.free(addresses.connect_peers);
        }
        self.listen_addresses_str = if (addresses.listen_addresses.len > 0)
            try addresses.listen_addresses[0].toString(allocator)
        else
            try allocator.dupe(u8, "");
        errdefer allocator.free(self.listen_addresses_str);
        self.connect_peers_str = try joinMultiaddrsCsv(allocator, addresses.connect_peers);
        errdefer allocator.free(self.connect_peers_str);

        self.network = try networks.EthLibp2p.init(allocator, &self.loop, .{
            .networkId = options.network_id,
            .fork_digest = chain_config.spec.fork_digest,
            .listen_addresses = self.listen_addresses_str,
            .connect_peers = self.connect_peers_str,
            .node_registry = options.node_registry,
            .host_identity_key_path = options.host_identity_key_path,
        }, options.logger_config.logger(.network));
        errdefer self.network.deinit();
        self.clock = try Clock.init(allocator, chain_config.genesis.genesis_time, &self.loop, options.logger_config);
        errdefer self.clock.deinit(allocator);

        var db = try database.Db.openBackend(
            allocator,
            options.logger_config.logger(.database),
            options.database_path,
            options.db_backend,
        );
        errdefer db.deinit();

        const anchorState: *types.BeamState = try allocator.create(types.BeamState);
        errdefer allocator.destroy(anchorState);
        self.anchor_state = anchorState;
        errdefer self.anchor_state.deinit();

        // load a valid local state available in db else genesis
        var local_finalized_state: types.BeamState = undefined;
        if (db.loadLatestFinalizedState(&local_finalized_state)) {
            if (local_finalized_state.config.genesis_time != chain_config.genesis.genesis_time) {
                self.logger.warn("database genesis time mismatch (db={d}, config={d}), wiping stale database", .{
                    local_finalized_state.config.genesis_time,
                    chain_config.genesis.genesis_time,
                });
                try wipeAndReopenDb(&db, allocator, options.database_path, options.logger_config, self.logger, options.db_backend, false);
                self.logger.info("stale database wiped, starting fresh & generating genesis", .{});

                local_finalized_state.deinit();
                try self.anchor_state.genGenesisState(allocator, chain_config.genesis);
            } else {
                self.anchor_state.* = local_finalized_state;
            }
        } else |_| {
            self.logger.info("no finalized state found in db, wiping database for a clean slate", .{});
            // ignore_not_found=true: db dir may not exist yet on a fresh install
            try wipeAndReopenDb(&db, allocator, options.database_path, options.logger_config, self.logger, options.db_backend, true);
            self.logger.info("starting fresh & generating genesis", .{});
            try self.anchor_state.genGenesisState(allocator, chain_config.genesis);
        }

        // check if a valid and more recent checkpoint finalized state is available
        if (options.checkpoint_sync_url) |checkpoint_url| {
            self.logger.info("checkpoint sync enabled, downloading state from: {s}", .{checkpoint_url});

            // Try checkpoint sync, fall back to database/genesis on failure
            if (downloadCheckpointState(allocator, checkpoint_url, self.logger)) |downloaded_state_const| {
                var downloaded_state = downloaded_state_const;
                // Verify state against genesis config
                if (verifyCheckpointState(allocator, &downloaded_state, &chain_config.genesis, self.logger)) {
                    if (downloaded_state.slot > self.anchor_state.slot) {
                        self.logger.info("checkpoint sync completed successfully with a recent state at slot={d} as anchor", .{downloaded_state.slot});
                        self.anchor_state.deinit();
                        self.anchor_state.* = downloaded_state;
                        // Fetch the real anchor block from the checkpoint provider and store
                        // it in the DB so blocks_by_root can serve it with the correct
                        // hash_tree_root.  Compute the finalized anchor: derive
                        // anchor_block_root + expected_state_root, then fetch + verify both
                        // root and state_root pairing before persisting.
                        //
                        // Fail closed: if any hash computation fails we skip the block fetch
                        // rather than passing null to downloadAndStoreCheckpointBlock and
                        // silently skipping the pairing check.
                        anchor_block_fetch: {
                            var anchor_state_root: types.Root = undefined;
                            zeam_utils.hashTreeRoot(types.BeamState, self.anchor_state.*, &anchor_state_root, allocator) catch |err| {
                                self.logger.warn("checkpoint block fetch: hashTreeRoot(BeamState) failed: {} — skipping", .{err});
                                break :anchor_block_fetch;
                            };
                            const hdr = self.anchor_state.genStateBlockHeader(allocator) catch |err| {
                                self.logger.warn("checkpoint block fetch: genStateBlockHeader failed: {} — skipping", .{err});
                                break :anchor_block_fetch;
                            };
                            var anchor_block_root: types.Root = undefined;
                            zeam_utils.hashTreeRoot(types.BeamBlockHeader, hdr, &anchor_block_root, allocator) catch |err| {
                                self.logger.warn("checkpoint block fetch: hashTreeRoot(BeamBlockHeader) failed: {} — skipping", .{err});
                                break :anchor_block_fetch;
                            };
                            downloadAndStoreCheckpointBlock(allocator, checkpoint_url, anchor_block_root, anchor_state_root, &db, self.logger);
                        }
                    } else {
                        self.logger.warn("skipping checkpoint sync downloaded stale/same state at slot={d}, falling back to database", .{downloaded_state.slot});
                        downloaded_state.deinit();
                    }
                } else |verify_err| {
                    self.logger.warn("checkpoint state verification failed: {}, falling back to database/genesis", .{verify_err});
                    downloaded_state.deinit();
                }
            } else |download_err| {
                self.logger.warn("checkpoint sync failed: {}, falling back to database/genesis", .{download_err});
            }
        }

        const num_validators: usize = @intCast(chain_config.genesis.numValidators());
        self.key_manager = key_manager_lib.KeyManager.init(allocator);
        errdefer self.key_manager.deinit();

        try self.loadValidatorKeypairs(num_validators);

        const validator_ids = try options.getValidatorIndices(allocator);
        errdefer allocator.free(validator_ids);

        // Initialize metrics BEFORE beam_node so that metrics set during
        // initialization (like lean_validators_count) are captured on real
        // metrics instead of being discarded by noop metrics.
        if (options.metrics_enable) {
            try api.init(allocator);
            zeam_metrics.metrics.lean_node_start_time_seconds.set(@intCast(zeam_utils.unixTimestampSeconds()));
        }

        const cpu_count = std.Thread.getCpuCount() catch 2;
        const reserved_system_threads: usize = 4; // main, p2p, api server, metrics server
        const desired_workers = @max(@as(usize, 1), cpu_count -| reserved_system_threads);

        // Aggregators: XMSS recursive prove (rayon) is the per-slot bottleneck. Keep a
        // small Zig pool for capped in-flight aggregate workers.
        const worker_count = if (options.is_aggregator) blk: {
            const aggregator_zig_workers = @max(@as(usize, 2), desired_workers / 4);
            break :blk @min(@as(usize, ThreadPool.max_thread_count), aggregator_zig_workers);
        } else blk: {
            const zig_worker_budget = @max(@as(usize, 1), (desired_workers + 1) / 2);
            break :blk @min(zig_worker_budget, @as(usize, ThreadPool.max_thread_count));
        };

        // Coordinate the Zig worker pool and the rayon pool used by the XMSS
        // aggregate prover from the same post-system-thread budget so they do
        // not independently claim every remaining CPU. Prefer the Zig pool for
        // the extra worker on odd counts since aggregate verification enters
        // rayon from Zig workers. Both pools still keep a minimum of one worker
        // so tiny/cgroup-limited systems remain functional.
        //
        // Operators can override the rayon worker count with `--rayon-threads`.
        // The automatic split is conservative — it deliberately leaves
        // half of the post-system-thread budget to the Zig pool because
        // verification and gossip work also enter rayon. On a CPU-rich
        // aggregator that bottleneck is the produce path instead, so giving
        // rayon more cores measurably shortens the per-pass build time.
        //
        // Must be called before setupXmssAggregation since rayon’s global
        // pool is initialized lazily on first use.
        const rayon_threads = if (options.rayon_threads) |override| blk: {
            const requested = @max(@as(usize, 1), @as(usize, override));
            // Cap explicit overrides to the post-system budget so hosts
            // with `--rayon-threads 12` on 8 vCPUs do not oversubscribe XMSS
            // prove and inflate worker p50.
            const effective = @min(requested, desired_workers);
            if (effective < requested) {
                self.logger.warn(
                    "--rayon-threads {d} exceeds post-system budget {d}; using {d} rayon threads",
                    .{ requested, desired_workers, effective },
                );
            }
            break :blk effective;
        } else if (options.is_aggregator)
            desired_workers
        else
            @max(@as(usize, 1), desired_workers -| worker_count);
        // One outer aggregate tick at a time on aggregators. A tick may prove
        // multiple justification-path att_data in parallel (bounded by
        // --max-aggregations-per-tick), so Rayon gets the full post-system CPU
        // budget while we avoid stacking multiple concurrent aggregate ticks
        // on top of ThreadPool × Rayon work (#925).
        const aggregate_max_inflight: u32 = if (options.is_aggregator) 1 else 4;
        const rayon_source: []const u8 = if (options.rayon_threads != null)
            " (--rayon-threads override)"
        else if (options.is_aggregator)
            " (aggregator auto-tune: full post-system budget)"
        else
            " (non-aggregator auto-split)";
        // Operator-typo guard for --rayon-threads.
        // Rayon tolerates over-subscription, but values like `--rayon-threads 160`
        // on a 4-vCPU box silently degrade throughput. Warn (don't reject) so the
        // operator notices in startup logs without blocking deliberate edge cases
        // (e.g. fractional cgroup quotas where `getCpuCount` reports more CPUs
        // than the container can actually use).
        if (options.rayon_threads) |override| {
            if (@as(usize, override) > cpu_count) {
                self.logger.warn(
                    "--rayon-threads {d} exceeds detected cpu_count={d}; rayon over-subscription typically reduces throughput. Verify this is intentional.",
                    .{ override, cpu_count },
                );
            }
        }
        xmss.setRayonThreads(rayon_threads);

        // Single XMSS aggregation setup for both prover and verifier paths
        try xmss.setupXmssAggregation();

        self.thread_pool = try ThreadPool.init(.{
            .allocator = allocator,
            .io = std.Io.Threaded.global_single_threaded.io(),
            .thread_count = @intCast(worker_count),
        });
        errdefer self.thread_pool.deinit();

        // Log the aggregator threshold on startup so operators can see
        // exactly how `--min-aggregation-inputs` was resolved (default vs
        // override). The threshold is enforced by the aggregator-side
        // pre-filter in `pruneTrivialFromAggregateSnapshot`,
        // not inside the core `computeAggregatedSignatures` FFI.
        self.logger.info(
            "aggregator threshold: min_aggregation_inputs={d}{s}",
            .{
                options.min_aggregation_inputs,
                if (options.min_aggregation_inputs != types.default_min_aggregation_inputs)
                    " (override via --min-aggregation-inputs)"
                else
                    "",
            },
        );

        self.logger.info(
            "aggregator cap: max_aggregations_per_tick={d}{s}",
            .{
                options.max_aggregations_per_tick,
                if (options.max_aggregations_per_tick != types.default_max_aggregations_per_tick)
                    " (override via --max-aggregations-per-tick)"
                else
                    "",
            },
        );

        try self.beam_node.init(allocator, .{
            .nodeId = @intCast(options.node_key_index),
            .config = chain_config,
            .anchorState = self.anchor_state,
            .backend = self.network.getNetworkInterface(),
            .clock = &self.clock,
            .validator_ids = validator_ids,
            .key_manager = &self.key_manager,
            .db = db,
            .logger_config = options.logger_config,
            .node_registry = options.node_registry,
            .is_aggregator = options.is_aggregator,
            .aggregation_subnet_ids = options.aggregation_subnet_ids,
            .thread_pool = self.thread_pool,
            .chain_worker_enabled = options.chain_worker_enabled,
            .min_aggregation_inputs = options.min_aggregation_inputs,
            .max_aggregation_children = options.max_aggregation_children,
            .max_aggregations_per_tick = options.max_aggregations_per_tick,
            .aggregate_max_inflight = aggregate_max_inflight,
            .type1_aggregation_deadline_pct = options.type1_aggregation_deadline_pct,
        });
        errdefer self.beam_node.deinit();

        // Start API and metrics servers
        // Note: api.init() was already called above before beam_node.init()
        if (options.metrics_enable) {
            // Validate that API and metrics ports are different
            if (options.api_port == options.metrics_port) {
                std.log.err("API port and metrics port cannot be the same (both set to {d})", .{options.api_port});
                return error.PortConflict;
            }

            // Start metrics server (doesn't need chain reference)
            self.metrics_server_handle = try metrics_server.startMetricsServer(
                allocator,
                options.metrics_port,
                options.logger_config,
            );
            // Clean up metrics server if subsequent init operations fail
            errdefer if (self.metrics_server_handle) |handle| handle.stop();

            // Set validator status gauges on node start
            zeam_metrics.metrics.lean_is_aggregator.set(if (options.is_aggregator) 1 else 0);
            // Set committee count from config
            const committee_count = chain_config.spec.attestation_committee_count;
            zeam_metrics.metrics.lean_attestation_committee_count.set(committee_count);
            // Set subnet for the first validator (if any)
            if (validator_ids.len > 0) {
                const first_validator_id: types.ValidatorIndex = @intCast(validator_ids[0]);
                const subnet_id = types.computeSubnetId(first_validator_id, committee_count) catch 0;
                zeam_metrics.metrics.lean_attestation_committee_subnet.set(subnet_id);
            } else {
                zeam_metrics.metrics.lean_attestation_committee_subnet.set(0);
            }

            // Start API server (pass chain pointer for chain-dependent endpoints)
            self.api_server_handle = try api_server.startAPIServer(
                allocator,
                options.api_port,
                options.logger_config,
                self.beam_node.chain,
            );

            // Set node lifecycle metrics
            zeam_metrics.metrics.lean_node_info.set(.{ .name = "zeam", .version = build_options.version }, 1) catch {};
        }

        self.logStartupThreadBudget(.{
            .cpu_count = cpu_count,
            .zig_worker_pool = worker_count,
            .rayon = rayon_threads,
            .aggregate_max_inflight = aggregate_max_inflight,
            .chain_worker = if (options.chain_worker_enabled) 1 else 0,
            .metrics_server = if (options.metrics_enable) 1 else 0,
            .api_server = if (options.metrics_enable) 1 else 0,
            .rayon_source = rayon_source,
        });

        self.logger = options.logger_config.logger(.node);
    }

    pub fn deinit(self: *Self) void {
        if (self.slot_driver_watchdog) |*wd| {
            wd.stop();
        }
        if (self.api_server_handle) |handle| {
            handle.stop();
        }
        if (self.metrics_server_handle) |handle| {
            handle.stop();
        }
        self.clock.deinit(self.allocator);
        self.beam_node.deinit();
        self.thread_pool.deinit();
        self.allocator.free(self.listen_addresses_str);
        self.allocator.free(self.connect_peers_str);
        self.key_manager.deinit();
        self.network.deinit();
        self.enr.deinit();
        self.db.deinit();
        self.loop.deinit();
        event_broadcaster.deinitGlobalBroadcaster();
        self.anchor_state.deinit();
        self.allocator.destroy(self.anchor_state);
    }

    pub fn run(self: *Node) !void {
        // Start the Rust libp2p network before BeamNode:
        // `BeamNode.run()` calls `gossip.subscribe(...)`, which enqueues
        // `SwarmCommand::SubscribeGossip` on the per-network command channel.
        // That channel only exists after `EthLibp2p.run()` returns from
        // `wait_for_network_ready`. Reversing the order drops every subscribe
        // with `error.GossipMeshSubscribeFailed`. The dev `beam`
        // command already does network-first; this is the matching swap for
        // the production node path.
        //
        // Peer + req-resp handlers are subscribed before the network starts so
        // the one-shot peer-connect event and early STATUS requests are not
        // lost to a handler-not-yet-registered race; only gossip mesh subscribe
        // needs the running channel and stays inside `BeamNode.run()`.
        try self.beam_node.subscribeNetworkEventHandlers();
        try self.network.run();
        try self.beam_node.run();

        const ascii_art =
            \\  ███████████████████████████████████████████████████████
            \\  ██████████████                         ████  ██████████
            \\  ███████████        ████████████████       █████████████
            \\  █████████      ████████████████████████     ███████████
            \\  ██████    █████████████████████████████████     ███████
            \\  █████    █████████████████████  █████████████    ██████
            \\  ███     ██████████       █   █████   █████████    █████
            \\  ███    ███████████  █████ █ █ █ ███████████████     ███
            \\  ██    ██████████ ██ ██ ████ ███ ██    ██████████   ████
            \\  ██   ██████████         ███ ████       █████████    ███
            \\  ██   ███████████  █  ██████ ████        █████████   ███
            \\  █    █████████ ████ █████     █████ █████████████   ███
            \\  █    ██████████ █   ████     ██   █████   ███████   ███
            \\  ██   ██████████       ████████ ██    █    ██████    ███
            \\  ██    █████████       ███████ █       ██████████    ███
            \\  ███   ██████████      ███   ███       █████████    █ ██
            \\  ███    ████████████ ███ ██ ███ █     █ ███████    █████
            \\  ███     ████████████ ████   █████   █████████    ██████
            \\  █████     █████████   ███   █████   ████████    ███████
            \\  ████████      ██████████████████████████     ██████████
            \\  ████████  █      ████████████████████      ████████████
            \\  █████  ██████         ██████████         ██████████████
            \\  █████████████████                    ██████████████████
            \\  ███████████████████████████████████████████████████████
            \\
            \\           ███████╗███████╗ █████╗ ███╗   ███╗
            \\           ╚══███╔╝██╔════╝██╔══██╗████╗ ████║
            \\             ███╔╝ █████╗  ███████║██╔████╔██║
            \\            ███╔╝  ██╔══╝  ██╔══██║██║╚██╔╝██║
            \\           ███████╗███████╗██║  ██║██║ ╚═╝ ██║
            \\           ╚══════╝╚══════╝╚═╝  ╚═╝╚═╝     ╚═╝
            \\
            \\          A blazing fast lean consensus client
        ;

        var encoded_txt_buf: [1000]u8 = undefined;
        const encoded_txt = try self.enr.encodeToTxt(&encoded_txt_buf);

        const quic_port = try self.enr.getQUIC();

        // Use logger.info instead of std.debug.print
        self.logger.info("\n{s}", .{ascii_art});
        self.logger.info("════════════════════════════════════════════════════════", .{});
        self.logger.info("  🚀 Zeam Lean Node Started Successfully!", .{});
        self.logger.info("════════════════════════════════════════════════════════", .{});
        self.logger.info("  Node ID: {d}", .{self.options.node_key_index});
        self.logger.info("  Listening on QUIC port: {?d}", .{quic_port});
        self.logger.info("  ENR: {s}", .{encoded_txt});
        self.logger.info("────────────────────────────────────────────────────────", .{});

        // Spawn the slot-driver stall watchdog so multi-second
        // libxev stalls surface in the log + metrics even if the main
        // loop is stuck inside one completion or syscall. Failure to
        // spawn is non-fatal — log and continue without it.
        //
        // The stall callback flips an atomic flag on `BeamNode` that the
        // next libxev tick observes and acts on, forcing a peer status
        // refresh outside the normal 8-slot cadence. This bootstraps
        // catch-up as soon as the slot driver resumes.
        self.slot_driver_watchdog = SlotDriverWatchdog.init(
            &self.clock,
            self.options.logger_config.logger(.clock),
            .{
                .on_stall = .{
                    .ptr = &self.beam_node,
                    .onStall = BeamNode.onSlotDriverStall,
                },
            },
        );
        if (self.slot_driver_watchdog) |*wd| {
            wd.start() catch |err| {
                self.logger.warn("failed to start slot-driver watchdog: {any}", .{err});
                self.slot_driver_watchdog = null;
            };
        }

        try self.clock.run();
    }

    fn constructMultiaddrs(self: *Self) !struct { listen_addresses: []const Multiaddr, connect_peers: []const Multiaddr } {
        if (std.mem.eql(u8, self.options.validator_config, "genesis_bootnode")) {
            try ENR.decodeTxtInto(&self.enr, self.options.bootnodes[self.options.node_key_index]);
        } else {
            // Parse validator config to get ENR fields
            const validator_config_filepath = try std.mem.concat(self.allocator, u8, &[_][]const u8{
                self.options.validator_config,
                "/validator-config.yaml",
            });
            defer self.allocator.free(validator_config_filepath);

            var parsed_validator_config = try utils_lib.loadFromYAMLFile(self.allocator, validator_config_filepath);
            defer parsed_validator_config.deinit(self.allocator);

            // Get ENR fields from validator config
            var enr_fields = try getEnrFieldsFromValidatorConfig(self.allocator, self.options.node_key, parsed_validator_config);
            defer enr_fields.deinit(self.allocator);

            // Construct ENR from fields and private key
            self.enr = try constructENRFromFields(
                self.allocator,
                self.options.local_priv_key,
                enr_fields,
                self.options.is_aggregator,
            );
        }

        // Overriding the IP to 0.0.0.0 to listen on all interfaces
        try self.enr.kvs.put("ip", "\x00\x00\x00\x00");

        var node_multiaddrs = try self.enr.multiaddrP2PQUIC(self.allocator);
        defer node_multiaddrs.deinit(self.allocator);
        // move the ownership to the `EthLibp2p`, will be freed in its deinit
        const listen_addresses = try node_multiaddrs.toOwnedSlice(self.allocator);
        errdefer {
            for (listen_addresses) |addr| addr.deinit();
            self.allocator.free(listen_addresses);
        }
        var connect_peer_list: std.ArrayList(Multiaddr) = .empty;
        defer connect_peer_list.deinit(self.allocator);

        for (self.options.bootnodes, 0..) |n, i| {
            // don't exclude any entry from nodes.yaml if this is not a genesis bootnode
            if (i != self.options.node_key_index or !std.mem.eql(u8, self.options.validator_config, "genesis_bootnode")) {
                var n_enr: ENR = undefined;
                try ENR.decodeTxtInto(&n_enr, n);
                var peer_multiaddr_list = try n_enr.multiaddrP2PQUIC(self.allocator);
                defer peer_multiaddr_list.deinit(self.allocator);
                const peer_multiaddrs = try peer_multiaddr_list.toOwnedSlice(self.allocator);
                defer self.allocator.free(peer_multiaddrs);
                try connect_peer_list.appendSlice(self.allocator, peer_multiaddrs);
            }
        }

        // move the ownership to the `EthLibp2p`, will be freed in its deinit
        const connect_peers = try connect_peer_list.toOwnedSlice(self.allocator);
        errdefer {
            for (connect_peers) |addr| addr.deinit();
            self.allocator.free(connect_peers);
        }

        return .{ .listen_addresses = listen_addresses, .connect_peers = connect_peers };
    }

    fn loadValidatorKeypairs(
        self: *Self,
        num_validators: usize,
    ) !void {
        if (self.options.validator_assignments.len == 0) {
            return error.NoValidatorAssignments;
        }

        const hash_sig_key_dir = self.options.hash_sig_key_dir;

        // First pass: group assignments by validator index, routing by filename.
        // If the filename contains "attester" it goes to att_base; "proposer" to prop_base.
        // A filename with neither is rejected with error.InvalidPrivkeyFileFormat.
        // Slices point into validator_assignments memory which outlives this function.
        const FileSlots = struct {
            att_base: ?[]const u8 = null,
            prop_base: ?[]const u8 = null,
        };
        var file_map = std.AutoHashMap(usize, FileSlots).init(self.allocator);
        defer file_map.deinit();

        for (self.options.validator_assignments) |assignment| {
            if (assignment.index >= num_validators) {
                return error.HashSigValidatorIndexOutOfRange;
            }
            const privkey_file = assignment.privkey_file;
            if (!std.mem.endsWith(u8, privkey_file, "_sk.ssz")) {
                return error.InvalidPrivkeyFileFormat;
            }
            const base = privkey_file[0 .. privkey_file.len - 7]; // Remove "_sk.ssz"
            const slots = try file_map.getOrPutValue(assignment.index, .{});
            if (std.mem.indexOf(u8, privkey_file, "attester") != null) {
                slots.value_ptr.att_base = base;
            } else if (std.mem.indexOf(u8, privkey_file, "proposer") != null) {
                slots.value_ptr.prop_base = base;
            } else {
                // Filename must contain "attester" or "proposer" to unambiguously
                // assign the key to a role. A file with neither is an error.
                return error.InvalidPrivkeyFileFormat;
            }
        }

        // Second pass: load each keypair from disk and register with the key manager.
        // If only one role's file was provided, fall back to using it for both roles.
        var map_it = file_map.iterator();
        while (map_it.next()) |entry| {
            const index = entry.key_ptr.*;
            const slots = entry.value_ptr.*;

            const att_base = slots.att_base orelse slots.prop_base orelse return error.HashSigSecretKeyMissing;
            const prop_base = slots.prop_base orelse slots.att_base orelse return error.HashSigSecretKeyMissing;

            const att_sk = try std.fmt.allocPrint(self.allocator, "{s}/{s}_sk.ssz", .{ hash_sig_key_dir, att_base });
            defer self.allocator.free(att_sk);
            const att_pk = try std.fmt.allocPrint(self.allocator, "{s}/{s}_pk.ssz", .{ hash_sig_key_dir, att_base });
            defer self.allocator.free(att_pk);

            var att_keypair = key_manager_lib.loadKeypairFromFiles(self.allocator, att_sk, att_pk) catch |err| switch (err) {
                error.SecretKeyFileNotFound => return error.HashSigSecretKeyMissing,
                error.PublicKeyFileNotFound => return error.HashSigPublicKeyMissing,
                else => return err,
            };
            errdefer att_keypair.deinit();

            const prop_sk = try std.fmt.allocPrint(self.allocator, "{s}/{s}_sk.ssz", .{ hash_sig_key_dir, prop_base });
            defer self.allocator.free(prop_sk);
            const prop_pk = try std.fmt.allocPrint(self.allocator, "{s}/{s}_pk.ssz", .{ hash_sig_key_dir, prop_base });
            defer self.allocator.free(prop_pk);

            var prop_keypair = key_manager_lib.loadKeypairFromFiles(self.allocator, prop_sk, prop_pk) catch |err| switch (err) {
                error.SecretKeyFileNotFound => return error.HashSigSecretKeyMissing,
                error.PublicKeyFileNotFound => return error.HashSigPublicKeyMissing,
                else => return err,
            };
            errdefer prop_keypair.deinit();

            const validator_keys = key_manager_lib.ValidatorKeys{
                .attestation_keypair = att_keypair,
                .proposal_keypair = prop_keypair,
            };
            try self.key_manager.addKeypair(index, validator_keys);
        }
    }
};

/// Reads ATTESTATION_COMMITTEE_COUNT from a parsed config.yaml Yaml document.
/// Returns null if the field is absent or cannot be parsed.
/// Comma-separated multiaddr string for `EthLibp2pParams.connect_peers`.
/// Returns an allocator-owned slice (empty string for the no-bootnodes case).
fn joinMultiaddrsCsv(allocator: std.mem.Allocator, addrs: []const Multiaddr) ![]u8 {
    if (addrs.len == 0) return try allocator.dupe(u8, "");
    var buf: std.ArrayList(u8) = .empty;
    defer buf.deinit(allocator);
    for (addrs, 0..) |addr, i| {
        if (i > 0) try buf.append(allocator, ',');
        const s = try addr.toString(allocator);
        defer allocator.free(s);
        try buf.appendSlice(allocator, s);
    }
    return try buf.toOwnedSlice(allocator);
}

fn attestationCommitteeCountFromYAML(config: Yaml) ?u64 {
    if (config.docs.items.len == 0) return null;
    const root = config.docs.items[0];
    if (root != .map) return null;
    const value = root.map.get("ATTESTATION_COMMITTEE_COUNT") orelse return null;
    return switch (value) {
        .scalar => |s| std.fmt.parseInt(u64, s, 10) catch null,
        else => null,
    };
}

/// Reads MAX_ATTESTATIONS_DATA from a parsed config.yaml Yaml document.
/// Returns null if the field is absent or cannot be parsed.
fn maxAttestationsDataFromYAML(config: Yaml) ?u8 {
    if (config.docs.items.len == 0) return null;
    const root = config.docs.items[0];
    if (root != .map) return null;
    const value = root.map.get("MAX_ATTESTATIONS_DATA") orelse return null;
    return switch (value) {
        .scalar => |s| std.fmt.parseInt(u8, s, 10) catch null,
        else => null,
    };
}

/// Builds the start options for a node based on the provided command and options.
/// It loads the necessary configuration files, parses them, and populates the
/// `StartNodeOptions` structure.
/// The caller is responsible for freeing the allocated resources in `StartNodeOptions`.
pub fn buildStartOptions(
    allocator: std.mem.Allocator,
    node_cmd: NodeCommand,
    opts: *NodeOptions,
) !void {
    try utils_lib.checkDIRExists(node_cmd.@"custom-genesis");

    const config_filepath = try std.mem.concat(allocator, u8, &[_][]const u8{ node_cmd.@"custom-genesis", "/config.yaml" });
    defer allocator.free(config_filepath);
    const bootnodes_filepath = try std.mem.concat(allocator, u8, &[_][]const u8{ node_cmd.@"custom-genesis", "/nodes.yaml" });
    defer allocator.free(bootnodes_filepath);
    const validators_filepath = try std.mem.concat(allocator, u8, &[_][]const u8{
        if (std.mem.eql(u8, node_cmd.@"validator-config", "genesis_bootnode"))
            //
            node_cmd.@"custom-genesis"
        else
            node_cmd.@"validator-config",
        "/annotated_validators.yaml",
    });
    defer allocator.free(validators_filepath);
    const validator_config_filepath = try std.mem.concat(allocator, u8, &[_][]const u8{
        if (std.mem.eql(u8, node_cmd.@"validator-config", "genesis_bootnode"))
            //
            node_cmd.@"custom-genesis"
        else
            node_cmd.@"validator-config",
        "/validator-config.yaml",
    });
    defer allocator.free(validator_config_filepath);
    // TODO: support genesis file loading when ssz library supports it
    // const genesis_filepath = try std.mem.concat(allocator, &[_][]const u8{custom_genesis, "/genesis.ssz"});
    // defer allocator.free(genesis_filepath);

    var parsed_bootnodes = try utils_lib.loadFromYAMLFile(allocator, bootnodes_filepath);
    defer parsed_bootnodes.deinit(allocator);

    var parsed_config = try utils_lib.loadFromYAMLFile(allocator, config_filepath);
    defer parsed_config.deinit(allocator);

    var parsed_validators = try utils_lib.loadFromYAMLFile(allocator, validators_filepath);
    defer parsed_validators.deinit(allocator);

    var parsed_validator_config = try utils_lib.loadFromYAMLFile(allocator, validator_config_filepath);
    defer parsed_validator_config.deinit(allocator);

    const bootnodes = try nodesFromYAML(allocator, parsed_bootnodes);
    errdefer {
        for (bootnodes) |b| allocator.free(b);
        allocator.free(bootnodes);
    }
    if (bootnodes.len == 0) {
        return error.InvalidNodesConfig;
    }
    const genesis_spec = try configs.genesisConfigFromYAML(allocator, parsed_config, node_cmd.@"override-genesis-time");

    const validator_assignments = try validatorAssignmentsFromYAML(allocator, opts.node_key, parsed_validators);
    errdefer {
        for (validator_assignments) |*a| {
            @constCast(a).deinit(allocator);
        }
        allocator.free(validator_assignments);
    }
    if (validator_assignments.len == 0) {
        return error.InvalidValidatorConfig;
    }
    const local_priv_key = try getPrivateKeyFromValidatorConfig(allocator, opts.node_key, parsed_validator_config);

    const node_key_index = try nodeKeyIndexFromYaml(opts.node_key, parsed_validator_config);

    const hash_sig_key_dir = try std.mem.concat(allocator, u8, &[_][]const u8{
        node_cmd.@"custom-genesis",
        "/",
        node_cmd.@"sig-keys-dir",
    });

    // Populate node name registry with peer information
    populateNodeNameRegistry(allocator, opts.node_registry, validator_config_filepath, validators_filepath) catch |err| {
        std.log.warn("Failed to populate node name registry: {any}", .{err});
    };

    opts.bootnodes = bootnodes;
    opts.validator_assignments = validator_assignments;
    opts.local_priv_key = local_priv_key;
    opts.genesis_spec = genesis_spec;
    opts.node_key_index = node_key_index;
    opts.hash_sig_key_dir = hash_sig_key_dir;
    opts.checkpoint_sync_url = node_cmd.@"checkpoint-sync-url";
    opts.chain_spec = node_cmd.@"chain-spec";
    opts.is_aggregator = node_cmd.@"is-aggregator";
    opts.chain_worker_enabled = node_cmd.@"chain-worker";

    // Parse --aggregate-subnet-ids (comma-separated list of subnet ids, e.g. "0,1,2")
    // Require --is-aggregator to be set when --aggregate-subnet-ids is provided.
    if (node_cmd.@"aggregate-subnet-ids" != null and !node_cmd.@"is-aggregator") {
        std.log.err("--aggregate-subnet-ids requires --is-aggregator to be set", .{});
        return error.AggregateSubnetIdsRequiresIsAggregator;
    }
    if (node_cmd.@"aggregate-subnet-ids") |subnet_ids_str| {
        var list: std.ArrayList(u32) = .empty;
        var it = std.mem.splitScalar(u8, subnet_ids_str, ',');
        while (it.next()) |part| {
            const trimmed = std.mem.trim(u8, part, " ");
            if (trimmed.len == 0) continue;
            const id = std.fmt.parseInt(u32, trimmed, 10) catch |err| {
                std.log.warn("invalid subnet id '{s}': {any}", .{ trimmed, err });
                list.deinit(allocator);
                return error.InvalidSubnetId;
            };
            try list.append(allocator, id);
        }
        opts.aggregation_subnet_ids = try list.toOwnedSlice(allocator);
    }

    // Resolve attestation_committee_count: CLI flag takes precedence over config.yaml.
    if (node_cmd.@"attestation-committee-count") |count| {
        opts.attestation_committee_count = count;
    } else {
        // Try to read ATTESTATION_COMMITTEE_COUNT from config.yaml
        opts.attestation_committee_count = attestationCommitteeCountFromYAML(parsed_config);
    }

    // Validate: attestation_committee_count must be >= 1.
    // If the resolved value is 0 (an invalid input), log a warning and fall back to 1.
    if (opts.attestation_committee_count) |count| {
        if (count == 0) {
            std.log.warn(
                "attestation-committee-count must be >= 1 (got 0); defaulting to 1",
                .{},
            );
            opts.attestation_committee_count = 1;
        }
    }

    // Resolve max_attestations_data from config.yaml (no CLI flag; defaults to 16 in ChainConfig).
    opts.max_attestations_data = maxAttestationsDataFromYAML(parsed_config);
}

/// Downloads finalized checkpoint state from the given URL and deserializes it
/// Returns the deserialized state. The caller is responsible for calling deinit on it.
fn downloadCheckpointState(
    allocator: std.mem.Allocator,
    url: []const u8,
    logger: zeam_utils.ModuleLogger,
) !types.BeamState {
    logger.info("downloading checkpoint state from: {s}", .{url});

    const io = std.Io.Threaded.global_single_threaded.io();
    var client = std.http.Client{
        .allocator = allocator,
        .io = io,
    };
    defer client.deinit();

    // Use an Allocating writer so client.fetch handles both Content-Length and
    // Transfer-Encoding: chunked transparently. The previous manual readSliceShort
    // loop panicked when the server switched to chunked encoding for responses
    // larger than ~3 MB because readSliceShort → readVec → defaultReadVec →
    // contentLengthStream panics when the body union field is 'ready' (chunked)
    // rather than 'body_remaining_content_length'.
    var body_writer = std.Io.Writer.Allocating.init(allocator);
    defer body_writer.deinit();

    const result = client.fetch(.{
        .location = .{ .url = url },
        .method = .GET,
        .response_writer = &body_writer.writer,
    }) catch |err| {
        logger.err("checkpoint sync request failed: {any}", .{err});
        return error.RequestFailed;
    };

    if (result.status != .ok) {
        logger.err("checkpoint sync failed: HTTP {d}", .{@intFromEnum(result.status)});
        return error.HttpError;
    }

    // Transfer ownership out of the writer (writer buffer becomes empty so the
    // deferred deinit above is safe to call).
    var ssz_data = body_writer.toArrayList();
    defer ssz_data.deinit(allocator);

    logger.info("downloaded checkpoint state: {d} bytes", .{ssz_data.items.len});

    // Deserialize SSZ state
    // Use arena allocator for deserialization as SSZ types may allocate
    var arena = std.heap.ArenaAllocator.init(allocator);
    defer arena.deinit();

    var checkpoint_state: types.BeamState = undefined;
    try ssz.deserialize(types.BeamState, ssz_data.items, &checkpoint_state, arena.allocator());

    logger.info("successfully deserialized checkpoint state at slot {d}", .{checkpoint_state.slot});

    // Clone the state to move it out of the arena using the proper cloning function
    const cloned_state = try zeam_utils.clone(types.BeamState, &checkpoint_state, allocator);

    return cloned_state;
}

/// Verifies checkpoint state against the genesis configuration
/// Validates that the downloaded state is consistent with expected genesis parameters
/// Also computes and logs the state root and block root
fn verifyCheckpointState(
    allocator: std.mem.Allocator,
    state: *const types.BeamState,
    genesis_spec: *const types.GenesisSpec,
    logger: zeam_utils.ModuleLogger,
) !void {
    // Verify genesis timestamp matches
    if (state.config.genesis_time != genesis_spec.genesis_time) {
        logger.err("checkpoint state verification failed: genesis time mismatch (expected={d}, got={d})", .{
            genesis_spec.genesis_time,
            state.config.genesis_time,
        });
        return error.GenesisTimeMismatch;
    }

    // Verify number of validators matches genesis config
    const expected_validators = genesis_spec.numValidators();
    const actual_validators = state.validators.len();
    if (actual_validators != expected_validators) {
        logger.err("checkpoint state verification failed: validator count mismatch (expected={d}, got={d})", .{
            expected_validators,
            actual_validators,
        });
        return error.ValidatorCountMismatch;
    }

    // Verify state has validators
    if (actual_validators == 0) {
        logger.err("checkpoint state verification failed: no validators in state", .{});
        return error.NoValidators;
    }

    // Verify each validator pubkey matches genesis config
    const state_validators = state.validators.constSlice();
    for (genesis_spec.validator_attestation_pubkeys, genesis_spec.validator_proposal_pubkeys, 0..) |expected_att_pubkey, expected_prop_pubkey, i| {
        const actual_att_pubkey = state_validators[i].attestation_pubkey;
        if (!std.mem.eql(u8, &expected_att_pubkey, &actual_att_pubkey)) {
            logger.err("checkpoint state verification failed: attestation pubkey mismatch at index {d}", .{i});
            return error.ValidatorAttestationPubkeyMismatch;
        }
        const actual_prop_pubkey = state_validators[i].proposal_pubkey;
        if (!std.mem.eql(u8, &expected_prop_pubkey, &actual_prop_pubkey)) {
            logger.err("checkpoint state verification failed: proposal pubkey mismatch at index {d}", .{i});
            return error.ValidatorProposalPubkeyMismatch;
        }
    }

    // Generate state block header with correct state_root
    // (latest_block_header.state_root is zero; genStateBlockHeader computes and sets it)
    const state_block_header = try state.genStateBlockHeader(allocator);

    // Calculate the block root from the properly constructed block header
    var block_root: types.Root = undefined;
    try zeam_utils.hashTreeRoot(types.BeamBlockHeader, state_block_header, &block_root, allocator);

    logger.info("checkpoint state verified: slot={d}, genesis_time={d}, validators={d}, state_root=0x{x}, block_root=0x{x}", .{
        state.slot,
        state.config.genesis_time,
        actual_validators,
        &state_block_header.state_root,
        &block_root,
    });
}

/// Path suffix that the checkpoint-sync state URL must end with.
/// Used to derive the block URL (same base, different path tail).
const FINALIZED_STATE_PATH = "/lean/v0/states/finalized";
/// Path for the finalized block endpoint (spec FINALIZED_BLOCK_ENDPOINT).
const FINALIZED_BLOCK_PATH = "/lean/v0/blocks/finalized";

/// Tries to fetch the real SignedBlock for the checkpoint anchor from the
/// checkpoint provider and persist it to the DB.
///
/// The block URL is derived from the state URL by stripping the known
/// `FINALIZED_STATE_PATH` suffix and appending `FINALIZED_BLOCK_PATH`.
///
/// On any failure the error is logged and the function returns without storing
/// anything — a missing anchor block is non-fatal: blocks_by_root will return
/// empty for this root until the real block arrives via reqresp or gossip.
fn downloadAndStoreCheckpointBlock(
    allocator: std.mem.Allocator,
    state_url: []const u8,
    expected_root: types.Root,
    /// hash_tree_root(anchor_state) — required. Block's state_root is checked
    /// against this for block/state pairing (spec fetch_finalized_anchor).
    expected_state_root: types.Root,
    db: *database.Db,
    logger: zeam_utils.ModuleLogger,
) void {
    // Derive block URL: strip the known state path suffix and append the block path.
    // Using endsWith avoids ambiguous first-occurrence matches in hostname/query params.
    const trimmed = std.mem.trimEnd(u8, state_url, "/");
    const base_url = if (std.mem.endsWith(u8, trimmed, FINALIZED_STATE_PATH))
        trimmed[0 .. trimmed.len - FINALIZED_STATE_PATH.len]
    else {
        logger.warn(
            "checkpoint block fetch: state URL '{s}' does not end with '{s}', cannot derive block URL",
            .{ state_url, FINALIZED_STATE_PATH },
        );
        return;
    };
    const block_url = std.fmt.allocPrint(allocator, "{s}{s}", .{ base_url, FINALIZED_BLOCK_PATH }) catch |err| {
        logger.warn("checkpoint block fetch: failed to allocate block URL: {}", .{err});
        return;
    };
    defer allocator.free(block_url);

    logger.info("checkpoint block fetch: downloading anchor block from: {s}", .{block_url});

    const io = std.Io.Threaded.global_single_threaded.io();
    var client = std.http.Client{ .allocator = allocator, .io = io };
    defer client.deinit();

    var body_writer = std.Io.Writer.Allocating.init(allocator);
    defer body_writer.deinit();

    const result = client.fetch(.{
        .location = .{ .url = block_url },
        .method = .GET,
        .response_writer = &body_writer.writer,
        // Spec: Accept: application/octet-stream for SSZ binary response.
        .extra_headers = &.{.{ .name = "Accept", .value = "application/octet-stream" }},
    }) catch |err| {
        logger.warn("checkpoint block fetch: HTTP request failed: {}", .{err});
        return;
    };

    if (result.status != .ok) {
        logger.warn("checkpoint block fetch: HTTP {d} from {s}", .{ @intFromEnum(result.status), block_url });
        return;
    }

    var ssz_data = body_writer.toArrayList();
    defer ssz_data.deinit(allocator);

    // Deserialize into arena so block fields don't need explicit cleanup.
    var arena = std.heap.ArenaAllocator.init(allocator);
    defer arena.deinit();

    var block: types.SignedBlock = undefined;
    ssz.deserialize(types.SignedBlock, ssz_data.items, &block, arena.allocator()) catch |err| {
        logger.warn("checkpoint block fetch: SSZ deserialize failed: {}", .{err});
        return;
    };

    // Verify block root: hash_tree_root(BeamBlock) must equal expected_root.
    // Fail closed: if hashing fails something is seriously wrong (hashTreeRoot is
    // deterministic over well-typed values); drop rather than store unverified.
    var computed_root: types.Root = undefined;
    zeam_utils.hashTreeRoot(types.BeamBlock, block.block, &computed_root, allocator) catch |err| {
        logger.warn("checkpoint block fetch: hash verification failed: {}, discarding", .{err});
        return;
    };
    if (!std.mem.eql(u8, &computed_root, &expected_root)) {
        logger.warn("checkpoint block fetch: root mismatch — computed=0x{x} expected=0x{x}, discarding", .{ &computed_root, &expected_root });
        return;
    }

    // Pairing check: block.state_root must equal hash_tree_root(anchor_state).
    // This is the finalized-anchor assertion: it detects a server advancing
    // finalization between the two requests (state then block).
    if (!std.mem.eql(u8, &block.block.state_root, &expected_state_root)) {
        logger.warn(
            "checkpoint block fetch: anchor block/state mismatch — block.state_root=0x{x} hash_tree_root(state)=0x{x}; server may have advanced finalization, discarding",
            .{ &block.block.state_root, &expected_state_root },
        );
        return;
    }

    // Store the original SSZ bytes directly — no re-serialise round-trip.
    // The bytes already passed deserialise + hash_tree_root verification so they
    // are valid; re-encoding would waste CPU and could diverge on non-canonical
    // encodings (though hash_tree_root catches those too).
    var batch = db.initWriteBatch() catch |err| {
        logger.warn("checkpoint block fetch: write-batch init failed: {}", .{err});
        return;
    };
    defer batch.deinit();
    batch.putBlockSerialized(database.DbBlocksNamespace, expected_root, ssz_data.items);
    // Also index the checkpoint block in the finalized slot index so that
    // blocks_by_range can serve it.  Without this, `onReqRespRequest`'s
    // finalized-range path (slot <= finalized_slot) finds no entry in
    // DbFinalizedSlotsNamespace and the unfinalized-range walk is skipped
    // (its guard is `end_slot_exclusive > finalized_slot + 1`, which is false
    // when start_slot == finalized_slot == checkpoint_slot).  The result is an
    // empty blocks_by_range response for the checkpoint slot even though the
    // block is present in DbBlocksNamespace and is served correctly via
    // blocks_by_root.
    batch.putFinalizedSlotIndex(database.DbFinalizedSlotsNamespace, block.block.slot, expected_root);
    db.commit(&batch) catch |err| {
        logger.warn("checkpoint block fetch: DB commit failed: {}", .{err});
        return;
    };

    logger.info(
        "checkpoint block fetch: stored real anchor block root=0x{x} slot={d} ({d} bytes)",
        .{ &expected_root, block.block.slot, ssz_data.items.len },
    );
}

/// Parses the nodes from a YAML configuration.
/// Expects a YAML structure like:
/// ```yaml
///   - enr1...
///   - enr2...
/// ```
/// Returns a set of ENR strings. The caller is responsible for freeing the returned slice.
fn nodesFromYAML(allocator: std.mem.Allocator, nodes_config: Yaml) ![]const []const u8 {
    // Directly access yaml structure to avoid yaml library's parse() double allocation bug
    if (nodes_config.docs.items.len == 0) return error.InvalidYamlShape;
    const root = nodes_config.docs.items[0];
    if (root != .list) return error.InvalidYamlShape;

    var nodes = try allocator.alloc([]const u8, root.list.len);
    errdefer {
        for (nodes) |node| allocator.free(node);
        allocator.free(nodes);
    }

    for (root.list, 0..) |item, i| {
        if (item != .scalar) return error.InvalidYamlShape;
        nodes[i] = try allocator.dupe(u8, item.scalar);
    }

    return nodes;
}

/// Parses the validator indices for a given node from a YAML configuration.
/// Expects a YAML structure like:
/// ```yaml
/// node_0:
///   - 0
///   - 1
/// node_1:
/// Parses the validator assignments for a given node from a YAML configuration.
/// Expects a YAML structure like:
/// ```yaml
/// zeam_0:
///   - index: 0
///     pubkey_hex: 812f8540481ce70515d43b451cedcf6b4e3177312821fa541df6375c20ced55196ad245a00335d425e86817bdbde75536c986c25
///     privkey_file: validator_0_sk.json
///   - index: 3
///     pubkey_hex: a47e01144cd43e3efddef56da069f418d9aac406c29589314f74b00158437375e51b1213ecbbf23d47fd9e05933cfb24ac84e72d
///     privkey_file: validator_3_sk.json
/// ```
/// where `node_key` (e.g., "zeam_0") is the key for the node's validator assignments.
/// Returns a slice of ValidatorAssignment. The caller is responsible for freeing the returned slice.
fn validatorAssignmentsFromYAML(allocator: std.mem.Allocator, node_key: []const u8, validators: Yaml) ![]ValidatorAssignment {
    var assignments: std.ArrayList(ValidatorAssignment) = .empty;
    defer assignments.deinit(allocator);
    errdefer {
        for (assignments.items) |*a| {
            @constCast(a).deinit(allocator);
        }
    }

    const node_validators = validators.docs.items[0].map.get(node_key) orelse return error.InvalidNodeKey;
    if (node_validators != .list) return error.InvalidValidatorConfig;

    for (node_validators.list) |item| {
        if (item != .map) return error.InvalidValidatorConfig;

        const index_value = item.map.get("index") orelse return error.InvalidValidatorConfig;
        if (index_value != .scalar) return error.InvalidValidatorConfig;

        const pubkey_value = item.map.get("pubkey_hex") orelse return error.InvalidValidatorConfig;
        if (pubkey_value != .scalar) return error.InvalidValidatorConfig;

        const privkey_value = item.map.get("privkey_file") orelse return error.InvalidValidatorConfig;
        if (privkey_value != .scalar) return error.InvalidValidatorConfig;

        const assignment = ValidatorAssignment{
            .index = @intCast(std.fmt.parseInt(usize, index_value.scalar, 10) catch return error.InvalidValidatorConfig),
            .pubkey_hex = try allocator.dupe(u8, pubkey_value.scalar),
            .privkey_file = try allocator.dupe(u8, privkey_value.scalar),
        };
        try assignments.append(allocator, assignment);
    }
    return try assignments.toOwnedSlice(allocator);
}

// Parses the index for a given node key from a YAML configuration.
// ```yaml
// shuffle: roundrobin
// validators:
//   - name: "zeam_0"
//     # node id 7d0904dc6d8d7130e0e68d5d3175d0c3cf470f8725f67bd8320882f5b9753cc0
//     # peer id 16Uiu2HAkvi2sxT75Bpq1c7yV2FjnSQJJ432d6jeshbmfdJss1i6f
//     privkey: "bdf953adc161873ba026330c56450453f582e3c4ee6cb713644794bcfdd85fe5"
//     enrFields:
//       # verify /ip4/127.0.0.1/udp/9000/quic-v1/p2p/16Uiu2HAkvi2sxT75Bpq1c7yV2FjnSQJJ432d6jeshbmfdJss1i6f
//       ip: "127.0.0.1"
//       quic: 9000
//     count: 1 # number of indices for this node
//```

fn nodeKeyIndexFromYaml(node_key: []const u8, validator_config: Yaml) !usize {
    for (validator_config.docs.items[0].map.get("validators").?.list, 0..) |entry, index| {
        const name_value = entry.map.get("name").?;
        if (name_value == .scalar and std.mem.eql(u8, name_value.scalar, node_key)) {
            return index;
        }
    }
    return error.InvalidNodeKey;
}

fn getPrivateKeyFromValidatorConfig(allocator: std.mem.Allocator, node_key: []const u8, validator_config: Yaml) ![]const u8 {
    for (validator_config.docs.items[0].map.get("validators").?.list) |entry| {
        const name_value = entry.map.get("name").?;
        if (name_value == .scalar and std.mem.eql(u8, name_value.scalar, node_key)) {
            const privkey_value = entry.map.get("privkey").?;
            if (privkey_value == .scalar) {
                return try allocator.dupe(u8, privkey_value.scalar);
            } else {
                return error.InvalidPrivateKeyFormat;
            }
        }
    }
    return error.InvalidNodeKey;
}

fn getIsAggregatorFromValidatorConfig(node_key: []const u8, validator_config: Yaml) !bool {
    for (validator_config.docs.items[0].map.get("validators").?.list) |entry| {
        const name_value = entry.map.get("name").?;
        if (name_value == .scalar and std.mem.eql(u8, name_value.scalar, node_key)) {
            const value = entry.map.get("is_aggregator") orelse return false;
            return switch (value) {
                .boolean => |b| b,
                .scalar => |s| blk: {
                    if (std.ascii.eqlIgnoreCase(s, "true")) break :blk true;
                    if (std.ascii.eqlIgnoreCase(s, "false")) break :blk false;
                    const i = std.fmt.parseInt(i64, s, 10) catch break :blk error.InvalidAggregatorFlag;
                    return switch (i) {
                        0 => false,
                        1 => true,
                        else => error.InvalidAggregatorFlag,
                    };
                },
                else => error.InvalidAggregatorFlag,
            };
        }
    }
    return error.InvalidNodeKey;
}

fn getEnrFieldsFromValidatorConfig(allocator: std.mem.Allocator, node_key: []const u8, validator_config: Yaml) !EnrFields {
    for (validator_config.docs.items[0].map.get("validators").?.list) |entry| {
        const name_value = entry.map.get("name").?;
        if (name_value == .scalar and std.mem.eql(u8, name_value.scalar, node_key)) {
            const enr_fields_value = entry.map.get("enrFields");
            if (enr_fields_value == null) {
                return error.MissingEnrFields;
            }

            var enr_fields = EnrFields{
                .custom_fields = std.StringHashMap([]const u8).init(allocator),
            };
            errdefer enr_fields.deinit(allocator);

            const fields_map = enr_fields_value.?.map;

            // Parse known fields
            if (fields_map.get("ip")) |ip_value| {
                if (ip_value == .scalar) {
                    enr_fields.ip = try allocator.dupe(u8, ip_value.scalar);
                }
            }

            if (fields_map.get("ip6")) |ip6_value| {
                if (ip6_value == .scalar) {
                    enr_fields.ip6 = try allocator.dupe(u8, ip6_value.scalar);
                }
            }

            if (fields_map.get("tcp")) |tcp_value| {
                if (tcp_value == .scalar) {
                    enr_fields.tcp = std.fmt.parseInt(u16, tcp_value.scalar, 10) catch 0;
                }
            }

            if (fields_map.get("udp")) |udp_value| {
                if (udp_value == .scalar) {
                    enr_fields.udp = std.fmt.parseInt(u16, udp_value.scalar, 10) catch 0;
                }
            }

            if (fields_map.get("quic")) |quic_value| {
                if (quic_value == .scalar) {
                    enr_fields.quic = std.fmt.parseInt(u16, quic_value.scalar, 10) catch 0;
                }
            }

            if (fields_map.get("seq")) |seq_value| {
                if (seq_value == .scalar) {
                    enr_fields.seq = std.fmt.parseInt(u64, seq_value.scalar, 10) catch 0;
                }
            }

            // Parse custom fields
            var iterator = fields_map.iterator();
            while (iterator.next()) |kv| {
                const key = kv.key_ptr.*;
                const value = kv.value_ptr.*;

                // Skip known fields
                if (std.mem.eql(u8, key, "ip") or
                    std.mem.eql(u8, key, "ip6") or
                    std.mem.eql(u8, key, "tcp") or
                    std.mem.eql(u8, key, "udp") or
                    std.mem.eql(u8, key, "quic") or
                    std.mem.eql(u8, key, "seq"))
                {
                    continue;
                }

                // Handle custom field based on type
                if (value == .scalar) {
                    const key_copy = try allocator.dupe(u8, key);
                    const value_copy = try allocator.dupe(u8, value.scalar);
                    try enr_fields.custom_fields.put(key_copy, value_copy);
                }
            }

            return enr_fields;
        }
    }
    return error.InvalidNodeKey;
}

fn constructENRFromFields(
    allocator: std.mem.Allocator,
    private_key: []const u8,
    enr_fields: EnrFields,
    is_aggregator: bool,
) !ENR {
    // Clean up private key (remove 0x prefix if present)
    const secret_key_str = if (std.mem.startsWith(u8, private_key, "0x"))
        private_key[2..]
    else
        private_key;

    if (secret_key_str.len != 64) {
        return error.InvalidSecretKeyLength;
    }

    // Create SignableENR from private key
    var signable_enr = enr_lib.SignableENR.fromSecretKeyString(secret_key_str) catch {
        return error.ENRCreationFailed;
    };

    // Set IP address (IPv4)
    if (enr_fields.ip) |ip_str| {
        const ip_addr = std.Io.net.Ip4Address.parse(ip_str, 0) catch {
            return error.InvalidIPAddress;
        };
        const ip_addr_bytes = &ip_addr.bytes;
        signable_enr.set("ip", ip_addr_bytes) catch {
            return error.ENRSetIPFailed;
        };
    }

    // Set IP address (IPv6)
    if (enr_fields.ip6) |ip6_str| {
        const ip6_addr = std.Io.net.Ip6Address.parse(ip6_str, 0) catch {
            return error.InvalidIP6Address;
        };
        const ip6_addr_bytes = &ip6_addr.bytes;
        signable_enr.set("ip6", ip6_addr_bytes) catch {
            return error.ENRSetIP6Failed;
        };
    }

    // Set TCP port
    if (enr_fields.tcp) |tcp_port| {
        var tcp_bytes: [2]u8 = undefined;
        std.mem.writeInt(u16, &tcp_bytes, tcp_port, .big);
        signable_enr.set("tcp", &tcp_bytes) catch {
            return error.ENRSetTCPFailed;
        };
    }

    // Set UDP port
    if (enr_fields.udp) |udp_port| {
        var udp_bytes: [2]u8 = undefined;
        std.mem.writeInt(u16, &udp_bytes, udp_port, .big);
        signable_enr.set("udp", &udp_bytes) catch {
            return error.ENRSetUDPFailed;
        };
    }

    // Set QUIC port
    if (enr_fields.quic) |quic_port| {
        var quic_bytes: [2]u8 = undefined;
        std.mem.writeInt(u16, &quic_bytes, quic_port, .big);
        signable_enr.set("quic", &quic_bytes) catch {
            return error.ENRSetQUICFailed;
        };
    }

    // Set sequence number
    if (enr_fields.seq) |seq_num| {
        var seq_bytes: [8]u8 = undefined;
        std.mem.writeInt(u64, &seq_bytes, seq_num, .big);
        signable_enr.set("seq", &seq_bytes) catch {
            return error.ENRSetSEQFailed;
        };
    }

    // Advertise aggregator capability in ENR.
    // 0x00 = false, 0x01 = true.
    const is_aggregator_bytes = [_]u8{if (is_aggregator) 0x01 else 0x00};
    signable_enr.set("is_aggregator", &is_aggregator_bytes) catch {
        return error.ENRSetIsAggregatorFailed;
    };

    // Set custom fields
    var custom_iterator = enr_fields.custom_fields.iterator();
    while (custom_iterator.next()) |kv| {
        const key = kv.key_ptr.*;
        const value = kv.value_ptr.*;

        // Try to parse as hex if it starts with 0x
        if (std.mem.startsWith(u8, value, "0x")) {
            const hex_value = value[2..];
            if (hex_value.len % 2 != 0) {
                return error.InvalidHexValue;
            }
            const bytes = try allocator.alloc(u8, hex_value.len / 2);
            defer allocator.free(bytes);

            _ = std.fmt.hexToBytes(bytes, hex_value) catch {
                return error.InvalidHexFormat;
            };

            signable_enr.set(key, bytes) catch {
                return error.ENRSetCustomFieldFailed;
            };
        } else {
            // Treat as string
            signable_enr.set(key, value) catch {
                return error.ENRSetCustomFieldFailed;
            };
        }
    }

    // Convert SignableENR to ENR
    var writer_alloc: std.Io.Writer.Allocating = .init(allocator);
    defer writer_alloc.deinit();

    try enr_lib.writeSignableENR(&writer_alloc.writer, &signable_enr);
    const enr_text = writer_alloc.writer.buffered();

    var enr: ENR = undefined;
    try ENR.decodeTxtInto(&enr, enr_text);

    return enr;
}

/// Populate a NodeNameRegistry from validator-config.yaml and validators.yaml
/// This creates mappings from peer IDs and validator indices to node names
pub fn populateNodeNameRegistry(
    allocator: std.mem.Allocator,
    registry: *node_lib.NodeNameRegistry,
    validator_config_path: []const u8,
    validators_path: []const u8,
) !void {

    // Parse validator-config.yaml to get node names and their ENRs/privkeys
    var parsed_validator_config = try utils_lib.loadFromYAMLFile(allocator, validator_config_path);
    defer parsed_validator_config.deinit(allocator);

    // Parse validators.yaml to get validator indices for each node
    var parsed_validators = try utils_lib.loadFromYAMLFile(allocator, validators_path);
    defer parsed_validators.deinit(allocator);

    const validators_list = parsed_validator_config.docs.items[0].map.get("validators");
    if (validators_list == null) return;

    for (validators_list.?.list) |entry| {
        const name_value = entry.map.get("name");
        if (name_value == null or name_value.? != .scalar) continue;
        const node_name = name_value.?.scalar;

        // Get peer ID from ENR or private key
        const peer_id_str = blk: {
            var peer_id_buf: [256]u8 = undefined;
            // Try to get ENR first
            if (entry.map.get("enr")) |enr_value| {
                if (enr_value == .scalar) {
                    var enr: ENR = undefined;
                    ENR.decodeTxtInto(&enr, enr_value.scalar) catch break :blk null;
                    const pid = enr.peerId(allocator) catch break :blk null;
                    const pid_str_slice = pid.toBase58(&peer_id_buf) catch break :blk null;
                    const pid_str = allocator.dupe(u8, pid_str_slice) catch break :blk null;
                    break :blk pid_str;
                }
            }

            // Try to construct ENR from privkey and enrFields
            if (entry.map.get("privkey")) |privkey_value| {
                if (privkey_value == .scalar) {
                    const enr_fields_value = entry.map.get("enrFields");
                    if (enr_fields_value != null) {
                        const is_aggregator = getIsAggregatorFromValidatorConfig(node_name, parsed_validator_config) catch break :blk null;
                        var enr_fields = getEnrFieldsFromValidatorConfig(allocator, node_name, parsed_validator_config) catch break :blk null;
                        defer enr_fields.deinit(allocator);
                        var enr = constructENRFromFields(allocator, privkey_value.scalar, enr_fields, is_aggregator) catch break :blk null;
                        defer enr.deinit();
                        const pid = enr.peerId(allocator) catch break :blk null;
                        const pid_str_slice = pid.toBase58(&peer_id_buf) catch break :blk null;
                        const pid_str = allocator.dupe(u8, pid_str_slice) catch break :blk null;
                        break :blk pid_str;
                    }
                }
            }
            break :blk null;
        };

        // Add peer ID mapping if we got a valid peer ID string
        if (peer_id_str) |pid_str| {
            defer allocator.free(pid_str);
            registry.addPeerMapping(pid_str, node_name) catch |err| {
                std.log.warn("Failed to add peer mapping for node {s}: {any}", .{ node_name, err });
            };
        }

        // Add validator index mappings
        const node_validators = parsed_validators.docs.items[0].map.get(node_name);
        if (node_validators) |validators| {
            if (validators == .list) {
                for (validators.list) |item| {
                    if (item == .scalar) {
                        const validator_index: usize = std.fmt.parseInt(usize, item.scalar, 10) catch continue;
                        registry.addValidatorMapping(validator_index, node_name) catch |err| {
                            std.log.warn("Failed to add validator mapping for node {s} index {d}: {any}", .{ node_name, validator_index, err });
                        };
                    }
                }
            }
        }
    }
}

test "configs yaml parsing" {
    var config_file = try utils_lib.loadFromYAMLFile(std.testing.allocator, "pkgs/cli/test/fixtures/config.yaml");
    defer config_file.deinit(std.testing.allocator);
    const genesis_spec = try configs.genesisConfigFromYAML(std.testing.allocator, config_file, null);
    defer std.testing.allocator.free(genesis_spec.validator_attestation_pubkeys);
    defer std.testing.allocator.free(genesis_spec.validator_proposal_pubkeys);
    try std.testing.expectEqual(@as(u64, 9), genesis_spec.numValidators());
    try std.testing.expectEqual(@as(u64, 1704085200), genesis_spec.genesis_time);

    var validators_file = try utils_lib.loadFromYAMLFile(std.testing.allocator, "pkgs/cli/test/fixtures/annotated_validators.yaml");
    defer validators_file.deinit(std.testing.allocator);
    const validator_assignments = try validatorAssignmentsFromYAML(std.testing.allocator, "zeam_0", validators_file);
    defer {
        for (validator_assignments) |*a| {
            @constCast(a).deinit(std.testing.allocator);
        }
        std.testing.allocator.free(validator_assignments);
    }
    try std.testing.expectEqual(3, validator_assignments.len);
    try std.testing.expectEqual(1, validator_assignments[0].index);
    try std.testing.expectEqual(4, validator_assignments[1].index);
    try std.testing.expectEqual(7, validator_assignments[2].index);

    var nodes_file = try utils_lib.loadFromYAMLFile(std.testing.allocator, "pkgs/cli/test/fixtures/nodes.yaml");
    defer nodes_file.deinit(std.testing.allocator);
    const nodes = try nodesFromYAML(std.testing.allocator, nodes_file);
    defer {
        for (nodes) |node| std.testing.allocator.free(node);
        std.testing.allocator.free(nodes);
    }
    try std.testing.expectEqual(3, nodes.len);
    try std.testing.expectEqualStrings("enr:-IW4QA0pljjdLfxS_EyUxNAxJSoGCwmOVNJauYWsTiYHyWG5Bky-7yCEktSvu_w-PWUrmzbc8vYL_Mx5pgsAix2OfOMBgmlkgnY0gmlwhKwUAAGEcXVpY4IfkIlzZWNwMjU2azGhA6mw8mfwe-3TpjMMSk7GHe3cURhOn9-ufyAqy40wEyui", nodes[0]);
    try std.testing.expectEqualStrings("enr:-IW4QNx7F6OKXCmx9igmSwOAOdUEiQ9Et73HNygWV1BbuFgkXZLMslJVgpLYmKAzBF-AO0qJYq40TtqvtFkfeh2jzqYBgmlkgnY0gmlwhKwUAAKEcXVpY4IfkIlzZWNwMjU2azGhA2hqUIfSG58w4lGPMiPp9llh1pjFuoSRUuoHmwNdHELw", nodes[1]);
    try std.testing.expectEqualStrings("enr:-IW4QOh370UNQipE8qYlVRK3MpT7I0hcOmrTgLO9agIxuPS2B485Se8LTQZ4Rhgo6eUuEXgMAa66Wt7lRYNHQo9zk8QBgmlkgnY0gmlwhKwUAAOEcXVpY4IfkIlzZWNwMjU2azGhA7NTxgfOmGE2EQa4HhsXxFOeHdTLYIc2MEBczymm9IUN", nodes[2]);
}

test "ENR fields parsing from validator config" {
    var validator_config = try utils_lib.loadFromYAMLFile(std.testing.allocator, "pkgs/cli/test/fixtures/validator-config.yaml");
    defer validator_config.deinit(std.testing.allocator);

    // Test parsing ENR fields for zeam_0
    var enr_fields = try getEnrFieldsFromValidatorConfig(std.testing.allocator, "zeam_0", validator_config);
    defer enr_fields.deinit(std.testing.allocator);

    // Verify the parsed fields match expected values
    try std.testing.expectEqualStrings("172.20.0.100", enr_fields.ip.?);
    try std.testing.expectEqual(@as(u16, 9000), enr_fields.tcp.?);
    try std.testing.expectEqual(@as(u16, 9001), enr_fields.quic.?);
    try std.testing.expectEqual(@as(u64, 1), enr_fields.seq.?);

    // Test parsing ENR fields for quadrivium_0
    var enr_fields_1 = try getEnrFieldsFromValidatorConfig(std.testing.allocator, "quadrivium_0", validator_config);
    defer enr_fields_1.deinit(std.testing.allocator);

    try std.testing.expectEqualStrings("2001:db8:85a3::8a2e:370:7334", enr_fields_1.ip6.?);
    try std.testing.expectEqual(@as(u16, 30303), enr_fields_1.tcp.?);
    try std.testing.expectEqual(@as(u16, 8080), enr_fields_1.quic.?);
    try std.testing.expectEqual(@as(u64, 1), enr_fields_1.seq.?);

    // Test custom field parsing
    // Check if the custom field exists
    const whatever_field = enr_fields.custom_fields.get("whatever");
    if (whatever_field) |value| {
        try std.testing.expectEqualStrings("0x01000000", value);
    } else {
        // If the field doesn't exist, that's also a test failure
        try std.testing.expect(false);
    }
    // quadrivium_0 doesn't have custom fields, so just verify the custom_fields map is empty
    try std.testing.expectEqual(@as(usize, 0), enr_fields_1.custom_fields.count());
}

test "ENR construction from fields" {
    var validator_config = try utils_lib.loadFromYAMLFile(std.testing.allocator, "pkgs/cli/test/fixtures/validator-config.yaml");
    defer validator_config.deinit(std.testing.allocator);

    // Get ENR fields for zeam_0
    var enr_fields = try getEnrFieldsFromValidatorConfig(std.testing.allocator, "zeam_0", validator_config);
    defer enr_fields.deinit(std.testing.allocator);

    // Get private key for zeam_0
    const private_key = try getPrivateKeyFromValidatorConfig(std.testing.allocator, "zeam_0", validator_config);
    defer std.testing.allocator.free(private_key);

    // Construct ENR from fields
    const constructed_enr = try constructENRFromFields(std.testing.allocator, private_key, enr_fields, true);

    // Verify the ENR was constructed successfully
    // We can't easily verify the exact ENR content without knowing the exact signature,
    // but we can verify that specific fields are present in the constructed ENR
    try std.testing.expect(constructed_enr.kvs.get("ip") != null);
    try std.testing.expect(constructed_enr.kvs.get("quic") != null);
    try std.testing.expect(constructed_enr.kvs.get("tcp") != null);
    try std.testing.expect(constructed_enr.kvs.get("seq") != null);
    try std.testing.expect(constructed_enr.kvs.get("is_aggregator") != null);
}

test "compare roots from genGensisBlock and genGenesisState and genStateBlockHeader" {
    var arena_allocator = std.heap.ArenaAllocator.init(std.testing.allocator);
    defer arena_allocator.deinit();
    const allocator = arena_allocator.allocator();

    // Load config.yaml from test fixtures
    const config_filepath = "pkgs/cli/test/fixtures/config.yaml";
    var parsed_config = try utils.loadFromYAMLFile(allocator, config_filepath);
    defer parsed_config.deinit(allocator);

    // Parse genesis config from YAML
    const genesis_spec = try configs.genesisConfigFromYAML(allocator, parsed_config, null);
    defer allocator.free(genesis_spec.validator_attestation_pubkeys);
    defer allocator.free(genesis_spec.validator_proposal_pubkeys);

    // Generate genesis state
    var genesis_state: types.BeamState = undefined;
    try genesis_state.genGenesisState(allocator, genesis_spec);
    defer genesis_state.deinit();

    std.debug.print("\nGenesis state: {s}\n", .{try genesis_state.toJsonString(allocator)});

    // Generate genesis block using genGenesisBlock
    var genesis_block: types.BeamBlock = undefined;
    try genesis_state.genGenesisBlock(allocator, &genesis_block);
    defer genesis_block.deinit();

    // Get state root by hashing the state directly
    var state_root_from_genesis: [32]u8 = undefined;
    try zeam_utils.hashTreeRoot(types.BeamState, genesis_state, &state_root_from_genesis, allocator);

    // Generate block header using genStateBlockHeader
    const state_block_header = try genesis_state.genStateBlockHeader(allocator);
    const state_root_from_block_header = state_block_header.state_root;

    // Compare the roots - they should be equal
    try std.testing.expect(std.mem.eql(u8, &genesis_block.state_root, &state_root_from_block_header));
    try std.testing.expect(std.mem.eql(u8, &state_root_from_genesis, &state_root_from_block_header));

    // Verify the state root matches the expected value
    const state_root_from_genesis_hex = try std.fmt.allocPrint(allocator, "0x{x}", .{&state_root_from_genesis});
    defer allocator.free(state_root_from_genesis_hex);
    try std.testing.expectEqualStrings(state_root_from_genesis_hex, "0x9a77892fc5afa43bbfa1462bd9c96d239ee8b0fd863b1c3a08e69b6f7efbb8d2");
}

test "populateNodeNameRegistry" {
    const allocator = std.testing.allocator;

    const validator_config_path = "pkgs/cli/test/fixtures/validator-config.yaml";
    const validators_path = "pkgs/cli/test/fixtures/validators.yaml";

    // Create an empty registry and populate it from test fixtures
    var registry = node_lib.NodeNameRegistry.init(allocator);
    defer registry.deinit();
    try populateNodeNameRegistry(allocator, &registry, validator_config_path, validators_path);

    try std.testing.expectEqual(@as(usize, 9), registry.validator_index_to_name.count());
    try std.testing.expectEqual(@as(usize, 3), registry.peer_id_to_name.count());

    try std.testing.expectEqualStrings("zeam_0", registry.getNodeNameFromValidatorIndex(1).name.?);
    try std.testing.expectEqualStrings("zeam_0", registry.getNodeNameFromValidatorIndex(4).name.?);
    try std.testing.expectEqualStrings("zeam_0", registry.getNodeNameFromValidatorIndex(7).name.?);

    try std.testing.expectEqualStrings("ream_0", registry.getNodeNameFromValidatorIndex(0).name.?);
    try std.testing.expectEqualStrings("ream_0", registry.getNodeNameFromValidatorIndex(3).name.?);
    try std.testing.expectEqualStrings("ream_0", registry.getNodeNameFromValidatorIndex(6).name.?);

    try std.testing.expectEqualStrings("quadrivium_0", registry.getNodeNameFromValidatorIndex(2).name.?);
    try std.testing.expectEqualStrings("quadrivium_0", registry.getNodeNameFromValidatorIndex(5).name.?);
    try std.testing.expectEqualStrings("quadrivium_0", registry.getNodeNameFromValidatorIndex(8).name.?);

    try std.testing.expectEqualStrings("zeam_0", registry.getNodeNameFromPeerId("16Uiu2HAmKgamysJowVqBeftDWr3XBETpmwvjcusbcuai17uWFgLf").name.?);
    try std.testing.expectEqualStrings("ream_0", registry.getNodeNameFromPeerId("16Uiu2HAmSH2XVgZqYHWucap5kuPzLnt2TsNQkoppVxB5eJGvaXwm").name.?);
    try std.testing.expectEqualStrings("quadrivium_0", registry.getNodeNameFromPeerId("16Uiu2HAmQj1RDNAxopeeeCFPRr3zhJYmH6DEPHYKmxLViLahWcFE").name.?);
}

test "checkpoint-sync-url parameter is optional" {
    // Verify that the NodeCommand struct has checkpoint-sync-url as optional
    const node_cmd = NodeCommand{
        .@"custom-genesis" = "test",
        .@"node-id" = "test",
        .@"validator-config" = "test",
        .@"override-genesis-time" = null,
        .@"checkpoint-sync-url" = null, // Should compile and work with null
    };

    try std.testing.expect(node_cmd.@"checkpoint-sync-url" == null);

    const node_cmd_with_url = NodeCommand{
        .@"custom-genesis" = "test",
        .@"node-id" = "test",
        .@"validator-config" = "test",
        .@"override-genesis-time" = null,
        .@"checkpoint-sync-url" = "http://localhost:5052/lean/v0/states/finalized",
    };

    try std.testing.expect(node_cmd_with_url.@"checkpoint-sync-url" != null);
    try std.testing.expectEqualStrings(node_cmd_with_url.@"checkpoint-sync-url".?, "http://localhost:5052/lean/v0/states/finalized");
}

test "NodeOptions checkpoint_sync_url field is optional" {
    // Verify NodeOptions can be created with null checkpoint_sync_url
    const allocator = std.testing.allocator;

    // Create a minimal NodeOptions structure for testing
    var registry = node_lib.NodeNameRegistry.init(allocator);
    defer registry.deinit();

    var logger_config = utils_lib.getLoggerConfig(null, null);

    // Create a minimal genesis spec for testing
    const genesis_spec = types.GenesisSpec{
        .genesis_time = 1000,
        .validator_attestation_pubkeys = try allocator.alloc(types.Bytes52, 0),
        .validator_proposal_pubkeys = try allocator.alloc(types.Bytes52, 0),
    };
    defer allocator.free(genesis_spec.validator_attestation_pubkeys);
    defer allocator.free(genesis_spec.validator_proposal_pubkeys);

    var node_options = NodeOptions{
        .network_id = 0,
        .node_key = "test",
        .node_key_index = 0,
        .validator_config = "test",
        .bootnodes = &[_][]const u8{},
        .validator_assignments = &[_]ValidatorAssignment{},
        .genesis_spec = genesis_spec,
        .metrics_enable = false,
        .is_aggregator = false,
        .api_port = 5052,
        .metrics_port = 5053,
        .local_priv_key = try allocator.dupe(u8, "test"),
        .logger_config = &logger_config,
        .database_path = "test",
        .hash_sig_key_dir = try allocator.dupe(u8, "test"),
        .node_registry = &registry,
        .checkpoint_sync_url = null, // Should work with null
    };
    defer {
        allocator.free(node_options.local_priv_key);
        allocator.free(node_options.hash_sig_key_dir);
    }

    try std.testing.expect(node_options.checkpoint_sync_url == null);

    // Test with a URL
    node_options.checkpoint_sync_url = "http://localhost:5052/lean/v0/states/finalized";
    try std.testing.expect(node_options.checkpoint_sync_url != null);
}

test "attestationCommitteeCountFromYAML reads ATTESTATION_COMMITTEE_COUNT from config.yaml" {
    var config_file = try utils_lib.loadFromYAMLFile(std.testing.allocator, "pkgs/cli/test/fixtures/config.yaml");
    defer config_file.deinit(std.testing.allocator);

    const count = attestationCommitteeCountFromYAML(config_file);
    try std.testing.expect(count != null);
    try std.testing.expectEqual(@as(u64, 4), count.?);
}

test "attestationCommitteeCountFromYAML returns null when field is absent" {
    // validator-config.yaml has no ATTESTATION_COMMITTEE_COUNT field
    var validator_config = try utils_lib.loadFromYAMLFile(std.testing.allocator, "pkgs/cli/test/fixtures/validator-config.yaml");
    defer validator_config.deinit(std.testing.allocator);

    const count = attestationCommitteeCountFromYAML(validator_config);
    try std.testing.expect(count == null);
}

test "attestation_committee_count: zero value is clamped to 1 with a warning" {
    // Simulate opts with count=0 — the validation block should reset it to 1.
    var opts: NodeOptions = undefined;
    opts.attestation_committee_count = 0;

    // Mirror the validation logic from buildStartOptions.
    if (opts.attestation_committee_count) |count| {
        if (count == 0) {
            opts.attestation_committee_count = 1;
        }
    }

    try std.testing.expectEqual(@as(?u64, 1), opts.attestation_committee_count);
}