feat: validator monitor

Implements the missing validator monitor metrics from lodestar-ts.

Note that there are some metrics intentionally omitted because they
probably require even bigger changes on the zig side, this is just
the minimal change required to track validator perf. 

The following metrics are excluded:

- `validator_monitor_prev_epoch_on_chain_attester_hit_total`
- `validator_monitor_prev_epoch_on_chain_attester_miss_total`
- `validator_monitor_prev_epoch_on_chain_attester_correct_head_total`
- `validator_monitor_prev_epoch_on_chain_attester_incorrect_head_total`
- `validator_monitor_prev_epoch_on_chain_inclusion_distance`

These will be updated on the TS side to run on `onceEveryEndOfEpoch`
rather than inside `registerValidatorStatuses`, which is now zig native

Author: spiral-ladder

bench/state_transition/process_block.zig

@@ -529,6 +529,7 @@ fn runBenchmark(
         cached_state,
         block_slot,
         .{},
+        null,
     );
     try cached_state.state.commit();
     try state_transition.buildSlashingsCacheFromStateIfNeeded(allocator, cached_state.state, &cached_state.slashings_cache);

bindings/napi/BeaconStateView.zig

@@ -17,6 +17,7 @@ const js_types = @import("./js_types.zig");
 const sszValueToNapiValue = @import("./to_napi_value.zig").sszValueToNapiValue;
 const numberSliceToNapiValue = @import("./to_napi_value.zig").numberSliceToNapiValue;
 const napi_io = @import("./io.zig");
+const validator_monitor = @import("./validator_monitor.zig");
 
 /// Allocator used for all BeaconStateView instances.
 var gpa: std.heap.DebugAllocator(.{}) = .init;
@@ -1289,7 +1290,14 @@ pub fn processSlots(self: *const BeaconStateView, slot_arg: js.Number, options:
         allocator.destroy(post_state);
     }
 
