[Fix] (MG_Remote, Transport): give the inproc doorbell a death state so Shutdown can join a parked waiter, and bound a ring record at half the capacity so a refusal can never look like backpressure

- T-1: InProcessChannel::Close rang each CondVarDoorbell once and claimed that unparks a peer mid-frame. It does not. Doorbell::Wait consumes the one ring, re-tests a condition nothing published, finds the bell alive (CondVarDoorbell never overrode Dead(); Doorbell.cpp had no death state at all) and with kWaitForever parks again for good - so Shutdown could never join a server thread sitting in the design's own steady state (spun, set consumerParked, blocked; plan section 8.1 inheriting the earlier plan's 6.2a). CondVarDoorbell now carries an atomic death latch: Kill() sets it under the mutex and notify_all's, Dead() reports it, Park returns false at once on a dead bell (and the wait predicate includes it, so a Kill cannot slip between the test and the wait), and Close kills both bells instead of ringing them. Same shape as SocketDoorbell's EOF latch; Notify stays the ordinary wakeup.
- T-2: RingProducer::Reserve refused only total > capacity, but a record with capacity/2 < total <= capacity is unplaceable at every head offset where neither the space to the wrap boundary nor the space before it holds it - even in an EMPTY ring, because a wrap pad costs spaceToEnd bytes on top of the record. Concretely: head offset 16 of an empty 256-byte ring, a 248-byte record; FreeBytes() says 256, Reserve says nullptr, forever, and a producer waiting for FreeBytes() >= 248 stalls with nothing logged. The bound is now capacity/2, which is exact rather than conservative (worst case 2*total-8 <= capacity-8), exposed as MaxRecordBytes() for the emitter to chunk against; the minimum ring is two headers so the smallest record still fits the bound. Ring.h states Capacity()/2 as the chunking bound and the G3 header comment in gen_pipe.py now states the chunking rule plan section 8.2 asks G3 to define (PipeWire.inc regenerated).
- Tests, each shown red with only the fix site reverted and green with it: InProcessTransportTest.ShutdownUnparksAWaiterWithNoDeadline (bounded join through a shared_ptr-owned waiter: 5 s red instead of a hung job; reverted it hangs and fails at 5051 ms), RingTest.RecordLargerThanHalfTheRingIsRefused (reverted, the 248-byte record is accepted), RingTest.RecordPlaceabilityDoesNotDependOnTheHeadOffset (the offset-0 vs offset-16 negative control), RingTest.HalfCapacityRecordFitsAtEveryHeadOffset (the positive half: the maximal record at all 32 head offsets of a 256-byte ring) and RingTest.RejectsARingTooSmallForTheSmallestRecord.

MobileGL/MG_Pipe/generated/PipeWire.inc

@@ -21,6 +21,16 @@
 // a record that is shorter than its own type, longer than what is left in the buffer, or
 // not a multiple of 8 is protocol corruption and is fatal. There is no recovery path -
 // silently applying a truncated record is how a corrupt stream becomes a wrong picture.
