[Fix] (IntegrationTest): assert actual staging backpressure in the small ring lane

MobileGL/MG_IntegrationTest/Harness/SplitRuntimePeek.cpp

@@ -19,10 +19,21 @@
 
 #include <MG_Remote/Client/ClientSession.h>
 #include <MG_Remote/Client/EmitTables.h>
+#include <MG_Remote/Server/ServerLoop.h>
+#include <chrono>
+#include <thread>
 #define MGITEST_SPLIT_RUNTIME_PEEK_LIVE 1
 #endif
 
 namespace MGITest {
+    void DelaySplitRetirementForTesting(bool enabled) {
+#if defined(MGITEST_SPLIT_RUNTIME_PEEK_LIVE)
+        MobileGL::MG_Remote::Server::ServerLoopInstance().SetBeforeRetireHookForTesting(
+            enabled ? +[] { std::this_thread::sleep_for(std::chrono::milliseconds(30)); } : nullptr);
+#else
+        (void)enabled;
+#endif
+    }
 
     SplitRuntimeState PeekSplitRuntime() {
         SplitRuntimeState state;

MobileGL/MG_IntegrationTest/Harness/SplitRuntimePeek.h

@@ -91,9 +91,8 @@ namespace MGITest {
         // than asserted: a uniform record stride over a power-of-two ring lands on the
         // boundary exactly and never straddles it.
         //
-        // stageReclaimWaits: SEG_STAGE allocations that only fitted after the encoder reclaimed
-        // what the server had retired - P5's one real producer wait (see PipeWireCodec.h for
-        // why the command ring has none while the verb barrier is armed).
+        // stageReclaimWaits: allocations blocked after immediate reclamation failed;
+        // already-retired bytes reclaimed lazily do not count as a wait.
         unsigned long long maxRecordBytes = 0;
         unsigned long long maxRecordBytesCap = 0;
         unsigned long long cmdWraps = 0;
@@ -103,6 +102,8 @@ namespace MGITest {
     };
 
     SplitRuntimeState PeekSplitRuntime();
+    // Scheduling-only perturbation; never changes a watermark or counter.
+    void DelaySplitRetirementForTesting(bool enabled);
 
     // Empty when this process is a real split run that can be asserted about; otherwise the
     // reason to GTEST_SKIP() with, naming the first fact that is not true and the package that

MobileGL/MG_IntegrationTest/Harness/WireLedgerChecks.h

@@ -134,6 +134,9 @@ namespace MGITest::WireLedger {
                "kSmallRingLaneCmdByteTarget, which is sized for the 1 MiB ring this lane "
                "declares (MGL_ITEST_GLES_SPLIT_SMALL_RING_ENVIRONMENT); a larger ring needs a "
                "larger workload and is not what this lane is for";
+        EXPECT_GE(state.stageReclaimWaits, 1u)
+            << where << ": exit gate E3(e) - producer NEVER WAITED for staging retirement; "
+               "lazy reclamation of already-retired bytes is not back-pressure";
         ::testing::Test::RecordProperty("ring_wraps", static_cast<int>(state.cmdWraps));
         ::testing::Test::RecordProperty("ring_wrap_pads", static_cast<int>(state.cmdWrapPads));
         ::testing::Test::RecordProperty("ring_waits", static_cast<int>(state.stageReclaimWaits));

MobileGL/MG_IntegrationTest/Scenarios/TriangleScenario.cpp

@@ -264,6 +264,21 @@ void main() { oColor = vec4(vColor, 1.0); }
         // the lane is that IT is the arm with a ring the workload can fill.
         if (SplitLane::IsSmallRingLane()) {
             const unsigned long long driven = DriveUntilSmallRingOverruns();
+            // Ordinary uploads, each fitting by itself, jointly exceed the lane's 1 MiB
+            // staging segment. Delay only scheduling between applied and retired: the
+            // production allocator, not this test, must observe capacity and wait.
+            GLuint pressureBuffer = 0;
+            glGenBuffers(1, &pressureBuffer);
+            glBindBuffer(GL_COPY_WRITE_BUFFER, pressureBuffer);
+            std::vector<unsigned char> upload(768 * 1024, 0x5a);
+            glBufferData(GL_COPY_WRITE_BUFFER, upload.size(), nullptr, GL_DYNAMIC_DRAW);
+            DelaySplitRetirementForTesting(true);
+            glBufferSubData(GL_COPY_WRITE_BUFFER, 0, upload.size(), upload.data());
+            upload[0] = 0xa5;
+            glBufferSubData(GL_COPY_WRITE_BUFFER, 0, upload.size(), upload.data());
+            DelaySplitRetirementForTesting(false);
+            glBindBuffer(GL_COPY_WRITE_BUFFER, 0);
+            glDeleteBuffers(1, &pressureBuffer);
             Gl().EndFrame();
             WireLedger::ExpectSmallRingWrappedAtLeastOnce(
                 "TriangleScenario.TheSameVboAndVaoRedrawAcrossAFrameBoundary", driven);

MobileGL/MG_Remote/Client/ClientSession.cpp

@@ -516,6 +516,7 @@ namespace MobileGL::MG_Remote::Client {
         // a live RingProducer over a cursor triple nobody consumes would be the
         // half-wired shape this session exists not to have.
         m_encoder = Wire::PipeWireEncoder(control, &m_cmd, nullptr, &m_segments);
+        m_encoder.SetStageRetirementDoorbell(m_producer.SelfDoorbell());
 
         // ---- 7. the first CapsSnapshot, if the server had a backend to publish one from.
         // ONE DRAIN, ONE ADOPTER (c1): PumpControlPlane below is the only thing in the client

MobileGL/MG_Remote/Server/ServerLoop.cpp

@@ -530,6 +530,7 @@ namespace MobileGL::MG_Remote::Server {
             // a loop that applies and never retires ends the first MOBILEGL_IPC_STAGE_MB of
             // staging in Fatal{RingOverrun, "SEG_STAGE"} (w1-v1 5). Once per drain batch, not
             // once per record: retiring LATE is always legal, retiring EARLY never is.
+            if (const auto hook = m_beforeRetireHook.load(std::memory_order_acquire)) hook();
             consumer.RetireThrough(consumer.AppliedSeq());
             // LEAVING THE APPLIER (p1's M-5) IS *NOT* DONE HERE. It is done inside
             // PipeApplier::ApplyOne, before s1's SessionConsumer::ApplyOne publishes appliedSeq

MobileGL/MG_Remote/Server/ServerLoop.h

@@ -146,6 +146,11 @@ namespace MobileGL::MG_Remote::Server {
         // fail for its own reason.
         Uint64 DrainedRecords() const;
         Uint64 ParkCount() const;
+        // Scheduling perturbation only: the hook runs after application, before retirement.
+        // Integration tests use it to observe real producer back-pressure from GL uploads.
+        void SetBeforeRetireHookForTesting(void (*hook)()) {
+            m_beforeRetireHook.store(hook, std::memory_order_release);
+        }
 
         // C7 / ID-54 diagnostics, read by ServerLoopTest's C7 and N-3 controls. NativeBindCount is
         // how many times ApplyMakeCurrent FORWARDED a bind to the backend (a tuple it did not
@@ -230,6 +235,7 @@ namespace MobileGL::MG_Remote::Server {
         Uint64 m_affinityMask = 0;
         std::atomic<Uint64> m_drained{0};
         std::atomic<Uint64> m_parks{0};
+        std::atomic<void (*)()> m_beforeRetireHook{nullptr};
 
         // C7 / ID-54: the (dpy, draw, read, ctx) currently bound on the apply thread. Written and
         // read ONLY on the apply thread inside ApplyMakeCurrent, so it needs no lock; the two

MobileGL/MG_Remote/Wire/PipeWireCodec.cpp

@@ -776,10 +776,12 @@ namespace MobileGL::MG_Remote::Wire {
             std::abort();
         }
 
-        for (int attempt = 0; attempt < 2; ++attempt) {
+        bool reclaimed = false;
+        bool waited = false;
+        for (;;) {
             // THE WRAP SKIP MAY ONLY BE CHARGED AGAINST BYTES THAT ARE STILL IN FLIGHT. When
             // there are none the allocator starts over at offset zero, so a blob the segment
-            // can hold whole is never refused (see RebaseEmptyStage). On attempt 1 this runs
+            // can hold whole is never refused (see RebaseEmptyStage). On retry this runs
             // AFTER ReclaimStagedBytes, which is the case the finding describes: 8 MiB
             // allocated, then retired, then a 28 MiB request that used to abort.
             RebaseEmptyStage();
@@ -793,21 +795,28 @@ namespace MobileGL::MG_Remote::Wire {
                 m_stageHead += skip + need;
                 return m_stageBase + at;
             }
-            if (attempt == 0) {
-                // One try at reclaiming what the server has already retired. A second failure
-                // means the bytes genuinely do not fit, which R-10 says P5 does not chunk and
-                // must instead prove it never needs to.
-                //
-                // AND THIS IS P5'S ONE REAL BACK-PRESSURE EVENT, so it is counted here and
-                // published as `ringwaits=`. Reaching this line means the producer could not
-                // place a blob until the CONSUMER had retired earlier ones - the producer's
-                // progress depended on retiredSeq, which is exactly what R-9's "batching may
-                // only delay a watermark" is about. Exit gate E3(e)'s small-ring lane exists
-                // to make it happen at least once; a lane that never reaches it has a ring
-                // that is small only in its environment block.
-                ++m_stageReclaimWaits;
+            if (!reclaimed) {
+                // Lazy reclamation of already-retired bytes is NOT a producer wait.
                 ReclaimStagedBytes();
+                reclaimed = true;
+                continue;
             }
+            if (m_stageRetirementBell == nullptr || m_stageMarkFront == m_stageMarks.size()) break;
+            const Uint64 pending = m_stageMarks[m_stageMarkFront].Seq;
+            const auto ready = [&] {
+                return m_control->retiredSeq.load(std::memory_order_acquire) >= pending;
+            };
+            if (!ready()) {
+                // The allocation still cannot progress after reclamation. Count this
+                // blocked allocation once, not each watermark poll or each reclaimed mark.
+                if (!waited) { ++m_stageReclaimWaits; waited = true; }
+                if (!m_stageRetirementBell->Wait(m_control->producerParked, ready, 0, 5000)) {
+                    MGLOG_F("MGPipe: Fatal{RetirementWaitFailed, \"SEG_STAGE\"} producer wait "
+                            "ended before the pending allocation retired (shutdown or timeout)");
+                    std::abort();
+                }
+            }
+            ReclaimStagedBytes();
         }
         MGLOG_F("MGPipe: Fatal{RingOverrun, \"SEG_STAGE\"} a %llu byte blob does not fit a %llu "
                 "byte staging segment with %llu bytes still in flight (retiredSeq=%llu); P5 "

MobileGL/MG_Remote/Wire/PipeWireCodec.h

@@ -44,6 +44,7 @@
 #include <MG_Pipe/MGPipe.h>
 
 #include "../Transport/Ring.h"
+#include "../Transport/Doorbell.h"
 
 namespace MobileGL::MG_Remote::Wire {
 
@@ -311,17 +312,16 @@ namespace MobileGL::MG_Remote::Wire {
         // this number would have been red for the arithmetic of the record catalogue rather
         // than for anything about the ring.
         //
-        // `StageReclaimWaits()` counts every SEG_STAGE allocation that did not fit until the
-        // encoder reclaimed the runs the server had already retired - i.e. every time the
-        // producer's progress depended on the consumer's retiredSeq. That is the honest
-        // back-pressure reading in P5, and the reason the command ring has none: the verb
-        // barrier makes EmitAndWait wait for appliedSeq after EVERY record (R-1), so at most
-        // one record is ever in flight on SEG_CMD and a full command ring is not a wait but a
-        // Fatal{RingOverrun} (ClientSession.cpp). Publishing a "command ring waits" counter
-        // that can only ever be zero-or-dead is the decoration this file's counters are not.
+        // `StageReclaimWaits()` counts allocations blocked on an outstanding retiredSeq
+        // after immediate reclamation still left insufficient space. Reclaiming bytes
+        // the consumer had already retired does not increment it. One allocation counts
+        // once even if it waits for several marks; this is staging, not command-ring pressure.
         Uint64 CmdWraps() const;
         Uint64 CmdWrapPads() const;
         Uint64 StageReclaimWaits() const;
+        // The live session supplies its shutdown-aware producer doorbell. Standalone codecs
+        // without a consumer cannot wait for retirement and retain the named refusal.
+        void SetStageRetirementDoorbell(Transport::Doorbell* bell) { m_stageRetirementBell = bell; }
 
         // Bytes this encoder has ever written into SEG_CMD, pad fillers included: the
         // producer's monotonic head cursor. It is the DENOMINATOR the wrap count only means
@@ -365,6 +365,7 @@ namespace MobileGL::MG_Remote::Wire {
         Uint64 m_cmdWraps = 0;
         Uint64 m_cmdWrapPads = 0;
         Uint64 m_stageReclaimWaits = 0;
+        Transport::Doorbell* m_stageRetirementBell = nullptr;
         Vector<StageMark> m_stageMarks;
         SizeT m_stageMarkFront = 0;
         Uint8* m_stageBase = nullptr;

MobileGL/MG_Test/Wire/PipeWireCodecTest.cpp

@@ -1595,6 +1595,20 @@ TEST_F(PipeWireCodecTest, ABigProgramArchiveDoesNotGrowItsRecordAtAll) {
     EXPECT_GE(wire.Encoder().StagedBytesInFlight(), archive.size());
 }
 
+TEST_F(PipeWireCodecTest, AlreadyRetiredStagingReclamationIsNotAProducerWait) {
+    Wire2 wire;
+    std::vector<std::uint8_t> payload(Wire2::kStageBytes * 3 / 4, 0x5a);
+    wire.Encoder().StageBytes(payload.data(), payload.size());
+    MGPBindRenderState bind{};
+    ASSERT_NE(wire.Encoder().EncodeRecord(MGPWireOp::BindRenderState, &bind, sizeof(bind)), kInvalidSeq);
+    bool applied = false;
+    ASSERT_TRUE(wire.PumpOne(&applied));
+    // Do not explicitly reclaim: the second real allocation must do that itself.
+    wire.Encoder().StageBytes(payload.data(), payload.size());
+    EXPECT_EQ(wire.Encoder().StageReclaimWaits(), 0u)
+        << "already-retired lazy reclamation is not a producer wait";
+}
+
 TEST_F(PipeWireCodecTest, StagedBytesAreReclaimedOnlyBehindRetiredSeq) {
     Wire2 wire;
     const std::uint8_t payload[64] = {};