From f7d1a3d46c25aa3bb7e3da0be24471af0ec9c298 Mon Sep 17 00:00:00 2001 From: suguanYang Date: Wed, 19 Aug 2026 04:12:23 +0800 Subject: [PATCH 1/5] fix: sanitize NUL characters before persistence --- .../jobs/lifecycle/success_finalizer.py | 22 ++++++++++++++---- .../shared-python/shared/utils/json_utils.py | 23 +++++++++++++++++++ 2 files changed, 40 insertions(+), 5 deletions(-) diff --git a/packages/shared-python/shared/services/jobs/lifecycle/success_finalizer.py b/packages/shared-python/shared/services/jobs/lifecycle/success_finalizer.py index 2edfc2b66..3706ed998 100644 --- a/packages/shared-python/shared/services/jobs/lifecycle/success_finalizer.py +++ b/packages/shared-python/shared/services/jobs/lifecycle/success_finalizer.py @@ -1,7 +1,7 @@ from __future__ import annotations from dataclasses import dataclass -from typing import Any, Literal +from typing import Any, Literal, cast from loguru import logger from sqlalchemy.orm import Session @@ -11,6 +11,7 @@ from shared.services.jobs.lifecycle.publication import SyncJobPublicationFinalizer from shared.services.jobs.lifecycle.result_writer import SyncJobResultWriter from shared.services.jobs.lifecycle.webhook_outbox import SyncJobWebhookOutbox +from shared.utils.json_utils import remove_nul_characters @dataclass(frozen=True) @@ -81,6 +82,17 @@ def finalize( section_summaries: dict[str, str] | None, document_top_summary: str | None = None, ) -> JobSuccessFinalization: + safe_chunks = cast( + list[dict[str, Any]], remove_nul_characters(chunks) + ) + safe_section_summaries = cast( + dict[str, str] | None, + remove_nul_characters(section_summaries), + ) + safe_document_top_summary = cast( + str | None, + remove_nul_characters(document_top_summary), + ) job_result = self._result_writer.upsert_job_result( db, job_id, @@ -89,14 +101,14 @@ def finalize( result_s3_key=result_s3_key, result_size=zip_size, ) - self._result_writer.replace_chunks(db, job_result.id, chunks) + self._result_writer.replace_chunks(db, job_result.id, safe_chunks) publication_outcome = self._publication_finalizer.publish_result( db, job_id=job_id, job_result_id=job_result.id, - chunks=chunks, - section_summaries=section_summaries, - document_top_summary=document_top_summary, + chunks=safe_chunks, + section_summaries=safe_section_summaries, + document_top_summary=safe_document_top_summary, ) transition_outcome = self._state_machine.mark_completed_outcome( diff --git a/packages/shared-python/shared/utils/json_utils.py b/packages/shared-python/shared/utils/json_utils.py index fa2de8c73..275c6be32 100644 --- a/packages/shared-python/shared/utils/json_utils.py +++ b/packages/shared-python/shared/utils/json_utils.py @@ -7,6 +7,29 @@ from typing import Any, Mapping, MutableSet +def remove_nul_characters(value: object) -> object: + """Remove PostgreSQL-incompatible NUL characters from JSON-like values. + + Parser output is persisted both as JSON metadata and as text columns. The + PostgreSQL text encoders reject U+0000, so clean it at the persistence + boundary while preserving the shape and non-string scalar values of the + payload. + """ + if isinstance(value, str): + return value.replace("\x00", "") + if isinstance(value, Mapping): + return { + remove_nul_characters(key) if isinstance(key, str) else key: + remove_nul_characters(item) + for key, item in value.items() + } + if isinstance(value, list): + return [remove_nul_characters(item) for item in value] + if isinstance(value, tuple): + return tuple(remove_nul_characters(item) for item in value) + return value + + def make_json_safe( value: Any, *, max_preview_rows: int = 5, _visited: MutableSet[int] | None = None ) -> Any: From ea57524d1fe363d942f9427e41a4d80fb3b835d2 Mon Sep 17 00:00:00 2001 From: suguanYang Date: Wed, 19 Aug 2026 04:15:57 +0800 Subject: [PATCH 2/5] fix: make document ingestion publication idempotent --- .../document_ingestion/handoff_service.py | 24 +++ .../tests/contract/test_s3_event_contract.py | 87 +++++++++ .../document_ingestion/processing_run.py | 45 +++-- .../contract/test_parse_task_contract.py | 173 ++++++++++++++++++ .../contract/test_processing_run_contract.py | 65 +++++++ .../services/jobs/lifecycle/result_writer.py | 18 +- .../services/retrieval/publication_content.py | 51 +++++- .../services/retrieval/publication_service.py | 3 +- 8 files changed, 439 insertions(+), 27 deletions(-) create mode 100644 apps/worker/tests/contract/test_processing_run_contract.py diff --git a/apps/api/app/services/document_ingestion/handoff_service.py b/apps/api/app/services/document_ingestion/handoff_service.py index c8de180bd..fe6af5029 100644 --- a/apps/api/app/services/document_ingestion/handoff_service.py +++ b/apps/api/app/services/document_ingestion/handoff_service.py @@ -29,6 +29,7 @@ class _UploadedFileJob(Protocol): job_id: str job_type: str + status: str class DocumentIngestionHandoffService: @@ -66,6 +67,17 @@ async def start_uploaded_file_workflow( ], ) + # Upload completion can arrive through both the S3 notification and the + # confirm-upload endpoint. Once either path has moved the job out of + # waiting-file, the other path must be a no-op instead of dispatching a + # second worker task for the same logical job. + if job.status != JobStatus.WAITING_FILE.value: + logger.info( + "Upload handoff already completed: " + f"job_id={job.job_id}, status={job.status}" + ) + return + outcome = await self._state_machine.transition_outcome( db, job.job_id, @@ -75,6 +87,18 @@ async def start_uploaded_file_workflow( "system", ) if not outcome.succeeded: + # A concurrent handoff may have won the CAS transition after this + # caller loaded the waiting-file snapshot. The state machine + # reports the winner's state as ``from_state``; treat that result + # as an idempotent no-op and do not enqueue another task. + if outcome.reason == "invalid_transition" and outcome.from_state != ( + JobStatus.WAITING_FILE.value + ): + logger.info( + "Upload handoff won by another trigger: " + f"job_id={job.job_id}, status={outcome.from_state}" + ) + return logger.warning( "Upload handoff transition rejected: " f"job_id={job.job_id}, reason={outcome.reason}" diff --git a/apps/api/tests/contract/test_s3_event_contract.py b/apps/api/tests/contract/test_s3_event_contract.py index d91c37951..6507016b1 100644 --- a/apps/api/tests/contract/test_s3_event_contract.py +++ b/apps/api/tests/contract/test_s3_event_contract.py @@ -119,6 +119,93 @@ async def start_uploaded_file_parse( ] +@pytest.mark.asyncio +async def test_should_not_dispatch_a_second_task_for_a_replayed_upload_event( + api_client_factory: Callable[[], AbstractAsyncContextManager[AsyncClient]], + monkeypatch: MonkeyPatch, +) -> None: + workflow_calls: list[dict[str, str]] = [] + + class FakeDocumentIngestionWorkerDispatcher: + async def start_uploaded_file_parse( + self, + *, + job_id: str, + user_id: str, + ) -> str: + workflow_calls.append({"job_id": job_id, "user_id": user_id}) + return "contract-task-id" + + async with api_client_factory() as api_client: + user_id, job_id = await _insert_waiting_file_job() + handoff_service = importlib.import_module( + "app.services.document_ingestion.handoff_service" + ) + monkeypatch.setattr( + handoff_service, + "DocumentIngestionWorkerDispatcher", + FakeDocumentIngestionWorkerDispatcher, + ) + + first_response = await api_client.post( + "/api/v1/internal/s3-events", + json=_build_s3_event_payload(job_id), + ) + replay_response = await api_client.post( + "/api/v1/internal/s3-events", + json=_build_s3_event_payload(job_id), + ) + + assert first_response.status_code == 200 + assert replay_response.status_code == 200 + assert workflow_calls == [{"job_id": job_id, "user_id": user_id}] + + +@pytest.mark.asyncio +async def test_should_treat_a_concurrent_upload_handoff_cas_winner_as_a_no_op() -> None: + from app.services.document_ingestion.handoff_service import ( + DocumentIngestionHandoffService, + ) + from shared.core.state_machine.transition_outcome import JobTransitionOutcome + + class FakeStateMachine: + async def transition_outcome(self, *args: object, **kwargs: object) -> object: + del args, kwargs + return JobTransitionOutcome.rejected( + job_id="job-race", + to_state="pending", + reason="invalid_transition", + attempts=1, + from_state="pending", + ) + + class FakeDispatcher: + async def start_uploaded_file_parse( + self, + *, + job_id: str, + user_id: str, + ) -> str: + del job_id, user_id + raise AssertionError("CAS loser must not dispatch a duplicate task") + + service = DocumentIngestionHandoffService( + state_machine=FakeStateMachine(), + worker_dispatcher=FakeDispatcher(), + ) + + await service.start_uploaded_file_workflow( + db=cast(object, None), + job=SimpleNamespace( + job_id="job-race", + job_type="document_ingestion", + status="waiting-file", + ), + user_id="contract-user", + trigger="s3_upload_completed", + ) + + @pytest.mark.asyncio async def test_should_accept_a_pre_rename_waiting_file_job_type_during_upload_handoff( api_client_factory: Callable[[], AbstractAsyncContextManager[AsyncClient]], diff --git a/apps/worker/app/services/document_ingestion/processing_run.py b/apps/worker/app/services/document_ingestion/processing_run.py index 4eacc259d..cd5bcfce0 100644 --- a/apps/worker/app/services/document_ingestion/processing_run.py +++ b/apps/worker/app/services/document_ingestion/processing_run.py @@ -32,7 +32,10 @@ cleanup_stage_tracker, init_stage_tracker, ) -from shared.core.exceptions.domain_exceptions import ValidationException +from shared.core.exceptions.domain_exceptions import ( + UnavailableException, + ValidationException, +) from shared.models.schemas.job_metadata import JobMetadataHelper from shared.services.ai.llm_overrides import cleanup_llm_overrides, init_llm_overrides from shared.services.ai.token_tracking import cleanup_token_tracker, init_token_tracker @@ -63,21 +66,39 @@ def execute(self, job_id: str, user_id: str | None) -> dict[str, object]: "reason": "job_already_terminal", } - with RedisJobLock(job_context.redis_service, job_id): - task_workspace = TemporaryParseWorkspace.create(job_id) - try: - result = _run_parse_job( - job_id=job_id, - job_context=job_context, - lifecycle_service=lifecycle_service, - task_workspace=task_workspace, - ) - finally: - task_workspace.cleanup() + try: + with RedisJobLock(job_context.redis_service, job_id): + task_workspace = TemporaryParseWorkspace.create(job_id) + try: + result = _run_parse_job( + job_id=job_id, + job_context=job_context, + lifecycle_service=lifecycle_service, + task_workspace=task_workspace, + ) + finally: + task_workspace.cleanup() + except UnavailableException as exc: + if not _is_processing_lock_contention(exc): + raise + logger.info( + "Skipping duplicate parse delivery while another worker owns " + f"the processing lock: job_id={job_id}" + ) + return { + "status": "skipped", + "job_id": job_id, + "reason": "job_already_processing", + } return result +def _is_processing_lock_contention(error: UnavailableException) -> bool: + """Identify lock contention without swallowing unrelated 503 failures.""" + return error.internal_message.startswith("Could not acquire processing lock") + + def _run_parse_job( *, job_id: str, diff --git a/apps/worker/tests/contract/test_parse_task_contract.py b/apps/worker/tests/contract/test_parse_task_contract.py index 66267f357..86f336d0b 100644 --- a/apps/worker/tests/contract/test_parse_task_contract.py +++ b/apps/worker/tests/contract/test_parse_task_contract.py @@ -121,6 +121,98 @@ def test_parse_task_should_process_uploaded_file_through_real_contract_boundarie assert contract.find_task_workspaces(tmp_path, job["job_id"]) == [] +def test_parse_task_should_publish_each_non_null_source_path_once( + 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) + + source_file_name = "contract-duplicate-path.xlsx" + job = contract.create_file_job( + source_file_name=source_file_name, + job_id_prefix="job_duplicate_path", + ) + contract.upload_source_file( + local_file_path=_SAMPLE_XLSX_PATH, + s3_key=job["s3_key"], + ) + + def fake_execute_document_parse( + *, + job_id: str, + job_context: object, + prepared_source: object, + output_dir: str, + ) -> ParseOutput: + del job_id, job_context, prepared_source + parsed_df = pd.DataFrame( + [ + { + "know_id": "first-shared-path", + "type": "text", + "content": "first shared-path chunk", + "path": f"{source_file_name}/Root/Shared", + }, + { + "know_id": "duplicate-shared-path", + "type": "text", + "content": "duplicate shared-path chunk", + "path": f"{source_file_name}/Root/Shared", + }, + { + "know_id": "first-null-path", + "type": "text", + "content": "first null-path chunk", + "path": "", + }, + { + "know_id": "second-null-path", + "type": "text", + "content": "second null-path chunk", + "path": "", + }, + ] + ) + return ParseOutput(output_dir=output_dir, parsed_df=parsed_df) + + monkeypatch.setattr( + "app.services.document_ingestion.processing_run.execute_document_parse", + fake_execute_document_parse, + ) + + celery_result = contract.enqueue_parse_task( + job_id=job["job_id"], + user_id=job["user_id"], + ) + + assert celery_result.successful() + assert celery_result.result["status"] == "success" + + observed = contract.observe_successful_job(job["job_id"]) + job_chunks = observed["job_chunks"] + document_chunks = observed["document_chunks"] + + assert [row["chunk_id"] for row in job_chunks] == [ + "first-shared-path", + "duplicate-shared-path", + "first-null-path", + "second-null-path", + ] + assert [row["chunk_id"] for row in document_chunks] == [ + "first-shared-path", + "first-null-path", + "second-null-path", + ] + assert [row["source_chunk_path"] for row in document_chunks] == [ + f"{source_file_name}/Root/Shared", + None, + None, + ] + + def test_parse_task_result_zip_includes_page_citation_assets( worker_contract_environment: None, monkeypatch: pytest.MonkeyPatch, @@ -223,6 +315,87 @@ def fake_execute_document_parse( ] +def test_parse_task_sanitizes_nul_characters_at_database_boundary( + 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) + + source_file = tmp_path / "nul-source.txt" + source_file.write_text("source", encoding="utf-8") + job = contract.create_file_job( + source_file_name=source_file.name, + job_id_prefix="job_parse_nul", + ) + contract.upload_source_file( + local_file_path=source_file, + s3_key=job["s3_key"], + ) + + def fake_execute_document_parse( + *, + job_id: str, + job_context: object, + prepared_source: object, + output_dir: str, + ) -> ParseOutput: + del job_id, job_context, prepared_source + parsed_df = pd.DataFrame( + [ + { + "know_id": "chunk-with-nul", + "type": "text", + "content": "Text\x00content", + "path": f"{source_file.name}/Root\x00/Introduction", + "length": 12, + "keywords": "", + "summary": "Summary\x00value", + "tokens": "", + "connectto": "", + "page_nums": "1", + "extra_metadata": json.dumps({"note": "Nested\x00value"}), + } + ] + ) + return ParseOutput(output_dir=output_dir, parsed_df=parsed_df) + + monkeypatch.setattr( + "app.services.document_ingestion.processing_run.execute_document_parse", + fake_execute_document_parse, + ) + + celery_result = contract.enqueue_parse_task( + job_id=job["job_id"], + user_id=job["user_id"], + ) + + assert celery_result.successful() + assert celery_result.result["status"] == "success" + + observed = contract.observe_successful_job(job["job_id"]) + job_chunk = observed["job_chunks"][0] + document_chunk = observed["document_chunks"][0] + + assert job_chunk["text"] == "Textcontent" + assert job_chunk["path"] == f"{source_file.name}/Root/Introduction" + assert job_chunk["chunk_metadata"]["summary"] == "Summaryvalue" + assert job_chunk["chunk_metadata"]["note"] == "Nestedvalue" + assert document_chunk["content"] == "Textcontent" + assert document_chunk["source_chunk_path"] == ( + f"{source_file.name}/Root/Introduction" + ) + + result_zip = contract.read_result_zip( + result_s3_key=observed["result"]["result_s3_key"], + tmp_path=tmp_path, + ) + archived_chunk = result_zip["chunks"]["chunks"][0] + assert archived_chunk["content"] == "Text\x00content" + + def test_parse_task_should_charge_user_when_billing_is_enabled( worker_contract_environment: None, monkeypatch: pytest.MonkeyPatch, diff --git a/apps/worker/tests/contract/test_processing_run_contract.py b/apps/worker/tests/contract/test_processing_run_contract.py new file mode 100644 index 000000000..5078ebe63 --- /dev/null +++ b/apps/worker/tests/contract/test_processing_run_contract.py @@ -0,0 +1,65 @@ +from __future__ import annotations + +from types import SimpleNamespace + +import pytest + +from shared.core.exceptions.domain_exceptions import UnavailableException + + +def test_should_skip_duplicate_delivery_when_processing_lock_is_held( + worker_contract_environment: None, + monkeypatch: pytest.MonkeyPatch, +) -> None: + import app.services.document_ingestion.processing_run as processing_run + + class FakeLock: + def __init__(self, *args: object, **kwargs: object) -> None: + del args, kwargs + + def __enter__(self) -> "FakeLock": + raise UnavailableException( + internal_message="Could not acquire processing lock for job job-lock", + retry_after=120, + ) + + def __exit__(self, *args: object) -> bool: + del args + return False + + monkeypatch.setattr( + processing_run, + "load_parse_job_context", + lambda *args, **kwargs: SimpleNamespace(redis_service=object()), + ) + monkeypatch.setattr(processing_run, "mark_job_running", lambda *args: True) + monkeypatch.setattr( + processing_run, + "get_sync_job_lifecycle_service", + lambda: object(), + ) + monkeypatch.setattr(processing_run, "RedisJobLock", FakeLock) + + result = processing_run.DocumentProcessingRun().execute( + job_id="job-lock", + user_id="contract-user", + ) + + assert result == { + "status": "skipped", + "job_id": "job-lock", + "reason": "job_already_processing", + } + + +def test_should_not_treat_other_unavailable_errors_as_lock_contention( + worker_contract_environment: None, +) -> None: + import app.services.document_ingestion.processing_run as processing_run + + error = UnavailableException( + internal_message="Job state is still settling", + retry_after=120, + ) + + assert processing_run._is_processing_lock_contention(error) is False diff --git a/packages/shared-python/shared/services/jobs/lifecycle/result_writer.py b/packages/shared-python/shared/services/jobs/lifecycle/result_writer.py index 22983a1ec..36a60abf5 100644 --- a/packages/shared-python/shared/services/jobs/lifecycle/result_writer.py +++ b/packages/shared-python/shared/services/jobs/lifecycle/result_writer.py @@ -1,12 +1,13 @@ from __future__ import annotations -from typing import Any +from typing import Any, cast from uuid import uuid4 from sqlalchemy import delete, select from sqlalchemy.orm import Session from shared.models.database.job_result import JobChunk, JobResult +from shared.utils.json_utils import remove_nul_characters class SyncJobResultWriter: @@ -59,23 +60,24 @@ def replace_chunks( chunk_models = [] for index, chunk in enumerate(chunks): - chunk_identifier = chunk.get("chunk_id") or str(uuid4()) - metadata = chunk.get("metadata") - chunk_text = chunk.get("text") or chunk.get("content") + safe_chunk = cast(dict[str, Any], remove_nul_characters(chunk)) + chunk_identifier = safe_chunk.get("chunk_id") or str(uuid4()) + metadata = safe_chunk.get("metadata") + chunk_text = safe_chunk.get("text") or safe_chunk.get("content") chunk_path = ( metadata.get("path") if isinstance(metadata, dict) and metadata.get("path") - else chunk.get("path") + else safe_chunk.get("path") ) chunk_models.append( JobChunk( job_result_id=job_result_id, - chunk_id=chunk_identifier, - chunk_type=chunk.get("type", "paragraph"), + chunk_id=str(chunk_identifier), + chunk_type=str(safe_chunk.get("type", "paragraph")), text=str(chunk_text) if chunk_text is not None else None, path=str(chunk_path) if chunk_path is not None else None, chunk_metadata=metadata, - sort_order=chunk.get("order", index), + sort_order=safe_chunk.get("order", index), ) ) db.add_all(chunk_models) diff --git a/packages/shared-python/shared/services/retrieval/publication_content.py b/packages/shared-python/shared/services/retrieval/publication_content.py index deed9135e..cf81e0362 100644 --- a/packages/shared-python/shared/services/retrieval/publication_content.py +++ b/packages/shared-python/shared/services/retrieval/publication_content.py @@ -1,6 +1,6 @@ from __future__ import annotations -from typing import Any +from typing import Any, cast from uuid import uuid4 from sqlalchemy import delete @@ -16,6 +16,34 @@ build_term_search_text, section_path_from_chunk_path, ) +from shared.utils.json_utils import remove_nul_characters + + +def deduplicate_chunks_by_source_path( + chunks: list[dict[str, Any]], +) -> list[dict[str, Any]]: + """Keep the first chunk for each non-null source path. + + The revision-path uniqueness constraint applies to non-null + ``source_chunk_path`` values. Text chunks without a source path are + intentionally retained because PostgreSQL permits multiple NULL values + for that constraint and those chunks can carry distinct content. + """ + seen_source_paths: set[str] = set() + deduplicated_chunks: list[dict[str, Any]] = [] + + for chunk in chunks: + source_path = _get_source_path( + chunk=chunk, + chunk_metadata=_get_chunk_metadata(chunk), + ) + if source_path is not None: + if source_path in seen_source_paths: + continue + seen_source_paths.add(source_path) + deduplicated_chunks.append(chunk) + + return deduplicated_chunks def replace_document_revision_content( @@ -31,8 +59,12 @@ def replace_document_revision_content( db=db, scope=scope, section_summaries=section_summaries, ) for index, chunk in enumerate(chunks): - chunk_metadata = _get_chunk_metadata(chunk) - source_path = _get_source_path(chunk=chunk, chunk_metadata=chunk_metadata) + safe_chunk = cast(dict[str, Any], remove_nul_characters(chunk)) + chunk_metadata = _get_chunk_metadata(safe_chunk) + source_path = _get_source_path( + chunk=safe_chunk, + chunk_metadata=chunk_metadata, + ) section_path = section_path_from_chunk_path( source_path, source_file_name=scope.source_file_name, @@ -40,7 +72,7 @@ def replace_document_revision_content( section = section_publisher.ensure_section(section_path) db.add( _build_document_chunk( - chunk=chunk, + chunk=safe_chunk, chunk_metadata=chunk_metadata, source_path=source_path, section=section, @@ -60,7 +92,10 @@ def __init__( ) -> None: self._db = db self._scope = scope - self._section_summaries = section_summaries or {} + self._section_summaries = cast( + dict[str, str], + remove_nul_characters(section_summaries or {}), + ) self._sections_by_path: dict[str, DocumentSection] = {} def ensure_section(self, section_path: str) -> DocumentSection: @@ -179,7 +214,11 @@ def _get_source_path( chunk_metadata: dict[str, Any], ) -> str | None: source_path = chunk_metadata.get("path") or chunk.get("path") - return str(source_path) if source_path is not None else None + if source_path is None: + return None + + normalized_source_path = str(source_path) + return normalized_source_path or None def _get_sort_order(chunk: dict[str, Any], fallback_sort_order: int) -> int: diff --git a/packages/shared-python/shared/services/retrieval/publication_service.py b/packages/shared-python/shared/services/retrieval/publication_service.py index 8911c3b7c..e201916e5 100644 --- a/packages/shared-python/shared/services/retrieval/publication_service.py +++ b/packages/shared-python/shared/services/retrieval/publication_service.py @@ -23,6 +23,7 @@ from shared.models.schemas.retrieval_namespace import normalize_retrieval_namespace from shared.services.retrieval.graph.service import DocumentGraphService, GraphScope from shared.services.retrieval.publication_content import ( + deduplicate_chunks_by_source_path, replace_document_revision_content, ) from shared.services.retrieval.publication_models import ( @@ -106,7 +107,7 @@ def _publish_document_state_for_job( ) document_metadata = JobMetadataHelper.get_document_metadata(job_metadata) - deduped_chunks = chunks + deduped_chunks = deduplicate_chunks_by_source_path(chunks) # If ALL chunks are duplicates → skip document creation entirely if not deduped_chunks: From 08c2fbd9da1e5e4cd7314ed2e44c2ad156567b49 Mon Sep 17 00:00:00 2001 From: suguanYang Date: Wed, 19 Aug 2026 04:17:10 +0800 Subject: [PATCH 3/5] fix: deduplicate sanitized chunk paths --- .../shared/services/retrieval/publication_content.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/packages/shared-python/shared/services/retrieval/publication_content.py b/packages/shared-python/shared/services/retrieval/publication_content.py index cf81e0362..9ebbdd74a 100644 --- a/packages/shared-python/shared/services/retrieval/publication_content.py +++ b/packages/shared-python/shared/services/retrieval/publication_content.py @@ -33,9 +33,10 @@ def deduplicate_chunks_by_source_path( deduplicated_chunks: list[dict[str, Any]] = [] for chunk in chunks: + safe_chunk = cast(dict[str, Any], remove_nul_characters(chunk)) source_path = _get_source_path( - chunk=chunk, - chunk_metadata=_get_chunk_metadata(chunk), + chunk=safe_chunk, + chunk_metadata=_get_chunk_metadata(safe_chunk), ) if source_path is not None: if source_path in seen_source_paths: From 330737ab8a58bdcc97cff2f6cfb21d0910c47ec3 Mon Sep 17 00:00:00 2001 From: suguanYang Date: Wed, 19 Aug 2026 04:31:29 +0800 Subject: [PATCH 4/5] test: cover sanitized chunk path collisions --- apps/worker/tests/contract/test_parse_task_contract.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/apps/worker/tests/contract/test_parse_task_contract.py b/apps/worker/tests/contract/test_parse_task_contract.py index 86f336d0b..398da4e31 100644 --- a/apps/worker/tests/contract/test_parse_task_contract.py +++ b/apps/worker/tests/contract/test_parse_task_contract.py @@ -162,6 +162,12 @@ def fake_execute_document_parse( "content": "duplicate shared-path chunk", "path": f"{source_file_name}/Root/Shared", }, + { + "know_id": "nul-duplicate-shared-path", + "type": "text", + "content": "NUL duplicate shared-path chunk", + "path": f"{source_file_name}/Root\x00/Shared", + }, { "know_id": "first-null-path", "type": "text", @@ -198,6 +204,7 @@ def fake_execute_document_parse( assert [row["chunk_id"] for row in job_chunks] == [ "first-shared-path", "duplicate-shared-path", + "nul-duplicate-shared-path", "first-null-path", "second-null-path", ] From 3c6f41999ca74da7bcc494f149593ddd3296feeb Mon Sep 17 00:00:00 2001 From: suguanYang Date: Wed, 19 Aug 2026 10:28:28 +0800 Subject: [PATCH 5/5] test: address ingestion review comments --- .../tests/contract/test_processing_run_contract.py | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/apps/worker/tests/contract/test_processing_run_contract.py b/apps/worker/tests/contract/test_processing_run_contract.py index 5078ebe63..d219627fb 100644 --- a/apps/worker/tests/contract/test_processing_run_contract.py +++ b/apps/worker/tests/contract/test_processing_run_contract.py @@ -14,8 +14,8 @@ def test_should_skip_duplicate_delivery_when_processing_lock_is_held( import app.services.document_ingestion.processing_run as processing_run class FakeLock: - def __init__(self, *args: object, **kwargs: object) -> None: - del args, kwargs + def __init__(self, _redis_service: object, _job_id: str) -> None: + pass def __enter__(self) -> "FakeLock": raise UnavailableException( @@ -23,8 +23,7 @@ def __enter__(self) -> "FakeLock": retry_after=120, ) - def __exit__(self, *args: object) -> bool: - del args + def __exit__(self, *_args: object) -> bool: return False monkeypatch.setattr( @@ -36,7 +35,7 @@ def __exit__(self, *args: object) -> bool: monkeypatch.setattr( processing_run, "get_sync_job_lifecycle_service", - lambda: object(), + object, ) monkeypatch.setattr(processing_run, "RedisJobLock", FakeLock)