refactor: address PR #851 review round 4 (ingest transport)

- 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) <noreply@anthropic.com>

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()

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()
 

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.