diff --git a/backend/src/cyclone/clearhouse/__init__.py b/backend/src/cyclone/clearhouse/__init__.py index 2144375..56e6069 100644 --- a/backend/src/cyclone/clearhouse/__init__.py +++ b/backend/src/cyclone/clearhouse/__init__.py @@ -24,6 +24,7 @@ stub secret and the paramiko auth will fail loudly at connect time. from __future__ import annotations +import asyncio import io import logging import os @@ -40,6 +41,59 @@ from cyclone.providers import SftpBlock log = logging.getLogger(__name__) +# --------------------------------------------------------------------------- +# Per-op SFTP timeout (SP27 Task 8) +# +# paramiko is synchronous; without a timeout, a hung ``listdir_attr`` +# freezes the worker thread indefinitely. The 06/25 silent hang was +# exactly this — the MFT server TCP-acked but stopped responding, the +# scheduler's ``asyncio.to_thread`` waited forever, and the operator +# had no signal that polling had stalled. +# +# The async wrappers below apply ``asyncio.wait_for`` to every SFTP +# call site so the event loop can give up after the configured bound. +# The bound is read fresh on every call (env-var-only, no module-level +# cache) so an operator who tunes the value at runtime picks it up on +# the next poll. +# --------------------------------------------------------------------------- + +_DEFAULT_SFTP_OP_TIMEOUT_SECONDS = 30.0 + + +def _op_timeout_seconds() -> float: + """Per-op SFTP timeout from ``CYCLONE_SFTP_OP_TIMEOUT_SECONDS``. + + Default 30s. Picked to comfortably outlast Gainwell's p99 + listdir_attr (~2s) while still surfacing real hangs inside one + scheduler tick. Operators who hit repeated timeouts should drop + this — but the right answer is to fix the MFT server, not to + paper over it here. + """ + raw = os.environ.get("CYCLONE_SFTP_OP_TIMEOUT_SECONDS") + if not raw: + return _DEFAULT_SFTP_OP_TIMEOUT_SECONDS + try: + value = float(raw) + except ValueError: + log.warning( + "CYCLONE_SFTP_OP_TIMEOUT_SECONDS=%r is not a float; using default %.1fs", + raw, _DEFAULT_SFTP_OP_TIMEOUT_SECONDS, + ) + return _DEFAULT_SFTP_OP_TIMEOUT_SECONDS + if value <= 0: + # ``asyncio.wait_for(timeout=0)`` raises immediately, and + # ``wait_for(timeout<0)`` is undefined per the asyncio docs. + # A zero/negative setting would silently turn every SFTP call + # into an instant timeout (a wave of bogus "list_inbound: + # timeout" errors). Treat the value as bad and fall back. + log.warning( + "CYCLONE_SFTP_OP_TIMEOUT_SECONDS=%r must be positive; using default %.1fs", + raw, _DEFAULT_SFTP_OP_TIMEOUT_SECONDS, + ) + return _DEFAULT_SFTP_OP_TIMEOUT_SECONDS + return value + + @dataclass class InboundFile: """A single file observed in the inbound MFT path.""" @@ -178,6 +232,37 @@ class SftpClient: return secrets.STUB_SECRET return value + # ---- Async surface (SP27 Task 8) ----------------------------------- + # + # Every sync SFTP call has an async wrapper that runs the paramiko + # call on a worker thread and applies an ``asyncio.wait_for(... + # timeout=N)`` around it. The wait_for cancels the awaiter but + # leaves the worker thread running until paramiko returns on its + # own (paramiko is not asyncio-aware, so we can't cancel the + # underlying socket cleanly). The scheduler should treat + # ``asyncio.TimeoutError`` from these wrappers as a transient + # SFTP error and surface it in ``Scheduler.status()`` (Task 9). + # + # The timeout is read on every call (see ``_op_timeout_seconds``) + # so an operator who tunes ``CYCLONE_SFTP_OP_TIMEOUT_SECONDS`` at + # runtime sees the new value on the next tick. + + async def async_list_inbound(self) -> list["InboundFile"]: + """Async-wrapped :meth:`list_inbound` with a per-op timeout. + + The 06/25 silent hang was a hung ``listdir_attr`` that froze + the worker thread indefinitely. This wrapper applies + ``asyncio.wait_for(...)`` so the event loop can give up after + ``CYCLONE_SFTP_OP_TIMEOUT_SECONDS`` (default 30s) and the + scheduler tick can surface the timeout in + ``result.errors`` (and, once Task 9 lands, in + ``Scheduler.status()``). + """ + return await asyncio.wait_for( + asyncio.to_thread(self.list_inbound), + timeout=_op_timeout_seconds(), + ) + # ---- Stub implementations (SP9) ------------------------------------- def _write_bytes_stub(self, remote_path: str, content: bytes) -> Path: diff --git a/backend/src/cyclone/scheduler.py b/backend/src/cyclone/scheduler.py index 20df257..a226f86 100644 --- a/backend/src/cyclone/scheduler.py +++ b/backend/src/cyclone/scheduler.py @@ -373,8 +373,21 @@ class Scheduler: started = datetime.now(timezone.utc) result = TickResult(started_at=started) + # SP27 Task 8: use the async-wrapped SFTP client so a hung + # ``listdir_attr`` (the 06/25 silent-hang failure mode) is + # bounded by ``CYCLONE_SFTP_OP_TIMEOUT_SECONDS`` instead of + # waiting on paramiko forever. ``asyncio.TimeoutError`` is + # handled explicitly so the tick surfaces a clear error in + # ``Scheduler.status()`` (Task 9) rather than a generic + # ``Exception`` catch-all. + client = self._sftp_client_factory(self._sftp_block) try: - files = await asyncio.to_thread(self._list_inbound) + files = await client.async_list_inbound() + except asyncio.TimeoutError: + log.exception("SFTP list_inbound timed out") + result.errors.append("list_inbound: timeout") + 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}") @@ -390,11 +403,6 @@ class Scheduler: result.finished_at = datetime.now(timezone.utc) return result - def _list_inbound(self) -> list[InboundFile]: - """Return files in the inbound MFT path. Runs on a thread.""" - client = self._sftp_client_factory(self._sftp_block) - return client.list_inbound() - async def _handle_one(self, f: InboundFile, result: TickResult) -> None: """Process one inbound file: skip-if-seen, classify, parse, record.""" if await self._already_processed(f.name): diff --git a/backend/tests/test_sftp_op_timeout.py b/backend/tests/test_sftp_op_timeout.py new file mode 100644 index 0000000..049e4d6 --- /dev/null +++ b/backend/tests/test_sftp_op_timeout.py @@ -0,0 +1,139 @@ +"""Per-op timeout guard for SFTP operations (SP27 Task 8). + +paramiko is synchronous; without a timeout, a hung ``listdir_attr`` +freezes the worker thread indefinitely. The 06/25 silent hang was +exactly this — the MFT server TCP-acked but stopped responding, the +scheduler's ``asyncio.to_thread`` waited forever, and the operator +had no signal that polling had stalled. + +This test pins the fix: the SFTP client's only async-wrapped call +site (``async_list_inbound``) applies an ``asyncio.wait_for(... +timeout=N)`` so the event loop can give up after the configured +bound. The bound is read from ``CYCLONE_SFTP_OP_TIMEOUT_SECONDS`` +at call time (default 30s). + +The other SFTP methods (``list_inbound_names``, ``download_inbound``, +``read_file``, ``write_file``) intentionally remain sync — they're +called from operator-triggered paths (admin endpoints, CLI) where +the operator can Ctrl-C the request. Wrapping them would require +making the FastAPI handlers themselves async, which is out of +scope for Task 8. Tracked as a follow-up. +""" +from __future__ import annotations + +import asyncio +import time + +import pytest + +from cyclone.clearhouse import SftpClient +from cyclone.providers import SftpBlock + + +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, + ) + + +class _HangingClient(SftpClient): + """SftpClient that hangs on list_inbound — exercises the + asyncio.wait_for timeout. The hang runs in a thread, so + ``asyncio.to_thread`` alone would never give up; only the + wait_for wrapper can break it. + + The hang is short (3s) so the per-test thread pool doesn't + accumulate stragglers across the timeout tests. Each test sets + ``CYCLONE_SFTP_OP_TIMEOUT_SECONDS=1`` so wait_for fires well + before the sleep completes. + """ + + def list_inbound(self): + time.sleep(3) + return [] + + +# ---- env-var parsing -------------------------------------------------------- + + +def test_op_timeout_default_is_30_seconds(monkeypatch): + """Default is 30s when CYCLONE_SFTP_OP_TIMEOUT_SECONDS is unset. + + Pins the documented default so a refactor that changes it has to + update both the constant and the spec. + """ + monkeypatch.delenv("CYCLONE_SFTP_OP_TIMEOUT_SECONDS", raising=False) + from cyclone.clearhouse import _op_timeout_seconds + assert _op_timeout_seconds() == 30.0 + + +def test_op_timeout_reads_env_var(monkeypatch): + """Operator can override via env var without a restart-rebuild.""" + monkeypatch.setenv("CYCLONE_SFTP_OP_TIMEOUT_SECONDS", "7.5") + from cyclone.clearhouse import _op_timeout_seconds + assert _op_timeout_seconds() == 7.5 + + +def test_op_timeout_rejects_unparseable_env_var(monkeypatch): + """A non-float value (e.g. "30s") falls back to the default. + + Pins the ValueError branch — a refactor that drops the try/except + would crash the SFTP wrapper on every call when the env var is + set to anything non-numeric. + """ + monkeypatch.setenv("CYCLONE_SFTP_OP_TIMEOUT_SECONDS", "30s") + from cyclone.clearhouse import _op_timeout_seconds + assert _op_timeout_seconds() == 30.0 + + +def test_op_timeout_rejects_zero_and_negative(monkeypatch): + """Zero / negative timeouts would silently turn every SFTP call + into an instant failure (asyncio.wait_for(timeout=0) raises + immediately; timeout<0 is undefined). They fall back to default. + """ + from cyclone.clearhouse import _op_timeout_seconds + for bad in ("0", "0.0", "-1", "-0.5"): + monkeypatch.setenv("CYCLONE_SFTP_OP_TIMEOUT_SECONDS", bad) + assert _op_timeout_seconds() == 30.0, f"value {bad!r} should fall back to default" + + +# ---- async wrapper behavior ------------------------------------------------- + + +@pytest.mark.asyncio +async def test_async_list_inbound_times_out_when_paramiko_hangs(monkeypatch): + """A hanging list_inbound must be cancelled by wait_for, not the thread. + + The 06/25 silent hang was exactly this scenario — paramiko's + listdir_attr TCP-acked then went silent, freezing the worker + thread. The wait_for wrapper is the only thing that can break + out of that. + """ + monkeypatch.setenv("CYCLONE_SFTP_OP_TIMEOUT_SECONDS", "1") + client = _HangingClient(_block()) + start = time.monotonic() + with pytest.raises(asyncio.TimeoutError): + await client.async_list_inbound() + elapsed = time.monotonic() - start + # Loose upper bound: timeout fires at ~1s, allow 3s for asyncio + # scheduling jitter on slow CI runners. If this ever fires at the + # hang duration (3s) the wait_for wrapper is missing. + assert elapsed < 3, f"timeout fired too late: {elapsed:.2f}s" + + +@pytest.mark.asyncio +async def test_async_list_inbound_returns_normally_when_fast(monkeypatch, tmp_path): + """Sanity check: a non-hanging call returns the stub's empty list + without raising — pins that the wrapper doesn't break the happy path.""" + monkeypatch.setenv("CYCLONE_SFTP_OP_TIMEOUT_SECONDS", "5") + block = _block() + block.staging_dir = str(tmp_path) + client = SftpClient(block) + # stub mode + no files staged = empty list, no hang, no timeout + result = await client.async_list_inbound() + assert result == []