From 655d608fb73d801629997cf022e56ca54b03e243 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Thu, 4 Jun 2026 20:07:19 +0200 Subject: [PATCH] refactor: address PR #851 review round 1 (ingest transport) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add IngestTransport.active_consumer_count (0 by default; LocalTransport stores the started count) so app.py logs the worker count without re-checking INGEST_QUEUE — the last backend-knowledge leak in the lifespan is gone. - Document that DistributedTransport is postgres/procrastinate-specific by design (aclose() calls ProcrastinateTaskProducer.drain()); other distributed backends would be separate IngestTransport subclasses. - Clarify the _wire_vector_sync_state log line (writes app.state + singleton, not only the singleton). - Strengthen the LocalTransport test: assert active_consumer_count transitions 0→N and that each worker receives a distinct cloned receive stream. Refs: Deck #196 Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/app.py | 10 +++----- .../vector/queue/transport.py | 23 +++++++++++++++++++ tests/unit/vector/test_ingest_transport.py | 11 ++++++++- 3 files changed, 36 insertions(+), 8 deletions(-) diff --git a/nextcloud_mcp_server/app.py b/nextcloud_mcp_server/app.py index 4145acbb..f514a29e 100644 --- a/nextcloud_mcp_server/app.py +++ b/nextcloud_mcp_server/app.py @@ -377,7 +377,7 @@ def _wire_vector_sync_state( # app.state (Starlette) + the module singleton share the same attribute names. _apply(app.state) _apply(_vector_sync_state) - logger.info("Vector sync state stored in module singleton") + logger.info("Vector sync state published (app.state + module singleton)") # Also share with the mounted /app browser sub-app, if present. for route in app.routes: @@ -1763,9 +1763,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)", - 0 - if settings.ingest_queue == "postgres" - else settings.vector_sync_processor_workers, + transport.active_consumer_count, settings.ingest_queue, ) @@ -1969,9 +1967,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)", - 0 - if settings.ingest_queue == "postgres" - else settings.vector_sync_processor_workers, + transport.active_consumer_count, settings.ingest_queue, ) diff --git a/nextcloud_mcp_server/vector/queue/transport.py b/nextcloud_mcp_server/vector/queue/transport.py index 9ce6f62d..fb4fcbbf 100644 --- a/nextcloud_mcp_server/vector/queue/transport.py +++ b/nextcloud_mcp_server/vector/queue/transport.py @@ -104,6 +104,17 @@ class IngestTransport(abc.ABC): reads procrastinate job counts via the producer instead).""" return None + @property + def active_consumer_count(self) -> int: + """In-process consumers started by :meth:`run_consumers` for this process. + + ``0`` by default — distributed backends run their consumers as a separate + ``worker`` process. Lets the lifespan log the worker count without + re-inspecting ``INGEST_QUEUE`` (keeping the backend choice inside the + transport). + """ + return 0 + async def run_consumers( self, task_group: TaskGroup, spawn_worker: SpawnWorker, count: int ) -> None: @@ -139,6 +150,7 @@ class LocalTransport(IngestTransport): self._send_stream = send_stream self._receive_stream = receive_stream self._producer = MemoryTaskProducer(send_stream) + self._active_consumer_count = 0 @property def producer(self) -> TaskProducer: @@ -152,6 +164,10 @@ class LocalTransport(IngestTransport): def receive_stream(self) -> MemoryObjectReceiveStream[DocumentTask]: return self._receive_stream + @property + def active_consumer_count(self) -> int: + return self._active_consumer_count + async def run_consumers( self, task_group: TaskGroup, spawn_worker: SpawnWorker, count: int ) -> None: @@ -162,6 +178,7 @@ class LocalTransport(IngestTransport): # the prior inline lifespan behaviour. for i in range(count): await task_group.start(spawn_worker, i, self._receive_stream.clone()) + self._active_consumer_count = count class DistributedTransport(IngestTransport): @@ -171,6 +188,12 @@ class DistributedTransport(IngestTransport): by the lifespan; :meth:`run_consumers` is the inherited no-op (the ``nextcloud-mcp-server worker`` process drains the queue) and :meth:`aclose` closes the pool once on shutdown. + + This adapter is postgres/procrastinate-specific by design: :meth:`aclose` + calls ``ProcrastinateTaskProducer.drain()`` (the narrow ``_producer`` type + confirms it). A different distributed backend (Redis/NATS/SQS) with its own + shutdown semantics would be a separate :class:`IngestTransport` subclass, not + a reconfiguration of this one. """ def __init__(self, producer: ProcrastinateTaskProducer): diff --git a/tests/unit/vector/test_ingest_transport.py b/tests/unit/vector/test_ingest_transport.py index 93242874..d7fb4b40 100644 --- a/tests/unit/vector/test_ingest_transport.py +++ b/tests/unit/vector/test_ingest_transport.py @@ -70,6 +70,8 @@ class TestBuildTransport: class TestLocalTransport: async def test_run_consumers_starts_count_workers_off_shared_stream(self): transport = LocalTransport(max_buffer_size=5) + # Not yet started → no active consumers. + assert transport.active_consumer_count == 0 started: list[int] = [] received_streams: list[object] = [] @@ -85,9 +87,12 @@ class TestLocalTransport: await transport.run_consumers(tg, fake_worker, 3) assert sorted(started) == [0, 1, 2] - # Each worker gets its own (cloned) receive handle, none None. + assert transport.active_consumer_count == 3 + # Each worker gets its own distinct cloned receive handle (so each + # observes end-of-stream when the senders all close), none None. assert len(received_streams) == 3 assert all(s is not None for s in received_streams) + assert len({id(s) for s in received_streams}) == 3 class TestDistributedTransport: @@ -95,6 +100,9 @@ class TestDistributedTransport: producer = AsyncMock() transport = DistributedTransport(producer) + # No in-process consumers for the distributed backend. + assert transport.active_consumer_count == 0 + # Passing sentinel task group / spawn callback proves the no-op never # touches them (the external worker is the consumer). The casts satisfy # the signature; the values are deliberately unusable to catch any @@ -104,6 +112,7 @@ class TestDistributedTransport: await transport.run_consumers(sentinel_tg, sentinel_spawn, count=3) producer.assert_not_awaited() + assert transport.active_consumer_count == 0 async def test_aclose_drains_producer(self): producer = AsyncMock()