Chris CoutinhoandClaude Opus 4.8
cfdef3c2c5
🔴 nc_get_vector_sync_status reported pending=0 for INGEST_QUEUE=postgres: the AppContext/OAuthAppContext per-session yields snapshotted the stream fields but never forwarded task_producer, so lifespan_ctx.task_producer was always None. Convert task_producer to a @property that reads _vector_sync_state live (like eviction_task_group), removing the snapshot field so the yields can't drop it. Add a regression test pinning the contract on both contexts. 🟡 Remove the unused _RECLAIM_TASK_NAME constant. 🟡 get_procrastinate_conninfo: warn + document that DATABASE_URL query params (application_name, connect_timeout, …) are dropped. 🟡 worker: open the procrastinate App once — apply_ingest_queue_schema gains manage_connection=False so the worker reuses its own open connector instead of a redundant open/close before run_worker_async. 🟢 Clarify the apply-schema broad-except comment (non-race errors re-raise) and document the deliberate Any typing in ingest_status.get_ingest_pending. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
63 lines
2.5 KiB
Python
63 lines
2.5 KiB
Python
"""Shared read model for the vector-sync status surface (Deck #183).
|
|
|
|
The status endpoints (``/api/v1/vector-sync/status``, the userinfo route, and
|
|
the ``nc_get_vector_sync_status`` MCP tool) all need the same "how much work is
|
|
outstanding" figure, computed differently per ``INGEST_QUEUE`` backend:
|
|
|
|
- ``memory`` — the in-process anyio stream's buffer depth (today's behavior).
|
|
- ``postgres`` — procrastinate job counts read from the per-tenant Postgres
|
|
(``todo`` + ``doing``), plus the per-status breakdown for observability.
|
|
|
|
``indexed_documents`` (the Qdrant placeholder count) is backend-independent and
|
|
stays at each call site.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class IngestPending:
|
|
"""Outstanding-work view for the active ingest queue backend."""
|
|
|
|
pending: int
|
|
# Per-status counts (todo/doing/failed/…) on the postgres backend; None on
|
|
# the memory backend, which has no durable per-status breakdown.
|
|
job_counts: dict[str, int] | None = None
|
|
|
|
|
|
async def get_ingest_pending(
|
|
*, task_producer: Any, document_receive_stream: Any, ingest_queue: str | None
|
|
) -> IngestPending:
|
|
"""Compute outstanding ingest work for the configured queue backend.
|
|
|
|
``task_producer`` and ``document_receive_stream`` are intentionally typed
|
|
``Any``: they're duck-typed across backends. Only ``ProcrastinateTaskProducer``
|
|
exposes ``job_counts`` (the ``TaskProducer`` protocol doesn't), and the memory
|
|
backend reads the anyio stream's ``statistics()`` — so no single concrete type
|
|
or Protocol fits both branches, and we probe with ``hasattr`` instead.
|
|
|
|
Never raises — a status surface must stay available even if the queue is
|
|
unreachable; failures degrade to ``pending=0``.
|
|
"""
|
|
if ingest_queue == "postgres":
|
|
counts: dict[str, int] = {}
|
|
if task_producer is not None and hasattr(task_producer, "job_counts"):
|
|
try:
|
|
counts = await task_producer.job_counts()
|
|
except Exception as e:
|
|
logger.warning("Failed to read ingest job counts: %s", e)
|
|
pending = counts.get("todo", 0) + counts.get("doing", 0)
|
|
return IngestPending(pending=pending, job_counts=counts)
|
|
|
|
if document_receive_stream is None:
|
|
return IngestPending(pending=0)
|
|
return IngestPending(
|
|
pending=document_receive_stream.statistics().current_buffer_used
|
|
)
|