Prevent overlapping autonomous release runs #336
12
README.md
12
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).
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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])
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user