test(harness): select whole signed frames without mutating payloads

src/test/jtx/SimTransport.h

@@ -31,6 +31,7 @@
 #include <memory>
 #include <mutex>
 #include <optional>
+#include <span>
 #include <stdexcept>
 #include <utility>
 #include <vector>
@@ -283,6 +284,10 @@ public:
     // from its (type, size); installed at a stepping boundary, empty = off.
     using WriteFault =
         std::function<SimFault(std::uint16_t type, std::size_t size)>;
+    // Optional content-aware selection of whole frames. Read-only and valid
+    // only during the callback: selectors cannot rewrite signed messages.
+    using Frame = std::span<std::uint8_t const>;
+    using FrameFault = std::function<SimFault(std::uint16_t type, Frame bytes)>;
     // Installed by simConnect in stepping mode: schedule a raw write of
     // `bytes` into THIS pipe `delay` from now (rides scheduleDelivery, so it
     // inherits the draining/inactive-node guards). Fault delays require it.
@@ -291,7 +296,7 @@ public:
 
 private:
     DeliveryRouter router_;
-    WriteFault writeFault_;
+    FrameFault writeFault_;
     DelayedWrite delayedWrite_;
     // (type, bytes remaining) of every framed message currently buffered, in
     // order — writeRaw pushes, drainInto consumes. Guarded by m_. Purely
@@ -415,6 +420,17 @@ public:
     // stepping boundary; single-threaded stepping makes live install race-free.
     void
     setWriteFault(WriteFault f)
+    {
+        if (!f)
+            writeFault_ = {};
+        else
+            writeFault_ = [f = std::move(f)](std::uint16_t type, Frame bytes) {
+                return f(type, bytes.size());
+            };
+    }
+
+    void
+    setFrameFault(FrameFault f)
     {
         writeFault_ = std::move(f);
     }
@@ -436,7 +452,7 @@ public:
         if (!writeFault_)
             return writeRaw(std::move(data));
 
-        auto const fault = writeFault_(peekMessageType(data), data.size());
+        auto const fault = writeFault_(peekMessageType(data), Frame{data});
         auto const copies = (fault.drop ? 0 : 1) + fault.duplicates;
         if (copies == 0)
         {
@@ -704,6 +720,12 @@ public:
         (aToB ? a2b_ : b2a_)->setWriteFault(std::move(f));
     }
 
+    void
+    setFrameFault(bool aToB, SimPipe::FrameFault f)
+    {
+        (aToB ? a2b_ : b2a_)->setFrameFault(std::move(f));
+    }
+
     /** Inject one already-framed protocol message from endpoint A toward B. */
     void
     injectAToB(std::vector<std::uint8_t> bytes)

src/test/jtx/SteppingNetwork.h

@@ -785,6 +785,21 @@ public:
         throw std::logic_error("SteppingNetwork::faultLink: no live wire between nodes");
     }
 
+    // Content-aware whole-frame faults, e.g. lose only share-bearing proposals.
+    SteppingNetwork&
+    faultFrames(std::uint32_t from, std::uint32_t to, SimPipe::FrameFault injector)
+    {
+        requireSlot(from, "faultFrames");
+        requireSlot(to, "faultFrames");
+        for (auto it = links_.rbegin(); it != links_.rend(); ++it)
+            if (it->connects(from, to) && it->wire && !it->wire->severed())
+            {
+                it->wire->setFrameFault(it->a == from, std::move(injector));
+                return *this;
+            }
+        throw std::logic_error("SteppingNetwork::faultFrames: no live wire between nodes");
+    }
+
     SteppingNetwork&
     linkDelays(latency::Profile profile)
     {

src/test/overlay/SimTransport_test.cpp

@@ -131,6 +131,43 @@ class SimTransport_test : public beast::unit_test::suite
         BEAST_EXPECT(wire.bufferedBytes() == 0);
     }
 
+    void
+    testReadOnlyFrameSelection()
+    {
+        testcase("content faults select whole frames and clearing restores exact bytes");
+        SimPipe pipe;
+        Bytes const kept{0, 0, 0, 1, 0, 7, 'K'};
+        Bytes const lost{0, 0, 0, 1, 0, 7, 'L'};
+        unsigned inspected = 0;
+        pipe.setFrameFault([&](std::uint16_t type, SimPipe::Frame frame) {
+            ++inspected;
+            BEAST_EXPECT(type == 7);
+            BEAST_EXPECT(frame.size() == kept.size());
+            return SimFault{frame.back() == 'L'};
+        });
+        BEAST_EXPECT(pipe.write(lost));
+        BEAST_EXPECT(pipe.bufferedBytes() == 0);
+        BEAST_EXPECT(pipe.write(kept));
+        BEAST_EXPECT(pipe.bufferedBytes() == kept.size());
+        pipe.setWriteFault({});
+        BEAST_EXPECT(pipe.write(lost));
+        BEAST_EXPECT(inspected == 2);
+
+        boost::asio::io_context io;
+        Bytes actual(kept.size() + lost.size());
+        bool read = false;
+        pipe.read({boost::asio::buffer(actual)}, io.get_executor(),
+                  [&](Transport::error_code ec, std::size_t n) {
+                      BEAST_EXPECT(!ec && n == actual.size());
+                      read = true;
+                  });
+        io.run();
+        BEAST_EXPECT(read);
+        Bytes expected = kept;
+        expected.insert(expected.end(), lost.begin(), lost.end());
+        BEAST_EXPECT(actual == expected);
+    }
+
     void
     testBufferedBytesDrainBeforeEof()
     {
@@ -174,6 +211,7 @@ public:
         testDelayedWriteCannotCrossClose();
         testEndpointWriteAfterSever();
         testFaultDropIsSuccessfulWrite();
+        testReadOnlyFrameSelection();
         testBufferedBytesDrainBeforeEof();
     }
 };