diff --git a/reflexio/server/services/playbook/README.md b/reflexio/server/services/playbook/README.md index b3896212..7de6ae6c 100644 --- a/reflexio/server/services/playbook/README.md +++ b/reflexio/server/services/playbook/README.md @@ -112,6 +112,13 @@ scheduled cycle; new durable signals preserve that due time so changes coalesce, while a remaining backlog continues in bounded follow-up units. The manual `/api/run_playbook_aggregation` route remains a fenced administrative full rerun. +The scheduler processes organizations sequentially. An exception escaping an +organization's execution defers that organization for five minutes using +monotonic time; later organizations continue after the failing call returns. +Failed repair attempts also respect the five-minute repair interval. Logs +identify the organization and failing stage, and provider enumeration failures +remain separate tick-level errors. Durable retries and lease fencing are unchanged. + Incremental work is isolated by `agent_version`: 1. Admit only the newest configured window of CURRENT, nonempty rows that have diff --git a/reflexio/server/services/playbook/aggregation_scheduler.py b/reflexio/server/services/playbook/aggregation_scheduler.py index 3b9e9e22..ce8a3bb7 100644 --- a/reflexio/server/services/playbook/aggregation_scheduler.py +++ b/reflexio/server/services/playbook/aggregation_scheduler.py @@ -112,8 +112,12 @@ def __init__( self._poll_interval_seconds = poll_interval_seconds self._worker_id = worker_id or uuid.uuid4().hex self._last_repair_at: dict[str, float] = {} + self._retry_after: dict[str, float] = {} + # Organization execution is sequential on the scheduler thread. + self._active_stage = "configuration" def _run_context(self, context: RequestContext) -> None: + self._active_stage = "configuration" storage = context.storage playbook_config = getattr( context.configurator.get_config(), "user_playbook_extractor_config", None @@ -140,8 +144,9 @@ def _run_context(self, context: RequestContext) -> None: last_repair_at is None or repair_now - last_repair_at >= _REPAIR_INTERVAL_SECONDS ): - repaired = storage.repair_playbook_aggregation_pending_state() self._last_repair_at[context.org_id] = repair_now + self._active_stage = "repair" + repaired = storage.repair_playbook_aggregation_pending_state() for agent_version in repaired: logger.info( "event=playbook_aggregation_progress state=scheduled org_id=%s " @@ -149,6 +154,7 @@ def _run_context(self, context: RequestContext) -> None: context.org_id, agent_version, ) + self._active_stage = "claim" claim = storage.claim_due_playbook_aggregation( owner=f"aggregation:{self._worker_id}:{context.org_id}", lease_seconds=AGGREGATION_LEASE_SECONDS, @@ -163,6 +169,7 @@ def _run_context(self, context: RequestContext) -> None: claim.agent_version, claim.fence, ) + self._active_stage = "aggregation" heartbeat = AggregationLeaseHeartbeat(storage, claim) heartbeat.start() success = False @@ -239,6 +246,7 @@ def _run_context(self, context: RequestContext) -> None: claim.fence, ) finally: + self._active_stage = "finalization" heartbeat.stop() active_claim = heartbeat.claim finished = storage.finish_playbook_aggregation_claim( @@ -285,9 +293,28 @@ def _run_once(self) -> float: for context in self._context_provider(): if self._stop_event.is_set(): break - self._run_context(context) + org_id = context.org_id + if time.monotonic() < self._retry_after.get(org_id, 0): + continue + try: + self._run_context(context) + except Exception: + self._retry_after[org_id] = ( + time.monotonic() + _REPAIR_INTERVAL_SECONDS + ) + logger.exception( + "event=playbook_aggregation_scheduler_org_failed org_id=%s " + "stage=%s retry_after_seconds=%s", + org_id, + self._active_stage, + _REPAIR_INTERVAL_SECONDS, + ) + else: + self._retry_after.pop(org_id, None) except Exception: - logger.exception("event=playbook_aggregation_scheduler_tick_failed") + logger.exception( + "event=playbook_aggregation_scheduler_tick_failed stage=context_provider" + ) return self._poll_interval_seconds def _on_started(self) -> None: diff --git a/tests/server/services/playbook/test_aggregation_scheduler.py b/tests/server/services/playbook/test_aggregation_scheduler.py index f0680954..f9f4a252 100644 --- a/tests/server/services/playbook/test_aggregation_scheduler.py +++ b/tests/server/services/playbook/test_aggregation_scheduler.py @@ -190,6 +190,104 @@ def test_lease_heartbeat_marks_renewal_exception_as_lost() -> None: heartbeat.require_live() +@pytest.mark.parametrize( + "failing_operation", + [ + "repair_playbook_aggregation_pending_state", + "claim_due_playbook_aggregation", + "finish_playbook_aggregation_claim", + ], +) +def test_org_failure_defers_only_that_org_and_recovers( + failing_operation, monkeypatch, caplog +) -> None: + clock = [1000.0] + monkeypatch.setattr(aggregation_scheduler.time, "monotonic", lambda: clock[0]) + broken = MagicMock(supports_incremental_playbook_aggregation=True) + healthy = MagicMock(supports_incremental_playbook_aggregation=True) + for storage in (broken, healthy): + storage.repair_playbook_aggregation_pending_state.return_value = [] + storage.claim_due_playbook_aggregation.return_value = None + healthy.claim_due_playbook_aggregation.return_value = PlaybookAggregationClaim( + "v1", "healthy-owner", 1, 0, 2000 + ) + healthy.get_playbook_aggregation_invalidations.return_value = [ + SimpleNamespace(invalidation_id=i) + for i in range(aggregation_scheduler.AGGREGATION_INVALIDATION_BATCH_SIZE + 1) + ] + monkeypatch.setattr( + aggregation_scheduler.AggregationLeaseHeartbeat, "start", lambda _: None + ) + if failing_operation == "finish_playbook_aggregation_claim": + broken.claim_due_playbook_aggregation.return_value = PlaybookAggregationClaim( + "v1", "owner", 1, 0, 2000 + ) + broken.get_playbook_aggregation_invalidations.return_value = [ + SimpleNamespace(invalidation_id=i) + for i in range( + aggregation_scheduler.AGGREGATION_INVALIDATION_BATCH_SIZE + 1 + ) + ] + failure = getattr(broken, failing_operation) + failure.side_effect = RuntimeError("database unavailable") + contexts = [_context(broken), _context(healthy)] + contexts[1].org_id = "org-2" + scheduler = aggregation_scheduler.PlaybookAggregationScheduler( + context_provider=lambda: contexts + ) + + scheduler._run_once() + assert failure.call_count == 1 + assert healthy.claim_due_playbook_aggregation.call_count == 1 + assert healthy.finish_playbook_aggregation_claim.call_count == 1 + assert healthy.finish_playbook_aggregation_claim.call_args.kwargs["success"] is True + assert "org_id=org-1" in caplog.text + expected_stage = { + "repair_playbook_aggregation_pending_state": "repair", + "claim_due_playbook_aggregation": "claim", + "finish_playbook_aggregation_claim": "finalization", + }[failing_operation] + assert f"stage={expected_stage}" in caplog.text + clock[0] += 299 + scheduler._run_once() + assert failure.call_count == 1 + assert healthy.claim_due_playbook_aggregation.call_count == 2 + assert healthy.finish_playbook_aggregation_claim.call_count == 2 + + failure.side_effect = None + broken.claim_due_playbook_aggregation.return_value = None + clock[0] += 1 + scheduler._run_once() + assert "org-1" not in scheduler._retry_after + assert healthy.claim_due_playbook_aggregation.call_count == 3 + assert healthy.finish_playbook_aggregation_claim.call_count == 3 + + +def test_failed_repair_attempt_is_throttled(monkeypatch) -> None: + storage = MagicMock(supports_incremental_playbook_aggregation=True) + storage.repair_playbook_aggregation_pending_state.side_effect = RuntimeError("down") + storage.claim_due_playbook_aggregation.return_value = None + monkeypatch.setattr(aggregation_scheduler.time, "monotonic", lambda: 1000.0) + scheduler = aggregation_scheduler.PlaybookAggregationScheduler( + context_provider=lambda: [] + ) + with pytest.raises(RuntimeError, match="down"): + scheduler._run_context(_context(storage)) + scheduler._run_context(_context(storage)) + storage.repair_playbook_aggregation_pending_state.assert_called_once() + + +def test_context_provider_failure_remains_recoverable(caplog) -> None: + provider = MagicMock(side_effect=[RuntimeError("repository unavailable"), []]) + scheduler = aggregation_scheduler.PlaybookAggregationScheduler( + context_provider=provider + ) + scheduler._run_once() + scheduler._run_once() + assert provider.call_count == 2 + assert "stage=context_provider" in caplog.text + + def test_stopped_local_scheduler_drops_captured_context(monkeypatch) -> None: monkeypatch.delenv("DEPLOYMENT_MODE", raising=False) aggregation_scheduler._LOCAL_SCHEDULERS.clear()