diff --git a/backend/app/services/pipeline_service.py b/backend/app/services/pipeline_service.py index df21d28..b9c04ed 100644 --- a/backend/app/services/pipeline_service.py +++ b/backend/app/services/pipeline_service.py @@ -33,6 +33,12 @@ PK = "PIPELINE" PK_RUN = "PIPELINERUN" + + +def _run_pointer_key(run_id: str) -> dict: + """Each run also writes ``PK=PIPELINERUN#, SK=META`` holding the run's + sort key, so a run is found by id however many runs came after it.""" + return {"PK": f"{PK_RUN}#{run_id}", "SK": "META"} SOURCE = "pipeline" MAX_SCRIPT = 100_000 MAX_HISTORY = 10 @@ -292,11 +298,21 @@ def pct(sorted_vals: list[float], p: int) -> float | None: }) def get_run(self, run_id: str) -> dict | None: + pointer = self.table.get_item(Key=_run_pointer_key(run_id)).get("Item") + if pointer: + return self._get_run_by_sk(pointer["run_sk"]) + # Runs created before the pointer existed: look in the recent window. for run in self.list_runs(limit=50): if run["id"] == run_id: return run return None + def _get_run_by_sk(self, sk: str) -> dict | None: + # Strongly consistent: callers read a run right after writing it + # (start_run returns it; run_sync reads the final status to report ok). + item = self.table.get_item(Key={"PK": PK_RUN, "SK": sk}, ConsistentRead=True).get("Item") + return self._run_public(item) if item else None + def _update_run(self, sk: str, **fields) -> None: # alias every name: status/error are DynamoDB reserved words expr, names, values = [], {}, {} @@ -331,6 +347,7 @@ def _create_run(self, pipeline: str, user: str, source: str, if parent_run: item["parent_run"] = parent_run self.table.put_item(Item=item) + self.table.put_item(Item={**_run_pointer_key(run_id), "run_sk": sk}) return sk # ------------------------------------------------------------ entrypoints @@ -347,7 +364,7 @@ def start_run(self, *, name: str, user: str, args=None) -> dict: ) self._tasks.add(task) task.add_done_callback(self._tasks.discard) - return self.get_run(sk.partition("#")[2]) or { + return self._get_run_by_sk(sk) or { "id": sk.partition("#")[2], "pipeline": name, "status": "running"} def run_sync(self, name: str, user: str = "scheduler", *, args=None, @@ -365,7 +382,7 @@ def run_sync(self, name: str, user: str = "scheduler", *, args=None, sk = self._create_run(name, user, source, parent_run=parent_run) self._execute(sk, pipe, user, args, allow_nested=not nested, timeout_s=timeout_s) - run = self.get_run(sk.partition("#")[2]) or {} + run = self._get_run_by_sk(sk) or {} return {"ok": run.get("status") == "completed", "run_id": run.get("id"), "result": _plain(run.get("result"))} diff --git a/backend/tests/test_pipeline_runs.py b/backend/tests/test_pipeline_runs.py new file mode 100644 index 0000000..e7bd10a --- /dev/null +++ b/backend/tests/test_pipeline_runs.py @@ -0,0 +1,97 @@ +"""Pipeline runs are found by id, and their final status is read consistently.""" + +import importlib +import sys +import types +import unittest +from pathlib import Path +from unittest.mock import patch + +import boto3 + + +class FakeTable: + """Items keyed by (PK, SK) and the Query shape list_runs uses.""" + + def __init__(self): + self.items: dict[tuple[str, str], dict] = {} + self.consistent_reads: list[tuple[str, str]] = [] + + def put_item(self, **kwargs): + item = kwargs["Item"] + self.items[(item["PK"], item["SK"])] = dict(item) + + def get_item(self, **kwargs): + key = (kwargs["Key"]["PK"], kwargs["Key"]["SK"]) + if kwargs.get("ConsistentRead"): + self.consistent_reads.append(key) + return {"Item": dict(self.items[key])} if key in self.items else {} + + def update_item(self, **kwargs): + key = (kwargs["Key"]["PK"], kwargs["Key"]["SK"]) + names = kwargs.get("ExpressionAttributeNames", {}) + values = kwargs["ExpressionAttributeValues"] + item = self.items.setdefault(key, {"PK": key[0], "SK": key[1]}) + for index in range(len(names)): + item[names[f"#f{index}"]] = values[f":v{index}"] + + def query(self, **kwargs): + pk = kwargs["ExpressionAttributeValues"][":pk"] + rows = [dict(v) for (p, _), v in self.items.items() if p == pk] + rows.sort(key=lambda r: r["SK"], reverse=not kwargs.get("ScanIndexForward", True)) + return {"Items": rows[: kwargs["Limit"]] if "Limit" in kwargs else rows} + + +def load_pipeline_module(): + sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + config = types.ModuleType("app.config") + config.settings = types.SimpleNamespace(aws_region="us-east-1", dynamo_table="test") + resource = types.SimpleNamespace(Table=lambda name: FakeTable()) + with patch.dict(sys.modules, {"app.config": config}), patch.object(boto3, "resource", return_value=resource): + return importlib.import_module("app.services.pipeline_service") + + +class PipelineRunLookupTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.module = load_pipeline_module() + + @classmethod + def tearDownClass(cls): + sys.modules.pop("app.services.pipeline_service", None) + + def setUp(self): + self.service = self.module.PipelineService.__new__(self.module.PipelineService) + self.service.table = FakeTable() + self.service._tasks = set() + + def test_run_is_found_by_id_after_many_newer_runs(self): + sk = self.service._create_run("nightly", "tester", "api") + run_id = sk.partition("#")[2] + for i in range(60): # newer than the 50-run window list_runs reads + self.service._create_run("other", "tester", "api") + self.assertNotIn(run_id, [r["id"] for r in self.service.list_runs(limit=50)]) + self.assertEqual(self.service.get_run(run_id)["id"], run_id) + + def test_run_sync_reports_final_status_with_a_consistent_read(self): + def execute(sk, pipe, user, args, **kwargs): + self.service._update_run(sk, status="completed", result={"ok": True}) + + with patch.object(self.service, "get_pipeline", return_value={"name": "nightly"}), \ + patch.object(self.service, "_execute", side_effect=execute): + out = self.service.run_sync("nightly", "tester") + self.assertTrue(out["ok"]) + self.assertIn(("PIPELINERUN", next(sk for (pk, sk) in self.service.table.items if pk == "PIPELINERUN")), + self.service.table.consistent_reads) + + def test_run_created_before_pointers_is_still_found_in_recent_window(self): + self.service.table.put_item(Item={ + "PK": "PIPELINERUN", "SK": "2026-01-01T00:00:00+00:00#legacy1", "run_id": "legacy1", + "pipeline": "nightly", "status": "completed", "agents": [], "logs": [], + }) + self.assertEqual(self.service.get_run("legacy1")["id"], "legacy1") + self.assertIsNone(self.service.get_run("missing")) + + +if __name__ == "__main__": + unittest.main()