[MGPipe] cut buffer content ranges at the stage segment's chunk budget

One resource_subdata record's blob is staged whole in SEG_STAGE, so a range
larger than that arena is Fatal{RingOverrun, SEG_STAGE} at the encoder rather
than a split: the CI traces abort there on a 128 MiB arena's whole-buffer
follow-up against a 32 MiB segment.

- ClientSession::StageCapacityBytes(), the twin of EventRingCapacityBytes, so
  the emitter can size against the arena the codec allocates from.
- MGPipeStageChunkBytes(): clamp(segment/4, 4096, segment), 0 when there is
  nothing to fit (monolith, the server role's own uploads, no session).
- PipeFill.cpp's two buffer content walks take that cap instead of the
  record's own 2^32-1 bound, so an over-long range becomes several contiguous
  ascending records. StagedShadowStore::Adopt already accumulates coverage.

ClientSession.h/.cpp, GpuWritePending.h/.cpp, PipeFill.cpp.

MobileGL/MG_Impl/Pipe/PipeFill.cpp

@@ -984,6 +984,25 @@ namespace MobileGL::MG_Pipe {
         MGPipeRouteResourceRespecify(desc, initialBytes);
     }
 
+    // THE CAP THE TWO CONTENT WALKS BELOW CUT A RANGE AT. The record's own bound (2^32-1) is not
+    // the one a real upload meets first: one piece's bytes are staged WHOLE in SEG_STAGE, a
+    // linear arena, and a blob larger than that arena is Fatal{RingOverrun, "SEG_STAGE"} at the
+    // encoder rather than a split (PipeWireCodec.cpp:856-864). Measured on the CI traces: a
+    // 128 MiB arena's whole-buffer follow-up against a 32 MiB segment aborted there, which is
+    // the RingOverrun this walk exists to prevent.
+    //
+    // 0 from the helper means "nothing to fit" - monolith, the server role's own uploads (which
+    // run the monolith adapter), or a process with no session - and the record's own bound is
+    // then the answer, exactly as it was before the cap existed.
+#if MOBILEGL_BUILD_DISAGGREGATED
+    Uint64 MGPipeContentChunkCap() {
+        const SizeT stageChunk = MG_Remote::Client::MGPipeStageChunkBytes();
+        return stageChunk == 0 ? kMGPipeSubDataMaxRecordSize : static_cast<Uint64>(stageChunk);
+    }
+#else
+    constexpr Uint64 MGPipeContentChunkCap() { return kMGPipeSubDataMaxRecordSize; }
+#endif
+
     void MGPipeEmitResourceSubData(BufferObject& buffer, SizeT offset, SizeT size) {
         const MGPipeHandle handle = ContentHandleFor(buffer, "resource_subdata");
         if (MGPipeHandleIsNull(handle)) return;
@@ -1009,7 +1028,7 @@ namespace MobileGL::MG_Pipe {
                 // rather than re-read so the staged run and the record's own claim come from
                 // one number.
                 MGPipeRouteResourceSubData(record, base + at, length);
-            });
+            }, MGPipeContentChunkCap());
         if (!encodable) {
             MGLOG_E_ONCE("MGPipe: resource_subdata range [%llu, +%llu) on buffer %u cannot be encoded - "
                          "one record's destination box caps the offset at 2^31-1",
@@ -1040,7 +1059,7 @@ namespace MobileGL::MG_Pipe {
 #endif
                 // The application's STAGING store, valid for the duration of the call only.
                 MGPipeRouteBufferSubDataResident(record, base + (at - offset), length);
-            });
+            }, MGPipeContentChunkCap());
         if (!encodable) {
             MGLOG_E_ONCE("MGPipe: buffer_subdata_resident range [%llu, +%llu) on buffer %u cannot be encoded",
                          static_cast<unsigned long long>(offset), static_cast<unsigned long long>(size),

MobileGL/MG_Remote/Client/ClientSession.cpp

@@ -1467,6 +1467,8 @@ namespace MobileGL::MG_Remote::Client {
 
     Uint64 ClientSession::EventRingCapacityBytes() const { return m_shm.EventRingCapacity(); }
 
+    Uint64 ClientSession::StageCapacityBytes() const { return m_shm.StageBytes(); }
+
     Wire::SegmentTable& ClientSession::Segments() { return m_segments; }
 
     Transport::RingControl* ClientSession::Control() { return m_shm.CmdControl(); }

MobileGL/MG_Remote/Client/ClientSession.h

@@ -314,6 +314,12 @@ namespace MobileGL::MG_Remote::Client {
         // capacity/2, Ring.h:288).
         Uint64 EventRingCapacityBytes() const;
 
+        // SEG_STAGE's capacity, exposed so a content emitter can cut a range into records that
+        // always stage whole: the staging area is a linear arena of exactly these bytes (w1),
+        // and one blob larger than it is Fatal{RingOverrun, "SEG_STAGE"} at the encoder rather
+        // than a split (PipeWireCodec.cpp:856-864).
+        Uint64 StageCapacityBytes() const;
+
         // The CLIENT's own segment table (P5c ev, CONTRACT-P5C §4.3): the writeback
         // consumer resolves a SEG_EVENT blobref through it. Never the process resolver -
         // table 3 installs that one on the server role only.

MobileGL/MG_Remote/Client/GpuWritePending.cpp

@@ -9,6 +9,9 @@
 #include "GpuWritePending.h"
 
 #include "ClientSession.h"
+// RunsAsTheServerRole: declared beside the wire emitters that need the same predicate, so this
+// TU does not pull a server header in for it.
+#include "WireTables.h"
 
 #include <MG_Pipe/PipeMutation.h>
 #include <MG_State/GLState/BufferState/BufferObject.h>
@@ -199,6 +202,36 @@ namespace MobileGL::MG_Remote::Client {
         return static_cast<SizeT>(slice < 4096 ? 4096 : slice);
     }
 
+    SizeT MGPipeStageChunkBytesFor(Uint64 stageCapacityBytes) {
+        if (stageCapacityBytes == 0) return 0;
+        // A quarter of the arena, not all of it, for the readback slice's reason one ring over
+        // (above): the pieces of one range are staged one after another, an allocation is only
+        // reclaimable once the record carrying it has retired, and the allocator skips a
+        // remainder it cannot fill contiguously. A quarter leaves room for the pieces still in
+        // flight, so an ordinary whole-buffer upload never waits on a retirement it could have
+        // avoided.
+        const Uint64 quarter = stageCapacityBytes / 4;
+        // A floor so a pathologically small operator-supplied segment cannot make the piece
+        // width zero, which the walk reads as "refuse" (ResourceTracker.h:254); such a segment
+        // is broken anyway, and the producer's Fatal{RingOverrun} names it on the first stage.
+        const Uint64 chunk = quarter < 4096 ? 4096 : quarter;
+        return static_cast<SizeT>(chunk > stageCapacityBytes ? stageCapacityBytes : chunk);
+    }
+
+    SizeT MGPipeStageChunkBytes() {
+        // The server role's own uploads run the monolith adapter (RunsAsTheServerRole's comment
+        // in WireTables.h): cutting them would change how many calls the server's own backend
+        // sees for one application call, which is a monolith behaviour change on the apply
+        // thread and not this emitter's to make.
+        if (MG_Config::Transport == MG_Config::TransportMode::Monolith) return 0;
+        if (RunsAsTheServerRole()) return 0;
+        ClientSession* session = ClientSession::Active();
+        // No session: the emission WAS the application (every unit gate, and a monolith-shaped
+        // lane in a split build), so there is no arena to fit and the record's own bound stands.
+        if (session == nullptr) return 0;
+        return MGPipeStageChunkBytesFor(session->StageCapacityBytes());
+    }
+
     void AwaitBufferWriteback(BufferObject& buffer) {
         // THE WAIT IS THE BARRIER'S WAIT (R-3). The reply-slot id IS the record's seq, so
         // "appliedSeq reached my readback" and "my answer is back" are one condition, and

MobileGL/MG_Remote/Client/GpuWritePending.h

@@ -157,4 +157,21 @@ namespace MobileGL::MG_Remote::Client {
     // active - the synchronous arm never slices.
     SizeT BufferWritebackSliceBytes();
 
+    // The same question for CONTENT: how many bytes one resource_subdata record may stage
+    // before the range has to be cut. SEG_STAGE is a linear arena, one record's blob is
+    // allocated from it whole, and a blob larger than the arena is
+    // Fatal{RingOverrun, "SEG_STAGE"} at the encoder rather than a split
+    // (PipeWireCodec.cpp:856-864) - measured on the CI traces, where a 128 MiB arena's
+    // whole-buffer follow-up against a 32 MiB segment aborted. 0 means "do not cut", i.e. the
+    // record's own bound (MGPipeForEachSubDataRecordRange's default), which is the answer for
+    // monolith, for the server role's own uploads (they run the monolith adapter) and for a
+    // process with no session - every unit gate.
+    SizeT MGPipeStageChunkBytes();
+
+    // The clamp itself, a pure function of the segment's size so that a unit gate can drive the
+    // splitter at exactly the value a live session produces. A quarter of the segment, never
+    // below 4096 (a zero-width piece is a refusal, ResourceTracker.h:254) and never above the
+    // segment (a larger cap would stage the very blob the arena refuses).
+    SizeT MGPipeStageChunkBytesFor(Uint64 stageCapacityBytes);
+
 } // namespace MobileGL::MG_Remote::Client