+//
+// OVERSIZED PAYLOADS ARE CHUNKED, NEVER EMITTED WHOLE (plan section 8.2: G3 has to define
+// the path for a record larger than the segment). The bound is the ring's,
+// RingProducer::MaxRecordBytes() == Capacity()/2, and it is exact rather than
+// conservative: a record has to be placeable at every head offset of an empty ring, the
+// wrap pad in front of it costs up to total-8 bytes, and only a record of at most half the
+// ring survives that at every offset. An emitter holding more than Capacity()/2 bytes of
+// record (a large resource_subdata, a create_shader_state archive) splits it into several
+// records of at most that size; the transport refuses a bigger one outright - nullptr plus
+// an MGLOG_E - rather than let the producer wait on free bytes that can never suffice.
 
 struct MGPWireRecHeader {
     Uint16 Op;    // MGPWireOp

MobileGL/MG_Remote/Transport/Doorbell.cpp

@@ -48,6 +48,12 @@ namespace MobileGL::MG_Remote::Transport {
 
     bool CondVarDoorbell::Park(std::uint32_t timeoutMs) {
         std::unique_lock<std::mutex> lock(m_impl->mutex);
+        // The death latch is tested under the same mutex Kill sets it under, so
+        // a Kill cannot slip between this test and the wait below: it either
+        // returns here or wakes the predicate.
+        if (m_dead.load(std::memory_order_relaxed)) {
+            return false;
+        }
         if (m_impl->signals != 0) {
             --m_impl->signals;
             return true;
@@ -55,16 +61,33 @@ namespace MobileGL::MG_Remote::Transport {
         if (timeoutMs == 0) {
             return false;
         }
+        const auto woken = [this] {
+            return m_impl->signals != 0 || m_dead.load(std::memory_order_relaxed);
+        };
         if (timeoutMs == kWaitForever) {
-            m_impl->cv.wait(lock, [this] { return m_impl->signals != 0; });
+            m_impl->cv.wait(lock, woken);
-        } else if (!m_impl->cv.wait_for(lock, std::chrono::milliseconds(timeoutMs),
+        } else if (!m_impl->cv.wait_for(lock, std::chrono::milliseconds(timeoutMs), woken)) {
-                                        [this] { return m_impl->signals != 0; })) {
+            return false;
+        }
+        if (m_dead.load(std::memory_order_relaxed)) {
+            // Woken by Kill, not by an event. The caller re-tests its condition
+            // regardless (Doorbell::Wait always does) and then sees Dead().
             return false;
         }
         --m_impl->signals;
         return true;
     }
 
+    void CondVarDoorbell::Kill() {
+        {
+            std::lock_guard<std::mutex> lock(m_impl->mutex);
+            m_dead.store(true, std::memory_order_release);
+        }
+        // notify_all, not notify_one: both a raw Park and a Doorbell::Wait may
+        // be parked here, and after this nobody will ring again.
+        m_impl->cv.notify_all();
+    }
+
     void CondVarDoorbell::Reset() {
         std::lock_guard<std::mutex> lock(m_impl->mutex);
         m_impl->signals = 0;

MobileGL/MG_Remote/Transport/Doorbell.h

@@ -104,12 +104,15 @@ namespace MobileGL::MG_Remote::Transport {
         // does not make the next Park return spuriously forever.
         virtual void Reset() = 0;
 
-        // True once the wakeup channel is permanently unusable, e.g. the peer
+        // True once the wakeup channel is permanently unusable: the peer closed
-        // closed its end of the socket. A dead doorbell can never deliver
+        // its end of the socket, or the inproc channel was shut down. A dead
-        // another wakeup AND its descriptor is permanently poll-ready, so Wait
+        // doorbell can never deliver another wakeup, and Wait must stop
-        // must stop re-parking on it: otherwise a waiter with no deadline
+        // re-parking on it - for the socket because its descriptor is
-        // burns a big core at full clock, which is the exact pathology the
+        // permanently poll-ready and a waiter with no deadline would burn a
-        // bidirectional doorbell exists to prevent.
+        // big core at full clock, for the condvar because Park would otherwise
+        // block forever and Shutdown could never join the waiter. Every
+        // implementation has a death state; the base default is only for a
+        // bell that cannot die.
         virtual bool Dead() const { return false; }
 
         // Spin `spinUs`, then park until `ready()` or the deadline.
@@ -204,10 +207,23 @@ namespace MobileGL::MG_Remote::Transport {
         void Notify() override;
         bool Park(std::uint32_t timeoutMs) override;
         void Reset() override;
+        bool Dead() const override { return m_dead.load(std::memory_order_acquire); }
+
+        // Hangs the bell up for good: every parked waiter returns false now and
+        // every later Park returns false at once. The inproc twin of the socket
+        // peer closing its end (SocketDoorbell latches m_dead on EOF), and what
+        // InProcessChannel::Close rings instead of Notify. A Notify is consumed
+        // by ONE Park; Doorbell::Wait then re-tests its condition, finds
+        // nothing published, finds the bell alive, and with kWaitForever parks
+        // again - so a Shutdown that only rang could never join a server thread
+        // sitting in the design's own steady state (spun, set consumerParked,
+        // blocked). Irreversible by design, like the socket's.
+        void Kill();
 
     private:
         struct Impl;
         Impl* m_impl;
+        std::atomic<bool> m_dead{false};
     };
 
 #if !defined(_WIN32)

MobileGL/MG_Remote/Transport/InProcessTransport.cpp

@@ -81,9 +81,13 @@ namespace MobileGL::MG_Remote::Transport {
                 dir.fdCv.notify_all();
             }
             // Anything parked on a ring doorbell has to come back too, or a
-            // shutdown mid-frame hangs the peer forever.
+            // shutdown mid-frame hangs the peer forever. Kill, not Notify: a
+            // ring is consumed by one Park, after which Doorbell::Wait re-tests
+            // a condition nothing published and - the bell still reporting
+            // alive - parks again, with no deadline forever. Only Dead() ends
+            // that loop.
             for (CondVarDoorbell& bell : m_bells) {
-                bell.Notify();
+                bell.Kill();
             }
         }
 
@@ -277,8 +281,9 @@ namespace MobileGL::MG_Remote::Transport {
     }
 
     // Whole-connection teardown, as ITransport::Shutdown documents: both
-    // directions are half-closed and both ring doorbells are rung, because a
+    // directions are half-closed and both ring doorbells are KILLED, because a
-    // peer parked on a ring doorbell mid-frame would otherwise never come back.
+    // peer parked on a ring doorbell mid-frame would otherwise never come back
+    // (a mere ring is consumed once and the waiter parks again).
     void InProcessTransport::Shutdown() { m_channel->Close(); }
 
     Doorbell& InProcessTransport::PeerDoorbell() { return m_channel->Bell(1 - m_endpoint); }

MobileGL/MG_Remote/Transport/Ring.cpp

@@ -87,11 +87,13 @@ namespace MobileGL::MG_Remote::Transport {
         : m_control(control), m_base(static_cast<std::uint8_t*>(base)), m_capacity(capacityBytes),
           m_mask(capacityBytes - 1), m_cursors(cursors) {
         if (control == nullptr || base == nullptr || !IsPowerOfTwo(capacityBytes) ||
-            capacityBytes < sizeof(RingRecordHeader) || capacityBytes > kMaxRingCapacity) {
+            capacityBytes < kMinRingCapacity || capacityBytes > kMaxRingCapacity) {
             MGLOG_E("MG_Remote ring: producer rejected, capacity %llu must be a power of two "
-                    "between %zu and %llu bytes over a non-null mapping (the record header's size "
+                    "between %llu and %llu bytes over a non-null mapping (a record may be at most "
-                    "field is 32-bit, so a bigger ring would truncate it)",
+                    "half the ring, and the record header's size field is 32-bit, so a bigger ring "
-                    static_cast<unsigned long long>(capacityBytes), sizeof(RingRecordHeader),
+                    "would truncate it)",
+                    static_cast<unsigned long long>(capacityBytes),
+                    static_cast<unsigned long long>(kMinRingCapacity),
                     static_cast<unsigned long long>(kMaxRingCapacity));
             m_control = nullptr;
             m_base = nullptr;
@@ -123,11 +125,23 @@ namespace MobileGL::MG_Remote::Transport {
             return nullptr;
         }
         const std::uint64_t total = Align8(sizeof(RingRecordHeader) + payloadBytes);
-        if (total > m_capacity) {
+        if (total > MaxRecordBytes()) {
-            // A single record larger than the whole ring is a caller bug: the
+            // A single record larger than HALF the ring is a caller bug: the
             // record catalogue has to chunk oversized payloads (large subdata
             // becomes several records) rather than emit one giant record.
-            MGLOG_E("MG_Remote ring: record kind %u of %llu bytes does not fit a %llu byte ring; "
+            //
+            // Half, not the whole ring, because a record has to be placeable at
+            // EVERY head offset of an empty ring. Straddling the wrap boundary
+            // costs a pad of spaceToEnd bytes on top of the record, and with
+            // spaceToEnd < total that is at most 2*total-8, which stays within
+            // the capacity exactly up to capacity/2. Above it the record is
+            // placeable at some offsets and not at others: at head offset 16 of
+            // an empty 256-byte ring a 248-byte record needs 240+248 bytes while
+            // FreeBytes() reports 256, so a producer that waits for FreeBytes()
+            // >= total stalls forever, and nothing is ever logged. Refusing here
+            // makes that impossible - a nullptr with FreeBytes() >= total can no
+            // longer mean "wait".
+            MGLOG_E("MG_Remote ring: record kind %u of %llu bytes exceeds half of a %llu byte ring; "
                     "the emitter must chunk it",
                     static_cast<unsigned>(kind), static_cast<unsigned long long>(total),
                     static_cast<unsigned long long>(m_capacity));
@@ -184,11 +198,13 @@ namespace MobileGL::MG_Remote::Transport {
         : m_control(control), m_base(static_cast<const std::uint8_t*>(base)),
           m_capacity(capacityBytes), m_mask(capacityBytes - 1), m_cursors(cursors) {
         if (control == nullptr || base == nullptr || !IsPowerOfTwo(capacityBytes) ||
-            capacityBytes < sizeof(RingRecordHeader) || capacityBytes > kMaxRingCapacity) {
+            capacityBytes < kMinRingCapacity || capacityBytes > kMaxRingCapacity) {
             MGLOG_E("MG_Remote ring: consumer rejected, capacity %llu must be a power of two "
-                    "between %zu and %llu bytes over a non-null mapping (the record header's size "
+                    "between %llu and %llu bytes over a non-null mapping (a record may be at most "
-                    "field is 32-bit, so a bigger ring would truncate it)",
+                    "half the ring, and the record header's size field is 32-bit, so a bigger ring "
-                    static_cast<unsigned long long>(capacityBytes), sizeof(RingRecordHeader),
+                    "would truncate it)",
+                    static_cast<unsigned long long>(capacityBytes),
+                    static_cast<unsigned long long>(kMinRingCapacity),
                     static_cast<unsigned long long>(kMaxRingCapacity));
             m_control = nullptr;
             m_base = nullptr;

MobileGL/MG_Remote/Transport/Ring.h

@@ -120,6 +120,11 @@ namespace MobileGL::MG_Remote::Transport {
     // class of construction-time guard as the power-of-two check beside it.
     inline constexpr std::uint64_t kMaxRingCapacity = 0xFFFFFFFFull;
 
+    // Smallest ring: two record headers. A record may be at most HALF the ring
+    // (see RingProducer::Reserve), so a ring of one header could carry nothing
+    // at all - not even the smallest record, a bare header.
+    inline constexpr std::uint64_t kMinRingCapacity = 2 * sizeof(RingRecordHeader);
+
     // Which cursor triple a producer/consumer pair drives.
     enum class RingCursorSet : std::uint32_t {
         Cmd = 0,
@@ -156,8 +161,8 @@ namespace MobileGL::MG_Remote::Transport {
     public:
         RingProducer() = default;
         // `base` is the ring's byte area (NOT the control page) and
-        // `capacityBytes` must be a power of two of at least one record header
+        // `capacityBytes` must be a power of two between kMinRingCapacity and
-        // and at most kMaxRingCapacity. Anything else leaves Valid() false.
+        // kMaxRingCapacity. Anything else leaves Valid() false.
         RingProducer(RingControl* control, void* base, std::uint64_t capacityBytes,
                      RingCursorSet cursors);
 
@@ -167,12 +172,28 @@ namespace MobileGL::MG_Remote::Transport {
         std::uint64_t FreeBytes() const;
 
         // Reserves room for one record and returns a pointer to its payload,
-        // or nullptr when the ring is full (or the record cannot fit at all).
+        // or nullptr when the ring is full. The payload is uninitialized;
-        // The payload is uninitialized; alignment padding at its tail is NOT
+        // alignment padding at its tail is NOT zeroed. Emits a pad record
-        // zeroed. Emits a pad record automatically when the record would
+        // automatically when the record would straddle the wrap boundary, so
-        // straddle the wrap boundary, so every record is contiguous.
+        // every record is contiguous.
+        //
+        // A record whose total (header + payload, rounded up to 8) exceeds
+        // MaxRecordBytes() == Capacity()/2 is refused outright, with an error
+        // log and however empty the ring is: chunking it is the emitter's job
+        // (plan section 8.2, the G3 chunking rule). Half is exact, not
+        // conservative - it is the largest record EVERY head offset can place,
+        // because a wrap pad costs at most total-8 bytes on top of the record
+        // and 2*total-8 <= capacity-8 holds exactly up to capacity/2. Above it
+        // a record is placeable at some offsets and not at others, and a
+        // producer waiting for FreeBytes() >= total stalls forever on an empty
+        // ring. So: nullptr with FreeBytes() >= total never means "wait"; it
+        // can only mean "too big, chunk".
         void* Reserve(std::uint16_t kind, std::uint16_t flags, std::uint64_t payloadBytes);
 
+        // The largest header+payload total Reserve accepts: Capacity()/2. This
+        // is the number the emitter chunks against.
+        std::uint64_t MaxRecordBytes() const { return m_capacity / 2; }
+
         // Makes every reserved record visible to the consumer (release store on
         // the head cursor). Cheap: publishing per record is fine, batching 8-16
         // only amortizes the doorbell store.

MobileGL/MG_Test/Wire/InProcessTransportTest.cpp

@@ -18,6 +18,7 @@
 #include <atomic>
 #include <chrono>
 #include <cstring>
+#include <memory>
 #include <string>
 #include <thread>
 #include <vector>
@@ -330,3 +331,68 @@ TEST(InProcessTransportTest, DoorbellTimesOutWhenNothingHappens) {
               20);
     EXPECT_EQ(parked.load(), 0u);
 }
+
+// The design's own steady state: the consumer spun, set consumerParked and blocked
+// with NO deadline. Shutdown has to bring that thread back, and a single Notify
+// cannot - Doorbell::Wait consumes it, re-tests a condition that is still false,
+// and with kWaitForever parks again. Only a bell that reports Dead() ends the
+// loop, which is what InProcessChannel::Close rings now.
+//
+// A regression here is a HANG, so the join is bounded: the waiter owns its state
+// through a shared_ptr and is detached on timeout, and the test fails red after
+// five seconds instead of wedging the CI job.
+TEST(InProcessTransportTest, ShutdownUnparksAWaiterWithNoDeadline) {
+    struct Shared {
+        std::unique_ptr<InProcessTransport> client;
+        std::unique_ptr<InProcessTransport> server;
+        std::atomic<std::uint32_t> parked{0};
+        std::atomic<bool> returned{false};
+        std::atomic<bool> woke{true};
+    };
+    auto shared = std::make_shared<Shared>();
+    InProcessTransport::CreatePair(shared->client, shared->server);
+
+    std::thread waiter([shared] {
+        shared->woke.store(shared->server->SelfDoorbell().Wait(
+            shared->parked, [] { return false; }, kDefaultSpinUs, kWaitForever));
+        shared->returned.store(true, std::memory_order_release);
+    });
+    // Past the spin and announced as parked; a little longer and it is inside
+    // Park. (A Kill that lands before the Park is handled too - Park returns at
+    // once on a dead bell - but the case under test is the parked one.)
+    while (shared->parked.load() == 0) {
+        std::this_thread::yield();
+    }
+    std::this_thread::sleep_for(std::chrono::milliseconds(50));
+    ASSERT_FALSE(shared->returned.load());
+
+    shared->client->Shutdown();
+
+    const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(5);
+    while (!shared->returned.load(std::memory_order_acquire) &&
+           std::chrono::steady_clock::now() < deadline) {
+        std::this_thread::sleep_for(std::chrono::milliseconds(1));
+    }
+    if (!shared->returned.load(std::memory_order_acquire)) {
+        waiter.detach();
+        FAIL() << "Shutdown did not unpark a waiter with no deadline within 5 s: the inproc doorbell "
+                  "has no death state, so the waiter consumed the ring and parked again";
+    }
+    waiter.join();
+
+    // No wakeup was consumed - the bell died - and the park flag is clear.
+    EXPECT_FALSE(shared->woke.load());
+    EXPECT_TRUE(shared->server->SelfDoorbell().Dead());
+    EXPECT_TRUE(shared->client->SelfDoorbell().Dead());
+    EXPECT_EQ(shared->parked.load(), 0u);
+
+    // Sticky: a wait with no deadline that ARRIVES after the Shutdown returns at
+    // once rather than parking, so a late thread cannot hang either.
+    const auto start = std::chrono::steady_clock::now();
+    EXPECT_FALSE(shared->server->SelfDoorbell().Wait(
+        shared->parked, [] { return false; }, 0, kWaitForever));
+    EXPECT_LT(std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() - start)
+                  .count(),
+              1000);
+    EXPECT_EQ(shared->parked.load(), 0u);
+}

MobileGL/MG_Test/Wire/RingTest.cpp

@@ -231,6 +231,94 @@ TEST(RingTest, RecordLargerThanTheRingIsRefused) {
     EXPECT_TRUE(ring.Invariants());
 }
 
+TEST(RingTest, RecordLargerThanHalfTheRingIsRefused) {
+    // 256-byte ring: the bound is 128 bytes of header + payload.
+    RingFixture ring(256);
+    EXPECT_EQ(ring.Producer().MaxRecordBytes(), 128u);
+    // 8 + 240 = 248: fits the whole ring, does not fit half of it.
+    EXPECT_EQ(ring.Producer().Reserve(1, kRecNone, 240), nullptr);
+    // 8 + 128 = 136: one step over the bound, refused the same way...
+    EXPECT_EQ(ring.Producer().Reserve(1, kRecNone, 128), nullptr);
+    // ...and 8 + 120 = 128, exactly the bound, is accepted.
+    EXPECT_NE(ring.Producer().Reserve(1, kRecNone, 120), nullptr);
+    EXPECT_TRUE(ring.Invariants());
+}
+
+// The scenario that motivated the bound, as the negative control. Whether a record
+// can be placed must not depend on where the head happens to be. With "total <=
+// capacity" as the only rule, a 248-byte record is accepted at head offset 0 of an
+// empty 256-byte ring and refused forever at head offset 16 of the same empty
+// ring - it would need a 240-byte wrap pad plus itself, 488 bytes - while
+// FreeBytes() reports 256 the whole time, so a producer waiting for FreeBytes()
+// >= 248 spins on nullptr with nothing logged. Both answers have to be the same
+// refusal, and it has to be the loud one.
+TEST(RingTest, RecordPlaceabilityDoesNotDependOnTheHeadOffset) {
+    RingFixture atOffsetZero(256);
+    void* atZero = atOffsetZero.Producer().Reserve(1, kRecNone, 240);
+
+    RingFixture atOffsetSixteen(256);
+    ASSERT_TRUE(atOffsetSixteen.WriteRecord(1, 8, 0x01)); // 8 + 8 = 16 bytes
+    RingRecordView view{};
+    ASSERT_TRUE(atOffsetSixteen.Consumer().Pop(view));
+    atOffsetSixteen.Consumer().PublishRetired();
+    ASSERT_EQ(atOffsetSixteen.Producer().LocalHead(), 16u);
+    ASSERT_EQ(atOffsetSixteen.Producer().FreeBytes(), 256u);
+    void* atSixteen = atOffsetSixteen.Producer().Reserve(1, kRecNone, 240);
+
+    EXPECT_EQ(atSixteen, nullptr);
+    EXPECT_EQ(atZero, nullptr)
+        << "a 248-byte record was accepted at head offset 0 but is unplaceable at head offset 16 of "
+           "the same empty ring: the emitter cannot tell a refusal it must chunk from a full ring it "
+           "must wait on";
+    EXPECT_TRUE(atOffsetZero.Invariants());
+    EXPECT_TRUE(atOffsetSixteen.Invariants());
+}
+
+// The positive half of the same argument: a record of exactly half the capacity is
+// placeable at EVERY head offset of an empty ring, because the wrap pad in front of
+// it costs at most total-8 bytes. Walk the head to each 8-byte offset with bare
+// header records and reserve the maximal record there.
+TEST(RingTest, HalfCapacityRecordFitsAtEveryHeadOffset) {
+    RingFixture ring(256);
+    const std::uint64_t mask = ring.Capacity() - 1;
+    const std::uint64_t maximal = ring.Producer().MaxRecordBytes() - sizeof(RingRecordHeader); // 120
+    for (std::uint64_t target = 0; target < ring.Capacity(); target += 8) {
+        // A bare header never straddles the boundary, so no pad appears on the way.
+        while ((ring.Producer().LocalHead() & mask) != target) {
+            ASSERT_TRUE(ring.WriteRecord(1, 0, 0));
+            RingRecordView filler{};
+            ASSERT_TRUE(ring.Consumer().Pop(filler));
+            ring.Consumer().PublishRetired();
+        }
+        ASSERT_EQ(ring.Producer().FreeBytes(), ring.Capacity()) << "head offset " << target;
+        void* payload = ring.Producer().Reserve(2, kRecNone, maximal);
+        ASSERT_NE(payload, nullptr) << "head offset " << target;
+        ring.Producer().Publish();
+        RingRecordView view{};
+        bool corrupt = false;
+        ASSERT_TRUE(ring.Consumer().Pop(view, &corrupt)) << "head offset " << target;
+        ASSERT_FALSE(corrupt);
+        EXPECT_EQ(view.kind, 2u);
+        EXPECT_EQ(view.payloadSize, maximal);
+        ring.Consumer().PublishRetired();
+        ASSERT_TRUE(ring.Invariants()) << "head offset " << target;
+    }
+}
+
+TEST(RingTest, RejectsARingTooSmallForTheSmallestRecord) {
+    alignas(4096) RingControl control{};
+    InitRingControl(control);
+    std::uint8_t bytes[16] = {};
+    // One header's worth of ring can carry nothing once a record may be at most
+    // half the ring; two headers' worth carries a bare header.
+    RingProducer tooSmall(&control, bytes, sizeof(RingRecordHeader), RingCursorSet::Cmd);
+    EXPECT_FALSE(tooSmall.Valid());
+    RingProducer smallest(&control, bytes, kMinRingCapacity, RingCursorSet::Cmd);
+    ASSERT_TRUE(smallest.Valid());
+    EXPECT_EQ(smallest.MaxRecordBytes(), sizeof(RingRecordHeader));
+    EXPECT_NE(smallest.Reserve(1, kRecNone, 0), nullptr);
+}
+
 TEST(RingTest, HardDrainBumpsTheGenerationOnlyWhenQuiesced) {
     RingFixture ring(256);
     ASSERT_TRUE(ring.WriteRecord(1, 32, 0x01));

scripts/gen_pipe.py

@@ -272,6 +272,16 @@ def gen_wire(calls):
 // a record that is shorter than its own type, longer than what is left in the buffer, or
 // not a multiple of 8 is protocol corruption and is fatal. There is no recovery path -
 // silently applying a truncated record is how a corrupt stream becomes a wrong picture.
+//
+// OVERSIZED PAYLOADS ARE CHUNKED, NEVER EMITTED WHOLE (plan section 8.2: G3 has to define
+// the path for a record larger than the segment). The bound is the ring's,
+// RingProducer::MaxRecordBytes() == Capacity()/2, and it is exact rather than
+// conservative: a record has to be placeable at every head offset of an empty ring, the
+// wrap pad in front of it costs up to total-8 bytes, and only a record of at most half the
+// ring survives that at every offset. An emitter holding more than Capacity()/2 bytes of
+// record (a large resource_subdata, a create_shader_state archive) splits it into several
+// records of at most that size; the transport refuses a bigger one outright - nullptr plus
+// an MGLOG_E - rather than let the producer wait on free bytes that can never suffice.
 
 struct MGPWireRecHeader {
     Uint16 Op;    // MGPWireOp