feat(ops-runs): expire claims that keep losing their lease

Lapsed claims were reopened with no attempt ceiling, so a run whose worker
lost its lease (including a 409 before dispatch) could cycle forever.
Activity Core now reopens a lapsed claim below OPS_RUN_MAX_ATTEMPTS and sets
terminal 'expired' at the ceiling. The owner is kept for audit, and
/ops-runs/expire-leases reports reopened and expired counts. Answers the
REINAH-WP-0003-T05 open point on thread 27cd4c47 (ACTIVITY-WP-ADHOC-2026-10-19).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 244572@bnt-lap001
Assistant-Session: 3b235e1a-ae45-4d0f-ae70-17a9bd78ece8
This commit is contained in:
tegwick 2026-10-19 15:30:29 +02:00
parent 3b25aea7b0
commit 42347da012
6 changed files with 192 additions and 14 deletions

View file

@ -219,17 +219,47 @@ async def create_ops_run_from_spec(
async def reopen_stale_claims(session: AsyncSession) -> int:
"""Return claimed rows with expired leases to open."""
"""Return claimed rows with expired leases to open; expire exhausted ones.
A claim whose lease lapsed without a terminal close (crash, lost lease,
or a worker that refused dispatch after a 409) is Activity Core's to
recover: the worker that lost the lease must not close it. Below the
attempt ceiling the row returns to ``open``; at or above
``OPS_RUN_MAX_ATTEMPTS`` it becomes terminal ``expired`` so a row that
keeps losing its lease cannot cycle forever. Returns the number of rows
reopened (see :func:`expire_stale_claims` for both counts).
"""
counts = await expire_stale_claims(session)
return counts["reopened"]
async def expire_stale_claims(session: AsyncSession) -> dict[str, int]:
"""Reopen or terminally expire claimed rows whose lease has lapsed."""
now = _utcnow()
stmt = (
stale = and_(
OpsRun.state == "claimed",
OpsRun.lease_until.is_not(None),
OpsRun.lease_until <= now,
)
ceiling = max_attempts()
expired = await session.execute(
update(OpsRun)
.where(
and_(
OpsRun.state == "claimed",
OpsRun.lease_until.is_not(None),
OpsRun.lease_until <= now,
)
.where(stale, OpsRun.attempt >= ceiling)
.values(
state="expired",
lease_until=None,
result={
"error": (
f"lease expired without terminal close after {ceiling} "
"attempt(s); expired by activity-core"
)
},
updated_at=now,
)
)
reopened = await session.execute(
update(OpsRun)
.where(stale, OpsRun.attempt < ceiling)
.values(
state="open",
claim_owner=None,
@ -237,8 +267,10 @@ async def reopen_stale_claims(session: AsyncSession) -> int:
updated_at=now,
)
)
result = await session.execute(stmt)
return int(result.rowcount or 0)
return {
"reopened": int(reopened.rowcount or 0),
"expired": int(expired.rowcount or 0),
}
async def claim_ops_runs(

View file

@ -30,7 +30,7 @@ from activity_core.ops_run_queue import (
list_ops_runs,
ops_run_counts,
ops_run_to_dict,
reopen_stale_claims,
expire_stale_claims,
)
router = APIRouter(prefix="/ops-runs", tags=["ops-runs"])
@ -479,5 +479,5 @@ async def post_expire_leases(
Session = _db()
async with Session() as session:
async with session.begin():
n = await reopen_stale_claims(session)
return {"reopened": n}
counts = await expire_stale_claims(session)
return counts