From c4401af9c619596715875f034410e17d6e010f9b Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Thu, 4 Jun 2026 20:42:57 +0200 Subject: [PATCH] refactor: address PR #851 review round 4 (ingest transport) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Clear the module-singleton ingest references (task_producer, document_send_stream, document_receive_stream) on lifespan shutdown via a new _clear_vector_sync_state() helper, mirroring the eviction_task_group cleanup. Defense-in-depth so a late webhook (or a module-singleton integration test) can't touch a producer/stream backed by an already-closed resource. - Add IngestTransport.backend_name ("memory"/"postgres") and use it in both lifespan log lines, removing the last settings.ingest_queue read from the background-sync setup — the lifespan no longer inspects the backend at all. - Cover backend_name in the build_transport adapter-selection tests. Refs: Deck #196 Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/app.py | 23 +++++++++++++++++-- .../vector/queue/transport.py | 17 ++++++++++++++ tests/unit/vector/test_ingest_transport.py | 2 ++ 3 files changed, 40 insertions(+), 2 deletions(-) diff --git a/nextcloud_mcp_server/app.py b/nextcloud_mcp_server/app.py index 9c7149a1..329ec51b 100644 --- a/nextcloud_mcp_server/app.py +++ b/nextcloud_mcp_server/app.py @@ -392,6 +392,21 @@ def _wire_vector_sync_state( break +def _clear_vector_sync_state() -> None: + """Drop the module-singleton ingest references on lifespan shutdown. + + Mirrors the ``eviction_task_group = None`` cleanup so that any code reaching + the singleton in the narrow window between transport teardown and process + exit (e.g. a late webhook) sees ``None`` rather than a producer/stream backed + by an already-closed resource. The per-request ``shutdown_event`` gate is the + primary guard; this is defense-in-depth. Integration tests with module-level + singletons also benefit (no stale closed producer leaks between runs). + """ + _vector_sync_state.task_producer = None + _vector_sync_state.document_send_stream = None + _vector_sync_state.document_receive_stream = None + + @dataclass class AppContext: """Application context for BasicAuth mode.""" @@ -1771,7 +1786,7 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = logger.info( "Background sync tasks started: 1 scanner + %s processors (queue=%s)", ingest_transport.active_consumer_count, - settings.ingest_queue, + ingest_transport.backend_name, ) # Run MCP session manager and yield @@ -1789,6 +1804,8 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = # for the memory stream, which task-group cancellation # closes). await ingest_transport.aclose() + # Drop stale singleton refs to the now-closed transport. + _clear_vector_sync_state() await client.close() # TaskGroup automatically cancels all tasks on exit @@ -1976,7 +1993,7 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = logger.info( "Background sync tasks started: 1 user manager + %s processors (queue=%s)", ingest_transport.active_consumer_count, - settings.ingest_queue, + ingest_transport.backend_name, ) # Run MCP session manager and yield @@ -1993,6 +2010,8 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = # procrastinate connector pool in postgres mode; # no-op for the memory stream). await ingest_transport.aclose() + # Drop stale singleton refs to the now-closed transport. + _clear_vector_sync_state() # Close token broker HTTP client if token_broker._http_client: await token_broker._http_client.aclose() diff --git a/nextcloud_mcp_server/vector/queue/transport.py b/nextcloud_mcp_server/vector/queue/transport.py index fb2fd4c4..6b27e36d 100644 --- a/nextcloud_mcp_server/vector/queue/transport.py +++ b/nextcloud_mcp_server/vector/queue/transport.py @@ -86,6 +86,15 @@ class IngestTransport(abc.ABC): def producer(self) -> TaskProducer: """The :class:`TaskProducer` to wire into ``app.state`` / the scanner.""" + @property + @abc.abstractmethod + def backend_name(self) -> str: + """Short backend identifier for logs/metrics (``memory``/``postgres``). + + Lets the lifespan log which ingest backend is active without reading + ``settings.ingest_queue`` — the backend choice stays inside the transport. + """ + @property def send_stream(self) -> MemoryObjectSendStream[DocumentTask] | None: """Memory backend's raw send end; ``None`` for distributed backends. @@ -156,6 +165,10 @@ class LocalTransport(IngestTransport): def producer(self) -> TaskProducer: return self._producer + @property + def backend_name(self) -> str: + return "memory" + @property def send_stream(self) -> MemoryObjectSendStream[DocumentTask]: return self._send_stream @@ -221,6 +234,10 @@ class DistributedTransport(IngestTransport): def producer(self) -> TaskProducer: return self._producer + @property + def backend_name(self) -> str: + return "postgres" + async def aclose(self) -> None: await self._producer.drain() diff --git a/tests/unit/vector/test_ingest_transport.py b/tests/unit/vector/test_ingest_transport.py index c5ddbadc..afb3b606 100644 --- a/tests/unit/vector/test_ingest_transport.py +++ b/tests/unit/vector/test_ingest_transport.py @@ -44,6 +44,7 @@ class TestBuildTransport: assert isinstance(transport, LocalTransport) assert isinstance(transport.producer, MemoryTaskProducer) + assert transport.backend_name == "memory" # Memory backend exposes both raw stream ends. assert transport.send_stream is not None assert transport.receive_stream is not None @@ -61,6 +62,7 @@ class TestBuildTransport: assert isinstance(transport, DistributedTransport) assert transport.producer is producer + assert transport.backend_name == "postgres" # Schema applied once on the open pool before any defer. producer.ensure_schema.assert_awaited_once() # No in-process stream for the distributed backend.