From 6a4e82edb7f1e0520853362eaf802ea40bb19980 Mon Sep 17 00:00:00 2001 From: suguanYang Date: Sat, 10 Oct 2026 17:19:39 +0800 Subject: [PATCH] fix: reject duplicate document source content --- CONTEXT.md | 12 + ...f7a8b9c0d_add_source_content_hash_index.py | 71 +++ .../document_ingestion/creation_service.py | 39 -- .../document_ingestion/scope_service.py | 40 -- .../contract/test_job_creation_contract.py | 90 ++-- .../tests/migrations/test_schema_contract.py | 108 ++++- ...t_document_ingestion_duplicate_filename.py | 207 --------- .../source_content_admission.py | 163 +++++++ .../document_ingestion/source_preparation.py | 9 +- .../contract/test_parse_task_contract.py | 121 ++++- .../test_source_content_admission_contract.py | 420 ++++++++++++++++++ docs/source-content-deduplication-review.md | 58 +++ .../shared/models/database/job.py | 6 + .../shared/models/schemas/job_metadata.py | 6 + 14 files changed, 1005 insertions(+), 345 deletions(-) create mode 100644 apps/api/alembic/versions/5e6f7a8b9c0d_add_source_content_hash_index.py delete mode 100644 apps/api/tests/unit/test_document_ingestion_duplicate_filename.py create mode 100644 apps/worker/app/services/document_ingestion/source_content_admission.py create mode 100644 apps/worker/tests/contract/test_source_content_admission_contract.py create mode 100644 docs/source-content-deduplication-review.md diff --git a/CONTEXT.md b/CONTEXT.md index 5b465357f..faaae583a 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -69,6 +69,18 @@ state, and starts parsing work. The API-side handoff that advances an uploaded file Job to pending state and enqueues the worker parse task with the user-aware Celery queue policy. +### Source Content Admission + +The check that rejects identical original file bytes within one User and +Namespace before parsing or billing, including unchanged Document updates. +Filenames do not determine duplicates. Only current active Document revisions +and in-progress Jobs participate; failed Jobs and archived or superseded +revisions do not block retries. The check happens after source upload or URL +download, so a duplicate is reported through the normal Job failure contract. +Demo Corpus publication retains its separate admission policy. +Existing sources without recorded fingerprints do not participate, and their +original files are not fetched for comparison. + ### Worker Document Parsing The worker-side workflow that turns a source file into parsed DataFrame rows, diff --git a/apps/api/alembic/versions/5e6f7a8b9c0d_add_source_content_hash_index.py b/apps/api/alembic/versions/5e6f7a8b9c0d_add_source_content_hash_index.py new file mode 100644 index 000000000..b47198205 --- /dev/null +++ b/apps/api/alembic/versions/5e6f7a8b9c0d_add_source_content_hash_index.py @@ -0,0 +1,71 @@ +"""Index future source content fingerprints without backfilling existing Jobs.""" + +from __future__ import annotations + +from collections.abc import Sequence + +from alembic import op +from sqlalchemy import text + +revision: str = "5e6f7a8b9c0d" +down_revision: str = "4d5e6f7a8b9c" +branch_labels: Sequence[str] | None = None +depends_on: Sequence[str] | None = None + + +def _create_index(*, concurrently: bool) -> None: + existing_index: tuple[bool, str, str] | None = ( + op.get_bind() + .execute( + text( + "SELECT indexes.indisvalid AND indexes.indisready, " + "tables.relname, pg_get_indexdef(indexes.indexrelid) " + "FROM pg_index AS indexes " + "JOIN pg_class AS classes ON classes.oid = indexes.indexrelid " + "JOIN pg_class AS tables ON tables.oid = indexes.indrelid " + "JOIN pg_namespace AS namespaces ON namespaces.oid = classes.relnamespace " + "WHERE namespaces.nspname = current_schema() " + "AND classes.relname = 'idx_jobs_user_source_content_hash'" + ) + ) + .tuples() + .first() + ) + if existing_index is not None: + is_ready, table_name, definition = existing_index + normalized_definition: str = " ".join(definition.split()) + intended_definition: str = ( + "USING btree (user_id, ((job_metadata ->> 'source_content_sha256'::text))) " + "WHERE ((job_metadata ->> 'source_content_sha256'::text) IS NOT NULL)" + ) + if ( + is_ready + and table_name == "jobs" + and normalized_definition.startswith("CREATE INDEX ") + and normalized_definition.endswith(intended_definition) + ): + return + concurrent_clause: str = "CONCURRENTLY " if concurrently else "" + op.execute( + f"DROP INDEX {concurrent_clause}IF EXISTS idx_jobs_user_source_content_hash" + ) + op.execute( + f"CREATE INDEX {concurrent_clause}IF NOT EXISTS idx_jobs_user_source_content_hash " + "ON jobs (user_id, (job_metadata ->> 'source_content_sha256')) " + "WHERE job_metadata ->> 'source_content_sha256' IS NOT NULL" + ) + + +def upgrade() -> None: + uses_external_transaction: bool = bool( + op.get_context().opts.get("knowhere_external_transaction", False) + ) + if uses_external_transaction: + _create_index(concurrently=False) + return + with op.get_context().autocommit_block(): + _create_index(concurrently=True) + + +def downgrade() -> None: + op.execute("DROP INDEX IF EXISTS idx_jobs_user_source_content_hash") diff --git a/apps/api/app/services/document_ingestion/creation_service.py b/apps/api/app/services/document_ingestion/creation_service.py index 79e2227ca..ec88c65ce 100644 --- a/apps/api/app/services/document_ingestion/creation_service.py +++ b/apps/api/app/services/document_ingestion/creation_service.py @@ -10,10 +10,8 @@ from app.repositories.job_repository import JobRepository from app.services.document_ingestion.command import DocumentIngestionCommand from app.services.document_ingestion.scope_service import ( - find_active_document_by_source_file_name, is_active_document_job_unique_violation, raise_document_ingestion_conflict, - raise_duplicate_filename_conflict, ) from app.services.jobs.result_projection import to_job_status_value from app.services.rate_limit.data_structures import CurrentUser @@ -147,29 +145,6 @@ async def _cache_job_creation_state( } await job_info_service.save_job_info(job_id, job_info) - async def _reject_duplicate_source_file_name( - self, - db: AsyncSession, - *, - payload: JobCreateBase, - current_user: CurrentUser, - scope: ResolvedDocumentIngestionScope, - source_file_name: str, - ) -> None: - if payload.document_id or scope.job_metadata.get("corpus_target") == "DEMO": - return - existing = await find_active_document_by_source_file_name( - db, - user_id=current_user.user_id, - namespace=scope.namespace, - source_file_name=source_file_name, - ) - if existing is not None: - raise_duplicate_filename_conflict( - existing_document_id=existing.document_id, - source_file_name=source_file_name, - ) - async def _create_file_job( self, db: AsyncSession, @@ -180,13 +155,6 @@ async def _create_file_job( scope: ResolvedDocumentIngestionScope, ) -> JobResponse: assert payload.file_name is not None - await self._reject_duplicate_source_file_name( - db, - payload=payload, - current_user=current_user, - scope=scope, - source_file_name=payload.file_name, - ) file_extension = os.path.splitext(payload.file_name)[1] s3_key = f"uploads/{job_id}{file_extension}" JobMetadataHelper.set_file_source( @@ -264,13 +232,6 @@ async def _create_url_job( source_url=payload.source_url, file_extension=file_extension, ) - await self._reject_duplicate_source_file_name( - db, - payload=payload, - current_user=current_user, - scope=scope, - source_file_name=source_file_name, - ) s3_key = f"uploads/{job_id}{file_extension}" JobMetadataHelper.set_url_source( scope.job_metadata, diff --git a/apps/api/app/services/document_ingestion/scope_service.py b/apps/api/app/services/document_ingestion/scope_service.py index 842bade73..61e3ea268 100644 --- a/apps/api/app/services/document_ingestion/scope_service.py +++ b/apps/api/app/services/document_ingestion/scope_service.py @@ -16,7 +16,6 @@ NotFoundException, ValidationException, ) -from shared.models.database.document import Document from shared.models.database.job import Job from shared.models.schemas.retrieval_namespace import normalize_retrieval_namespace @@ -68,45 +67,6 @@ def raise_document_ingestion_conflict( ) -async def find_active_document_by_source_file_name( - db: AsyncSession, - *, - user_id: str, - namespace: str, - source_file_name: str, -) -> Document | None: - result = await db.execute( - select(Document) - .where(Document.user_id == user_id) - .where(Document.namespace == namespace) - .where(Document.status == "active") - .where(Document.source_file_name == source_file_name) - .order_by(Document.created_at.asc(), Document.document_id.asc()) - .limit(1) - ) - return result.scalars().first() - - -def raise_duplicate_filename_conflict( - *, - existing_document_id: str, - source_file_name: str, -) -> None: - raise ConflictException( - user_message=( - f"A document named {source_file_name!r} already exists. " - "To replace it, retry with document_id set to details.id." - ), - reason="ALREADY_EXISTS", - resource="Document", - resource_id=existing_document_id, - internal_message=( - "Duplicate source_file_name=" - f"{source_file_name!r} document_id={existing_document_id}" - ), - ) - - async def resolve_effective_document_scope( db: AsyncSession, *, diff --git a/apps/api/tests/contract/test_job_creation_contract.py b/apps/api/tests/contract/test_job_creation_contract.py index 4c578eefe..b47bdad6d 100644 --- a/apps/api/tests/contract/test_job_creation_contract.py +++ b/apps/api/tests/contract/test_job_creation_contract.py @@ -874,7 +874,7 @@ async def test_should_inherit_existing_document_namespace_when_update_namespace_ @pytest.mark.asyncio -async def test_should_return_conflict_when_creating_a_job_for_a_duplicate_file_name( +async def test_should_accept_same_file_name_before_source_content_is_available( developer_api_client_factory: Callable[ [], AbstractAsyncContextManager[AsyncClient] ], @@ -894,21 +894,12 @@ async def test_should_return_conflict_when_creating_a_job_for_a_duplicate_file_n response = await api_client.post("/api/v1/jobs", json=payload) jobs_after = await _count_jobs() - assert response.status_code == 409 + assert response.status_code == 200 response_json: dict[str, object] = response.json() - error = cast(dict[str, object], response_json["error"]) - assert response_json["success"] is False - assert error["code"] == "ALREADY_EXISTS" - assert error["message"] == ( - f"A document named {file_name!r} already exists. " - "To replace it, retry with document_id set to details.id." - ) - assert error["details"] == { - "reason": "ALREADY_EXISTS", - "resource": "Document", - "id": document_id, - } - assert jobs_after == jobs_before + assert response_json["status"] == "waiting-file" + assert response_json["document_id"] != document_id + assert response_json["upload_url"] + assert jobs_after == jobs_before + 1 @pytest.mark.asyncio @@ -967,7 +958,7 @@ async def test_should_allow_same_file_name_in_a_different_namespace( @pytest.mark.asyncio -async def test_should_return_conflict_when_url_source_resolves_to_a_duplicate_file_name( +async def test_should_accept_url_source_with_same_resolved_file_name( monkeypatch: MonkeyPatch, developer_api_client_factory: Callable[ [], AbstractAsyncContextManager[AsyncClient] @@ -981,6 +972,7 @@ async def test_should_return_conflict_when_url_source_resolves_to_a_duplicate_fi "source_url": f"https://example.com/contracts/{file_name}", "data_id": "contract-job-duplicate-filename-url", } + scheduled_tasks: list[dict[str, object]] = [] class _FakeHeadResponse: def __init__(self) -> None: @@ -998,6 +990,25 @@ async def head( assert follow_redirects is False return _FakeHeadResponse() + class _FakeCeleryTask: + def apply_async( + self, + *, + args: list[object], + kwargs: dict[str, object], + ) -> None: + scheduled_tasks.append({"args": args, "kwargs": kwargs}) + + class _FakeCeleryApp: + def __init__(self) -> None: + from types import SimpleNamespace + + self.conf = SimpleNamespace(task_routes={}) + + def signature(self, task_name: str) -> _FakeCeleryTask: + assert task_name == "app.core.tasks.document_ingestion_tasks.upload_url_file_task" + return _FakeCeleryTask() + def resolve_public_address( host: str, port: int | None, @@ -1007,6 +1018,7 @@ def resolve_public_address( del host, port, args, kwargs return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 0))] + import shared.core.celery_app as celery_app_module import shared.services.http.client_pool as client_pool_module monkeypatch.setattr(socket, "getaddrinfo", resolve_public_address) @@ -1015,26 +1027,38 @@ def resolve_public_address( "get_async_client", lambda: _FakeAsyncHttpClient(), ) + monkeypatch.setattr( + celery_app_module, + "get_celery_app", + lambda: _FakeCeleryApp(), + ) async with developer_api_client_factory() as api_client: await _insert_document(document_id=document_id, source_file_name=file_name) jobs_before = await _count_jobs() response = await api_client.post("/api/v1/jobs", json=payload) jobs_after = await _count_jobs() + response_json: dict[str, object] = response.json() + job_id = cast(str, response_json["job_id"]) + job_metadata = cast( + dict[str, object], (await _load_job_record(job_id))["job_metadata"] + ) - assert response.status_code == 409 - error = cast(dict[str, object], response.json()["error"]) - assert error["code"] == "ALREADY_EXISTS" - assert error["details"] == { - "reason": "ALREADY_EXISTS", - "resource": "Document", - "id": document_id, - } - assert jobs_after == jobs_before + assert response.status_code == 200 + assert response_json["source_type"] == "url" + assert response_json["document_id"] != document_id + assert job_metadata["source_file_name"] == file_name + assert jobs_after == jobs_before + 1 + assert scheduled_tasks == [ + { + "args": [job_id, payload["source_url"], "local-dev-user"], + "kwargs": {"job_type": "document_ingestion"}, + } + ] @pytest.mark.asyncio -async def test_should_return_conflict_when_creating_a_v2_job_for_a_duplicate_file_name( +async def test_should_accept_v2_job_with_same_file_name_before_upload( developer_api_client_factory: Callable[ [], AbstractAsyncContextManager[AsyncClient] ], @@ -1054,15 +1078,11 @@ async def test_should_return_conflict_when_creating_a_v2_job_for_a_duplicate_fil response = await api_client.post("/api/v2/jobs", json=payload) jobs_after = await _count_jobs() - assert response.status_code == 409 - error = cast(dict[str, object], response.json()["error"]) - assert error["code"] == "ALREADY_EXISTS" - assert error["details"] == { - "reason": "ALREADY_EXISTS", - "resource": "Document", - "id": document_id, - } - assert jobs_after == jobs_before + assert response.status_code == 200 + response_json: dict[str, object] = response.json() + assert response_json["status"] == "waiting-file" + assert response_json["document_id"] != document_id + assert jobs_after == jobs_before + 1 @pytest.mark.asyncio diff --git a/apps/api/tests/migrations/test_schema_contract.py b/apps/api/tests/migrations/test_schema_contract.py index 9c0438184..5961dadf2 100644 --- a/apps/api/tests/migrations/test_schema_contract.py +++ b/apps/api/tests/migrations/test_schema_contract.py @@ -9,7 +9,7 @@ import pytest from alembic import command from alembic.config import Config -from sqlalchemy import text +from sqlalchemy import Select, select, text from sqlalchemy.engine import Connection, Engine from sqlalchemy.exc import IntegrityError @@ -206,6 +206,112 @@ def test_should_allow_a_new_active_document_job_after_a_terminal_job( assert int(result.scalar_one()) == 2 +@pytest.mark.parametrize("external_transaction", [False, True]) +@pytest.mark.parametrize( + "existing_index", + [ + "absent", "wrong-columns", "wrong-predicate", "wrong-key-case", + "wrong-key-suffix", "intended", + ], +) +def test_should_index_future_source_hashes_without_backfilling_existing_jobs( + alembic_engine: Engine, + external_transaction: bool, + existing_index: str, +) -> None: + from shared.models.database.job import Job + + config: Config = _build_alembic_command_config(engine=alembic_engine) + command.upgrade(config, "4d5e6f7a8b9c") + user_id: str = f"hash-index-user-{uuid4().hex[:12]}" + job_id: str = f"job_hash_{uuid4().hex[:12]}" + with alembic_engine.begin() as connection: + insert_contract_user(connection, user_id=user_id) + _insert_job( + connection, + job_id=job_id, + user_id=user_id, + document_id=f"doc_hash_{uuid4().hex[:12]}", + status="done", + ) + index_definitions: dict[str, str] = { + "wrong-columns": "ON jobs (status)", + "wrong-predicate": ( + "ON jobs (user_id, (job_metadata ->> 'source_content_sha256')) " + "WHERE status = 'done'" + ), + "wrong-key-case": ( + "ON jobs (user_id, (job_metadata ->> 'SOURCE_CONTENT_SHA256')) " + "WHERE job_metadata ->> 'SOURCE_CONTENT_SHA256' IS NOT NULL" + ), + "wrong-key-suffix": ( + "ON jobs (user_id, (job_metadata ->> 'source_content_sha256::text')) " + "WHERE job_metadata ->> 'source_content_sha256::text' IS NOT NULL" + ), + "intended": ( + "ON jobs (user_id, (job_metadata ->> 'source_content_sha256')) " + "WHERE job_metadata ->> 'source_content_sha256' IS NOT NULL" + ), + } + if existing_index != "absent": + connection.execute( + text( + "CREATE INDEX idx_jobs_user_source_content_hash " + + index_definitions[existing_index] + ) + ) + previous_index_id: int | None = connection.execute( + text("SELECT to_regclass('idx_jobs_user_source_content_hash')::oid") + ).scalar_one() + if external_transaction: + _upgrade_to_heads_with_external_connection(engine=alembic_engine) + else: + _upgrade_to_heads(engine=alembic_engine) + + with alembic_engine.begin() as connection: + current_index_id: int = connection.execute( + text("SELECT to_regclass('idx_jobs_user_source_content_hash')::oid") + ).scalar_one() + if existing_index == "intended": + assert current_index_id == previous_index_id + elif existing_index != "absent": + assert current_index_id != previous_index_id + assert connection.execute( + text("SELECT job_metadata ->> 'source_content_sha256' FROM jobs WHERE job_id = :job_id"), + {"job_id": job_id}, + ).scalar_one() is None + connection.execute( + text("UPDATE jobs SET job_metadata = jsonb_set(job_metadata::jsonb, '{source_content_sha256}', to_jsonb(CAST(:hash AS text)))::json WHERE job_id = :job_id"), + {"job_id": job_id, "hash": "a" * 64}, + ) + connection.execute(text("SET LOCAL enable_seqscan = off")) + lookup: Select[tuple[str]] = ( + select(Job.job_id) + .where(Job.user_id == user_id) + .where(Job.job_metadata["source_content_sha256"].as_string() == "a" * 64) + .limit(1) + ) + compiled_lookup: str = str( + lookup.compile( + dialect=connection.dialect, + compile_kwargs={"literal_binds": True}, + ) + ) + plan: str = "\n".join( + str(row[0]) for row in connection.execute( + text(f"EXPLAIN {compiled_lookup}"), + ) + ) + assert "idx_jobs_user_source_content_hash" in plan + index_conditions: list[str] = [ + line for line in plan.splitlines() if "Index Cond:" in line + ] + assert any( + "user_id" in condition and "source_content_sha256" in condition + for condition in index_conditions + ) + + def test_should_seed_v2_job_polling_system_limit( migrated_head_engine: Engine, ) -> None: diff --git a/apps/api/tests/unit/test_document_ingestion_duplicate_filename.py b/apps/api/tests/unit/test_document_ingestion_duplicate_filename.py deleted file mode 100644 index 7a9ba68f2..000000000 --- a/apps/api/tests/unit/test_document_ingestion_duplicate_filename.py +++ /dev/null @@ -1,207 +0,0 @@ -"""Service-entry tests for same-name document conflict on create (no document_id).""" - -from __future__ import annotations - -from datetime import datetime - -import pytest -from sqlalchemy import create_engine, event -from sqlalchemy.orm import Session -from tests.support.import_environment import ( - configure_import_environment, - ensure_import_paths, -) - -configure_import_environment() -ensure_import_paths() - -from app.services.document_ingestion.creation_service import ( # noqa: E402 - DocumentIngestionCreationService, - ResolvedDocumentIngestionScope, -) -from app.services.document_ingestion.scope_service import ( # noqa: E402 - find_active_document_by_source_file_name, -) -from app.services.rate_limit.data_structures import CurrentUser # noqa: E402 -from shared.core.exceptions.domain_exceptions import ConflictException # noqa: E402 -from shared.models.database.document import Document # noqa: E402 -from shared.models.schemas.job import JobCreate # noqa: E402 - - -USER_ID = "user_dup" -NAMESPACE = "default" -FILE_NAME = "01-p1-70.pdf" -EXISTING_DOC = "doc_existing12" -OLDER_DOC = "doc_older00001" -NEWER_DOC = "doc_newer00002" - - -class _AsyncSessionAdapter: - def __init__(self, session: Session) -> None: - self._session = session - - async def execute(self, statement): # noqa: ANN001 - return self._session.execute(statement) - - -def _seed_document( - session: Session, - *, - document_id: str, - source_file_name: str, - namespace: str = NAMESPACE, - status: str = "active", - user_id: str = USER_ID, - created_at: datetime | None = None, -) -> None: - now = created_at or datetime(2026, 1, 1) - session.add( - Document( - document_id=document_id, - user_id=user_id, - namespace=namespace, - status=status, - source_file_name=source_file_name, - parse_track="chunk", - created_at=now, - updated_at=now, - ) - ) - session.commit() - - -@pytest.fixture -def db_session() -> Session: - engine = create_engine("sqlite:///:memory:") - - @event.listens_for(engine, "connect") - def _disable_fk(dbapi_connection, _connection_record) -> None: # noqa: ANN001 - dbapi_connection.execute("PRAGMA foreign_keys=OFF") - - Document.__table__.create(engine) - session = Session(engine) - _seed_document(session, document_id=EXISTING_DOC, source_file_name=FILE_NAME) - return session - - -def _file_payload(*, document_id: str | None = None) -> JobCreate: - payload: dict[str, str] = { - "source_type": "file", - "file_name": FILE_NAME, - "namespace": NAMESPACE, - } - if document_id: - payload["document_id"] = document_id - return JobCreate.model_validate(payload) - - -def _scope() -> ResolvedDocumentIngestionScope: - return ResolvedDocumentIngestionScope( - job_metadata={}, - document_id="doc_newscope01", - namespace=NAMESPACE, - ) - - -async def _reject( - db_session: Session, - *, - payload: JobCreate, -) -> None: - await DocumentIngestionCreationService()._reject_duplicate_source_file_name( - _AsyncSessionAdapter(db_session), # type: ignore[arg-type] - payload=payload, - current_user=CurrentUser(user_id=USER_ID, user_tier="pro"), - scope=_scope(), - source_file_name=FILE_NAME, - ) - - -@pytest.mark.asyncio -async def test_find_returns_oldest_active_document_for_exact_name( - db_session: Session, -) -> None: - _seed_document( - db_session, - document_id=NEWER_DOC, - source_file_name=FILE_NAME, - created_at=datetime(2026, 3, 1), - ) - _seed_document( - db_session, - document_id=OLDER_DOC, - source_file_name=FILE_NAME, - created_at=datetime(2025, 12, 1), - ) - found = await find_active_document_by_source_file_name( - _AsyncSessionAdapter(db_session), # type: ignore[arg-type] - user_id=USER_ID, - namespace=NAMESPACE, - source_file_name=FILE_NAME, - ) - assert found is not None - assert found.document_id == OLDER_DOC - - -@pytest.mark.asyncio -async def test_find_ignores_other_user_other_namespace_archived_and_case( - db_session: Session, -) -> None: - adapter = _AsyncSessionAdapter(db_session) - other_user = await find_active_document_by_source_file_name( - adapter, # type: ignore[arg-type] - user_id="someone_else", - namespace=NAMESPACE, - source_file_name=FILE_NAME, - ) - other_ns = await find_active_document_by_source_file_name( - adapter, # type: ignore[arg-type] - user_id=USER_ID, - namespace="other", - source_file_name=FILE_NAME, - ) - different_case = await find_active_document_by_source_file_name( - adapter, # type: ignore[arg-type] - user_id=USER_ID, - namespace=NAMESPACE, - source_file_name="01-P1-70.PDF", - ) - db_session.query(Document).filter(Document.document_id == EXISTING_DOC).update( - {"status": "archived"} - ) - db_session.commit() - archived = await find_active_document_by_source_file_name( - adapter, # type: ignore[arg-type] - user_id=USER_ID, - namespace=NAMESPACE, - source_file_name=FILE_NAME, - ) - assert other_user is None - assert other_ns is None - assert different_case is None - assert archived is None - - -@pytest.mark.asyncio -async def test_reject_duplicate_file_name_without_document_id( - db_session: Session, -) -> None: - with pytest.raises(ConflictException) as caught: - await _reject(db_session, payload=_file_payload()) - error = caught.value - assert error.details == { - "reason": "ALREADY_EXISTS", - "resource": "Document", - "id": EXISTING_DOC, - } - assert error.user_message == ( - f"A document named {FILE_NAME!r} already exists. " - "To replace it, retry with document_id set to details.id." - ) - - -@pytest.mark.asyncio -async def test_reject_skips_when_document_id_is_provided( - db_session: Session, -) -> None: - await _reject(db_session, payload=_file_payload(document_id=EXISTING_DOC)) diff --git a/apps/worker/app/services/document_ingestion/source_content_admission.py b/apps/worker/app/services/document_ingestion/source_content_admission.py new file mode 100644 index 000000000..e943d1eae --- /dev/null +++ b/apps/worker/app/services/document_ingestion/source_content_admission.py @@ -0,0 +1,163 @@ +from __future__ import annotations + +import hashlib +import json +from dataclasses import dataclass +from pathlib import Path + +from sqlalchemy import CompoundSelect, literal, or_, select, text +from sqlalchemy.orm import Session + +from shared.core.database_sync import get_sync_db_context +from shared.core.exceptions.domain_exceptions import ( + ConflictException, + NotFoundException, +) +from shared.models.database.document import Document +from shared.models.database.job import Job +from shared.models.database.job_result import JobResult +from shared.models.schemas.retrieval_namespace import normalize_retrieval_namespace + +_HASH_FIELD: str = "source_content_sha256" +_SIZE_FIELD: str = "source_size_bytes" +_ACTIVE_JOB_STATUSES: tuple[str, ...] = ( + "waiting-file", + "pending", + "running", + "converting", +) + + +@dataclass(frozen=True) +class _SourceScope: + user_id: str + namespace: str + is_demo: bool + + +class SourceContentAdmission: + """Reject identical original bytes within a user's active document scope. + + Fingerprints live on the source Job, so joining the current Job Result + compares only active revisions. Sources without fingerprints are ignored. + A transaction lock serializes content claims before parsing starts; running + Jobs reserve their content until they publish or become terminal. Both reads + use one statement snapshot to cover the transition from running to published. + """ + + def admit_source(self, *, job_id: str, local_file_path: str) -> dict[str, object]: + with Path(local_file_path).open("rb") as source_file: + digest: str = hashlib.file_digest(source_file, "sha256").hexdigest() + size: int = Path(local_file_path).stat().st_size + fingerprint: dict[str, object] = {_HASH_FIELD: digest, _SIZE_FIELD: size} + scope: _SourceScope = self._load_scope(job_id) + with get_sync_db_context() as database: + if not scope.is_demo: + self._lock_content_claim(database, scope, digest) + self._reject_existing_content(database, scope, digest, job_id) + self._save_fingerprint(database, job_id, fingerprint) + return fingerprint + + def _load_scope(self, job_id: str) -> _SourceScope: + with get_sync_db_context() as database: + job: Job | None = database.get(Job, job_id) + if job is None: + raise NotFoundException(resource="Job", resource_id=job_id) + metadata: dict[str, object] = dict(job.job_metadata or {}) + namespace: object = metadata.get("namespace") + return _SourceScope( + user_id=job.user_id, + namespace=normalize_retrieval_namespace( + namespace if isinstance(namespace, str) else None, + ), + is_demo=metadata.get("corpus_target") == "DEMO", + ) + + @staticmethod + def _lock_content_claim( + database: Session, scope: _SourceScope, digest: str + ) -> None: + identity: str = json.dumps( + ["source-content", scope.user_id, scope.namespace, digest] + ) + lock_key: int = int.from_bytes( + hashlib.sha256(identity.encode("utf-8")).digest()[:8], + "big", + signed=True, + ) + database.execute( + text("SELECT pg_advisory_xact_lock(:lock_key)"), {"lock_key": lock_key} + ) + + def _reject_existing_content( + self, + database: Session, + scope: _SourceScope, + digest: str, + job_id: str, + ) -> None: + match: tuple[str, str] | None = ( + database.execute(self._build_duplicate_query(scope, digest, job_id)) + .tuples() + .first() + ) + if match is None: + return + resource, resource_id = match + user_message: str = ( + "A document with identical file content already exists in this namespace." + if resource == "Document" + else "A job with identical file content is already being processed in this namespace." + ) + raise ConflictException( + user_message=user_message, + resource=resource, + resource_id=resource_id, + internal_message=f"Duplicate source content: resource={resource} id={resource_id}", + ) + + @staticmethod + def _build_duplicate_query( + scope: _SourceScope, + digest: str, + job_id: str, + ) -> CompoundSelect[tuple[str, str]]: + return ( + select(literal("Document"), Document.document_id) + .select_from(Job) + .join(JobResult, JobResult.job_id == Job.job_id) + .join(Document, Document.current_job_result_id == JobResult.id) + .where(Job.user_id == scope.user_id) + .where(Job.job_metadata[_HASH_FIELD].as_string() == digest) + .where(Document.user_id == scope.user_id) + .where(Document.namespace == scope.namespace) + .where(Document.status == "active") + .union_all( + select(literal("Job"), Job.job_id) + .where(Job.user_id == scope.user_id) + .where(Job.job_id != job_id) + .where(Job.status.in_(_ACTIVE_JOB_STATUSES)) + .where(Job.job_metadata["namespace"].as_string() == scope.namespace) + .where(Job.job_metadata[_HASH_FIELD].as_string() == digest) + .where( + or_( + Job.job_metadata["corpus_target"].as_string().is_(None), + Job.job_metadata["corpus_target"].as_string() != "DEMO", + ) + ) + ) + .limit(1) + ) + + @staticmethod + def _save_fingerprint( + database: Session, + job_id: str, + fingerprint: dict[str, object], + ) -> None: + job: Job | None = database.execute( + select(Job).where(Job.job_id == job_id).with_for_update() + ).scalar_one_or_none() + if job is None: + raise NotFoundException(resource="Job", resource_id=job_id) + job.job_metadata = {**dict(job.job_metadata or {}), **fingerprint} diff --git a/apps/worker/app/services/document_ingestion/source_preparation.py b/apps/worker/app/services/document_ingestion/source_preparation.py index ca27344c8..b14b3cd40 100644 --- a/apps/worker/app/services/document_ingestion/source_preparation.py +++ b/apps/worker/app/services/document_ingestion/source_preparation.py @@ -13,6 +13,7 @@ ParseJobContext, persist_job_metadata_updates, ) +from app.services.document_ingestion.source_content_admission import SourceContentAdmission from app.services.document_parser.support.internal_parse_name import ( prepare_internal_parse_input, ) @@ -56,8 +57,12 @@ def prepare_source_file( suffix=file_extension, temp_dir=input_dir, ) - if job_context.job_metadata.get("corpus_target") == "DEMO": - persist_job_metadata_updates(job_id=job_id, job_context=job_context, metadata_updates={"source_size_bytes": os.path.getsize(local_file_path)}) + fingerprint: dict[str, object] = SourceContentAdmission().admit_source( + job_id=job_id, + local_file_path=local_file_path, + ) + job_context.metadata_service.update_metadata(job_id, fingerprint) + job_context.job_metadata.update(fingerprint) logger.info(f"File downloaded: job_id={job_id}, local_path={local_file_path}") prepared_parse_input = prepare_internal_parse_input( diff --git a/apps/worker/tests/contract/test_parse_task_contract.py b/apps/worker/tests/contract/test_parse_task_contract.py index da630eaac..0fdf813b8 100644 --- a/apps/worker/tests/contract/test_parse_task_contract.py +++ b/apps/worker/tests/contract/test_parse_task_contract.py @@ -467,14 +467,14 @@ def test_parse_task_should_charge_user_when_billing_is_enabled( ] -def test_parse_task_should_export_full_result_when_same_content_was_already_published( +def test_parse_task_should_reject_identical_source_before_billing_or_parsing( worker_contract_environment: None, monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: contract = WorkerParseContract.create() contract.use_workspace_root(monkeypatch, tmp_path) - contract.use_billing(monkeypatch, is_enabled=False) + contract.use_billing(monkeypatch, is_enabled=True) user_id = f"worker-contract-user-{uuid4().hex[:12]}" first_job = contract.create_file_job( @@ -487,6 +487,29 @@ def test_parse_task_should_export_full_result_when_same_content_was_already_publ source_file_name="contract-duplicate.xlsx", job_id_prefix="job_parse_duplicate", ) + published_events: list[str] = [] + + class CapturedWebhookPublisher: + def publish_event(self, event_id: str) -> str: + published_events.append(event_id) + return "contract-webhook-message" + + from sqlalchemy import text + import shared.services.webhook.qstash_publisher as publisher_module + + monkeypatch.setattr( + publisher_module, + "get_qstash_webhook_publisher", + lambda: CapturedWebhookPublisher(), + ) + with contract.engine.begin() as connection: + connection.execute( + text( + "UPDATE jobs SET webhook_enabled = true, webhook_url = :url " + "WHERE job_id = :job_id" + ), + {"job_id": second_job["job_id"], "url": "https://example.com/duplicate-hook"}, + ) for job in [first_job, second_job]: contract.upload_source_file( local_file_path=_SAMPLE_XLSX_PATH, @@ -497,28 +520,80 @@ def test_parse_task_should_export_full_result_when_same_content_was_already_publ job_id=first_job["job_id"], user_id=user_id, ) + assert first_result.successful() + billing_before_duplicate = contract.observe_user_billing(user_id) second_result = contract.enqueue_parse_task( job_id=second_job["job_id"], user_id=user_id, ) - assert first_result.successful() - assert second_result.successful() - - observed = contract.observe_successful_job(second_job["job_id"]) - result_row = observed["result"] - job_chunks = observed["job_chunks"] - document_chunks = observed["document_chunks"] - result_zip = contract.read_result_zip( - result_s3_key=result_row["result_s3_key"], - tmp_path=tmp_path, - ) - - assert len(job_chunks) > 0 - assert len(document_chunks) == len(job_chunks) - assert len(result_zip["chunks"]["chunks"]) == len(job_chunks) - assert any(member.startswith("tables/") for member in result_zip["members"]) - assert "chunk_overlap" not in dict(result_row["document_metadata"] or {}) + assert second_result.failed() + observed = contract.observe_job_status(second_job["job_id"]) + assert observed["status"] == "failed" + assert observed["error_code"] == "ALREADY_EXISTS" + assert observed["credits_charged"] == 0 + assert observed["page_count"] is None + assert contract.observe_user_billing(user_id) == billing_before_duplicate + with contract.engine.connect() as connection: + assert connection.execute( + text("SELECT count(*) FROM job_results WHERE job_id = :job_id"), + {"job_id": second_job["job_id"]}, + ).scalar_one() == 0 + webhook = connection.execute( + text("SELECT id, payload FROM webhook_events WHERE job_id = :job_id"), + {"job_id": second_job["job_id"]}, + ).mappings().one() + assert webhook["payload"]["event"] == "job.failed" + assert webhook["payload"]["error"]["code"] == "ALREADY_EXISTS" + assert webhook["payload"]["error"]["details"]["resource"] == "Document" + assert published_events == [webhook["id"]] + assert contract.find_task_workspaces(tmp_path, second_job["job_id"]) == [] + + +def test_parse_task_should_export_full_results_for_distinct_bytes_with_same_content( + worker_contract_environment: None, + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + contract: WorkerParseContract = WorkerParseContract.create() + contract.use_workspace_root(monkeypatch, tmp_path) + contract.use_billing(monkeypatch, is_enabled=False) + user_id: str = f"worker-contract-user-{uuid4().hex[:12]}" + jobs: list[dict[str, object]] = [ + contract.create_file_job( + user_id=user_id, + source_file_name="same-name.xlsx", + ) + for _ in range(2) + ] + chunk_identifier_groups: list[list[str]] = [] + for index, job in enumerate(jobs): + source_path: Path = tmp_path / f"distinct-bytes-{index}.xlsx" + source_path.write_bytes(_SAMPLE_XLSX_PATH.read_bytes()) + with zipfile.ZipFile(source_path, mode="a") as archive: + archive.comment = f"distinct-bytes-{index}".encode("utf-8") + contract.upload_source_file( + local_file_path=source_path, + s3_key=str(job["s3_key"]), + ) + result = contract.enqueue_parse_task( + job_id=str(job["job_id"]), + user_id=user_id, + ) + assert result.successful() + observed = contract.observe_successful_job(str(job["job_id"])) + chunk_identifier_groups.append( + [str(chunk["chunk_id"]) for chunk in observed["job_chunks"]] + ) + result_zip = contract.read_result_zip( + result_s3_key=observed["result"]["result_s3_key"], + tmp_path=tmp_path, + ) + assert len(observed["job_chunks"]) > 0 + assert len(observed["document_chunks"]) == len(observed["job_chunks"]) + assert len(result_zip["chunks"]["chunks"]) == len(observed["job_chunks"]) + assert any(member.startswith("tables/") for member in result_zip["members"]) + assert chunk_identifier_groups[0] == chunk_identifier_groups[1] def test_parse_task_should_initialize_billing_once_for_concurrent_parse_tasks( @@ -539,9 +614,13 @@ def test_parse_task_should_initialize_billing_once_for_concurrent_parse_tasks( ) for index in range(2) ] - for job in jobs: + for index, job in enumerate(jobs): + source_path: Path = tmp_path / f"concurrent-source-{index}.xlsx" + source_path.write_bytes(_SAMPLE_XLSX_PATH.read_bytes()) + with zipfile.ZipFile(source_path, mode="a") as archive: + archive.comment = f"concurrent-source-{index}".encode("utf-8") contract.upload_source_file( - local_file_path=_SAMPLE_XLSX_PATH, + local_file_path=source_path, s3_key=job["s3_key"], ) diff --git a/apps/worker/tests/contract/test_source_content_admission_contract.py b/apps/worker/tests/contract/test_source_content_admission_contract.py new file mode 100644 index 000000000..fcbfb6867 --- /dev/null +++ b/apps/worker/tests/contract/test_source_content_admission_contract.py @@ -0,0 +1,420 @@ +from __future__ import annotations + +from concurrent.futures import Future, ThreadPoolExecutor +from dataclasses import dataclass +from pathlib import Path +from threading import Event +from time import monotonic, sleep +from uuid import uuid4 + +import pytest +from sqlalchemy import event, select, text +from sqlalchemy.engine import Connection +from support.worker_parse_contract import WorkerParseContract + + +@dataclass(frozen=True) +class SourceJob: + job_id: str + user_id: str + document_id: str + local_path: Path + storage_key: str + + +def create_source_job( + contract: WorkerParseContract, + directory: Path, + *, + content: bytes, + user_id: str | None = None, + namespace: str = "worker-contract", + file_name: str = "original.md", + document_id: str | None = None, + source_type: str = "file", +) -> SourceJob: + from shared.core.database_sync import get_sync_db_context + from shared.models.database.job import Job + + record: dict[str, object] = contract.create_file_job( + source_file_name=file_name, + user_id=user_id, + ) + job_id: str = str(record["job_id"]) + effective_document_id: str = document_id or f"doc_{uuid4().hex[:12]}" + path: Path = directory / job_id + path.write_bytes(content) + storage_key: str = str(record["s3_key"]) + contract.upload_source_file(local_file_path=path, s3_key=storage_key) + with get_sync_db_context() as database: + job: Job | None = database.get(Job, job_id) + assert job is not None + job.source_type = source_type + job.job_metadata = { + **dict(job.job_metadata or {}), + "namespace": namespace, + "document_id": effective_document_id, + "original_request": {"document_id": document_id}, + } + return SourceJob( + job_id=job_id, + user_id=str(record["user_id"]), + document_id=effective_document_id, + local_path=path, + storage_key=storage_key, + ) + + +def admit_source(source: SourceJob) -> dict[str, object]: + from app.services.document_ingestion.source_content_admission import ( + SourceContentAdmission, + ) + + return SourceContentAdmission().admit_source( + job_id=source.job_id, + local_file_path=str(source.local_path), + ) + + +def publish_source(source: SourceJob, *, namespace: str = "worker-contract") -> None: + from shared.core.database_sync import get_sync_db_context + from shared.models.database.document import Document + from shared.models.database.job import Job + from shared.models.database.job_result import JobResult + + with get_sync_db_context() as database: + job: Job | None = database.get(Job, source.job_id) + assert job is not None + job.status = "done" + document: Document | None = database.get(Document, source.document_id) + if document is None: + document = Document( + document_id=source.document_id, + user_id=source.user_id, + namespace=namespace, + source_file_name="original.md", + ) + database.add(document) + database.flush() + result: JobResult = JobResult( + job_id=source.job_id, + document_id=source.document_id, + delivery_mode="inline", + ) + database.add(result) + database.flush() + document.current_job_result_id = result.id + + +@pytest.mark.parametrize("file_name", ["original.md", "renamed.md"]) +@pytest.mark.parametrize("source_type", ["file", "url"]) +def test_rejects_identical_content_independently_of_filename_and_source_type( + worker_contract_environment: None, + tmp_path: Path, + file_name: str, + source_type: str, +) -> None: + from shared.core.exceptions.domain_exceptions import ConflictException + + contract: WorkerParseContract = WorkerParseContract.create() + existing: SourceJob = create_source_job( + contract, tmp_path, content=b"original content" + ) + admit_source(existing) + publish_source(existing) + incoming: SourceJob = create_source_job( + contract, + tmp_path, + content=b"original content", + user_id=existing.user_id, + file_name=file_name, + source_type=source_type, + ) + + with pytest.raises(ConflictException) as failure: + admit_source(incoming) + assert failure.value.to_client(incoming.job_id)["error"]["details"] == { + "reason": "ALREADY_EXISTS", + "resource": "Document", + "id": existing.document_id, + } + + +def test_accepts_same_filename_and_same_size_when_bytes_differ( + worker_contract_environment: None, + tmp_path: Path, +) -> None: + contract: WorkerParseContract = WorkerParseContract.create() + existing: SourceJob = create_source_job( + contract, tmp_path, content=b"first content" + ) + admit_source(existing) + publish_source(existing) + incoming: SourceJob = create_source_job( + contract, + tmp_path, + content=b"other content", + user_id=existing.user_id, + ) + fingerprint: dict[str, object] = admit_source(incoming) + assert fingerprint["source_size_bytes"] == len(b"other content") + assert ( + fingerprint["source_content_sha256"] + != contract.get_job_metadata(existing.job_id)["source_content_sha256"] + ) + + +def test_rejects_unchanged_explicit_document_update( + worker_contract_environment: None, + tmp_path: Path, +) -> None: + from shared.core.exceptions.domain_exceptions import ConflictException + + contract: WorkerParseContract = WorkerParseContract.create() + existing: SourceJob = create_source_job(contract, tmp_path, content=b"unchanged") + admit_source(existing) + publish_source(existing) + incoming: SourceJob = create_source_job( + contract, + tmp_path, + content=b"unchanged", + user_id=existing.user_id, + document_id=existing.document_id, + ) + with pytest.raises(ConflictException) as failure: + admit_source(incoming) + assert failure.value.to_client(incoming.job_id)["error"]["details"]["id"] == existing.document_id + + +@pytest.mark.parametrize("scope_change", ["user", "namespace", "archived"]) +def test_scopes_duplicate_content_to_current_active_user_documents( + worker_contract_environment: None, + tmp_path: Path, + scope_change: str, +) -> None: + from shared.core.database_sync import get_sync_db_context + from shared.models.database.document import Document + + contract: WorkerParseContract = WorkerParseContract.create() + existing: SourceJob = create_source_job(contract, tmp_path, content=b"shared bytes") + admit_source(existing) + publish_source(existing) + if scope_change == "archived": + with get_sync_db_context() as database: + document: Document | None = database.get(Document, existing.document_id) + assert document is not None + document.status = "archived" + incoming: SourceJob = create_source_job( + contract, + tmp_path, + content=b"shared bytes", + user_id=None if scope_change == "user" else existing.user_id, + namespace="another-namespace" + if scope_change == "namespace" + else "worker-contract", + ) + assert admit_source(incoming)["source_content_sha256"] + + +def test_compares_only_the_current_revision_and_allows_changed_updates( + worker_contract_environment: None, + tmp_path: Path, +) -> None: + contract: WorkerParseContract = WorkerParseContract.create() + existing: SourceJob = create_source_job(contract, tmp_path, content=b"old revision") + admit_source(existing) + publish_source(existing) + revision: SourceJob = create_source_job( + contract, + tmp_path, + content=b"new revision", + user_id=existing.user_id, + document_id=existing.document_id, + ) + admit_source(revision) + publish_source(revision) + incoming: SourceJob = create_source_job( + contract, + tmp_path, + content=b"old revision", + user_id=existing.user_id, + ) + assert admit_source(incoming)["source_content_sha256"] + + +def test_ignores_existing_sources_without_hashes_and_records_future_uploads( + worker_contract_environment: None, + tmp_path: Path, +) -> None: + from shared.core.exceptions.domain_exceptions import ConflictException + + contract: WorkerParseContract = WorkerParseContract.create() + existing: SourceJob = create_source_job(contract, tmp_path, content=b"legacy bytes") + publish_source(existing) + incoming: SourceJob = create_source_job( + contract, + tmp_path, + content=b"legacy bytes", + user_id=existing.user_id, + ) + # An unavailable historical original does not participate in admission. + contract.storage.delete_upload_file(existing.storage_key) + admit_source(incoming) + publish_source(incoming) + assert "source_content_sha256" not in contract.get_job_metadata(existing.job_id) + repeat: SourceJob = create_source_job( + contract, + tmp_path, + content=b"legacy bytes", + user_id=existing.user_id, + ) + with pytest.raises(ConflictException): + admit_source(repeat) + + +def test_keeps_rejecting_a_duplicate_when_publication_commits_during_the_lookup( + worker_contract_environment: None, + tmp_path: Path, +) -> None: + from shared.core.exceptions.domain_exceptions import ConflictException + + contract: WorkerParseContract = WorkerParseContract.create() + existing: SourceJob = create_source_job( + contract, tmp_path, content=b"publishing bytes" + ) + admit_source(existing) + incoming: SourceJob = create_source_job( + contract, + tmp_path, + content=b"publishing bytes", + user_id=existing.user_id, + ) + has_published: bool = False + + def publish_after_document_lookup( + connection: Connection, + cursor: object, + statement: str, + parameters: object, + context: object, + is_many: bool, + ) -> None: + nonlocal has_published + if has_published or "JOIN job_results" not in statement: + return + has_published = True + publish_source(existing) + + event.listen(contract.engine, "after_cursor_execute", publish_after_document_lookup) + try: + with pytest.raises(ConflictException): + admit_source(incoming) + finally: + event.remove( + contract.engine, "after_cursor_execute", publish_after_document_lookup + ) + assert has_published + + +def test_preserves_demo_corpus_admission_when_source_bytes_match_a_personal_document( + worker_contract_environment: None, + tmp_path: Path, +) -> None: + from shared.core.database_sync import get_sync_db_context + from shared.models.database.job import Job + + contract: WorkerParseContract = WorkerParseContract.create() + existing: SourceJob = create_source_job(contract, tmp_path, content=b"demo bytes") + admit_source(existing) + publish_source(existing) + incoming: SourceJob = create_source_job( + contract, + tmp_path, + content=b"demo bytes", + user_id=existing.user_id, + ) + with get_sync_db_context() as database: + job: Job | None = database.get(Job, incoming.job_id) + assert job is not None + job.job_metadata = {**dict(job.job_metadata or {}), "corpus_target": "DEMO"} + assert admit_source(incoming)["source_content_sha256"] + + +def test_serializes_concurrent_content_claims_and_allows_retries_after_failure( + worker_contract_environment: None, + tmp_path: Path, +) -> None: + from shared.core.database_sync import get_sync_db_context + from shared.core.exceptions.domain_exceptions import ConflictException + from shared.models.database.job import Job + + contract: WorkerParseContract = WorkerParseContract.create() + first: SourceJob = create_source_job( + contract, tmp_path, content=b"concurrent bytes" + ) + second: SourceJob = create_source_job( + contract, tmp_path, content=b"concurrent bytes", user_id=first.user_id + ) + has_read_candidates: Event = Event() + can_save_claim: Event = Event() + + def pause_first_claim_after_lookup( + connection: Connection, + cursor: object, + statement: str, + parameters: object, + context: object, + is_many: bool, + ) -> None: + if "JOIN job_results" not in statement or has_read_candidates.is_set(): + return + has_read_candidates.set() + assert can_save_claim.wait(timeout=10) + + def claim_content(source: SourceJob) -> bool: + try: + admit_source(source) + return True + except ConflictException: + return False + + # Hold the first transaction after its empty snapshot. The second must wait + # for its advisory lock, rather than also admitting from an empty snapshot. + event.listen(contract.engine, "after_cursor_execute", pause_first_claim_after_lookup) + try: + with ThreadPoolExecutor(max_workers=2) as executor: + first_claim: Future[bool] = executor.submit(claim_content, first) + try: + assert has_read_candidates.wait(timeout=10) + second_claim: Future[bool] = executor.submit(claim_content, second) + deadline: float = monotonic() + 10 + with contract.engine.connect() as connection: + while not second_claim.done(): + is_waiting_for_claim: bool = connection.execute( + text( + "SELECT EXISTS (SELECT 1 FROM pg_locks " + "WHERE locktype = 'advisory' AND NOT granted)" + ) + ).scalar_one() + if is_waiting_for_claim: + break + assert monotonic() < deadline + sleep(0.01) + finally: + can_save_claim.set() + results: list[bool] = [first_claim.result(), second_claim.result()] + finally: + event.remove( + contract.engine, "after_cursor_execute", pause_first_claim_after_lookup + ) + assert sorted(results) == [False, True] + winner: SourceJob = first if results[0] else second + loser: SourceJob = second if results[0] else first + # A redelivered task may claim its own content again. + assert admit_source(winner)["source_content_sha256"] + with get_sync_db_context() as database: + job: Job = database.execute( + select(Job).where(Job.job_id == winner.job_id) + ).scalar_one() + job.status = "failed" + assert admit_source(loser)["source_content_sha256"] diff --git a/docs/source-content-deduplication-review.md b/docs/source-content-deduplication-review.md new file mode 100644 index 000000000..9aff23b4b --- /dev/null +++ b/docs/source-content-deduplication-review.md @@ -0,0 +1,58 @@ +# Source content deduplication review and ablations + +Reviewed on 2026-10-10 against `5264697481567b76abf8abcf463573018847701c`, including the uncommitted implementation and new files. The specification is the agreed conversation: compare original bytes within one user and namespace, reject unchanged updates, use current active revisions and in-progress reservations, and perform no historical backfill. + +## Standards + +The review found one concrete migration defect: a valid, ready index with the expected name but incorrect columns or predicate was retained. Two real PostgreSQL contracts reproduced the defect under ordinary and caller-owned transactions. The migration now validates the table, nonunique B-tree definition, column order and expression, predicate, validity, and readiness before retaining an index. Twelve cases pass, including retention of the correct index without rebuilding it and replacement of indexes whose JSON keys differ in case or contain a cast-like suffix. Definition comparison preserves JSON string literals rather than normalizing their case or stripping cast-like text from them. + +No remaining blocker was found. Optional improvements include a direct interrupted-index recovery contract, concurrent index removal during ordinary downgrade, and a precise fingerprint return type. Canonical index-definition formatting is verified on the local PostgreSQL version; a future formatting change could cause an unnecessary rebuild. + +## Spec + +No blocking behavior mismatch was found. Publication and the terminal job transition commit together, accepted fingerprints survive ordinary metadata updates, and completed task redelivery exits before admission. + +The review identified a lost export contract. It is restored for two same-named XLSX files differing only in ZIP comments: both succeed, retain full document and ZIP chunks, and produce identical parsed chunk identifiers. The duplicate worker contract now enables billing and verifies that the duplicate leaves account balance unchanged, has no result, and creates an `ALREADY_EXISTS` webhook outbox event and dispatch request. The dispatch is captured locally; external webhook delivery is not exercised. + +Exact byte equality remains the agreed boundary. Re-exported files with changed container metadata can pass even when their visible content is identical. An unchanged source is also rejected when only parsing options change. Existing unhashed sources remain outside admission. + +## Component ablations + +Every variant ran the same 14 admission contracts against an isolated real PostgreSQL fixture and local source files. Mutations were test-scoped patches in separate processes; production code was not replaced with ablated implementations. All runs below had zero setup errors. Failures in the altered variants are intentional evidence that the contracts detect the removed behavior. + +| Variant | Passed | Failed | Observed failure | +| --- | ---: | ---: | --- | +| Unchanged implementation | 14 | 0 | Agreed behavior holds | +| Use file size instead of hashing bytes | 12 | 2 | Different bytes of equal length are falsely rejected | +| Omit fingerprint persistence | 5 | 9 | Later duplicates are accepted and durable fingerprints are unavailable | +| Remove advisory transaction lock | 13 | 1 | Both concurrent claims are admitted | +| Check published documents only | 12 | 2 | Unpublished reservations and the publication transition are missed | +| Split the union into two separate reads | 13 | 1 | Publication between the snapshots allows a duplicate | + +The concurrency contract pauses the first transaction after its empty lookup and starts the second claim before releasing it. With the lock present, PostgreSQL reports the second transaction waiting for its advisory lock. Without the lock, the second claim can finish against the empty state. The publication contract commits publication after the first lookup executes, exposing the gap in separate statement snapshots. + +## Index ablation and skew experiment + +The experiment used the actual ORM union query and the ordinary runtime database role, with planner defaults enabled. The first dataset contained 5,000 active current document revisions with distinct hashes and one incoming job. The last dataset additionally contained 2,000 running jobs with the requested hash in other namespaces. Every query correctly returned no duplicate in the incoming namespace. + +| Local workload | SQL execution time | Shared buffer hits | Query-plan evidence | +| --- | ---: | ---: | --- | +| Hash index, absent hash | 0.093 ms | 6 | Published branch uses user/hash index; no revision joins execute | +| Same dataset, hash index removed | 37.883 ms | 30,210 | Visits 5,000 documents and performs 5,000 result/job lookups | +| Hash index, same hash in 2,000 other namespaces | 6.948 ms | 4,177 | Published branch probes 2,000 candidate results; processing branch filters 2,001 jobs | + +These are single local `EXPLAIN (ANALYZE, BUFFERS)` probes with zero shared disk reads, not production latency measurements or a five-million-document benchmark. SQL execution time excludes download, hashing, pool acquisition, network round trips, and lock waits. Buffer counts and visited rows provide the stronger evidence. + +The index is necessary, but matching hash multiplicity still matters because namespace and lifecycle state are later filters. The earlier 5–20 ms planning estimate assumes few matching user/hash rows and is not a worst-case guarantee. We keep the current design for this change; a workload dominated by repeated content across namespaces would justify revisiting index scope and the published-candidate query. Terminal and archived history were not measured by this skew experiment. + +## Reproduction and validation + +Local experiment artifacts are in `/tmp/knowhere-source-content-ablation/`: `source_content_ablation.py`, `run_matrix.py`, per-variant logs and JUnit reports, `results.json`, `test_query_plan_experiment.py`, and the complete `query-plans.json`. + +The matrix can be repeated with `python3 /tmp/knowhere-source-content-ablation/run_matrix.py`; it uses the existing uv environment, applies one mutation per process, and runs the admission contract file. Local PostgreSQL sockets must be permitted. The index experiment loads `contract.conftest` as a pytest plugin, executes its temporary experiment file, and drops/recreates the hash index only in the isolated fixture database. + +The strengthened worker suite passes all 33 contracts, and the API and migration suite passes all 61 contracts. The final expanded index migration check passes all 12 cases; this check overlaps the broader migration suite and is not an additional 12 distinct contracts. Ruff, Pyright (production paths, changed API contracts, and the new migration), and `git diff --check` pass. URL transport and real external webhook delivery remain outside these local contracts. + +Before PR publication, the change was rebased onto `main` at `65c0445a` and revalidated: all 33 worker contracts and all 65 API/migration contracts pass, including the expanded migration cases. Ruff, Pyright, and the diff whitespace check also pass on that base. + +Standards: one confirmed defect, fixed; optional operational and typing suggestions remain. Spec: no blocking mismatch; export, billing, webhook, and concurrency coverage strengthened. diff --git a/packages/shared-python/shared/models/database/job.py b/packages/shared-python/shared/models/database/job.py index 8a4fb6bf3..d4d4831bc 100644 --- a/packages/shared-python/shared/models/database/job.py +++ b/packages/shared-python/shared/models/database/job.py @@ -126,6 +126,12 @@ class Job(Base): # Indexes __table_args__ = ( + Index( + "idx_jobs_user_source_content_hash", + "user_id", + text("(job_metadata ->> 'source_content_sha256')"), + postgresql_where=text("job_metadata ->> 'source_content_sha256' IS NOT NULL"), + ), Index("idx_job_status", "status"), Index("idx_job_type", "job_type"), Index("idx_job_created_at", "created_at"), diff --git a/packages/shared-python/shared/models/schemas/job_metadata.py b/packages/shared-python/shared/models/schemas/job_metadata.py index ec7d23454..9b5c7d930 100644 --- a/packages/shared-python/shared/models/schemas/job_metadata.py +++ b/packages/shared-python/shared/models/schemas/job_metadata.py @@ -49,6 +49,12 @@ class JobMetadataBase(BaseModel): # Source-file fields. source_type: Optional[str] = Field(None, description="Source type") source_file_name: Optional[str] = Field(None, description="Source file name") + source_content_sha256: Optional[str] = Field( + None, pattern=r"^[0-9a-f]{64}$", description="SHA-256 of the original source bytes" + ) + source_size_bytes: Optional[int] = Field( + None, ge=0, description="Original source file size in bytes" + ) source_url: Optional[str] = Field(None, description="Source URL") file_url: Optional[str] = Field(None, description="File URL")