-    try st.processSlots(allocator, napi_io.get(), post_state, slot_value, .{});
+    try st.processSlots(
+        allocator,
+        napi_io.get(),
+        post_state,
+        slot_value,
+        .{},
+        validator_monitor.get(),
+    );
     return .{
         .cached_state = post_state,
         .pool_rc = pool.state.poolRc().ref(),
@@ -1318,7 +1326,14 @@ pub fn stateTransition(self: *const BeaconStateView, signed_block_bytes: js.Uint
     const signed_block = try AnySignedBeaconBlock.deserialize(allocator, .full, fork_seq, bytes);
     defer signed_block.deinit(allocator);
 
-    const post_state = try st.stateTransition(allocator, napi_io.get(), cached_state, signed_block, opts);
+    const post_state = try st.stateTransition(
+        allocator,
+        napi_io.get(),
+        cached_state,
+        signed_block,
+        opts,
+        validator_monitor.get(),
+    );
     return .{
         .cached_state = post_state,
         .pool_rc = pool.state.poolRc().ref(),

bindings/napi/metrics.zig

@@ -12,13 +12,34 @@ else
 
 var initialized: bool = false;
 
+const validator_monitor = @import("./validator_monitor.zig");
+
 /// JS: metrics.init() → void
 pub fn init() !void {
     if (initialized) return;
     try state_transition.metrics.init(allocator, napi_io.get(), .{});
     initialized = true;
 }
 
+/// JS: metrics.registerLocalValidator(index) → void
+///
+/// Adds a validator index to the process-wide validator monitor so that
+/// metrics are recorded for it on every epoch transition.
+pub fn registerLocalValidator(index: js.Number) !void {
+    const value = try index.toI64();
+    if (value < 0) return error.InvalidValidatorIndex;
+    try validator_monitor.get().registerLocalValidator(napi_io.get(), @intCast(value));
+}
+
+/// JS: metrics.unregisterLocalValidator(index) → void
+///
+/// Prunes a validator index from the process-wide validator monitor.
+pub fn unregisterLocalValidator(index: js.Number) !void {
+    const value = try index.toI64();
+    if (value < 0) return error.InvalidValidatorIndex;
+    validator_monitor.get().unregisterLocalValidator(napi_io.get(), @intCast(value));
+}
+
 /// JS: metrics.scrapeMetrics() → string
 pub fn scrapeMetrics() !js.String {
     var aw: std.Io.Writer.Allocating = .init(allocator);
@@ -29,7 +50,8 @@ pub fn scrapeMetrics() !js.String {
 }
 
 pub fn deinit() void {
+    validator_monitor.deinit();
     if (!initialized) return;
-    state_transition.metrics.state_transition.deinit();
+    state_transition.metrics.deinit();
     initialized = false;
 }

bindings/napi/validator_monitor.zig

@@ -0,0 +1,31 @@
+//! Process-wide validator monitor shared by the NAPI bindings.
+//!
+//! This file is intentionally NOT exported as a JS module in `root.zig`:
+//! it only holds native state. JS interacts with it through
+//! `metrics.registerLocalValidator()` and the metrics scraped via
+//! `metrics.scrapeMetrics()`.
+
+const std = @import("std");
+const builtin = @import("builtin");
+const state_transition = @import("state_transition");
+
+var gpa: std.heap.DebugAllocator(.{}) = .init;
+const allocator = if (builtin.mode == .Debug)
+    gpa.allocator()
+else
+    std.heap.c_allocator;
+
+/// Fed by `BeaconStateView.processSlots`/`stateTransition` on every epoch
+/// transition. Only records metrics for validators registered via
+/// `metrics.registerLocalValidator()`.
+var monitor = state_transition.ValidatorMonitor.init(allocator);
+
+/// Returns the process-wide validator monitor.
+pub fn get() *state_transition.ValidatorMonitor {
+    return &monitor;
+}
+
+/// Frees all monitor state. Meant to be called once on module cleanup.
+pub fn deinit() void {
+    monitor.deinit();
+}

bindings/src/index.d.ts

@@ -384,6 +384,8 @@ declare const bindings: {
   metrics: {
     init: () => void;
     scrapeMetrics: () => string;
+    registerLocalValidator: (index: number) => void;
+    unregisterLocalValidator: (index: number) => void;
   };
   BeaconStateView: typeof BeaconStateView;
 };

bindings/src/metrics.d.ts

@@ -3,3 +3,16 @@ export declare function init(): void;
 
 /** Scrape native state-transition metrics in Prometheus text format. */
 export declare function scrapeMetrics(): string;
+
+/**
+ * Register a validator index with the native validator monitor. Metrics
+ * are recorded for registered validators on every epoch transition.
+ */
+export declare function registerLocalValidator(index: number): void;
+
+/**
+ * Remove a validator index from the native validator monitor, so its
+ * `validator_monitor_*` status metrics stop being recorded. Mirrors the pruning
+ * of stale registrations in lodestar's validator monitor.
+ */
+export declare function unregisterLocalValidator(index: number): void;

bindings/src/metrics.js

@@ -4,3 +4,5 @@ const native = bindings.metrics;
 
 export const init = native.init;
 export const scrapeMetrics = native.scrapeMetrics;
+export const registerLocalValidator = native.registerLocalValidator;
+export const unregisterLocalValidator = native.unregisterLocalValidator;

src/state_transition/ValidatorMonitor.zig

@@ -0,0 +1,194 @@
+//! Consumes the per-validator status data produced by the epoch transition
+//! and records validator metrics for the validators registered with `registerLocalValidator`.
+
+const std = @import("std");
+
+const types = @import("consensus_types");
+const metrics = @import("metrics.zig");
+const attester_status = @import("utils/attester_status.zig");
+const hasMarkers = attester_status.hasMarkers;
+
+const Allocator = std.mem.Allocator;
+const Epoch = types.primitive.Epoch.Type;
+const ValidatorIndex = types.primitive.ValidatorIndex.Type;
+
+pub const ValidatorMonitor = @This();
+
+allocator: Allocator,
+/// Unordered list of validators that require additional monitoring.
+validators: std.AutoArrayHashMapUnmanaged(ValidatorIndex, void),
+/// Prevents registering statuses for the same epoch twice.
+/// processEpoch() may be run more than once for the same epoch.
+last_registered_status_epoch: ?Epoch,
+
+pub fn init(allocator: Allocator) ValidatorMonitor {
+    return .{
+        .allocator = allocator,
+        .validators = .empty,
+        .last_registered_status_epoch = null,
+    };
+}
+
+pub fn deinit(self: *ValidatorMonitor) void {
+    self.validators.deinit(self.allocator);
+    self.* = undefined;
+}
+
+/// Adds a validator to the list of monitored validators.
+///
+/// Registering an already-monitored validator is a no-op.
+pub fn registerLocalValidator(self: *ValidatorMonitor, io: std.Io, index: ValidatorIndex) !void {
+    try self.validators.put(self.allocator, index, {});
+}
+
+/// Prunes a validator from the list of monitored validators.
+///
+/// Unregistering an unknown validator is a no-op.
+pub fn unregisterLocalValidator(self: *ValidatorMonitor, io: std.Io, index: ValidatorIndex) void {
+    _ = self.validators.swapRemove(index);
+}
+
+/// Registers the per-validator statuses produced by one epoch transition and
+/// records metrics for all monitored validators.
+///
+/// `flags` are the packed attester flags of `EpochTransitionCache` (see `utils/attester_status.zig`).
+/// `balances` is optional; when present the total balance of all monitored
+/// validators is reported.
+///
+/// NOTE: Gossip-derived metrics (inclusion distance, attester hit/miss, correct head)
+/// are recorded by the lodestar-ts validator monitor, which observes attestations
+/// on the network; the state transition cannot see them post-altair.
+/// TODO(bing): port the rest of validator monitor, this is just a minimal
+/// impl for stf integration
+pub fn registerValidatorStatuses(
+    self: *ValidatorMonitor,
+    current_epoch: Epoch,
+    flags: []const u8,
+    balances: ?[]const u64,
+) void {
+
+    // Prevent registering statuses for the same epoch twice.
+    if (self.last_registered_status_epoch) |last_registered_status_epoch|
+        if (current_epoch <= last_registered_status_epoch) return;
+
+    self.last_registered_status_epoch = current_epoch;
+
+    // There won't be any validator activity in epoch -1.
+    if (current_epoch == 0) return;
+
+    const vm = &metrics.validator_monitor;
+
+    // Track total balance instead of per-validator balance to reduce metric cardinality.
+    var total_balance: u64 = 0;
+
+    for (self.validators.keys()) |index| {
+        // The monitored validator may not be in the state yet.
+        if (index >= flags.len) continue;
+
+        const flag = flags[index];
+
+        if (hasMarkers(flag, attester_status.FLAG_PREV_SOURCE_ATTESTER)) {
+            vm.prev_epoch_on_chain_source_attester_hit.incr();
+        } else {
+            vm.prev_epoch_on_chain_source_attester_miss.incr();
+        }
+        if (hasMarkers(flag, attester_status.FLAG_PREV_HEAD_ATTESTER)) {
+            vm.prev_epoch_on_chain_head_attester_hit.incr();
+        } else {
+            vm.prev_epoch_on_chain_head_attester_miss.incr();
+        }
+        if (hasMarkers(flag, attester_status.FLAG_PREV_TARGET_ATTESTER)) {
+            vm.prev_epoch_on_chain_target_attester_hit.incr();
+        } else {
+            vm.prev_epoch_on_chain_target_attester_miss.incr();
+        }
+
+        if (balances) |b| {
+            if (index < b.len) total_balance += b[index];
+        }
+    }
+
+    if (balances != null) {
+        vm.prev_epoch_on_chain_balance.set(total_balance);
+    }
+}
+
+test "registerValidatorStatuses records metrics" {
+    const allocator = std.testing.allocator;
+    try metrics.init(allocator, std.testing.io, .{});
+    defer metrics.deinit();
+
+    var monitor = ValidatorMonitor.init(allocator);
+    defer monitor.deinit();
+    try monitor.registerLocalValidator(std.testing.io, 0);
+    try monitor.registerLocalValidator(std.testing.io, 1);
+
+    const flags = [_]u8{
+        attester_status.FLAG_PREV_SOURCE_ATTESTER |
+            attester_status.FLAG_PREV_TARGET_ATTESTER |
+            attester_status.FLAG_PREV_HEAD_ATTESTER,
+        0,
+    };
+    const balances = [_]u64{ 32_000_000_000, 31_000_000_000 };
+    monitor.registerValidatorStatuses(1, &flags, &balances);
+
+    var aw: std.Io.Writer.Allocating = .init(allocator);
+    var list = _: {
+        errdefer aw.deinit();
+        try metrics.write(&aw.writer);
+        break :_ aw.toArrayList();
+    };
+    defer list.deinit(allocator);
+
+    const expectations = [_][]const u8{
+        "validator_monitor_prev_epoch_on_chain_source_attester_hit_total 1",
+        "validator_monitor_prev_epoch_on_chain_source_attester_miss_total 1",
+        "validator_monitor_prev_epoch_on_chain_target_attester_hit_total 1",
+        "validator_monitor_prev_epoch_on_chain_target_attester_miss_total 1",
+        "validator_monitor_prev_epoch_on_chain_head_attester_hit_total 1",
+        "validator_monitor_prev_epoch_on_chain_head_attester_miss_total 1",
+        "validator_monitor_prev_epoch_on_chain_balance 63000000000",
+    };
+    for (expectations) |expected| {
+        std.testing.expect(std.mem.indexOf(u8, list.items, expected) != null) catch |err| {
+            std.debug.print("expected metric not found: {s}\n", .{expected});
+            return err;
+        };
+    }
+}
+
+test "registerValidatorStatuses guards" {
+    var monitor = ValidatorMonitor.init(std.testing.allocator);
+    defer monitor.deinit();
+
+    try monitor.registerLocalValidator(0);
+    try monitor.registerLocalValidator(2);
+    // registering twice is a no-op
+    try monitor.registerLocalValidator(0);
+    try std.testing.expectEqual(@as(usize, 2), monitor.validators.count());
+
+    // unregistering removes; unknown index is a no-op
+    monitor.unregisterLocalValidator(2);
+    monitor.unregisterLocalValidator(99);
+    try std.testing.expectEqual(@as(usize, 1), monitor.validators.count());
+    try monitor.registerLocalValidator(2);
+    try std.testing.expectEqual(@as(usize, 2), monitor.validators.count());
+
+    const flags = [_]u8{
+        attester_status.FLAG_PREV_SOURCE_ATTESTER | attester_status.FLAG_PREV_TARGET_ATTESTER,
+        0,
+        attester_status.FLAG_PREV_HEAD_ATTESTER,
+    };
+    const balances = [_]u64{ 32_000_000_000, 31_000_000_000, 33_000_000_000 };
+
+    // epoch 0 is registered but has no previous epoch activity
+    monitor.registerValidatorStatuses(std.testing.io, 0, &flags, &balances);
+    try std.testing.expectEqual(@as(?Epoch, 0), monitor.last_registered_status_epoch);
+
+    monitor.registerValidatorStatuses(std.testing.io, 1, &flags, &balances);
+    try std.testing.expectEqual(@as(?Epoch, 1), monitor.last_registered_status_epoch);
+
+    // same epoch twice is a no-op
+    monitor.registerValidatorStatuses(std.testing.io, 1, &flags, &balances);
+    try std.testing.expectEqual(@as(?Epoch, 1), monitor.last_registered_status_epoch);
+}