[Fix] (MG_Remote, Wire): stop the decoder writing RingControl - appliedSeq has exactly one writer and it is the session, so the decoder keeps its own tally and the two are compared instead

MobileGL/MG_Remote/Wire/PipeWireCodec.cpp

@@ -949,15 +949,13 @@ namespace MobileGL::MG_Remote::Wire {
 
         PoisonResolvedRuns();
 
+        // THE DECODER'S OWN TALLY, AND NOT THE SHARED WATERMARK. RingControl::appliedSeq has
+        // exactly one writer - s1's SessionConsumer::ApplyOne, +1 per record, pads never
+        // counted - and this class does not write RingControl at all. Keeping a private count
+        // beside it is what makes R-9's batching ban CHECKABLE rather than merely stated: the
+        // session's watermark and this number must agree after every record, and a test that
+        // compares them catches a batched publish that a single counter could not.
         ++m_applySeq;
-        // R-9: EVERY record, never batched. The client's verb barrier and every reply wait
-        // read appliedSeq, and a batched watermark makes a waiter resume on work the server
-        // has not done. retiredSeq goes with it because nothing in P5 borrows a ring slot into
-        // the GPU timeline - the day something does, this is the line that splits.
-        if (m_control != nullptr) {
-            m_control->appliedSeq.store(m_applySeq, std::memory_order_release);
-            m_control->retiredSeq.store(m_applySeq, std::memory_order_release);
-        }
         return applied;
     }
 

MobileGL/MG_Remote/Wire/PipeWireCodec.h

