[obvious-fleet] Serialize MemoryState event sequence assignment

backend/server/common/backends/state_memory.py

@@ -49,6 +49,7 @@ def __init__(self) -> None:
         self._spend: dict[str, float] = {}
         self._open_by_owner: dict[str, list[str]] = {}
         self._claim_lock = threading.Lock()
+        self._event_lock = threading.Lock()
 
     def get_capture(self, capture_id: str) -> CaptureRecord | None:
         rec = self._captures.get(capture_id)
@@ -137,18 +138,21 @@ def append_event(
         run_id: str | None = None,
         stage: StageName | None = None,
     ) -> EventRecord:
-        prior = self._events.get(capture_id, [])
-        seq = (prior[-1].seq + 1) if prior else 1
-        rec = EventRecord(
-            capture_id=capture_id,
-            seq=seq,
-            type=type,
-            at=datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ"),
-            data=dict(data),
-            run_id=run_id,
-            stage=stage,
-        )
-        self._events.setdefault(capture_id, []).append(rec)
+        with self._event_lock:
+            # Read the tail, assign the seq and append as one atomic step, so two
+            # producers never observe the same last event.
+            prior = self._events.get(capture_id, [])
+            seq = (prior[-1].seq + 1) if prior else 1
+            rec = EventRecord(
+                capture_id=capture_id,
+                seq=seq,
+                type=type,
+                at=datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ"),
+                data=dict(data),
+                run_id=run_id,
+                stage=stage,
+            )
+            self._events.setdefault(capture_id, []).append(rec)
         return rec.model_copy(deep=True)
 
     def events_after(self, capture_id: str, seq: int, limit: int = 100) -> list[EventRecord]:

backend/server/common/tests/test_state_memory_events.py

@@ -0,0 +1,78 @@
+"""Concurrent appends on MemoryState keep a gap-free sequence."""
+
+from __future__ import annotations
+
+import threading
+
+import pytest
+
+from server.common.backends import state_memory
+from server.common.backends.state_memory import MemoryState
+
+_APPENDS_PER_PRODUCER = 40
+_ROUNDS = 5
+
+
+def _emit(
+    state: MemoryState,
+    capture_id: str,
+    seqs: list[int],
+    errors: list[Exception],
+    guard: threading.Lock,
+) -> None:
+    try:
+        for _ in range(_APPENDS_PER_PRODUCER):
+            rec = state.append_event(capture_id, "hint", {"code": "frames_blurry", "message": "m"})
+            rec.data["code"] = "mutated"
+            with guard:
+                seqs.append(rec.seq)
+    except Exception as exc:
+        with guard:
+            errors.append(exc)
+
+
+def _run_round(
+    monkeypatch: pytest.MonkeyPatch,
+    real_event_record: type,
+    capture_id: str,
+) -> tuple[list[int], list[Exception], MemoryState]:
+    state = MemoryState()
+    barrier = threading.Barrier(2)
+
+    def gated_event_record(**kwargs: object):
+        # Pre-fix, both producers reach event construction unlocked, meet at the
+        # barrier and read the same tail. With the event lock held, the gate is
+        # a no-op and the producers never meet.
+        lock = getattr(state, "_event_lock", None)
+        if lock is None or not lock.locked():
+            barrier.wait(timeout=30)
+        return real_event_record(**kwargs)
+
+    monkeypatch.setattr(state_memory, "EventRecord", gated_event_record)
+    seqs: list[int] = []
+    errors: list[Exception] = []
+    guard = threading.Lock()
+    threads = [threading.Thread(target=_emit, args=(state, capture_id, seqs, errors, guard)) for _ in range(2)]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join()
+    monkeypatch.undo()
+    return seqs, errors, state
+
+
+def test_append_event_two_producers_stay_contiguous_unique_and_ordered(monkeypatch: pytest.MonkeyPatch) -> None:
+    capture_id = "cap_seq2"
+    total = 2 * _APPENDS_PER_PRODUCER
+    real_event_record = state_memory.EventRecord
+    for _round in range(_ROUNDS):
+        seqs, errors, state = _run_round(monkeypatch, real_event_record, capture_id)
+        assert errors == []
+        ordered = sorted(seqs)
+        assert ordered == list(range(1, total + 1))
+        assert len(set(ordered)) == total
+        assert [rec.seq for rec in state.events_after(capture_id, 0, limit=total + 10)] == list(range(1, total + 1))
+        assert [rec.seq for rec in state.events_after(capture_id, 40)] == list(range(41, total + 1))
+        assert [rec.seq for rec in state.events_after(capture_id, 40, limit=5)] == [41, 42, 43, 44, 45]
+        stored = state.events_after(capture_id, 0, limit=1)[0]
+        assert stored.data["code"] == "frames_blurry"