[MG_Remote, MG_Test] (P12 review): a lost server window is answered before every ring record, and a stopping in-process server leaves a streaming client's queue unapplied

- Major: the window-lost release was picked up only between drain batches (PumpControlRequest), and a client streaming frames keeps the command ring from emptying for seconds - surfaceDestroyed waited out its 3 s bound on the UI thread, answered TimedOut, and the session kept presenting into the destroyed window. DrainRing now checks the request after its latch check, before every pop. ServerLoopLatchTest.ALostServerWindowIsReleasedMidBatchWhileTheClientStreams streams a 4 s batch under Detach.
- Stop while streaming: mobilegl_server_stop_inprocess ended the session through loop.Stop(), whose batch and final drain of a ring the client keeps refilling could outlast the bounded join (Fatal{ApplyThreadJoinTimeout} aborts the display Activity's process). RunSession now calls ServerLoop::AbandonQueuedRecords() first when the server is stopping: the drain stops after the record in hand and the exit path declines the rest, as a ReverseChannelForfeit does. ServerLoopTest.StoppingTheServerUnderAStreamingClientLeavesTheQueueAndJoinsPromptly.

MobileGL/MG_Remote/Server/ServerLoop.cpp

@@ -210,6 +210,7 @@ namespace MobileGL::MG_Remote::Server {
         }
         m_session = &session;
         m_stopRequested.store(false, std::memory_order_release);
+        m_abandonQueue.store(false, std::memory_order_release);
         // PH-6 fix round: the session's copy of the request, for ReserveEventOrBlock's wait. Lowered
         // with the loop's own, so a loop restarted on the same session does not forfeit at once.
         session.ClearApplyStopRequest();
@@ -506,13 +507,22 @@ namespace MobileGL::MG_Remote::Server {
         // a peer that keeps publishing could keep this final drain going as long as it liked,
         // which is exactly the unbounded hold the forfeit exists to end. The queue is left
         // unapplied; the rings die with the session.
+        //
+        // AND AFTER AbandonQueuedRecords (P12 review fix), for the forfeit's reason: the server is
+        // stopping under a client that may still be streaming, and draining what it keeps queuing
+        // is the unbounded hold Stop()'s bounded join turns into an abort.
         const Bool forfeited = session.ReverseChannelForfeited();
+        const Bool abandoned = m_abandonQueue.load(std::memory_order_acquire);
         PumpControlRequest();
         if (forfeited) {
             MGLOG_E("MG_Remote server: mgl-srv-apply stops on ReverseChannelForfeit - %llu reverse-channel "
                     "event(s) dropped, applied seq %llu; the records still queued are not applied",
                     static_cast<unsigned long long>(session.ForfeitDrops()),
                     static_cast<unsigned long long>(session.Consumer().AppliedSeq()));
+        } else if (abandoned) {
+            MGLOG_W("MG_Remote server: mgl-srv-apply stops because the server is stopping - applied seq %llu; "
+                    "the records still queued are not applied",
+                    static_cast<unsigned long long>(session.Consumer().AppliedSeq()));
         } else {
             DrainRing();
         }
@@ -712,12 +722,25 @@ namespace MobileGL::MG_Remote::Server {
         //
         // The idle poll's cost is unchanged: this load replaces the function-top check the first
         // version made, and the empty-ring answer is still it plus Pop's cmdHead load.
+        //
+        // P12 (D6) REVIEW FIX: A LOST SERVER WINDOW IS ANSWERED BEFORE EVERY POP TOO, not only by
+        // PumpControlRequest between batches. A batch ends when the ring is empty, and a client
+        // streaming frames keeps it from emptying for seconds: surfaceDestroyed (ServerDisplay::
+        // Detach) waited that long on the UI thread - past its 3 s bound, so it gave up and the
+        // session went on presenting into the destroyed window. Checked AFTER the latch check (the
+        // MECHANICS row in ph_latch_sites.py pins that one as the loop's first statement); the
+        // release latches ServerWindowLost, so `continue` takes the latch check's way out. One more
+        // acquire load per pop; the empty-ring answer pays it once.
         ServerSession& session = *m_session;
         Transport::SessionConsumer& consumer = session.Consumer();
         PipeApplier& applier = session.Applier();
         Uint64 applied = 0;
         for (;;) {
             if (SessionLatched()) break;
+            if (m_windowLostRequested.load(std::memory_order_acquire)) {
+                ReleaseLostServerWindow();
+                continue;
+            }
             bool corrupt = false;
             const bool popped = consumer.ApplyOne(
                 [this, &applier](const Transport::RingRecordView& record) {
@@ -756,6 +779,9 @@ namespace MobileGL::MG_Remote::Server {
             // loop-head check on the next iteration - ph_latch_sites.py's MECHANICS row pins that
             // check as the first statement of this loop.)
             if (session.ReverseChannelForfeited()) break;
+            // P12 review fix: and the record in hand is the last one when the server is stopping
+            // (AbandonQueuedRecords); same placement, same free empty-ring answer.
+            if (m_abandonQueue.load(std::memory_order_acquire)) break;
             // TEST-ONLY scheduling point BETWEEN two records: after this record's own checks and
             // before the next pop's latch check. A control-thread latch stored here is seen by that
             // check (the first version let it through); one stored after the check and before the
@@ -935,6 +961,8 @@ namespace MobileGL::MG_Remote::Server {
         m_exitCv.notify_all();
     }
 
+    void ServerLoop::AbandonQueuedRecords() { m_abandonQueue.store(true, std::memory_order_release); }
+
     void ServerLoop::Stop() {
         if (!m_thread.joinable()) {
             // Never started, or already stopped. The backend may still exist - the hook builds

MobileGL/MG_Remote/Server/ServerLoop.h

@@ -216,6 +216,15 @@ namespace MobileGL::MG_Remote::Server {
         // Kill the doorbell, join the thread (bounded), then destroy the private backend object
         // ON THAT THREAD before it exits. Blocking by contract - see the header note.
         void Stop();
+        // P12 review fix: THE NEXT Stop() LEAVES THE QUEUE UNAPPLIED. A Stop() normally lets the
+        // apply thread finish its batch and drain what is still queued - the client's own drain
+        // bounds that. A SERVER that is stopping (the in-process display server's
+        // mobilegl_server_stop_inprocess) has a client that may still be streaming, and a batch plus
+        // a final drain of a ring that client keeps refilling can outlast the bounded join
+        // (Fatal{ApplyThreadJoinTimeout} aborts the display Activity's process). Called before Stop():
+        // the drain stops after the record in hand and the exit path declines the rest, as a
+        // ReverseChannelForfeit does. Reset by Start().
+        void AbandonQueuedRecords();
 
         Bool Running() const;
 
@@ -486,6 +495,8 @@ namespace MobileGL::MG_Remote::Server {
         ServerSession* m_session = nullptr;
         std::atomic<Bool> m_running{false};
         std::atomic<Bool> m_stopRequested{false};
+        // P12 review fix: AbandonQueuedRecords() - read after every popped record and at the exit.
+        std::atomic<Bool> m_abandonQueue{false};
         std::thread m_thread;
         // The apply thread's identity used to live here as an atomic<std::thread::id>. It is
         // Detail::g_applyThreadKey now - see the block at the top of this header for why the

MobileGL/MG_Remote/Server/ServerMain.cpp

@@ -437,6 +437,7 @@ int RunSession(std::unique_ptr<SocketTransport> control, std::vector<std::uint8_
         if (backendBuilt) {
             // Stop() joins a running apply thread (it destroys the backend on it), or - never
             // started - drops the backend here; either way the display lease goes with it.
+            if (g_inProcessStop.load(std::memory_order_acquire)) loop.AbandonQueuedRecords();
             loop.Stop();
             session.SetBackend(nullptr);
         }
@@ -723,6 +724,11 @@ int RunSession(std::unique_ptr<SocketTransport> control, std::vector<std::uint8_
         }
     }
     loop.SetControlProgressSink(nullptr, nullptr);
+    // P12 review fix: A STOPPING SERVER DOES NOT DRAIN A STREAMING CLIENT. The session ends because
+    // the display server is going away, not because its client finished - and a client still
+    // streaming would keep the apply thread's final drain going past Stop()'s bounded join, whose
+    // Fatal{ApplyThreadJoinTimeout} aborts this (the display Activity's) process.
+    if (inProcess && g_inProcessStop.load(std::memory_order_acquire)) loop.AbandonQueuedRecords();
     loop.Stop();
     // Publish the final server window while log forwarding is still attached.
     MobileGL::MG_Util::PipeStats::Shutdown();

MobileGL/MG_Test/Wire/ServerLoopTest.cpp

@@ -3305,6 +3305,174 @@ TEST(ServerLoopLatchTest, ALostServerWindowIsReleasedOnTheApplyThreadBeforeDetac
            "latch; 16 = ServerWindowsLost() != 1; 32 = the apply thread stayed (64+ = setup)";
 }
 
+namespace {
+    // P12 review fix (major): THE LOST WINDOW IS ANSWERED IN THE MIDDLE OF A BATCH. A client streaming
+    // frames keeps the command ring from emptying, and the first version answered a window-lost request
+    // only between batches (PumpControlRequest). The between-records hook is that stream here: after
+    // the first record it runs the UI thread's surfaceDestroyed (Detach, on a thread of its own) and
+    // waits until the lost hook has been called; after every record it takes kStreamRecordMs, so the
+    // 200-record batch is a 4 s stream - longer than Detach's 3 s bound.
+    constexpr int kStreamRecords = 200;
+    constexpr int kStreamRecordMs = 20;
+    enum : int {
+        kStreamNotLeased = 1,       // the probe could not take the lease
+        kStreamDetach = 2,          // Detach did not answer ReleasedBySession
+        kStreamDetachSlow = 4,      // Detach took 1 s or more (the batch ran on under it)
+        kStreamDrainedOn = 8,       // more than two records were applied (the stream was not cut)
+        kStreamNotLatched = 16,     // the session did not latch ServerWindowLost
+        kStreamThreadStayed = 32,   // the apply thread was still running 3 s after the latch
+    };
+
+    std::atomic<int> g_streamRecords{0};
+    std::thread g_streamDetachThread;
+    std::atomic<int> g_streamDetachResult{-1};
+    std::atomic<long long> g_streamDetachMs{-1};
+
+    void StreamOneRecordAndLoseTheWindowAfterTheFirst() {
+        if (g_streamRecords.fetch_add(1, std::memory_order_acq_rel) == 0) {
+            g_streamDetachThread = std::thread([] {
+                const auto started = std::chrono::steady_clock::now();
+                const Server::ServerWindowDetach detached = Server::ServerDisplayInstance().Detach(3000);
+                g_streamDetachMs.store(std::chrono::duration_cast<std::chrono::milliseconds>(
+                                           std::chrono::steady_clock::now() - started)
+                                           .count(),
+                                       std::memory_order_release);
+                g_streamDetachResult.store(static_cast<int>(detached), std::memory_order_release);
+            });
+            // Detach clears the window and calls the lost hook under one lock hold, so once the window
+            // reads detached the request is published.
+            (void)PollUntil([] { return !Server::ServerDisplayInstance().Attached(); }, 1000);
+        }
+        std::this_thread::sleep_for(std::chrono::milliseconds(kStreamRecordMs));
+    }
+
+    [[noreturn]] void LoseTheServerWindowMidStreamAndExit() {
+        Remote::ArmSessionLatch();
+        ServerFixture fixture;
+        if (!fixture.Handshake() || !fixture.StartLoop()) ::_exit(64);
+        if (!fixture.WaitUntilTrulyParked()) ::_exit(65);
+        Server::ServerLoop& loop = Server::ServerLoopInstance();
+        Server::ServerDisplay& display = Server::ServerDisplayInstance();
+        display.Install(CountingWindowHooks::Hooks());
+        display.Attach(&g_fakeServerWindow, 64, 48);
+        ServerWindowProbe probe;
+        probe.width = 64;
+        probe.height = 48;
+        if (loop.RunProbeOnApplyThreadForTesting(&AcquireServerWindowOnTheApplyThread, &probe) != MOBILEGL_OK) ::_exit(66);
+        int failed = 0;
+        if (probe.rc != MOBILEGL_OK || probe.lease.window != &g_fakeServerWindow) failed |= kStreamNotLeased;
+        if (!fixture.WaitUntilTrulyParked()) ::_exit(67);
+        loop.SetBetweenRecordsHookForTesting(&StreamOneRecordAndLoseTheWindowAfterTheFirst);
+        // LEGAL draws under ONE publish (ALatchFromAnotherThread...'s records), so only the lost window
+        // can cut the batch short.
+        MG_Pipe::MGPDrawInfo info{};
+        info.Mode = 0x0004; // GL_TRIANGLES
+        info.InstanceCount = 1;
+        info.MinIndex = ~0u;
+        info.MaxIndex = ~0u;
+        info.NumDraws = 1;
+        const MG_Pipe::MGPDrawRange range{0, 3, 0};
+        Uint64 last = Codec::kInvalidSeq;
+        for (int i = 0; i < kStreamRecords; ++i) {
+            last = fixture.encoder.EncodeRecord(MG_Pipe::MGPWireOp::DrawVbo, &info, sizeof(info), &range,
+                                                sizeof(range));
+            if (last == Codec::kInvalidSeq) ::_exit(68);
+        }
+        fixture.encoder.Publish();
+        fixture.producer.PublishAndNotify(last);
+
+        if (!PollUntil([] { return g_streamDetachResult.load(std::memory_order_acquire) >= 0; }, 8000)) ::_exit(69);
+        if (g_streamDetachThread.joinable()) g_streamDetachThread.join();
+        if (g_streamDetachResult.load() != static_cast<int>(Server::ServerWindowDetach::ReleasedBySession))
+            failed |= kStreamDetach;
+        if (g_streamDetachMs.load() >= 1000) failed |= kStreamDetachSlow;
+        if (!PollUntil([&] { return !loop.Running(); }, 3000)) failed |= kStreamThreadStayed;
+        if (loop.DrainedRecords() > 2) failed |= kStreamDrainedOn;
+        if (!Remote::SessionLatched() || Remote::SessionLatchedFamily() != Remote::MGFatalFamily::ServerWindowLost)
+            failed |= kStreamNotLatched;
+        std::fprintf(stderr, "[stream] Detach answered %s in %lld ms after %llu of %d streamed records\n",
+                     Server::ServerWindowDetachName(
+                         static_cast<Server::ServerWindowDetach>(g_streamDetachResult.load())),
+                     g_streamDetachMs.load(), static_cast<unsigned long long>(loop.DrainedRecords()),
+                     kStreamRecords);
+        loop.SetBetweenRecordsHookForTesting(nullptr);
+        fixture.Stop();
+        ::_exit(failed);
+    }
+} // namespace
+
+// P12 review fix (major). Red with DrainRing's pre-pop lost-window check deleted: the whole 4 s batch
+// runs under the UI thread's Detach, which gives up at its 3 s bound and answers TimedOut (bits 2|4|8).
+TEST(ServerLoopLatchTest, ALostServerWindowIsReleasedMidBatchWhileTheClientStreams) {
+    EXPECT_EXIT(LoseTheServerWindowMidStreamAndExit(), ::testing::ExitedWithCode(0), ".*")
+        << "bits: 1 = the lease was not taken; 2 = Detach did not answer ReleasedBySession; 4 = Detach took "
+           "1 s or more; 8 = more than two records applied after the window went; 16 = no ServerWindowLost "
+           "latch; 32 = the apply thread stayed (64+ = setup)";
+}
+
+namespace {
+    // P12 review fix: A STOPPING SERVER UNDER A STREAMING CLIENT. 200 records at 40 ms each is an 8 s
+    // stream, longer than Stop()'s 5 s bounded join. The display server's stop marks the queue
+    // abandoned first; the join must then come back within a record or two.
+    std::atomic<int> g_stopStreamRecords{0};
+    void StreamSlowly() {
+        g_stopStreamRecords.fetch_add(1, std::memory_order_acq_rel);
+        std::this_thread::sleep_for(std::chrono::milliseconds(40));
+    }
+
+    enum : int {
+        kStopSlow = 1,      // Stop() took 1 s or more
+        kStopDrainedOn = 2, // more than three records were applied
+    };
+
+    [[noreturn]] void StopTheServerMidStreamAndExit() {
+        Remote::ArmSessionLatch();
+        ServerFixture fixture;
+        if (!fixture.Handshake() || !fixture.StartLoop()) ::_exit(64);
+        if (!fixture.WaitUntilTrulyParked()) ::_exit(65);
+        Server::ServerLoop& loop = Server::ServerLoopInstance();
+        loop.SetBetweenRecordsHookForTesting(&StreamSlowly);
+        MG_Pipe::MGPDrawInfo info{};
+        info.Mode = 0x0004; // GL_TRIANGLES
+        info.InstanceCount = 1;
+        info.MinIndex = ~0u;
+        info.MaxIndex = ~0u;
+        info.NumDraws = 1;
+        const MG_Pipe::MGPDrawRange range{0, 3, 0};
+        Uint64 last = Codec::kInvalidSeq;
+        for (int i = 0; i < 200; ++i) {
+            last = fixture.encoder.EncodeRecord(MG_Pipe::MGPWireOp::DrawVbo, &info, sizeof(info), &range,
+                                                sizeof(range));
+            if (last == Codec::kInvalidSeq) ::_exit(66);
+        }
+        fixture.encoder.Publish();
+        fixture.producer.PublishAndNotify(last);
+        if (!PollUntil([] { return g_stopStreamRecords.load(std::memory_order_acquire) >= 1; }, 3000)) ::_exit(67);
+        // RunSession's order when mobilegl_server_stop_inprocess raised the stop.
+        const auto started = std::chrono::steady_clock::now();
+        loop.AbandonQueuedRecords();
+        loop.Stop();
+        const auto stopMs =
+            std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() - started).count();
+        int failed = 0;
+        if (stopMs >= 1000) failed |= kStopSlow;
+        if (loop.DrainedRecords() > 3) failed |= kStopDrainedOn;
+        std::fprintf(stderr, "[stop] Stop() returned in %lld ms after %llu of 200 streamed records\n",
+                     static_cast<long long>(stopMs), static_cast<unsigned long long>(loop.DrainedRecords()));
+        loop.SetBetweenRecordsHookForTesting(nullptr);
+        fixture.Stop();
+        ::_exit(failed);
+    }
+} // namespace
+
+// P12 review fix. Red with AbandonQueuedRecords a no-op: the batch and the exit-path drain apply the
+// whole stream, Stop()'s join gives up at 5 s and Fatal{ApplyThreadJoinTimeout} aborts the process.
+TEST(ServerLoopTest, StoppingTheServerUnderAStreamingClientLeavesTheQueueAndJoinsPromptly) {
+    EXPECT_EXIT(StopTheServerMidStreamAndExit(), ::testing::ExitedWithCode(0), ".*")
+        << "bits: 1 = Stop() took 1 s or more; 2 = more than three records applied after the stop "
+           "(64+ = setup; SIGABRT = Fatal{ApplyThreadJoinTimeout})";
+}
+
 // D3. A server that owns no display - the host, the exec'd supervisor's children - refuses a request
 // for its window BY NAME and does NOT latch: it is the client's configuration, not corrupt bytes. Red
 // with the refusal turned into a latch: this unarmed process aborts.