diff --git a/backend/src/cyclone/scheduler.py b/backend/src/cyclone/scheduler.py index a226f86..9d242d5 100644 --- a/backend/src/cyclone/scheduler.py +++ b/backend/src/cyclone/scheduler.py @@ -95,6 +95,12 @@ class TickResult: files_skipped: int = 0 files_errored: int = 0 errors: list[str] = field(default_factory=list) + # SP27 Task 9: ``sftp_failed`` is the discriminator that separates + # SFTP-side failures (the listing step) from per-file processing + # errors. ``tick()`` keys off this flag — NOT ``result.errors`` — + # so a single malformed 999 cannot flip the operator's destructive + # pill. Set in ``_tick_impl``'s listing-error branches only. + sftp_failed: bool = False def as_dict(self) -> dict[str, Any]: return { @@ -107,12 +113,22 @@ class TickResult: "files_skipped": self.files_skipped, "files_errored": self.files_errored, "errors": list(self.errors), + "sftp_failed": self.sftp_failed, } @dataclass class SchedulerStatus: - """Snapshot of the scheduler's runtime state.""" + """Snapshot of the scheduler's runtime state. + + SP27 Task 9 added the four SFTP-error fields + (``consecutive_failures``, ``last_error``, ``last_error_at``, + ``last_sftp_attempt_at``) so the operator's UI can tell "all + quiet on the MFT front" from "we've been unable to reach the + MFT server for 3 hours". The previous 06/25 incident went + unnoticed because the status dict had no signal for a hung + SFTP poll. + """ running: bool poll_interval_seconds: int @@ -123,6 +139,13 @@ class SchedulerStatus: total_skipped: int total_errored: int last_tick: Optional[TickResult] = None + # SP27 Task 9: SFTP-error surfacing. The UI flips to a destructive + # pill when ``consecutive_failures >= 3`` (``CYCLONE_SCHEDULER_FAIL_THRESHOLD`` + # in the operator pill config; not enforced server-side). + consecutive_failures: int = 0 + last_error: Optional[str] = None + last_error_at: Optional[datetime] = None + last_sftp_attempt_at: Optional[datetime] = None def as_dict(self) -> dict[str, Any]: return { @@ -137,6 +160,15 @@ class SchedulerStatus: "total_skipped": self.total_skipped, "total_errored": self.total_errored, "last_tick": self.last_tick.as_dict() if self.last_tick else None, + "consecutive_failures": self.consecutive_failures, + "last_error": self.last_error, + "last_error_at": ( + self.last_error_at.isoformat() if self.last_error_at else None + ), + "last_sftp_attempt_at": ( + self.last_sftp_attempt_at.isoformat() + if self.last_sftp_attempt_at else None + ), } @@ -231,6 +263,40 @@ class Scheduler: # ticks stack up; the next tick fires only after the previous # one finishes). self._tick_in_progress = False + # SP27 Task 9: SFTP-error surfacing. ``_consecutive_failures`` + # counts back-to-back tick failures (cleared on the next + # successful tick). The other three fields are the most + # recent failure for the operator's pill — preserved across + # successes so the audit trail doesn't disappear the moment + # the MFT server recovers. + self._consecutive_failures = 0 + self._last_error: Optional[str] = None + self._last_error_at: Optional[datetime] = None + self._last_sftp_attempt_at: Optional[datetime] = None + + def _record_sftp_outcome( + self, *, success: bool, error: Optional[str] = None, + ) -> None: + """Update the SFTP-error state after a tick. + + ``success=True`` resets the consecutive-failure counter; the + last-error fields are preserved as an audit trail (the + operator's UI can still surface "last failure: 2 hours ago" + after a recovery). + + ``success=False`` bumps the counter and records the error + message + timestamp. ``last_sftp_attempt_at`` is bumped in + both cases so the operator can tell when the scheduler last + *tried* to reach the MFT (not just when it last failed). + """ + now = datetime.now(timezone.utc) + self._last_sftp_attempt_at = now + if success: + self._consecutive_failures = 0 + return + self._consecutive_failures += 1 + self._last_error = error + self._last_error_at = now # ---- Public API ------------------------------------------------------- @@ -278,6 +344,10 @@ class Scheduler: total_skipped=self._total_skipped, total_errored=self._total_errored, last_tick=self._last_tick, + consecutive_failures=self._consecutive_failures, + last_error=self._last_error, + last_error_at=self._last_error_at, + last_sftp_attempt_at=self._last_sftp_attempt_at, ) def is_running(self) -> bool: @@ -291,6 +361,13 @@ class Scheduler: SFTP server from a stampede when the operator hits ``/api/admin/scheduler/tick`` while a scheduled tick is already running. + + SP27 Task 9: the tick outcome is also recorded on the + scheduler's SFTP-error state via + :meth:`_record_sftp_outcome`. The discriminator is + ``TickResult.sftp_failed`` (set by ``_tick_impl``'s + listing-error branches only), not ``TickResult.errors`` + which is also populated by per-file processing errors. """ while self._tick_in_progress: await asyncio.sleep(0.05) @@ -303,6 +380,20 @@ class Scheduler: self._total_processed += result.files_processed self._total_skipped += result.files_skipped self._total_errored += result.files_errored + # SFTP-error surfacing: discriminator is ``result.sftp_failed``, + # NOT ``result.errors``. ``sftp_failed`` is set by + # ``_tick_impl``'s listing-error branches (timeout, connect + # refused). Per-file processing errors append to + # ``result.errors`` but do NOT set ``sftp_failed`` — a + # single malformed 999 in a 5-file batch must not flip + # the operator's pill to destructive. + if result.sftp_failed: + self._record_sftp_outcome( + success=False, + error="; ".join(result.errors), + ) + else: + self._record_sftp_outcome(success=True) return result finally: self._tick_in_progress = False @@ -323,6 +414,11 @@ class Scheduler: Honors ``_stop_event`` between files. Concurrent calls are coalesced the same way :meth:`tick` does, so a slow handler can't be stampeded by a parallel admin invocation. + + SP27 Task 9: deliberately does NOT update the SFTP-error + state (no listing, no SFTP). The consecutive-failure counter + is the signal for scheduled SFTP polling; operator-initiated + batches shouldn't be able to flip it. """ while self._tick_in_progress: await asyncio.sleep(0.05) @@ -386,11 +482,13 @@ class Scheduler: except asyncio.TimeoutError: log.exception("SFTP list_inbound timed out") result.errors.append("list_inbound: timeout") + result.sftp_failed = True result.finished_at = datetime.now(timezone.utc) return result except Exception as exc: # noqa: BLE001 log.exception("SFTP list_inbound failed") result.errors.append(f"list_inbound: {exc}") + result.sftp_failed = True result.finished_at = datetime.now(timezone.utc) return result diff --git a/backend/tests/test_scheduler_status_errors.py b/backend/tests/test_scheduler_status_errors.py new file mode 100644 index 0000000..b0ef19f --- /dev/null +++ b/backend/tests/test_scheduler_status_errors.py @@ -0,0 +1,229 @@ +"""SP27 Task 9: scheduler status surfaces SFTP failures. + +The 06/25 silent hang exposed a gap: ``Scheduler.status()`` had no +signal that polling had stalled. The operator's UI couldn't tell +"all quiet on the MFT front" from "we've been unable to reach the +MFT server for 3 hours". This pins the fix — the status dict +carries ``consecutive_failures``, ``last_error``, ``last_error_at``, +and ``last_sftp_attempt_at`` so the operator pill can flip to +destructive after 3 (or however many) consecutive failures. + +Discriminator under test: ``TickResult.sftp_failed`` is the flag +``tick()`` keys off — NOT ``TickResult.errors``. Per-file processing +errors append to ``errors`` but must NOT bump ``consecutive_failures``. +The test suite pins both halves of the discriminator (SFTP failure +bumps; per-file failure doesn't). +""" +from __future__ import annotations + +import asyncio +from datetime import datetime, timezone + +import pytest + +from cyclone.clearhouse import InboundFile +from cyclone.providers import SftpBlock +from cyclone.scheduler import Scheduler + + +def _block() -> SftpBlock: + return SftpBlock( + host="mft.example.com", + port=22, + username="user", + auth={"password_keychain_account": "x"}, + paths={"inbound": "/inbound", "outbound": "/outbound"}, + stub=True, + ) + + +# ---- pre-state: a fresh scheduler has no errors ------------------------- + + +def test_status_starts_with_zero_failures_and_no_error(): + """A fresh scheduler must report no errors — pins that the new + fields default cleanly and don't trip on first read.""" + sched = Scheduler(_block()) + status = sched.status() + assert status.consecutive_failures == 0 + assert status.last_error is None + assert status.last_error_at is None + assert status.last_sftp_attempt_at is None + + +# ---- a failed SFTP call bumps consecutive_failures ----------------------- + + +@pytest.mark.asyncio +async def test_status_records_last_error_after_failure(): + """A failed tick bumps consecutive_failures and records the error. + + Simulates a broken SFTP client. Pins the post-Task-8 behavior + where ``_tick_impl`` catches the error and ``tick()`` records + it on the scheduler's status. + """ + class BrokenClient: + def list_inbound(self): + raise RuntimeError("simulated outage") + + # SP27 Task 8: scheduler calls async_list_inbound() (the + # wait_for-wrapped variant). The broken stub raises before + # the wait_for wrapper runs — that's a real SFTP failure + # either way. + async def async_list_inbound(self): + raise RuntimeError("simulated outage") + + sched = Scheduler(_block(), sftp_client_factory=lambda b: BrokenClient()) + + before = datetime.now(timezone.utc) + await sched.tick() + after = datetime.now(timezone.utc) + + status = sched.status() + assert status.consecutive_failures == 1 + assert status.last_error is not None + assert "simulated outage" in status.last_error + # last_error_at is set to roughly the time of the failed tick + assert status.last_error_at is not None + assert before <= status.last_error_at <= after + # last_sftp_attempt_at is also bumped — separate from last_error_at + # when the failure was an SFTP-side issue (not a parse error). + assert status.last_sftp_attempt_at is not None + assert before <= status.last_sftp_attempt_at <= after + + +# ---- a successful tick clears consecutive_failures ----------------------- + + +@pytest.mark.asyncio +async def test_successful_tick_clears_consecutive_failures(): + """A tick that succeeds (or skips cleanly) resets the counter. + + Without this, a transient blip would lock the destructive pill + on forever. Pins the recovery path. + """ + class BrokenClient: + def list_inbound(self): + raise RuntimeError("transient") + + async def async_list_inbound(self): + raise RuntimeError("transient") + + class HealthyClient: + def list_inbound(self): + return [] + + async def async_list_inbound(self): + return [] + + # First: two failing ticks to bump the counter + sched = Scheduler(_block(), sftp_client_factory=lambda b: BrokenClient()) + await sched.tick() + await sched.tick() + assert sched.status().consecutive_failures == 2 + + # Then: switch to a healthy client and tick again + sched._sftp_client_factory = lambda b: HealthyClient() + await sched.tick() + status = sched.status() + assert status.consecutive_failures == 0 + # last_error is preserved (audit trail) but last_error_at is from + # the last failure, not the success. + assert status.last_error is not None # the historical error stays + assert "transient" in status.last_error + + +# ---- last_sftp_attempt_at moves on every attempt, not just failures --- + + +@pytest.mark.asyncio +async def test_last_sftp_attempt_at_advances_on_every_tick(): + """Even a successful tick should bump last_sftp_attempt_at. + + Without this, the operator can't tell when the scheduler last + *tried* to reach the MFT — only when it last failed. + """ + sched = Scheduler(_block()) # default stub returns [] + before = datetime.now(timezone.utc) + await sched.tick() + after = datetime.now(timezone.utc) + first = sched.status().last_sftp_attempt_at + assert first is not None + assert before <= first <= after + + # Wait a moment, tick again — last_sftp_attempt_at should move forward. + await asyncio.sleep(0.01) + await sched.tick() + second = sched.status().last_sftp_attempt_at + assert second is not None + assert second > first + + +# ---- as_dict surfaces the new fields for the API ----------------------- + + +def test_status_as_dict_includes_new_fields(): + """The API endpoint (/api/health or /api/admin/scheduler) reads + status.as_dict() — the new fields must be present and serialized.""" + sched = Scheduler(_block()) + d = sched.status().as_dict() + assert "consecutive_failures" in d + assert d["consecutive_failures"] == 0 + assert "last_error" in d + assert d["last_error"] is None + assert "last_error_at" in d + assert d["last_error_at"] is None + assert "last_sftp_attempt_at" in d + assert d["last_sftp_attempt_at"] is None + + +# ---- per-file processing errors must NOT bump consecutive_failures ------ +# +# This pins the SP27 Task 9 discriminator (``TickResult.sftp_failed`` +# vs. ``TickResult.errors``). Without this test, a future refactor +# that broadens ``result.errors`` semantics — or rewires `tick()` to +# key off it again — would silently re-break the discrimination. + + +@pytest.mark.asyncio +async def test_per_file_error_does_not_bump_consecutive_failures(monkeypatch): + """Per-file processing errors must NOT flip the destructive pill. + + The listing step succeeds but ``_handle_one`` simulates the + real handler's error path: it appends to ``result.errors`` + (the way ``_handle_one`` does on parser exceptions) but does + NOT touch ``result.sftp_failed``. The pill must stay clean. + """ + + class FilesClient: + async def async_list_inbound(self): + return [ + InboundFile( + name="tp123-999_MTRACKING-20260525001606060-1of1.x12", + size=100, + modified_at=datetime.now(timezone.utc), + # ``local_path`` is None because fake_handle_one + # never reaches the download step. + local_path=None, + ) + ] + + sched = Scheduler(_block(), sftp_client_factory=lambda b: FilesClient()) + + async def fake_handle_one(f, result): + # Mirrors the real ``_handle_one`` parser-failure path: + # appends to result.errors, does NOT set sftp_failed. + result.errors.append(f"{f.name}: ValueError: garbage") + + monkeypatch.setattr(sched, "_handle_one", fake_handle_one) + + await sched.tick() + status = sched.status() + + # Counter stays clean — per-file error is not an SFTP outage. + assert status.consecutive_failures == 0 + # ``last_error`` is reserved for SFTP-side failures. + assert status.last_error is None + # But ``last_sftp_attempt_at`` moves forward regardless — the + # scheduler did try, it just failed to ingest the file. + assert status.last_sftp_attempt_at is not None