proj/portage:master commit in: lib/_emerge/, lib/portage/tests/emerge/

commit:     a709d40e2882278fb61666d7810add0a117f1b75
Author:     Matt Turner <mattst88 <AT> gentoo <DOT> org>
AuthorDate: Tue Aug 11 22:35:56 2026 +0000
Commit:     Matt Turner <mattst88 <AT> gentoo <DOT> org>
CommitDate: Fri Aug 14 20:37:44 2026 +0000
URL:        https://gitweb.gentoo.org/proj/portage.git/commit/?id=a709d40e

observability: forget clients that disconnect

_client_connected appended each client's writer to self._writers with no
removal path other than _broadcast dropping a writer whose write() raised.
It never does: asyncio's stream protocol keeps the transport open after
EOF, to permit half-close, and once connection_lost() has fired a write is
counted and discarded rather than raising. Every client that connected
therefore stayed in the list for the lifetime of the emerge, and every
update walked all of them.

Await EOF on the client's reader and drop the writer when it arrives.
Also do not record a writer whose initial snapshot could not be sent.

This is minor today, but becomes routine once "emerge --status" reads the
socket, where each invocation connects, reads one line, and exits.

Signed-off-by: Matt Turner <mattst88 <AT> gentoo.org>

 lib/_emerge/_observability.py                  | 22 +++++++++++++--
 lib/portage/tests/emerge/test_observability.py | 39 ++++++++++++++++++++++++++
 2 files changed, 58 insertions(+), 3 deletions(-)

diff --git a/lib/_emerge/_observability.py b/lib/_emerge/_observability.py
index 9226ff752..255d826b3 100644
--- a/lib/_emerge/_observability.py
+++ b/lib/_emerge/_observability.py
@@ -382,13 +382,29 @@ class ObservabilityMonitor:
                 noiselevel=-1,
             )
 
-    def _client_connected(self, reader, writer):
-        self._writers.append(writer)
+    async def _client_connected(self, reader, writer):
         if self._last_snapshot is not None:
             data = (json.dumps(self._last_snapshot, sort_keys=True) + "\n").encode(
                 "utf_8"
             )
-            self._send(writer, data)
+            if not self._send(writer, data):
+                return
+        self._writers.append(writer)
+        try:
+            # Clients are not expected to send anything; this waits for the
+            # peer to go away.  Without it a disconnected client is never
+            # forgotten: asyncio's stream protocol keeps the transport open
+            # after EOF, to permit half-close, so nothing else notices.
+            # A client that shuts down only its write side is therefore
+            # treated as gone: read the stream without half-closing.
+            while await reader.read(4096):
+                pass
+        except OSError:
+            pass
+        finally:
+            if writer in self._writers:
+                self._writers.remove(writer)
+            writer.close()
 
     def _broadcast(self, snapshot):
         if not self._writers:

diff --git a/lib/portage/tests/emerge/test_observability.py b/lib/portage/tests/emerge/test_observability.py
index 152fbe7b5..3ed138420 100644
--- a/lib/portage/tests/emerge/test_observability.py
+++ b/lib/portage/tests/emerge/test_observability.py
@@ -1,6 +1,7 @@
 # Copyright 2026 Gentoo Authors
 # Distributed under the terms of the GNU General Public License v2
 
+import asyncio
 import json
 import os
 import tempfile
@@ -17,6 +18,7 @@ from _emerge._observability import (
 from _emerge.PackageMerge import PackageMerge as _RealPackageMerge
 
 from portage.tests import TestCase
+from portage.util._eventloop.global_event_loop import global_event_loop
 
 
 class _Pkg:
@@ -178,6 +180,43 @@ class ObservabilitySnapshotTestCase(TestCase):
             monitor.close()
             self.assertFalse(os.path.exists(path))
 
+    def test_client_disconnect_removes_writer(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            build = EbuildBuild(_Pkg("dev-libs/foo-1.2"), pid=99)
+            sched = _make_scheduler(eprefix=tmp, tasks=[build])
+            monitor = ObservabilityMonitor(sched)
+            monitor.note_task_started(build)
+
+            async def exercise():
+                monitor.update(force=True)
+                for _ in range(100):
+                    if monitor._server is not None:
+                        break
+                    await asyncio.sleep(0.01)
+                self.assertIsNotNone(monitor._server)
+
+                reader, writer = await asyncio.open_unix_connection(
+                    monitor._socket_path
+                )
+                snap = json.loads(await reader.readline())
+                self.assertEqual(snap["tasks"][0]["cpv"], "dev-libs/foo-1.2")
+                self.assertEqual(len(monitor._writers), 1)
+
+                writer.close()
+                await writer.wait_closed()
+                # The server side must forget the client on EOF alone, without
+                # needing a broadcast to discover the write is going nowhere.
+                for _ in range(100):
+                    if not monitor._writers:
+                        break
+                    await asyncio.sleep(0.01)
+                self.assertEqual(monitor._writers, [])
+
+            try:
+                global_event_loop().run_until_complete(exercise())
+            finally:
+                monitor.close()
+
     def test_hint_when_nothing_read_and_feature_absent(self):
         self.assertIn("observability", missing_feature_hint([], features=set()))