cf1a0a80d8
The 06/25 silent hang was a hung ``sftp.listdir_attr()`` on the
worker thread — paramiko TCP-acked then went silent, the scheduler's
``asyncio.to_thread`` waited forever, and the operator had no
signal that polling had stalled. Every poll cycle since was
suspected of the same failure mode.
``backend/src/cyclone/clearhouse/__init__.``
- New ``_op_timeout_seconds()`` helper reads
``CYCLONE_SFTP_OP_TIMEOUT_SECONDS`` (default 30s). Rejects
unparseable, zero, and negative values — ``asyncio.wait_for(timeout=0)``
raises immediately and ``timeout<0`` is undefined per the asyncio
docs, so a typo would silently turn every SFTP call into an
instant failure. Logs a WARNING and falls back to the default.
- New ``async_list_inbound()`` async wrapper applies
``asyncio.wait_for(asyncio.to_thread(self.list_inbound),
timeout=N)``. ``wait_for`` cancels the awaiter after N
seconds but the worker thread keeps running until paramiko returns
on its own (paramiko is not asyncio-aware, can't be cancelled
cleanly). ``asyncio.TimeoutError`` propagates so the scheduler
can surface it as a transient SFTP error.
``backend/src/cyclone/scheduler.py``
- ``_tick_impl`` now calls ``client.async_list_inbound()``
instead of ``asyncio.to_thread(self._list_inbound)``. Catches
``asyncio.TimeoutError`` explicitly with a clear error message
("list_inbound: timeout") — separate from the generic
``Exception`` catch-all so the operator's tick-result error
reads as the actual cause.
- Deleted the now-dead ``Scheduler._list_inbound()`` shim.
``backend/tests/test_sftp_op_timeout.py`` (new, 6 tests):
1. Default timeout is 30s when the env var is unset.
2. Operator can override via env var without restart-rebuild.
3. Unparseable value (e.g. "30s") falls back to default.
4. Zero and negative values fall back to default.
5. A hanging list_inbound raises ``asyncio.TimeoutError`` within
the configured bound (the 06/25 hang pin).
6. A non-hanging call returns the stub's empty list normally
(sanity check that the wrapper doesn't break the happy path).
Scope note: only ``async_list_inbound`` is wired up. The other
SFTP methods (``list_inbound_names``, ``download_inbound``,
``read_file``, ``write_file``) are called from
operator-triggered paths (admin endpoints, CLI, claim submission)
where the operator can Ctrl-C the request, so a hang is at least
visible. Wrapping them would require making the FastAPI handlers
async, which is out of scope for Task 8. Tracked as a follow-up —
worth folding into the same shape when those endpoints are next
touched.
140 lines
5.2 KiB
Python
140 lines
5.2 KiB
Python
"""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 == []
|