diff --git a/src/k1link/observatory/portable_queue_binding.py b/src/k1link/observatory/portable_queue_binding.py index 2e6f54b..75f9ce1 100644 --- a/src/k1link/observatory/portable_queue_binding.py +++ b/src/k1link/observatory/portable_queue_binding.py @@ -214,6 +214,7 @@ class PortableRecordedQueueBindingService: return self._queue.submit( preparation.intent(idempotency_key=idempotency_key), enqueue=enqueue, + reject_duplicate_computation=True, ) def _resolve_definition( diff --git a/src/k1link/observatory/recorded_jobs.py b/src/k1link/observatory/recorded_jobs.py index 5ac9385..dfbc3fd 100644 --- a/src/k1link/observatory/recorded_jobs.py +++ b/src/k1link/observatory/recorded_jobs.py @@ -239,6 +239,14 @@ class ObservatoryRecordedQueueConflictError(ObservatoryRecordedQueueError): """An idempotency identity or state transition conflicts with history.""" +class ObservatoryRecordedQueueDuplicateError(ObservatoryRecordedQueueConflictError): + """Another request already owns this exact recorded computation.""" + + def __init__(self, job_id: str) -> None: + self.job_id = job_id + super().__init__("an identical recorded computation already exists") + + class ObservatoryRecordedQueueNotFoundError(ObservatoryRecordedQueueError): """A requested job or live lease does not exist.""" @@ -1177,6 +1185,7 @@ class ObservatoryRecordedJobQueue: intent: ObservatoryRecordedJobIntent, *, enqueue: bool = False, + reject_duplicate_computation: bool = False, ) -> tuple[ObservatoryRecordedJob, bool]: """Accept one immutable identity and optionally queue it atomically.""" @@ -1199,16 +1208,32 @@ class ObservatoryRecordedJobQueue: ) record = self._get_job(connection, record.job_id) return record, False + definition = self._definitions.resolve(intent.setup_id, intent.definition_sha256) + identity_sha256 = _job_identity_sha256(intent, definition) + if reject_duplicate_computation: + # This check shares the INSERT transaction: two clients with + # different idempotency keys cannot race into duplicate jobs. + # A sealed result awaiting publication is not a reason to run + # the models again. Published-cache validity is a separate gate. + duplicate = connection.execute( + "SELECT job_id FROM observatory_recorded_jobs " + "WHERE identity_sha256 = ? AND (" + "state IN ('accepted', 'queued', 'claimed', 'running', 'paused', " + "'preemption-pending', 'reconciliation-required') OR " + "(state = 'succeeded' AND publication_state IN ('pending', 'failed'))) " + "ORDER BY created_at_utc DESC, job_id DESC LIMIT 1", + (identity_sha256,), + ).fetchone() + if duplicate is not None: + raise ObservatoryRecordedQueueDuplicateError(duplicate["job_id"]) self._require_capacity( connection, table="observatory_recorded_jobs", limit=self._max_jobs, label="recorded job", ) - definition = self._definitions.resolve(intent.setup_id, intent.definition_sha256) now = self._timestamp() job_id = f"observatory-run-{uuid4().hex}" - identity_sha256 = _job_identity_sha256(intent, definition) receipt_sha256 = _submission_receipt_sha256( job_id=job_id, request_sha256=intent.request_sha256, diff --git a/src/k1link/web/observatory_api.py b/src/k1link/web/observatory_api.py index 822d9fe..180e938 100644 --- a/src/k1link/web/observatory_api.py +++ b/src/k1link/web/observatory_api.py @@ -53,6 +53,7 @@ from k1link.observatory.recorded_jobs import ( ObservatoryRecordedJobQueue, ObservatoryRecordedQueueCapacityError, ObservatoryRecordedQueueConflictError, + ObservatoryRecordedQueueDuplicateError, ObservatoryRecordedQueueError, ObservatoryRecordedQueueNotFoundError, ) @@ -929,6 +930,14 @@ def build_observatory_router( status_code=409, detail="Portable-привязка источника не прошла проверку целостности.", ) from exc + except ObservatoryRecordedQueueDuplicateError as exc: + raise HTTPException( + status_code=409, + detail=( + "Такой расчёт уже выполняется или ожидает публикации. " + "Обновите список расчётов; повторный запуск не создан." + ), + ) from exc except ObservatoryRecordedQueueConflictError as exc: try: raced = recorded_job_queue.get_by_idempotency_key(request.idempotency_key) diff --git a/tests/test_observatory_portable_queue_binding.py b/tests/test_observatory_portable_queue_binding.py index 1a62bf8..b66d71f 100644 --- a/tests/test_observatory_portable_queue_binding.py +++ b/tests/test_observatory_portable_queue_binding.py @@ -18,7 +18,10 @@ from k1link.observatory.portable_run_definitions import ( PortableRunDefinitionUnavailableError, canonical_sha256, ) -from k1link.observatory.recorded_jobs import ObservatoryRecordedJobQueue +from k1link.observatory.recorded_jobs import ( + ObservatoryRecordedJobQueue, + ObservatoryRecordedQueueDuplicateError, +) from k1link.observatory.source_admission import ( PORTABLE_SOURCE_DOCUMENT_DIRECTORY, PortableRecordedSourceAdmission, @@ -534,6 +537,16 @@ def test_admit_persists_exact_documents_and_submit_queues_same_identity( ) assert queue is not None assert queue.list_jobs() == (job,) + with pytest.raises(ObservatoryRecordedQueueDuplicateError) as caught: + service.submit( + source_session_id=SESSION_ID, + setup_id=definition.setup_id, + definition_sha256=definition.definition_sha256, + expected_check_sha256=checked.check_sha256, + idempotency_key="second-browser", + ) + assert caught.value.job_id == job.job_id + assert queue.list_jobs() == (job,) def test_admit_rejects_catalog_change_since_check_without_writes_or_job( diff --git a/tests/test_observatory_portable_setup_api.py b/tests/test_observatory_portable_setup_api.py index 549b980..90316c0 100644 --- a/tests/test_observatory_portable_setup_api.py +++ b/tests/test_observatory_portable_setup_api.py @@ -242,6 +242,7 @@ class _PortableBinding: definition_sha256=definition_sha256, ), enqueue=True, + reject_duplicate_computation=True, ) @@ -299,6 +300,23 @@ def test_portable_api_check_sha_fences_ready_submission(tmp_path: Path) -> None: assert submitted.json()["state"] == "queued" assert submitted.json()["setup"]["setup_id"] == definition.setup_id assert binding.submit_count == 1 + request = { + "schema_version": "missioncore.observatory-recorded-run-submit/v1", + "idempotency_key": "portable:lab-v1:source-005", + "source_session_id": SOURCE_SESSION_ID, + "setup_id": definition.setup_id, + "definition_sha256": definition.definition_sha256, + "check_sha256": binding.check_sha256, + } + repeated = client.post("/api/v1/observatory/runs", json=request) + assert repeated.status_code == 202 + assert repeated.json()["job_id"] == submitted.json()["job_id"] + assert binding.submit_count == 1 + request["idempotency_key"] = "second-browser" + duplicate = client.post("/api/v1/observatory/runs", json=request) + assert duplicate.status_code == 409 + assert "повторный запуск не создан" in duplicate.json()["detail"] + assert len(queue.list_jobs()) == 1 def test_portable_api_rejects_blocked_lab_executor_before_binding_submit( diff --git a/tests/test_observatory_recorded_jobs.py b/tests/test_observatory_recorded_jobs.py index a608e0c..34f9ce6 100644 --- a/tests/test_observatory_recorded_jobs.py +++ b/tests/test_observatory_recorded_jobs.py @@ -1,8 +1,10 @@ from __future__ import annotations import sqlite3 +from concurrent.futures import ThreadPoolExecutor from dataclasses import replace from pathlib import Path +from threading import Barrier import pytest @@ -17,6 +19,7 @@ from k1link.observatory.recorded_jobs import ( ObservatoryRecordedPreemptionError, ObservatoryRecordedQueueBusyError, ObservatoryRecordedQueueConflictError, + ObservatoryRecordedQueueDuplicateError, ObservatoryRecordedQueueIntegrityError, ObservatoryRecordedQueueStaleClaimError, ObservatoryRecordedReconciliationRequest, @@ -198,6 +201,160 @@ def _reconciliation_request( ) +@pytest.mark.parametrize("enqueue", [False, True]) +def test_portable_duplicate_guard_preserves_original_request( + tmp_path: Path, + enqueue: bool, +) -> None: + queue = _queue(tmp_path) + first, created = queue.submit( + _intent(), + enqueue=enqueue, + reject_duplicate_computation=True, + ) + assert created + repeated, created = queue.submit( + _intent(), + enqueue=enqueue, + reject_duplicate_computation=True, + ) + assert repeated == first and not created + reopened = _queue(tmp_path) + with pytest.raises(ObservatoryRecordedQueueDuplicateError) as caught: + reopened.submit( + _intent(idempotency_key="second-browser"), + enqueue=True, + reject_duplicate_computation=True, + ) + assert caught.value.job_id == first.job_id + assert len(reopened.list_jobs()) == 1 + # Reject, rather than silently rebinding a second idempotency key. + with pytest.raises(ObservatoryRecordedQueueConflictError): + reopened.submit( + _intent(source_catalog_sha256="0" * 64), + reject_duplicate_computation=True, + ) + + +@pytest.mark.parametrize( + "field", + [ + "source_session_id", + "source_catalog_sha256", + "source_bundle_sha256", + "source_capability_manifest_sha256", + "setup_id", + ], +) +def test_duplicate_guard_keeps_distinct_sources_and_profiles( + tmp_path: Path, + field: str, +) -> None: + queue = _queue(tmp_path) + queue.submit(_intent(), reject_duplicate_computation=True) + values = { + "source_session_id": "another-source", + "source_catalog_sha256": "0" * 64, + "source_bundle_sha256": "0" * 64, + "source_capability_manifest_sha256": "0" * 64, + "setup_id": "legacy-monolith-v1", + } + changes = {field: values[field]} + if field == "setup_id": + changes["definition_sha256"] = NON_CHECKPOINTABLE_DEFINITION_SHA + distinct = replace(_intent(idempotency_key="distinct-request"), **changes) + _, created = queue.submit(distinct, reject_duplicate_computation=True) + assert created and len(queue.list_jobs()) == 2 + + +@pytest.mark.parametrize("publication_failed", [False, True]) +def test_duplicate_guard_never_recomputes_pending_publication( + tmp_path: Path, + publication_failed: bool, +) -> None: + queue = _queue(tmp_path) + job, claim = _running_job(queue) + with pytest.raises(ObservatoryRecordedQueueDuplicateError): + queue.submit( + _intent(idempotency_key="while-running"), + reject_duplicate_computation=True, + ) + queue.complete_for_publication( + job.job_id, + claim_token=claim.claim_token, + result_id="portable-result-001", + result_sha256=RESULT_SHA, + ) + if publication_failed: + queue.mark_publication_failed(job.job_id, message="temporary publication outage") + with pytest.raises(ObservatoryRecordedQueueDuplicateError): + queue.submit( + _intent(idempotency_key="while-publishing"), + reject_duplicate_computation=True, + ) + assert len(queue.list_jobs()) == 1 + + +def test_duplicate_guard_allows_retry_after_computation_failure(tmp_path: Path) -> None: + queue = _queue(tmp_path) + job, claim = _running_job(queue) + queue.fail( + job.job_id, + claim_token=claim.claim_token, + error_code="executor-error", + message="bounded test failure", + ) + retried, created = queue.submit( + _intent(idempotency_key="explicit-retry"), + enqueue=True, + reject_duplicate_computation=True, + ) + assert created and retried.job_id != job.job_id + + +def test_duplicate_guard_is_atomic_across_two_queue_instances(tmp_path: Path) -> None: + queues = (_queue(tmp_path), _queue(tmp_path)) + barrier = Barrier(2) + + def submit(index: int) -> str: + barrier.wait(timeout=5) + try: + job, created = queues[index].submit( + _intent(idempotency_key=f"browser-{index}"), + enqueue=True, + reject_duplicate_computation=True, + ) + assert created + return job.job_id + except ObservatoryRecordedQueueDuplicateError as error: + return error.job_id + + with ThreadPoolExecutor(max_workers=2) as executor: + results = tuple(executor.map(submit, (0, 1))) + assert results[0] == results[1] + assert len(queues[0].list_jobs()) == 1 + + +def test_duplicate_guard_keeps_new_version_of_same_setup(tmp_path: Path) -> None: + queue = _queue(tmp_path) + queue.submit(_intent(), reject_duplicate_computation=True) + revised = replace( + _definitions().definitions[0], + definition_version=2, + definition_sha256="0" * 64, + ) + reopened = ObservatoryRecordedJobQueue( + tmp_path, + definitions=RecordedRunDefinitionRegistry((revised,)), + clock=lambda: NOW, + ) + _, created = reopened.submit( + _intent(idempotency_key="new-version", definition_sha256="0" * 64), + reject_duplicate_computation=True, + ) + assert created and len(reopened.list_jobs()) == 2 + + def test_submission_is_durable_exactly_idempotent_and_path_free(tmp_path: Path) -> None: queue = _queue(tmp_path) first, first_created = queue.submit(_intent())