From 4969cadd82550adb5eafd30dec96634e3d989962 Mon Sep 17 00:00:00 2001 From: timmy Date: Sat, 8 Aug 2026 19:54:20 +0000 Subject: [PATCH] feat: serialize autonomous release runs (#335) --- README.md | 12 +++++- docs/release-engine-spec.md | 19 +++++++++ src/release_engine.py | 76 ++++++++++++++++++++++++++++++++++-- tests/test_release_engine.py | 74 +++++++++++++++++++++++++++++++++++ 4 files changed, 176 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index 9c1e730..81446b3 100644 --- a/README.md +++ b/README.md @@ -310,6 +310,14 @@ without advancing the release. The equivalent environment variables are Durable state defaults to `.release-engine/state.json`. Successful checkpoints record the repository, issue branch, and exact commit SHA. A restart resumes at tests, push, or PR creation only after validating that identity and a clean -worktree; a pushed checkpoint also verifies the remote branch SHA. Full -behavior and safety gates are documented in +worktree; a pushed checkpoint also verifies the remote branch SHA. Invocations +are serialized by an OS-backed, non-blocking lease at +`.release-engine/run.lock` under the canonical repository path. If an hourly +run overlaps a manual or slow prior run, the contender exits with `release run +already active` and owner diagnostics before Gitea discovery or any Git/agent +work. The kernel releases ownership when the process exits, so never delete a +lock file to recover from a crash; stale file metadata cannot retain the lease. +Checkpoint replacement uses writer-unique, flushed temporary files and an +atomic rename, preventing concurrent writers from colliding or exposing partial +JSON. Full behavior and safety gates are documented in [`docs/release-engine-spec.md`](docs/release-engine-spec.md). diff --git a/docs/release-engine-spec.md b/docs/release-engine-spec.md index 3015f30..9d86d50 100644 --- a/docs/release-engine-spec.md +++ b/docs/release-engine-spec.md @@ -44,6 +44,19 @@ repeating or advancing delivery. 4. Otherwise sort by priority label (`P0`, `P1`, `P2`, unlabeled) then issue number. 5. Select exactly one issue per invocation. +## Single-active-run lease + +- Before issue discovery, every invocation acquires a non-blocking advisory lock + at `.release-engine/run.lock` under the canonical repository path. +- The lease is held through the final checkpoint and pull-request request. A + contended invocation fails immediately with `release run already active` + before contacting Gitea or running Git, agent, or test commands. +- The lock file records bounded owner PID and acquisition-time diagnostics. + These are informational only: kernel lock ownership is authoritative, so a + killed or crashed process releases the lease automatically and stale metadata + cannot block the next run. +- Different canonical repository paths use independent leases. + ## Claiming - PATCH the issue with the configured assignee. @@ -78,6 +91,10 @@ repeating or advancing delivery. - `--dry-run` performs discovery and planning only: no claim, git mutation, agent command, push, or PR. - State is written atomically after each successful transition. +- Each checkpoint write uses a writer-unique temporary file, flushes file data, + atomically replaces the checkpoint, and flushes its parent directory. A + failed or concurrent writer therefore cannot collide on a shared temp path or + expose partial JSON. - Resumable state carries repository and commit identity; successful stages are skipped only after local (and, after push, remote) Git verification. - One invocation handles at most one issue. @@ -101,3 +118,5 @@ repeating or advancing delivery. 13. Dirty or identity-mismatched checkpoints stop before push or PR creation. 14. Parent credentials are absent from coding-agent, test, and Git subprocesses while explicit agent variables and issue context reach only the coding stage. 15. Reserved or malformed agent environment allowlist entries abort before claim. +16. An overlapping invocation fails before Gitea discovery or local command execution and reports bounded owner diagnostics. +17. Concurrent checkpoint-save stress completes without temp-file collisions or partial state. diff --git a/src/release_engine.py b/src/release_engine.py index 3f0b4f1..7d87325 100644 --- a/src/release_engine.py +++ b/src/release_engine.py @@ -2,12 +2,16 @@ from __future__ import annotations import argparse +import errno +import fcntl import json import os import re import signal import shlex import subprocess +import tempfile +import time from dataclasses import asdict, dataclass from pathlib import Path from typing import Any, Iterable, Protocol @@ -97,9 +101,71 @@ class StateStore: def save(self, state: RunState) -> None: self.path.parent.mkdir(parents=True, exist_ok=True) - tmp = self.path.with_name(self.path.name + ".tmp") - tmp.write_text(json.dumps(asdict(state), indent=2) + "\n") - tmp.replace(self.path) + temporary_path = None + try: + with tempfile.NamedTemporaryFile( + mode="w", + encoding="utf-8", + dir=self.path.parent, + prefix=self.path.name + ".", + suffix=".tmp", + delete=False, + ) as temporary: + temporary_path = Path(temporary.name) + temporary.write(json.dumps(asdict(state), indent=2) + "\n") + temporary.flush() + os.fsync(temporary.fileno()) + os.replace(temporary_path, self.path) + directory = os.open(self.path.parent, os.O_RDONLY | os.O_DIRECTORY) + try: + os.fsync(directory) + finally: + os.close(directory) + finally: + if temporary_path is not None: + temporary_path.unlink(missing_ok=True) + + +class RepositoryLease: + """Non-blocking, process-backed single-writer lease for one worktree.""" + + def __init__(self, repo_path: Path): + self.path = Path(repo_path).resolve() / ".release-engine" / "run.lock" + self._file = None + + def __enter__(self): + self.path.parent.mkdir(parents=True, exist_ok=True) + lock_file = self.path.open("a+", encoding="utf-8") + try: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + except OSError as exc: + if exc.errno in (errno.EACCES, errno.EAGAIN): + detail = "" + try: + lock_file.seek(0) + metadata = json.loads(lock_file.read(512)) + owner_pid = int(metadata["pid"]) + acquired_at = float(metadata["acquired_at"]) + detail = f" (owner pid={owner_pid}, acquired_at={acquired_at:.3f})" + except (KeyError, TypeError, ValueError, json.JSONDecodeError): + pass + lock_file.close() + raise RuntimeError(f"release run already active{detail}") from exc + lock_file.close() + raise + self._file = lock_file + lock_file.seek(0) + lock_file.truncate() + lock_file.write(json.dumps({"pid": os.getpid(), "acquired_at": time.time()}) + "\n") + lock_file.flush() + os.fsync(lock_file.fileno()) + return self + + def __exit__(self, exc_type, exc, traceback): + if self._file is not None: + fcntl.flock(self._file.fileno(), fcntl.LOCK_UN) + self._file.close() + self._file = None def _label_names(labels: Iterable[Any]) -> list[str]: @@ -267,6 +333,10 @@ class ReleaseEngine: self.store = StateStore(config.state_path) def execute(self, issue_number: int | None = None, dry_run: bool = False) -> dict[str, Any]: + with RepositoryLease(self.config.repo_path): + return self._execute_locked(issue_number=issue_number, dry_run=dry_run) + + def _execute_locked(self, issue_number: int | None = None, dry_run: bool = False) -> dict[str, Any]: deadlines = { "agent timeout seconds": self.config.agent_timeout_seconds, "test timeout seconds": self.config.test_timeout_seconds, diff --git a/tests/test_release_engine.py b/tests/test_release_engine.py index d3419f4..e875d47 100644 --- a/tests/test_release_engine.py +++ b/tests/test_release_engine.py @@ -461,6 +461,51 @@ def test_dry_run_does_not_claim_or_execute(tmp_path): assert runner.commands == [] +def test_overlapping_run_fails_before_gitea_or_git_side_effects(tmp_path): + entered = threading.Event() + release_owner = threading.Event() + + class BlockingClient(FakeClient): + def __init__(self): + super().__init__([]) + self.discovery_calls = 0 + + def list_open_issues(self, repo): + self.discovery_calls += 1 + if self.discovery_calls > 1: + raise AssertionError("contended run reached Gitea discovery") + entered.set() + release_owner.wait(timeout=2) + raise RuntimeError("stop lease owner") + + client = BlockingClient() + owner_runner = FakeRunner([]) + contender_runner = FakeRunner([]) + owner = ReleaseEngine(config(tmp_path), client, owner_runner) + contender = ReleaseEngine(config(tmp_path), client, contender_runner) + + def run_owner(): + with pytest.raises(RuntimeError, match="stop lease owner"): + owner.execute(dry_run=True) + + thread = threading.Thread(target=run_owner) + thread.start() + assert entered.wait(timeout=1) + try: + with pytest.raises(RuntimeError, match="release run already active") as error: + contender.execute(dry_run=True) + assert f"owner pid={os.getpid()}" in str(error.value) + assert "acquired_at=" in str(error.value) + finally: + release_owner.set() + thread.join(timeout=2) + + assert not thread.is_alive() + assert client.discovery_calls == 1 + assert owner_runner.commands == [] + assert contender_runner.commands == [] + + def test_existing_pr_is_reused_without_agent_execution(tmp_path): issue = Issue(14, "Engine MVP", "body", "open", [], "timmy") client = FakeClient([issue], existing_pr={"number": 9, "html_url": "https://forge/pr/9"}) @@ -482,6 +527,35 @@ def test_state_store_round_trip_is_atomic(tmp_path): assert not (tmp_path / "state.json.tmp").exists() +def test_concurrent_checkpoint_writers_never_collide_or_leave_partial_state(tmp_path): + store = StateStore(tmp_path / "state.json") + states = [ + RunState(issue=index, branch=f"timmy/{index}-engine", status="tests_passed") + for index in range(8) + ] + barrier = threading.Barrier(len(states)) + errors = [] + + def save_repeatedly(state): + barrier.wait() + for _ in range(100): + try: + store.save(state) + except Exception as exc: + errors.append(exc) + + threads = [threading.Thread(target=save_repeatedly, args=(state,)) for state in states] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=5) + + assert not any(thread.is_alive() for thread in threads) + assert errors == [] + assert store.load() in states + assert list(tmp_path.glob("state.json.*.tmp")) == [] + + def test_restart_from_agent_complete_skips_claim_and_coding(tmp_path): issue = Issue(331, "Resume releases", "body", "open", [], "timmy") client = FakeClient([issue])