Overlay V2: Update e2e infrastructure

src/ledger/LedgerManager.h

@@ -302,11 +302,11 @@ class LedgerManager
 
     // Records the submission time of a self-submitted transaction for the
     // tx-latency metrics. No-op unless
-    // Config::LOADGEN_MEASURE_TX_LATENCY_FOR_TESTING is set.
+    // Config::LOADGEN_MEASURE_TX_E2E_LATENCY_FOR_TESTING is set.
     virtual void recordTxSubmission(Hash const& contentsHash) = 0;
 
     // Begins/ends a load-generation latency measurement window. No-op unless
-    // Config::LOADGEN_MEASURE_TX_LATENCY_FOR_TESTING is set.
+    // Config::LOADGEN_MEASURE_TX_E2E_LATENCY_FOR_TESTING is set.
     virtual void beginTxLatencyMeasurement(uint32_t expectedTxCount) = 0;
     virtual void finalizeTxLatencyMeasurement() = 0;
 #endif

src/ledger/LedgerManagerImpl.cpp

@@ -1362,22 +1362,23 @@ LedgerManagerImpl::recordTxSubmission(Hash const& contentsHash)
 }
 
 void
-LedgerManagerImpl::recordTxMetaEmissionLatency(LedgerCloseMeta const& lcm)
+LedgerManagerImpl::recordTxE2eLatency(ApplicableTxSetFrame const& txSet)
 {
     if (!mApp.getConfig().LOADGEN_MEASURE_TX_E2E_LATENCY_FOR_TESTING)
     {
         return;
     }
-    VirtualClock::time_point const emitTime = mApp.getClock().now();
+    VirtualClock::time_point const applyEndTime = mApp.getClock().now();
     MutexLocker guard(mTxLatencyMetrics.mMutex);
-    auto probe = [&](auto const& txProcessing) {
-        for (auto const& txp : txProcessing)
+    for (auto const& phase : txSet.getPhases())
+    {
+        for (auto const& tx : phase)
         {
             if (auto submitted = mTxLatencyMetrics.mTxSubmitTimes.find(
-                    txp.result.transactionHash);
+                    tx->getContentsHash());
                 submitted != mTxLatencyMetrics.mTxSubmitTimes.end())
             {
-                auto const latency = emitTime - submitted->second;
+                auto const latency = applyEndTime - submitted->second;
                 mTxLatencyMetrics.mTxsExternalized.inc();
                 int64_t const ms =
                     std::chrono::duration_cast<std::chrono::milliseconds>(
@@ -1388,20 +1389,6 @@ LedgerManagerImpl::recordTxMetaEmissionLatency(LedgerCloseMeta const& lcm)
                 mTxLatencyMetrics.mTxSubmitTimes.erase(submitted);
             }
         }
-    };
-    switch (lcm.v())
-    {
-    case 0:
-        probe(lcm.v0().txProcessing);
-        break;
-    case 1:
-        probe(lcm.v1().txProcessing);
-        break;
-    case 2:
-        probe(lcm.v2().txProcessing);
-        break;
-    default:
-        releaseAssert(false);
     }
 }
 
@@ -1451,7 +1438,8 @@ LedgerManagerImpl::finalizeTxLatencyMeasurement()
     std::sort(samples.begin(), samples.end());
     size_t const n = samples.size();
 
-    auto percentile = [&](size_t p) -> uint32_t {
+    auto percentile = [&](size_t p)
+                          REQUIRES(mTxLatencyMetrics.mMutex) -> uint32_t {
         // Calculate ceiling of rank
         size_t rank = (p * n + 99) / 100;
         rank = std::clamp<size_t>(rank, 1, n);
@@ -1939,7 +1927,6 @@ LedgerManagerImpl::applyLedger(LedgerCloseData const& ledgerData,
             appliedLedgerState->getLastClosedLedgerHeader();
         // Copy this before we move it into mNextMetaToEmit below
         mLastLedgerCloseMeta = *ledgerCloseMeta;
-        recordTxMetaEmissionLatency(ledgerCloseMeta->getXDR());
     }
 #endif
 
@@ -2041,6 +2028,12 @@ LedgerManagerImpl::applyLedger(LedgerCloseData const& ledgerData,
     // to be from the same ledger.
     maybeRunSnapshotInvariantFromLedgerState(mApplyState.copyApplyLedgerView());
 
+#ifdef BUILD_TESTS
+    maybeSimulateSleep(mApp.getConfig(), txSet->sizeOpTotalForLogging(),
+                       applyLedgerTime, mApplySleepRng);
+    recordTxE2eLatency(*applicableTxSet);
+#endif
+
     // Steps 6, 7, 8 are done in `advanceLedgerStateAndPublish`
     // NB: appliedLedgerState is invalidated after this call.
     if (threadIsMain())
@@ -2061,10 +2054,6 @@ LedgerManagerImpl::applyLedger(LedgerCloseData const& ledgerData,
         mApp.postOnMainThread(std::move(cb), "advanceLedgerStateAndPublish");
     }
 
-#ifdef BUILD_TESTS
-    maybeSimulateSleep(mApp.getConfig(), txSet->sizeOpTotalForLogging(),
-                       applyLedgerTime, mApplySleepRng);
-#endif
     std::chrono::duration<double> ledgerTimeSeconds = ledgerTime.Stop();
     CLOG_DEBUG(Perf, "Applied ledger {} in {} seconds", ledgerSeq,
                ledgerTimeSeconds.count());

src/ledger/LedgerManagerImpl.h

@@ -451,15 +451,15 @@ class LedgerManagerImpl : public LedgerManager
         ANNOTATED_MUTEX(mMutex);
         UnorderedMap<Hash, VirtualClock::time_point>
             mTxSubmitTimes GUARDED_BY(mMutex);
-        // Each recorded submission -> meta-emission latency in ms
+        // Each recorded submission -> post-apply latency in ms
         std::vector<uint32_t> mSamples GUARDED_BY(mMutex);
 
         TxLatencyMetrics(MetricsRegistry& registry);
     } mTxLatencyMetrics;
 
-    // End point of the tx-latency metric: matches the ledger-close meta's
-    // txProcessing entries against mTxSubmitTimes and records each latency.
-    void recordTxMetaEmissionLatency(LedgerCloseMeta const& lcm);
+    // End point of the tx-latency metric: records the submission to post-apply
+    // latency for each externalized transaction.
+    void recordTxE2eLatency(ApplicableTxSetFrame const& txSet);
 #endif
 
     void setState(State s);