From ef3c1c2aeefc59a6ee548f82bc234547712a77b4 Mon Sep 17 00:00:00 2001 From: lau0708 <1581784147@qq.com> Date: Tue, 22 Sep 2026 11:38:43 +0800 Subject: [PATCH] fix(runtime): share one change position across an atomic thread operation (#5663) * fix(runtime): share one change position across an atomic thread operation MemoryRunStore.create_thread_operation_atomic marked each row it touched separately, so one interrupt-and-replace consumed two positions in the (change_seq, run_id) cursor that list_changed consumers page with. The SQL store allocates a single position for the whole set, and runtime/AGENTS.md documents that as the contract for an atomic thread operation. The memory store is the backend an install runs whenever no durable database is configured, so a cursor reader could observe the interrupted row and its replacement under two positions instead of one, ordered by run_id. Allocate one position after the raise-only scan -- so a rejected operation still consumes nothing -- and reuse it for the claimed rows and the new row. * test(runtime): pin that a rejected thread operation consumes no position create_thread_operation_atomic allocates its change position only after the raise-only candidate scan, so a ConflictError consumes nothing -- the memory store's counterpart of the SQL store rolling back and leaving its clock untouched. Nothing asserted that ordering. A consumed-but-unused position leaves no trace in the rows themselves, so a test comparing list_changed output passes even with the allocation hoisted above the scan. Assert instead on the position the next accepted operation lands on: that surfaces a hoist as a gap. --- .../deerflow/runtime/runs/store/memory.py | 22 ++++- .../tests/test_multi_worker_run_ownership.py | 91 +++++++++++++++++++ 2 files changed, 109 insertions(+), 4 deletions(-) diff --git a/backend/packages/harness/deerflow/runtime/runs/store/memory.py b/backend/packages/harness/deerflow/runtime/runs/store/memory.py index cd67bdff6..cbbd42900 100644 --- a/backend/packages/harness/deerflow/runtime/runs/store/memory.py +++ b/backend/packages/harness/deerflow/runtime/runs/store/memory.py @@ -40,9 +40,12 @@ class MemoryRunStore(RunStore): if not bucket: self._runs_by_thread.pop(thread_id, None) - def _mark_changed(self, run: dict[str, Any]) -> None: + def _next_change_seq(self) -> int: self._change_seq += 1 - run["change_seq"] = self._change_seq + return self._change_seq + + def _mark_changed(self, run: dict[str, Any]) -> None: + run["change_seq"] = self._next_change_seq() async def put( self, @@ -502,6 +505,7 @@ class MemoryRunStore(RunStore): # interrupted state on raise, diverging from SQL where a raise rolls # the whole transaction back. claimed = [] + change_seq: int | None = None if multitask_strategy in ("interrupt", "rollback"): candidates: list[dict[str, Any]] = [] for r in self._runs.values(): @@ -533,12 +537,20 @@ class MemoryRunStore(RunStore): if r.get("operation_kind", "run") != "run" and not lease_expired: raise ConflictError(f"Thread {thread_id} has an active checkpoint write") candidates.append(r) + # One position covers this atomic set of changes, with ``run_id`` + # ordering ties. The SQL store allocates the same single value for + # the set from its singleton clock (``runtime/AGENTS.md``); marking + # each row separately split one interrupt-and-replace into two + # positions in the ``(change_seq, run_id)`` cursor consumers page + # with. Allocate after the raise-only scan above so a rejected + # operation does not consume a position. + change_seq = self._next_change_seq() for r in candidates: r["status"] = "interrupted" r["error"] = "Cancelled by newer run" r["owner_worker_id"] = owner_worker_id r["updated_at"] = now - self._mark_changed(r) + r["change_seq"] = change_seq claimed.append(r) new_row = { @@ -561,7 +573,9 @@ class MemoryRunStore(RunStore): "created_at": created_at or now, "updated_at": now, } + if change_seq is None: + change_seq = self._next_change_seq() + new_row["change_seq"] = change_seq self._runs[run_id] = new_row - self._mark_changed(new_row) self._index_run(run_id, thread_id) return new_row, claimed diff --git a/backend/tests/test_multi_worker_run_ownership.py b/backend/tests/test_multi_worker_run_ownership.py index 09b25b2e4..7b4af73a5 100644 --- a/backend/tests/test_multi_worker_run_ownership.py +++ b/backend/tests/test_multi_worker_run_ownership.py @@ -1296,6 +1296,97 @@ async def test_create_thread_operation_atomic_interrupt_claims_and_creates(): assert old_row["status"] == "interrupted" +@pytest.mark.anyio +async def test_create_thread_operation_atomic_uses_one_change_position(): + """One atomic thread operation advances the change cursor once. + + ``runtime/AGENTS.md`` documents that an atomic thread operation uses one + position for its interrupted rows and its new row, with ``run_id`` + ordering ties, and ``test_run_evidence_reader`` pins that for the SQL + store. The memory store must agree: it is the backend an install runs + whenever no durable database is configured. Advancing once per row splits + a single interrupt-and-replace into two positions in the + ``(change_seq, run_id)`` cursor extensions page with. + """ + store = MemoryRunStore() + config = _lease_config() + expired_lease = (datetime.now(UTC) - timedelta(seconds=60)).isoformat() + + await store.create_thread_operation_atomic( + run_id="run-old", + thread_id="thread-1", + owner_worker_id="w1", + lease_expires_at=expired_lease, + multitask_strategy="reject", + grace_seconds=config.grace_seconds, + ) + + new_row, claimed = await store.create_thread_operation_atomic( + run_id="run-new", + thread_id="thread-1", + owner_worker_id="w2", + lease_expires_at=(datetime.now(UTC) + timedelta(seconds=30)).isoformat(), + multitask_strategy="interrupt", + grace_seconds=config.grace_seconds, + ) + + assert [row["run_id"] for row in claimed] == ["run-old"] + assert claimed[0]["change_seq"] == new_row["change_seq"] + + # The shared position is what a paging consumer observes: the interrupted + # row and its replacement surface under one cursor value, ordered by + # ``run_id``, exactly as the SQL store reports them. + changed = await store.list_changed(after_change_seq=-1, after_run_id="") + positions = {row["run_id"]: row["change_seq"] for row in changed} + assert positions["run-old"] == positions["run-new"] + assert [row["run_id"] for row in changed if row["change_seq"] == positions["run-new"]] == ["run-new", "run-old"] + + +@pytest.mark.anyio +async def test_create_thread_operation_atomic_rejection_consumes_no_change_position(): + """A ``ConflictError``-rejected operation advances no change position. + + ``create_thread_operation_atomic`` allocates its position only after the + raise-only candidate scan, so a rejected operation consumes nothing — the + memory-store counterpart of the SQL store's rollback leaving the clock + untouched. A consumed-but-unused position leaves no trace in the rows + themselves, so the assertion is on the position the next accepted + operation lands on. Hoisting the allocation above the scan keeps every + other test in this file green and shows up here as a gap. + """ + store = MemoryRunStore() + config = _lease_config(grace_seconds=10) + + accepted, _ = await store.create_thread_operation_atomic( + run_id="valid-lease-run", + thread_id="thread-1", + owner_worker_id="other-worker", + lease_expires_at=(datetime.now(UTC) + timedelta(seconds=30)).isoformat(), + multitask_strategy="reject", + grace_seconds=config.grace_seconds, + ) + + with pytest.raises(ConflictError, match="another worker"): + await store.create_thread_operation_atomic( + run_id="run-new", + thread_id="thread-1", + owner_worker_id="w2", + lease_expires_at=(datetime.now(UTC) + timedelta(seconds=30)).isoformat(), + multitask_strategy="interrupt", + grace_seconds=config.grace_seconds, + ) + + next_row, _ = await store.create_thread_operation_atomic( + run_id="run-after", + thread_id="thread-2", + owner_worker_id="w2", + lease_expires_at=(datetime.now(UTC) + timedelta(seconds=30)).isoformat(), + multitask_strategy="reject", + grace_seconds=config.grace_seconds, + ) + assert next_row["change_seq"] == accepted["change_seq"] + 1 + + @pytest.mark.anyio async def test_create_thread_operation_atomic_interrupt_rejects_other_worker_valid_lease(): """Interrupt must raise ConflictError when a valid-lease run is owned by another worker.