fix(v2): preserve execution cancellation and retry safety
This commit is contained in:
@@ -164,7 +164,18 @@ async def retry(db: AsyncSession, execution_id: str) -> Execution | None:
|
||||
execution.heartbeat_at = None
|
||||
execution.reason_code = None
|
||||
execution.operator_message = None
|
||||
await db.commit()
|
||||
try:
|
||||
await db.commit()
|
||||
except IntegrityError as error:
|
||||
await db.rollback()
|
||||
active_id = await db.scalar(
|
||||
select(Execution.id).where(
|
||||
Execution.job_id == execution.job_id, Execution.state.in_(ACTIVE_STATES)
|
||||
)
|
||||
)
|
||||
raise EnqueueError(
|
||||
"execution_active", "Job already has an active execution.", active_id
|
||||
) from error
|
||||
await db.refresh(execution)
|
||||
return execution
|
||||
|
||||
@@ -199,10 +210,22 @@ def public_event(execution: Execution) -> dict[str, object]:
|
||||
|
||||
async def recover_stale(db: AsyncSession) -> int:
|
||||
now = datetime.now(UTC)
|
||||
result = await db.execute(
|
||||
cancelled = await db.execute(
|
||||
update(Execution)
|
||||
.where(Execution.state == "cancelling", Execution.lease_expires_at < now)
|
||||
.values(
|
||||
state="cancelled",
|
||||
completed_at=now,
|
||||
lease_owner=None,
|
||||
lease_expires_at=None,
|
||||
heartbeat_at=None,
|
||||
reason_code="cancellation_requested",
|
||||
)
|
||||
)
|
||||
recovered = await db.execute(
|
||||
update(Execution)
|
||||
.where(
|
||||
Execution.state.in_({"preparing", "running", "verifying", "cancelling"}),
|
||||
Execution.state.in_({"preparing", "running", "verifying"}),
|
||||
Execution.lease_expires_at < now,
|
||||
)
|
||||
.values(
|
||||
@@ -214,4 +237,4 @@ async def recover_stale(db: AsyncSession) -> int:
|
||||
)
|
||||
)
|
||||
await db.commit()
|
||||
return getattr(result, "rowcount", 0) or 0
|
||||
return (getattr(cancelled, "rowcount", 0) or 0) + (getattr(recovered, "rowcount", 0) or 0)
|
||||
|
||||
@@ -45,7 +45,8 @@ class Worker:
|
||||
execution = await claim(db, execution_id, self.owner)
|
||||
if execution is None:
|
||||
return False
|
||||
# M6 supplies source/repository work; this slice proves durable ownership.
|
||||
# Reload after the claim: a control request can race the lease acquisition.
|
||||
await db.refresh(execution)
|
||||
if execution.state == "cancelling":
|
||||
await complete_cancellation(db, execution.id, self.owner)
|
||||
return True
|
||||
|
||||
Reference in New Issue
Block a user