@@ -364,16 +364,19 @@ namespace MobileGL::MG_Remote::Wire {
         // here Fatals, because a pad that reached the decoder has already been counted.
         Bool DecodeAndApply(const Transport::RingRecordView& record);
 
-        // Advanced by exactly one per applied non-pad record. P5 FORBIDS BATCHING IT (R-9):
+        // THE DECODER'S OWN TALLY, NOT THE SHARED WATERMARK. Advanced by exactly one per
-        // the verb barrier's waiter reads it, and a batched watermark makes the client wait
+        // applied non-pad record.
-        // for records the server has not run.
         //
-        // DecodeAndApply PUBLISHES RingControl::appliedSeq AND retiredSeq to this value after
+        // RingControl::appliedSeq has exactly ONE writer - s1's SessionConsumer::ApplyOne, +1
-        // every record, because the decoder is the thing that knows when a record's SEG_STAGE
+        // per record, pads never counted - and this class writes NO RingControl field at all.
-        // runs stopped being read (R-11, table 1's "retires: apply"). v1's PipeApplier must
+        // That is deliberate rather than a division of labour: two writers of a watermark is
-        // therefore NOT advance either watermark a second time - a double advance makes the
+        // how a waiter resumes on a record the server has not run, which is what R-9's "never
-        // client's barrier resume on a record the server has not run, which is precisely the
+        // publish a watermark early" forbids, and there is no checksum on this ring that would
-        // failure R-9's "never publish a watermark early" exists to forbid.
+        // catch it.
+        //
+        // Keeping a private count beside the session's is what makes the batching ban
+        // CHECKABLE instead of merely stated: after every record the two numbers must agree,
+        // and a single counter could not tell a batched publish from an honest one.
         Uint64 AppliedSeq() const;
 
         // v1 installs the backend bridge for contract §7's five class-B verbs. Null - the

MobileGL/MG_Test/Wire/PipeWireCodecTest.cpp

@@ -111,9 +111,14 @@ namespace {
         Transport::RingConsumer& Consumer() { return m_consumer; }
         std::uint8_t* StageBase() { return m_stageBytes.data(); }
 
-        // Pops one record and decodes it. Returns whether the decoder reported "applied";
+        // Pops one record, decodes it, and then does what s1's SessionConsumer::ApplyOne does:
-        // `popped` says whether there was a record at all, so a case cannot pass because
+        // advance RingControl's watermarks by ONE. THE DECODER DOES NOT WRITE RingControl -
-        // nothing was there.
+        // appliedSeq has exactly one writer and it is the session - so this fixture has to
+        // play that role, which is also what lets a case compare the session's watermark
+        // against the decoder's own tally and catch a batched publish.
+        //
+        // Returns whether there was a record at all, so a case cannot pass because nothing was
+        // there; `applied` is what the decoder reported.
         bool PumpOne(bool* applied) {
             m_encoder.Publish();
             Transport::RingRecordView view{};
@@ -125,6 +130,11 @@ namespace {
                 return false;
             }
             const bool result = m_decoder.DecodeAndApply(view);
+            ++m_sessionApplied;
+            m_control.appliedSeq.store(m_sessionApplied, std::memory_order_release);
+            // Nothing in P5 borrows a ring slot into the GPU timeline, so a record's SEG_STAGE
+            // runs retire as soon as it is applied (table 1's "retires: apply").
+            m_control.retiredSeq.store(m_sessionApplied, std::memory_order_release);
             m_consumer.PublishRetired();
             if (applied != nullptr) {
                 *applied = result;
@@ -132,6 +142,8 @@ namespace {
             return true;
         }
 
+        std::uint64_t SessionAppliedSeq() const { return m_sessionApplied; }
+
         // ---- recorded answers ----
         struct Reply {
             std::uint64_t Seq = 0;
@@ -216,6 +228,7 @@ namespace {
         PipeWireDecoder m_decoder;
         Replies m_replies;
         Verbs m_verbs;
+        std::uint64_t m_sessionApplied = 0;
     };
 
     MGPipeHandle MakeHandle(Uint32 slot, Uint32 gen = 1) {
@@ -899,6 +912,10 @@ TEST_F(PipeWireCodecTest, AClassBVerbWithNoSinkDeclinesRatherThanInventsASemanti
 // =====================================================================================
 
 TEST_F(PipeWireCodecTest, AppliedSeqAdvancesByExactlyOnePerRecordAndIsNeverBatched) {
+    // TWO COUNTS, ON PURPOSE. RingControl::appliedSeq has exactly one writer - the session -
+    // and the decoder keeps its own tally. They must agree after every record, and a single
+    // counter could not tell a batched publish from an honest one (R-9: a batched watermark
+    // makes the client's barrier resume on work the server has not run).
     Wire2 wire;
     MGPPresent present{};
     for (Uint64 i = 1; i <= 5; ++i) {
@@ -911,12 +928,44 @@ TEST_F(PipeWireCodecTest, AppliedSeqAdvancesByExactlyOnePerRecordAndIsNeverBatch
         ASSERT_TRUE(wire.PumpOne(&applied));
         EXPECT_EQ(wire.Decoder().AppliedSeq(), i);
         EXPECT_EQ(wire.Control().appliedSeq.load(), i);
+        EXPECT_EQ(wire.SessionAppliedSeq(), wire.Decoder().AppliedSeq());
     }
     EXPECT_EQ(wire.Encoder().EmitSeq(), 5u);
 }
 
+TEST_F(PipeWireCodecTest, TheDecoderWritesNoRingControlFieldOfItsOwn) {
+    // s1 owns RingControl::appliedSeq; the codec's job ends at "this record was applied". Two
+    // writers of a watermark is how a waiter resumes on a record the server has not run, and
+    // there is no checksum on this ring that would catch it - so this is asserted rather than
+    // documented.
+    Wire2 wire;
+    MGPPresent present{};
+    ASSERT_NE(wire.Encoder().EncodeRecord(MGPWireOp::Present, &present, sizeof(present)),
+              kInvalidSeq);
+    wire.Encoder().Publish();
+
+    Transport::RingRecordView view{};
+    bool corrupt = false;
+    ASSERT_TRUE(wire.Consumer().Pop(view, &corrupt));
+    ASSERT_FALSE(corrupt);
+    const std::uint64_t appliedBefore = wire.Control().appliedSeq.load();
+    const std::uint64_t retiredBefore = wire.Control().retiredSeq.load();
+
+    (void)wire.Decoder().DecodeAndApply(view);
+
+    EXPECT_EQ(wire.Control().appliedSeq.load(), appliedBefore);
+    EXPECT_EQ(wire.Control().retiredSeq.load(), retiredBefore);
+    EXPECT_EQ(wire.Decoder().AppliedSeq(), 1u); // the decoder's own tally did move
+}
+
 TEST_F(PipeWireCodecTest, MaxRecordBytesSeenStaysFarBelowHalfTheRing) {
     // R-10's proof obligation. P5 does no chunking and must instead show it never needed any.
+    //
+    // THE CAP IS ASKED FOR AT RUNTIME AND NEVER DERIVED FROM MOBILEGL_IPC_RING_MB. s1 found
+    // that SEG_CMD's 8 MiB holds a 4096-byte control page plus a POWER-OF-TWO ring, so the
+    // ring is 4 MiB and MaxRecordBytes() is 2 MiB - half of what both CONTRACT-P5.md §5 and
+    // Config.h's comment say. Every comparison in the codec goes through
+    // RingProducer::MaxRecordBytes() for exactly this reason.
     Wire2 wire;
     MGPDrawInfo info{};
     info.NumDraws = 64;
@@ -934,6 +983,31 @@ TEST_F(PipeWireCodecTest, MaxRecordBytesSeenStaysFarBelowHalfTheRing) {
     EXPECT_LT(8u + sizeof(MGPFramebufferState), wire.Cmd().MaxRecordBytes());
 }
 
+TEST_F(PipeWireCodecTest, ABigProgramArchiveDoesNotGrowItsRecordAtAll) {
+    // THE POINT OF R-10's "every blob goes through SEG_STAGE": create_shader_state's RECORD is
+    // 8 + sizeof(MGPProgramDesc) == 200 bytes whether the archive is one kilobyte or one
+    // megabyte, because the record carries {Seg, Offset, Size} and nothing else. So
+    // create_shader_state is one of the SMALLEST records in the catalogue, not the one most
+    // likely to approach MaxRecordBytes(); what approaches that cap is a var-tail, and the
+    // archive's own bound is MOBILEGL_IPC_STAGE_MB with a Fatal of its own.
+    Wire2 wire;
+    MG_State::GLState::LinkArtifacts link;
+    MG_State::GLState::SpirvArtifacts spirv;
+    spirv.generatedSpirv.resize(1);
+    spirv.generatedSpirv[0].assign(8 * 1024, 0x07230203u); // 32 KiB of module words
+    Vector<Uint8> archive;
+    MG_State::GLState::EncodeProgramArtifacts(link, spirv, archive);
+    ASSERT_GT(archive.size(), 32u * 1024u);
+
+    MGPProgramDesc desc{};
+    desc.Cso = MakeHandle(91);
+    desc.Reflection = wire.Encoder().StageBytes(archive.data(), archive.size());
+    ASSERT_NE(wire.Encoder().EncodeRecord(MGPWireOp::CreateShaderState, &desc, sizeof(desc)),
+              kInvalidSeq);
+    EXPECT_EQ(wire.Encoder().MaxRecordBytesSeen(), 8u + sizeof(MGPProgramDesc));
+    EXPECT_GE(wire.Encoder().StagedBytesInFlight(), archive.size());
+}
+
 TEST_F(PipeWireCodecTest, StagedBytesAreReclaimedOnlyBehindRetiredSeq) {
     Wire2 wire;
     const std::uint8_t payload[64] = {};