From d5eb9caacfa8ce962945290841eaeb99b4686dc9 Mon Sep 17 00:00:00 2001
From: roboomp <omp@can.ac>
Date: Thu, 4 Jun 2026 10:48:35 +0000
Subject: [PATCH] fix(robomp): serialized same-issue event claims

Prevented the durable queue from claiming a queued event while another event with the same issue key is still running, so duplicate worker instances cannot resume the same RpcClient session concurrently.

Added regression coverage for blocked same-issue events and for skipping blocked queue heads without stalling unrelated issues.

Fixes #1840
---
 python/robomp/src/db.py        | 25 ++++++++++----
 python/robomp/tests/test_db.py | 60 ++++++++++++++++++++++++++++++++++
 2 files changed, 79 insertions(+), 6 deletions(-)

diff --git a/python/robomp/src/db.py b/python/robomp/src/db.py
index 0e5762ba9..06906b59c 100644
--- a/python/robomp/src/db.py
+++ b/python/robomp/src/db.py
@@ -50,6 +50,9 @@ CREATE TABLE IF NOT EXISTS events (
 CREATE INDEX IF NOT EXISTS events_state_received
   ON events(state, received_at);
 
+CREATE INDEX IF NOT EXISTS events_issue_state
+  ON events(issue_key, state);
+
 CREATE TABLE IF NOT EXISTS issues (
   key            TEXT PRIMARY KEY,
   repo           TEXT NOT NULL,
@@ -293,15 +296,25 @@ class Database:
             return cur.rowcount > 0
 
     def claim_next_event(self) -> EventRow | None:
-        """Atomically dequeue one queued event into running state."""
+        """Atomically dequeue one unblocked queued event into running state."""
         with self._txn() as conn:
             row = conn.execute(
                 """
-                SELECT delivery_id, event_type, repo, issue_key, payload_json, received_at,
-                       state, attempts, last_error
-                FROM events
-                WHERE state = 'queued'
-                ORDER BY received_at
+                SELECT queued.delivery_id, queued.event_type, queued.repo, queued.issue_key,
+                       queued.payload_json, queued.received_at, queued.state, queued.attempts,
+                       queued.last_error
+                FROM events AS queued
+                WHERE queued.state = 'queued'
+                  AND (
+                    queued.issue_key IS NULL
+                    OR NOT EXISTS (
+                      SELECT 1
+                      FROM events AS running
+                      WHERE running.state = 'running'
+                        AND running.issue_key = queued.issue_key
+                    )
+                  )
+                ORDER BY queued.received_at
                 LIMIT 1
                 """
             ).fetchone()
diff --git a/python/robomp/tests/test_db.py b/python/robomp/tests/test_db.py
index 55aa05584..9ff0644e6 100644
--- a/python/robomp/tests/test_db.py
+++ b/python/robomp/tests/test_db.py
@@ -55,6 +55,66 @@ def test_claim_next_event_singleton_under_contention(db: Database) -> None:
     assert all(db.get_event(f"d-{i}").state == "running" for i in range(5))
 
 
+def test_claim_next_event_leaves_same_issue_queued_while_running(db: Database) -> None:
+    key = issue_key("octo/widget", 4)
+    db.record_event(
+        delivery_id="running",
+        event_type="issue_comment",
+        repo="octo/widget",
+        issue_key=key,
+        payload={"action": "created"},
+        state="running",
+    )
+    db.record_event(
+        delivery_id="queued",
+        event_type="issue_comment",
+        repo="octo/widget",
+        issue_key=key,
+        payload={"action": "created"},
+    )
+
+    assert db.claim_next_event() is None
+    assert db.get_event("queued").state == "queued"
+
+    db.mark_event("running", "done")
+    claimed = db.claim_next_event()
+    assert claimed is not None
+    assert claimed.delivery_id == "queued"
+    assert db.get_event("queued").state == "running"
+
+
+def test_claim_next_event_skips_blocked_issue_without_stalling_others(db: Database) -> None:
+    blocked = issue_key("octo/widget", 4)
+    ready = issue_key("octo/widget", 5)
+    db.record_event(
+        delivery_id="running",
+        event_type="issue_comment",
+        repo="octo/widget",
+        issue_key=blocked,
+        payload={"action": "created"},
+        state="running",
+    )
+    db.record_event(
+        delivery_id="blocked",
+        event_type="issue_comment",
+        repo="octo/widget",
+        issue_key=blocked,
+        payload={"action": "created"},
+    )
+    db.record_event(
+        delivery_id="ready",
+        event_type="issues",
+        repo="octo/widget",
+        issue_key=ready,
+        payload={"action": "opened"},
+    )
+
+    claimed = db.claim_next_event()
+    assert claimed is not None
+    assert claimed.delivery_id == "ready"
+    assert db.get_event("blocked").state == "queued"
+
+
 def test_requeue_event_can_be_restricted_by_source_state(db: Database) -> None:
     db.record_event(
         delivery_id="done-event",