From ad4736cd39df35767fbf062cf50ca182368c14f6 Mon Sep 17 00:00:00 2001 From: Nicholas Dudfield Date: Fri, 18 Sep 2026 17:49:47 +0700 Subject: [PATCH] test(harness): isolate deferred callbacks and observe real job boundaries --- .../consensus/SteppingController_test.cpp | 42 +++++++++++++++++++ src/test/jtx/SteppingController.h | 37 +++++++++++++++- src/xrpld/app/consensus/ConsensusExtensions.h | 3 ++ 3 files changed, 80 insertions(+), 2 deletions(-) diff --git a/src/test/consensus/SteppingController_test.cpp b/src/test/consensus/SteppingController_test.cpp index efe8118c30..a5dd5ffac0 100644 --- a/src/test/consensus/SteppingController_test.cpp +++ b/src/test/consensus/SteppingController_test.cpp @@ -183,6 +183,47 @@ class SteppingController_test : public beast::unit_test::suite } } + void + testNamedJobLagAndObservation() + { + testcase("named lag delays only the selected job; observations precede real bodies"); + SteppingController c; + c.setSyncClock([](SteppingController::time_point) {}); + c.setJobLag(0, Tier::process, ms{2}); + c.setJobLag(0, jtADVANCE, "validatedLedgerWork", ms{7}); + auto hook0 = c.makeJobHook(0); + auto hook1 = c.makeJobHook(1); + std::vector order; + std::vector times; + bool observed = false; + c.observeJobs([&](std::uint32_t, JobType, std::string const&) { + BEAST_EXPECT(!observed); + observed = true; + }); + auto body = [&](int id) { + BEAST_EXPECT(observed); + observed = false; + order.push_back(id); + times.push_back(c.now()); + }; + using D = JobQueue::JobDisposition; + BEAST_EXPECT(hook0(jtADVANCE, "validatedLedgerWork", [&] { body(3); }) == D::claimedQueued); + BEAST_EXPECT(hook0(jtADVANCE, "getConsensusLedger2", [&] { body(1); }) == D::claimedQueued); + BEAST_EXPECT(hook0(jtPROPOSAL_t, "checkPropose", [&] { body(2); }) == D::claimedQueued); + BEAST_EXPECT(hook1(jtADVANCE, "validatedLedgerWork", [&] { body(0); }) == D::claimedQueued); + while (c.stepOne()) {} + BEAST_EXPECT((order == std::vector{0, 1, 2, 3})); + BEAST_EXPECT((times == std::vector{ + {}, SteppingController::time_point{ms{2}}, SteppingController::time_point{ms{2}}, + SteppingController::time_point{ms{7}}})); + c.clearJobLag(0); + auto const now = c.now(); + BEAST_EXPECT(hook0(jtADVANCE, "validatedLedgerWork", [&] { body(4); }) == D::claimedQueued); + BEAST_EXPECT(c.stepOne()); + BEAST_EXPECT(times.back() == now); + BEAST_EXPECT(except([&] { c.setJobLag(0, jtADVANCE, "W", ms{-1}); })); + } + void testCanonicalAllJobsPolicy() { @@ -448,6 +489,7 @@ public: { testClassify(); testExtensionWorkScheduled(); + testNamedJobLagAndObservation(); testJobHookClosedWorld(); testCanonicalAllJobsPolicy(); testCanonicalNestedJobNotSwallowed(); diff --git a/src/test/jtx/SteppingController.h b/src/test/jtx/SteppingController.h index 8fde8a82f7..088d50e1ea 100644 --- a/src/test/jtx/SteppingController.h +++ b/src/test/jtx/SteppingController.h @@ -155,6 +155,8 @@ private: std::vector failedJobs_; static constexpr std::size_t maxRecentJobs_ = 40; std::map, duration> jobLags_; + std::map, duration> namedJobLags_; + std::function beforeJob_; [[nodiscard]] static char const* jobTypeName(JobType t) @@ -401,8 +403,10 @@ private: } [[nodiscard]] duration - jobLag(std::uint32_t nodeId, Tier tier) const + jobLag(std::uint32_t nodeId, Tier tier, JobType type, std::string const& name) const { + if (auto const it = namedJobLags_.find({nodeId, type, name}); it != namedJobLags_.end()) + return it->second; auto const it = jobLags_.find({nodeId, static_cast(tier)}); return it == jobLags_.end() ? duration::zero() : it->second; } @@ -508,6 +512,28 @@ public: jobLags_[key] = lag; } + void + setJobLag(std::uint32_t nodeId, JobType type, std::string name, duration lag) + { + requireSteppingThread("setJobLag"); + if (lag < duration::zero()) + Throw("SteppingController::setJobLag: lag must be non-negative"); + auto const key = std::make_tuple(nodeId, type, std::move(name)); + if (lag == duration::zero()) + namedJobLags_.erase(key); + else + namedJobLags_[key] = lag; + } + + // Passive scenario inspection after clock sync, immediately before the real + // job body. This does not enqueue, suppress or replace that body. + void + observeJobs(std::function observer) + { + requireSteppingThread("observeJobs"); + beforeJob_ = std::move(observer); + } + void clearJobLag(std::uint32_t nodeId) { @@ -519,6 +545,11 @@ public: else ++it; } + for (auto it = namedJobLags_.begin(); it != namedJobLags_.end();) + if (std::get<0>(it->first) == nodeId) + it = namedJobLags_.erase(it); + else + ++it; } [[nodiscard]] std::size_t @@ -857,7 +888,7 @@ public: "' arrived off the stepping thread [thread='" + std::string(beast::getCurrentThreadName()) + "']"); } - auto const lag = jobLag(nodeId, classification.tier); + auto const lag = jobLag(nodeId, classification.tier, t, name); recordJob( nodeId, t, name, lag == duration::zero() ? "queued" : "queued:lagged"); if (lag != duration::zero()) @@ -872,6 +903,8 @@ public: if (lag != duration::zero()) clearLaggedPending(nodeId, tier, t, name, lag); recordJob(nodeId, t, name, "run"); + if (beforeJob_) + beforeJob_(nodeId, t, name); f(); }), HarnessScheduler::Kind::job, diff --git a/src/xrpld/app/consensus/ConsensusExtensions.h b/src/xrpld/app/consensus/ConsensusExtensions.h index 83e52e239b..b884812b0a 100644 --- a/src/xrpld/app/consensus/ConsensusExtensions.h +++ b/src/xrpld/app/consensus/ConsensusExtensions.h @@ -39,6 +39,7 @@ class STTx; namespace test { class ConsensusExtensions_test; +class SteppingExtensions_test; } // namespace test /// Concrete alias for the consensus tick context. @@ -52,6 +53,8 @@ using TickContext = ConsensusTick; class ConsensusExtensions { friend class test::ConsensusExtensions_test; + // DSF observes accepted evidence at real job boundaries; it does not seed it. + friend class test::SteppingExtensions_test; Application& app_; ExportSigCollector postValidationExportSigCollector_;