diff --git a/src/wf_artifacts/store.py b/src/wf_artifacts/store.py index 2daa6836..0fc1a4db 100644 --- a/src/wf_artifacts/store.py +++ b/src/wf_artifacts/store.py @@ -3,6 +3,7 @@ from __future__ import annotations import json import re from pathlib import Path +from threading import RLock from .models import WorkflowArtifact, WorkflowDeployment @@ -65,6 +66,7 @@ class FileWorkflowArtifactStore(WorkflowArtifactStore): def __init__(self, root: Path) -> None: self.root = root + self._lock = RLock() self.artifacts_dir.mkdir(parents=True, exist_ok=True) self.deployments_dir.mkdir(parents=True, exist_ok=True) @@ -111,20 +113,25 @@ class FileWorkflowArtifactStore(WorkflowArtifactStore): return self.get_artifact(artifact_id, max(versions)) def save_deployment(self, deployment: WorkflowDeployment) -> None: - path = self._deployment_path(deployment.id) - if path.exists(): - existing = WorkflowDeployment.model_validate_json( - path.read_text(encoding="utf-8") + # Revision advancement is store-owned and must cover the read, + # increment, and write as one process-local transition. Callers + # intentionally continue sending the existing deployment payload + # without a mandatory expected-revision field. + with self._lock: + path = self._deployment_path(deployment.id) + if path.exists(): + existing = WorkflowDeployment.model_validate_json( + path.read_text(encoding="utf-8") + ) + deployment = deployment.model_copy( + update={"revision": existing.revision + 1} + ) + elif deployment.revision != 1: + deployment = deployment.model_copy(update={"revision": 1}) + path.write_text( + json.dumps(deployment.model_dump(mode="json"), indent=2), + encoding="utf-8", ) - deployment = deployment.model_copy( - update={"revision": existing.revision + 1} - ) - elif deployment.revision != 1: - deployment = deployment.model_copy(update={"revision": 1}) - path.write_text( - json.dumps(deployment.model_dump(mode="json"), indent=2), - encoding="utf-8", - ) def get_deployment(self, deployment_id: str) -> WorkflowDeployment: path = self._deployment_path(deployment_id) diff --git a/tests/artifacts/test_store.py b/tests/artifacts/test_store.py index a2516736..566b9fa5 100644 --- a/tests/artifacts/test_store.py +++ b/tests/artifacts/test_store.py @@ -1,6 +1,7 @@ from __future__ import annotations import json +from concurrent.futures import ThreadPoolExecutor import pytest @@ -67,6 +68,32 @@ def test_file_store_round_trips_deployment(tmp_path) -> None: assert loaded.binding_map()["context7"] == "context7.personal" +def test_concurrent_deployment_saves_advance_revision_without_lost_updates( + tmp_path, +) -> None: + store = FileWorkflowArtifactStore(tmp_path) + store.save_deployment( + WorkflowDeployment( + id="concurrent.personal", + artifact_id="summarize_docs", + artifact_version=1, + ) + ) + updates = [ + WorkflowDeployment( + id="concurrent.personal", + artifact_id="summarize_docs", + artifact_version=version, + ) + for version in range(2, 10) + ] + + with ThreadPoolExecutor(max_workers=len(updates)) as executor: + list(executor.map(store.save_deployment, updates)) + + assert store.get_deployment("concurrent.personal").revision == 9 + + def test_file_store_loads_legacy_artifact_and_rewrites_canonical_shape( tmp_path, ) -> None: