From 2ae9a96d176b599ba7f71760332b55cc45904cb0 Mon Sep 17 00:00:00 2001 From: hamza-syndica Date: Thu, 30 Jul 2026 18:29:08 +0600 Subject: [PATCH 1/4] v2: log when bootstrap is blocked (#1746) \nAdd periodic observability logging to gossip, snapshot downloader, accounts_db, and replay services while they are blocked waiting on upstream bootstrap data.\n\nEach service:\n- Logs at info level every 10s while waiting\n- Escalates to warn after a service-specific timeout\n (gossip: 60s, snapshot: 60s, accounts_db: 5min, replay: 15min)\n- Emits a one-shot info log when data arrives\n\nIntroduces a shared ThrottledLogger utility in v2/lib/telemetry.zig\nthat gates log frequency and tracks escalation state.\n\nCloses #1746 --- v2/components/snapshot/download.zig | 80 ++++++++++++++++++++++++++++ v2/lib/telemetry.zig | 69 ++++++++++++++++++++++++ v2/services/accounts_db.zig | 73 ++++++++++++++++++++++++- v2/services/gossip.zig | 43 +++++++++++++++ v2/services/replay.zig | 82 ++++++++++++++++++++++++++++- 5 files changed, 344 insertions(+), 3 deletions(-) diff --git a/v2/components/snapshot/download.zig b/v2/components/snapshot/download.zig index 82d6660f18..155b02d4fc 100644 --- a/v2/components/snapshot/download.zig +++ b/v2/components/snapshot/download.zig @@ -929,6 +929,19 @@ pub const Downloader = struct { metrics: Metrics, logger: tel.Logger("snapshot"), + // Bootstrap-blocked observability (issue #1746). + // Emits a periodic info log while the downloader has no candidates, and + // escalates to warn once after `awaiting_warn_after_ns` has elapsed. + // If we ever logged an "awaiting" message, we also emit a one-shot info + // log when a usable peer is found. + bootstrap_start_ns: u64, + awaiting_gate: tel.ThrottledLogger, + logged_awaiting: bool, + logged_ready: bool, + + const AWAITING_LOG_INTERVAL_NS: u64 = 10 * std.time.ns_per_s; + const AWAITING_WARN_AFTER_NS: u64 = 60 * std.time.ns_per_s; + pub fn init( gossip_to_snapshot: *SnapshotSourceRing, known_validators: KnownValidators, @@ -936,6 +949,7 @@ pub const Downloader = struct { metrics: Metrics, logger: tel.Logger("snapshot"), ) !Downloader { + const now_ns = lib.clock.monotonic(.ns); return .{ .ring = try IoUring.init(IO_URING_ENTRIES, 0), .gossip_iter = gossip_to_snapshot.get(.reader), @@ -950,6 +964,10 @@ pub const Downloader = struct { .run_result = null, .metrics = metrics, .logger = logger, + .bootstrap_start_ns = now_ns, + .awaiting_gate = .init(AWAITING_LOG_INTERVAL_NS, now_ns), + .logged_awaiting = false, + .logged_ready = false, }; } @@ -974,6 +992,7 @@ pub const Downloader = struct { while (true) { try self.drainGossip(); + self.maybeLogAwaitingPeers(); _ = try self.ring.submit_and_wait(0); const n = try self.ring.copy_cqes(&cqes, 0); @@ -2665,6 +2684,67 @@ pub const Downloader = struct { }; } + /// Bootstrap observability: log at most once per 10s while we have no + /// download candidates from gossip. Distinguishes between "no peers arrived + /// at all" and "peers arrived but none are usable". Once a candidate has + /// begun racing (or a winner is picked), emits a one-shot "ready" info log + /// if we previously logged that we were waiting. + /// + /// See issue #1746. + fn maybeLogAwaitingPeers(self: *Downloader) void { + // We only care about the pre-race window. Once `.racing` starts, we + // may still transition through failures, but the "awaiting peers" + // phase is over — the download-side already logs failures. + const still_awaiting = self.download_race.phase == .idle; + + if (!still_awaiting) { + if (self.logged_awaiting and !self.logged_ready) { + self.logged_ready = true; + self.logger.info().log( + "snapshot: found usable peer, starting download", + ); + } + return; + } + + const now_ns = lib.clock.monotonic(.ns); + if (!self.awaiting_gate.tick(now_ns)) return; + + self.logged_awaiting = true; + const elapsed_ns = now_ns -| self.bootstrap_start_ns; + const elapsed_s = elapsed_ns / std.time.ns_per_s; + const peers_seen = self.dedupe_map.len; + const escalate = !self.awaiting_gate.escalated and + elapsed_ns >= AWAITING_WARN_AFTER_NS; + if (escalate) self.awaiting_gate.escalated = true; + + if (peers_seen == 0) { + if (escalate) { + self.logger.warn().logf( + "snapshot: awaiting peers from gossip ({d}s, none received)", + .{elapsed_s}, + ); + } else { + self.logger.info().logf( + "snapshot: awaiting peers from gossip ({d}s, none received)", + .{elapsed_s}, + ); + } + } else { + if (escalate) { + self.logger.warn().logf( + "snapshot: no usable peers so far ({d}s, received={d}, active_probes={d})", + .{ elapsed_s, peers_seen, self.active_probes }, + ); + } else { + self.logger.info().logf( + "snapshot: no usable peers so far ({d}s, received={d}, active_probes={d})", + .{ elapsed_s, peers_seen, self.active_probes }, + ); + } + } + } + /// Retires all active download connections except the one at `keep_index`. /// Used after the race reaches a terminal state (completed or failed) to /// clean up losers. Late CQEs from retired slots are ignored via gen mismatch. diff --git a/v2/lib/telemetry.zig b/v2/lib/telemetry.zig index db8706ed5c..16bf862278 100644 --- a/v2/lib/telemetry.zig +++ b/v2/lib/telemetry.zig @@ -453,6 +453,75 @@ pub const Gauge = struct { } }; +/// Emits at most one log per `interval_ns`, with an optional one-shot +/// escalation flag that callers can use to promote the level from info to warn +/// exactly once after some deadline has elapsed. +/// +/// This is intended for use in bootstrap wait loops where a service is blocked +/// on an upstream signal (e.g. gossip peers, snapshot bytes, runtime metadata) +/// and needs to periodically inform the operator without spamming the log. +/// +/// Callers pass `now_ns` from a monotonic clock so this stays testable and +/// avoids taking a dependency on the wall clock here. The first tick fires +/// only after `interval_ns` has elapsed since `start_ns`. +pub const ThrottledLogger = struct { + interval_ns: u64, + last_ns: u64, + /// Set to true once the caller has emitted the escalated (e.g. warn) log. + /// Use to keep escalation to a single emission per stage. + escalated: bool = false, + + pub fn init(interval_ns: u64, start_ns: u64) ThrottledLogger { + return .{ .interval_ns = interval_ns, .last_ns = start_ns }; + } + + /// Returns true if at least `interval_ns` has elapsed since the last tick + /// that returned true (or since construction). When it returns true, the + /// internal timer is advanced to `now_ns`. + pub fn tick(self: *ThrottledLogger, now_ns: u64) bool { + // Saturating subtract guards against a hypothetical clock regression. + if (now_ns -| self.last_ns >= self.interval_ns) { + self.last_ns = now_ns; + return true; + } + return false; + } +}; + +test "ThrottledLogger: does not fire until interval has elapsed" { + var t: ThrottledLogger = .init(1_000, 0); + try std.testing.expect(!t.tick(0)); + try std.testing.expect(!t.tick(1)); + try std.testing.expect(!t.tick(999)); + try std.testing.expect(t.tick(1_000)); +} + +test "ThrottledLogger: subsequent ticks respect the interval" { + var t: ThrottledLogger = .init(1_000, 0); + try std.testing.expect(t.tick(1_000)); + try std.testing.expect(!t.tick(1_500)); + try std.testing.expect(t.tick(2_001)); + try std.testing.expect(!t.tick(2_500)); + try std.testing.expect(t.tick(3_500)); +} + +test "ThrottledLogger: escalated flag is caller-owned and starts false" { + var t: ThrottledLogger = .init(1_000, 0); + try std.testing.expect(!t.escalated); + t.escalated = true; + try std.testing.expect(t.escalated); + // Escalation flag does not affect tick behavior. + try std.testing.expect(t.tick(2_000)); +} + +test "ThrottledLogger: start_ns suppresses early ticks" { + // Simulates production usage: monotonic clock is already large at construction. + var t: ThrottledLogger = .init(1_000, 50_000); + try std.testing.expect(!t.tick(50_000)); // same instant as construction + try std.testing.expect(!t.tick(50_500)); // within interval + try std.testing.expect(t.tick(51_000)); // exactly one interval later +} + /// Can be used as a counter or a gauge. pub fn Variant(comptime V: type) type { return struct { diff --git a/v2/services/accounts_db.zig b/v2/services/accounts_db.zig index 201f24aded..d445ad578c 100644 --- a/v2/services/accounts_db.zig +++ b/v2/services/accounts_db.zig @@ -20,6 +20,51 @@ pub const std_options = start.options; pub const ReadOnly = services.accounts_db.ReadOnly; pub const ReadWrite = services.accounts_db.ReadWrite; +const ServiceLogger = lib.telemetry.Logger("main"); + +/// Bootstrap-blocked observability for the snapshot-bytes wait. Held by pointer +/// on the `SnapshotBufReader` so it survives across pass-by-value copies of the +/// reader. See issue #1746. +const SnapshotWaitState = struct { + logger: ServiceLogger, + start_ns: u64, + gate: lib.telemetry.ThrottledLogger, + logged_awaiting: bool = false, + logged_ready: bool = false, + + const AWAITING_LOG_INTERVAL_NS: u64 = 10 * std.time.ns_per_s; + const AWAITING_WARN_AFTER_NS: u64 = 5 * 60 * std.time.ns_per_s; + + fn maybeLogAwaiting(self: *SnapshotWaitState) void { + const now_ns = lib.clock.monotonic(.ns); + if (!self.gate.tick(now_ns)) return; + self.logged_awaiting = true; + const elapsed_ns = now_ns -| self.start_ns; + const elapsed_s = elapsed_ns / std.time.ns_per_s; + const escalate = + !self.gate.escalated and elapsed_ns >= AWAITING_WARN_AFTER_NS; + if (escalate) { + self.gate.escalated = true; + self.logger.warn().logf( + "accounts_db: awaiting snapshot bytes from snapshot service ({d}s)", + .{elapsed_s}, + ); + } else { + self.logger.info().logf( + "accounts_db: awaiting snapshot bytes from snapshot service ({d}s)", + .{elapsed_s}, + ); + } + } + + fn markReady(self: *SnapshotWaitState) void { + if (self.logged_awaiting and !self.logged_ready) { + self.logged_ready = true; + self.logger.info().log("accounts_db: receiving snapshot bytes"); + } + } +}; + pub fn serviceMain(runner: lib.runner.Connection, _: ReadOnly, rw: ReadWrite) !noreturn { const logger = rw.tel.acquireLogger(@tagName(name), "main"); rw.tel.signalReady(); @@ -30,6 +75,7 @@ pub fn serviceMain(runner: lib.runner.Connection, _: ReadOnly, rw: ReadWrite) !n const Global = struct { var fba_memory: [32 * 1024 * 1024]u8 = undefined; var rooted: Rooted = undefined; + var wait_state: SnapshotWaitState = undefined; }; const rooted = &Global.rooted; @@ -50,20 +96,42 @@ pub fn serviceMain(runner: lib.runner.Connection, _: ReadOnly, rw: ReadWrite) !n if (rooted.table.count() == 0) { logger.info().log("no existing rooted db. reading from snapshot"); + const now_ns = lib.clock.monotonic(.ns); + Global.wait_state = .{ + .logger = logger, + .start_ns = now_ns, + .gate = .init(SnapshotWaitState.AWAITING_LOG_INTERVAL_NS, now_ns), + }; + const SnapshotDataRingReader = @TypeOf(in); const SnapshotBufReader = struct { in_: *SnapshotDataRingReader, runner_: lib.runner.Connection, completion_: *std.atomic.Value(f64), + wait_: *SnapshotWaitState, pub fn percentCompleted(self: @This()) f64 { return self.completion_.load(.monotonic); } pub fn getBuffer(self: @This()) []const u8 { - return self.in_.getBufferBlocking(self.runner_) catch |err| switch (err) { - error.Canceled => return &.{}, // cancel -> EOF + // Fast path: bytes already available, no waiting. + if (self.in_.getBuffer()) |buf| { + self.wait_.markReady(); + return buf; + } + + // Slow path: mirror `getBufferBlocking`'s idle/active signaling, + // and additionally emit throttled "still awaiting" logs so an + // operator can tell the service is blocked upstream (see #1746). + const buf = while (true) { + self.wait_.maybeLogAwaiting(); + self.runner_.activity.signalIdleSpinning() catch return &.{}; + if (self.in_.getBuffer()) |b| break b; }; + self.runner_.activity.signalActive() catch return &.{}; + self.wait_.markReady(); + return buf; } pub fn advance(self: @This(), n: usize) void { @@ -76,6 +144,7 @@ pub fn serviceMain(runner: lib.runner.Connection, _: ReadOnly, rw: ReadWrite) !n .in_ = &in, .runner_ = runner, .completion_ = &rw.ready_snapshot_in.completion, + .wait_ = &Global.wait_state, }); logger.info().log("reading snapshot accounts"); diff --git a/v2/services/gossip.zig b/v2/services/gossip.zig index 9f30e31192..6943f91924 100644 --- a/v2/services/gossip.zig +++ b/v2/services/gossip.zig @@ -112,6 +112,16 @@ pub fn serviceMain(runner: lib.runner.Connection, ro: ReadOnly, rw: ReadWrite) ! }); var it = rw.net_pair.recv.get(.reader); + + // Bootstrap-blocked observability: emit periodic logs while gossip has not + // yet received any inbound packet. Silent on the healthy path (after the + // first packet arrives, these variables are never touched again). + // See issue #1746. + const bootstrap_start_ns = lib.clock.monotonic(.ns); + var awaiting_gate: lib.telemetry.ThrottledLogger = .init(10 * std.time.ns_per_s, bootstrap_start_ns); + const warn_after_ns = 60 * std.time.ns_per_s; + var first_packet_received = false; + while (true) { now = lib.clock.wallclock(.ms); try gossip_node.poll(.from(logger), now); @@ -122,10 +132,43 @@ pub fn serviceMain(runner: lib.runner.Connection, ro: ReadOnly, rw: ReadWrite) ! // For now this should work fine, but in theory there's a very slim chance // of a race condition (it should be basically impossible to manifest // in the one black-box test that currently exists for this). + + if (!first_packet_received) { + const now_ns = lib.clock.monotonic(.ns); + if (awaiting_gate.tick(now_ns)) { + const elapsed_s = (now_ns -| bootstrap_start_ns) / std.time.ns_per_s; + if (!awaiting_gate.escalated and + now_ns -| bootstrap_start_ns >= warn_after_ns) + { + awaiting_gate.escalated = true; + logger.warn().logf( + "gossip has received no packets from cluster ({d}s)", + .{elapsed_s}, + ); + } else { + logger.info().logf( + "gossip has received no packets from cluster ({d}s)", + .{elapsed_s}, + ); + } + } + } + try runner.activity.signalIdleSpinning(); continue; }; try runner.activity.signalActive(); + + if (!first_packet_received) { + first_packet_received = true; + const elapsed_s = + (lib.clock.monotonic(.ns) -| bootstrap_start_ns) / std.time.ns_per_s; + logger.info().logf( + "gossip received first packets from cluster ({d}s)", + .{elapsed_s}, + ); + } + gossip_node.processPacket(.from(logger), now, packet); it.markUsed(); } diff --git a/v2/services/replay.zig b/v2/services/replay.zig index 55c571e9bd..59960d4e35 100644 --- a/v2/services/replay.zig +++ b/v2/services/replay.zig @@ -275,6 +275,74 @@ pub fn serviceMain(runner: lib.runner.Connection, _: ReadOnly, rw: ReadWrite) !n } } +/// Bootstrap-blocked observability for replay's wait on runtime metadata +/// (blockhash queue + root slot) from accounts_db. Kept next to `bootstrap` +/// because the waits happen only during bootstrap. See issue #1746. +const RuntimeMetadataWaitState = struct { + logger: tel.Logger("main"), + start_ns: u64, + gate: tel.ThrottledLogger, + logged_awaiting: bool = false, + logged_ready: bool = false, + + const AWAITING_LOG_INTERVAL_NS: u64 = 10 * std.time.ns_per_s; + // accounts_db loading a mainnet snapshot can take longer than 10 minutes; + // avoid alerting the operator until we're well past the realistic tail. + const AWAITING_WARN_AFTER_NS: u64 = 15 * 60 * std.time.ns_per_s; + + fn maybeLogAwaiting(self: *RuntimeMetadataWaitState) void { + const now_ns = lib.clock.monotonic(.ns); + if (!self.gate.tick(now_ns)) return; + self.logged_awaiting = true; + const elapsed_ns = now_ns -| self.start_ns; + const elapsed_s = elapsed_ns / std.time.ns_per_s; + const escalate = + !self.gate.escalated and elapsed_ns >= AWAITING_WARN_AFTER_NS; + if (escalate) { + self.gate.escalated = true; + self.logger.warn().logf( + "replay: awaiting runtime metadata from accounts_db ({d}s)", + .{elapsed_s}, + ); + } else { + self.logger.info().logf( + "replay: awaiting runtime metadata from accounts_db ({d}s)", + .{elapsed_s}, + ); + } + } + + fn markReady(self: *RuntimeMetadataWaitState) void { + if (self.logged_awaiting and !self.logged_ready) { + self.logged_ready = true; + self.logger.info().log("replay: receiving runtime metadata"); + } + } +}; + +/// Non-blocking + throttled-logging wrapper around +/// `blockhashes_in.getBufferBlocking(runner)`. Mirrors the idle/active signaling +/// of `getBufferBlocking` and additionally emits periodic "still awaiting" logs +/// while the reader has nothing to return. +fn waitForBlockhashes( + blockhashes_in: anytype, + runner: lib.runner.Connection, + wait_state: *RuntimeMetadataWaitState, +) ![]const Hash { + if (blockhashes_in.getBuffer()) |buf| { + wait_state.markReady(); + return buf; + } + const buf = while (true) { + wait_state.maybeLogAwaiting(); + try runner.activity.signalIdleSpinning(); + if (blockhashes_in.getBuffer()) |b| break b; + }; + try runner.activity.signalActive(); + wait_state.markReady(); + return buf; +} + /// Reads all the RuntimeMetadata provided by accountsdb from the snapshot or /// its internal state. This bootstraps replay with information about its /// starting root slot, and some older info like the history of blockhashes. @@ -299,6 +367,18 @@ fn bootstrap( exec_states: *BlockExecStates, blockhash_states: *BlockHashStates, ) !void { + // Bootstrap-blocked observability (issue #1746). Emits a periodic info log + // while replay has not yet received any runtime-metadata bytes from + // accounts_db, and escalates to warn after 15 minutes. Silent on the + // healthy path once the first chunk arrives. + logger.info().log("replay: awaiting runtime metadata from accounts_db"); + const now_ns = lib.clock.monotonic(.ns); + var wait_state: RuntimeMetadataWaitState = .{ + .logger = logger, + .start_ns = now_ns, + .gate = .init(RuntimeMetadataWaitState.AWAITING_LOG_INTERVAL_NS, now_ns), + }; + var num_hashes: usize = 0; // Drain the blockhash queue into the block tree. accountsdb writes into // this ring blocks waiting for the reader (us). @@ -307,7 +387,7 @@ fn bootstrap( defer blockhashes_in.close(); var last_block: ?api.BlockRef = null; while (true) { - const hashes = try blockhashes_in.getBufferBlocking(runner); + const hashes = try waitForBlockhashes(&blockhashes_in, runner, &wait_state); if (hashes.len == 0) break; // blockhashes_out closed their end for (hashes) |*hash| { const block = try block_pool.createId(); From c0d66d96577d7e244ea77ee1b69a5b933d2d900f Mon Sep 17 00:00:00 2001 From: hamza-syndica Date: Thu, 30 Jul 2026 19:42:59 +0600 Subject: [PATCH 2/4] fix: zig build ci --- v2/services/gossip.zig | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/v2/services/gossip.zig b/v2/services/gossip.zig index 6943f91924..bd23cc0c1a 100644 --- a/v2/services/gossip.zig +++ b/v2/services/gossip.zig @@ -118,7 +118,8 @@ pub fn serviceMain(runner: lib.runner.Connection, ro: ReadOnly, rw: ReadWrite) ! // first packet arrives, these variables are never touched again). // See issue #1746. const bootstrap_start_ns = lib.clock.monotonic(.ns); - var awaiting_gate: lib.telemetry.ThrottledLogger = .init(10 * std.time.ns_per_s, bootstrap_start_ns); + var awaiting_gate: lib.telemetry.ThrottledLogger = + .init(10 * std.time.ns_per_s, bootstrap_start_ns); const warn_after_ns = 60 * std.time.ns_per_s; var first_packet_received = false; From 07ae235ea77196bc67ae5859e4bdd39d179ce772 Mon Sep 17 00:00:00 2001 From: hamza-syndica Date: Fri, 31 Jul 2026 16:51:18 +0600 Subject: [PATCH 3/4] restructure for better testing and codecov --- v2/components/snapshot/download.zig | 71 ++++-------- v2/lib/telemetry.zig | 170 ++++++++++++++++++++++++++++ v2/services/accounts_db.zig | 87 ++++++-------- v2/services/gossip.zig | 46 ++++---- v2/services/replay.zig | 94 ++++++--------- 5 files changed, 288 insertions(+), 180 deletions(-) diff --git a/v2/components/snapshot/download.zig b/v2/components/snapshot/download.zig index 155b02d4fc..9e78fb823b 100644 --- a/v2/components/snapshot/download.zig +++ b/v2/components/snapshot/download.zig @@ -929,15 +929,10 @@ pub const Downloader = struct { metrics: Metrics, logger: tel.Logger("snapshot"), - // Bootstrap-blocked observability (issue #1746). - // Emits a periodic info log while the downloader has no candidates, and - // escalates to warn once after `awaiting_warn_after_ns` has elapsed. - // If we ever logged an "awaiting" message, we also emit a one-shot info - // log when a usable peer is found. - bootstrap_start_ns: u64, - awaiting_gate: tel.ThrottledLogger, - logged_awaiting: bool, - logged_ready: bool, + // Bootstrap-blocked observability (issue #1746). Emits a periodic info log + // while the downloader has no candidates, escalates to warn once after + // 60s, and emits a one-shot info log when a usable peer is found. + await_state: tel.BootstrapWait, const AWAITING_LOG_INTERVAL_NS: u64 = 10 * std.time.ns_per_s; const AWAITING_WARN_AFTER_NS: u64 = 60 * std.time.ns_per_s; @@ -964,10 +959,7 @@ pub const Downloader = struct { .run_result = null, .metrics = metrics, .logger = logger, - .bootstrap_start_ns = now_ns, - .awaiting_gate = .init(AWAITING_LOG_INTERVAL_NS, now_ns), - .logged_awaiting = false, - .logged_ready = false, + .await_state = .init(AWAITING_LOG_INTERVAL_NS, AWAITING_WARN_AFTER_NS, now_ns), }; } @@ -2695,11 +2687,8 @@ pub const Downloader = struct { // We only care about the pre-race window. Once `.racing` starts, we // may still transition through failures, but the "awaiting peers" // phase is over — the download-side already logs failures. - const still_awaiting = self.download_race.phase == .idle; - - if (!still_awaiting) { - if (self.logged_awaiting and !self.logged_ready) { - self.logged_ready = true; + if (self.download_race.phase != .idle) { + if (self.await_state.markReady()) { self.logger.info().log( "snapshot: found usable peer, starting download", ); @@ -2707,41 +2696,29 @@ pub const Downloader = struct { return; } - const now_ns = lib.clock.monotonic(.ns); - if (!self.awaiting_gate.tick(now_ns)) return; - - self.logged_awaiting = true; - const elapsed_ns = now_ns -| self.bootstrap_start_ns; - const elapsed_s = elapsed_ns / std.time.ns_per_s; const peers_seen = self.dedupe_map.len; - const escalate = !self.awaiting_gate.escalated and - elapsed_ns >= AWAITING_WARN_AFTER_NS; - if (escalate) self.awaiting_gate.escalated = true; - - if (peers_seen == 0) { - if (escalate) { - self.logger.warn().logf( + switch (self.await_state.tick(lib.clock.monotonic(.ns))) { + .none => {}, + .info => |s| if (peers_seen == 0) + self.logger.info().logf( "snapshot: awaiting peers from gossip ({d}s, none received)", - .{elapsed_s}, - ); - } else { + .{s}, + ) + else self.logger.info().logf( + "snapshot: no usable peers so far ({d}s, received={d}, active_probes={d})", + .{ s, peers_seen, self.active_probes }, + ), + .warn => |s| if (peers_seen == 0) + self.logger.warn().logf( "snapshot: awaiting peers from gossip ({d}s, none received)", - .{elapsed_s}, - ); - } - } else { - if (escalate) { + .{s}, + ) + else self.logger.warn().logf( "snapshot: no usable peers so far ({d}s, received={d}, active_probes={d})", - .{ elapsed_s, peers_seen, self.active_probes }, - ); - } else { - self.logger.info().logf( - "snapshot: no usable peers so far ({d}s, received={d}, active_probes={d})", - .{ elapsed_s, peers_seen, self.active_probes }, - ); - } + .{ s, peers_seen, self.active_probes }, + ), } } diff --git a/v2/lib/telemetry.zig b/v2/lib/telemetry.zig index 16bf862278..a932c67d0a 100644 --- a/v2/lib/telemetry.zig +++ b/v2/lib/telemetry.zig @@ -522,6 +522,176 @@ test "ThrottledLogger: start_ns suppresses early ticks" { try std.testing.expect(t.tick(51_000)); // exactly one interval later } +/// Bootstrap-blocked observability state machine (issue #1746). +/// +/// Encapsulates the "throttled awaiting log, escalate to warn once, emit a +/// one-shot ready log when the wait ends" pattern used by v2 bootstrap +/// services (gossip, snapshot downloader, accounts_db, replay). +/// +/// The state machine is pure — callers pass a monotonic timestamp into +/// `tick`. Tests exercise every branch without any wall-clock dependency. +/// +/// Usage: +/// ``` +/// var wait: BootstrapWait = .init(10 * ns_per_s, 60 * ns_per_s, start_ns); +/// // While the upstream signal is still missing: +/// switch (wait.tick(lib.clock.monotonic(.ns))) { +/// .none => {}, +/// .info => |elapsed_s| logger.info().logf("still waiting ({d}s)", .{elapsed_s}), +/// .warn => |elapsed_s| logger.warn().logf("still waiting ({d}s)", .{elapsed_s}), +/// } +/// // On the transition to ready: +/// if (wait.markReady()) logger.info().log("got it"); +/// ``` +pub const BootstrapWait = struct { + gate: ThrottledLogger, + start_ns: u64, + /// Duration after which `tick` returns `.warn` exactly once instead of `.info`. + warn_after_ns: u64, + /// Whether `tick` has ever returned `.info` or `.warn`. Gates `markReady`. + logged_awaiting: bool = false, + /// Whether `markReady` has ever returned true. One-shot. + logged_ready: bool = false, + + pub const Action = union(enum) { + /// The throttle interval has not elapsed; the caller should not log. + none, + /// Emit an info-level "still awaiting" log with the given elapsed seconds. + info: u64, + /// Emit a warn-level "still awaiting" log with the given elapsed seconds. + /// Only returned once per BootstrapWait; subsequent overdue ticks + /// return `.info` again. + warn: u64, + }; + + pub fn init(interval_ns: u64, warn_after_ns: u64, start_ns: u64) BootstrapWait { + return .{ + .gate = .init(interval_ns, start_ns), + .start_ns = start_ns, + .warn_after_ns = warn_after_ns, + }; + } + + /// Call while the upstream signal is still missing. Returns which log + /// action (if any) the caller should emit at this instant. + pub fn tick(self: *BootstrapWait, now_ns: u64) Action { + if (!self.gate.tick(now_ns)) return .none; + self.logged_awaiting = true; + const elapsed_ns = now_ns -| self.start_ns; + const elapsed_s = elapsed_ns / std.time.ns_per_s; + if (!self.gate.escalated and elapsed_ns >= self.warn_after_ns) { + self.gate.escalated = true; + return .{ .warn = elapsed_s }; + } + return .{ .info = elapsed_s }; + } + + /// Call on the transition to "ready" (upstream signal arrived). Returns + /// true exactly once, and only if a preceding `tick` ever returned + /// `.info` or `.warn`. Callers use this to gate a one-shot info log so + /// the healthy-startup path stays silent. + pub fn markReady(self: *BootstrapWait) bool { + if (self.logged_awaiting and !self.logged_ready) { + self.logged_ready = true; + return true; + } + return false; + } +}; + +test "BootstrapWait: no tick fires before interval elapses" { + var w: BootstrapWait = .init(10 * std.time.ns_per_s, 60 * std.time.ns_per_s, 0); + try std.testing.expectEqual(BootstrapWait.Action.none, w.tick(0)); + try std.testing.expectEqual(BootstrapWait.Action.none, w.tick(9 * std.time.ns_per_s)); + try std.testing.expect(!w.logged_awaiting); +} + +test "BootstrapWait: first tick after interval returns .info" { + var w: BootstrapWait = .init(10 * std.time.ns_per_s, 60 * std.time.ns_per_s, 0); + const action = w.tick(10 * std.time.ns_per_s); + try std.testing.expectEqual(@as(u64, 10), action.info); + try std.testing.expect(w.logged_awaiting); + try std.testing.expect(!w.gate.escalated); +} + +test "BootstrapWait: escalates to .warn exactly once at warn_after_ns" { + var w: BootstrapWait = .init(10 * std.time.ns_per_s, 60 * std.time.ns_per_s, 0); + // Ticks below warn_after_ns stay at .info. + try std.testing.expectEqual(BootstrapWait.Action{ .info = 10 }, w.tick(10 * std.time.ns_per_s)); + try std.testing.expectEqual(BootstrapWait.Action{ .info = 30 }, w.tick(30 * std.time.ns_per_s)); + // First tick past the escalation threshold returns .warn. + try std.testing.expectEqual(BootstrapWait.Action{ .warn = 60 }, w.tick(60 * std.time.ns_per_s)); + try std.testing.expect(w.gate.escalated); + // Subsequent ticks return .info again (warn is one-shot). + try std.testing.expectEqual( + BootstrapWait.Action{ .info = 90 }, + w.tick(90 * std.time.ns_per_s), + ); + try std.testing.expectEqual( + BootstrapWait.Action{ .info = 120 }, + w.tick(120 * std.time.ns_per_s), + ); +} + +test "BootstrapWait: throttle applies between ticks" { + var w: BootstrapWait = .init(10 * std.time.ns_per_s, 600 * std.time.ns_per_s, 0); + try std.testing.expectEqual(BootstrapWait.Action{ .info = 10 }, w.tick(10 * std.time.ns_per_s)); + // Within one interval of the last successful tick. + try std.testing.expectEqual(BootstrapWait.Action.none, w.tick(15 * std.time.ns_per_s)); + try std.testing.expectEqual(BootstrapWait.Action.none, w.tick(19 * std.time.ns_per_s)); + // Exactly one interval later. + try std.testing.expectEqual(BootstrapWait.Action{ .info = 20 }, w.tick(20 * std.time.ns_per_s)); +} + +test "BootstrapWait: escalation deadline before first interval still triggers .warn on first fire" { + // warn_after_ns < interval_ns: the very first log should be at warn level. + var w: BootstrapWait = .init(10 * std.time.ns_per_s, 1 * std.time.ns_per_s, 0); + try std.testing.expectEqual(BootstrapWait.Action{ .warn = 10 }, w.tick(10 * std.time.ns_per_s)); + try std.testing.expect(w.gate.escalated); +} + +test "BootstrapWait: markReady returns false when awaiting never logged" { + var w: BootstrapWait = .init(10 * std.time.ns_per_s, 60 * std.time.ns_per_s, 0); + try std.testing.expect(!w.markReady()); + // Even after ticks that returned .none. + _ = w.tick(5 * std.time.ns_per_s); + try std.testing.expect(!w.markReady()); +} + +test "BootstrapWait: markReady returns true once after tick fired" { + var w: BootstrapWait = .init(10 * std.time.ns_per_s, 60 * std.time.ns_per_s, 0); + _ = w.tick(10 * std.time.ns_per_s); // first .info tick sets logged_awaiting + try std.testing.expect(w.markReady()); + // Second call is a no-op. + try std.testing.expect(!w.markReady()); + try std.testing.expect(!w.markReady()); +} + +test "BootstrapWait: markReady still gated by logged_awaiting after later ticks" { + const start_ns = 100 * std.time.ns_per_s; + var w: BootstrapWait = .init(10 * std.time.ns_per_s, 60 * std.time.ns_per_s, start_ns); + // start_ns=100s means the first tick at now=start_ns does not fire. + _ = w.tick(start_ns); + try std.testing.expect(!w.markReady()); + // Time advances past interval; tick fires; markReady now works. + _ = w.tick(start_ns + 11 * std.time.ns_per_s); + try std.testing.expect(w.markReady()); + try std.testing.expect(!w.markReady()); +} + +test "BootstrapWait: saturating subtract survives clock regression" { + const start_ns = 100 * std.time.ns_per_s; + var w: BootstrapWait = .init(10 * std.time.ns_per_s, 60 * std.time.ns_per_s, start_ns); + // now_ns lower than start_ns — elapsed should saturate to 0, no tick. + try std.testing.expectEqual(BootstrapWait.Action.none, w.tick(50 * std.time.ns_per_s)); + try std.testing.expectEqual(BootstrapWait.Action.none, w.tick(start_ns - 1)); + // Normal progression from start_ns still works. + try std.testing.expectEqual( + BootstrapWait.Action{ .info = 10 }, + w.tick(start_ns + 10 * std.time.ns_per_s), + ); +} + /// Can be used as a counter or a gauge. pub fn Variant(comptime V: type) type { return struct { diff --git a/v2/services/accounts_db.zig b/v2/services/accounts_db.zig index d445ad578c..be5533a3e1 100644 --- a/v2/services/accounts_db.zig +++ b/v2/services/accounts_db.zig @@ -22,48 +22,11 @@ pub const ReadWrite = services.accounts_db.ReadWrite; const ServiceLogger = lib.telemetry.Logger("main"); -/// Bootstrap-blocked observability for the snapshot-bytes wait. Held by pointer -/// on the `SnapshotBufReader` so it survives across pass-by-value copies of the -/// reader. See issue #1746. -const SnapshotWaitState = struct { - logger: ServiceLogger, - start_ns: u64, - gate: lib.telemetry.ThrottledLogger, - logged_awaiting: bool = false, - logged_ready: bool = false, - - const AWAITING_LOG_INTERVAL_NS: u64 = 10 * std.time.ns_per_s; - const AWAITING_WARN_AFTER_NS: u64 = 5 * 60 * std.time.ns_per_s; - - fn maybeLogAwaiting(self: *SnapshotWaitState) void { - const now_ns = lib.clock.monotonic(.ns); - if (!self.gate.tick(now_ns)) return; - self.logged_awaiting = true; - const elapsed_ns = now_ns -| self.start_ns; - const elapsed_s = elapsed_ns / std.time.ns_per_s; - const escalate = - !self.gate.escalated and elapsed_ns >= AWAITING_WARN_AFTER_NS; - if (escalate) { - self.gate.escalated = true; - self.logger.warn().logf( - "accounts_db: awaiting snapshot bytes from snapshot service ({d}s)", - .{elapsed_s}, - ); - } else { - self.logger.info().logf( - "accounts_db: awaiting snapshot bytes from snapshot service ({d}s)", - .{elapsed_s}, - ); - } - } - - fn markReady(self: *SnapshotWaitState) void { - if (self.logged_awaiting and !self.logged_ready) { - self.logged_ready = true; - self.logger.info().log("accounts_db: receiving snapshot bytes"); - } - } -}; +/// Bootstrap-blocked observability thresholds for the snapshot-bytes wait. +/// See issue #1746. Kept as file-scope constants so they're trivially editable +/// without touching call-site logic. +const SNAPSHOT_WAIT_INTERVAL_NS: u64 = 10 * std.time.ns_per_s; +const SNAPSHOT_WAIT_WARN_AFTER_NS: u64 = 5 * 60 * std.time.ns_per_s; pub fn serviceMain(runner: lib.runner.Connection, _: ReadOnly, rw: ReadWrite) !noreturn { const logger = rw.tel.acquireLogger(@tagName(name), "main"); @@ -75,7 +38,9 @@ pub fn serviceMain(runner: lib.runner.Connection, _: ReadOnly, rw: ReadWrite) !n const Global = struct { var fba_memory: [32 * 1024 * 1024]u8 = undefined; var rooted: Rooted = undefined; - var wait_state: SnapshotWaitState = undefined; + // Held by pointer on the `SnapshotBufReader` so it survives across + // pass-by-value copies of the reader. + var wait_state: lib.telemetry.BootstrapWait = undefined; }; const rooted = &Global.rooted; @@ -97,18 +62,19 @@ pub fn serviceMain(runner: lib.runner.Connection, _: ReadOnly, rw: ReadWrite) !n logger.info().log("no existing rooted db. reading from snapshot"); const now_ns = lib.clock.monotonic(.ns); - Global.wait_state = .{ - .logger = logger, - .start_ns = now_ns, - .gate = .init(SnapshotWaitState.AWAITING_LOG_INTERVAL_NS, now_ns), - }; + Global.wait_state = .init( + SNAPSHOT_WAIT_INTERVAL_NS, + SNAPSHOT_WAIT_WARN_AFTER_NS, + now_ns, + ); const SnapshotDataRingReader = @TypeOf(in); const SnapshotBufReader = struct { in_: *SnapshotDataRingReader, runner_: lib.runner.Connection, completion_: *std.atomic.Value(f64), - wait_: *SnapshotWaitState, + wait_: *lib.telemetry.BootstrapWait, + logger_: ServiceLogger, pub fn percentCompleted(self: @This()) f64 { return self.completion_.load(.monotonic); @@ -117,7 +83,11 @@ pub fn serviceMain(runner: lib.runner.Connection, _: ReadOnly, rw: ReadWrite) !n pub fn getBuffer(self: @This()) []const u8 { // Fast path: bytes already available, no waiting. if (self.in_.getBuffer()) |buf| { - self.wait_.markReady(); + // Only mark ready on real data, not on empty-slice EOF + // (see v2/lib/ipc/ring.zig:60-64). + if (buf.len != 0 and self.wait_.markReady()) { + self.logger_.info().log("accounts_db: receiving snapshot bytes"); + } return buf; } @@ -125,12 +95,24 @@ pub fn serviceMain(runner: lib.runner.Connection, _: ReadOnly, rw: ReadWrite) !n // and additionally emit throttled "still awaiting" logs so an // operator can tell the service is blocked upstream (see #1746). const buf = while (true) { - self.wait_.maybeLogAwaiting(); + switch (self.wait_.tick(lib.clock.monotonic(.ns))) { + .none => {}, + .info => |s| self.logger_.info().logf( + "accounts_db: awaiting snapshot bytes from snapshot service ({d}s)", + .{s}, + ), + .warn => |s| self.logger_.warn().logf( + "accounts_db: awaiting snapshot bytes from snapshot service ({d}s)", + .{s}, + ), + } self.runner_.activity.signalIdleSpinning() catch return &.{}; if (self.in_.getBuffer()) |b| break b; }; self.runner_.activity.signalActive() catch return &.{}; - self.wait_.markReady(); + if (buf.len != 0 and self.wait_.markReady()) { + self.logger_.info().log("accounts_db: receiving snapshot bytes"); + } return buf; } @@ -145,6 +127,7 @@ pub fn serviceMain(runner: lib.runner.Connection, _: ReadOnly, rw: ReadWrite) !n .runner_ = runner, .completion_ = &rw.ready_snapshot_in.completion, .wait_ = &Global.wait_state, + .logger_ = logger, }); logger.info().log("reading snapshot accounts"); diff --git a/v2/services/gossip.zig b/v2/services/gossip.zig index bd23cc0c1a..6eddc76bfb 100644 --- a/v2/services/gossip.zig +++ b/v2/services/gossip.zig @@ -113,14 +113,16 @@ pub fn serviceMain(runner: lib.runner.Connection, ro: ReadOnly, rw: ReadWrite) ! var it = rw.net_pair.recv.get(.reader); - // Bootstrap-blocked observability: emit periodic logs while gossip has not - // yet received any inbound packet. Silent on the healthy path (after the - // first packet arrives, these variables are never touched again). - // See issue #1746. + // Bootstrap-blocked observability (issue #1746). Emits periodic logs + // while gossip has not yet received any inbound packet, and escalates + // to warn once after 60s. On the healthy path (first packet arrives + // before the first tick fires) the operator sees nothing. const bootstrap_start_ns = lib.clock.monotonic(.ns); - var awaiting_gate: lib.telemetry.ThrottledLogger = - .init(10 * std.time.ns_per_s, bootstrap_start_ns); - const warn_after_ns = 60 * std.time.ns_per_s; + var wait_state: lib.telemetry.BootstrapWait = .init( + 10 * std.time.ns_per_s, + 60 * std.time.ns_per_s, + bootstrap_start_ns, + ); var first_packet_received = false; while (true) { @@ -135,23 +137,16 @@ pub fn serviceMain(runner: lib.runner.Connection, ro: ReadOnly, rw: ReadWrite) ! // in the one black-box test that currently exists for this). if (!first_packet_received) { - const now_ns = lib.clock.monotonic(.ns); - if (awaiting_gate.tick(now_ns)) { - const elapsed_s = (now_ns -| bootstrap_start_ns) / std.time.ns_per_s; - if (!awaiting_gate.escalated and - now_ns -| bootstrap_start_ns >= warn_after_ns) - { - awaiting_gate.escalated = true; - logger.warn().logf( - "gossip has received no packets from cluster ({d}s)", - .{elapsed_s}, - ); - } else { - logger.info().logf( - "gossip has received no packets from cluster ({d}s)", - .{elapsed_s}, - ); - } + switch (wait_state.tick(lib.clock.monotonic(.ns))) { + .none => {}, + .info => |s| logger.info().logf( + "gossip has received no packets from cluster ({d}s)", + .{s}, + ), + .warn => |s| logger.warn().logf( + "gossip has received no packets from cluster ({d}s)", + .{s}, + ), } } @@ -162,6 +157,9 @@ pub fn serviceMain(runner: lib.runner.Connection, ro: ReadOnly, rw: ReadWrite) ! if (!first_packet_received) { first_packet_received = true; + // `markReady` returns true only if an "awaiting" log was ever emitted; + // we ignore it and always emit the milestone log for gossip. + _ = wait_state.markReady(); const elapsed_s = (lib.clock.monotonic(.ns) -| bootstrap_start_ns) / std.time.ns_per_s; logger.info().logf( diff --git a/v2/services/replay.zig b/v2/services/replay.zig index 59960d4e35..c6c4633e65 100644 --- a/v2/services/replay.zig +++ b/v2/services/replay.zig @@ -275,71 +275,52 @@ pub fn serviceMain(runner: lib.runner.Connection, _: ReadOnly, rw: ReadWrite) !n } } -/// Bootstrap-blocked observability for replay's wait on runtime metadata -/// (blockhash queue + root slot) from accounts_db. Kept next to `bootstrap` -/// because the waits happen only during bootstrap. See issue #1746. -const RuntimeMetadataWaitState = struct { - logger: tel.Logger("main"), - start_ns: u64, - gate: tel.ThrottledLogger, - logged_awaiting: bool = false, - logged_ready: bool = false, - - const AWAITING_LOG_INTERVAL_NS: u64 = 10 * std.time.ns_per_s; - // accounts_db loading a mainnet snapshot can take longer than 10 minutes; - // avoid alerting the operator until we're well past the realistic tail. - const AWAITING_WARN_AFTER_NS: u64 = 15 * 60 * std.time.ns_per_s; - - fn maybeLogAwaiting(self: *RuntimeMetadataWaitState) void { - const now_ns = lib.clock.monotonic(.ns); - if (!self.gate.tick(now_ns)) return; - self.logged_awaiting = true; - const elapsed_ns = now_ns -| self.start_ns; - const elapsed_s = elapsed_ns / std.time.ns_per_s; - const escalate = - !self.gate.escalated and elapsed_ns >= AWAITING_WARN_AFTER_NS; - if (escalate) { - self.gate.escalated = true; - self.logger.warn().logf( - "replay: awaiting runtime metadata from accounts_db ({d}s)", - .{elapsed_s}, - ); - } else { - self.logger.info().logf( - "replay: awaiting runtime metadata from accounts_db ({d}s)", - .{elapsed_s}, - ); - } - } - - fn markReady(self: *RuntimeMetadataWaitState) void { - if (self.logged_awaiting and !self.logged_ready) { - self.logged_ready = true; - self.logger.info().log("replay: receiving runtime metadata"); - } - } -}; +/// Bootstrap-blocked observability thresholds for replay's wait on runtime +/// metadata (blockhash queue + root slot) from accounts_db. See issue #1746. +/// +/// accounts_db loading a mainnet snapshot can take longer than 10 minutes; +/// the warn threshold is set well past that realistic tail so we don't page +/// operators during normal startup. +const RUNTIME_METADATA_WAIT_INTERVAL_NS: u64 = 10 * std.time.ns_per_s; +const RUNTIME_METADATA_WAIT_WARN_AFTER_NS: u64 = 15 * 60 * std.time.ns_per_s; /// Non-blocking + throttled-logging wrapper around /// `blockhashes_in.getBufferBlocking(runner)`. Mirrors the idle/active signaling /// of `getBufferBlocking` and additionally emits periodic "still awaiting" logs -/// while the reader has nothing to return. +/// while the reader has nothing to return (issue #1746). fn waitForBlockhashes( blockhashes_in: anytype, runner: lib.runner.Connection, - wait_state: *RuntimeMetadataWaitState, + logger: tel.Logger("main"), + wait_state: *tel.BootstrapWait, ) ![]const Hash { if (blockhashes_in.getBuffer()) |buf| { - wait_state.markReady(); + // Only mark ready on real data, not on empty-slice EOF + // (see v2/lib/ipc/ring.zig:60-64). + if (buf.len != 0 and wait_state.markReady()) { + logger.info().log("replay: receiving runtime metadata"); + } return buf; } const buf = while (true) { - wait_state.maybeLogAwaiting(); + switch (wait_state.tick(lib.clock.monotonic(.ns))) { + .none => {}, + .info => |s| logger.info().logf( + "replay: awaiting runtime metadata from accounts_db ({d}s)", + .{s}, + ), + .warn => |s| logger.warn().logf( + "replay: awaiting runtime metadata from accounts_db ({d}s)", + .{s}, + ), + } try runner.activity.signalIdleSpinning(); if (blockhashes_in.getBuffer()) |b| break b; }; try runner.activity.signalActive(); - wait_state.markReady(); + if (buf.len != 0 and wait_state.markReady()) { + logger.info().log("replay: receiving runtime metadata"); + } return buf; } @@ -370,14 +351,13 @@ fn bootstrap( // Bootstrap-blocked observability (issue #1746). Emits a periodic info log // while replay has not yet received any runtime-metadata bytes from // accounts_db, and escalates to warn after 15 minutes. Silent on the - // healthy path once the first chunk arrives. - logger.info().log("replay: awaiting runtime metadata from accounts_db"); + // healthy path where the first chunk is already available. const now_ns = lib.clock.monotonic(.ns); - var wait_state: RuntimeMetadataWaitState = .{ - .logger = logger, - .start_ns = now_ns, - .gate = .init(RuntimeMetadataWaitState.AWAITING_LOG_INTERVAL_NS, now_ns), - }; + var wait_state: tel.BootstrapWait = .init( + RUNTIME_METADATA_WAIT_INTERVAL_NS, + RUNTIME_METADATA_WAIT_WARN_AFTER_NS, + now_ns, + ); var num_hashes: usize = 0; // Drain the blockhash queue into the block tree. accountsdb writes into @@ -387,7 +367,7 @@ fn bootstrap( defer blockhashes_in.close(); var last_block: ?api.BlockRef = null; while (true) { - const hashes = try waitForBlockhashes(&blockhashes_in, runner, &wait_state); + const hashes = try waitForBlockhashes(&blockhashes_in, runner, logger, &wait_state); if (hashes.len == 0) break; // blockhashes_out closed their end for (hashes) |*hash| { const block = try block_pool.createId(); From b323d794db2af5b4da0060a0e0ac659173889e22 Mon Sep 17 00:00:00 2001 From: hamza-syndica Date: Fri, 31 Jul 2026 18:33:49 +0600 Subject: [PATCH 4/4] another approach to fix codecov --- v2/components/snapshot/download.zig | 59 ++---- v2/lib/telemetry.zig | 287 +++++++++++++++++++++------- v2/services/accounts_db.zig | 47 +---- v2/services/gossip.zig | 25 +-- v2/services/replay.zig | 52 +---- 5 files changed, 265 insertions(+), 205 deletions(-) diff --git a/v2/components/snapshot/download.zig b/v2/components/snapshot/download.zig index 9e78fb823b..7b44cb10db 100644 --- a/v2/components/snapshot/download.zig +++ b/v2/components/snapshot/download.zig @@ -929,9 +929,7 @@ pub const Downloader = struct { metrics: Metrics, logger: tel.Logger("snapshot"), - // Bootstrap-blocked observability (issue #1746). Emits a periodic info log - // while the downloader has no candidates, escalates to warn once after - // 60s, and emits a one-shot info log when a usable peer is found. + // See issue #1746. await_state: tel.BootstrapWait, const AWAITING_LOG_INTERVAL_NS: u64 = 10 * std.time.ns_per_s; @@ -2676,49 +2674,32 @@ pub const Downloader = struct { }; } - /// Bootstrap observability: log at most once per 10s while we have no - /// download candidates from gossip. Distinguishes between "no peers arrived - /// at all" and "peers arrived but none are usable". Once a candidate has - /// begun racing (or a winner is picked), emits a one-shot "ready" info log - /// if we previously logged that we were waiting. - /// /// See issue #1746. fn maybeLogAwaitingPeers(self: *Downloader) void { - // We only care about the pre-race window. Once `.racing` starts, we - // may still transition through failures, but the "awaiting peers" - // phase is over — the download-side already logs failures. if (self.download_race.phase != .idle) { - if (self.await_state.markReady()) { - self.logger.info().log( - "snapshot: found usable peer, starting download", - ); - } + self.await_state.logReady( + self.logger, + "snapshot: found usable peer, starting download", + ); return; } + const now_ns = lib.clock.monotonic(.ns); const peers_seen = self.dedupe_map.len; - switch (self.await_state.tick(lib.clock.monotonic(.ns))) { - .none => {}, - .info => |s| if (peers_seen == 0) - self.logger.info().logf( - "snapshot: awaiting peers from gossip ({d}s, none received)", - .{s}, - ) - else - self.logger.info().logf( - "snapshot: no usable peers so far ({d}s, received={d}, active_probes={d})", - .{ s, peers_seen, self.active_probes }, - ), - .warn => |s| if (peers_seen == 0) - self.logger.warn().logf( - "snapshot: awaiting peers from gossip ({d}s, none received)", - .{s}, - ) - else - self.logger.warn().logf( - "snapshot: no usable peers so far ({d}s, received={d}, active_probes={d})", - .{ s, peers_seen, self.active_probes }, - ), + if (peers_seen == 0) { + self.await_state.logAwaiting( + now_ns, + self.logger, + "snapshot: awaiting peers from gossip ({d}s, none received)", + .{}, + ); + } else { + self.await_state.logAwaiting( + now_ns, + self.logger, + "snapshot: no usable peers so far (received={d}, active_probes={d}, {d}s)", + .{ peers_seen, self.active_probes }, + ); } } diff --git a/v2/lib/telemetry.zig b/v2/lib/telemetry.zig index a932c67d0a..60c19cda27 100644 --- a/v2/lib/telemetry.zig +++ b/v2/lib/telemetry.zig @@ -453,31 +453,18 @@ pub const Gauge = struct { } }; -/// Emits at most one log per `interval_ns`, with an optional one-shot -/// escalation flag that callers can use to promote the level from info to warn -/// exactly once after some deadline has elapsed. -/// -/// This is intended for use in bootstrap wait loops where a service is blocked -/// on an upstream signal (e.g. gossip peers, snapshot bytes, runtime metadata) -/// and needs to periodically inform the operator without spamming the log. -/// -/// Callers pass `now_ns` from a monotonic clock so this stays testable and -/// avoids taking a dependency on the wall clock here. The first tick fires -/// only after `interval_ns` has elapsed since `start_ns`. +/// Emits at most one log per `interval_ns`. Callers own the `escalated` flag +/// to promote the log level exactly once. Takes `now_ns` from a monotonic +/// clock to keep the state machine testable. pub const ThrottledLogger = struct { interval_ns: u64, last_ns: u64, - /// Set to true once the caller has emitted the escalated (e.g. warn) log. - /// Use to keep escalation to a single emission per stage. escalated: bool = false, pub fn init(interval_ns: u64, start_ns: u64) ThrottledLogger { return .{ .interval_ns = interval_ns, .last_ns = start_ns }; } - /// Returns true if at least `interval_ns` has elapsed since the last tick - /// that returned true (or since construction). When it returns true, the - /// internal timer is advanced to `now_ns`. pub fn tick(self: *ThrottledLogger, now_ns: u64) bool { // Saturating subtract guards against a hypothetical clock regression. if (now_ns -| self.last_ns >= self.interval_ns) { @@ -510,57 +497,29 @@ test "ThrottledLogger: escalated flag is caller-owned and starts false" { try std.testing.expect(!t.escalated); t.escalated = true; try std.testing.expect(t.escalated); - // Escalation flag does not affect tick behavior. try std.testing.expect(t.tick(2_000)); } test "ThrottledLogger: start_ns suppresses early ticks" { - // Simulates production usage: monotonic clock is already large at construction. var t: ThrottledLogger = .init(1_000, 50_000); - try std.testing.expect(!t.tick(50_000)); // same instant as construction - try std.testing.expect(!t.tick(50_500)); // within interval - try std.testing.expect(t.tick(51_000)); // exactly one interval later + try std.testing.expect(!t.tick(50_000)); + try std.testing.expect(!t.tick(50_500)); + try std.testing.expect(t.tick(51_000)); } -/// Bootstrap-blocked observability state machine (issue #1746). -/// -/// Encapsulates the "throttled awaiting log, escalate to warn once, emit a -/// one-shot ready log when the wait ends" pattern used by v2 bootstrap -/// services (gossip, snapshot downloader, accounts_db, replay). -/// -/// The state machine is pure — callers pass a monotonic timestamp into -/// `tick`. Tests exercise every branch without any wall-clock dependency. -/// -/// Usage: -/// ``` -/// var wait: BootstrapWait = .init(10 * ns_per_s, 60 * ns_per_s, start_ns); -/// // While the upstream signal is still missing: -/// switch (wait.tick(lib.clock.monotonic(.ns))) { -/// .none => {}, -/// .info => |elapsed_s| logger.info().logf("still waiting ({d}s)", .{elapsed_s}), -/// .warn => |elapsed_s| logger.warn().logf("still waiting ({d}s)", .{elapsed_s}), -/// } -/// // On the transition to ready: -/// if (wait.markReady()) logger.info().log("got it"); -/// ``` +/// Throttled awaiting-log + one-shot warn escalation + one-shot ready log, +/// used by v2 bootstrap services blocked on an upstream signal. See #1746. pub const BootstrapWait = struct { gate: ThrottledLogger, start_ns: u64, - /// Duration after which `tick` returns `.warn` exactly once instead of `.info`. warn_after_ns: u64, - /// Whether `tick` has ever returned `.info` or `.warn`. Gates `markReady`. logged_awaiting: bool = false, - /// Whether `markReady` has ever returned true. One-shot. logged_ready: bool = false, pub const Action = union(enum) { - /// The throttle interval has not elapsed; the caller should not log. none, - /// Emit an info-level "still awaiting" log with the given elapsed seconds. info: u64, - /// Emit a warn-level "still awaiting" log with the given elapsed seconds. - /// Only returned once per BootstrapWait; subsequent overdue ticks - /// return `.info` again. + /// Returned at most once; subsequent overdue ticks return `.info`. warn: u64, }; @@ -572,8 +531,6 @@ pub const BootstrapWait = struct { }; } - /// Call while the upstream signal is still missing. Returns which log - /// action (if any) the caller should emit at this instant. pub fn tick(self: *BootstrapWait, now_ns: u64) Action { if (!self.gate.tick(now_ns)) return .none; self.logged_awaiting = true; @@ -586,10 +543,7 @@ pub const BootstrapWait = struct { return .{ .info = elapsed_s }; } - /// Call on the transition to "ready" (upstream signal arrived). Returns - /// true exactly once, and only if a preceding `tick` ever returned - /// `.info` or `.warn`. Callers use this to gate a one-shot info log so - /// the healthy-startup path stays silent. + /// Returns true exactly once, and only if a prior `tick` fired. pub fn markReady(self: *BootstrapWait) bool { if (self.logged_awaiting and !self.logged_ready) { self.logged_ready = true; @@ -597,8 +551,55 @@ pub const BootstrapWait = struct { } return false; } + + /// `tick` + emit. `fmt` must end with `({d}s)`; elapsed_s is appended last. + pub fn logAwaiting( + self: *BootstrapWait, + now_ns: u64, + logger: anytype, + comptime fmt: []const u8, + args: anytype, + ) void { + switch (self.tick(now_ns)) { + .none => {}, + .info => |s| logger.info().logf(fmt, args ++ .{s}), + .warn => |s| logger.warn().logf(fmt, args ++ .{s}), + } + } + + pub fn logReady( + self: *BootstrapWait, + logger: anytype, + comptime msg: []const u8, + ) void { + if (self.markReady()) logger.info().log(msg); + } }; +/// `View.getBufferBlocking(runner)` with bootstrap-awaiting logs woven in. +/// Skips the ready log on empty-slice EOF (see `v2/lib/ipc/ring.zig:60-64`). +pub fn waitForBufferWithAwaitingLog( + view: anytype, + runner: anytype, + wait: *BootstrapWait, + logger: anytype, + comptime awaiting_fmt: []const u8, + comptime ready_msg: []const u8, +) !@typeInfo(@TypeOf(view.getBuffer())).optional.child { + if (view.getBuffer()) |buf| { + if (buf.len != 0) wait.logReady(logger, ready_msg); + return buf; + } + const buf = while (true) { + wait.logAwaiting(clock.monotonic(.ns), logger, awaiting_fmt, .{}); + try runner.activity.signalIdleSpinning(); + if (view.getBuffer()) |b| break b; + }; + try runner.activity.signalActive(); + if (buf.len != 0) wait.logReady(logger, ready_msg); + return buf; +} + test "BootstrapWait: no tick fires before interval elapses" { var w: BootstrapWait = .init(10 * std.time.ns_per_s, 60 * std.time.ns_per_s, 0); try std.testing.expectEqual(BootstrapWait.Action.none, w.tick(0)); @@ -616,13 +617,10 @@ test "BootstrapWait: first tick after interval returns .info" { test "BootstrapWait: escalates to .warn exactly once at warn_after_ns" { var w: BootstrapWait = .init(10 * std.time.ns_per_s, 60 * std.time.ns_per_s, 0); - // Ticks below warn_after_ns stay at .info. try std.testing.expectEqual(BootstrapWait.Action{ .info = 10 }, w.tick(10 * std.time.ns_per_s)); try std.testing.expectEqual(BootstrapWait.Action{ .info = 30 }, w.tick(30 * std.time.ns_per_s)); - // First tick past the escalation threshold returns .warn. try std.testing.expectEqual(BootstrapWait.Action{ .warn = 60 }, w.tick(60 * std.time.ns_per_s)); try std.testing.expect(w.gate.escalated); - // Subsequent ticks return .info again (warn is one-shot). try std.testing.expectEqual( BootstrapWait.Action{ .info = 90 }, w.tick(90 * std.time.ns_per_s), @@ -636,15 +634,12 @@ test "BootstrapWait: escalates to .warn exactly once at warn_after_ns" { test "BootstrapWait: throttle applies between ticks" { var w: BootstrapWait = .init(10 * std.time.ns_per_s, 600 * std.time.ns_per_s, 0); try std.testing.expectEqual(BootstrapWait.Action{ .info = 10 }, w.tick(10 * std.time.ns_per_s)); - // Within one interval of the last successful tick. try std.testing.expectEqual(BootstrapWait.Action.none, w.tick(15 * std.time.ns_per_s)); try std.testing.expectEqual(BootstrapWait.Action.none, w.tick(19 * std.time.ns_per_s)); - // Exactly one interval later. try std.testing.expectEqual(BootstrapWait.Action{ .info = 20 }, w.tick(20 * std.time.ns_per_s)); } test "BootstrapWait: escalation deadline before first interval still triggers .warn on first fire" { - // warn_after_ns < interval_ns: the very first log should be at warn level. var w: BootstrapWait = .init(10 * std.time.ns_per_s, 1 * std.time.ns_per_s, 0); try std.testing.expectEqual(BootstrapWait.Action{ .warn = 10 }, w.tick(10 * std.time.ns_per_s)); try std.testing.expect(w.gate.escalated); @@ -653,16 +648,14 @@ test "BootstrapWait: escalation deadline before first interval still triggers .w test "BootstrapWait: markReady returns false when awaiting never logged" { var w: BootstrapWait = .init(10 * std.time.ns_per_s, 60 * std.time.ns_per_s, 0); try std.testing.expect(!w.markReady()); - // Even after ticks that returned .none. _ = w.tick(5 * std.time.ns_per_s); try std.testing.expect(!w.markReady()); } test "BootstrapWait: markReady returns true once after tick fired" { var w: BootstrapWait = .init(10 * std.time.ns_per_s, 60 * std.time.ns_per_s, 0); - _ = w.tick(10 * std.time.ns_per_s); // first .info tick sets logged_awaiting + _ = w.tick(10 * std.time.ns_per_s); try std.testing.expect(w.markReady()); - // Second call is a no-op. try std.testing.expect(!w.markReady()); try std.testing.expect(!w.markReady()); } @@ -670,10 +663,8 @@ test "BootstrapWait: markReady returns true once after tick fired" { test "BootstrapWait: markReady still gated by logged_awaiting after later ticks" { const start_ns = 100 * std.time.ns_per_s; var w: BootstrapWait = .init(10 * std.time.ns_per_s, 60 * std.time.ns_per_s, start_ns); - // start_ns=100s means the first tick at now=start_ns does not fire. _ = w.tick(start_ns); try std.testing.expect(!w.markReady()); - // Time advances past interval; tick fires; markReady now works. _ = w.tick(start_ns + 11 * std.time.ns_per_s); try std.testing.expect(w.markReady()); try std.testing.expect(!w.markReady()); @@ -682,16 +673,174 @@ test "BootstrapWait: markReady still gated by logged_awaiting after later ticks" test "BootstrapWait: saturating subtract survives clock regression" { const start_ns = 100 * std.time.ns_per_s; var w: BootstrapWait = .init(10 * std.time.ns_per_s, 60 * std.time.ns_per_s, start_ns); - // now_ns lower than start_ns — elapsed should saturate to 0, no tick. try std.testing.expectEqual(BootstrapWait.Action.none, w.tick(50 * std.time.ns_per_s)); try std.testing.expectEqual(BootstrapWait.Action.none, w.tick(start_ns - 1)); - // Normal progression from start_ns still works. try std.testing.expectEqual( BootstrapWait.Action{ .info = 10 }, w.tick(start_ns + 10 * std.time.ns_per_s), ); } +test "BootstrapWait: logAwaiting dispatches all three branches through a noop logger" { + const logger = Logger("test").noop; + var w: BootstrapWait = .init(10 * std.time.ns_per_s, 60 * std.time.ns_per_s, 0); + + w.logAwaiting(5 * std.time.ns_per_s, logger, "test: waiting ({d}s)", .{}); + try std.testing.expect(!w.logged_awaiting); + + w.logAwaiting(10 * std.time.ns_per_s, logger, "test: waiting ({d}s)", .{}); + try std.testing.expect(w.logged_awaiting); + try std.testing.expect(!w.gate.escalated); + + w.logAwaiting(60 * std.time.ns_per_s, logger, "test: waiting ({d}s)", .{}); + try std.testing.expect(w.gate.escalated); + + w.logAwaiting(90 * std.time.ns_per_s, logger, "test: waiting ({d}s)", .{}); + try std.testing.expect(w.gate.escalated); +} + +test "BootstrapWait: logAwaiting appends elapsed_s to arbitrary leading args" { + const logger = Logger("test").noop; + var w: BootstrapWait = .init(10 * std.time.ns_per_s, 60 * std.time.ns_per_s, 0); + const peers_seen: usize = 3; + const active_probes: u8 = 2; + w.logAwaiting( + 10 * std.time.ns_per_s, + logger, + "test: no usable peers (received={d}, active_probes={d}, {d}s)", + .{ peers_seen, active_probes }, + ); + try std.testing.expect(w.logged_awaiting); +} + +test "BootstrapWait: logReady is a no-op when awaiting was never logged" { + const logger = Logger("test").noop; + var w: BootstrapWait = .init(10 * std.time.ns_per_s, 60 * std.time.ns_per_s, 0); + w.logReady(logger, "test: ready"); + try std.testing.expect(!w.logged_ready); +} + +test "BootstrapWait: logReady emits exactly once after a prior awaiting log" { + const logger = Logger("test").noop; + var w: BootstrapWait = .init(10 * std.time.ns_per_s, 60 * std.time.ns_per_s, 0); + w.logAwaiting(10 * std.time.ns_per_s, logger, "test: waiting ({d}s)", .{}); + w.logReady(logger, "test: ready"); + try std.testing.expect(w.logged_ready); + w.logReady(logger, "test: ready"); + try std.testing.expect(w.logged_ready); +} + +const TestMocks = struct { + const View = struct { + script: []const ?[]const u8, + cursor: usize = 0, + pub fn getBuffer(self: *View) ?[]const u8 { + const i = self.cursor; + self.cursor += 1; + return if (i < self.script.len) self.script[i] else null; + } + }; + const Activity = struct { + idle_calls: u32 = 0, + active_calls: u32 = 0, + fail_after: ?u32 = null, + pub fn signalIdleSpinning(self: *Activity) !void { + if (self.fail_after) |n| if (self.idle_calls >= n) return error.Canceled; + self.idle_calls += 1; + } + pub fn signalActive(self: *Activity) !void { + self.active_calls += 1; + } + }; + const Runner = struct { + activity: *Activity, + }; +}; + +test "waitForBufferWithAwaitingLog: fast path returns immediately without signaling" { + var view: TestMocks.View = .{ .script = &.{"hello"} }; + var activity: TestMocks.Activity = .{}; + const runner: TestMocks.Runner = .{ .activity = &activity }; + var wait: BootstrapWait = .init(10 * std.time.ns_per_s, 60 * std.time.ns_per_s, 0); + const logger = Logger("test").noop; + + const buf = try waitForBufferWithAwaitingLog( + &view, + runner, + &wait, + logger, + "test: awaiting ({d}s)", + "test: ready", + ); + + try std.testing.expectEqualStrings("hello", buf); + try std.testing.expectEqual(@as(u32, 0), activity.idle_calls); + try std.testing.expectEqual(@as(u32, 0), activity.active_calls); + try std.testing.expect(!wait.logged_awaiting); + try std.testing.expect(!wait.logged_ready); +} + +test "waitForBufferWithAwaitingLog: slow path signals idle then active, emits ready log" { + var view: TestMocks.View = .{ .script = &.{ null, null, null, "delayed" } }; + var activity: TestMocks.Activity = .{}; + const runner: TestMocks.Runner = .{ .activity = &activity }; + var wait: BootstrapWait = .init(10 * std.time.ns_per_s, 60 * std.time.ns_per_s, 0); + const logger = Logger("test").noop; + + const buf = try waitForBufferWithAwaitingLog( + &view, + runner, + &wait, + logger, + "test: awaiting ({d}s)", + "test: ready", + ); + + try std.testing.expectEqualStrings("delayed", buf); + try std.testing.expect(activity.idle_calls >= 1); + try std.testing.expectEqual(@as(u32, 1), activity.active_calls); +} + +test "waitForBufferWithAwaitingLog: empty-slice EOF does not trigger ready log" { + const empty: []const u8 = &.{}; + var view: TestMocks.View = .{ .script = &.{empty} }; + var activity: TestMocks.Activity = .{}; + const runner: TestMocks.Runner = .{ .activity = &activity }; + var wait: BootstrapWait = .init(10 * std.time.ns_per_s, 60 * std.time.ns_per_s, 0); + wait.logAwaiting(10 * std.time.ns_per_s, Logger("test").noop, "test: awaiting ({d}s)", .{}); + + const buf = try waitForBufferWithAwaitingLog( + &view, + runner, + &wait, + Logger("test").noop, + "test: awaiting ({d}s)", + "test: ready", + ); + + try std.testing.expectEqual(@as(usize, 0), buf.len); + try std.testing.expect(!wait.logged_ready); +} + +test "waitForBufferWithAwaitingLog: signalIdleSpinning error propagates" { + var view: TestMocks.View = .{ .script = &.{null} }; + var activity: TestMocks.Activity = .{ .fail_after = 0 }; + const runner: TestMocks.Runner = .{ .activity = &activity }; + var wait: BootstrapWait = .init(10 * std.time.ns_per_s, 60 * std.time.ns_per_s, 0); + const logger = Logger("test").noop; + + const result = waitForBufferWithAwaitingLog( + &view, + runner, + &wait, + logger, + "test: awaiting ({d}s)", + "test: ready", + ); + + try std.testing.expectError(error.Canceled, result); +} + /// Can be used as a counter or a gauge. pub fn Variant(comptime V: type) type { return struct { diff --git a/v2/services/accounts_db.zig b/v2/services/accounts_db.zig index be5533a3e1..448452a7fa 100644 --- a/v2/services/accounts_db.zig +++ b/v2/services/accounts_db.zig @@ -22,9 +22,6 @@ pub const ReadWrite = services.accounts_db.ReadWrite; const ServiceLogger = lib.telemetry.Logger("main"); -/// Bootstrap-blocked observability thresholds for the snapshot-bytes wait. -/// See issue #1746. Kept as file-scope constants so they're trivially editable -/// without touching call-site logic. const SNAPSHOT_WAIT_INTERVAL_NS: u64 = 10 * std.time.ns_per_s; const SNAPSHOT_WAIT_WARN_AFTER_NS: u64 = 5 * 60 * std.time.ns_per_s; @@ -38,8 +35,7 @@ pub fn serviceMain(runner: lib.runner.Connection, _: ReadOnly, rw: ReadWrite) !n const Global = struct { var fba_memory: [32 * 1024 * 1024]u8 = undefined; var rooted: Rooted = undefined; - // Held by pointer on the `SnapshotBufReader` so it survives across - // pass-by-value copies of the reader. + // Held by pointer on `SnapshotBufReader` to survive pass-by-value copies. var wait_state: lib.telemetry.BootstrapWait = undefined; }; @@ -81,39 +77,14 @@ pub fn serviceMain(runner: lib.runner.Connection, _: ReadOnly, rw: ReadWrite) !n } pub fn getBuffer(self: @This()) []const u8 { - // Fast path: bytes already available, no waiting. - if (self.in_.getBuffer()) |buf| { - // Only mark ready on real data, not on empty-slice EOF - // (see v2/lib/ipc/ring.zig:60-64). - if (buf.len != 0 and self.wait_.markReady()) { - self.logger_.info().log("accounts_db: receiving snapshot bytes"); - } - return buf; - } - - // Slow path: mirror `getBufferBlocking`'s idle/active signaling, - // and additionally emit throttled "still awaiting" logs so an - // operator can tell the service is blocked upstream (see #1746). - const buf = while (true) { - switch (self.wait_.tick(lib.clock.monotonic(.ns))) { - .none => {}, - .info => |s| self.logger_.info().logf( - "accounts_db: awaiting snapshot bytes from snapshot service ({d}s)", - .{s}, - ), - .warn => |s| self.logger_.warn().logf( - "accounts_db: awaiting snapshot bytes from snapshot service ({d}s)", - .{s}, - ), - } - self.runner_.activity.signalIdleSpinning() catch return &.{}; - if (self.in_.getBuffer()) |b| break b; - }; - self.runner_.activity.signalActive() catch return &.{}; - if (buf.len != 0 and self.wait_.markReady()) { - self.logger_.info().log("accounts_db: receiving snapshot bytes"); - } - return buf; + return lib.telemetry.waitForBufferWithAwaitingLog( + self.in_, + self.runner_, + self.wait_, + self.logger_, + "accounts_db: awaiting snapshot bytes from snapshot service ({d}s)", + "accounts_db: receiving snapshot bytes", + ) catch return &.{}; } pub fn advance(self: @This(), n: usize) void { diff --git a/v2/services/gossip.zig b/v2/services/gossip.zig index 6eddc76bfb..7466714c6d 100644 --- a/v2/services/gossip.zig +++ b/v2/services/gossip.zig @@ -113,10 +113,7 @@ pub fn serviceMain(runner: lib.runner.Connection, ro: ReadOnly, rw: ReadWrite) ! var it = rw.net_pair.recv.get(.reader); - // Bootstrap-blocked observability (issue #1746). Emits periodic logs - // while gossip has not yet received any inbound packet, and escalates - // to warn once after 60s. On the healthy path (first packet arrives - // before the first tick fires) the operator sees nothing. + // See issue #1746. const bootstrap_start_ns = lib.clock.monotonic(.ns); var wait_state: lib.telemetry.BootstrapWait = .init( 10 * std.time.ns_per_s, @@ -137,17 +134,12 @@ pub fn serviceMain(runner: lib.runner.Connection, ro: ReadOnly, rw: ReadWrite) ! // in the one black-box test that currently exists for this). if (!first_packet_received) { - switch (wait_state.tick(lib.clock.monotonic(.ns))) { - .none => {}, - .info => |s| logger.info().logf( - "gossip has received no packets from cluster ({d}s)", - .{s}, - ), - .warn => |s| logger.warn().logf( - "gossip has received no packets from cluster ({d}s)", - .{s}, - ), - } + wait_state.logAwaiting( + lib.clock.monotonic(.ns), + logger, + "gossip has received no packets from cluster ({d}s)", + .{}, + ); } try runner.activity.signalIdleSpinning(); @@ -157,8 +149,7 @@ pub fn serviceMain(runner: lib.runner.Connection, ro: ReadOnly, rw: ReadWrite) ! if (!first_packet_received) { first_packet_received = true; - // `markReady` returns true only if an "awaiting" log was ever emitted; - // we ignore it and always emit the milestone log for gossip. + // Milestone log fires unconditionally; ignore markReady's gate. _ = wait_state.markReady(); const elapsed_s = (lib.clock.monotonic(.ns) -| bootstrap_start_ns) / std.time.ns_per_s; diff --git a/v2/services/replay.zig b/v2/services/replay.zig index c6c4633e65..6d1f5eaca1 100644 --- a/v2/services/replay.zig +++ b/v2/services/replay.zig @@ -275,53 +275,25 @@ pub fn serviceMain(runner: lib.runner.Connection, _: ReadOnly, rw: ReadWrite) !n } } -/// Bootstrap-blocked observability thresholds for replay's wait on runtime -/// metadata (blockhash queue + root slot) from accounts_db. See issue #1746. -/// -/// accounts_db loading a mainnet snapshot can take longer than 10 minutes; -/// the warn threshold is set well past that realistic tail so we don't page -/// operators during normal startup. +// Warn threshold sits past the realistic tail of mainnet snapshot loading +// so operators aren't paged during normal startup. See issue #1746. const RUNTIME_METADATA_WAIT_INTERVAL_NS: u64 = 10 * std.time.ns_per_s; const RUNTIME_METADATA_WAIT_WARN_AFTER_NS: u64 = 15 * 60 * std.time.ns_per_s; -/// Non-blocking + throttled-logging wrapper around -/// `blockhashes_in.getBufferBlocking(runner)`. Mirrors the idle/active signaling -/// of `getBufferBlocking` and additionally emits periodic "still awaiting" logs -/// while the reader has nothing to return (issue #1746). fn waitForBlockhashes( blockhashes_in: anytype, runner: lib.runner.Connection, logger: tel.Logger("main"), wait_state: *tel.BootstrapWait, ) ![]const Hash { - if (blockhashes_in.getBuffer()) |buf| { - // Only mark ready on real data, not on empty-slice EOF - // (see v2/lib/ipc/ring.zig:60-64). - if (buf.len != 0 and wait_state.markReady()) { - logger.info().log("replay: receiving runtime metadata"); - } - return buf; - } - const buf = while (true) { - switch (wait_state.tick(lib.clock.monotonic(.ns))) { - .none => {}, - .info => |s| logger.info().logf( - "replay: awaiting runtime metadata from accounts_db ({d}s)", - .{s}, - ), - .warn => |s| logger.warn().logf( - "replay: awaiting runtime metadata from accounts_db ({d}s)", - .{s}, - ), - } - try runner.activity.signalIdleSpinning(); - if (blockhashes_in.getBuffer()) |b| break b; - }; - try runner.activity.signalActive(); - if (buf.len != 0 and wait_state.markReady()) { - logger.info().log("replay: receiving runtime metadata"); - } - return buf; + return tel.waitForBufferWithAwaitingLog( + blockhashes_in, + runner, + wait_state, + logger, + "replay: awaiting runtime metadata from accounts_db ({d}s)", + "replay: receiving runtime metadata", + ); } /// Reads all the RuntimeMetadata provided by accountsdb from the snapshot or @@ -348,10 +320,6 @@ fn bootstrap( exec_states: *BlockExecStates, blockhash_states: *BlockHashStates, ) !void { - // Bootstrap-blocked observability (issue #1746). Emits a periodic info log - // while replay has not yet received any runtime-metadata bytes from - // accounts_db, and escalates to warn after 15 minutes. Silent on the - // healthy path where the first chunk is already available. const now_ns = lib.clock.monotonic(.ns); var wait_state: tel.BootstrapWait = .init( RUNTIME_METADATA_WAIT_INTERVAL_NS,