[Test] (MG_Remote, Wire): pin the hung-up doorbell, the wakeup that must not be eaten, the ring's capacity ceiling and the publish-then-ring handoff

- FdPassingTest.SocketDoorbellStopsParkingWhenThePeerHangsUp builds a
  SOCK_STREAM socketpair - deliberately not FdPassing::CreateSocketPair's
  datagram pair, because only a stream end reports the hangup at all - closes
  the notifier, and asserts Park returns false, latches Dead(), stays latched,
  and that a Wait with kWaitForever gives up in under a second instead of
  spinning.

- FdPassingTest.SocketDoorbellStillDeliversTheLastRingBeforeAHangup rings and
  then closes: detecting death must not swallow the wakeup already sitting in
  the socket buffer, since the peer's last publish is the one a waiter is most
  likely to be blocked on.

- InProcessTransportTest.AFrameWakeupIsNotEatenByAWaiterOnDescriptors blocks
  two readers on one endpoint with two different predicates and requires the
  frame to arrive within 2s rather than "eventually, when a receive timed out".

- RingTest.RejectsACapacityTheRecordHeaderCannotDescribe refuses 4 GiB from
  both roles without mapping anything (the constructor rejects before it
  touches the base pointer) and keeps 2 GiB accepted as the positive control.

- RingTest.DoorbellHandoffWakesBothSidesOnEveryPublish runs 2000 records
  through the ring with real parking in both directions, in the publish-then-
  NotifyIfParked order the fences assume. It cannot prove the fence pairing -
  no test can, since x86 has to actually hold the release store in the store
  buffer across the flag read - but it exercises the exact call order, and a
  lost wakeup surfaces as a Wait that times out with work available (a red
  test) rather than as a hung CI job.

- Result: 1429 unit tests pass with MOBILEGL_BUILD_DISAGGREGATED=ON (1424
  before this commit; the wire subset is 47), 1382 pass with it OFF, and the
  same three pre-existing skips appear in both. Every new case was verified to
  fail with its fix reverted and to pass again with it restored.

MobileGL/MG_Test/Wire/FdPassingTest.cpp

@@ -25,6 +25,7 @@
 #include <chrono>
 #include <cstdint>
 #include <string>
+#include <sys/socket.h>
 #include <sys/wait.h>
 #include <thread>
 #include <unistd.h>
@@ -248,3 +249,57 @@ TEST(FdPassingTest, SocketDoorbellTimesOutAndRemembersAnEarlyWakeup) {
     ::close(sockets[0]);
     ::close(sockets[1]);
 }
+
+// A doorbell whose peer has hung up must report that, not keep saying "ready".
+// Park used to treat any `poll` return > 0 as a wakeup without ever looking at
+// revents, and a closed peer leaves a stream socket permanently poll-ready
+// with nothing to read - so Doorbell::Wait re-parked in a tight loop at full
+// clock, unbounded when the caller passed kWaitForever. That is the pathology
+// the bidirectional doorbell exists to prevent, arrived at from the other
+// side.
+TEST(FdPassingTest, SocketDoorbellStopsParkingWhenThePeerHangsUp) {
+    // A SOCK_STREAM pair, not FdPassing::CreateSocketPair's datagram pair:
+    // measured on Linux, a closed peer makes a stream end report
+    // POLLIN|POLLHUP with recv()==0, while a datagram end reports no readiness
+    // at all. The stream shape is what the spawn transport will use, and it is
+    // the shape that used to spin.
+    int sockets[2] = {-1, -1};
+    ASSERT_EQ(::socketpair(AF_UNIX, SOCK_STREAM, 0, sockets), 0);
+
+    SocketDoorbell waiterBell(sockets[0], kDoorbellRingAdvanced, /*ownsFd=*/true);
+    ASSERT_EQ(::close(sockets[1]), 0);
+
+    const auto start = std::chrono::steady_clock::now();
+    EXPECT_FALSE(waiterBell.Park(kWaitForever));
+    EXPECT_TRUE(waiterBell.Dead());
+    // Latched: no second syscall storm either.
+    EXPECT_FALSE(waiterBell.Park(kWaitForever));
+
+    // ...and a Wait with no deadline at all gives up instead of re-parking.
+    std::atomic<std::uint32_t> parked{0};
+    EXPECT_FALSE(waiterBell.Wait(
+        parked, [] { return false; }, /*spinUs=*/0, kWaitForever));
+    EXPECT_EQ(parked.load(), 0u);
+    EXPECT_LT(std::chrono::duration_cast<std::chrono::milliseconds>(
+                  std::chrono::steady_clock::now() - start)
+                  .count(),
+              1000);
+}
+
+TEST(FdPassingTest, SocketDoorbellStillDeliversTheLastRingBeforeAHangup) {
+    int sockets[2] = {-1, -1};
+    ASSERT_EQ(::socketpair(AF_UNIX, SOCK_STREAM, 0, sockets), 0);
+
+    SocketDoorbell waiterBell(sockets[0], kDoorbellRingAdvanced, /*ownsFd=*/true);
+    SocketDoorbell notifierBell(sockets[1], kDoorbellRingAdvanced, /*ownsFd=*/false);
+
+    // Ring, then die. Detecting the hangup must not swallow the wakeup that
+    // was already queued - the peer's last publish is the one a waiter is
+    // most likely to be blocked on.
+    notifierBell.Notify();
+    ASSERT_EQ(::close(sockets[1]), 0);
+
+    EXPECT_TRUE(waiterBell.Park(1000));
+    EXPECT_TRUE(waiterBell.Dead());
+    EXPECT_FALSE(waiterBell.Park(kWaitForever));
+}

