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