test(threaded): wait for sim transport quiescence before post-run activity asserts

src/test/consensus/ThreadedConsensus_test.cpp

@@ -246,6 +246,7 @@ class ThreadedConsensus_test : public beast::unit_test::suite
             return std::nullopt;
         }
 
+        BEAST_EXPECT(net.waitForSimQuiescence(options.stallTimeout));
         auto const endActivity = net.simActivitySnapshot();
         out.readStarted = endActivity.readStarted - startActivity.readStarted;
         out.writeStarted =

src/test/consensus/ThreadedTraffic_test.cpp

@@ -922,6 +922,7 @@ class ThreadedTraffic_test : public beast::unit_test::suite
                 return std::nullopt;
             }
 
+            BEAST_EXPECT(net.waitForSimQuiescence(tickOptions.stallTimeout));
             auto const activity = net.simActivitySnapshot();
             out.maxBufferedBytes = std::max<std::size_t>(
                 out.maxBufferedBytes,

src/test/jtx/MultiNode.h

@@ -1054,6 +1054,25 @@ public:
                             : SimTransportActivitySnapshot{};
     }
 
+    // Wait until no sim transport post is in flight and no wire holds
+    // buffered bytes. threadedTick can return while a slow post is still
+    // running on a node's io thread, so a caller that asserts quiescence
+    // after its last tick waits here first. Returns false on timeout.
+    [[nodiscard]] bool
+    waitForSimQuiescence(std::chrono::milliseconds timeout) const
+    {
+        auto const deadline = std::chrono::steady_clock::now() + timeout;
+        for (;;)
+        {
+            if (simBufferedBytes() == 0 &&
+                simActivitySnapshot().inFlightPosts == 0)
+                return true;
+            if (std::chrono::steady_clock::now() >= deadline)
+                return false;
+            std::this_thread::sleep_for(std::chrono::milliseconds{1});
+        }
+    }
+
     // Wait until every node has at least `expected` active (post-handshake)
     // peers.
     bool