[MG_Remote, ci] (P7): a bound session closes its data-connection source, so a later DataBind is refused by name instead of queued into a child that stopped reading, and the single-session listener closes with it

- ServerMain.cpp: RunSession takes the descriptor its data-connection source reads from (the child's hand-off end under --serve, the listener itself without it) and closes it the moment Accept returns; the source object goes with it
- the supervisor's hand-off send is MSG_DONTWAIT (FdPassing::SendFd dontWait), so a peer that stopped reading answers at once; "peer gone" on a datagram pair (ECONNREFUSED, ENOTCONN) is an answer the caller names, not a logged fault
- before: every DataBind after the bind sat in the child's receive queue with no reply, its TCP connection held open until the session ended, and past the queue limit (net.unix.max_dgram_qlen 10 on the phone; ~288 datagrams of sender buffer on this host) the supervisor blocked in sendmsg for the rest of the session - no accepts, no Busy, no reaping
- before, without --serve: the listener stayed open after the data connection bound, so a second client or a retry completed a TCP handshake nobody would accept and waited out its 20 s connect budget instead of the ECONNREFUSED it got before PH-7 (4)
- tcp_supervisor_smoke.py: data_bind_after_bound (12 extra DataBinds on a bound session each get Refuse{Authentication} "data connection could not be handed to the live session", and a fresh control connection is still Busy) and single_session_listener (a connect after `ready` is ECONNREFUSED); connect_control retries ECONNREFUSED at startup for pair() and held_session() alike
- red-once at da313010: smoke TimeoutError on the first extra's receive; probe with qlen+2 fire-and-forget DataBinds: 288 queued silently, then the supervisor stopped accepting (fresh control: connect timed out) while still alive; single-session connect after ready succeeded

MobileGL/MG_Remote/Server/ServerMain.cpp

@@ -74,6 +74,18 @@ void RefuseBusy(SocketTransport& transport, const char* detail) {
 // the child that minted it. This replaces AcceptPair's rule that the second connection to
 // arrive inside the window IS the data plane: over a Windows `adb forward` the two arrived
 // reordered, the data connection was read as control, and every child exited 67.
+//
+// THE HAND-OFF HAS ONE READER, AND IT STOPS READING (F fix round, review of slices 1-3). The
+// child reads the hand-off only inside BindDataConnection; once its data connection is bound
+// it never reads it again. The first shape of this function kept forwarding every later
+// DataBind regardless: each descriptor sat in the child's receive queue with no reply, its TCP
+// connection was held open until the session ended, and once the queue reached
+// net.unix.max_dgram_qlen (10 on the phone's kernel, 512 on a desktop's) the supervisor BLOCKED
+// in sendmsg for the rest of the session - no accepts, no Busy refusals, no reaping. Two halves
+// close it: the child closes its end of the pair the moment Accept returns (RunSession), so the
+// send fails with "peer gone" and the refusal below fires by name; and the send is MSG_DONTWAIT,
+// so a queue that is full for any other reason refuses too rather than stalling the one process
+// that answers the port.
 constexpr std::uint32_t kFirstFrameWaitMs = 2000;
 constexpr std::uint64_t kFirstFrameMaxBytes = 1024 * 1024;
 
@@ -83,7 +95,8 @@ void RouteWhileBusy(SocketTransport& connection, int handoff) {
     const auto read = SocketTransport::ReceiveOneFrame(fd, kFirstFrameWaitMs, kFirstFrameMaxBytes, &frame);
     std::uint8_t nonce[kDataNonceBytes] = {};
     if (read == MOBILEGL_OK && DecodeDataBind(frame, nonce)) {
-        if (handoff >= 0 && FdPassing::SendFd(handoff, fd, MobileGLByteSpan{nonce, sizeof(nonce)}) == MOBILEGL_OK) {
+        if (handoff >= 0 &&
+            FdPassing::SendFd(handoff, fd, MobileGLByteSpan{nonce, sizeof(nonce)}, /*dontWait=*/true) == MOBILEGL_OK) {
             // The child holds its own descriptor for this connection now; ours goes without a
             // shutdown(2), which would disconnect the child's too.
             connection.CloseLocalCopy();
@@ -172,9 +185,11 @@ void SendLogAck(void* pointer) {
 
 // Called only after fork: the supervisor never creates a backend or EGL context.
 // `dataSource` is set on TCP (PH-7 (4)) and empty on a unix endpoint, whose second connection
-// is the SCM_RIGHTS socket AcceptPair already paired.
+// is the SCM_RIGHTS socket AcceptPair already paired. `sourceFd` is the descriptor that source
+// reads from - the child's end of the hand-off socketpair under --serve, the listener itself in
+// the single-session shape - and -1 when there is none; it is closed the moment Accept returns.
 [[noreturn]] void RunSession(std::unique_ptr<SocketTransport> control,
-                             Server::ServerSession::DataConnectionSource dataSource) {
+                             Server::ServerSession::DataConnectionSource dataSource, int sourceFd) {
     const int selfPid = static_cast<int>(::getpid());
     const bool tcp = control->IsTcp();
     std::vector<std::uint8_t> helloFrame;
@@ -237,6 +252,21 @@ void SendLogAck(void* pointer) {
     }
     if (tcp) session.SetDataConnectionSource(std::move(dataSource));
     const auto accepted = session.Accept(*control, &helloFrame);
+    // THE SOURCE IS DONE WHEN ACCEPT IS, whichever way Accept went (F fix round). Under --serve
+    // this is the child's end of the hand-off pair: closing it turns every later DataBind the
+    // supervisor would have queued here into a send failure the supervisor refuses by name on
+    // that connection ("data connection could not be handed to the live session"), instead of a
+    // descriptor nobody will read holding a peer's connection open and, past max_dgram_qlen,
+    // holding the supervisor itself. In the single-session shape it is the listener: kept open
+    // only until the data connection arrived on it, and closed here so a second client or a
+    // retry is refused at connect (ECONNREFUSED, as before PH-7 (4)) rather than completing a
+    // TCP handshake this process would never accept and waiting out its whole connect budget.
+    // The source object goes with it, so no later call can poll a descriptor number the kernel
+    // may have reused.
+    if (sourceFd >= 0) {
+        ::close(sourceFd);
+        session.SetDataConnectionSource({});
+    }
     if (accepted != MOBILEGL_OK) {
         WireLogError("MG_Remote server: pid=%d Accept failed (rc=%d)", selfPid, static_cast<int>(accepted));
         std::fflush(nullptr);
@@ -498,13 +528,14 @@ extern "C" __attribute__((visibility("default"))) int mobilegl_server_main(int a
         if (accepted != MOBILEGL_OK) { LogSupervisorSummary(sessionsFaulted); ::close(listener); return 73; }
         if (!serve) {
             if (tcpEndpoint) {
-                // The listener stays open: this session's data connection has yet to arrive on it.
+                // The listener stays open until this session's data connection has arrived on
+                // it; RunSession closes it the moment Accept returns.
                 const int controlFd = control->StreamFd();
-                RunSession(std::move(control), ListenerSource(listener, controlFd));
+                RunSession(std::move(control), ListenerSource(listener, controlFd), listener);
             }
             ::close(listener);
             if (endpoint[0] != '@') ::unlink(endpoint.c_str());
-            RunSession(std::move(control), {});
+            RunSession(std::move(control), {}, -1);
         }
         reap();
         // exit_group closes files just before waitpid can observe the exit.
@@ -531,7 +562,8 @@ extern "C" __attribute__((visibility("default"))) int mobilegl_server_main(int a
             if (pair[0] >= 0) ::close(pair[0]);
             const int controlFd = control->StreamFd();
             RunSession(std::move(control), tcpEndpoint ? HandoffSource(pair[1], controlFd)
-                                                       : Server::ServerSession::DataConnectionSource{});
+                                                       : Server::ServerSession::DataConnectionSource{},
+                       pair[1]);
         }
         if (child < 0) {
             if (pair[0] >= 0) ::close(pair[0]);

MobileGL/MG_Remote/Transport/FdPassing.cpp

@@ -38,7 +38,7 @@ namespace MobileGL::MG_Remote::Transport::FdPassing {
 
     MobileGLResult CreateSocketPair(int[2]) { return MOBILEGL_ERR_UNSUPPORTED; }
 
-    MobileGLResult SendFd(int, int, MobileGLByteSpan) { return MOBILEGL_ERR_UNSUPPORTED; }
+    MobileGLResult SendFd(int, int, MobileGLByteSpan, bool) { return MOBILEGL_ERR_UNSUPPORTED; }
 
     MobileGLResult ReceiveFd(int, int*, MobileGLMutableByteSpan, std::uint64_t*, std::uint32_t) {
         return MOBILEGL_ERR_UNSUPPORTED;
@@ -105,7 +105,7 @@ namespace MobileGL::MG_Remote::Transport::FdPassing {
         return MOBILEGL_OK;
     }
 
-    MobileGLResult SendFd(int socket, int fd, MobileGLByteSpan sideband) {
+    MobileGLResult SendFd(int socket, int fd, MobileGLByteSpan sideband, bool dontWait) {
         if (socket < 0 || fd < 0) {
             return MOBILEGL_ERR_INVALID_ARGUMENT;
         }
@@ -148,8 +148,9 @@ namespace MobileGL::MG_Remote::Transport::FdPassing {
         cmsg->cmsg_len = CMSG_LEN(sizeof(int));
         std::memcpy(CMSG_DATA(cmsg), &fd, sizeof(fd));
 
+        const int flags = MSG_NOSIGNAL | (dontWait ? MSG_DONTWAIT : 0);
         for (;;) {
-            const ssize_t sent = ::sendmsg(socket, &msg, MSG_NOSIGNAL);
+            const ssize_t sent = ::sendmsg(socket, &msg, flags);
             if (sent >= 0) {
                 if (static_cast<std::size_t>(sent) != payloadSize) {
                     // A datagram socket sends all or nothing.
@@ -162,9 +163,18 @@ namespace MobileGL::MG_Remote::Transport::FdPassing {
             if (errno == EINTR) {
                 continue;
             }
-            if (errno == EPIPE || errno == ECONNRESET) {
+            // A connected AF_UNIX datagram socket whose peer closed answers ECONNREFUSED on the
+            // first send after the close and ENOTCONN on every later one; a stream peer answers
+            // EPIPE / ECONNRESET. All four are "the peer is gone", which is an answer and not a
+            // fault, so none of them is logged here - the caller names what it does about it.
+            if (errno == EPIPE || errno == ECONNRESET || errno == ECONNREFUSED || errno == ENOTCONN) {
                 return MOBILEGL_ERR_TRANSPORT_CLOSED;
             }
+            if (dontWait && (errno == EAGAIN || errno == EWOULDBLOCK)) {
+                MGLOG_E("MG_Remote fd passing: the peer's receive queue is full and it is not "
+                        "reading; the descriptor was not queued");
+                return MOBILEGL_ERR_TIMEOUT;
+            }
             MGLOG_E("MG_Remote fd passing: sendmsg failed (errno=%d)", errno);
             return MOBILEGL_ERR_TRANSPORT_CLOSED;
         }

MobileGL/MG_Remote/Transport/FdPassing.h

@@ -48,7 +48,15 @@ namespace MobileGL::MG_Remote::Transport::FdPassing {
     // Sends `fd` with `sideband` attached. The caller keeps ownership of `fd`
     // (the peer gets its own descriptor for the same open file description).
     // sideband.size must be <= kMaxSidebandBytes.
-    MobileGLResult SendFd(int socket, int fd, MobileGLByteSpan sideband);
+    //
+    // `dontWait` sends with MSG_DONTWAIT: a receive queue the peer is not draining
+    // (net.unix.max_dgram_qlen datagrams, 10 on an Android kernel) answers
+    // MOBILEGL_ERR_TIMEOUT at once instead of blocking the sender until the peer
+    // reads. A peer that has closed its end answers MOBILEGL_ERR_TRANSPORT_CLOSED
+    // either way. The TCP supervisor's hand-off (ServerMain.cpp RouteWhileBusy)
+    // sends this way because the session child stops reading the hand-off once its
+    // data connection is bound, and a supervisor blocked in sendmsg serves nobody.
+    MobileGLResult SendFd(int socket, int fd, MobileGLByteSpan sideband, bool dontWait = false);
 
     // Receives one descriptor and its sideband bytes.
     //

scripts/ci/tcp_supervisor_smoke.py

@@ -222,13 +222,16 @@ def refused_listen(server, log, endpoint, token):
 
 
 @contextmanager
-def supervisor(server, log, same_build=False):
+def supervisor(server, log, same_build=False, serve=True):
+    """A supervisor on its own free loopback port. `serve=False` is the single-session shape
+    (no fork; the process IS the session and exits with it), which `single_session_listener`
+    below needs and nothing else does."""
     port = free_loopback_port()
     base = log.with_suffix('.library.log')
     env = supervisor_environment(base, kToken, same_build=same_build)
     with log.open('wb') as output:
-        child = subprocess.Popen([server, f'tcp://127.0.0.1:{port}', '--serve'], env=env,
-                                 stdout=output, stderr=output, start_new_session=True)
+        child = subprocess.Popen([server, f'tcp://127.0.0.1:{port}'] + (['--serve'] if serve else []),
+                                 env=env, stdout=output, stderr=output, start_new_session=True)
     try:
         # Log.cpp's RoleLogPath: the base name is a BASE, and the server role writes
         # `<stem>.server<ext>`. Yielded rather than recomputed at the call site, so the one
@@ -242,6 +245,20 @@ def supervisor(server, log, same_build=False):
         child.wait(timeout=10)
 
 
+def connect_control(port, seconds=5):
+    """A connection to the supervisor's port, retried on ECONNREFUSED for `seconds`: a supervisor
+    just started may not have reached listen() yet, and a refused connect before it has is
+    startup, not a finding."""
+    deadline = time.monotonic() + seconds
+    while True:
+        try:
+            return socket.create_connection(('127.0.0.1', port), timeout=3)
+        except ConnectionRefusedError:
+            if time.monotonic() >= deadline:
+                raise
+            time.sleep(.02)
+
+
 @contextmanager
 def held_session(port, schema, flatbuffers, message):
     """A welcomed session whose sockets STAY OPEN, so its child is still alive to be killed.
@@ -258,7 +275,7 @@ def held_session(port, schema, flatbuffers, message):
     data = None
     reply = None
     for _ in range(30):
-        control = socket.create_connection(('127.0.0.1', port), timeout=3)
+        control = connect_control(port)
         control.sendall(message)
         reply = receive(control, schema)
         if reply.get('code') != 7:
@@ -305,15 +322,7 @@ def pair(port):
     two connections paired by arrival order, and this opened both. The data connection is now
     opened after Welcome with the nonce it carries (open_data), so a control that stops at the
     handshake - every one but held_session's - never opens one."""
-    deadline = time.monotonic() + 5
-    while True:
-        try:
-            control = socket.create_connection(('127.0.0.1', port), timeout=3)
-            break
-        except ConnectionRefusedError:
-            if time.monotonic() >= deadline:
-                raise
-            time.sleep(.02)
+    control = connect_control(port)
     try:
         yield control
     finally:
@@ -543,10 +552,76 @@ def main():
                 assert peak < ask, f'session pid={welcome["welcome"]} reached VmPeak={peak} for a 64 GiB ask'
             evidence['hello_asks_64_gib'] = {'asked': ask, 'terms': terms, 'childVmPeakBytes': peak}
 
+            # ---- F fix round: A BOUND SESSION'S SUPERVISOR REFUSES EVERY LATER DataBind BY NAME --
+            #
+            # The session child reads its hand-off socketpair only while binding; after that it
+            # never reads it again. Until this control the supervisor kept forwarding every later
+            # DataBind into it anyway: each descriptor sat in the child's receive queue with NO
+            # reply, its TCP connection was held open until the session ended, and once the queue
+            # reached net.unix.max_dgram_qlen (10 on the phone's kernel, 512 on this host's) the
+            # supervisor blocked in sendmsg for the rest of the session - no accepts, no Busy, no
+            # reaping. Now the child closes its end the moment Accept returns and the send is
+            # non-blocking, so each of these twelve is answered on its own connection with
+            # Refuse{Authentication} "data connection could not be handed to the live session",
+            # and a fresh control connection afterwards is still answered Busy. RED before the fix
+            # at the very first `receive`: a 3 s socket timeout, because nothing ever answered.
+            with held_session(port, schema, flatbuffers,
+                              hello(schema, flatbuffers, fingerprint=fingerprint)) as (control, welcome):
+                assert welcome.get('welcome', 0), welcome
+                # Bound, not merely welcomed: an extra DataBind that arrived while the child was
+                # still in BindDataConnection would be forwarded and refused as a nonce mismatch,
+                # which is the other control's line, not this one's.
+                ready = wait_for_log(serverLog, f'pid={welcome["welcome"]} transport=spawn role=server ready')
+                assert ready is not None, f'the held session never came up; see {serverLog}'
+                answered = []
+                for _ in range(12):
+                    extra = open_data(port, schema, flatbuffers, os.urandom(16))
+                    try:
+                        answered.append(receive(extra, schema))
+                    finally:
+                        extra.close()
+                for answer in answered:
+                    assert answer.get('code') == 4 and answer.get('detail') == (
+                        'data connection could not be handed to the live session'), answer
+                with pair(port) as fresh:
+                    fresh.sendall(hello(schema, flatbuffers, fingerprint=fingerprint))
+                    still_busy = receive(fresh, schema)
+                assert still_busy.get('code') == 7, still_busy
+            evidence['data_bind_after_bound'] = {'refused': len(answered), 'last': answered[-1],
+                                                 'fresh_control': still_busy}
+            assert process.poll() is None, 'the refused DataBinds killed the supervisor'
+
         with supervisor(args.server, args.out / 'same-build-supervisor.log', same_build=True) as (port, _, _log):
             strict = exchange(port, schema, hello(schema, flatbuffers, fingerprint=fingerprint))
             assert strict.get('code') == 3, strict
             evidence['same_build_required'] = strict
+
+        # ---- F fix round: THE SINGLE-SESSION SERVER CLOSES ITS LISTENER ONCE THE SESSION IS BOUND
+        #
+        # Without --serve the process is the session; it keeps the listener open only so the
+        # session's data connection can arrive on it (ListenerSource). Until this control nothing
+        # closed it afterwards, so the kernel went on completing TCP handshakes into a backlog no
+        # accept() would ever drain: a second client, or a retry, hung for its whole connect budget
+        # (kSpawnConnectTimeoutMs, 20 s) instead of the ECONNREFUSED it got before PH-7 (4). Now
+        # RunSession closes the listener the moment Accept returns, and a connect after `ready`
+        # is refused at once. RED before the fix: the connect below succeeds.
+        with supervisor(args.server, args.out / 'single-session.log', serve=False) as (port, process, serverLog):
+            with held_session(port, schema, flatbuffers,
+                              hello(schema, flatbuffers, fingerprint=fingerprint)) as (control, welcome):
+                assert welcome.get('welcome', 0), welcome
+                ready = wait_for_log(serverLog, f'pid={welcome["welcome"]} transport=spawn role=server ready')
+                assert ready is not None, f'the single session never came up; see {serverLog}'
+                try:
+                    second = socket.create_connection(('127.0.0.1', port), timeout=3)
+                except ConnectionRefusedError as refused:
+                    second = None
+                    at_connect = repr(refused)
+                else:
+                    second.close()
+                assert second is None, (
+                    'the single-session server still completed a TCP handshake after its data '
+                    'connection bound; the listener was not closed')
+            evidence['single_session_listener'] = {'second_connect': at_connect}
         (args.out / 'protocol-controls.json').write_text(json.dumps(evidence, indent=2))
         print(json.dumps(evidence, indent=2))
     return 0