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