test(harness): isolate deferred callbacks and observe real job boundaries

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<int> order;
+        std::vector<SteppingController::time_point> 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<int>{0, 1, 2, 3}));
+        BEAST_EXPECT((times == std::vector<SteppingController::time_point>{
+            {}, 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<std::logic_error>([&] { c.setJobLag(0, jtADVANCE, "W", ms{-1}); }));
+    }
+
     void
     testCanonicalAllJobsPolicy()
     {
@@ -448,6 +489,7 @@ public:
     {
         testClassify();
         testExtensionWorkScheduled();
+        testNamedJobLagAndObservation();
         testJobHookClosedWorld();
         testCanonicalAllJobsPolicy();
         testCanonicalNestedJobNotSwallowed();

src/test/jtx/SteppingController.h

@@ -155,6 +155,8 @@ private:
     std::vector<std::string> failedJobs_;
     static constexpr std::size_t maxRecentJobs_ = 40;
     std::map<std::pair<std::uint32_t, int>, duration> jobLags_;
+    std::map<std::tuple<std::uint32_t, JobType, std::string>, duration> namedJobLags_;
+    std::function<void(std::uint32_t, JobType, std::string const&)> 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<int>(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<std::logic_error>("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<void(std::uint32_t, JobType, std::string const&)> 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,

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<ExtendedPosition, RCLCxPeerPos, RCLTxSet>;
 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_;