From fdd7a14e9ccd7779698cb6448234c043360c9818 Mon Sep 17 00:00:00 2001 From: Jay Moran Date: Thu, 1 Oct 2026 19:01:35 -0700 Subject: [PATCH] Lock protocol run row in update_node_run Parallel supervisor workers each merged their own key into node_runs via an unlocked read-modify-write, so one worker's stale snapshot could overwrite a sibling's completed status. --- src/asaree/services/protocol_runs.py | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/src/asaree/services/protocol_runs.py b/src/asaree/services/protocol_runs.py index 182be238..14e807a1 100644 --- a/src/asaree/services/protocol_runs.py +++ b/src/asaree/services/protocol_runs.py @@ -327,8 +327,19 @@ async def update_node_run( ) -> ProtocolRun | None: """Shallow-merge *patch* into ``node_runs[node_id]`` -- the same read-modify-write idiom ``upsert_replicate`` uses for its JSONB columns, one - level deeper (merging into one key of the blob, not the blob itself).""" - run = await get_protocol_run(db, protocol_run_id) + level deeper (merging into one key of the blob, not the blob itself). + + The row is locked for the merge: a supervisor's parallel workers each write + their own key concurrently, and an unlocked read-modify-write let one + worker's stale snapshot overwrite a sibling's finished status.""" + run = ( + await db.execute( + select(ProtocolRun) + .where(ProtocolRun.id == protocol_run_id) + .with_for_update() + .execution_options(populate_existing=True) + ) + ).scalar_one_or_none() if run is None: return None node_runs = dict(run.node_runs or {})