fix: back off after ops claim errors
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a02b6f-7db1-7222-918b-e813a6bda38d
This commit is contained in:
parent
81d6222c6b
commit
db588010f7
3 changed files with 37 additions and 4 deletions
|
|
@ -42,6 +42,7 @@ logger = logging.getLogger("rein_aharness.claim_loop")
|
||||||
class ProcessResult:
|
class ProcessResult:
|
||||||
claimed: bool
|
claimed: bool
|
||||||
empty: bool = False
|
empty: bool = False
|
||||||
|
retry_full_interval: bool = False
|
||||||
run_id: str | None = None
|
run_id: str | None = None
|
||||||
approach: str | None = None
|
approach: str | None = None
|
||||||
ok: bool | None = None
|
ok: bool | None = None
|
||||||
|
|
@ -104,7 +105,11 @@ def process_one(
|
||||||
try:
|
try:
|
||||||
claimed = client.claim(limit=1)
|
claimed = client.claim(limit=1)
|
||||||
except OpsRunError as exc:
|
except OpsRunError as exc:
|
||||||
return ProcessResult(claimed=False, reason=f"claim error: {exc}")
|
return ProcessResult(
|
||||||
|
claimed=False,
|
||||||
|
retry_full_interval=True,
|
||||||
|
reason=f"claim error: {exc}",
|
||||||
|
)
|
||||||
|
|
||||||
if not claimed:
|
if not claimed:
|
||||||
return ProcessResult(claimed=False, empty=True, reason="queue empty")
|
return ProcessResult(claimed=False, empty=True, reason="queue empty")
|
||||||
|
|
@ -396,8 +401,13 @@ def run_claim_loop(
|
||||||
break
|
break
|
||||||
if max_iterations is not None and iterations >= max_iterations:
|
if max_iterations is not None and iterations >= max_iterations:
|
||||||
break
|
break
|
||||||
# Sleep full interval only when empty; short pause after work
|
# Empty queues and upstream errors both use the configured backoff.
|
||||||
sleep_for = interval_seconds if result.empty else min(2.0, interval_seconds)
|
# The short pause is only for a cycle that actually claimed work.
|
||||||
|
sleep_for = (
|
||||||
|
interval_seconds
|
||||||
|
if result.empty or result.retry_full_interval
|
||||||
|
else min(2.0, interval_seconds)
|
||||||
|
)
|
||||||
stop.wait(sleep_for)
|
stop.wait(sleep_for)
|
||||||
|
|
||||||
logger.info("claim-loop stop iterations=%s exit=%s", iterations, exit_code)
|
logger.info("claim-loop stop iterations=%s exit=%s", iterations, exit_code)
|
||||||
|
|
|
||||||
|
|
@ -7,7 +7,12 @@ from unittest.mock import MagicMock, patch
|
||||||
from rein_aharness.approaches import ApproachResult, APPROACH_FI_RESEARCH_BRIEF
|
from rein_aharness.approaches import ApproachResult, APPROACH_FI_RESEARCH_BRIEF
|
||||||
from rein_aharness.claim_loop import process_one, poll_peek
|
from rein_aharness.claim_loop import process_one, poll_peek
|
||||||
from rein_aharness.glas_execution import GLAS_APPROACH, GlasExecutionError
|
from rein_aharness.glas_execution import GLAS_APPROACH, GlasExecutionError
|
||||||
from rein_aharness.ops_run_client import OpsRun, OpsRunConfig, ActivityCoreOpsClient
|
from rein_aharness.ops_run_client import (
|
||||||
|
ActivityCoreOpsClient,
|
||||||
|
OpsRun,
|
||||||
|
OpsRunConfig,
|
||||||
|
OpsRunError,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
def _claimed_run() -> OpsRun:
|
def _claimed_run() -> OpsRun:
|
||||||
|
|
@ -34,6 +39,16 @@ def test_process_one_empty() -> None:
|
||||||
assert r.claimed is False
|
assert r.claimed is False
|
||||||
|
|
||||||
|
|
||||||
|
def test_process_one_claim_error_requests_full_backoff() -> None:
|
||||||
|
client = MagicMock(spec=ActivityCoreOpsClient)
|
||||||
|
client.config = OpsRunConfig(worker_id="w", lease_seconds=90)
|
||||||
|
client.claim.side_effect = OpsRunError("upstream failed")
|
||||||
|
result = process_one(client)
|
||||||
|
assert result.claimed is False
|
||||||
|
assert result.empty is False
|
||||||
|
assert result.retry_full_interval is True
|
||||||
|
|
||||||
|
|
||||||
def test_process_one_success_completes() -> None:
|
def test_process_one_success_completes() -> None:
|
||||||
client = MagicMock(spec=ActivityCoreOpsClient)
|
client = MagicMock(spec=ActivityCoreOpsClient)
|
||||||
client.config = OpsRunConfig(worker_id="w", lease_seconds=90)
|
client.config = OpsRunConfig(worker_id="w", lease_seconds=90)
|
||||||
|
|
|
||||||
|
|
@ -106,3 +106,11 @@ deployment rsync now excludes `.venv`, preventing workstation-bound shebangs
|
||||||
from overwriting the live worker environment. railiance01 is healthy on the
|
from overwriting the live worker environment. railiance01 is healthy on the
|
||||||
legacy worker but still needs the Glas checkout/package install, deployment,
|
legacy worker but still needs the Glas checkout/package install, deployment,
|
||||||
restart, and coordinated pilot.
|
restart, and coordinated pilot.
|
||||||
|
|
||||||
|
Deployment then exposed a pre-existing upstream database mismatch: the live
|
||||||
|
Activity Core image selects `ops_runs.harness_profile_ref`, but production
|
||||||
|
PostgreSQL has not applied migration `0008`, so every claim returns 500. The
|
||||||
|
worker's prior error path retried that failure every two seconds; it now uses
|
||||||
|
the configured 30-second backoff for claim errors and reserves the short pause
|
||||||
|
for cycles that actually claimed work. Activity Core must apply/verify its
|
||||||
|
migration before the coordinated pilot can run.
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue