diff --git a/CONTEXT.md b/CONTEXT.md index b67e16d0..8609579f 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -221,7 +221,7 @@ the serving-index latency target until the derived data is repaired. ### Retrieval Serving Generation The namespace-scoped version that identifies one coherent set of active -document revisions and their serving-index statistics. Retrieval captures one +document revisions and their serving-index data. Retrieval captures one generation and retries or falls back if publication changes it during capture. ### Retrieval Semantic Parity diff --git a/apps/api/alembic/versions/9f0a1b2c3d4e_drop_retrieval_namespace_statistics.py b/apps/api/alembic/versions/9f0a1b2c3d4e_drop_retrieval_namespace_statistics.py new file mode 100644 index 00000000..55069014 --- /dev/null +++ b/apps/api/alembic/versions/9f0a1b2c3d4e_drop_retrieval_namespace_statistics.py @@ -0,0 +1,119 @@ +"""Drop unused retrieval namespace statistics tables. + +Query-time BM25 scoring reads only ``document_map_unit_tokens``, +``document_map_units``, and ``document_map_unit_indexes``. The per-revision and +namespace statistics tables were only written by publication/backfill and never +read by the retrieval path, so they are removed here. +""" + +from __future__ import annotations + +from collections.abc import Sequence + +from alembic import op +import sqlalchemy as sa + + +revision: str = "9f0a1b2c3d4e" +down_revision: str | None = "8e9f0a1b2c3d" +branch_labels: Sequence[str] | None = None +depends_on: Sequence[str] | None = None + +__all__ = [ + "revision", + "down_revision", + "branch_labels", + "depends_on", + "upgrade", + "downgrade", +] + + +def upgrade() -> None: + op.drop_index( + "idx_retrieval_namespace_token_stats_lookup", + table_name="retrieval_namespace_token_stats", + if_exists=True, + ) + op.drop_table("retrieval_namespace_token_stats", if_exists=True) + op.drop_table("retrieval_namespace_stats", if_exists=True) + op.drop_index( + "idx_retrieval_serving_revision_stats_scope", + table_name="retrieval_serving_revision_stats", + if_exists=True, + ) + op.drop_table("retrieval_serving_revision_stats", if_exists=True) + + +def downgrade() -> None: + bind = op.get_bind() + inspector = sa.inspect(bind) + if not inspector.has_table("retrieval_serving_revision_stats"): + op.create_table( + "retrieval_serving_revision_stats", + sa.Column("id", sa.String(length=100), nullable=False), + sa.Column("user_id", sa.Text(), nullable=False), + sa.Column("namespace", sa.String(length=255), nullable=False), + sa.Column("document_id", sa.String(length=36), nullable=False), + sa.Column("job_result_id", sa.String(length=36), nullable=False), + sa.Column("format_version", sa.Integer(), nullable=False), + sa.Column("payload_zlib", sa.LargeBinary(), nullable=False), + sa.Column("checksum", sa.String(length=64), nullable=False), + sa.Column("created_at", sa.DateTime(), nullable=False), + sa.ForeignKeyConstraint( + ["document_id"], ["documents.document_id"], ondelete="CASCADE" + ), + sa.ForeignKeyConstraint( + ["job_result_id"], ["job_results.id"], ondelete="CASCADE" + ), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint( + "document_id", + "job_result_id", + name="uq_retrieval_serving_revision_stats_revision", + ), + ) + op.create_index( + "idx_retrieval_serving_revision_stats_scope", + "retrieval_serving_revision_stats", + ["user_id", "namespace", "document_id", "job_result_id"], + ) + if not inspector.has_table("retrieval_namespace_stats"): + op.create_table( + "retrieval_namespace_stats", + sa.Column("id", sa.String(length=100), nullable=False), + sa.Column("user_id", sa.Text(), nullable=False), + sa.Column("namespace", sa.String(length=255), nullable=False), + sa.Column("generation", sa.BigInteger(), nullable=False), + sa.Column("payload_zlib", sa.LargeBinary(), nullable=False), + sa.Column("checksum", sa.String(length=64), nullable=False), + sa.Column("updated_at", sa.DateTime(), nullable=False), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint( + "user_id", "namespace", name="uq_retrieval_namespace_stats_scope" + ), + ) + if not inspector.has_table("retrieval_namespace_token_stats"): + op.create_table( + "retrieval_namespace_token_stats", + sa.Column("id", sa.String(length=100), nullable=False), + sa.Column("user_id", sa.Text(), nullable=False), + sa.Column("namespace", sa.String(length=255), nullable=False), + sa.Column("generation", sa.BigInteger(), nullable=False), + sa.Column("channel", sa.String(length=32), nullable=False), + sa.Column("token_hash", sa.String(length=64), nullable=False), + sa.Column("document_frequency", sa.Integer(), nullable=False), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint( + "user_id", + "namespace", + "channel", + "token_hash", + name="uq_retrieval_namespace_token_stats_key", + ), + ) + op.create_index( + "idx_retrieval_namespace_token_stats_lookup", + "retrieval_namespace_token_stats", + ["user_id", "namespace", "generation", "channel", "token_hash"], + ) diff --git a/apps/api/app/services/documents/lifecycle_service.py b/apps/api/app/services/documents/lifecycle_service.py index 3ea08f05..fd97a9ba 100644 --- a/apps/api/app/services/documents/lifecycle_service.py +++ b/apps/api/app/services/documents/lifecycle_service.py @@ -8,13 +8,11 @@ from app.repositories.document_repository import DocumentRepository from loguru import logger -from sqlalchemy import delete from sqlalchemy.ext.asyncio import AsyncSession from shared.models.database.document import ( DocumentChunk, DocumentSection, - RetrievalServingRevisionStat, ) from shared.services.retrieval.cache_service import ( invalidate_retrieval_cache_namespaces, @@ -27,9 +25,6 @@ advance_namespace_generation, lock_namespace_generation, ) -from shared.services.retrieval.serving_manifest import ( - rebuild_namespace_serving_statistics, -) from shared.services.storage.result_storage import ResultStorage, get_result_storage _DOCUMENT_CHUNK_ASSET_URL_EXPIRES_SECONDS = 7 * 24 * 60 * 60 @@ -469,23 +464,6 @@ async def archive_document( ) ) await self._repository.archive_document(db, document=document) - current_revision = document.current_job_result_id - if current_revision: - await db.run_sync( - lambda sync_db: sync_db.execute( - delete(RetrievalServingRevisionStat).where( - RetrievalServingRevisionStat.document_id == document_id, - RetrievalServingRevisionStat.job_result_id == current_revision, - ) - ) - ) - await db.run_sync( - lambda sync_db: rebuild_namespace_serving_statistics( - sync_db, - user_id=user_id, - namespace=previous_namespace, - ) - ) await db.run_sync( lambda sync_db: remove_document_from_namespace_map_snapshot( sync_db, diff --git a/apps/api/scripts/backfill_map_unit_indexes.py b/apps/api/scripts/backfill_map_unit_indexes.py index 6eed78ef..ebf1a04d 100644 --- a/apps/api/scripts/backfill_map_unit_indexes.py +++ b/apps/api/scripts/backfill_map_unit_indexes.py @@ -1,8 +1,8 @@ """Backfill persisted MAP-NAV lexical indexes for existing revisions. Rebuilds, per active revision: the map-unit index, the revision serving -manifest, that document's subtree in the namespace MAP snapshot, namespace -statistics, and the namespace generation. The migrations that create these +manifest, that document's subtree in the namespace MAP snapshot, and the +namespace generation. The migrations that create these derived tables leave them empty intentionally. Run this command after deployment with ``--apply`` so each revision is rebuilt and committed independently; without ``--apply`` it is a read-only inventory. @@ -69,7 +69,6 @@ def _bootstrap_python_path() -> None: from shared.services.retrieval.serving_manifest import ( decode_serving_manifest, persist_revision_serving_state, - rebuild_namespace_serving_statistics, ) @@ -346,11 +345,6 @@ def backfill_map_unit_indexes(*, apply: bool, document_id: str = "") -> int: patch_namespace_map_snapshot( db, scope=scope, manifest_payload=manifest_payload ) - rebuild_namespace_serving_statistics( - db, - user_id=scope.user_id, - namespace=scope.namespace, - ) advance_namespace_generation( db, user_id=scope.user_id, diff --git a/apps/api/tests/contract/test_documents_contract.py b/apps/api/tests/contract/test_documents_contract.py index a71d4029..3e33f202 100644 --- a/apps/api/tests/contract/test_documents_contract.py +++ b/apps/api/tests/contract/test_documents_contract.py @@ -13,7 +13,6 @@ from tests.support.contract_database import ContractDatabase from shared.testing.contract_runtime import get_contract_database_url -from shared.services.retrieval.serving_manifest import encode_serving_manifest async def _create_contract_engine() -> AsyncEngine: @@ -1333,84 +1332,3 @@ async def test_should_archive_a_document_via_the_legacy_archive_route( assert response_json["archived_at"] assert persisted_document["status"] == "archived" assert persisted_document["archived_at"] is not None - - -@pytest.mark.asyncio -async def test_archive_removes_revision_serving_stats_and_advances_generation( - developer_api_client_factory: Callable[ - [], AbstractAsyncContextManager[AsyncClient] - ], -) -> None: - document_id = f"doc_{uuid4().hex[:12]}" - namespace = f"archive-serving-{uuid4().hex[:8]}" - async with developer_api_client_factory() as api_client: - revision = await _insert_document_revision_with_chunks( - document_id=document_id, - namespace=namespace, - chunks=[ - { - "id": f"dchk_{uuid4().hex[:12]}", - "chunk_id": "archive-serving-chunk", - "chunk_type": "text", - "content": "serving contribution", - "source_chunk_path": "Archive/Serving", - "metadata": {}, - } - ], - ) - payload_bytes, checksum, version = encode_serving_manifest( - { - "document_id": document_id, - "job_result_id": revision["job_result_id"], - "unit_count": 1, - "path_token_count": 1, - "content_token_count": 2, - "token_frequencies": { - "path": {"archive": 1}, - "content": {"serving": 1}, - }, - } - ) - await ContractDatabase.execute( - """ - INSERT INTO retrieval_serving_revision_stats ( - id, user_id, namespace, document_id, job_result_id, - format_version, payload_zlib, checksum, created_at - ) VALUES ( - :id, :user_id, :namespace, :document_id, :job_result_id, - :format_version, :payload_zlib, :checksum, NOW() - ) - """, - { - "id": f"rss_{uuid4().hex[:12]}", - "user_id": "local-dev-user", - "namespace": namespace, - "document_id": document_id, - "job_result_id": revision["job_result_id"], - "format_version": version, - "payload_zlib": payload_bytes, - "checksum": checksum, - }, - ) - response = await api_client.post(f"/api/v1/documents/{document_id}/archive") - - assert response.status_code == 200 - remaining_revision_stats = await ContractDatabase.fetch_one( - """ - SELECT id - FROM retrieval_serving_revision_stats - WHERE document_id = :document_id AND job_result_id = :job_result_id - """, - {"document_id": document_id, "job_result_id": revision["job_result_id"]}, - ) - namespace_stats = await ContractDatabase.fetch_one( - """ - SELECT generation - FROM retrieval_namespace_stats - WHERE user_id = :user_id AND namespace = :namespace - """, - {"user_id": "local-dev-user", "namespace": namespace}, - ) - assert remaining_revision_stats is None - assert namespace_stats is not None - assert int(namespace_stats["generation"]) >= 1 diff --git a/docs/adr/0006-atomically-publish-retrieval-serving-index.md b/docs/adr/0006-atomically-publish-retrieval-serving-index.md index 6f0b0800..49e3e472 100644 --- a/docs/adr/0006-atomically-publish-retrieval-serving-index.md +++ b/docs/adr/0006-atomically-publish-retrieval-serving-index.md @@ -2,5 +2,5 @@ - Status: Accepted - Context: Retrieval will use a persistent derived serving index to avoid rebuilding a large namespace on every first request. A document revision without a complete index would have unpredictable latency and could produce inconsistent scoring metadata. -- Decision: Build the serving manifest and scoring statistics in the same database transaction as the document revision. Write the completeness marker last. If serving-index construction fails, roll back the publication and retry the job; do not expose an active revision with a partial serving index. +- Decision: Build the map-unit index and serving manifest in the same database transaction as the document revision. Write the completeness marker last. If serving-index construction fails, roll back the publication and retry the job; do not expose an active revision with a partial serving index. - Consequences: Active revisions have a simple completeness invariant and predictable first-request behavior. Publication takes more work and storage, and an index failure can delay publication, but retrieval can retain a guarded legacy fallback for migrations or already-existing incomplete revisions. diff --git a/docs/adr/0007-use-coherent-retrieval-serving-generations.md b/docs/adr/0007-use-coherent-retrieval-serving-generations.md index 89d2ca05..f875480d 100644 --- a/docs/adr/0007-use-coherent-retrieval-serving-generations.md +++ b/docs/adr/0007-use-coherent-retrieval-serving-generations.md @@ -1,6 +1,6 @@ # Use coherent retrieval-serving generations - Status: Accepted -- Context: A namespace can contain many active document revisions, and publication can replace them while a retrieval request is loading serving metadata and scoring statistics. +- Context: A namespace can contain many active document revisions, and publication can replace them while a retrieval request is loading serving metadata. - Decision: Assign each namespace a serving generation. Retrieval captures one generation and verifies it across serving reads; if it changes, retry once and use the exact legacy path if consistency cannot be established. -- Consequences: Retrieval never combines incompatible revision metadata and scoring statistics. Publication and retrieval need a small amount of generation bookkeeping, and rare concurrent updates may cause a retry or slower fallback. +- Consequences: Retrieval never combines incompatible revision metadata. Publication and retrieval need a small amount of generation bookkeeping, and rare concurrent updates may cause a retry or slower fallback. diff --git a/docs/design/retrieval-serving-index-plan.md b/docs/design/retrieval-serving-index-plan.md deleted file mode 100644 index 5e66f6d6..00000000 --- a/docs/design/retrieval-serving-index-plan.md +++ /dev/null @@ -1,425 +0,0 @@ -# Retrieval-serving index: online performance plan - -**Status:** Proposed for review -**Reviewed against:** current `knowhere` retrieval and publication code, 2026-08-29 -**Scope:** first-request retrieval performance; LLM planner/harvest/control time is excluded - -## 1. Goal and non-goals - -The target is predictable, bounded **Retrieval Non-LLM Work** on the current -production-sized namespace (about 643 active documents, 50k sections, and 60k -chunks). The measured baseline is roughly 160-170 seconds before map-nav's LLM -episode begins. We do not require a one-second absolute target in this phase; -we require that the request path avoid repeated full-corpus work and have a -clear linear/bounded complexity profile. - -The target includes: - -- snapshot or serving-index loading; -- classic or map-nav lexical scoring; -- ranking; -- selected-result hydration; -- citation and asset-reference assembly. - -It excludes planner, harvest, control, and answer-generation model time. Those -stages must continue to have separate timings. - -The plan does not change prompts, models, tokenization, BM25 formulas, RRF -weights, cache semantics, citation rules, or public HTTP response shapes. -It does not add LLM-response caching, process-wide serving caches, or startup -prewarming. The acceptance benchmark is a cold, uncached retrieval request, -and planner/harvest/control model time remains a separately reported -dependency. Episode-local reuse is allowed only while one request is active. - -## 2. Verified current behavior - -The current code has two retrieval routes in -`shared/services/retrieval/execution/routes.py`: - -- `use_agentic=false` runs `bottom_discovery()` and then ranking/hydration. -- The default map-nav route calls `load_nav_snapshot(..., lazy=True)`, runs the - synchronous navigation episode, then opens a fresh database context for - reference resolution and result assembly. - -`load_nav_snapshot()` currently: - -1. Reads active documents and their current job-result IDs. -2. Loads all matching sections into memory. -3. Loads all chunk identities and `connect_to` metadata into memory. -4. Uses a lazy store for selected chunk content and asset paths. - -The current persisted map scorer in -`shared/services/retrieval/nav/nav_knowhere.py` still loads all eligible map -units, then reads query frequencies and term scores from the persisted tables. -The scorer itself is fast; the broad database projection is not. - -`bottom_discovery()` currently executes path, content, and term channels -sequentially. `term_channel()` uses substring predicates over lowercased text, -without a trigram index. - -Publication currently writes sections/chunks and then calls -`replace_document_map_units()` in the same SQLAlchemy transaction. The existing -`document_map_unit_indexes` row is a per-document-revision completeness marker. -The existing backfill script only rebuilds that map-unit index; it does not yet -build the proposed serving manifest or namespace statistics. - -The working tree already contains an uncommitted keyset-pagination change and -the `3f4a5b6c7d8e` section-order migration. Keep those changes separate from -the serving-index implementation and from unrelated documentation edits. - -## 3. Architecture decision - -Use PostgreSQL as the source of truth and add a persistent, revision-pinned -serving read model. Do not add OpenSearch, Elasticsearch, Tantivy, or another -search service in this version. PostgreSQL's existing FTS plus `pg_trgm` keeps -the current scoring and tie-breaking behavior directly testable. - -### 3.1 Revision serving manifest - -Add one compressed manifest row per `(document_id, job_result_id)`. The payload -contains ordered metadata only: - -- document, revision, source filename, and job identity; -- section IDs, parent IDs, paths, titles, levels, summaries, and sort order; -- chunk IDs, section IDs, types, sort order, and `connect_to` target IDs; -- map-unit row IDs, unit IDs, unit kinds, token lengths, and sort order; -- root-asset IDs and remounted asset owners. - -It must not contain full chunk content or asset file paths. Those remain in the -canonical tables and are loaded lazily for selected evidence. - -Store the payload as canonical JSON compressed with the standard-library zlib -implementation, with a format version and checksum. The serving loader must -reject an unknown version, checksum mismatch, or incomplete payload. - -### 3.2 Serving generations and statistics - -Add namespace-scoped generation metadata and persistent scoring statistics: - -- `retrieval_namespace_generations`: current generation per user/namespace; -- `retrieval_serving_revision_stats`: compressed per-revision contributions; -- `retrieval_namespace_stats`: aggregate unit counts, total lengths, vocabulary - frequency histograms, and generation; -- `retrieval_namespace_token_stats`: queryable document frequency per channel - and token hash. - -The generation is a consistency marker, not a replacement for revision IDs. -Retrieval captures active revision IDs and one generation. If generation changes -while the snapshot is being captured or before scoring starts, retry once; if -consistency still cannot be proven, use the exact legacy reader. - -Every retrieval route must carry that capture as an immutable revision pin set: -`{document_id, namespace, job_result_id}` plus the captured generation. The pin -set is the source of truth for the request after capture. Downstream queries -must constrain sections, map units, chunks, connected assets, ranking lookups, -reference resolution, and result assembly by the pinned `job_result_id`; they -must not re-join through the live `Document.current_job_result_id`. A generation -change after scoring has started must never cause a mix of old and new rows: -finish against the captured pins (or return an exact legacy result), and only -retry before work that depends on the snapshot begins. -Snapshot admission is therefore decided at capture time. If a later archive -must suppress an in-flight result, discard/retry the whole request; do not -replace its pinned revision with the document's new current revision. - -Cache hits occur before route execution, so every operation that changes the -serving generation (publication, republish, archive, or namespace move) must -also advance the namespace retrieval-cache version, or store the generation in -the cache entry and reject mismatches. This keeps cached responses from -outliving the generation they represent without changing the public response -shape. - -### 3.3 Indexes - -Additive migrations should provide: - -- a token-first covering index for map-unit token candidates. The existing - `idx_document_map_unit_tokens_lookup` is token-first but does not cover the - selected columns; the pending `2e3f4a5b6c7d` migration adds a unit-first - covering index for a different access pattern, so the serving reader may - need one additional token-first covering index; -- a revision/section lookup index for map units; -- a trigram GIN index on `document_map_units.term_search_text_lower`; -- a generated lowercased term field and trigram GIN index for - `document_chunks.term_search_text`; -- the existing chunk ordering index plus the pending token-covering and - section-order migrations (`2e3f4a5b6c7d` and `3f4a5b6c7d8e`). - -Enable PostgreSQL's built-in `pg_trgm` extension. No separate search service is -required. - -## 4. Retrieval changes - -Capture the revision pin set and generation at retrieval-route entry, before -the small-corpus count or route selection. Pass that capture into whichever -route is selected; a route-local capture is allowed only when it is performed -as the same snapshot transaction. This prevents the count/load pair in the -small-corpus optimization from straddling a publication. - -### 4.1 Fast map-nav snapshot - -Extend `load_nav_snapshot()` to try the serving manifest first: - -1. Capture active documents, current revision IDs, and namespace generation in a - short read-only transaction, returning the immutable revision pin set with - the snapshot. -2. Fetch one manifest row per active revision. -3. Decode and validate manifests. -4. Apply the existing document and section exclusion predicates. -5. Build the current `LazyKnowhereProvider` and `LazyChunkRefIndex` from the - decoded metadata. -6. Pin the lazy chunk store to the captured revision IDs. -7. Verify generation stability before returning the snapshot. - -For an unfiltered map-nav request, route selection may count chunks from these -same validated manifests instead of scanning `document_chunks`; filtered and -classic requests retain the exact SQL counter. The count shortcut must fall -back when any manifest is missing or invalid. - -If any manifest is absent or invalid, use the existing legacy snapshot loader. -The legacy loader must return the same revision pin set and apply the same -downstream predicates. This fallback is automatic and exact; it is not a public -feature flag. - -The serving path must preserve current ordering, duplicate bare/document-scoped -reference keys, root-asset remounting, section filtering, and revision pinning. -Keep the pin set available through the complete map-nav request. After the LLM -episode, either materialize selected rows (including connected assets) from the -pinned lazy store before closing it, or pass the pin set to -`resolve_workflow_references()` and `assemble_retrieval_results()`. Their SQL -must select the captured `(document_id, job_result_id)` rows directly, so a -republish during the episode cannot make final citations resolve against the -new current revision. - -### 4.2 Exact persisted map scoring - -Keep `PersistedScoreCorpus` and the existing scorer unchanged wherever possible. -Replace only the data-loading strategy: - -- Prepare the immutable, revision-pinned unit projection and namespace scoring - statistics once per retrieval episode. Checklist relight waves must reuse that - projection; they may fetch or compute only query-specific postings/scores. - A wave must not issue another full-namespace unit/statistics load for the same - pin set. Instrument the loader call count and include it in the benchmark - report so repeated projection loads cannot hide behind separate wave timings. -- use manifest map-unit metadata to represent all units, including zero-score - units; the serving reader should not re-query `document_map_units` for these - IDs, lengths, or section membership; -- query token postings only for tokens in the request; -- filter postings by captured revisions and allowed sections; -- discover term-channel candidates through the trigram index while retaining the - current exact substring/token-hit scoring. The candidate predicate must use - the trigram-indexed `LIKE '%term%'` form (with the same lowercased query and - tokens), then apply the existing exact score expression; do not scan every - unit's term text in Python. -- obtain normal-corpus lengths, document frequencies, and IDF-flooring data from - persistent statistics; -- preserve the existing lexical sort key and RRF ranking. - -Queries with document or section exclusions must remain exact. If adjusted -statistics cannot be calculated with certainty, use the legacy scorer for that -request rather than approximating them. - -### 4.3 Classic retrieval — one pinned revision snapshot - -Keep the existing channel implementations and result projection. In -`bottom_discovery()`: - -- capture one revision pin set and generation before starting any channel; -- execute enabled channels concurrently; -- give each channel its own short-lived database session; -- pass the same pin set to every channel and constrain every channel query to - those revisions; -- preserve channel limits, Python BM25, term scoring, RRF merge, score - normalization, and all-or-error behavior; -- use the new trigram index only to narrow term candidates. - -Do not share one `AsyncSession` across concurrent channel tasks. -Ranking lookups, duplicate suppression, connected-target hydration, and final -assembly must receive the same pin set as discovery. The classic result must -therefore contain rows from one revision per document even if publication -replaces a document while one of the channel sessions is running. The -small-corpus optimization must use this same captured snapshot/pin contract (or -the exact legacy equivalent), rather than loading all rows through live current -revision joins. - -## 5. Publication and lifecycle behavior - -Refactor publication so the same build pass produces: - -- canonical sections/chunks; -- existing map-unit rows and completeness marker; -- the revision serving manifest; -- revision statistics and namespace-statistics deltas. - -All of this happens synchronously in the existing publication transaction. The -completeness marker and generation update are written last. If serving-index -construction fails, the publication transaction rolls back. - -New publication remains online during backfill. First publication, republish, -archive, and namespace-move paths must update statistics under the same -namespace generation row lock. The lock covers the active revision set, -namespace membership, revision contributions, and the generation increment, so -readers and writers have one lifecycle ordering. - -Backfill must rebuild the complete derived serving state (map units, manifest, -revision contribution, and namespace-statistics delta), not only the existing -map-unit index. It must select only documents with `status = 'active'`, a non-null -`current_job_result_id`, and the intended user/namespace. Immediately before -writing a contribution, it must hold the namespace lock and re-read the -document, then require all of the following to remain true: active status, -unchanged user/namespace, and `current_job_result_id` equal to the captured -revision. Otherwise it skips that revision without adding statistics. This -active-status predicate is required in the selector as well as in the -commit-time guard; update `apps/api/scripts/backfill_map_unit_indexes.py` to -include it in the existing selector. - -Archiving must atomically remove or invalidate that document revision's serving -statistics contribution while holding the same lock and advance the namespace -generation. `archive` currently changes `status` without clearing -`current_job_result_id`, so checking the revision pointer alone is insufficient -and would allow an in-flight backfill to re-add an archived revision. - -## 6. Online rollout - -There is no planned downtime, runtime feature flag, or production shadow-read -mode. - -1. Deploy additive schema/index migrations, beginning with the pending - `2e3f4a5b6c7d` and `3f4a5b6c7d8e` migrations. -2. Deploy code that automatically uses the serving reader only for complete, - valid revisions and otherwise uses the legacy reader. -3. Run an explicit, idempotent, bounded backfill for existing active revisions. -4. Keep retrieval and publication online while backfill runs. -5. Verify manifest checksums, revision coverage, namespace statistics, and - generation consistency. -6. Run strict legacy-versus-serving differential checks before considering the - rollout complete. - -If backfill is incomplete, affected revisions continue on the exact legacy -path. If online serving data is corrupted, reject it, alert, repair it with the -backfill/rebuild script, and do not serve partial data. - -Before enabling the serving reader for a namespace, record an inventory of -active `(document_id, current_job_result_id)` pairs, manifest completeness, and -expected per-revision and aggregate unit counts. After backfill, reconcile those -same values and verify that every aggregate includes only active, namespace- -member revisions. Abort the fast-path rollout on any missing/extra revision, -checksum failure, count mismatch, archived contribution, or generation -discontinuity. - -Any migration, backfill, or other database write—especially against -production—requires explicit approval immediately before execution. Read-only -inspection and benchmarking may proceed without that approval. - -## 6.1 DevOps operations runbook - -DevOps owns the production rollout mechanics; application code does not run a -startup backfill or create serving tables implicitly. Execute the following in -order: - -1. **Preflight (read-only):** confirm the target account, database, migration - head, available disk, connection headroom, and a recent rollback point. Record - the active `(document_id, current_job_result_id)` inventory for each namespace - that will be backfilled. -2. **Schema rollout:** with explicit approval immediately beforehand, apply the - additive migrations in dependency order: `2e3f4a5b6c7d`, - `3f4a5b6c7d8e`, `4a5b6c7d8e9f`, then `5b6c7d8e9f0a`. Run the trigram-index - migration during a low-traffic window and monitor for blocking locks. -3. **Application rollout:** deploy the API and worker versions containing the - serving reader and atomic publication changes. Verify health, error rate, and - legacy fallback before starting the backfill. -4. **Bounded backfill:** with separate approval, run - `uv run python apps/api/scripts/backfill_map_unit_indexes.py --apply` from a - controlled operator environment. Limit concurrency, pause on database - saturation, and resume safely; the operation is idempotent and stale or - inactive revisions must be skipped. -5. **Reconciliation:** compare the preflight inventory with serving manifests, - checksums, per-revision unit counts, namespace aggregates, and generation - values. Confirm aggregates contain only active documents still belonging to - the namespace. Investigate every missing, extra, stale, or invalid revision. -6. **Acceptance:** run the production read-only legacy-versus-serving - differential harness and record latency, selected IDs, order, scores, - citations, section paths, asset references, and fallback behavior. Declare - the rollout complete only after zero semantic mismatches. - -If migration or backfill must be stopped, leave the serving tables in place and -stop the operator job. The reader will continue using the exact legacy path for -incomplete revisions. Roll back application code first if necessary; do not -drop serving tables or indexes as an emergency rollback action. Repair a failed -revision by rerunning the bounded backfill after the cause is understood. - -## 7. Contract tests and benchmarks - -Use contract tests only. Add contracts for: - -- manifest round-trip, checksum, version, and revision pinning; -- eager versus serving snapshot equivalence; -- exclusions, duplicate chunk IDs, document-scoped references, and root assets; -- exact Latin/CJK, empty, no-hit, phrase, token-only, and negative-IDF cases; -- incomplete serving data falling back to legacy; -- publication replacement, archive deltas, concurrent generation changes, and - stale backfill protection; -- cache invalidation racing with a generation change, proving an old cached - response is not returned for a newer serving generation; -- map-nav republish during the LLM episode, proving final hydration and - connected-asset resolution stay on the captured revisions; -- classic publication replacement during concurrent channels, ranking, and - final assembly, proving every returned row shares the channel's pin set; -- archive/backfill races proving archived or namespace-moved revisions never - contribute to serving statistics; -- concurrent classic channels preserving IDs, order, scores, citations, and - fallback behavior. - -Race tests must use barriers or an equivalent deterministic hook to force a -republish during the map-nav episode, a publication between classic channel -sessions, an archive during backfill, and a namespace move during backfill. -Each test must assert both the returned evidence and the persisted statistics, -not merely that the request completed. - -The validation harness must also inspect the generated SQL/query plans (or an -equivalent query-boundary assertion) to prove pinned reads do not use live -`Document.current_job_result_id` joins. Run cache-version/generation races and -verify an old cached response is rejected after a lifecycle change. - -Run a differential harness against the production read-only database and -compare selected IDs, ordering, rounded scores, citations, section paths, -asset references, and fallback behavior. - -Benchmark fresh processes and uncached queries. Report separately: - -- serving capture/decode; -- map index projection and scoring; -- episode-local projection reuse (number of full projection loads and per-wave - query-only scoring time); -- classic discovery; -- ranking; -- hydration/assembly; -- total Retrieval Non-LLM Work; -- planner/harvest/control LLM time. - -The complexity check is explicit: one request may perform one full pinned -snapshot/projection pass, relight work should scale with query postings rather -than reloading the corpus, and hydration should scale with selected evidence -(`top_k`/references), not namespace size. Navigation wave count must not -multiply full-corpus database loads. - -Record peak resident memory for a fresh worker during the same benchmark and -repeat it with the expected concurrent-request level. Memory is reported as an -operational trade-off rather than a latency acceptance gate for this phase; -before production rollout, any episode-local or process-local reuse still needs -an explicit byte/item budget and an agreed worker ceiling. - -The fast path is accepted only after zero semantic mismatches and evidence that -the cold request performs one bounded serving projection, does not repeat -full-corpus loads per navigation wave, and meets an agreed latency budget for -the current production-sized corpus. - -## 8. Main tradeoffs and risks - -- Publication becomes slower and uses more storage because derived data is built - synchronously. -- Existing documents need an explicit backfill before they use the fast path. -- A serving-index inconsistency causes a slower legacy request, not approximate - evidence. -- PostgreSQL remains a scaling dependency; a future search-engine migration - would require a new semantic-parity review. diff --git a/packages/shared-python/shared/models/database/document.py b/packages/shared-python/shared/models/database/document.py index 739221c4..3d630b91 100644 --- a/packages/shared-python/shared/models/database/document.py +++ b/packages/shared-python/shared/models/database/document.py @@ -443,105 +443,6 @@ class RetrievalServingRevisionManifest(Base): ) -class RetrievalServingRevisionStat(Base): - """Compressed scoring contribution for one document revision.""" - - __tablename__ = "retrieval_serving_revision_stats" - - id: Mapped[str] = mapped_column( - String(100), primary_key=True, default=lambda: f"rss_{uuid4().hex}" - ) - user_id: Mapped[str] = mapped_column(Text, nullable=False) - namespace: Mapped[str] = mapped_column(String(255), nullable=False) - document_id: Mapped[str] = mapped_column( - String(36), ForeignKey("documents.document_id", ondelete="CASCADE"), nullable=False - ) - job_result_id: Mapped[str] = mapped_column( - String(36), ForeignKey("job_results.id", ondelete="CASCADE"), nullable=False - ) - format_version: Mapped[int] = mapped_column(Integer, nullable=False) - payload_zlib: Mapped[bytes] = mapped_column(LargeBinary, nullable=False) - checksum: Mapped[str] = mapped_column(String(64), nullable=False) - created_at: Mapped[datetime] = mapped_column( - DateTime, default=utc_now_naive, nullable=False - ) - - __table_args__ = ( - UniqueConstraint( - "document_id", - "job_result_id", - name="uq_retrieval_serving_revision_stats_revision", - ), - Index( - "idx_retrieval_serving_revision_stats_scope", - "user_id", - "namespace", - "document_id", - "job_result_id", - ), - ) - - -class RetrievalNamespaceStat(Base): - """Compressed aggregate scoring statistics for one namespace generation.""" - - __tablename__ = "retrieval_namespace_stats" - - id: Mapped[str] = mapped_column( - String(100), primary_key=True, default=lambda: f"rns_{uuid4().hex}" - ) - user_id: Mapped[str] = mapped_column(Text, nullable=False) - namespace: Mapped[str] = mapped_column(String(255), nullable=False) - generation: Mapped[int] = mapped_column(BigInteger, nullable=False) - payload_zlib: Mapped[bytes] = mapped_column(LargeBinary, nullable=False) - checksum: Mapped[str] = mapped_column(String(64), nullable=False) - updated_at: Mapped[datetime] = mapped_column( - DateTime, default=utc_now_naive, onupdate=utc_now_naive, nullable=False - ) - - __table_args__ = ( - UniqueConstraint( - "user_id", - "namespace", - name="uq_retrieval_namespace_stats_scope", - ), - ) - - -class RetrievalNamespaceTokenStat(Base): - """Document frequency for one token/channel in a namespace generation.""" - - __tablename__ = "retrieval_namespace_token_stats" - - id: Mapped[str] = mapped_column( - String(100), primary_key=True, default=lambda: f"rnt_{uuid4().hex}" - ) - user_id: Mapped[str] = mapped_column(Text, nullable=False) - namespace: Mapped[str] = mapped_column(String(255), nullable=False) - generation: Mapped[int] = mapped_column(BigInteger, nullable=False) - channel: Mapped[str] = mapped_column(String(32), nullable=False) - token_hash: Mapped[str] = mapped_column(String(64), nullable=False) - document_frequency: Mapped[int] = mapped_column(Integer, nullable=False) - - __table_args__ = ( - UniqueConstraint( - "user_id", - "namespace", - "channel", - "token_hash", - name="uq_retrieval_namespace_token_stats_key", - ), - Index( - "idx_retrieval_namespace_token_stats_lookup", - "user_id", - "namespace", - "generation", - "channel", - "token_hash", - ), - ) - - class RetrievalNamespaceMapSnapshot(Base): """Persisted namespace-level MAP (sections + chunk index + map units). diff --git a/packages/shared-python/shared/services/retrieval/namespace_map_snapshot.py b/packages/shared-python/shared/services/retrieval/namespace_map_snapshot.py index bd8b7767..41fa6c6b 100644 --- a/packages/shared-python/shared/services/retrieval/namespace_map_snapshot.py +++ b/packages/shared-python/shared/services/retrieval/namespace_map_snapshot.py @@ -2,9 +2,8 @@ Callers must already hold the namespace generation lock (see ``serving_generation.lock_namespace_generation``) before calling either -function here, exactly as ``rebuild_namespace_serving_statistics`` requires. -Each call only touches one document's subtree; every other document's -subtree in the payload is left byte-for-byte unchanged. +function here. Each call only touches one document's subtree; every other +document's subtree in the payload is left byte-for-byte unchanged. """ from __future__ import annotations @@ -81,8 +80,7 @@ def remove_document_from_namespace_map_snapshot( def _target_generation(db: Session, *, user_id: str, namespace: str) -> int: """Namespace generation this snapshot is prepared for (current + 1). - Mirrors ``rebuild_namespace_serving_statistics``: callers advance the - generation after this write, in the same transaction. + Callers advance the generation after this write, in the same transaction. """ generation = db.execute( select(RetrievalNamespaceGeneration) diff --git a/packages/shared-python/shared/services/retrieval/publication_service.py b/packages/shared-python/shared/services/retrieval/publication_service.py index 84b3b2d2..7ee5dec2 100644 --- a/packages/shared-python/shared/services/retrieval/publication_service.py +++ b/packages/shared-python/shared/services/retrieval/publication_service.py @@ -38,9 +38,6 @@ advance_namespace_generation, lock_namespace_generation, ) -from shared.services.retrieval.serving_manifest import ( - rebuild_namespace_serving_statistics, -) def utc_now_naive() -> datetime: @@ -183,17 +180,7 @@ def _publish_document_state_for_job( ) db.flush() - rebuild_namespace_serving_statistics( - db, - user_id=scope.user_id, - namespace=scope.namespace, - ) if existing_namespace and str(existing_namespace) != scope.namespace: - rebuild_namespace_serving_statistics( - db, - user_id=scope.user_id, - namespace=str(existing_namespace), - ) remove_document_from_namespace_map_snapshot( db, user_id=scope.user_id, diff --git a/packages/shared-python/shared/services/retrieval/serving_manifest.py b/packages/shared-python/shared/services/retrieval/serving_manifest.py index 467d55c9..c66ae579 100644 --- a/packages/shared-python/shared/services/retrieval/serving_manifest.py +++ b/packages/shared-python/shared/services/retrieval/serving_manifest.py @@ -4,8 +4,6 @@ import hashlib import json -import logging -import time import zlib from typing import Any @@ -15,20 +13,13 @@ from shared.models.database.document import ( Document, DocumentChunk, - DocumentMapUnit, - DocumentMapUnitToken, DocumentSection, - RetrievalNamespaceGeneration, - RetrievalNamespaceStat, - RetrievalNamespaceTokenStat, RetrievalServingRevisionManifest, - RetrievalServingRevisionStat, ) from shared.models.database.job_result import JobResult from shared.services.retrieval.publication_models import DocumentPublicationScope SERVING_MANIFEST_FORMAT_VERSION = 1 -_logger = logging.getLogger(__name__) def build_revision_serving_payload( @@ -121,82 +112,25 @@ def build_revision_serving_payload( } -def build_revision_statistics_payload( - db: Session, - *, - scope: DocumentPublicationScope, -) -> dict[str, Any]: - """Build compressed scoring contributions for one revision. - - Stores aggregate token frequencies for namespace statistics rebuild. - Per-unit frequencies stay in ``document_map_unit_tokens`` and are loaded - at query time by the query tokens only. - """ - units = list( - db.scalars( - select(DocumentMapUnit) - .where(DocumentMapUnit.document_id == scope.document_id) - .where(DocumentMapUnit.job_result_id == scope.job_result_id) - ) - ) - unit_ids = [unit.id for unit in units] - frequencies: dict[str, dict[str, int]] = {"path": {}, "content": {}} - if unit_ids: - for _map_unit_id, channel, token, frequency in db.execute( - select( - DocumentMapUnitToken.map_unit_id, - DocumentMapUnitToken.channel, - DocumentMapUnitToken.token, - DocumentMapUnitToken.frequency, - ).where(DocumentMapUnitToken.map_unit_id.in_(unit_ids)) - ).all(): - channel_key = str(channel) - if channel_key not in frequencies: - continue - token_key = str(token) - frequencies[channel_key][token_key] = frequencies[channel_key].get( - token_key, 0 - ) + int(frequency) - return { - "document_id": scope.document_id, - "job_result_id": scope.job_result_id, - "unit_count": len(units), - "path_token_count": sum(int(unit.path_token_count or 0) for unit in units), - "content_token_count": sum( - int(unit.content_token_count or 0) for unit in units - ), - "token_frequencies": frequencies, - } - - def persist_revision_serving_state( db: Session, *, scope: DocumentPublicationScope, ) -> dict[str, Any]: - """Replace manifest and statistics rows for one revision atomically. + """Replace the serving manifest row for one revision atomically. Returns the manifest payload so callers can patch the namespace-level MAP snapshot without rebuilding it. """ manifest_payload = build_revision_serving_payload(db, scope=scope) - statistics_payload = build_revision_statistics_payload(db, scope=scope) manifest_bytes, manifest_checksum, manifest_version = encode_serving_manifest( manifest_payload ) - statistics_bytes, statistics_checksum, statistics_version = encode_serving_manifest( - statistics_payload - ) db.execute( delete(RetrievalServingRevisionManifest) .where(RetrievalServingRevisionManifest.document_id == scope.document_id) .where(RetrievalServingRevisionManifest.job_result_id == scope.job_result_id) ) - db.execute( - delete(RetrievalServingRevisionStat) - .where(RetrievalServingRevisionStat.document_id == scope.document_id) - .where(RetrievalServingRevisionStat.job_result_id == scope.job_result_id) - ) db.add( RetrievalServingRevisionManifest( user_id=scope.user_id, @@ -208,144 +142,9 @@ def persist_revision_serving_state( checksum=manifest_checksum, ) ) - db.add( - RetrievalServingRevisionStat( - user_id=scope.user_id, - namespace=scope.namespace, - document_id=scope.document_id, - job_result_id=scope.job_result_id, - format_version=statistics_version, - payload_zlib=statistics_bytes, - checksum=statistics_checksum, - ) - ) return manifest_payload -def rebuild_namespace_serving_statistics( - db: Session, - *, - user_id: str, - namespace: str, -) -> int: - """Recompute namespace aggregates from active current revisions. - - Callers hold the namespace generation lock. The aggregate is prepared for - the generation that the caller will publish next. - """ - started = time.perf_counter() - generation = db.execute( - select(RetrievalNamespaceGeneration) - .where(RetrievalNamespaceGeneration.user_id == user_id) - .where(RetrievalNamespaceGeneration.namespace == namespace) - .with_for_update() - ).scalar_one() - target_generation = int(generation.generation) + 1 - revisions = { - (str(document_id), str(job_result_id)) - for document_id, job_result_id in db.execute( - select(Document.document_id, Document.current_job_result_id) - .where(Document.user_id == user_id) - .where(Document.namespace == namespace) - .where(Document.status == "active") - .where(Document.current_job_result_id.is_not(None)) - ).all() - if document_id and job_result_id - } - aggregate: dict[str, Any] = { - "document_count": 0, - "unit_count": 0, - "path_token_count": 0, - "content_token_count": 0, - "token_frequencies": {"path": {}, "content": {}}, - } - document_frequencies: dict[tuple[str, str], int] = {} - for row in db.scalars( - select(RetrievalServingRevisionStat) - .where(RetrievalServingRevisionStat.user_id == user_id) - .where(RetrievalServingRevisionStat.namespace == namespace) - ): - if (row.document_id, row.job_result_id) not in revisions: - continue - payload = decode_serving_manifest( - row.payload_zlib, - checksum=row.checksum, - format_version=row.format_version, - ) - aggregate["document_count"] += 1 - aggregate["unit_count"] += int(payload.get("unit_count", 0)) - aggregate["path_token_count"] += int(payload.get("path_token_count", 0)) - aggregate["content_token_count"] += int(payload.get("content_token_count", 0)) - token_frequencies = payload.get("token_frequencies", {}) - if not isinstance(token_frequencies, dict): - continue - for channel, values in token_frequencies.items(): - if channel not in aggregate["token_frequencies"] or not isinstance( - values, dict - ): - continue - for token, value in values.items(): - token_key = str(token) - aggregate["token_frequencies"][channel][token_key] = aggregate[ - "token_frequencies" - ][channel].get(token_key, 0) + int(value) - if int(value) > 0: - key = (str(channel), token_key) - document_frequencies[key] = document_frequencies.get(key, 0) + 1 - - encoded, checksum, _version = encode_serving_manifest(aggregate) - namespace_stat = db.execute( - select(RetrievalNamespaceStat) - .where(RetrievalNamespaceStat.user_id == user_id) - .where(RetrievalNamespaceStat.namespace == namespace) - ).scalar_one_or_none() - if namespace_stat is None: - db.add( - RetrievalNamespaceStat( - user_id=user_id, - namespace=namespace, - generation=target_generation, - payload_zlib=encoded, - checksum=checksum, - ) - ) - else: - namespace_stat.generation = target_generation - namespace_stat.payload_zlib = encoded - namespace_stat.checksum = checksum - db.execute( - delete(RetrievalNamespaceTokenStat) - .where(RetrievalNamespaceTokenStat.user_id == user_id) - .where(RetrievalNamespaceTokenStat.namespace == namespace) - ) - db.add_all( - [ - RetrievalNamespaceTokenStat( - user_id=user_id, - namespace=namespace, - generation=target_generation, - channel=channel, - token_hash=hashlib.sha256(token.encode("utf-8")).hexdigest(), - document_frequency=frequency, - ) - for (channel, token), frequency in document_frequencies.items() - ] - ) - db.flush() - _logger.info( - "retrieval namespace statistics rebuilt user_id=%s namespace=%s " - "generation=%d documents=%d units=%d token_stats=%d seconds=%.3f", - user_id, - namespace, - target_generation, - aggregate["document_count"], - aggregate["unit_count"], - len(document_frequencies), - time.perf_counter() - started, - ) - return target_generation - - def _connection_target_ids(metadata: Any) -> list[str]: if not isinstance(metadata, dict): return []