diff --git a/src/tigerflow/cli/report.py b/src/tigerflow/cli/report.py index 10e07cf..4aba0c8 100644 --- a/src/tigerflow/cli/report.py +++ b/src/tigerflow/cli/report.py @@ -168,12 +168,12 @@ def fmt_duration(ms: float) -> str: if shown >= 5: break err_detail = ( - f"{err.exception_type}: {err.message}" - if err.exception_type - else err.message or "Unknown error" + f"{err.record.exception_type}: {err.record.message}" + if err.record.exception_type + else err.record.message or "Unknown error" ) lines.append( - f" [dim]{task_name}[/dim] {err.file} [red]{err_detail}[/red]" + f" [dim]{task_name}[/dim] {err.record.file} [red]{err_detail}[/red]" ) shown += 1 if shown >= 5: @@ -322,18 +322,18 @@ def report( } if "errors" in sections: result["errors"] = { - name: [ + task: [ { - "file": e.file, - "path": e.path, - "timestamp": e.timestamp.isoformat() if e.timestamp else None, - "exception_type": e.exception_type, - "message": e.message, - "traceback": e.traceback, + "file": err.record.file, + "path": err.path, + "timestamp": err.record.timestamp or None, + "exception_type": err.record.exception_type, + "message": err.record.message, + "traceback": err.record.traceback, } - for e in errs + for err in errors ] - for name, errs in pipeline_report.errors.items() + for task, errors in pipeline_report.errors.items() } print(json.dumps(result, indent=2, default=str)) diff --git a/src/tigerflow/models.py b/src/tigerflow/models.py index 12504fb..029bedb 100644 --- a/src/tigerflow/models.py +++ b/src/tigerflow/models.py @@ -2,6 +2,7 @@ import re import shlex import textwrap +from dataclasses import dataclass from datetime import datetime from enum import Enum from pathlib import Path @@ -14,6 +15,7 @@ from tigerflow.staging import StagingPipeline from tigerflow.utils import ( TEMP_FILE_PREFIX, + ErrorRecord, is_process_running, read_pid_file, validate_file_ext, @@ -461,15 +463,16 @@ class TaskProgress(BaseModel): failed: int = 0 -class FileError(BaseModel): - """Error information for a failed file.""" +@dataclass(slots=True) +class FileError: + """A failed file's error record paired with where it was found. + + `path` is the location of the .err file itself, which the record does + not carry: it is known only to the reader that discovers the file. + """ - file: str path: str - timestamp: datetime | None = None - exception_type: str = "" - message: str = "" - traceback: str = "" + record: ErrorRecord class TaskMeta(BaseModel): @@ -622,19 +625,17 @@ def report(self) -> PipelineReport: stem = file.name.removesuffix(".err") failed_stems.add(stem) try: - data = json.loads(file.read_text()) - task_errors.append( - FileError( - file=data.get("file", stem), - path=str(file), - timestamp=datetime.fromisoformat(data["timestamp"]), - exception_type=data.get("exception_type", ""), - message=data.get("message", ""), - traceback=data.get("traceback", ""), - ) + record = ErrorRecord.read(file) + record.file = record.file or stem + except (OSError, ValueError): + record = ErrorRecord( + timestamp="", + exception_type="", + message="", + traceback="", + file=stem, ) - except (OSError, json.JSONDecodeError, KeyError): - task_errors.append(FileError(file=stem, path=str(file))) + task_errors.append(FileError(path=str(file), record=record)) if task_errors: errors[task_dir.name] = task_errors diff --git a/src/tigerflow/tasks/local.py b/src/tigerflow/tasks/local.py index 9ba8753..a095741 100644 --- a/src/tigerflow/tasks/local.py +++ b/src/tigerflow/tasks/local.py @@ -11,10 +11,10 @@ from tigerflow.logconfig import logger from tigerflow.models import LocalTaskConfig from tigerflow.settings import settings -from tigerflow.utils import SetupContext, atomic_write +from tigerflow.utils import ErrorRecord, SetupContext, atomic_write from ._base import Task -from .utils import log_metrics, write_error_file +from .utils import log_metrics class LocalTask(Task): @@ -52,7 +52,7 @@ def task(input_file: Path, output_file: Path): output_file.name.removesuffix(self.config.output_ext) + ".err" ) error_file = output_dir / error_fname - write_error_file(error_file, input_file.name) + ErrorRecord.from_exception(file=input_file.name).write(error_file) logger.error("Failed processing: {}", input_file.name) # Clean up incomplete temporary files left behind by a prior process instance diff --git a/src/tigerflow/tasks/local_async.py b/src/tigerflow/tasks/local_async.py index 3a60a74..ff0f2b4 100644 --- a/src/tigerflow/tasks/local_async.py +++ b/src/tigerflow/tasks/local_async.py @@ -10,10 +10,10 @@ from tigerflow.logconfig import logger from tigerflow.models import LocalAsyncTaskConfig from tigerflow.settings import settings -from tigerflow.utils import SetupContext, atomic_write +from tigerflow.utils import ErrorRecord, SetupContext, atomic_write from ._base import Task -from .utils import log_metrics, write_error_file +from .utils import log_metrics class LocalAsyncTask(Task): @@ -53,7 +53,7 @@ async def task(input_file: Path, output_file: Path): output_file.name.removesuffix(self.config.output_ext) + ".err" ) error_file = self.config.output_dir / error_fname - write_error_file(error_file, input_file.name) + ErrorRecord.from_exception(file=input_file.name).write(error_file) logger.error("Failed processing: {}", input_file.name) async def worker(): diff --git a/src/tigerflow/tasks/slurm.py b/src/tigerflow/tasks/slurm.py index c1088b4..a6b4085 100644 --- a/src/tigerflow/tasks/slurm.py +++ b/src/tigerflow/tasks/slurm.py @@ -22,13 +22,14 @@ from tigerflow.settings import settings from tigerflow.utils import ( TEMP_FILE_PREFIX, + ErrorRecord, SetupContext, atomic_write, submit_to_slurm, ) from ._base import Task -from .utils import get_slurm_task_status, log_metrics, write_error_file +from .utils import get_slurm_task_status, log_metrics class SlurmTask(Task): @@ -82,7 +83,8 @@ async def setup(self, worker: Worker): logger.info("Task setup complete") except Exception: logger.exception("Task setup failed; aborting task") - setup_failed_sentinel.touch() + setup_failed_sentinel.touch() # ensure sentinel exists even if write below fails + ErrorRecord.from_exception().write(setup_failed_sentinel) async def teardown(self, worker: Worker): logger.info("Tearing down task") @@ -112,7 +114,7 @@ def task(input_file: Path, output_file: Path): metrics["status"] = "error" error_fname = output_file.name.removesuffix(output_ext) + ".err" error_file = output_dir / error_fname - write_error_file(error_file, input_file.name) + ErrorRecord.from_exception(file=input_file.name).write(error_file) logger.error("Failed processing: {}", input_file.name) # Define parameters for each Slurm job diff --git a/src/tigerflow/tasks/utils.py b/src/tigerflow/tasks/utils.py index 0a681a8..46dc00a 100644 --- a/src/tigerflow/tasks/utils.py +++ b/src/tigerflow/tasks/utils.py @@ -1,14 +1,10 @@ import json import subprocess -import sys -import traceback from contextlib import contextmanager from datetime import datetime, timezone -from pathlib import Path from tigerflow.logconfig import logger from tigerflow.models import TaskStatus, TaskStatusKind -from tigerflow.utils import atomic_write @contextmanager @@ -112,21 +108,3 @@ def get_pending_worker_ids(worker_job_name: str) -> list[int]: ).stdout return [int(job_id) for job_id in pending_ids.split() if job_id.isdigit()] - - -def write_error_file(error_path: Path, input_file: str) -> None: - """Write structured error JSON for a failed file. - - Must be called from within an exception handler. - """ - exc_type, exc_value, _ = sys.exc_info() - error_data = { - "file": input_file, - "timestamp": datetime.now(timezone.utc).isoformat(), - "exception_type": exc_type.__name__ if exc_type else "Unknown", - "message": str(exc_value) if exc_value else "", - "traceback": traceback.format_exc(), - } - with atomic_write(error_path) as temp_path: - with open(temp_path, "w") as f: - json.dump(error_data, f, indent=2) diff --git a/src/tigerflow/utils.py b/src/tigerflow/utils.py index 1f124d4..a665400 100644 --- a/src/tigerflow/utils.py +++ b/src/tigerflow/utils.py @@ -1,11 +1,15 @@ +import dataclasses import importlib +import json import os import re import subprocess import sys import tempfile +import traceback from collections.abc import Callable from contextlib import contextmanager +from datetime import datetime, timezone from pathlib import Path from subprocess import TimeoutExpired from types import SimpleNamespace @@ -226,3 +230,52 @@ def atomic_write(filepath: str | os.PathLike[str]): raise else: temp_path.replace(filepath) + + +@dataclasses.dataclass(slots=True) +class ErrorRecord: + """Structured error record for JSON serialization. + + Represents the on-disk schema used by .err and .setup-failed files. + """ + + timestamp: str + exception_type: str + message: str + traceback: str + file: str | None = None + + @classmethod + def from_exception(cls, file: str | None = None) -> "ErrorRecord": + """Capture error details from the current exception context. + + Must be called from within an exception handler. *file* is the + name of the input file being processed, if any; omit for errors + not associated with a specific file (e.g. task setup failures). + """ + exc_type, exc_value, _ = sys.exc_info() + return cls( + timestamp=datetime.now(timezone.utc).isoformat(), + exception_type=exc_type.__name__ if exc_type else "Unknown", + message=str(exc_value) if exc_value else "", + traceback=traceback.format_exc(), + file=file, + ) + + def write(self, path: Path) -> None: + """Write error record as JSON to *path* using atomic write.""" + with atomic_write(path) as temp_path: + with open(temp_path, "w") as f: + json.dump(dataclasses.asdict(self), f, indent=2) + + @classmethod + def read(cls, path: Path) -> "ErrorRecord": + """Read error record from a JSON file. + + Raises `ValueError` if the file content is malformed or incomplete. + """ + try: + data = json.loads(path.read_text()) + return cls(**data) + except (json.JSONDecodeError, TypeError) as exc: + raise ValueError(f"invalid error record: {path}") from exc diff --git a/tests/integration/test_slurm_task.py b/tests/integration/test_slurm_task.py index 6a9d5b2..079d9df 100644 --- a/tests/integration/test_slurm_task.py +++ b/tests/integration/test_slurm_task.py @@ -9,6 +9,7 @@ Set SLURM_TEST_DIR to a shared filesystem path accessible by compute nodes. """ +import json import os import shutil import signal @@ -270,6 +271,13 @@ def test_setup_failure_aborts(self, task_dirs, input_files, tasks_dir): sentinel_files = list(log_base.rglob(".setup-failed")) assert len(sentinel_files) > 0, "Expected .setup-failed sentinel file" + # Sentinel should contain expected error details + error_data = json.loads(sentinel_files[0].read_text()) + assert error_data["exception_type"] == "RuntimeError" + assert "Intentional setup failure" in error_data["message"] + assert "timestamp" in error_data + assert "traceback" in error_data + # No output files should have been produced output_files = [ f diff --git a/tests/unit/test_utils.py b/tests/unit/test_utils.py index 76ad121..b80b90d 100644 --- a/tests/unit/test_utils.py +++ b/tests/unit/test_utils.py @@ -1,3 +1,4 @@ +import json import os import textwrap from pathlib import Path @@ -6,6 +7,7 @@ from tigerflow.utils import ( TEMP_FILE_PREFIX, + ErrorRecord, atomic_write, has_running_pid, import_callable, @@ -205,3 +207,93 @@ def test_accepts_str_path(self, tmp_path: Path): assert temp.suffix == ".json" temp.write_text("{}") assert target.read_text() == "{}" + + +class TestErrorRecord: + def test_from_exception_captures_fields(self): + try: + raise ValueError("test error") + except ValueError: + record = ErrorRecord.from_exception() + + assert record.exception_type == "ValueError" + assert record.message == "test error" + assert "ValueError: test error" in record.traceback + assert record.timestamp # non-empty ISO string + + def test_from_exception_outside_handler(self): + record = ErrorRecord.from_exception() + assert record.exception_type == "Unknown" + assert record.message == "" + assert record.file is None + + def test_from_exception_captures_file(self): + try: + raise ValueError("test error") + except ValueError: + record = ErrorRecord.from_exception(file="input.txt") + assert record.file == "input.txt" + + def test_write_read_roundtrip(self, tmp_path: Path): + original = ErrorRecord( + timestamp="2026-01-01T00:00:00+00:00", + exception_type="RuntimeError", + message="boom", + traceback="Traceback ...", + file="input.txt", + ) + path = tmp_path / "error.err" + original.write(path) + loaded = ErrorRecord.read(path) + assert loaded == original + + def test_write_read_roundtrip_without_file(self, tmp_path: Path): + original = ErrorRecord( + timestamp="2026-01-01T00:00:00+00:00", + exception_type="RuntimeError", + message="boom", + traceback="Traceback ...", + ) + path = tmp_path / "error.err" + original.write(path) + loaded = ErrorRecord.read(path) + assert loaded == original + assert loaded.file is None + + def test_read_extra_keys_raises_value_error(self, tmp_path: Path): + path = tmp_path / "error.err" + data = { + "timestamp": "2026-01-01T00:00:00+00:00", + "exception_type": "RuntimeError", + "message": "boom", + "traceback": "Traceback ...", + "file": "input.txt", + "unexpected": "value", + } + path.write_text(json.dumps(data)) + with pytest.raises(ValueError, match="invalid error record"): + ErrorRecord.read(path) + + def test_read_without_file_key(self, tmp_path: Path): + path = tmp_path / "error.err" + data = { + "timestamp": "2026-01-01T00:00:00+00:00", + "exception_type": "RuntimeError", + "message": "boom", + "traceback": "Traceback ...", + } + path.write_text(json.dumps(data)) + loaded = ErrorRecord.read(path) + assert loaded.file is None + + def test_read_missing_keys_raises_value_error(self, tmp_path: Path): + path = tmp_path / "error.err" + path.write_text(json.dumps({"timestamp": "2026-01-01T00:00:00+00:00"})) + with pytest.raises(ValueError, match="invalid error record"): + ErrorRecord.read(path) + + def test_read_malformed_json_raises_value_error(self, tmp_path: Path): + path = tmp_path / "error.err" + path.write_text("not json") + with pytest.raises(ValueError, match="invalid error record"): + ErrorRecord.read(path)