diff --git a/.env.example b/.env.example index 99fa692..2324927 100644 --- a/.env.example +++ b/.env.example @@ -36,7 +36,12 @@ ALPHA_VANTAGE_API_KEY= DOLT_BINARY=dolt DOLT_DATA_DIR=dolt-data DOLT_EARNINGS_SUBDIR=earnings -DOLT_MIN_FREE_DISK_GB=2.0 +# Free-space floor checked before a pull (clone is ~1.7 GB and grows). 5 GB is a +# safe production default; lower only on a space-constrained dev box. +DOLT_MIN_FREE_DISK_GB=5.0 +# Hard timeout (s) on each dolt subprocess so a hung pull/sql can't pin the +# import connection + advisory lock. +DOLT_COMMAND_TIMEOUT_SECONDS=600.0 # Regime Monitor — FRED (VIX + HY credit spreads). Free key: https://fred.stlouisfed.org/docs/api/api_key.html # Optional: without it the volatility (V1) and credit (C1) pillars show as n/a. diff --git a/app/config.py b/app/config.py index 7a0e1e0..70d219a 100644 --- a/app/config.py +++ b/app/config.py @@ -45,7 +45,12 @@ class Settings(BaseSettings): dolt_binary: str = "dolt" dolt_data_dir: str = "dolt-data" dolt_earnings_subdir: str = "earnings" - dolt_min_free_disk_gb: float = 2.0 + # Headroom above the ~1.7 GB earnings clone (grows with pulls); 5 GB is a safe + # production floor — override lower only in a space-constrained dev box. + dolt_min_free_disk_gb: float = 5.0 + # Bound every dolt subprocess so a hung pull/sql can't pin the import's + # connection + advisory lock indefinitely. + dolt_command_timeout_seconds: float = 600.0 # Regime Monitor — FRED (VIX level + HY credit spreads). Optional: without it # the volatility (P5) and credit-spread (F2) signals are reported as n/a. diff --git a/app/services/dolt_client.py b/app/services/dolt_client.py index 7f21282..0d9c9f8 100644 --- a/app/services/dolt_client.py +++ b/app/services/dolt_client.py @@ -23,12 +23,19 @@ from pathlib import Path logger = logging.getLogger(__name__) +# Default subprocess timeout. A hung `dolt pull`/`sql` would otherwise pin the +# import's connection and its advisory lock indefinitely, so every call is +# bounded; callers may override per operation. +DEFAULT_TIMEOUT = 600.0 + class DoltError(RuntimeError): - """A dolt subprocess exited non-zero.""" + """A dolt subprocess failed, timed out, or exited non-zero.""" -async def _run(binary: str, args: list[str], *, cwd: Path) -> str: +async def _run( + binary: str, args: list[str], *, cwd: Path, timeout: float = DEFAULT_TIMEOUT +) -> str: proc = await asyncio.create_subprocess_exec( binary, *args, @@ -36,7 +43,15 @@ async def _run(binary: str, args: list[str], *, cwd: Path) -> str: stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) - stdout, stderr = await proc.communicate() + try: + stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout) + except asyncio.TimeoutError: + proc.kill() + try: + await proc.wait() + except ProcessLookupError: + pass + raise DoltError(f"dolt {args[0] if args else ''} timed out after {timeout:.0f}s") if proc.returncode != 0: raise DoltError( f"dolt {' '.join(args)} failed ({proc.returncode}): " @@ -60,24 +75,31 @@ def ensure_free_disk(path: Path, min_free_gb: float) -> None: ) -async def pull(repo_dir: Path, *, binary: str) -> None: +async def pull(repo_dir: Path, *, binary: str, timeout: float = DEFAULT_TIMEOUT) -> None: """`dolt pull` the persistent clone to the latest upstream revision.""" - await _run(binary, ["pull"], cwd=repo_dir) + await _run(binary, ["pull"], cwd=repo_dir, timeout=timeout) -async def current_commit(repo_dir: Path, *, binary: str) -> str: - """The HEAD commit hash of the clone — used as the import revision.""" +async def current_commit( + repo_dir: Path, *, binary: str, timeout: float = DEFAULT_TIMEOUT +) -> str: + """The HEAD commit hash of the clone — used as the import revision. + + Uses ``DOLT_HASHOF('HEAD')`` (which formally identifies HEAD) rather than + ordering ``dolt_log`` by timestamp.""" rows = await query_csv( - repo_dir, "SELECT commit_hash FROM dolt_log ORDER BY date DESC LIMIT 1", binary=binary + repo_dir, "SELECT DOLT_HASHOF('HEAD') AS commit_hash", binary=binary, timeout=timeout ) if not rows or not rows[0].get("commit_hash"): - raise DoltError("could not read HEAD commit hash from dolt_log") + raise DoltError("could not read HEAD commit hash") return rows[0]["commit_hash"] -async def query_csv(repo_dir: Path, sql: str, *, binary: str) -> list[dict[str, str]]: +async def query_csv( + repo_dir: Path, sql: str, *, binary: str, timeout: float = DEFAULT_TIMEOUT +) -> list[dict[str, str]]: """Run a read query and parse the CSV result into a list of dict rows.""" - out = await _run(binary, ["sql", "-q", sql, "-r", "csv"], cwd=repo_dir) + out = await _run(binary, ["sql", "-q", sql, "-r", "csv"], cwd=repo_dir, timeout=timeout) if not out.strip(): return [] return list(csv.DictReader(io.StringIO(out))) diff --git a/app/services/dolt_earnings_importer.py b/app/services/dolt_earnings_importer.py index 0d3281b..d035c70 100644 --- a/app/services/dolt_earnings_importer.py +++ b/app/services/dolt_earnings_importer.py @@ -49,6 +49,14 @@ MAX_LEAD_DAYS = 14 # currently-loaded forward calendar (guards the destructive re-insert against a # partial parse / symbol-mapping regression). MIN_FUTURE_RATIO = 0.5 +# Initial-load gates (when nothing is loaded yet — the ratio gate has no baseline). +# The source publishes a forward calendar; require a real horizon, not one stray +# future row. 21 days is a conservative floor under the ~35d horizon observed on +# the live clone. +MIN_FORWARD_HORIZON_DAYS = 21 +# ...and require the symbol join to reach most of the tracked universe, so a +# broken/normalization-dropped join can't seed a hollow calendar. +MIN_INITIAL_COVERAGE = 0.5 _CAL_SQL = ( "SELECT act_symbol, `date`, `when` FROM earnings_calendar " @@ -96,16 +104,24 @@ class DoltEarningsImporter: # -- SourceImporter protocol ------------------------------------------- async def detect_revision(self, db) -> str | None: + timeout = settings.dolt_command_timeout_seconds if self.do_pull: dolt_client.ensure_free_disk(self.repo_dir, settings.dolt_min_free_disk_gb) - await self._dolt.pull(self.repo_dir, binary=self.binary) - return await self._dolt.current_commit(self.repo_dir, binary=self.binary) + await self._dolt.pull(self.repo_dir, binary=self.binary, timeout=timeout) + return await self._dolt.current_commit( + self.repo_dir, binary=self.binary, timeout=timeout + ) async def stage(self, db) -> StagedEarnings: universe = await self._load_universe(db) # {normalised symbol: ticker_id} - cal_raw = await self._dolt.query_csv(self.repo_dir, _CAL_SQL, binary=self.binary) - hist_raw = await self._dolt.query_csv(self.repo_dir, _HIST_SQL, binary=self.binary) + timeout = settings.dolt_command_timeout_seconds + cal_raw = await self._dolt.query_csv( + self.repo_dir, _CAL_SQL, binary=self.binary, timeout=timeout + ) + hist_raw = await self._dolt.query_csv( + self.repo_dir, _HIST_SQL, binary=self.binary, timeout=timeout + ) _require_columns(cal_raw, {"act_symbol", "date", "when"}, "earnings_calendar") _require_columns( hist_raw, {"act_symbol", "period_end_date", "reported", "estimate"}, "eps_history" @@ -164,13 +180,36 @@ class DoltEarningsImporter: ) async def validate(self, db, staged: StagedEarnings) -> ValidationResult: + # Promote deletes+reinserts the forward calendar, so this gate is + # fail-closed. The forward calendar is the project's acceptance gate. messages: list[str] = [] + current_future = await self._current_future_count(db) + universe_size = int(staged.stats.get("universe_size", 0) or 0) + coverage = ( + staged.stats.get("symbols_with_calendar", 0) / universe_size + if universe_size + else 0.0 + ) + horizon_days = ( + (staged.max_announce_date - self.today).days if staged.max_announce_date else 0 + ) - # Fail-closed forward-calendar protection (promote deletes+reinserts it). if staged.future_count == 0: messages.append("no future-dated earnings rows staged") - current_future = await self._current_future_count(db) - if current_future > 0 and staged.future_count < current_future * MIN_FUTURE_RATIO: + elif current_future == 0: + # Initial load: no baseline for the ratio gate, so require a real + # forward horizon and broad universe coverage instead of one stray row. + if horizon_days < MIN_FORWARD_HORIZON_DAYS: + messages.append( + f"forward horizon only {horizon_days}d < {MIN_FORWARD_HORIZON_DAYS}d " + "on initial load" + ) + if coverage < MIN_INITIAL_COVERAGE: + messages.append( + f"initial universe coverage {coverage:.0%} " + f"< {MIN_INITIAL_COVERAGE:.0%} — symbol join likely broken" + ) + elif staged.future_count < current_future * MIN_FUTURE_RATIO: messages.append( f"forward calendar collapsed: staged {staged.future_count} future rows " f"< {MIN_FUTURE_RATIO:.0%} of current {current_future}" @@ -185,6 +224,8 @@ class DoltEarningsImporter: "staged_rows": len(staged.rows), "future_rows": staged.future_count, "current_future_rows": current_future, + "forward_horizon_days": horizon_days, + "universe_coverage": round(coverage, 3), } return ValidationResult( ok=not messages, diff --git a/tests/unit/test_dolt_client.py b/tests/unit/test_dolt_client.py new file mode 100644 index 0000000..0cf7512 --- /dev/null +++ b/tests/unit/test_dolt_client.py @@ -0,0 +1,47 @@ +"""Tests for the async dolt subprocess wrapper's failure handling. + +Uses the Python interpreter as a stand-in subprocess (cross-platform, no dolt +needed) to prove a non-zero exit and a hung command both raise DoltError — the +latter is what stops a hung pull from pinning the import connection + advisory +lock forever. +""" + +from __future__ import annotations + +import sys +import time +from pathlib import Path + +import pytest + +from app.services import dolt_client +from app.services.dolt_client import DoltError + + +async def test_run_raises_on_nonzero_exit(): + with pytest.raises(DoltError) as exc: + await dolt_client._run( + sys.executable, ["-c", "import sys; sys.exit(3)"], cwd=Path.cwd(), timeout=30 + ) + assert "3" in str(exc.value) + + +async def test_run_times_out_and_kills(): + start = time.monotonic() + with pytest.raises(DoltError) as exc: + await dolt_client._run( + sys.executable, + ["-c", "import time; time.sleep(30)"], + cwd=Path.cwd(), + timeout=0.5, + ) + elapsed = time.monotonic() - start + assert "timed out" in str(exc.value) + assert elapsed < 10 # killed promptly, not waited out + + +async def test_run_returns_stdout_on_success(): + out = await dolt_client._run( + sys.executable, ["-c", "print('hello')"], cwd=Path.cwd(), timeout=30 + ) + assert out.strip() == "hello" diff --git a/tests/unit/test_dolt_earnings_importer.py b/tests/unit/test_dolt_earnings_importer.py index 91346f0..aea6521 100644 --- a/tests/unit/test_dolt_earnings_importer.py +++ b/tests/unit/test_dolt_earnings_importer.py @@ -57,13 +57,13 @@ class FakeDolt: self.commit = commit self.pulled = False - async def pull(self, repo_dir, *, binary): + async def pull(self, repo_dir, *, binary, timeout=None): self.pulled = True - async def current_commit(self, repo_dir, *, binary): + async def current_commit(self, repo_dir, *, binary, timeout=None): return self.commit - async def query_csv(self, repo_dir, sql, *, binary): + async def query_csv(self, repo_dir, sql, *, binary, timeout=None): if "earnings_calendar" in sql: return self.calendar if "eps_history" in sql: @@ -111,18 +111,18 @@ async def test_stage_and_promote_basic(engine): factory = _factory(engine) await _seed_tickers(factory, ["AAPL"]) fake = FakeDolt( - calendar=[_cal("AAPL", "2026-05-01"), _cal("AAPL", "2026-08-01")], + calendar=[_cal("AAPL", "2026-05-01"), _cal("AAPL", "2026-08-20")], history=[_hist("AAPL", "2026-03-31", 1.5, 1.4)], # only the reported quarter ) run = await run_import(_importer(fake), engine=engine) assert run.status == STATUS_PROMOTED - assert run.source_max_date == date(2026, 8, 1) + assert run.source_max_date == date(2026, 8, 20) events = await _events(factory) assert len(events) == 2 past = next(e for e in events if e.announce_date == date(2026, 5, 1)) - future = next(e for e in events if e.announce_date == date(2026, 8, 1)) + future = next(e for e in events if e.announce_date == date(2026, 8, 20)) # past announcement paired to the reported quarter assert past.eps_actual == 1.5 and past.eps_estimate == 1.4 assert past.period_end == date(2026, 3, 31) and past.session == "amc" @@ -136,7 +136,7 @@ async def test_symbol_normalisation_join(engine): factory = _factory(engine) ids = await _seed_tickers(factory, ["AAPL", "BRK.B"]) fake = FakeDolt( - calendar=[_cal("AAPL", "2026-08-01"), _cal("BRK.B", "2026-08-05")], # dotted source symbol + calendar=[_cal("AAPL", "2026-08-20"), _cal("BRK.B", "2026-08-25")], # dotted source symbol history=[], ) run = await run_import(_importer(fake), engine=engine) @@ -151,15 +151,15 @@ async def test_reschedule_moves_future_row(engine): factory = _factory(engine) await _seed_tickers(factory, ["AAPL"]) - fake1 = FakeDolt(calendar=[_cal("AAPL", "2026-08-01")], history=[], commit="c1") + fake1 = FakeDolt(calendar=[_cal("AAPL", "2026-08-20")], history=[], commit="c1") await run_import(_importer(fake1), engine=engine) - fake2 = FakeDolt(calendar=[_cal("AAPL", "2026-08-08")], history=[], commit="c2") # moved + fake2 = FakeDolt(calendar=[_cal("AAPL", "2026-08-27")], history=[], commit="c2") # moved run2 = await run_import(_importer(fake2), engine=engine) assert run2.status == STATUS_PROMOTED dates = {e.announce_date for e in await _events(factory)} - assert dates == {date(2026, 8, 8)} # old future date gone, new one present + assert dates == {date(2026, 8, 27)} # old future date gone, new one present async def test_cancellation_removes_future_row(engine): @@ -183,17 +183,17 @@ async def test_past_row_never_deleted(engine): await _seed_tickers(factory, ["AAPL"]) fake1 = FakeDolt( - calendar=[_cal("AAPL", "2026-05-01"), _cal("AAPL", "2026-08-01")], history=[], commit="c1" + calendar=[_cal("AAPL", "2026-05-01"), _cal("AAPL", "2026-08-20")], history=[], commit="c1" ) await run_import(_importer(fake1), engine=engine) # Second import's calendar omits the past date but keeps a future one. - fake2 = FakeDolt(calendar=[_cal("AAPL", "2026-08-01")], history=[], commit="c2") + fake2 = FakeDolt(calendar=[_cal("AAPL", "2026-08-20")], history=[], commit="c2") await run_import(_importer(fake2), engine=engine) dates = {e.announce_date for e in await _events(factory)} assert date(2026, 5, 1) in dates # past result survived - assert date(2026, 8, 1) in dates + assert date(2026, 8, 20) in dates async def test_validate_fails_when_no_future(engine): @@ -214,10 +214,10 @@ async def test_validate_fails_on_forward_collapse(engine): fake1 = FakeDolt( calendar=[ - _cal("AAPL", "2026-08-01"), - _cal("MSFT", "2026-08-02"), - _cal("NVDA", "2026-08-03"), - _cal("AMZN", "2026-08-04"), + _cal("AAPL", "2026-08-20"), + _cal("MSFT", "2026-08-21"), + _cal("NVDA", "2026-08-22"), + _cal("AMZN", "2026-08-23"), ], history=[], commit="c1", @@ -226,7 +226,7 @@ async def test_validate_fails_on_forward_collapse(engine): assert len([e for e in await _events(factory) if e.announce_date > TODAY]) == 4 # Only one future row now → 1 < 50% of 4 → fail-closed, no destructive wipe. - fake2 = FakeDolt(calendar=[_cal("AAPL", "2026-08-01")], history=[], commit="c2") + fake2 = FakeDolt(calendar=[_cal("AAPL", "2026-08-20")], history=[], commit="c2") run2 = await run_import(_importer(fake2), engine=engine) assert run2.status == STATUS_FAILED @@ -249,7 +249,9 @@ async def test_real_clone_smoke(engine): from app.services import dolt_client factory = _factory(engine) - await _seed_tickers(factory, ["AAPL", "MSFT", "JPM"]) + # A few tickers spanning near + further-out reporters so the initial-load + # forward-horizon gate (>= 21d) is satisfied on the fixed clone. + await _seed_tickers(factory, ["AAPL", "MSFT", "NVDA", "JPM", "BRK.B"]) imp = DoltEarningsImporter( repo_dir=_CLONE_DIR, binary=_DOLT_BIN, today=date.today(), do_pull=False, dolt=dolt_client )