feat(v2:metrics): implement histograms on services

conformance/src/shred_parse.zig

@@ -197,7 +197,7 @@ const HarnessState = struct {
         ls.base_slot = 0;
         @memset(&ls.leaders, .ZEROES);
 
-        const receiver = try Receiver.init(allocator, IN_PROGRESS_CAPACITY, DONE_CAPACITY);
+        const receiver = try Receiver.init(allocator, IN_PROGRESS_CAPACITY, DONE_CAPACITY, .initNoop());
 
         self.* = .{
             .allocator = allocator,

v2/components/shred/ReceiverMetrics.zig

@@ -0,0 +1,84 @@
+const std = @import("std");
+const builtin = @import("builtin");
+
+comptime {
+    _ = std.testing.refAllDecls(@This());
+}
+
+const lib = @import("lib");
+const tel = lib.telemetry;
+
+const LatencyHistogram = tel.LatencyHistogram;
+const Receiver = @import("receiver.zig").Receiver;
+
+const ReceiverMetrics = @This();
+
+reset_elapsed_ns: LatencyHistogram,
+update_slot_range_elapsed_ns: LatencyHistogram,
+process_packet_elapsed_ns: ProcessPacketElapsedHistogram,
+
+/// One latency series per `processPacket` outcome: each `NonErrorStatus` variant plus each error,
+/// flattened into `variant="<label>"` labels. `observe` takes the raw `ProcessPacketError!...`.
+const ProcessPacketElapsedHistogram = tel.ResultLatencyHistogram(
+    Receiver.ProcessPacketError!Receiver.NonErrorStatus,
+    .{ .payload_prefix = .initComptime(.{
+        .{ "fec_set_finished", "full_" },
+        .{ "fec_set_already_finished", "early_" },
+        .{ "shred_already_seen", "early_" },
+        .{ "unfinished_fec_set", "early_" },
+    }) },
+);
+
+/// Backs every series on the heap instead of in a metric region, so a test can build a `Receiver`
+/// without standing up an `Appender`. The layout is irrelevant to what these tests assert; it only
+/// has to be valid.
+pub fn initForTest(gpa: std.mem.Allocator) std.mem.Allocator.Error!ReceiverMetrics {
+    if (!builtin.is_test) @compileError("initForTest is only valid in test builds");
+    const layout: LatencyHistogram.Layout = .{
+        .min_upper_bound_ns = 512,
+        .max_upper_bound_ns = 512 << 20,
+        .bounds_per_doubling = 4,
+    };
+    const reset: LatencyHistogram = try .initForTest(gpa, layout);
+    errdefer reset.deinitForTest(gpa);
+
+    const update_slot_range: LatencyHistogram = try .initForTest(gpa, layout);
+    errdefer update_slot_range.deinitForTest(gpa);
+
+    return .{
+        .reset_elapsed_ns = reset,
+        .update_slot_range_elapsed_ns = update_slot_range,
+        .process_packet_elapsed_ns = try .initForTest(gpa, layout),
+    };
+}
+
+/// Only valid if `self` was initialized using `initForTest`.
+pub fn deinitForTest(self: ReceiverMetrics, gpa: std.mem.Allocator) void {
+    if (!builtin.is_test) @compileError("deinitForTest is only valid in test builds");
+    self.reset_elapsed_ns.deinitForTest(gpa);
+    self.update_slot_range_elapsed_ns.deinitForTest(gpa);
+    self.process_packet_elapsed_ns.deinitForTest(gpa);
+}
+
+fn initNoopLatencyHistogram() LatencyHistogram {
+    const layout: LatencyHistogram.Layout = .{
+        .min_upper_bound_ns = 1,
+        .max_upper_bound_ns = 2,
+        .bounds_per_doubling = 1,
+    };
+    const storage = struct {
+        var elements: [1 + 2 * (2 + 3)]u64 = @splat(0);
+    };
+    comptime std.debug.assert(storage.elements.len == layout.elementsFromBucketCount());
+    return .fromRaw(layout, .{ .elements = &storage.elements });
+}
+
+pub fn initNoop() ReceiverMetrics {
+    return .{
+        .reset_elapsed_ns = initNoopLatencyHistogram(),
+        .update_slot_range_elapsed_ns = initNoopLatencyHistogram(),
+        .process_packet_elapsed_ns = .{
+            .inner = .{ .histograms = @splat(initNoopLatencyHistogram()) },
+        },
+    };
+}

v2/components/shred/api.zig

@@ -227,9 +227,28 @@ pub const Shred = extern struct {
     const merkle_node_size = 20;
     const merkle_root_size = 32;
 
+    pub const PacketCheckedError = error{
+        PacketUnderMinHeaderSize,
+        UnsupportedVariant,
+        PacketUnderHeaderSize,
+        DataSmallerThanHeader,
+        DataPacketUnderMinSize,
+        DataEffectiveSizeTooSmall,
+        CodeShredOverMaxSize,
+        PacketSizeUnderExpected3,
+        DataShredMarkedCompleteIsNotLastInSet,
+        BadOffset,
+        BadSlotOrParentOffset,
+        BadSlotIdx,
+        BadCodeShredIdx,
+        NoCodeOrDataCount,
+        CodeOrDataCountTooLarge,
+        SlotIndexTooHigh,
+    };
+
     // [firedancer] https://github.com/firedancer-io/firedancer/commit/7cbb71919ec9b8045c247957280e5b15d1e0cb85
     /// Makes sure that the *layout* of the Shred is valid.
-    pub fn fromPacketChecked(packet: *const Packet) !*const Shred {
+    pub fn fromPacketChecked(packet: *const Packet) PacketCheckedError!*const Shred {
         if (packet.len < min_header_size) return error.PacketUnderMinHeaderSize;
         if (!Shred.hasSupportedVariant(&packet.data)) return error.UnsupportedVariant;
 
@@ -442,8 +461,10 @@ pub const Shred = extern struct {
         @panic("unimplemented");
     }
 
+    pub const MerkleRootError = ComputeMerkleRootError;
+
     // Reconstructs the merkle root from a shred
-    pub fn merkleRoot(shred: *const Shred, out: *Hash) !void {
+    pub fn merkleRoot(shred: *const Shred, out: *Hash) MerkleRootError!void {
         const zone = tracy.Zone.init(@src(), .{ .name = "merkleRoot" });
         defer zone.deinit();
 
@@ -478,12 +499,16 @@ pub const Shred = extern struct {
         out.* = t;
     }
 
+    pub const ComputeMerkleRootError = error{
+        InvalidMerkleProof,
+    };
+
     fn computeMerkleRoot(
         shred_idx: u32,
         leaf_node: *const Hash,
         proof_nodes: []const MerkleProofNode,
         out: *Hash,
-    ) !void {
+    ) ComputeMerkleRootError!void {
         var idx = shred_idx;
         out.* = leaf_node.*;
 

v2/components/shred/component.zig

@@ -4,6 +4,7 @@
 
 comptime {
     if (@import("builtin").is_test) {
+        _ = @import("ReceiverMetrics.zig");
         _ = @import("receiver.zig");
         _ = @import("reed_solomon.zig");
     }
@@ -13,3 +14,4 @@ pub const api = @import("shred_api");
 
 pub const Receiver = @import("receiver.zig").Receiver;
 pub const FecSetCtx = @import("receiver.zig").FecSetCtx;
+pub const ReceiverMetrics = @import("ReceiverMetrics.zig");

v2/components/shred/receiver.zig

@@ -18,6 +18,7 @@ const DeshreddedFecSet = api.DeshreddedFecSet;
 const DeshredRing = api.DeshredRing;
 const FecSetId = api.FecSetId;
 const Shred = api.Shred;
+const ReceiverMetrics = @import("ReceiverMetrics.zig");
 
 /// Takes in shreds, and writes out deshredded fec sets.
 /// For full docs see `services/shred_receiver.zig`.
@@ -31,6 +32,8 @@ pub const Receiver = struct {
     in_progress: InProgressSets,
     done: DoneSets,
 
+    metrics: ReceiverMetrics,
+
     /// Per-feature activation slots. A feature is enforced for shreds
     /// whose slot is `>= activation_slot`; the default `maxInt(Slot)`
     /// keeps every feature inactive.
@@ -42,6 +45,7 @@ pub const Receiver = struct {
         allocator: std.mem.Allocator,
         in_progress_capacity: u32,
         done_capacity: u32,
+        metrics: ReceiverMetrics,
     ) !Receiver {
         var in_progress: InProgressSets = try .init(allocator, in_progress_capacity);
         errdefer in_progress.deinit(allocator);
@@ -56,6 +60,8 @@ pub const Receiver = struct {
             .root_slot = 0,
             .max_slot = std.math.maxInt(Slot),
             .features = .{},
+
+            .metrics = metrics,
         };
     }
 
@@ -69,6 +75,9 @@ pub const Receiver = struct {
     /// (e.g. the conformance shred-parse harness, which runs one fixture per
     /// invocation and must not leak state between them).
     pub fn reset(self: *Receiver) void {
+        const method_obs = self.metrics.reset_elapsed_ns.observer();
+        defer method_obs.observe();
+
         self.in_progress.reset();
         self.done.reset();
         self.root_slot = 0;
@@ -77,25 +86,80 @@ pub const Receiver = struct {
     }
 
     pub fn updateSlotRange(self: *Receiver, root_slot: Slot, max_slot: Slot) void {
+        const method_obs = self.metrics.update_slot_range_elapsed_ns.observer();
+        defer method_obs.observe();
+
+        self.updateSlotRangeImpl(root_slot, max_slot);
+    }
+
+    fn updateSlotRangeImpl(self: *Receiver, root_slot: Slot, max_slot: Slot) void {
         self.root_slot = root_slot;
         self.max_slot = max_slot;
 
         // TODO: this is where we would add code to prune entries outside of the new range.
     }
 
-    // TODO: report return values to observability
-    // TODO: report back equivocating shreds, so that we can construct and send out duplicate proofs
+    pub const ProcessPacketError = error{
+        ShredOlderThanRoot,
+        ShredTooNew,
+        ShredVersionMismatch,
+        FecSetIndexTooHigh,
+        SlotIndexTooHigh,
+        BadDataShredCount,
+        BadCodeShredCount,
+        BadCodeShredIdx,
+        UnexpectedDataCompleteShred,
+        InvalidFecSetIdx,
+        ShredIdxTooLarge,
+        MerkleCountTooLarge,
+        VariantMismatchFromFecSet,
+        MismatchedMerkleRoot,
+        EquivocationDifferentHashForSameFecSetId,
+        EquivocationMatchingFecSetWithDifferentSignatureAlreadyInProgress,
+        UnknownLeader,
+    } || (Shred.PacketCheckedError ||
+        std.fmt.BufPrintError ||
+        Shred.ComputeMerkleRootError ||
+        lib.crypto.ed25519.VerifySignatureError ||
+        InProgressSets.CreateFecSetCtxError);
+
     pub fn processPacket(
         state: *Receiver,
         leader_schedule: *const lib.solana.LeaderSchedule,
         network_shred_version: u16,
         packet: *const Packet,
         deshred_writer: *DeshredRing.Iterator(.writer),
         logger: lib.telemetry.Logger("processPacket"),
-    ) !NonErrorStatus {
+    ) ProcessPacketError!NonErrorStatus {
         const zone = tracy.Zone.init(@src(), .{ .name = "processPacket" });
         defer zone.deinit();
 
+        const method_obs = state.metrics.process_packet_elapsed_ns.observer();
+
+        const result = state.processPacketImpl(
+            leader_schedule,
+            network_shred_version,
+            packet,
+            deshred_writer,
+            logger,
+            zone,
+        );
+
+        method_obs.observe(result);
+        return result;
+    }
+
+    // TODO: report return values to observability
+    // TODO: report back equivocating shreds, so that we can construct and send out duplicate proofs
+    fn processPacketImpl(
+        state: *Receiver,
+        leader_schedule: *const lib.solana.LeaderSchedule,
+        network_shred_version: u16,
+        packet: *const Packet,
+        deshred_writer: *DeshredRing.Iterator(.writer),
+        logger: lib.telemetry.Logger("processPacket"),
+        zone: tracy.Zone,
+    ) ProcessPacketError!NonErrorStatus {
         // check that the shred variant is supported and the header is valid
         const shred = try Shred.fromPacketChecked(packet);
 
@@ -579,12 +643,14 @@ const InProgressSets = struct {
         return self.signature_map.getAdapted(signature, map_ctx);
     }
 
+    pub const CreateFecSetCtxError = error{};
+
     // returns undefined memory, which must be immediately set by the caller
     fn createFecSetCtx(
         self: *InProgressSets,
         id: FecSetId,
         signature: *const Signature,
-    ) !*FecSetCtx {
+    ) CreateFecSetCtxError!*FecSetCtx {
         const map_ctx = self.mapContext();
 
         self.assertCounts();
@@ -756,7 +822,10 @@ fn signTestDataPacket(packet: *Packet, keypair: *const lib.crypto.KeyPair) !void
 test "shred.receiver: empty packet" {
     const allocator = std.testing.allocator;
 
-    var receiver: Receiver = try .init(allocator, 1, 1);
+    const metrics: ReceiverMetrics = try .initForTest(allocator);
+    defer metrics.deinitForTest(allocator);
+
+    var receiver: Receiver = try .init(allocator, 1, 1, metrics);
     defer receiver.deinit(allocator);
 
     var packet: Packet = undefined;
@@ -780,7 +849,10 @@ test "shred.receiver: empty packet" {
 test "shred.receiver: shred version mismatch" {
     const allocator = std.testing.allocator;
 
-    var receiver: Receiver = try .init(allocator, 1, 1);
+    const metrics: ReceiverMetrics = try .initForTest(allocator);
+    defer metrics.deinitForTest(allocator);
+
+    var receiver: Receiver = try .init(allocator, 1, 1, metrics);
     defer receiver.deinit(allocator);
 
     var packet: Packet = undefined;
@@ -804,7 +876,10 @@ test "shred.receiver: shred version mismatch" {
 test "shred.receiver: one shred (unfinished fec set)" {
     const allocator = std.testing.allocator;
 
-    var receiver: Receiver = try .init(allocator, 1, 1);
+    const metrics: ReceiverMetrics = try .initForTest(allocator);
+    defer metrics.deinitForTest(allocator);
+
+    var receiver: Receiver = try .init(allocator, 1, 1, metrics);
     defer receiver.deinit(allocator);
 
     const std_keypair = try std.crypto.sign.Ed25519.KeyPair.generateDeterministic(@splat(1));
@@ -839,7 +914,10 @@ test "shred.receiver: one shred (unfinished fec set)" {
 test "shred.receiver: duplicate shred" {
     const allocator = std.testing.allocator;
 
-    var receiver: Receiver = try .init(allocator, 1, 1);
+    const metrics: ReceiverMetrics = try .initForTest(allocator);
+    defer metrics.deinitForTest(allocator);
+
+    var receiver: Receiver = try .init(allocator, 1, 1, metrics);
     defer receiver.deinit(allocator);
 
     const std_keypair = try std.crypto.sign.Ed25519.KeyPair.generateDeterministic(@splat(1));

v2/lib/crypto/ed25519.zig

@@ -126,6 +126,12 @@ pub fn verifyBatchOverSingleMessage(
     }
 }
 
+pub const VerifySignatureError = error{
+    InvalidSignature,
+} || (std.crypto.errors.NonCanonicalError ||
+    std.crypto.errors.EncodingError ||
+    AffineLowOrderError);
+
 /// See the doc-comment above `verifyBatchOverSingleMessage` for further detail,
 /// but this is that same thing, just for single messages, and with the ability to toggle
 /// between `verify` and `verify_strict` semantics (used in ed25519 precompile).
@@ -134,7 +140,7 @@ pub fn verifySignature(
     pubkey: *const Pubkey,
     message: []const u8,
     strict: bool,
-) !void {
+) VerifySignatureError!void {
     const s = signature.s;
     const r = signature.r;
     try Edwards25519.scalar.rejectNonCanonical(s);
@@ -166,6 +172,8 @@ pub fn affineEqual(a: Edwards25519, b: Edwards25519) bool {
     return x1.equivalent(a.x) and y1.equivalent(a.y);
 }
 
+pub const AffineLowOrderError = error{WeakPublicKey};
+
 /// Determines whether `a` is of small order (in the torision subgroup E[8]), but with the
 /// assumption that `a.Z == 1`.
 ///
@@ -189,7 +197,7 @@ pub fn affineEqual(a: Edwards25519, b: Edwards25519) bool {
 /// just checking a single coordinate of the point is enough to determine if it's in the blacklist,
 /// meaning we only need 4 equivalence checks to cover all of the pairs.
 ///
-pub fn affineLowOrder(a: Edwards25519) !void {
+pub fn affineLowOrder(a: Edwards25519) AffineLowOrderError!void {
     // y coordinate of points 5 and 6
     const y0: Edwards25519.Fe = .{ .limbs = .{
         0x4d3d706a17c7,