- 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>
- LocalTransport.run_consumers increments active_consumer_count per worker
(instead of once after the loop) so the count is accurate if a later
tg.start() raises mid-pool.
- Add LocalTransport.aclose() to explicitly close its owned send/receive stream
ends (belt-and-suspenders against unclosed-resource warnings; anyio aclose is
idempotent, and by shutdown the scanner is already winding down). Reworded the
base IngestTransport.aclose() docstring to point at the overrides.
- Inline ingest_transport.producer at the scanner/user_manager call sites,
dropping the single-use task_producer alias in both lifespan paths.
- Annotate DistributedTransport._producer explicitly as ProcrastinateTaskProducer
so the drain() coupling is visible and ty catches drift.
- Add a unit test for LocalTransport.aclose() (closes the owned streams,
idempotent).
Refs: Deck #196
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
- 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>
Finish the hexagonal ports-&-adapters split started in #183. The producer side
already had a TaskProducer port + adapters, but the consumer side was
unabstracted and the INGEST_QUEUE selection leaked into a duplicated
`if use_postgres:` branch across both app.py lifespan paths.
Introduce an IngestTransport ABC (vector/queue/transport.py) that bundles the
producer with running (or not running) the in-process consumer pool, built by a
single build_transport() factory:
- LocalTransport (INGEST_QUEUE=memory): in-process anyio stream drained by an
N-worker pool that run_consumers starts.
- DistributedTransport (INGEST_QUEUE=postgres): wraps ProcrastinateTaskProducer;
run_consumers is a no-op because the consumer is the external `worker` role.
Both lifespan paths now call build_transport + _wire_vector_sync_state (new
helper that centralizes the app.state / module-singleton / browser-app writes) +
transport.run_consumers + transport.aclose(), with no INGEST_QUEUE branching and
no getattr drain probe. Adding a future backend (Redis/NATS/SQS) is one new
adapter + one build_transport arm, with no app.py or scanner change.
Preserves the single-tenant parallelism invariant (one shared multiplexed queue
+ N-worker pool, per-document not per-user dispatch) and documents it in
ADR-028. The worker CLI is unchanged (it is the external consumer).
Refs: Deck #196 (Deck #197 tracks the explicit parallelism regression test)
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