MobileGL/MG_Test/Wire/InProcessTransportTest.cpp

@@ -217,6 +217,62 @@ TEST(InProcessTransportTest, HandsOverADescriptorAndItsSideband) {
     ::close(pipeFds[0]);
     ::close(pipeFds[1]);
 }
+
+TEST(InProcessTransportTest, AFrameWakeupIsNotEatenByAWaiterOnDescriptors) {
+    std::unique_ptr<InProcessTransport> client;
+    std::unique_ptr<InProcessTransport> server;
+    InProcessTransport::CreatePair(client, server);
+
+    // Two readers on the SAME endpoint, blocked on two different predicates.
+    // With one condition_variable per direction and notify_one, the SendFrame
+    // below could be delivered to the descriptor waiter, which re-tests its
+    // own predicate and goes back to sleep - and the message then sits
+    // undelivered until some unrelated later event. ITransport narrows the
+    // contract to one dedicated reader thread, but that is a comment, and the
+    // first caller that splits its reader should not have to discover this.
+    std::atomic<bool> fdWaiterStarted{false};
+    std::thread fdWaiter([&] {
+        std::vector<std::uint8_t> sideband(FdPassing::kMaxSidebandBytes);
+        MobileGLMutableByteSpan span{sideband.data(), sideband.size()};
+        int fd = -1;
+        std::uint64_t size = 0;
+        fdWaiterStarted.store(true);
+        // Never offered a descriptor: this one ends on the Shutdown below.
+        EXPECT_EQ(client->ReceiveFd(&fd, span, &size, kWaitForever),
+                  MOBILEGL_ERR_TRANSPORT_CLOSED);
+        EXPECT_EQ(fd, -1);
+    });
+    while (!fdWaiterStarted.load()) {
+        std::this_thread::yield();
+    }
+    std::this_thread::sleep_for(std::chrono::milliseconds(50));
+
+    std::atomic<bool> frameWaiterStarted{false};
+    std::string got;
+    std::thread frameWaiter([&] {
+        frameWaiterStarted.store(true);
+        got = Receive(*client, 4000);
+    });
+    while (!frameWaiterStarted.load()) {
+        std::this_thread::yield();
+    }
+    std::this_thread::sleep_for(std::chrono::milliseconds(50));
+
+    const std::string message = "wake the right waiter";
+    const auto start = std::chrono::steady_clock::now();
+    ASSERT_EQ(server->SendFrame(Span(message)), MOBILEGL_OK);
+    frameWaiter.join();
+    const auto elapsedMs = std::chrono::duration_cast<std::chrono::milliseconds>(
+                               std::chrono::steady_clock::now() - start)
+                               .count();
+
+    EXPECT_EQ(got, message);
+    // Not "eventually, when the receive timed out and re-checked".
+    EXPECT_LT(elapsedMs, 2000);
+
+    client->Shutdown();
+    fdWaiter.join();
+}
 #endif
 
 TEST(InProcessTransportTest, DoorbellWakesAParkedWaiter) {

MobileGL/MG_Test/Wire/RingTest.cpp

@@ -10,6 +10,7 @@
 // invariants, wrap-around, backpressure, the generation bump after a hard
 // drain, and a real two-thread producer/consumer run.
 
+#include <MG_Remote/Transport/Doorbell.h>
 #include <MG_Remote/Transport/Ring.h>
 
 #include <gtest/gtest.h>
@@ -120,6 +121,27 @@ TEST(RingTest, RejectsANonPowerOfTwoCapacity) {
     EXPECT_EQ(producer.Reserve(1, kRecNone, 8), nullptr);
 }
 
+TEST(RingTest, RejectsACapacityTheRecordHeaderCannotDescribe) {
+    alignas(4096) RingControl control{};
+    InitRingControl(control);
+    // 4 GiB is a legal power of two, but RingRecordHeader::size is 32 bits and
+    // both a record's size and a wrap filler's size are bounded only by the
+    // capacity: they would be truncated on the way in and then bounds-checked
+    // in their truncated form on the way out. Nothing is mapped here - the
+    // constructor rejects before it ever touches the base pointer.
+    std::uint8_t dummy = 0;
+    constexpr std::uint64_t kFourGiB = 4ull * 1024 * 1024 * 1024;
+    EXPECT_GT(kFourGiB, kMaxRingCapacity);
+    RingProducer producer(&control, &dummy, kFourGiB, RingCursorSet::Cmd);
+    EXPECT_FALSE(producer.Valid());
+    RingConsumer consumer(&control, &dummy, kFourGiB, RingCursorSet::Cmd);
+    EXPECT_FALSE(consumer.Valid());
+
+    // The largest ring the header CAN describe stays accepted.
+    RingProducer biggest(&control, &dummy, 1ull << 31, RingCursorSet::Cmd);
+    EXPECT_TRUE(biggest.Valid());
+}
+
 TEST(RingTest, RoundTripsRecordsInOrder) {
     RingFixture ring(4096);
     ASSERT_TRUE(ring.WriteRecord(1, 16, 0x10));
@@ -323,3 +345,87 @@ TEST(RingTest, SpscProducerConsumerThreadsAgreeOnEveryRecord) {
     EXPECT_TRUE(ring.Invariants());
     EXPECT_EQ(ring.Control().cmdRetiredTail.load(), ring.Control().cmdHead.load());
 }
+
+// The publish/park protocol end to end, in both directions: publish the
+// watermark, THEN NotifyIfParked; park with Doorbell::Wait. A lost wakeup on
+// either side shows up as a Wait that times out with work available rather
+// than as a hang, so the failure is a red test and not a stuck CI job.
+//
+// This cannot prove the seq_cst fence pairing (no test can - x86 needs the
+// store buffer to hold the release store across the flag read, and it usually
+// does not), but it does exercise the exact call order the fences assume, so a
+// future edit that rings the bell BEFORE publishing has somewhere to fail.
+TEST(RingTest, DoorbellHandoffWakesBothSidesOnEveryPublish) {
+    // 4 byte payloads: every record is exactly 16 bytes and 4096 is a multiple
+    // of that, so no wrap filler ever appears and "head != tail" is exactly
+    // "a record is waiting".
+    RingFixture ring(4096);
+    CondVarDoorbell consumerBell;
+    CondVarDoorbell producerBell;
+    std::atomic<bool> ok{true};
+    constexpr int kRecords = 2000;
+    constexpr std::uint64_t kRecordBytes = 16;
+
+    std::thread consumerThread([&] {
+        int seen = 0;
+        while (seen < kRecords) {
+            const bool woke = consumerBell.Wait(
+                ring.Control().consumerParked,
+                [&] {
+                    return ring.Control().cmdHead.load(std::memory_order_acquire) !=
+                           ring.Consumer().LocalTail();
+                },
+                kDefaultSpinUs, 5000);
+            if (!woke) {
+                ok.store(false); // a wakeup was lost, or the producer stalled
+                return;
+            }
+            RingRecordView view{};
+            bool corrupt = false;
+            while (ring.Consumer().Pop(view, &corrupt)) {
+                std::uint32_t value = 0;
+                std::memcpy(&value, view.payload, sizeof(value));
+                if (value != static_cast<std::uint32_t>(seen)) {
+                    ok.store(false);
+                    return;
+                }
+                ++seen;
+            }
+            if (corrupt) {
+                ok.store(false);
+                return;
+            }
+            ring.Consumer().PublishRetired();
+            NotifyIfParked(producerBell, ring.Control().producerParked);
+        }
+    });
+
+    for (int i = 0; i < kRecords && ok.load(); ++i) {
+        void* payload = nullptr;
+        while ((payload = ring.Producer().Reserve(1, kRecNone, sizeof(std::uint32_t))) == nullptr) {
+            if (!ok.load()) {
+                break;
+            }
+            if (!producerBell.Wait(
+                    ring.Control().producerParked,
+                    [&] { return ring.Producer().FreeBytes() >= kRecordBytes; }, kDefaultSpinUs,
+                    5000)) {
+                ok.store(false);
+                break;
+            }
+        }
+        if (payload == nullptr) {
+            break;
+        }
+        const std::uint32_t value = static_cast<std::uint32_t>(i);
+        std::memcpy(payload, &value, sizeof(value));
+        // Publish first, ring second. The other order reopens the lost-wakeup
+        // window no matter how strong the flag's memory order is.
+        ring.Producer().Publish();
+        NotifyIfParked(consumerBell, ring.Control().consumerParked);
+    }
+
+    consumerThread.join();
+    EXPECT_TRUE(ok.load());
+    EXPECT_TRUE(ring.Invariants());
+}