From 5f846d96e360ae314dd41e9c4e7d5223ca77c8ce Mon Sep 17 00:00:00 2001 From: Swung0x48 Date: Mon, 21 Sep 2026 09:47:32 -0400 Subject: [PATCH] [MG_Remote] (P6/so): SocketTransport, green over a real socketpair before any second process MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The first P6 code that runs. ITransport over a connected AF_UNIX SOCK_STREAM socket, with SCM_RIGHTS descriptor passing on a dedicated SOCK_DGRAM aux pair. No shared memory: this is the CONTROL plane, and the hot path bypasses ITransport entirely by its own header's account. What the socket carries is the handshake, surface ops, resync, aux requests and fatals - and, once sm exists, the four segment descriptors that let the peer map the data plane. It moves the keys, not the bytes. CONTRACT-P6 §8 and the spawn plan both require this green over a plain socketpair with two threads BEFORE a child is ever forked, for the reason SessionRings.h gives about the attach half: a transport first exercised on the day the process appears is a transport whose bugs all arrive at once, on the hardest day to debug them. That requirement is now met. It reinvents nothing. Framing.h already had AppendFrame and a real byte-stream reassembler with exactly the BUFFER_TOO_SMALL keep-the-message semantics ITransport specifies - written for this and never before driven by a socket. FdPassing.h already had SCM_RIGHTS over a dedicated datagram pair. What is new is ~250 lines joining them, plus the platform refusal: Windows has no SCM_RIGHTS and ShmSegment::Adopt is POSIX-only, so CreatePair refuses BY NAME at construction rather than handing back a transport that fails every later call. SOCK_STREAM rather than SOCK_SEQPACKET, deliberately: Doorbell.h:390-398 already chose a stream socket so a closed peer surfaces as POLLIN|POLLHUP with recv()==0 rather than as a hang, and the control plane may not disagree with the bell about what death looks like. Framing.h is what makes a stream safe - the message boundary is in the payload, not in the socket. TESTS: the same nine ITransport-contract cases InProcessTransportTest runs, copied rather than shared. Folding both suites into one parameterised fixture would rename the sixteen existing cases and G14 says names only grow; and the two suites do not share a handle anyway, since InProcessTransport's tests reach into condvar doorbells this class does not have. Nine names added, zero removed. The seven Doorbell* cases are deliberately absent: they exercise CondVarDoorbell, not ITransport. Plus three a stream has and a message queue cannot: a 256 KiB payload that provably crosses in several reads; a garbage stream that must latch failed and must NOT resync onto a boundary it invented; and adopting an already-connected fd with no aux channel, which answers UNSUPPORTED to ShareFd rather than pretending - rule G, a transport declares what it supports. Red-once, both run: drop the message on BUFFER_TOO_SMALL and six cases go red; make the framing latch recoverable and AGarbageStreamIsLatchedFailed fails by name. Gates: 12/12 in 677 ms. Unit suite 2340 tests, up 12, with the same two pre-existing environmental failures. G1 pull 0 added / 0 removed, .text size 10822147 unchanged, and zero SocketTransport symbols in the pull build. One note for the VM track this eventually serves: swapping AF_UNIX for AF_VSOCK is one line here and nothing at all in Framing. It is the shared-memory half that does not cross a kernel. --- CMakeLists.txt | 1 + .../MG_Remote/Transport/SocketTransport.cpp | 335 +++++++++++++++ .../MG_Remote/Transport/SocketTransport.h | 114 +++++ MobileGL/MG_Test/Wire/CMakeLists.txt | 4 + MobileGL/MG_Test/Wire/SocketTransportTest.cpp | 390 ++++++++++++++++++ 5 files changed, 844 insertions(+) create mode 100644 MobileGL/MG_Remote/Transport/SocketTransport.cpp create mode 100644 MobileGL/MG_Remote/Transport/SocketTransport.h create mode 100644 MobileGL/MG_Test/Wire/SocketTransportTest.cpp diff --git a/CMakeLists.txt b/CMakeLists.txt index 43e0a468..1a5e1750 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -556,6 +556,7 @@ if (MOBILEGL_BUILD_DISAGGREGATED) MobileGL/MG_Remote/Transport/ShmSegmentWin32.cpp MobileGL/MG_Remote/Transport/FdPassing.cpp MobileGL/MG_Remote/Transport/InProcessTransport.cpp + MobileGL/MG_Remote/Transport/SocketTransport.cpp # Keeps MG_Util/Debug/Log.h - and through it the GL frontend's # umbrella header - out of the header-only wire code (WireLog.h). MobileGL/MG_Remote/Transport/WireLog.cpp diff --git a/MobileGL/MG_Remote/Transport/SocketTransport.cpp b/MobileGL/MG_Remote/Transport/SocketTransport.cpp new file mode 100644 index 00000000..31abe6d6 --- /dev/null +++ b/MobileGL/MG_Remote/Transport/SocketTransport.cpp @@ -0,0 +1,335 @@ +// MobileGL - MobileGL/MG_Remote/Transport/SocketTransport.cpp +// Copyright (c) 2025-2026 MobileGL-Dev +// Licensed under the GNU Lesser General Public License v3.0: +// https://www.gnu.org/licenses/gpl-3.0.txt +// https://www.gnu.org/licenses/lgpl-3.0.txt +// SPDX-License-Identifier: LGPL-3.0-only +// End of Source File Header + +#include "SocketTransport.h" + +#include "Doorbell.h" // kWaitForever +#include "FdPassing.h" +#include "WireLog.h" + +#include +#include +#include + +#if defined(__unix__) || defined(__APPLE__) || defined(__ANDROID__) +# define MOBILEGL_SOCKET_TRANSPORT_POSIX 1 +# include +# include +# include +# include +#else +# define MOBILEGL_SOCKET_TRANSPORT_POSIX 0 +#endif + +namespace MobileGL::MG_Remote::Transport { + +#if MOBILEGL_SOCKET_TRANSPORT_POSIX + + namespace { + constexpr std::uint64_t kPumpChunkBytes = 64ull * 1024; + + void CloseIfOpen(int& fd) { + if (fd >= 0) { + ::close(fd); + fd = -1; + } + } + + // poll() for readability. Returns: + // 1 readable 0 timed out -1 error (errno set) + // EINTR is retried rather than reported: a signal is not a transport + // event, and reporting it as one would make every caller's timeout + // budget a lie on a process that takes signals (and `sm`'s child takes + // SIGCHLD by construction). + int WaitReadable(int fd, std::uint32_t timeoutMs) { + for (;;) { + struct pollfd pfd {}; + pfd.fd = fd; + pfd.events = POLLIN; + const int timeout = + timeoutMs == kWaitForever ? -1 : static_cast(timeoutMs); + const int rc = ::poll(&pfd, 1, timeout); + if (rc < 0 && errno == EINTR) { + continue; + } + return rc; + } + } + } // namespace + + MobileGLResult SocketTransport::CreatePair(std::unique_ptr& outClient, + std::unique_ptr& outServer) { + outClient.reset(); + outServer.reset(); + + int stream[2] = {-1, -1}; + // SOCK_STREAM, not SOCK_SEQPACKET: Doorbell.h:390-398 already chose a + // stream socket for the bell so that a closed peer surfaces as + // POLLIN|POLLHUP with recv()==0 rather than as a hang, and the control + // plane must not disagree with the bell about what death looks like. + // Framing.h is what makes a stream safe: the message boundary is in the + // payload, not in the socket. + if (::socketpair(AF_UNIX, SOCK_STREAM, 0, stream) != 0) { + WireLogError("MG_Remote SocketTransport: socketpair(SOCK_STREAM) failed: %s", + std::strerror(errno)); + return MOBILEGL_ERR_UNSUPPORTED; + } + + int aux[2] = {-1, -1}; + const MobileGLResult auxResult = FdPassing::CreateSocketPair(aux); + if (auxResult != MOBILEGL_OK) { + ::close(stream[0]); + ::close(stream[1]); + WireLogError("MG_Remote SocketTransport: the aux (SCM_RIGHTS) pair failed"); + return auxResult; + } + + outClient = std::make_unique(stream[0], aux[0], TransportRole::Client); + outServer = std::make_unique(stream[1], aux[1], TransportRole::Server); + return MOBILEGL_OK; + } + + SocketTransport::SocketTransport(int streamFd, int auxFd, TransportRole role) + : m_streamFd(streamFd), m_auxFd(auxFd), m_role(role) {} + + SocketTransport::~SocketTransport() { Shutdown(); } + + MobileGLResult SocketTransport::SendFrame(MobileGLByteSpan bytes) { + if (bytes.size > kMaxFramePayloadSize) { + return MOBILEGL_ERR_INVALID_ARGUMENT; + } + if (bytes.size != 0 && bytes.data == nullptr) { + return MOBILEGL_ERR_INVALID_ARGUMENT; + } + + std::lock_guard lock(m_sendMutex); + if (m_closed || m_streamFd < 0) { + return MOBILEGL_ERR_TRANSPORT_CLOSED; + } + + // Framed into ONE buffer and written as one span. Not because a stream + // needs it - it does not - but because a partial write of the header + // followed by a failure would leave the peer's reassembler parked on a + // length it will never receive, which is indistinguishable from a slow + // sender and therefore not detectable at all. + std::vector framed; + framed.reserve(static_cast(kFrameHeaderSize + bytes.size)); + const MobileGLResult appended = AppendFrame(framed, bytes.data, bytes.size); + if (appended != MOBILEGL_OK) { + return appended; + } + + std::uint64_t written = 0; + while (written < framed.size()) { + const ssize_t n = ::send(m_streamFd, framed.data() + written, + static_cast(framed.size() - written), + MSG_NOSIGNAL); + if (n > 0) { + written += static_cast(n); + continue; + } + if (n < 0 && errno == EINTR) { + continue; + } + // EPIPE / ECONNRESET are the peer being gone, which is a transport + // state and not an error to shout about: the device-lost latch (dl) + // is what turns it into a GL-visible fact. + if (n < 0 && (errno == EPIPE || errno == ECONNRESET)) { + m_closed = true; + return MOBILEGL_ERR_TRANSPORT_CLOSED; + } + WireLogError("MG_Remote SocketTransport: send failed after %llu of %llu bytes: %s", + static_cast(written), + static_cast(framed.size()), std::strerror(errno)); + m_closed = true; + return MOBILEGL_ERR_TRANSPORT_CLOSED; + } + return MOBILEGL_OK; + } + + MobileGLResult SocketTransport::PumpOnce(std::uint32_t timeoutMs) { + // Caller holds m_recvMutex. + if (m_failed) { + return MOBILEGL_ERR_PROTOCOL_MISMATCH; + } + if (m_streamFd < 0) { + return MOBILEGL_ERR_TRANSPORT_CLOSED; + } + + const int ready = WaitReadable(m_streamFd, timeoutMs); + if (ready == 0) { + return MOBILEGL_ERR_TIMEOUT; + } + if (ready < 0) { + WireLogError("MG_Remote SocketTransport: poll failed: %s", std::strerror(errno)); + return MOBILEGL_ERR_TRANSPORT_CLOSED; + } + + std::uint8_t chunk[kPumpChunkBytes]; + for (;;) { + const ssize_t n = ::recv(m_streamFd, chunk, sizeof(chunk), 0); + if (n > 0) { + const MobileGLResult fed = + m_reader.Feed(chunk, static_cast(n)); + if (fed != MOBILEGL_OK) { + // Bad magic or an over-long length. FrameReader latches + // itself failed and never recovers; mirror that here so a + // caller cannot retry its way past a corrupt stream. + m_failed = true; + WireLogError("MG_Remote SocketTransport: framing violated on the stream; " + "the transport is latched failed"); + return MOBILEGL_ERR_PROTOCOL_MISMATCH; + } + return MOBILEGL_OK; + } + if (n == 0) { + // Clean EOF. The peer is gone - but anything already reassembled + // stays readable, which ITransport promises in as many words. + m_closed = true; + return MOBILEGL_ERR_TRANSPORT_CLOSED; + } + if (errno == EINTR) { + continue; + } + if (errno == EAGAIN || errno == EWOULDBLOCK) { + return MOBILEGL_ERR_TIMEOUT; + } + if (errno == ECONNRESET) { + m_closed = true; + return MOBILEGL_ERR_TRANSPORT_CLOSED; + } + WireLogError("MG_Remote SocketTransport: recv failed: %s", std::strerror(errno)); + m_closed = true; + return MOBILEGL_ERR_TRANSPORT_CLOSED; + } + } + + MobileGLResult SocketTransport::ReceiveFrame(MobileGLMutableByteSpan buffer, + std::uint64_t* outSize, + std::uint32_t timeoutMs) { + if (outSize == nullptr) { + return MOBILEGL_ERR_INVALID_ARGUMENT; + } + *outSize = 0; + + std::lock_guard lock(m_recvMutex); + if (m_failed) { + return MOBILEGL_ERR_PROTOCOL_MISMATCH; + } + + for (;;) { + // Drain the reassembler FIRST, every time. A message that is already + // whole must be returned even when the peer has since closed, and + // must be returned without a syscall when the caller is polling. + if (m_reader.HasMessage()) { + return m_reader.TakeMessage(buffer, outSize); + } + if (m_closed) { + return MOBILEGL_ERR_TRANSPORT_CLOSED; + } + + const MobileGLResult pumped = PumpOnce(timeoutMs); + if (pumped == MOBILEGL_OK) { + continue; // maybe a whole message now; maybe a fragment - loop decides + } + if (pumped == MOBILEGL_ERR_TRANSPORT_CLOSED && m_reader.HasMessage()) { + return m_reader.TakeMessage(buffer, outSize); + } + return pumped; + } + } + + std::uint64_t SocketTransport::PeekFrameSize() { + std::lock_guard lock(m_recvMutex); + if (m_failed) { + return 0; + } + if (!m_reader.HasMessage() && !m_closed) { + // A non-blocking pump, so a caller that has not received yet can + // still size its buffer. Errors are not reported here: PeekFrameSize + // answers "how big is the next message", and 0 means "none buffered", + // which is the truthful answer for every failure this can hit. + (void)PumpOnce(0); + } + return m_reader.PendingMessageSize(); + } + + MobileGLResult SocketTransport::ShareFd(int fd, MobileGLByteSpan sideband) { + if (m_auxFd < 0) { + // A link that declared no descriptor passing. Rule G: it does not + // invent a third answer, and the caller transfers the bytes instead. + return MOBILEGL_ERR_UNSUPPORTED; + } + std::lock_guard lock(m_sendMutex); + if (m_closed) { + return MOBILEGL_ERR_TRANSPORT_CLOSED; + } + return FdPassing::SendFd(m_auxFd, fd, sideband); + } + + MobileGLResult SocketTransport::ReceiveFd(int* outFd, MobileGLMutableByteSpan sideband, + std::uint64_t* outSidebandSize, + std::uint32_t timeoutMs) { + if (m_auxFd < 0) { + return MOBILEGL_ERR_UNSUPPORTED; + } + return FdPassing::ReceiveFd(m_auxFd, outFd, sideband, outSidebandSize, timeoutMs); + } + + void SocketTransport::Shutdown() { + // Idempotent, and tears down BOTH directions - which on a socket is what + // closing does anyway. Taking both locks in a fixed order (send, then + // recv) so a concurrent sender and the dedicated reader cannot deadlock + // against each other here; every other path takes exactly one. + std::lock_guard sendLock(m_sendMutex); + std::lock_guard recvLock(m_recvMutex); + if (m_streamFd >= 0) { + // shutdown() before close() so a peer blocked in recv() wakes now + // rather than whenever the last duplicate of this fd goes away. + ::shutdown(m_streamFd, SHUT_RDWR); + } + CloseIfOpen(m_streamFd); + CloseIfOpen(m_auxFd); + m_closed = true; + } + +#else // !MOBILEGL_SOCKET_TRANSPORT_POSIX + + // Windows has no SCM_RIGHTS and ShmSegment::Adopt is POSIX-only, so P6 lands + // POSIX only (CONTRACT-P6 §2.6). The refusal is NAMED at construction rather + // than at first use: a transport that accepts a connection and then fails + // every operation is the silent-fallback shape this project forbids. + MobileGLResult SocketTransport::CreatePair(std::unique_ptr&, + std::unique_ptr&) { + WireLogError("MG_Remote SocketTransport: unsupported on this platform - P6 lands POSIX " + "only (CONTRACT-P6 §2.6); `unix:` and `pipe:` stay named refusals"); + return MOBILEGL_ERR_UNSUPPORTED; + } + + SocketTransport::SocketTransport(int, int, TransportRole role) : m_role(role) {} + SocketTransport::~SocketTransport() = default; + + MobileGLResult SocketTransport::SendFrame(MobileGLByteSpan) { return MOBILEGL_ERR_UNSUPPORTED; } + MobileGLResult SocketTransport::ReceiveFrame(MobileGLMutableByteSpan, std::uint64_t*, + std::uint32_t) { + return MOBILEGL_ERR_UNSUPPORTED; + } + std::uint64_t SocketTransport::PeekFrameSize() { return 0; } + MobileGLResult SocketTransport::ShareFd(int, MobileGLByteSpan) { + return MOBILEGL_ERR_UNSUPPORTED; + } + MobileGLResult SocketTransport::ReceiveFd(int*, MobileGLMutableByteSpan, std::uint64_t*, + std::uint32_t) { + return MOBILEGL_ERR_UNSUPPORTED; + } + MobileGLResult SocketTransport::PumpOnce(std::uint32_t) { return MOBILEGL_ERR_UNSUPPORTED; } + void SocketTransport::Shutdown() { m_closed = true; } + +#endif + +} // namespace MobileGL::MG_Remote::Transport diff --git a/MobileGL/MG_Remote/Transport/SocketTransport.h b/MobileGL/MG_Remote/Transport/SocketTransport.h new file mode 100644 index 00000000..6389afc1 --- /dev/null +++ b/MobileGL/MG_Remote/Transport/SocketTransport.h @@ -0,0 +1,114 @@ +// MobileGL - MobileGL/MG_Remote/Transport/SocketTransport.h +// Copyright (c) 2025-2026 MobileGL-Dev +// Licensed under the GNU Lesser General Public License v3.0: +// https://www.gnu.org/licenses/gpl-3.0.txt +// https://www.gnu.org/licenses/lgpl-3.0.txt +// SPDX-License-Identifier: LGPL-3.0-only +// End of Source File Header + +// ITransport over a connected stream socket. Package `so` (CONTRACT-P6.md). +// +// WHY THIS IS THE FIRST P6 CODE THAT RUNS. It is the whole control plane for a +// second process, and it needs NOTHING from the second process to be correct: +// CONTRACT-P6 and the spawn plan both require it green over a plain +// socketpair() with two threads, running the WHOLE InProcessTransportTest +// suite, BEFORE a child is ever forked. Same discipline as SessionRings.h's +// "the attach half is the half P6 replaces, so exercise it now" - a transport +// first exercised on the day the process appears is a transport whose bugs all +// arrive at once. +// +// WHAT IT IS NOT. It is not the data plane. The hot path bypasses ITransport +// entirely (ITransport.h) and crosses through ILink, which is a separate seam +// with a separate implementation. This class carries the handshake, surface +// ops, resync, aux requests and fatals - the rare, variable-length, +// must-evolve traffic. +// +// THREE THINGS IT DOES NOT REINVENT, because they are already in the tree and +// already tested: +// Framing.h - AppendFrame writes [u32 'MGLF'][u32 len][payload]; FrameReader +// is a real byte-stream reassembler with the BUFFER_TOO_SMALL +// keep-the-message semantics ITransport requires. A socket +// hands you arbitrary fragments, which is precisely what +// FrameReader was written for and what InProcessTransport's +// message queue never needed. +// FdPassing.h - SCM_RIGHTS over a DEDICATED SOCK_DGRAM aux socket, so an fd +// offer is one datagram and cannot be half-consumed. +// Doorbell.h - SocketDoorbell already exists, is unit-tested, and has NO +// production construction site until now. It was written for +// this and is priced as new work here, not as already paid for. +// +// LIFETIME. The transport owns its stream fd and its aux fd and closes both in +// Shutdown/dtor. Shutdown is idempotent and tears down BOTH directions, which +// is what ITransport promises and what a socket does anyway: after either end +// calls it, the peer's reads return 0 and its sends fail. + +#pragma once + +#include "Framing.h" +#include "ITransport.h" + +#include +#include +#include +#include + +namespace MobileGL::MG_Remote::Transport { + + class SocketTransport final : public ITransport { + public: + // Creates a connected pair and hands back one transport per end, with + // the aux (fd-passing) socketpair already made. POSIX only; on a + // platform without SCM_RIGHTS this refuses BY NAME rather than handing + // back a transport whose ShareFd would surprise its caller later. + static MobileGLResult CreatePair(std::unique_ptr& outClient, + std::unique_ptr& outServer); + + // Adopts an already-connected stream fd - what `sm` uses on each side of + // the fork, where fd 3 is the stream and fd 4 is the aux socket. + // `auxFd` may be -1: the transport then answers MOBILEGL_ERR_UNSUPPORTED + // to ShareFd/ReceiveFd, which is the honest answer for a link that + // cannot pass descriptors (CONTRACT-P6 rule G: a transport declares what + // it supports and may not invent a third answer). + SocketTransport(int streamFd, int auxFd, TransportRole role); + ~SocketTransport() override; + + MobileGLResult SendFrame(MobileGLByteSpan bytes) override; + MobileGLResult ReceiveFrame(MobileGLMutableByteSpan buffer, std::uint64_t* outSize, + std::uint32_t timeoutMs) override; + std::uint64_t PeekFrameSize() override; + + MobileGLResult ShareFd(int fd, MobileGLByteSpan sideband) override; + MobileGLResult ReceiveFd(int* outFd, MobileGLMutableByteSpan sideband, + std::uint64_t* outSidebandSize, std::uint32_t timeoutMs) override; + + void Shutdown() override; + TransportRole Role() const override { return m_role; } + + // For `sm`'s process discipline and for the arm-proof gate: the fds this + // transport owns, so a test can assert the child inherited exactly these. + int StreamFd() const { return m_streamFd; } + int AuxFd() const { return m_auxFd; } + + private: + // Pulls whatever the socket has into the reassembler. Returns + // TRANSPORT_CLOSED on a clean EOF with nothing buffered. + MobileGLResult PumpOnce(std::uint32_t timeoutMs); + + int m_streamFd = -1; + int m_auxFd = -1; + TransportRole m_role = TransportRole::Client; + + // ITransport's threading rule: callers serialise sends, and receives may + // run on one dedicated reader thread concurrently with those sends. So + // the two directions get one mutex each rather than one shared - a + // single lock would make a blocked receive stall every send, which is + // exactly the wedge the control plane must not have. + std::mutex m_sendMutex; + std::mutex m_recvMutex; + + FrameReader m_reader; + bool m_closed = false; + bool m_failed = false; // framing violated: latched, never recovers + }; + +} // namespace MobileGL::MG_Remote::Transport diff --git a/MobileGL/MG_Test/Wire/CMakeLists.txt b/MobileGL/MG_Test/Wire/CMakeLists.txt index 359c04fe..f508f08d 100644 --- a/MobileGL/MG_Test/Wire/CMakeLists.txt +++ b/MobileGL/MG_Test/Wire/CMakeLists.txt @@ -17,6 +17,10 @@ set(MOBILEGL_WIRE_TESTS if (NOT WIN32) # SCM_RIGHTS and fork(): POSIX only. list(APPEND MOBILEGL_WIRE_TESTS FdPassingTest) + # P6 so: the SAME nine ITransport-contract cases InProcessTransportTest runs, over a real + # socketpair with real threads, plus the three a stream has and a message queue does not. + # POSIX for the same reason as FdPassingTest - and because CONTRACT-P6 §2.6 lands POSIX only. + list(APPEND MOBILEGL_WIRE_TESTS SocketTransportTest) endif() include(GoogleTest) diff --git a/MobileGL/MG_Test/Wire/SocketTransportTest.cpp b/MobileGL/MG_Test/Wire/SocketTransportTest.cpp new file mode 100644 index 00000000..f5a7ff15 --- /dev/null +++ b/MobileGL/MG_Test/Wire/SocketTransportTest.cpp @@ -0,0 +1,390 @@ +// MobileGL - MobileGL/MG_Test/Wire/SocketTransportTest.cpp +// Copyright (c) 2025-2026 MobileGL-Dev +// Licensed under the GNU Lesser General Public License v3.0: +// https://www.gnu.org/licenses/gpl-3.0.txt +// https://www.gnu.org/licenses/lgpl-3.0.txt +// SPDX-License-Identifier: LGPL-3.0-only +// End of Source File Header + +// The SAME nine ITransport-contract cases InProcessTransportTest runs, driven +// over a real socketpair() with real threads. Package `so`. +// +// THIS FILE IS THE PACKAGE'S WHOLE POINT. CONTRACT-P6.md §8 and the spawn plan +// both require SocketTransport green over a plain socketpair BEFORE a second +// process exists, for the reason SessionRings.h gives about the attach half: +// a transport first exercised on the day the child appears is a transport +// whose bugs all arrive at once, on the hardest day to debug them. +// +// WHY THE BODIES ARE COPIED RATHER THAN SHARED. The two suites assert the same +// contract but not through the same handle: InProcessTransport hands out +// condvar doorbells its tests reach into, SocketTransport does not have them +// (the spawn design pairs it with SocketDoorbell, which FdPassingTest already +// covers). Folding both into one parameterised fixture would rename the +// sixteen existing cases, and G14 says test names only ever grow. Nine new +// names, zero removed. +// +// The seven Doorbell* cases in InProcessTransportTest are deliberately NOT +// here: they exercise CondVarDoorbell through a transport that happens to own +// one, not ITransport. SocketDoorbell's equivalents live in FdPassingTest. +// +// WHAT A SOCKET MAKES REAL THAT A MESSAGE QUEUE DID NOT. InProcessTransport +// hands the peer whole messages by construction; a stream hands you arbitrary +// fragments, so every one of these cases now also exercises Framing.h's +// reassembler - which is why `PreservesOrder` sends 64 messages without +// reading, forcing coalescing in the socket buffer, and why +// `BufferTooSmallKeepsTheMessage` matters more here: the message it must keep +// is one this side already reassembled out of bytes nobody else can re-deliver. + +#include // kWaitForever +#include +#include + +#include + +#include +#include +#include +#include +#include +#include +#include + +#if !defined(_WIN32) +#include +#include +#endif + +using namespace MobileGL::MG_Remote::Transport; + +#if !defined(_WIN32) + +namespace { + + MobileGLByteSpan Span(const std::string& text) { + return MobileGLByteSpan{text.data(), text.size()}; + } + + std::string Receive(ITransport& transport, std::uint32_t timeoutMs = 1000) { + std::vector buffer(4096); + std::uint64_t size = 0; + MobileGLMutableByteSpan span{buffer.data(), buffer.size()}; + const MobileGLResult result = transport.ReceiveFrame(span, &size, timeoutMs); + if (result != MOBILEGL_OK) { + return std::string("(result)) + ">"; + } + return std::string(reinterpret_cast(buffer.data()), + static_cast(size)); + } + + struct Pair { + std::unique_ptr client; + std::unique_ptr server; + }; + + Pair MakePair() { + Pair pair; + const MobileGLResult result = SocketTransport::CreatePair(pair.client, pair.server); + EXPECT_EQ(result, MOBILEGL_OK); + return pair; + } + +} // namespace + +TEST(SocketTransportTest, CarriesFramesInBothDirections) { + Pair pair = MakePair(); + ASSERT_TRUE(pair.client && pair.server); + EXPECT_EQ(pair.client->Role(), TransportRole::Client); + EXPECT_EQ(pair.server->Role(), TransportRole::Server); + + const std::string hello = "Hello{abiMajor=1}"; + const std::string welcome = "Welcome{serverPid=42}"; + ASSERT_EQ(pair.client->SendFrame(Span(hello)), MOBILEGL_OK); + // PeekFrameSize pumps, so it is allowed to need the bytes to have arrived; + // on a socketpair a completed send is already in the peer's buffer. + EXPECT_EQ(pair.server->PeekFrameSize(), hello.size()); + // A message goes to the PEER, never back to the sender. + EXPECT_EQ(pair.client->PeekFrameSize(), 0u); + EXPECT_EQ(Receive(*pair.server), hello); + + ASSERT_EQ(pair.server->SendFrame(Span(welcome)), MOBILEGL_OK); + EXPECT_EQ(Receive(*pair.client), welcome); +} + +TEST(SocketTransportTest, PreservesOrder) { + Pair pair = MakePair(); + + // 64 unread messages coalesce in the socket buffer, so the reassembler has + // to find every boundary itself. That is the case InProcessTransport could + // not have: its queue kept the boundaries for it. + for (int i = 0; i < 64; ++i) { + const std::string message = "msg-" + std::to_string(i); + ASSERT_EQ(pair.client->SendFrame(Span(message)), MOBILEGL_OK); + } + for (int i = 0; i < 64; ++i) { + EXPECT_EQ(Receive(*pair.server), "msg-" + std::to_string(i)); + } +} + +TEST(SocketTransportTest, BufferTooSmallKeepsTheMessage) { + Pair pair = MakePair(); + + const std::string message(300, 'x'); + ASSERT_EQ(pair.client->SendFrame(Span(message)), MOBILEGL_OK); + + std::vector small(16); + std::uint64_t required = 0; + MobileGLMutableByteSpan smallSpan{small.data(), small.size()}; + // Zero timeout: the bytes may still be in flight, so give the reassembler a + // real chance first. PeekFrameSize is the documented way to do that. + while (pair.server->PeekFrameSize() == 0) { + std::this_thread::yield(); + } + EXPECT_EQ(pair.server->ReceiveFrame(smallSpan, &required, 0), MOBILEGL_ERR_BUFFER_TOO_SMALL); + EXPECT_EQ(required, message.size()); + // Still there. On a stream this is the load-bearing half: the bytes have + // already been consumed from the socket, so if the reassembler dropped the + // message nothing could re-deliver it and the stream would be wedged. + EXPECT_EQ(pair.server->PeekFrameSize(), message.size()); + EXPECT_EQ(Receive(*pair.server), message); +} + +TEST(SocketTransportTest, PollAndTimeoutDoNotBlockForever) { + Pair pair = MakePair(); + + std::vector buffer(64); + std::uint64_t size = 0; + MobileGLMutableByteSpan span{buffer.data(), buffer.size()}; + EXPECT_EQ(pair.server->ReceiveFrame(span, &size, 0), MOBILEGL_ERR_TIMEOUT); + + const auto start = std::chrono::steady_clock::now(); + EXPECT_EQ(pair.server->ReceiveFrame(span, &size, 30), MOBILEGL_ERR_TIMEOUT); + EXPECT_GE(std::chrono::duration_cast( + std::chrono::steady_clock::now() - start) + .count(), + 20); +} + +TEST(SocketTransportTest, ShutdownDrainsBeforeItCloses) { + Pair pair = MakePair(); + + const std::string last = "Fatal{code=DeviceLost}"; + ASSERT_EQ(pair.client->SendFrame(Span(last)), MOBILEGL_OK); + pair.client->Shutdown(); + + // A peer that shuts down right after sending must not lose its last + // message - that is usually the one that says why it is going away. On a + // socket the bytes are already in the peer's receive buffer when close() + // runs, and the peer must reassemble them before it reports EOF. + EXPECT_EQ(Receive(*pair.server), last); + + std::vector buffer(64); + std::uint64_t size = 0; + MobileGLMutableByteSpan span{buffer.data(), buffer.size()}; + EXPECT_EQ(pair.server->ReceiveFrame(span, &size, 100), MOBILEGL_ERR_TRANSPORT_CLOSED); + EXPECT_EQ(pair.server->SendFrame(Span(last)), MOBILEGL_ERR_TRANSPORT_CLOSED); +} + +TEST(SocketTransportTest, BlockedReceiverWakesOnSendAndOnShutdown) { + Pair pair = MakePair(); + + std::atomic got{false}; + std::thread reader([&] { got.store(Receive(*pair.server, kWaitForever) == "wake"); }); + std::this_thread::sleep_for(std::chrono::milliseconds(20)); + ASSERT_EQ(pair.client->SendFrame(Span(std::string("wake"))), MOBILEGL_OK); + reader.join(); + EXPECT_TRUE(got.load()); + + std::thread closer([&] { + std::this_thread::sleep_for(std::chrono::milliseconds(20)); + pair.client->Shutdown(); + }); + std::vector buffer(64); + std::uint64_t size = 0; + MobileGLMutableByteSpan span{buffer.data(), buffer.size()}; + // The wake is a real POLLHUP here, not a condvar notify: this is the + // mechanism Doorbell.h:390-398 chose a stream socket for, and the reason + // death is an event rather than a hang. + EXPECT_EQ(pair.server->ReceiveFrame(span, &size, kWaitForever), MOBILEGL_ERR_TRANSPORT_CLOSED); + closer.join(); +} + +TEST(SocketTransportTest, RefusesAPayloadOverTheFrameCap) { + Pair pair = MakePair(); + + // Not allocated: the cap is checked before the bytes are touched, and it is + // the SAME cap InProcessTransport enforces, so nothing passes CI on inproc + // and then fails after the switch to spawn. + const std::uint8_t dummy = 0; + MobileGLByteSpan huge{&dummy, 64ull * 1024 * 1024 + 1}; + EXPECT_EQ(pair.client->SendFrame(huge), MOBILEGL_ERR_INVALID_ARGUMENT); +} + +TEST(SocketTransportTest, HandsOverADescriptorAndItsSideband) { + Pair pair = MakePair(); + + int pipeFds[2] = {-1, -1}; + ASSERT_EQ(::pipe(pipeFds), 0); + + const std::string sideband = "SegmentRef{id=1,kind=Cmd}"; + ASSERT_EQ(pair.client->ShareFd(pipeFds[0], Span(sideband)), MOBILEGL_OK); + + // A short sideband buffer is refused before anything is consumed, so the + // descriptor is never dropped. This is the real SCM_RIGHTS path, not an + // in-process hand-off: the fd the peer gets is renumbered by the kernel. + std::vector small(8); + int fd = -1; + std::uint64_t required = 0; + MobileGLMutableByteSpan smallSpan{small.data(), small.size()}; + EXPECT_EQ(pair.server->ReceiveFd(&fd, smallSpan, &required, 0), MOBILEGL_ERR_BUFFER_TOO_SMALL); + EXPECT_EQ(required, FdPassing::kMaxSidebandBytes); + EXPECT_EQ(fd, -1); + + std::vector big(FdPassing::kMaxSidebandBytes); + std::uint64_t sidebandSize = 0; + MobileGLMutableByteSpan bigSpan{big.data(), big.size()}; + ASSERT_EQ(pair.server->ReceiveFd(&fd, bigSpan, &sidebandSize, 100), MOBILEGL_OK); + ASSERT_GE(fd, 0); + EXPECT_EQ(std::string(reinterpret_cast(big.data()), + static_cast(sidebandSize)), + sideband); + + // Same open file description, independent descriptor - which is exactly + // what the four segments will ride on in `sm`. + const char payload[] = "bytes"; + ASSERT_EQ(::write(pipeFds[1], payload, sizeof(payload)), static_cast(sizeof(payload))); + char readBack[sizeof(payload)] = {}; + ASSERT_EQ(::read(fd, readBack, sizeof(readBack)), static_cast(sizeof(payload))); + EXPECT_STREQ(readBack, payload); + + ::close(fd); + ::close(pipeFds[0]); + ::close(pipeFds[1]); +} + +TEST(SocketTransportTest, AFrameWakeupIsNotEatenByAWaiterOnDescriptors) { + Pair pair = MakePair(); + + // Two readers on the SAME endpoint, blocked on two different predicates. + // The socket transport keeps them on two different descriptors - the stream + // and the aux - so neither can eat the other's wakeup by construction. That + // is a stronger property than the inproc transport's, and asserting it here + // is what would catch a future version that folded them onto one fd. + std::atomic fdWaiterStarted{false}; + std::thread fdWaiter([&] { + std::vector 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 its own deadline. NOT on + // Shutdown, the way the inproc suite's does - closing an fd out from + // under a thread already blocked in poll() does not reliably wake it, + // and a test that depended on that would be asserting a platform + // accident. The deadline is short because all this waiter has to do is + // BE BLOCKED while the frame below arrives. + const MobileGLResult result = pair.client->ReceiveFd(&fd, span, &size, 600); + EXPECT_NE(result, MOBILEGL_OK); + EXPECT_EQ(fd, -1); + }); + while (!fdWaiterStarted.load()) { + std::this_thread::yield(); + } + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + + std::atomic frameWaiterStarted{false}; + std::string got; + std::thread frameWaiter([&] { + frameWaiterStarted.store(true); + got = Receive(*pair.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(pair.server->SendFrame(Span(message)), MOBILEGL_OK); + frameWaiter.join(); + const auto elapsedMs = std::chrono::duration_cast( + 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); + + fdWaiter.join(); +} + +// ---- cases a socket has and a message queue does not ------------------------ + +TEST(SocketTransportTest, ReassemblesAMessageSplitAcrossReads) { + Pair pair = MakePair(); + + // 256 KiB is comfortably past a default socketpair buffer, so the sender + // blocks and the payload provably crosses in several reads. If PumpOnce + // returned a partial frame as a whole one, or if it dropped the tail, this + // is where it shows. + const std::string big(256 * 1024, 'q'); + std::thread sender([&] { EXPECT_EQ(pair.client->SendFrame(Span(big)), MOBILEGL_OK); }); + + std::vector buffer(big.size()); + std::uint64_t size = 0; + MobileGLMutableByteSpan span{buffer.data(), buffer.size()}; + ASSERT_EQ(pair.server->ReceiveFrame(span, &size, kWaitForever), MOBILEGL_OK); + sender.join(); + + EXPECT_EQ(size, big.size()); + EXPECT_EQ(std::string(reinterpret_cast(buffer.data()), + static_cast(size)), + big); +} + +TEST(SocketTransportTest, AGarbageStreamIsLatchedFailedAndNeverRecovers) { + Pair pair = MakePair(); + + // Write bytes that are not a frame directly onto the peer's stream fd, + // bypassing SendFrame. A corrupt control stream must latch, by name, and + // must not be retry-able past: the alternative is a reader that resyncs + // onto a byte boundary it invented, which is how one bad frame becomes a + // silently misparsed session. + const std::uint8_t garbage[16] = {0xDE, 0xAD, 0xBE, 0xEF, 0, 0, 0, 0, + 0xFF, 0xFF, 0xFF, 0xFF, 1, 2, 3, 4}; + ASSERT_EQ(::write(pair.client->StreamFd(), garbage, sizeof(garbage)), + static_cast(sizeof(garbage))); + + std::vector buffer(64); + std::uint64_t size = 0; + MobileGLMutableByteSpan span{buffer.data(), buffer.size()}; + EXPECT_EQ(pair.server->ReceiveFrame(span, &size, 200), MOBILEGL_ERR_PROTOCOL_MISMATCH); + // Latched: a second call says the same thing rather than resyncing. + EXPECT_EQ(pair.server->ReceiveFrame(span, &size, 200), MOBILEGL_ERR_PROTOCOL_MISMATCH); + EXPECT_EQ(pair.server->PeekFrameSize(), 0u); +} + +TEST(SocketTransportTest, AdoptsAnAlreadyConnectedFdAndRefusesFdPassingWithoutAnAux) { + // What `sm` does on each side of the fork: adopt fd 3, and - for a link + // that has no descriptor channel - answer UNSUPPORTED to ShareFd rather + // than pretending. Rule G: a transport declares what it supports. + int fds[2] = {-1, -1}; + ASSERT_EQ(::socketpair(AF_UNIX, SOCK_STREAM, 0, fds), 0); + + SocketTransport a(fds[0], -1, TransportRole::Client); + SocketTransport b(fds[1], -1, TransportRole::Server); + + const std::string message = "Hello{adopted}"; + ASSERT_EQ(a.SendFrame(Span(message)), MOBILEGL_OK); + EXPECT_EQ(Receive(b), message); + + const std::uint8_t dummy = 0; + EXPECT_EQ(a.ShareFd(1, MobileGLByteSpan{&dummy, 1}), MOBILEGL_ERR_UNSUPPORTED); + int fd = -1; + std::uint64_t size = 0; + std::vector sideband(FdPassing::kMaxSidebandBytes); + MobileGLMutableByteSpan span{sideband.data(), sideband.size()}; + EXPECT_EQ(b.ReceiveFd(&fd, span, &size, 0), MOBILEGL_ERR_UNSUPPORTED); + EXPECT_EQ(fd, -1); +} + +#endif // !_WIN32