Merge branch 'pratik/otel-phase9-metric-gap-fill' into pratik/otel-phase10-workload-validation

This commit is contained in:
Pratik Mankawde
2026-07-28 18:23:13 +01:00
5 changed files with 186 additions and 39 deletions

View File

@@ -34,6 +34,7 @@
#include <xrpl/rdb/DatabaseCon.h>
#include <algorithm>
#include <atomic>
#include <chrono>
#include <cstddef>
#include <cstdint>
@@ -123,6 +124,72 @@ private:
backend.store(object);
}
/**
* A DummyScheduler that also totals what the fetch reports carried.
*
* Database::fetchNodeObject() measures each fetch once and uses that one
* value for both its internal accumulator and the report it hands the
* scheduler, so the report total is an independent reading of the same
* measurement. Comparing the two pins the accumulator without asserting
* that a fetch took any particular amount of time -- which is a statement
* about the machine, not the code, and fails on a host fast enough to
* serve every read in under a microsecond.
*
* @note Counters are atomic because Scheduler is called from the
* nodestore's read threads in production. These tests fetch
* synchronously on one thread, so the totals are still exact.
*/
struct CountingScheduler : DummyScheduler
{
/**
* Sum of the durations carried by every fetch report received.
*/
std::atomic<std::uint64_t> reportedFetchUs{0};
/**
* Fetch reports received, however long each one measured.
*/
std::atomic<std::uint64_t> fetchReports{0};
void
onFetch(FetchReport const& report) override
{
reportedFetchUs += static_cast<std::uint64_t>(
std::chrono::duration_cast<std::chrono::microseconds>(report.elapsed).count());
++fetchReports;
}
};
/**
* Assert the read accumulator holds exactly what the fetch reports carried.
*
* Deliberately not `getFetchDurationUs() > 0`: an individual read served
* from NuDB's in-memory buckets can genuinely measure under one
* microsecond and truncate to zero, so on a fast enough host every read
* truncates and the total stays at zero with nothing wrong. That makes
* `> 0` an assertion about the machine rather than about the code, and it
* is what failed on the macOS runner.
*
* The equality below is machine-independent and strictly stronger: each
* fetch is measured once and that one value feeds both the accumulator
* and the report, so an accumulator that dropped a fetch, double-counted
* one, or reported the write member instead would break the equality
* however fast the host is.
*
* @param db Database whose read accumulator is checked.
* @param scheduler Scheduler that received the reports for @p db.
* @param expectedReports Fetches that must have been reported.
*/
void
expectFetchDurationMatchesReports(
Database const& db,
CountingScheduler const& scheduler,
std::uint64_t expectedReports)
{
BEAST_EXPECT(scheduler.fetchReports.load() == expectedReports);
BEAST_EXPECT(db.getFetchDurationUs() == scheduler.reportedFetchUs.load());
}
public:
void
testConfig()
@@ -705,7 +772,7 @@ public:
{
testcase("Fetch and store duration accessors");
DummyScheduler scheduler;
CountingScheduler scheduler;
beast::TempDir const nodeDb;
Section nodeParams;
@@ -762,7 +829,7 @@ public:
BEAST_EXPECT(db->getFetchTotalCount() == kNumStored);
BEAST_EXPECT(db->getFetchHitCount() == kNumStored);
BEAST_EXPECT(db->getFetchDurationUs() > 0);
expectFetchDurationMatchesReports(*db, scheduler, kNumStored);
// Reads must leave the write accumulator alone. This also proves the
// write accessor does not report the read member.
@@ -838,7 +905,7 @@ public:
{
testcase("Rotating store duration accessors");
DummyScheduler scheduler;
CountingScheduler scheduler;
beast::TempDir const writableDir;
beast::TempDir const archiveDir;
@@ -876,7 +943,7 @@ public:
BEAST_EXPECT(db.getFetchTotalCount() == kNumStored);
BEAST_EXPECT(db.getFetchHitCount() == kNumStored);
BEAST_EXPECT(db.getFetchDurationUs() > 0);
expectFetchDurationMatchesReports(db, scheduler, kNumStored);
BEAST_EXPECT(db.getStoreCount() == kNumStored);
BEAST_EXPECT(db.getStoreDurationUs() == storeDurationAfterWrites);
@@ -1076,7 +1143,8 @@ public:
// Exact counts, so the denominators below are pinned independently.
BEAST_EXPECT(busy.value("node_writes") == std::int64_t{kNumStored});
BEAST_EXPECT(busy.value("node_reads_total") == std::int64_t{2 * kNumStored + kNumMissing});
BEAST_EXPECT(
busy.value("node_reads_total") == std::int64_t{(2 * kNumStored) + kNumMissing});
BEAST_EXPECT(busy.value("node_reads_hit") == std::int64_t{2 * kNumStored});
// The two denominators differ, which is what makes the cross-checks
// below able to catch a swapped accessor pair.
@@ -1208,7 +1276,7 @@ public:
// A quiet node publishes all seven at zero. Unlike a mean, zero is the
// meaningful "no such event yet" reading for a counter, so omitting
// these would lose the ability to see that nothing happened.
AcquireStats quiet;
AcquireStats const quiet;
MetricSink fresh;
telemetry::MetricsRegistry::observeAcquireStats(quiet, fresh.fn());
BEAST_EXPECT(fresh.names() == kLabels);
@@ -1284,7 +1352,11 @@ public:
// above the total. Compared as unwrapped values, since comparing two
// optionals would also pass with both absent.
auto const running = sink.value("read_threads_running");
if (BEAST_EXPECT(running.has_value()))
BEAST_EXPECT(running.has_value());
// Plain `if` rather than branching on BEAST_EXPECT's result: the
// macro's return value hides the check from static analysis, which
// then reads the dereferences below as unguarded.
if (running.has_value())
{
BEAST_EXPECT(*running >= 0);
BEAST_EXPECT(*running <= 3);

View File

@@ -794,49 +794,73 @@ class TMGetObjectByHash_test : public beast::unit_test::Suite
return {
// Wholly free: at or below the free allowance nothing is
// billable, so only the small size band applies -- which is 0.
{k.free, k.free, k.bandSmall, 0, "at the free allowance"},
{0, 0, k.bandSmall, 0, "empty request"},
{1, 1, k.bandSmall, 0, "one object, one hit"},
{.requested = k.free,
.found = k.free,
.derived = k.bandSmall,
.literal = 0,
.why = "at the free allowance"},
{.requested = 0,
.found = 0,
.derived = k.bandSmall,
.literal = 0,
.why = "empty request"},
{.requested = 1,
.found = 1,
.derived = k.bandSmall,
.literal = 0,
.why = "one object, one hit"},
// All hits, one object past the allowance: one billable hit.
{k.free + 1, k.free + 1, k.hit + k.bandSmall, 1, "one billable hit"},
{.requested = k.free + 1,
.found = k.free + 1,
.derived = k.hit + k.bandSmall,
.literal = 1,
.why = "one billable hit"},
// All misses, one past the allowance: misses are billed first,
// so the single billable object is priced as a miss, not a hit.
{k.free + 1, 0, k.miss + k.bandSmall, 8, "one billable miss"},
{.requested = k.free + 1,
.found = 0,
.derived = k.miss + k.bandSmall,
.literal = 8,
.why = "one billable miss"},
// Mixed at the small-band edge: 64 requested, 32 found.
// Billable is 64-16 = 48; misses are 32 and all billable,
// leaving 16 billable hits.
{k.smallMax,
32,
(16 * k.hit) + (32 * k.miss) + k.bandSmall,
272,
"small-band edge, mixed"},
{.requested = k.smallMax,
.found = 32,
.derived = (16 * k.hit) + (32 * k.miss) + k.bandSmall,
.literal = 272,
.why = "small-band edge, mixed"},
// One past the small band moves to the medium surcharge.
{k.smallMax + 1,
k.smallMax + 1,
((k.smallMax + 1 - k.free) * k.hit) + k.bandMedium,
149,
"first medium-band size"},
{.requested = k.smallMax + 1,
.found = k.smallMax + 1,
.derived = ((k.smallMax + 1 - k.free) * k.hit) + k.bandMedium,
.literal = 149,
.why = "first medium-band size"},
// The medium band's last size, then one past it, which moves to
// the large surcharge.
{k.mediumMax,
k.mediumMax,
((k.mediumMax - k.free) * k.hit) + k.bandMedium,
1108,
"last medium-band size"},
{k.mediumMax + 1,
k.mediumMax + 1,
((k.mediumMax + 1 - k.free) * k.hit) + k.bandLarge,
2009,
"first large-band size"},
{.requested = k.mediumMax,
.found = k.mediumMax,
.derived = ((k.mediumMax - k.free) * k.hit) + k.bandMedium,
.literal = 1108,
.why = "last medium-band size"},
{.requested = k.mediumMax + 1,
.found = k.mediumMax + 1,
.derived = ((k.mediumMax + 1 - k.free) * k.hit) + k.bandLarge,
.literal = 2009,
.why = "first large-band size"},
// Clamp: found > requested cannot make the miss count negative,
// so the fee is the same as the all-hit case (1).
{k.free + 1, k.free + 10, k.hit + k.bandSmall, 1, "found exceeds requested"},
{.requested = k.free + 1,
.found = k.free + 10,
.derived = k.hit + k.bandSmall,
.literal = 1,
.why = "found exceeds requested"},
};
}

View File

@@ -143,6 +143,21 @@ private:
void
done();
/**
* Count this acquisition as completed, at most once.
*
* done() is not the only place an acquisition finishes: init() can satisfy
* one entirely from the local store and return without ever reaching
* done(). Both call this, and the completionCounted_ latch makes the
* second call a no-op, so a completion is counted exactly once however it
* was reached.
*
* Does nothing unless the acquisition actually succeeded, so a failed or
* still-running acquisition is never counted.
*/
void
recordCompletionOnce();
void
onTimer(bool progress, ScopedLockType& peerSetLock) override;
@@ -181,6 +196,16 @@ private:
bool haveState_{false};
bool haveTransactions_{false};
bool signaled_{false};
/**
* Whether this acquisition has already been counted as completed.
*
* Separate from signaled_ because the two guard different things:
* signaled_ makes the done() state machine idempotent, while this makes
* the counter idempotent across the two independent exits that finish an
* acquisition (done() and init()'s local-hit path). Always accessed under
* mtx_.
*/
bool completionCounted_{false};
bool byHash_{true};
std::uint32_t seq_;
Reason const reason_;

View File

@@ -148,6 +148,14 @@ InboundLedger::init(ScopedLockType& collectionLock)
"xrpl::InboundLedger::init : valid ledger fees");
ledger_->setImmutable();
// The local store satisfied the whole acquisition, so this is a genuine
// completion. It is counted here rather than by calling done(), because
// done() drives the state machine (it stores the ledger, dispatches
// AcqDone, and would double-store on the paths below) and a counter must
// not change behaviour. recordCompletionOnce() is idempotent, so if
// done() is later reached for this same object the count stays at one.
recordCompletionOnce();
if (reason_ == Reason::HISTORY)
return;
@@ -449,6 +457,25 @@ InboundLedger::pmDowncast()
return shared_from_this();
}
void
InboundLedger::recordCompletionOnce()
{
// A failed or still-running acquisition is not a completion. Checked here
// rather than at each caller so both exits share one definition of
// success.
if (!complete_ || failed_)
return;
// The latch, not the counter, is what makes this idempotent: the two
// exits that finish an acquisition are independent, and either can run
// first.
if (completionCounted_)
return;
completionCounted_ = true;
app_.getAcquireStats().recordCompletion();
}
void
InboundLedger::done()
{
@@ -458,12 +485,12 @@ InboundLedger::done()
signaled_ = true;
touch();
// Counted here rather than at any single caller because done() is the one
// funnel every peer-driven outcome passes through, and the signaled_ guard
// above makes it run at most once per acquisition. failed_ outcomes are
// done() is the funnel every peer-driven outcome passes through, but not
// every outcome: init() can satisfy an acquisition from the local store
// and return without reaching here. Both call the same idempotent helper
// so each completion is counted exactly once. failed_ outcomes are
// excluded; the give-up path counts those itself.
if (complete_ && !failed_)
app_.getAcquireStats().recordCompletion();
recordCompletionOnce();
// Finalize the acquire span with the outcome, timeout count, and peer
// count. Keep it active as the ambient context across the finalize and

View File

@@ -143,7 +143,6 @@
#include <limits>
#include <memory>
#include <optional>
#include <ranges>
#include <string>
#include <string_view>