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

- _clear_vector_sync_state also nulls shutdown_event / scanner_wake_event on
  shutdown, symmetric with the stream/producer fields (the next startup rebinds
  them via _wire_vector_sync_state).
- Comment that the "DocumentTask" string subscript in LocalTransport is
  intentional (TYPE_CHECKING-only class; anyio ignores the runtime type arg).
- Move app.py's annotation-only IngestTransport / TaskProducer imports under
  TYPE_CHECKING (the module uses `from __future__ import annotations`), keeping
  only build_transport at runtime.

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:49:17 +02:00
co-authored by Claude Opus 4.8
parent c4401af9c6
commit bf84db35b4
2 changed files with 14 additions and 6 deletions
+11 -6
View File
@@ -9,7 +9,7 @@ import traceback
from collections.abc import AsyncIterator from collections.abc import AsyncIterator
from contextlib import AsyncExitStack, asynccontextmanager from contextlib import AsyncExitStack, asynccontextmanager
from dataclasses import dataclass from dataclasses import dataclass
from typing import Any, cast from typing import TYPE_CHECKING, Any, cast
from urllib.parse import urlparse from urllib.parse import urlparse
import anyio import anyio
@@ -133,14 +133,15 @@ from nextcloud_mcp_server.vector.oauth_sync import (
from nextcloud_mcp_server.vector.placeholder import sweep_orphan_placeholders from nextcloud_mcp_server.vector.placeholder import sweep_orphan_placeholders
from nextcloud_mcp_server.vector.processor import processor_task from nextcloud_mcp_server.vector.processor import processor_task
from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client
from nextcloud_mcp_server.vector.queue import ( from nextcloud_mcp_server.vector.queue import build_transport
IngestTransport,
TaskProducer,
build_transport,
)
from nextcloud_mcp_server.vector.scanner import scanner_task from nextcloud_mcp_server.vector.scanner import scanner_task
from nextcloud_mcp_server.vector.webhook_receiver import handle_nextcloud_webhook from nextcloud_mcp_server.vector.webhook_receiver import handle_nextcloud_webhook
if TYPE_CHECKING:
# Annotation-only in this module (the file uses `from __future__ import
# annotations`, so these are never evaluated at runtime).
from nextcloud_mcp_server.vector.queue import IngestTransport, TaskProducer
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
HTTPXClientInstrumentor().instrument() HTTPXClientInstrumentor().instrument()
@@ -405,6 +406,10 @@ def _clear_vector_sync_state() -> None:
_vector_sync_state.task_producer = None _vector_sync_state.task_producer = None
_vector_sync_state.document_send_stream = None _vector_sync_state.document_send_stream = None
_vector_sync_state.document_receive_stream = None _vector_sync_state.document_receive_stream = None
# Symmetric with the fields above: the just-fired events belong to the
# closed lifespan; the next startup's _wire_vector_sync_state rebinds them.
_vector_sync_state.shutdown_event = None
_vector_sync_state.scanner_wake_event = None
@dataclass @dataclass
@@ -153,6 +153,9 @@ class LocalTransport(IngestTransport):
""" """
def __init__(self, max_buffer_size: float): def __init__(self, max_buffer_size: float):
# "DocumentTask" as a string (not the symbol): the class is
# TYPE_CHECKING-only here, and anyio ignores the runtime value of the
# type argument — so the string is intentional, not a typo.
send_stream, receive_stream = anyio.create_memory_object_stream["DocumentTask"]( send_stream, receive_stream = anyio.create_memory_object_stream["DocumentTask"](
max_buffer_size=max_buffer_size max_buffer_size=max_buffer_size
) )