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

- 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) <noreply@anthropic.com>
This commit is contained in:
Chris Coutinho
2026-06-04 20:07:19 +02:00
co-authored by Claude Opus 4.8
parent e7bcdb1950
commit 655d608fb7
3 changed files with 36 additions and 8 deletions
+3 -7
View File
@@ -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,
)
@@ -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):