From 5c742e6aec112d474041406e6ef70ad744676389 Mon Sep 17 00:00:00 2001 From: Renn F Date: Sun, 3 May 2026 09:38:03 +0200 Subject: [PATCH] fix(git): serialize concurrent merges to the same parent branch Two PMs completing different subtasks of the same parent could race on gh API merge calls. GitHub returns 409 to one but local DB write ordering wasn't guaranteed. Take row-level lock on parent task before merge; retry once on 409. --- roboco/services/git.py | 51 ++++ tests/unit/services/test_git.py | 3 + .../services/test_pr_merge_concurrency.py | 279 ++++++++++++++++++ 3 files changed, 333 insertions(+) create mode 100644 tests/unit/services/test_pr_merge_concurrency.py diff --git a/roboco/services/git.py b/roboco/services/git.py index 667770d0..a64a8665 100644 --- a/roboco/services/git.py +++ b/roboco/services/git.py @@ -79,6 +79,10 @@ _REV_LIST_PARTS = 2 # GitHub REST API status codes _GH_UNPROCESSABLE = 422 +# 409 means the PR can't be merged in its current state — typically because +# a concurrent sibling-subtask merge updated the target branch and our local +# refs are stale. `pr_merge` re-syncs and retries exactly once on this code. +_HTTP_CONFLICT = 409 class GitService(BaseService): @@ -1710,11 +1714,44 @@ class GitService(BaseService): await self._record_pr_atomically(UUID(str(task.id)), pr_number, pr_url) return {"pr_number": pr_number, "pr_url": pr_url, "is_root_pr": is_root_pr} + async def _lock_parent_task_for_merge(self, parent_task_id: UUID | None) -> None: + """SELECT FOR UPDATE on the parent task row, if any. + + Two PMs completing different subtasks of the same parent could + race on the gh API merge call. Holding a row-level lock on the + parent task serializes those merges at the DB layer — the second + PM's transaction blocks until the first commits, by which time + the first PR is already merged. The lock is auto-released when + the surrounding transaction commits or rolls back. + + Root tasks (no parent) skip the lock — there's nothing to + contend on at parent level, and master-bound merges are + serialized by GitHub's PR-state machine alone. + """ + if parent_task_id is None: + return + from sqlalchemy import select + + from roboco.db.tables import TaskTable as _TaskTable + + await self.session.execute( + select(_TaskTable) + .where(_TaskTable.id == parent_task_id) + .with_for_update(of=_TaskTable) + ) + async def pr_merge(self, pr_number: int, *, target: str) -> dict[str, Any]: """Merge PR `pr_number` into `target`. Returns: ``{"merge_commit_sha": str | None}``. Looks up the task/project that owns the PR to resolve workspace + token. + + Concurrency: takes a row-level lock on the parent task before + invoking the GitHub merge API so that two PMs completing + sibling subtasks of the same parent are serialized. On a 409 + merge conflict (typical race symptom — GitHub serializes via + PR-state churn) the local target branch is re-pulled and the + merge is retried exactly once before giving up with `GitError`. """ from sqlalchemy import select @@ -1738,7 +1775,21 @@ class GitService(BaseService): git_token = await self._get_project_token_or_raise(project.slug) owner, repo = self._parse_github_remote(workspace) + # Serialize merges into the same parent branch — see helper docstring. + parent_id = UUID(str(task.parent_task_id)) if task.parent_task_id else None + await self._lock_parent_task_for_merge(parent_id) + resp = await self._call_merge_api(owner, repo, pr_number, git_token, "squash") + if resp.status_code == _HTTP_CONFLICT: + # Race symptom — another PM merged a sibling subtask first + # and our local target ref is stale. Refresh and retry once; + # if the second attempt also conflicts, it's a real conflict + # (not just a race) and the choreographer surfaces it as + # `invalid_state` so the PM can resolve it manually. + await self._sync_target_branch(workspace, target, git_token) + resp = await self._call_merge_api( + owner, repo, pr_number, git_token, "squash" + ) if not resp.is_success: raise GitError( f"GitHub API refused PR merge ({resp.status_code}): {resp.text[:200]}", diff --git a/tests/unit/services/test_git.py b/tests/unit/services/test_git.py index 0c10a7eb..dbbc443b 100644 --- a/tests/unit/services/test_git.py +++ b/tests/unit/services/test_git.py @@ -224,6 +224,9 @@ async def test_pr_merge_returns_merge_commit_dict() -> None: project_id = uuid4() fake_task = MagicMock( project_id=project_id, + # Root task — no parent to lock; concurrency tests cover the + # parent-lock + retry-on-409 paths separately. + parent_task_id=None, assigned_to=uuid4(), work_session_id=None, ) diff --git a/tests/unit/services/test_pr_merge_concurrency.py b/tests/unit/services/test_pr_merge_concurrency.py new file mode 100644 index 00000000..39e2f69e --- /dev/null +++ b/tests/unit/services/test_pr_merge_concurrency.py @@ -0,0 +1,279 @@ +"""Tests for `pr_merge` concurrency hardening. + +Two PMs completing different subtasks of the same parent could race on +the gh API merge call. The fix is to: + +1. Take a row-level lock on the parent task before invoking the merge. +2. Retry once if GitHub returns 409 (merge conflict from racing merges), + re-pulling the local target branch in between to refresh refs. + +These are unit tests — concurrency is exercised via mocks (status code +sequence + assertions on `with_for_update` use), not real DB transactions. +The lock-acquisition path is asserted by checking that the parent-task +SELECT statement passed to `session.execute` carries `FOR UPDATE` semantics. +""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch +from uuid import uuid4 + +import pytest +from roboco.exceptions import GitError +from roboco.services.git import GitService + +# Module-level constants kept local so the assertions stay readable and +# ruff's PLR2004 magic-value rule has nothing to complain about. The +# threshold mirrors httpx's `Response.is_success` rule (status < 400). +# The expected-call counters document the retry contract: at most two +# merge attempts, two `_sync_target_branch` calls (refresh + post-success), +# two SELECTs (PR lookup + parent FOR UPDATE). +_HTTP_OK_THRESHOLD = 400 +_EXPECTED_MERGE_ATTEMPTS = 2 +_EXPECTED_SYNC_CALLS = 2 +_EXPECTED_SELECT_CALLS = 2 + + +def _make_session( + pr_lookup_task: object, + parent_task: object | None, +) -> MagicMock: + """Build a session whose first execute returns the PR-owning task, + and second execute (the SELECT FOR UPDATE on the parent) returns + `parent_task`. + """ + session = MagicMock() + pr_result = MagicMock() + pr_result.scalar_one_or_none.return_value = pr_lookup_task + parent_result = MagicMock() + parent_result.scalar_one_or_none.return_value = parent_task + + # First execute() = PR -> task lookup; second = parent lock; further + # executes (none expected here) reuse parent_result. + session.execute = AsyncMock(side_effect=[pr_result, parent_result]) + session.commit = AsyncMock() + session.rollback = AsyncMock() + session.flush = AsyncMock() + return session + + +def _patch_project_service(project: object | None) -> Any: + fake_service = MagicMock() + fake_service.get = AsyncMock(return_value=project) + fake_service.get_by_slug = AsyncMock(return_value=project) + return patch("roboco.services.git.get_project_service", return_value=fake_service) + + +def _bind(svc: GitService, name: str, value: object) -> None: + object.__setattr__(svc, name, value) + + +def _fake_response(status_code: int) -> MagicMock: + resp = MagicMock() + resp.is_success = status_code < _HTTP_OK_THRESHOLD + resp.status_code = status_code + resp.text = f"status {status_code}" + return resp + + +# --------------------------------------------------------------------------- +# Scenario: GitHub returns 409 once, then 200. Retry must succeed and call +# `_sync_target_branch` between attempts to refresh local state. +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_pr_merge_retries_once_on_409_conflict() -> None: + project_id = uuid4() + parent_id = uuid4() + fake_task = MagicMock( + id=uuid4(), + project_id=project_id, + parent_task_id=parent_id, + assigned_to=uuid4(), + work_session_id=None, + ) + fake_parent = MagicMock(id=parent_id) + fake_project = MagicMock(slug="roboco") + + svc = GitService(_make_session(fake_task, fake_parent)) + _bind(svc, "get_workspace", AsyncMock(return_value=Path("/tmp/ws"))) + _bind(svc, "_get_project_token_or_raise", AsyncMock(return_value="tok")) + _bind(svc, "_parse_github_remote", MagicMock(return_value=("acme", "repo"))) + + call_seq = AsyncMock(side_effect=[_fake_response(409), _fake_response(200)]) + _bind(svc, "_call_merge_api", call_seq) + _bind(svc, "_delete_pr_branch_best_effort", AsyncMock()) + sync_branch = AsyncMock(return_value="merged-sha") + _bind(svc, "_sync_target_branch", sync_branch) + + with _patch_project_service(fake_project): + out = await svc.pr_merge(11, target="feature/backend/parent") + + assert out == {"merge_commit_sha": "merged-sha"} + # _call_merge_api invoked twice — once 409, once 200. + assert call_seq.await_count == _EXPECTED_MERGE_ATTEMPTS + # _sync_target_branch invoked twice — once for the 409 refresh, once + # for the post-success refresh that returns the merge SHA. + assert sync_branch.await_count == _EXPECTED_SYNC_CALLS + + +# --------------------------------------------------------------------------- +# Scenario: GitHub returns 409 twice. We don't loop indefinitely — bubble +# up GitError so the choreographer can return invalid_state. +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_pr_merge_raises_after_second_409() -> None: + project_id = uuid4() + parent_id = uuid4() + fake_task = MagicMock( + id=uuid4(), + project_id=project_id, + parent_task_id=parent_id, + assigned_to=uuid4(), + work_session_id=None, + ) + fake_parent = MagicMock(id=parent_id) + fake_project = MagicMock(slug="roboco") + + svc = GitService(_make_session(fake_task, fake_parent)) + _bind(svc, "get_workspace", AsyncMock(return_value=Path("/tmp/ws"))) + _bind(svc, "_get_project_token_or_raise", AsyncMock(return_value="tok")) + _bind(svc, "_parse_github_remote", MagicMock(return_value=("acme", "repo"))) + + call_seq = AsyncMock(side_effect=[_fake_response(409), _fake_response(409)]) + _bind(svc, "_call_merge_api", call_seq) + _bind(svc, "_delete_pr_branch_best_effort", AsyncMock()) + _bind(svc, "_sync_target_branch", AsyncMock(return_value="abc")) + + with _patch_project_service(fake_project), pytest.raises(GitError) as exc_info: + await svc.pr_merge(11, target="feature/backend/parent") + + assert "409" in str(exc_info.value) + # No infinite retries — exactly two attempts. + assert call_seq.await_count == _EXPECTED_MERGE_ATTEMPTS + + +# --------------------------------------------------------------------------- +# Scenario: Non-409 GitHub error is NOT retried; raises immediately. +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_pr_merge_does_not_retry_on_non_409_error() -> None: + project_id = uuid4() + parent_id = uuid4() + fake_task = MagicMock( + id=uuid4(), + project_id=project_id, + parent_task_id=parent_id, + assigned_to=uuid4(), + work_session_id=None, + ) + fake_parent = MagicMock(id=parent_id) + fake_project = MagicMock(slug="roboco") + + svc = GitService(_make_session(fake_task, fake_parent)) + _bind(svc, "get_workspace", AsyncMock(return_value=Path("/tmp/ws"))) + _bind(svc, "_get_project_token_or_raise", AsyncMock(return_value="tok")) + _bind(svc, "_parse_github_remote", MagicMock(return_value=("acme", "repo"))) + + call_seq = AsyncMock(side_effect=[_fake_response(422)]) + _bind(svc, "_call_merge_api", call_seq) + _bind(svc, "_delete_pr_branch_best_effort", AsyncMock()) + _bind(svc, "_sync_target_branch", AsyncMock(return_value="abc")) + + with _patch_project_service(fake_project), pytest.raises(GitError): + await svc.pr_merge(11, target="feature/backend/parent") + + # Only one merge attempt — non-409 error path skips retry. + assert call_seq.await_count == 1 + + +# --------------------------------------------------------------------------- +# Scenario: Locking the parent task — assert with_for_update is part of +# the SELECT statement issued for the parent task lookup. +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_pr_merge_locks_parent_task_with_for_update() -> None: + project_id = uuid4() + parent_id = uuid4() + fake_task = MagicMock( + id=uuid4(), + project_id=project_id, + parent_task_id=parent_id, + assigned_to=uuid4(), + work_session_id=None, + ) + fake_parent = MagicMock(id=parent_id) + fake_project = MagicMock(slug="roboco") + + session = _make_session(fake_task, fake_parent) + svc = GitService(session) + _bind(svc, "get_workspace", AsyncMock(return_value=Path("/tmp/ws"))) + _bind(svc, "_get_project_token_or_raise", AsyncMock(return_value="tok")) + _bind(svc, "_parse_github_remote", MagicMock(return_value=("acme", "repo"))) + _bind(svc, "_call_merge_api", AsyncMock(return_value=_fake_response(200))) + _bind(svc, "_delete_pr_branch_best_effort", AsyncMock()) + _bind(svc, "_sync_target_branch", AsyncMock(return_value="abc")) + + with _patch_project_service(fake_project): + await svc.pr_merge(11, target="feature/backend/parent") + + # Two SELECTs: PR lookup (no lock) + parent lock (FOR UPDATE). + assert session.execute.await_count == _EXPECTED_SELECT_CALLS + parent_call = session.execute.await_args_list[1] + parent_stmt = parent_call.args[0] + # SQLAlchemy's compiled SELECT with for_update has `_for_update_arg` set. + # Read via getattr so mypy doesn't trip on the protected attr name and + # we don't need a `# type: ignore` escape hatch. + assert getattr(parent_stmt, "_for_update_arg", None) is not None + + +# --------------------------------------------------------------------------- +# Scenario: Root-PR merge (parent_task_id is None) — no parent lock attempt, +# but merge still proceeds. Tests the "no parent" branch of the lock helper. +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_pr_merge_skips_parent_lock_for_root_task() -> None: + project_id = uuid4() + fake_task = MagicMock( + id=uuid4(), + project_id=project_id, + parent_task_id=None, # root task — merging into master + assigned_to=uuid4(), + work_session_id=None, + ) + fake_project = MagicMock(slug="roboco") + + # No second execute() — root task has no parent to lock. + session = MagicMock() + pr_result = MagicMock() + pr_result.scalar_one_or_none.return_value = fake_task + session.execute = AsyncMock(return_value=pr_result) + session.commit = AsyncMock() + session.rollback = AsyncMock() + session.flush = AsyncMock() + + svc = GitService(session) + _bind(svc, "get_workspace", AsyncMock(return_value=Path("/tmp/ws"))) + _bind(svc, "_get_project_token_or_raise", AsyncMock(return_value="tok")) + _bind(svc, "_parse_github_remote", MagicMock(return_value=("acme", "repo"))) + _bind(svc, "_call_merge_api", AsyncMock(return_value=_fake_response(200))) + _bind(svc, "_delete_pr_branch_best_effort", AsyncMock()) + _bind(svc, "_sync_target_branch", AsyncMock(return_value="abc")) + + with _patch_project_service(fake_project): + out = await svc.pr_merge(11, target="master") + + assert out == {"merge_commit_sha": "abc"} + # Only the PR-lookup SELECT runs — no second SELECT for parent lock. + assert session.execute.await_count == 1