diff --git a/.gitignore b/.gitignore index 47e92eb..43673bb 100644 --- a/.gitignore +++ b/.gitignore @@ -182,4 +182,6 @@ mail_config.toml logs/* logs -PR_DESCRIPTION.md \ No newline at end of file +PR_DESCRIPTION.md + +FEATURE_PLAN.md \ No newline at end of file diff --git a/FEATURE_PLAN1.MD b/FEATURE_PLAN1.MD new file mode 100644 index 0000000..f04a0e6 --- /dev/null +++ b/FEATURE_PLAN1.MD @@ -0,0 +1,86 @@ +USER: Could ProcessResult be ditched and replaced by ProcessExecutionReport? Show impacts. Don't code yet. + +Claude: Could ProcessResult be replaced by ProcessExecutionReport? + +Structurally, yes — ProcessExecutionReport.entries is a superset of what ProcessResult carries (it adds function/args/kwargs/status/elapsed_seconds/attempts for every task, success/errored/skipped alike, in topological order). The raw data ProcessRunner already collects (passed_results, failed_results, skipped_tasks, process.tasks) is exactly what from_result consumes — so ProcessRunner.run() could build a ProcessExecutionReport directly instead of a ProcessResult, without from_result needing a ProcessResult input at all (avoids the current process.py ↔ execution_report.py TYPE_CHECKING coupling — execution_report.py becomes self-sufficient). + +What breaks / changes + +1. Lost: raw Exception object access +TaskResult.exception is a live Exception instance (tests do isinstance(result.exception, TimeoutError)). TaskReportEntry.error is ErrorData | None, whose .exception field is a str(exception). Any caller that needs to isinstance()-check or re-raise the original exception loses that — ErrorData would need an added field, or TaskReportEntry would need to carry the raw TaskResult too. + +2. API shape changes for every consumer +- result.passed_tasks_results["x"].result → report.successes["x"].result (or report.entries["x"].result) +- result.failed_tasks (a set[str]) → no direct equivalent; becomes set(report.errored) | set(report.skipped) +- result.errored_tasks / result.skipped_tasks → set(report.errored) / set(report.skipped) +- result.failed_tasks_results["x"] (a TaskResult) → report.entries["x"].error (an ErrorData, narrower — see #1) +- .worked / boolean checks → .status == TaskStatus.SUCCESS etc. + +3. Test impact — large +Every test file that touches these attributes needs rewriting: +test_args_kwargs.py, test_complex_dag_failures.py, test_normal_run_no_errors.py, test_parallel_race_conditions.py, test_timeout_retry.py, plus 3 manual-test scripts. This is the bulk of the effort — dozens of assertion sites. + +4. Docs impact +README's ProcessResult section (added in this session) and any usage examples would need replacing with ProcessExecutionReport equivalents; docs/reference.md would drop the ProcessResult mkdocstrings entry. + +5. Breaking change vs. additive discipline +This session deliberately kept everything additive (FEATURE_PLAN's stance). Removing ProcessResult and changing Process.run()'s return type is a hard breaking change — would warrant a major version bump and a migration note, not a patch/minor release. + +6. Minor ergonomics loss +Simple "did everything pass?" checks (if not result.failed_tasks) become if not report.errored and not report.skipped unless a convenience property (e.g. report.ok) is added. + +Net assessment + +Feasible and arguably a cleaner unified model long-term, but it's a breaking API change with wide blast radius across tests/docs, and needs at least one new field on ErrorData/TaskReportEntry to avoid losing the raw exception object. If you want this, I'd suggest either (a) a v6 major bump doing a clean replacement, or (b) keep ProcessResult as a thin deprecated wrapper for one release while Process.run() returns ProcessExecutionReport and ProcessResult.from_report(...) is offered for back-compat. + +✻ Cooked for 49s + +❯ /advisor off + ⎿ Advisor disabled + +❯ is ProcessExecutionReport being created anywhere yet? + + Searched for 1 pattern + +Every test file that touches these attributes needs rewriting: +test_args_kwargs.py, test_complex_dag_failures.py, test_normal_run_no_errors.py, test_parallel_race_conditions.py, test_timeout_retry.py, plus 3 manual-test scripts. This is the bulk of the effort — dozens of assertion sites. + +4. Docs impact +README's ProcessResult section (added in this session) and any usage examples would need replacing with ProcessExecutionReport equivalents; docs/reference.md would drop the ProcessResult mkdocstrings entry. + +5. Breaking change vs. additive discipline +This session deliberately kept everything additive (FEATURE_PLAN's stance). Removing ProcessResult and changing Process.run()'s return type is a hard breaking change — would warrant a major version bump and a migration note, not a patch/minor release. + +6. Minor ergonomics loss +Simple "did everything pass?" checks (if not result.failed_tasks) become if not report.errored and not report.skipped unless a convenience property (e.g. report.ok) is added. + +Net assessment + +Feasible and arguably a cleaner unified model long-term, but it's a breaking API change with wide blast radius across tests/docs, and needs at least one new field on ErrorData/TaskReportEntry to avoid losing the raw exception object. If you want this, I'd suggest either (a) a v6 major bump doing a clean replacement, or (b) keep ProcessResult as a thin deprecated wrapper for one release while Process.run() returns ProcessExecutionReport and ProcessResult.from_report(...) is offered for back-compat. + +USER: failed_results and passed_results and failed_tasks are separated. Shouldn't they be in a single dict with k:v task.name:task_result_obj ? + +Claude: Yes — and it'd fix a latent inconsistency: currently failed_tasks (a set) = errored_tasks ∪ skipped_tasks, but failed_results (a dict) only has entries for errored tasks, not skipped ones. So failed_tasks and failed_results.keys() silently diverge — anyone iterating failed_tasks and indexing into failed_results will KeyError on skipped tasks. + +Proposal + +Maintain one internal dict in ProcessRunner: self.results: dict[str, TaskResult], one entry per task regardless of outcome (success, errored, or skipped — synthesizing a TaskResult for skipped tasks the same way the parallel defensive branch already does for unexpected exceptions). + +Discriminator problem: TaskResult.worked: bool alone can't tell errored apart from skipped (both worked=False). Two options: +- Implicit: attempts == 0 ⇒ skipped. Works, but relies on an unstated convention — fragile if someone constructs a TaskResult manually. +- Explicit (cleaner): give TaskResult a status: TaskStatus field (reuse the enum from execution_report.py). Then results[name].status is authoritative everywhere. + +Impact + +- Internals: _is_done, _all_deps_met, _is_unrunnable currently check membership across passed_results/failed_tasks/skipped_tasks/failed_results — all rewritten to query self.results by status. Net simplification. +- ProcessResult public API: can stay byte-for-byte unchanged. passed_tasks_results, failed_tasks, skipped_tasks, errored_tasks, failed_tasks_results all become derived views filtered from self.results at the end of run() — zero breaking change, fixes the divergence bug above as a side effect. +- Ties into the ProcessExecutionReport question: if TaskResult gains status: TaskStatus, then self.results is already shaped almost exactly like ProcessExecutionReport.entries (just missing name/function/args/kwargs, which come from Task). from_result/ProcessRunner.run() could build the report directly from this dict with minimal mapping — makes that future migration much cheaper. +- Effort: moderate, contained to process.py (ProcessRunner) + adding a field to TaskResult (additive, default-valued — safe). No test changes needed if ProcessResult's shape is preserved. + +This seems like a good incremental step worth doing on its own regardless of the ProcessExecutionReport replacement decision. + +USER: Tasks could have an extra attr TaskStatus already defined. Status could be expanded to PENDING, SUCCESS, ERRORED, SKIPPED. These changes should simplify or solve some of the problems stated by you. + + + +INSTRUCTION: IMPLEMENT BOTH THINGS. Removal of ProcessResult and use of single dict self.results. \ No newline at end of file diff --git a/README.md b/README.md index 700f959..4916819 100644 --- a/README.md +++ b/README.md @@ -37,7 +37,7 @@ A `Process` holds a list of `Task`s. At construction it validates names, types, When you call `process.run()`, tasks are topologically sorted and scheduled: dependencies first, independent tasks in parallel. -A `TaskDependency` can forward an upstream result directly into a downstream function, as a positional or keyword argument. The result is a `ProcessResult` with `passed_tasks_results` and `failed_tasks` for inspection. +A `TaskDependency` can forward an upstream result directly into a downstream function, as a positional or keyword argument. The result is a `ProcessExecutionReport` with `successes`, `errored`, and `skipped` for inspection. --- @@ -70,7 +70,7 @@ tasks = [ with Process(tasks) as p: result = p.run(parallel=True) -print(result.passed_tasks_results["enrich"].result) +print(result.successes["enrich"].result) # [{'id': 1, 'name': 'user-1'}, {'id': 2, 'name': 'user-2'}, {'id': 3, 'name': 'user-3'}] ``` @@ -201,11 +201,11 @@ tasks = [ with Process(tasks) as process: result = process.run(parallel=True) -print("passed:", sorted(result.passed_tasks_results)) +print("passed:", sorted(result.successes)) # archive_report, build_report, compute_revenue, compute_stock, fetch_inventory, fetch_orders -print("failed:", sorted(result.failed_tasks)) +print("failed:", sorted(set(result.errored) | set(result.skipped))) # notify_slack -print("report:", result.passed_tasks_results["build_report"].result) +print("report:", result.successes["build_report"].result) # daily-report | revenue=59.50 stock=262.50 ``` @@ -265,24 +265,71 @@ TaskDependency( ```python Process(tasks: list[Task]) # validates types, names, deps, cycles -process.run(parallel: bool | None = None, max_workers: int = 4) -> ProcessResult +process.run(parallel: bool | None = None, max_workers: int = 4) -> ProcessExecutionReport ``` - Raises `DependencyNotFoundError`, `CircularDependencyError`, `TypeError`, `ValueError` on construction if the workflow is malformed. - `parallel=None` auto-parallelises when `len(tasks) >= 10`; `max_workers=1` is always sequential. - Use as a context manager — it cleans up `FileHandler`s on exit. -### `ProcessResult` +### `TaskResult` ```python -result.passed_tasks_results # dict[str, TaskResult] — name → TaskResult for every task that succeeded -result.failed_tasks # set[str] — all tasks that did not produce a result (errored + skipped) -result.errored_tasks # set[str] — tasks whose function actually raised -result.skipped_tasks # set[str] — tasks skipped because an upstream dependency failed +TaskResult( + status: TaskStatus, + result: Any, + exception: Exception | None, + error_data: ErrorData | None = None, + elapsed_seconds: float = 0.0, # wall-clock time across all attempts + attempts: int = 0, # attempts actually executed (0 if never run) +) +``` + +- `status` — `TaskStatus.PENDING | SUCCESS | ERRORED | SKIPPED`. +- `worked` — `True` if `status == TaskStatus.SUCCESS`. -TaskResult(worked: bool, result: Any, exception: Exception | None) +### `ProcessExecutionReport` + +```python +report.entries # dict[str, TaskReportEntry] — one entry per task, in topological order +report.successes # dict[str, TaskReportEntry] — entries with status == TaskStatus.SUCCESS +report.errored # dict[str, TaskReportEntry] — entries with status == TaskStatus.ERRORED +report.skipped # dict[str, TaskReportEntry] — entries with status == TaskStatus.SKIPPED ``` +A `TaskReportEntry` carries `name`, `function`, `args`, `kwargs`, `status` +(`TaskStatus.SUCCESS | ERRORED | SKIPPED`), `elapsed_seconds`, `attempts`, +plus `result` (set when `SUCCESS`) and `error: ErrorData | None` (set when +`ERRORED`). + +```python +with Process([load_task, apply_task, notify_task]) as process: + report = process.run() + +for name, entry in report.entries.items(): + if entry.status is TaskStatus.ERRORED: + print(f"{name} failed after {entry.attempts} attempt(s): {entry.error.exception}") +``` + +### `ErrorData` + +```python +ErrorData( + task_name: str, + function: str, + args: tuple[Any, ...], + kwargs: dict[str, Any], + downstream_impact: list[str], + exception: str, + traceback_str: str, + traced_vars: dict[str, str], + traced_vars_location: str, +) +``` + +Structured failure context for a single task, available via +`TaskResult.error_data` and `TaskReportEntry.error`. + ### `SMTPConfig` ```python @@ -389,10 +436,10 @@ body, so receivers can verify the payload wasn't tampered with. When a task raises: -1. The exception is caught and stored in `TaskResult.exception`; the task name goes into `failed_tasks` and `errored_tasks`. -2. **Every task that depends on it (directly or indirectly) is skipped** — added to `failed_tasks` and `skipped_tasks` without running. +1. The exception is caught and stored in `TaskResult.exception`; the task's entry in the report gets `status == TaskStatus.ERRORED`. +2. **Every task that depends on it (directly or indirectly) is skipped** — its entry gets `status == TaskStatus.SKIPPED` without running. 3. **Every other independent part of the workflow keeps running.** With `parallel=True` they keep running concurrently on the worker pool. -4. After `run()` returns, `ProcessResult.errored_tasks` and `ProcessResult.skipped_tasks` let you distinguish root failures from cascade skips for triage or alerting. +4. After `run()` returns, `ProcessExecutionReport.errored` and `ProcessExecutionReport.skipped` let you distinguish root failures from cascade skips for triage or alerting. When a task has `retries >= 1`, a failure matching `retry_on` triggers another attempt before the task is declared failed and its dependants are skipped. This gives transient errors (network blips, connection resets) a chance to resolve without aborting downstream work. @@ -422,7 +469,7 @@ This makes the library a good fit for fan-out / fan-in pipelines, "best-effort" - **Shared log file** — pass the same `log_path` to every `Task` for a single combined run.log; pass distinct paths for per-task isolation. - **Auto-parallel** — `Process.run()` with no argument runs sequentially for small workflows and switches to parallel for `len(tasks) >= 10`. Pass `parallel=True` or `parallel=False` to force the mode. -- **Result inspection** — iterate `result.passed_tasks_results.items()` to log or post-process every successful task; iterate `result.failed_tasks` for triage. +- **Result inspection** — iterate `report.successes.items()` to log or post-process every successful task; iterate `set(report.errored) | set(report.skipped)` for triage. - **Re-raising** — wrap `process.run()` in `try/except` if you need a non-zero exit code on any failure; the library itself does not raise on partial failure. diff --git a/docs/reference.md b/docs/reference.md index 20cb014..1ed66a3 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -3,8 +3,12 @@ This page is automatically generated from the source code docstrings. ::: processes.process.Process -::: processes.process.ProcessResult ::: processes.task.Task ::: processes.task.TaskDependency +::: processes.task.TaskResult +::: processes.execution_report.ProcessExecutionReport +::: processes.execution_report.TaskReportEntry +::: processes.execution_report.TaskStatus +::: processes._error_data.ErrorData ::: processes.email_config.SMTPConfig ::: processes.email_config.HTMLEmailStyle \ No newline at end of file diff --git a/src/processes/__init__.py b/src/processes/__init__.py index 535262a..0f8f50e 100644 --- a/src/processes/__init__.py +++ b/src/processes/__init__.py @@ -1,6 +1,7 @@ from importlib.metadata import PackageNotFoundError as _pnfe from importlib.metadata import version as _v +from ._error_data import ErrorData as ErrorData from .email_config import HTMLEmailStyle as HTMLEmailStyle from .email_config import SMTPConfig as SMTPConfig from .exceptions import ( @@ -12,6 +13,8 @@ from .exceptions import ( TaskNotFoundError as TaskNotFoundError, ) +from .execution_report import ProcessExecutionReport as ProcessExecutionReport +from .execution_report import TaskReportEntry as TaskReportEntry from .notification_channels import EmailChannel as EmailChannel from .notification_channels import NotificationChannel as NotificationChannel from .notification_channels import WebhookChannel as WebhookChannel @@ -19,6 +22,7 @@ from .task import Task as Task from .task import TaskDependency as TaskDependency from .task import TaskResult as TaskResult +from .task import TaskStatus as TaskStatus from .webhook_config import WebhookConfig as WebhookConfig try: diff --git a/src/processes/_error_data.py b/src/processes/_error_data.py index 061d3e1..2aa7a4b 100644 --- a/src/processes/_error_data.py +++ b/src/processes/_error_data.py @@ -6,7 +6,7 @@ @dataclass(frozen=True) -class _ErrorData: +class ErrorData: """Typed view of a task failure, extracted from ``record.task_context``. Attributes @@ -46,7 +46,7 @@ class _ErrorData: class _ErrorContextFormatter(logging.Formatter): """Base formatter providing typed access to a record's failure context.""" - def _error_data(self, record: logging.LogRecord) -> _ErrorData: + def _error_data(self, record: logging.LogRecord) -> ErrorData: """Extract the failure context from a log record. Parameters @@ -57,12 +57,12 @@ def _error_data(self, record: logging.LogRecord) -> _ErrorData: Returns ------- - _ErrorData + ErrorData Typed view of ``record.task_context``, with defaults filled in for any missing fields. """ ctx = getattr(record, "task_context", None) or {} - return _ErrorData( + return ErrorData( task_name=str(ctx.get("task_name", "?")), function=str(ctx.get("function", "?")), args=ctx.get("args", ()), diff --git a/src/processes/_webhook_internals.py b/src/processes/_webhook_internals.py index f8f7d1f..749b927 100644 --- a/src/processes/_webhook_internals.py +++ b/src/processes/_webhook_internals.py @@ -7,7 +7,7 @@ import urllib.request from typing import Any -from ._error_data import _ErrorContextFormatter, _ErrorData +from ._error_data import ErrorData, _ErrorContextFormatter from .webhook_config import WebhookConfig _SIGNATURE_HEADER = "X-Signature-SHA256" @@ -46,8 +46,8 @@ def format(self, record: logging.LogRecord) -> str: payload = {**generic_payload, **self._extra_payload} return json.dumps(payload) - def _build_payload(self, error: _ErrorData) -> dict[str, Any]: - """Build the JSON-serializable payload dict from ``_ErrorData``. + def _build_payload(self, error: ErrorData) -> dict[str, Any]: + """Build the JSON-serializable payload dict from ``ErrorData``. Subclasses targeting a specific webhook service can override this to reshape the payload, while reusing ``format`` and the rest of @@ -55,7 +55,7 @@ def _build_payload(self, error: _ErrorData) -> dict[str, Any]: Parameters ---------- - error : _ErrorData + error : ErrorData Typed failure context for the record being formatted. Returns diff --git a/src/processes/execution_report.py b/src/processes/execution_report.py new file mode 100644 index 0000000..06b1a95 --- /dev/null +++ b/src/processes/execution_report.py @@ -0,0 +1,116 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import TYPE_CHECKING, Any + +from ._error_data import ErrorData +from .task import TaskResult, TaskStatus + +if TYPE_CHECKING: + from .process import Process + + +@dataclass(frozen=True) +class TaskReportEntry: + """Per-task entry in a :class:`ProcessExecutionReport`. + + Attributes + ---------- + name : str + The task's name. + function : str + Name of the function the task runs. + args : tuple[Any, ...] + Positional arguments the task was constructed with. + kwargs : dict[str, Any] + Keyword arguments the task was constructed with. + status : TaskStatus + Outcome of the task: ``SUCCESS``, ``ERRORED``, or ``SKIPPED``. + elapsed_seconds : float + Wall-clock time spent running the task across all attempts. ``0.0`` + for skipped tasks. + attempts : int + Number of attempts actually executed. ``0`` for skipped tasks. + result : Any | None + The task's return value if ``status`` is ``SUCCESS``, else ``None``. + error : ErrorData | None + Structured failure context if ``status`` is ``ERRORED``, else ``None``. + """ + + name: str + function: str + args: tuple[Any, ...] + kwargs: dict[str, Any] + status: TaskStatus + elapsed_seconds: float + attempts: int + result: Any | None = None + error: ErrorData | None = None + + +@dataclass(frozen=True) +class ProcessExecutionReport: + """Per-task breakdown of a finished :meth:`Process.run` call. + + Attributes + ---------- + entries : dict[str, TaskReportEntry] + Mapping of task name to its report entry, ordered the same way as + ``process.tasks`` (topological order). + """ + + entries: dict[str, TaskReportEntry] = field(default_factory=dict) + + def _filter(self, status: TaskStatus) -> dict[str, TaskReportEntry]: + return {name: entry for name, entry in self.entries.items() if entry.status == status} + + @property + def successes(self) -> dict[str, TaskReportEntry]: + """Entries for tasks whose status is ``SUCCESS``.""" + return self._filter(TaskStatus.SUCCESS) + + @property + def errored(self) -> dict[str, TaskReportEntry]: + """Entries for tasks whose status is ``ERRORED``.""" + return self._filter(TaskStatus.ERRORED) + + @property + def skipped(self) -> dict[str, TaskReportEntry]: + """Entries for tasks whose status is ``SKIPPED``.""" + return self._filter(TaskStatus.SKIPPED) + + @classmethod + def from_results( + cls, process: Process, results: dict[str, TaskResult] + ) -> ProcessExecutionReport: + """Build a report from a finished process run. + + Parameters + ---------- + process : Process + The process that was run. Used for task definitions (name, + function, args, kwargs) and topological ordering. + results : dict[str, TaskResult] + One ``TaskResult`` per task, keyed by task name, as produced by + :class:`~processes.process.ProcessRunner`. + + Returns + ------- + ProcessExecutionReport + One entry per task in ``process.tasks``, in topological order. + """ + entries: dict[str, TaskReportEntry] = {} + for task in process.tasks: + res = results[task.name] + entries[task.name] = TaskReportEntry( + name=task.name, + function=task.func.__name__, + args=task.args, + kwargs=task.kwargs, + status=res.status, + elapsed_seconds=res.elapsed_seconds, + attempts=res.attempts, + result=res.result if res.worked else None, + error=res.error_data if res.status == TaskStatus.ERRORED else None, + ) + return cls(entries) diff --git a/src/processes/process.py b/src/processes/process.py index 8deaf41..2237b74 100644 --- a/src/processes/process.py +++ b/src/processes/process.py @@ -3,46 +3,14 @@ from types import TracebackType from typing import Literal, Self +from ._error_data import ErrorData from .exceptions import CircularDependencyError, DependencyNotFoundError, TaskNotFoundError -from .task import Task, TaskResult +from .execution_report import ProcessExecutionReport +from .task import Task, TaskResult, TaskStatus __all__ = ["CircularDependencyError", "DependencyNotFoundError", "TaskNotFoundError"] -class ProcessResult: - """ - Container for the results of a process execution. - - Holds the outcomes of all tasks executed in a process, separating successful - and failed tasks with their respective results. - - Attributes - ---------- - passed_tasks_results : dict[str, TaskResult] - Mapping of task names to TaskResult objects for all tasks that executed successfully. - failed_tasks : set[str] - All tasks that did not produce a result — union of ``errored_tasks`` and - ``skipped_tasks``. - skipped_tasks : set[str] - Subset of ``failed_tasks``: tasks that were never run because an upstream - dependency failed (cascade-skipped). - errored_tasks : set[str] - Subset of ``failed_tasks``: tasks whose function actually raised an exception - (``failed_tasks - skipped_tasks``). - """ - - def __init__( - self, - passed_tasks_results: dict[str, TaskResult], - failed_tasks: set[str], - skipped_tasks: set[str], - ): - self.passed_tasks_results = passed_tasks_results - self.failed_tasks = failed_tasks - self.skipped_tasks = skipped_tasks - self.errored_tasks: set[str] = failed_tasks - skipped_tasks - - class Process: """ Manages and executes a collection of interdependent tasks. @@ -211,7 +179,7 @@ def get_task(self, task_name: str) -> Task: except KeyError as err: raise TaskNotFoundError(task_name) from err - def run(self, parallel: bool | None = None, max_workers: int = 4) -> ProcessResult: + def run(self, parallel: bool | None = None, max_workers: int = 4) -> ProcessExecutionReport: """Execute all tasks in the process. Runs tasks sequentially or in parallel while respecting dependencies. @@ -230,9 +198,8 @@ def run(self, parallel: bool | None = None, max_workers: int = 4) -> ProcessResu Returns ------- - ProcessResult - Contains passed_tasks_results (dict mapping task names to TaskResult) - and failed_tasks (set of task names that failed). + ProcessExecutionReport + Per-task breakdown of the run, in topological order. """ if parallel is None: parallel = len(self.tasks) >= 10 @@ -241,8 +208,7 @@ def run(self, parallel: bool | None = None, max_workers: int = 4) -> ProcessResu if parallel: if max_workers == 1: parallel = False # Fallback to sequential if only one worker - process_result = self.runner.run(parallel, max_workers) - return process_result + return self.runner.run(parallel, max_workers) def get_dependant_tasks(self, task_name: str) -> list[Task]: """Retrieve all tasks that directly or indirectly depend on a given task. @@ -287,42 +253,38 @@ class ProcessRunner: """ Executes tasks in a Process, handling both sequential and parallel execution. - Manages task execution state, tracks passed and failed tasks, and coordinates - dependencies during execution. + Manages task execution state and coordinates dependencies during execution. Attributes ---------- process : Process Reference to the parent Process being executed. - passed_results : dict[str, TaskResult] - Results from successfully executed tasks. - failed_tasks : set[str] - Names of tasks that failed during execution. - skipped_tasks : set[str] - Names of tasks that were never run because an upstream dependency failed. + results : dict[str, TaskResult] + One entry per task, keyed by task name. Every task starts out + ``PENDING`` and transitions to ``SUCCESS``, ``ERRORED``, or + ``SKIPPED`` once it is resolved. submitted_tasks : set[str] Names of tasks that have been submitted for execution. """ def __init__(self, process_ref: Process): self.process = process_ref - self.passed_results: dict[str, TaskResult] = {} - self.failed_tasks: set[str] = set() - self.skipped_tasks: set[str] = set() + self.results: dict[str, TaskResult] = { + task.name: TaskResult.pending() for task in process_ref.tasks + } self.submitted_tasks: set[str] = set() def _is_done(self) -> bool: - """Check whether every task has either passed or failed. + """Check whether every task has been resolved. Returns ------- bool - True if every task has a recorded result (passed or failed), - False otherwise. + True if no task is still ``PENDING``, False otherwise. """ - return len(self.passed_results) + len(self.failed_tasks) >= len(self.process.tasks) + return all(res.status != TaskStatus.PENDING for res in self.results.values()) - def run(self, parallel: bool, max_workers: int) -> ProcessResult: + def run(self, parallel: bool, max_workers: int) -> ProcessExecutionReport: """Execute all tasks in the process using the specified execution mode. Parameters @@ -334,17 +296,20 @@ def run(self, parallel: bool, max_workers: int) -> ProcessResult: Returns ------- - ProcessResult - The combined results of all task executions. + ProcessExecutionReport + Per-task breakdown of the run, in topological order. """ if parallel: self._run_parallel(max_workers) else: self._run_sequential() - return ProcessResult(self.passed_results, self.failed_tasks, self.skipped_tasks) + return ProcessExecutionReport.from_results(self.process, self.results) + + def _has_failed_dep(self, task: Task) -> bool: + """Return whether any dependency errored or was skipped. - def _is_unrunnable(self, task: Task) -> bool: - """Check if a task cannot be run due to failed dependencies. + Pure query with no side effects: it does not record the task as + ``SKIPPED``. Callers decide when to mark the cascade-skip. Parameters ---------- @@ -354,15 +319,13 @@ def _is_unrunnable(self, task: Task) -> bool: Returns ------- bool - True if any of the task's dependencies have failed, False otherwise. - If True, the task is recorded in both ``failed_tasks`` and - ``skipped_tasks`` (cascade-skip). + True if any of the task's dependencies errored or were skipped, + False otherwise. """ - if any(d.task_name in self.failed_tasks for d in task.dependencies): - self.failed_tasks.add(task.name) - self.skipped_tasks.add(task.name) - return True - return False + return any( + self.results[d.task_name].status in (TaskStatus.ERRORED, TaskStatus.SKIPPED) + for d in task.dependencies + ) def _all_deps_met(self, task: Task) -> bool: """Check if all dependencies of a task have been successfully executed. @@ -375,21 +338,18 @@ def _all_deps_met(self, task: Task) -> bool: Returns ------- bool - True if all dependencies have passed, False otherwise. + True if all dependencies succeeded, False otherwise. """ - return all(d.task_name in self.passed_results for d in task.dependencies) + return all(self.results[d.task_name].worked for d in task.dependencies) def _run_sequential(self) -> None: """Execute all tasks sequentially in dependency order.""" for task in self.process.tasks: - if self._is_unrunnable(task): + if self._has_failed_dep(task): + self.results[task.name] = TaskResult.skipped() continue if self._all_deps_met(task): - res = task.run(self.process) - if res.worked: - self.passed_results[task.name] = res - else: - self.failed_tasks.add(task.name) + self.results[task.name] = task.run(self.process) def _run_parallel(self, max_workers: int) -> None: """Execute tasks in parallel using a thread pool while respecting dependencies. @@ -407,13 +367,21 @@ def _run_parallel(self, max_workers: int) -> None: with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: fut_to_name = {} while not self._is_done(): - # Look for candidates to execute now + # Record cascade-skips first (a dependency errored or was + # skipped), then look for candidates whose deps all succeeded. + for t in self.process.tasks: + if ( + t.name not in self.submitted_tasks + and self.results[t.name].status == TaskStatus.PENDING + and self._has_failed_dep(t) + ): + self.results[t.name] = TaskResult.skipped() + candidates = [ t for t in self.process.tasks if t.name not in self.submitted_tasks - and t.name not in self.failed_tasks - and not self._is_unrunnable(t) + and self.results[t.name].status == TaskStatus.PENDING and self._all_deps_met(t) ] @@ -424,7 +392,7 @@ def _run_parallel(self, max_workers: int) -> None: self.submitted_tasks.add(task.name) # If there are tasks pending, wait. As soon one is completed, - # save as passed or failed and remove from futures. + # save its result and remove from futures. if fut_to_name: done, _ = concurrent.futures.wait( fut_to_name.keys(), return_when="FIRST_COMPLETED" @@ -432,17 +400,15 @@ def _run_parallel(self, max_workers: int) -> None: for fut in done: name = fut_to_name.pop(fut) try: - res = fut.result() - if res.worked: - self.passed_results[name] = res - else: - self.failed_tasks.add(name) - except Exception: - self.failed_tasks.add(name) + self.results[name] = fut.result() + except Exception as e: + self.results[name] = TaskResult.errored( + e, error_data=ErrorData(task_name=name, exception=str(e)) + ) else: # No running tasks and no new candidates. The # ``_is_unrunnable`` side effect above may have - # marked the remaining tasks as failed, completing + # marked the remaining tasks as skipped, completing # the DAG. Re-check the loop condition before # declaring a stall. if not self._is_done(): diff --git a/src/processes/task.py b/src/processes/task.py index af17cf6..6abe8bd 100644 --- a/src/processes/task.py +++ b/src/processes/task.py @@ -1,7 +1,9 @@ from __future__ import annotations import concurrent.futures +import time from collections.abc import Callable +from enum import Enum from typing import TYPE_CHECKING, Any if TYPE_CHECKING: @@ -9,32 +11,123 @@ import logging +from ._error_data import ErrorData from ._tb_utils import _build_traced_vars, _build_traced_vars_location, _format_traceback from .exceptions import CircularDependencyError from .notification_channels import NotificationChannel, _FileChannel +class TaskStatus(Enum): + """Outcome of a task within a process execution. + + Attributes + ---------- + PENDING + The task has not been executed yet. + SUCCESS + The task ran and its function returned without raising. + ERRORED + The task ran and its function raised an exception (after exhausting + retries, if any). + SKIPPED + The task never ran because an upstream dependency failed. + """ + + PENDING = "pending" + SUCCESS = "success" + ERRORED = "errored" + SKIPPED = "skipped" + + class TaskResult: """ Container for the result of a task execution. - Holds the outcome of running a task, including whether it succeeded, - its return value, and any exception that occurred. + Holds the outcome of running a task, including its status, return value, + and any exception that occurred. Attributes ---------- + status : TaskStatus + Outcome of the task: ``PENDING``, ``SUCCESS``, ``ERRORED``, or ``SKIPPED``. worked : bool - True if the task executed successfully, False if an exception occurred. + True if ``status`` is ``SUCCESS``, False otherwise. result : Any The return value of the task's function if execution succeeded, None if failed. exception : Exception | None The exception object if execution failed, None if successful. + error_data : ErrorData | None + Structured failure context (function, args, kwargs, traceback, traced + variables, downstream impact) when execution failed; None if the task + succeeded. + elapsed_seconds : float + Wall-clock time spent running the task across all attempts, in + seconds. Defaults to ``0.0``. + attempts : int + Number of attempts actually executed (1 or more if the task ran, + 0 if it never ran). Defaults to ``0``. """ - def __init__(self, worked: bool, result: Any, exception: Exception | None): - self.worked = worked + def __init__( + self, + status: TaskStatus, + result: Any, + exception: Exception | None, + error_data: ErrorData | None = None, + elapsed_seconds: float = 0.0, + attempts: int = 0, + ): + self.status = status self.result = result self.exception = exception + self.error_data = error_data + self.elapsed_seconds = elapsed_seconds + self.attempts = attempts + + @property + def worked(self) -> bool: + """True if the task executed successfully, False otherwise.""" + return self.status == TaskStatus.SUCCESS + + @classmethod + def pending(cls) -> TaskResult: + """Build the placeholder result for a task that has not run yet.""" + return cls(TaskStatus.PENDING, None, None) + + @classmethod + def skipped(cls) -> TaskResult: + """Build the result for a task skipped after an upstream dependency failed.""" + return cls(TaskStatus.SKIPPED, None, None) + + @classmethod + def success(cls, result: Any, *, elapsed_seconds: float = 0.0, attempts: int = 0) -> TaskResult: + """Build the result for a task whose function returned without raising.""" + return cls( + TaskStatus.SUCCESS, + result, + None, + elapsed_seconds=elapsed_seconds, + attempts=attempts, + ) + + @classmethod + def errored( + cls, + exception: Exception, + *, + error_data: ErrorData | None = None, + elapsed_seconds: float = 0.0, + attempts: int = 0, + ) -> TaskResult: + """Build the result for a task whose function raised after exhausting retries.""" + return cls( + TaskStatus.ERRORED, + None, + exception, + error_data=error_data, + elapsed_seconds=elapsed_seconds, + attempts=attempts, + ) class TaskDependency: @@ -352,7 +445,7 @@ def _resolve_args(self, executing_process: Process | None) -> tuple[list[Any], d final_kwargs = self.kwargs.copy() if executing_process is not None: for dep in self.dependencies: - dep_result = executing_process.runner.passed_results[dep.task_name].result + dep_result = executing_process.runner.results[dep.task_name].result if dep.use_result_as_additional_args: final_args.append(dep_result) if dep.use_result_as_additional_kwargs: @@ -381,9 +474,30 @@ def _build_failure_context( "traced_vars_location": _build_traced_vars_location(exc_tb, self._frame_filter), } + def _errored_result( + self, + exc: Exception, + executing_process: Process | None, + elapsed_seconds: float, + attempts: int, + ) -> TaskResult: + """Log ``exc`` through this task's handlers and wrap it in a TaskResult.""" + task_context = self._build_failure_context(exc, executing_process) + self.logger.error(str(exc), exc_info=exc, extra={"task_context": task_context}) + return TaskResult.errored( + exc, + error_data=ErrorData(**task_context), + elapsed_seconds=elapsed_seconds, + attempts=attempts, + ) + def run(self, executing_process: Process | None = None) -> TaskResult: """Execute the task, retrying on transient failures up to ``retries`` times. + Never propagates: a failure while resolving dependency arguments (before + any attempt runs) is captured and returned as an ``ERRORED`` result with + ``attempts=0``, just like a failure inside the task's function. + Parameters ---------- executing_process : Process, optional @@ -396,21 +510,27 @@ def run(self, executing_process: Process | None = None) -> TaskResult: ``worked=True`` with the return value on success; ``worked=False`` with the last exception on failure. """ - final_args, final_kwargs = self._resolve_args(executing_process) - max_attempts = self.retries + 1 effective_retry_on = ( self.retry_on if self.retry_on is not None else (ConnectionError, TimeoutError) ) self.logger.info(f"Starting {self.name}.") - last_exc: Exception | None = None + start = time.monotonic() + + try: + final_args, final_kwargs = self._resolve_args(executing_process) + except Exception as e: + return self._errored_result(e, executing_process, time.monotonic() - start, attempts=0) + last_exc: Exception | None = None for attempt in range(1, max_attempts + 1): try: result = self._call_with_timeout(final_args, final_kwargs) self.logger.info(f"Finished {self.name}.") - return TaskResult(True, result, None) + return TaskResult.success( + result, elapsed_seconds=time.monotonic() - start, attempts=attempt + ) except Exception as e: last_exc = e retryable = self.retries >= 1 and attempt < max_attempts @@ -422,6 +542,6 @@ def run(self, executing_process: Process | None = None) -> TaskResult: break assert last_exc is not None - task_context = self._build_failure_context(last_exc, executing_process) - self.logger.error(str(last_exc), exc_info=last_exc, extra={"task_context": task_context}) - return TaskResult(False, None, last_exc) + return self._errored_result( + last_exc, executing_process, time.monotonic() - start, attempts=attempt + ) diff --git a/tests/manual_tests/manual_pipeline_inspect.py b/tests/manual_tests/manual_pipeline_inspect.py index ebc4a21..9982771 100644 --- a/tests/manual_tests/manual_pipeline_inspect.py +++ b/tests/manual_tests/manual_pipeline_inspect.py @@ -406,10 +406,10 @@ def main() -> int: else: print("-" * 72) print("passed:") - for name in sorted(result.passed_tasks_results): + for name in sorted(result.successes): print(f" + {name}") print("failed (includes cascading-skipped):") - for name in sorted(result.failed_tasks): + for name in sorted(set(result.errored) | set(result.skipped)): print(f" - {name}") if exit_code == 0: diff --git a/tests/manual_tests/manual_themed_tracebacks.py b/tests/manual_tests/manual_themed_tracebacks.py index 7a2c9de..1455421 100644 --- a/tests/manual_tests/manual_themed_tracebacks.py +++ b/tests/manual_tests/manual_themed_tracebacks.py @@ -224,20 +224,21 @@ def _run_one_combo( return False, 2 print("passed:") - for name in sorted(result.passed_tasks_results): + for name in sorted(result.successes): print(f" + {name}") print("failed (includes cascading-skipped):") - for name in sorted(result.failed_tasks): + failed = set(result.errored) | set(result.skipped) + for name in sorted(failed): print(f" - {name}") # Strict post-conditions: downstream tasks were never invoked. ok = ( - "risky_step" in result.failed_tasks - and "child_a" in result.failed_tasks - and "child_b" in result.failed_tasks - and "risky_step" not in result.passed_tasks_results - and "child_a" not in result.passed_tasks_results - and "child_b" not in result.passed_tasks_results + "risky_step" in failed + and "child_a" in failed + and "child_b" in failed + and "risky_step" not in result.successes + and "child_a" not in result.successes + and "child_b" not in result.successes ) if not ok: print( diff --git a/tests/manual_tests/manual_webhook_inspect.py b/tests/manual_tests/manual_webhook_inspect.py index ac4cf3b..fbfe0c8 100644 --- a/tests/manual_tests/manual_webhook_inspect.py +++ b/tests/manual_tests/manual_webhook_inspect.py @@ -171,10 +171,10 @@ def main() -> int: print("-" * 72) print("passed:") - for name in sorted(result.passed_tasks_results): + for name in sorted(result.successes): print(f" + {name}") print("failed (includes cascading-skipped):") - for name in sorted(result.failed_tasks): + for name in sorted(set(result.errored) | set(result.skipped)): print(f" - {name}") return 0 diff --git a/tests/test_args_kwargs.py b/tests/test_args_kwargs.py index b53f7f1..8b101b6 100644 --- a/tests/test_args_kwargs.py +++ b/tests/test_args_kwargs.py @@ -72,12 +72,13 @@ def div(a: int, b: int) -> int: dependencies=[TaskDependency("task_1", use_result_as_additional_args=True)], ) with Process([t1, t2]) as process: - process_result = process.run() + report = process.run() - assert len(process_result.failed_tasks) == 0 - assert len(process_result.passed_tasks_results) == 2 - assert process_result.passed_tasks_results["task_1"].result == 2 - assert process_result.passed_tasks_results["task_2"].result == 5 + assert len(report.errored) == 0 + assert len(report.skipped) == 0 + assert len(report.successes) == 2 + assert report.successes["task_1"].result == 2 + assert report.successes["task_2"].result == 5 def test_add_extra_args_kwargs(self) -> None: """Test passing extra arguments and keyword arguments to a Task @@ -116,10 +117,11 @@ def div(a: int, b: int, c: int = 5) -> int: ], ) with Process([t0, t1, t2, t3]) as process: - process_result = process.run() - - assert len(process_result.failed_tasks) == 0 - assert len(process_result.passed_tasks_results) == 4 - assert process_result.passed_tasks_results["task_1"].result == 10 - assert process_result.passed_tasks_results["task_2"].result == 5 - assert process_result.passed_tasks_results["task_3"].result == 4 + report = process.run() + + assert len(report.errored) == 0 + assert len(report.skipped) == 0 + assert len(report.successes) == 4 + assert report.successes["task_1"].result == 10 + assert report.successes["task_2"].result == 5 + assert report.successes["task_3"].result == 4 diff --git a/tests/test_complex_dag_failures.py b/tests/test_complex_dag_failures.py index 0a75f32..630a6d0 100644 --- a/tests/test_complex_dag_failures.py +++ b/tests/test_complex_dag_failures.py @@ -138,24 +138,21 @@ def make_task(name: str, deps, fail: bool = False) -> Task: assert call_counts[name] == 1, ( f"Independent task {name} did not execute exactly once (got {call_counts[name]})" ) - assert name in result.passed_tasks_results, ( - f"Independent task {name} missing from passed_results" - ) - assert result.passed_tasks_results[name].worked is True - assert result.passed_tasks_results[name].result == f"payload_{name}" - assert result.passed_tasks_results[name].exception is None - - assert len(result.passed_tasks_results) == len(independent_task_names) - assert len(result.failed_tasks) == len(expected_failed) - assert result.errored_tasks == failing_task_names, ( - f"errored_tasks should be exactly the two tasks whose func raised, " - f"got {result.errored_tasks}" + assert name in result.successes, f"Independent task {name} missing from successes" + assert result.successes[name].result == f"payload_{name}" + assert result.successes[name].error is None + + assert len(result.successes) == len(independent_task_names) + assert len(result.errored) + len(result.skipped) == len(expected_failed) + assert set(result.errored) == failing_task_names, ( + f"errored entries should be exactly the two tasks whose func raised, " + f"got {set(result.errored)}" ) - assert result.skipped_tasks == skipped_task_names, ( - f"skipped_tasks should be the three cascade-skipped tasks, got {result.skipped_tasks}" + assert set(result.skipped) == skipped_task_names, ( + f"skipped entries should be the three cascade-skipped tasks, got {set(result.skipped)}" ) - assert result.errored_tasks.isdisjoint(result.skipped_tasks), ( - "errored_tasks and skipped_tasks must be disjoint" + assert set(result.errored).isdisjoint(result.skipped), ( + "errored and skipped entries must be disjoint" ) # OUTCOME #2 — Cascading Skip Control @@ -167,9 +164,9 @@ def make_task(name: str, deps, fail: bool = False) -> Task: assert call_counts[name] == 0, ( f"Skipped task {name} func was called {call_counts[name]} times — must be 0" ) - assert result.failed_tasks == expected_failed + assert set(result.errored) | set(result.skipped) == expected_failed for name in skipped_task_names: - assert name not in result.passed_tasks_results + assert name not in result.successes # OUTCOME #3 — Data-Driven Logger Payload for name in failing_task_names: diff --git a/tests/test_execution_report.py b/tests/test_execution_report.py new file mode 100644 index 0000000..a36c99c --- /dev/null +++ b/tests/test_execution_report.py @@ -0,0 +1,120 @@ +"""Tests for ``ProcessExecutionReport`` as returned by ``Process.run()``. + +Covers status classification (success / errored / skipped), the data carried +by each ``TaskReportEntry`` (result, error, elapsed_seconds, attempts), entry +ordering, and the ``successes``/``errored``/``skipped`` filter accessors — +for both sequential and parallel execution. +""" + +from __future__ import annotations + +from processes import ( + ErrorData, + Process, + Task, + TaskDependency, + TaskStatus, +) + +from .base_test import BaseTest + + +class TestProcessExecutionReport(BaseTest): + def _build_process(self) -> tuple[Process, Task, Task, Task]: + def load() -> str: + return "data" + + def apply(_data: str) -> None: + raise ValueError("apply failed") + + def notify() -> None: + pass + + load_task = Task("load", load, self._log("report_load.log")) + apply_task = Task( + "apply", + apply, + self._log("report_apply.log"), + dependencies=[TaskDependency("load", use_result_as_additional_args=True)], + ) + notify_task = Task( + "notify", + notify, + self._log("report_notify.log"), + dependencies=[TaskDependency("apply")], + ) + + process = Process([load_task, apply_task, notify_task]) + return process, load_task, apply_task, notify_task + + def test_sequential_report_classifies_each_task(self) -> None: + process, load_task, apply_task, notify_task = self._build_process() + with process: + report = process.run(parallel=False) + + assert list(report.entries) == ["load", "apply", "notify"] + + load_entry = report.entries["load"] + assert load_entry.status == TaskStatus.SUCCESS + assert load_entry.result == "data" + assert load_entry.error is None + assert load_entry.attempts == 1 + assert load_entry.elapsed_seconds >= 0.0 + assert load_entry.function == "load" + assert load_entry.args == () + assert load_entry.kwargs == {} + + apply_entry = report.entries["apply"] + assert apply_entry.status == TaskStatus.ERRORED + assert apply_entry.result is None + assert isinstance(apply_entry.error, ErrorData) + assert apply_entry.error.task_name == "apply" + assert apply_entry.error.exception == "apply failed" + assert apply_entry.attempts == 1 + + notify_entry = report.entries["notify"] + assert notify_entry.status == TaskStatus.SKIPPED + assert notify_entry.result is None + assert notify_entry.error is None + assert notify_entry.attempts == 0 + assert notify_entry.elapsed_seconds == 0.0 + + self._close_handlers(load_task, apply_task, notify_task) + + def test_filter_accessors_partition_entries(self) -> None: + process, load_task, apply_task, notify_task = self._build_process() + with process: + report = process.run(parallel=False) + + assert set(report.successes) == {"load"} + assert set(report.errored) == {"apply"} + assert set(report.skipped) == {"notify"} + + self._close_handlers(load_task, apply_task, notify_task) + + def test_parallel_report_classifies_each_task(self) -> None: + process, load_task, apply_task, notify_task = self._build_process() + with process: + report = process.run(parallel=True, max_workers=4) + + assert report.entries["load"].status == TaskStatus.SUCCESS + assert report.entries["apply"].status == TaskStatus.ERRORED + assert report.entries["notify"].status == TaskStatus.SKIPPED + + self._close_handlers(load_task, apply_task, notify_task) + + def test_all_success_report_has_no_errored_or_skipped(self) -> None: + def step() -> int: + return 42 + + task = Task("step", step, self._log("report_all_success.log")) + process = Process([task]) + with process: + report = process.run(parallel=False) + + assert report.entries["step"].status == TaskStatus.SUCCESS + assert report.entries["step"].result == 42 + assert report.errored == {} + assert report.skipped == {} + + self._close_handlers(task) diff --git a/tests/test_normal_run_no_errors.py b/tests/test_normal_run_no_errors.py index 7fd3e58..95fe983 100644 --- a/tests/test_normal_run_no_errors.py +++ b/tests/test_normal_run_no_errors.py @@ -103,14 +103,12 @@ def _build_dependent_task_graph(self) -> list[Task]: def test_run_dependent_tasks_sequential(self) -> None: with Process(self._build_dependent_task_graph()) as process: t_start = time.time() - process_result = process.run(parallel=False) + report = process.run(parallel=False) t_end = time.time() - assert len(process_result.passed_tasks_results) == 6, ( - f"Expected 6 passed tasks. Got {len(process_result.passed_tasks_results)}" - ) - assert len(process_result.failed_tasks) == 0, ( - f"Expected 0 failed tasks. Got {len(process_result.failed_tasks)}" + assert len(report.successes) == 6, f"Expected 6 passed tasks. Got {len(report.successes)}" + assert len(report.errored) + len(report.skipped) == 0, ( + f"Expected 0 failed tasks. Got {len(report.errored) + len(report.skipped)}" ) elapsed = t_end - t_start assert 4.0 <= elapsed < 4.8, f"Sequential run took {elapsed} seconds. Expected ~4 seconds." @@ -121,14 +119,12 @@ def test_run_dependent_tasks_parallel(self) -> None: n_workers = os.cpu_count() with Process(self._build_dependent_task_graph()) as process: t_start = time.time() - process_result = process.run(parallel=True, max_workers=n_workers) + report = process.run(parallel=True, max_workers=n_workers) t_end = time.time() - assert len(process_result.passed_tasks_results) == 6, ( - f"Expected 6 passed tasks. Got {len(process_result.passed_tasks_results)}" - ) - assert len(process_result.failed_tasks) == 0, ( - f"Expected 0 failed tasks. Got {len(process_result.failed_tasks)}" + assert len(report.successes) == 6, f"Expected 6 passed tasks. Got {len(report.successes)}" + assert len(report.errored) + len(report.skipped) == 0, ( + f"Expected 0 failed tasks. Got {len(report.errored) + len(report.skipped)}" ) if n_workers and n_workers > 2: assert int(round(t_end - t_start, 0)) == 2, ( diff --git a/tests/test_parallel_race_conditions.py b/tests/test_parallel_race_conditions.py index 5e1e7be..a4f25a1 100644 --- a/tests/test_parallel_race_conditions.py +++ b/tests/test_parallel_race_conditions.py @@ -2,7 +2,7 @@ Parallel-execution race-condition stress tests. Two scenarios designed to maximise contention on ``ProcessRunner``'s -shared state (``passed_results``, ``failed_tasks``, ``submitted_tasks``): +shared state (``results``, ``submitted_tasks``): 1. ``test_diamond_fan_in_race`` — wide fan-in diamond. Many sibling tasks gate on a ``threading.Barrier`` and release simultaneously, @@ -100,10 +100,11 @@ def collector(*sibling_results): f"{tag} collector received {call_counts['__collector_arg_count']} " f"args, expected {sibling_count}" ) - assert not result.failed_tasks, f"{tag} unexpected failures: {result.failed_tasks}" - assert len(result.passed_tasks_results) == sibling_count + 2, f"{tag} passed count mismatch" + failed = set(result.errored) | set(result.skipped) + assert not failed, f"{tag} unexpected failures: {failed}" + assert len(result.successes) == sibling_count + 2, f"{tag} passed count mismatch" - stored = result.passed_tasks_results["collector"].result + stored = result.successes["collector"].result assert stored == sorted(f"sibling_{i}_payload" for i in range(sibling_count)), ( f"{tag} collector result missing sibling payloads: {stored}" ) @@ -201,23 +202,23 @@ def _child(*_args, **_kwargs): expected_failed = {n for chain in failing_chains for n in chain} expected_passed = {n for chain in passing_chains for n in chain} - assert result.failed_tasks == expected_failed, ( + failed = set(result.errored) | set(result.skipped) + passed = set(result.successes) + + assert failed == expected_failed, ( f"{tag} failed set mismatch: " - f"missing={expected_failed - result.failed_tasks}, " - f"extra={result.failed_tasks - expected_failed}" + f"missing={expected_failed - failed}, " + f"extra={failed - expected_failed}" ) - assert set(result.passed_tasks_results) == expected_passed, ( + assert passed == expected_passed, ( f"{tag} passed set mismatch: " - f"missing={expected_passed - set(result.passed_tasks_results)}, " - f"extra={set(result.passed_tasks_results) - expected_passed}" - ) - assert not (result.failed_tasks & set(result.passed_tasks_results)), ( - f"{tag} a task appears in both passed and failed sets" + f"missing={expected_passed - passed}, " + f"extra={passed - expected_passed}" ) - assert len(result.failed_tasks) + len(result.passed_tasks_results) == len(tasks), ( + assert not (failed & passed), f"{tag} a task appears in both passed and failed sets" + assert len(failed) + len(passed) == len(tasks), ( f"{tag} task accounting drifted: " - f"{len(result.failed_tasks)} failed + " - f"{len(result.passed_tasks_results)} passed != {len(tasks)} total" + f"{len(failed)} failed + {len(passed)} passed != {len(tasks)} total" ) diff --git a/tests/test_resolve_args_failure.py b/tests/test_resolve_args_failure.py new file mode 100644 index 0000000..9b9b0e9 --- /dev/null +++ b/tests/test_resolve_args_failure.py @@ -0,0 +1,52 @@ +"""``Task.run`` must never propagate. + +A failure while resolving dependency arguments (before any attempt runs) has to +become an ``ERRORED`` result, just like a failure inside the task's function. +This previously crashed sequential execution while parallel execution swallowed +it, an asymmetry these tests pin down. +""" + +from __future__ import annotations + +from typing import Any + +from processes import Process, Task, TaskDependency, TaskStatus + +from .base_test import BaseTest + + +def _boom(_executing_process: Any) -> Any: + raise RuntimeError("resolve blew up") + + +class TestResolveArgsFailure(BaseTest): + def test_run_wraps_resolution_failure(self) -> None: + """A failure in ``_resolve_args`` yields ``ERRORED`` (attempts=0), not a raise.""" + task = Task("t", lambda: 1, self._log("resolve.log")) + task._resolve_args = _boom # type: ignore[method-assign] + try: + result = task.run(executing_process=None) + finally: + self._close_handlers(task) + + assert not result.worked + assert result.status == TaskStatus.ERRORED + assert isinstance(result.exception, RuntimeError) + assert result.attempts == 0 + + def test_sequential_process_survives_resolution_failure(self) -> None: + """Sequential ``Process.run`` does not crash when a task's args fail to resolve.""" + producer = Task("producer", lambda: 1, self._log("producer.log")) + consumer = Task( + "consumer", + lambda x: x, + self._log("consumer.log"), + dependencies=[TaskDependency("producer", use_result_as_additional_args=True)], + ) + consumer._resolve_args = _boom # type: ignore[method-assign] + + with Process([producer, consumer]) as process: + report = process.run(parallel=False) + + assert report.entries["producer"].status == TaskStatus.SUCCESS + assert report.entries["consumer"].status == TaskStatus.ERRORED diff --git a/tests/test_timeout_retry.py b/tests/test_timeout_retry.py index e9f266b..7713714 100644 --- a/tests/test_timeout_retry.py +++ b/tests/test_timeout_retry.py @@ -44,6 +44,8 @@ def fast() -> int: assert result.worked assert result.result == 42 assert result.exception is None + assert result.attempts == 1 + assert result.elapsed_seconds >= 0.0 def test_timeout_in_sequential_process(self) -> None: """Timeout propagates through Process.run(parallel=False).""" @@ -54,13 +56,10 @@ def slow() -> str: task = Task("t_seq", slow, self._log("seq.log"), timeout=0.05) with Process([task]) as process: - pr = process.run(parallel=False) + report = process.run(parallel=False) - assert "t_seq" in pr.failed_tasks - assert "t_seq" in pr.errored_tasks - assert isinstance( - pr.passed_tasks_results.get("t_seq", None) or pr.failed_tasks, set - ) # task did not pass + assert "t_seq" in report.errored + assert "t_seq" not in report.successes def test_timeout_in_parallel_process(self) -> None: """Timeout propagates through Process.run(parallel=True).""" @@ -71,10 +70,10 @@ def slow() -> str: task = Task("t_par", slow, self._log("par.log"), timeout=0.05) with Process([task]) as process: - pr = process.run(parallel=True) + report = process.run(parallel=True) - assert "t_par" in pr.failed_tasks - assert "t_par" in pr.errored_tasks + assert "t_par" in report.errored + assert "t_par" not in report.successes def test_timeout_cascades_to_dependants(self) -> None: """A timed-out task causes its dependants to be cascade-skipped.""" @@ -94,11 +93,11 @@ def child(x: str) -> str: dependencies=[TaskDependency("root", use_result_as_additional_args=True)], ) with Process([t_root, t_child]) as process: - pr = process.run(parallel=False) + report = process.run(parallel=False) - assert "root" in pr.errored_tasks - assert "child" in pr.skipped_tasks - assert "child" not in pr.passed_tasks_results + assert "root" in report.errored + assert "child" in report.skipped + assert "child" not in report.successes # --- validation --- @@ -135,6 +134,7 @@ def flaky() -> str: assert result.worked assert result.result == "ok" assert len(calls) == 2 + assert result.attempts == 2 def test_retry_exhausted_returns_failed(self) -> None: """A task that fails all attempts returns TaskResult(False, ...).""" @@ -153,6 +153,8 @@ def always_fails() -> None: assert not result.worked assert isinstance(result.exception, ConnectionError) assert len(calls) == 3 # 1 original + 2 retries + assert result.attempts == 3 + assert result.elapsed_seconds >= 0.0 def test_retry_zero_means_no_retry(self) -> None: """retries=0 (default) never retries, even for a retryable exception.""" @@ -262,10 +264,10 @@ def flaky() -> str: task = Task("proc_retry", flaky, self._log("pr.log"), retries=1) with Process([task]) as process: - pr = process.run(parallel=False) + report = process.run(parallel=False) - assert "proc_retry" in pr.passed_tasks_results - assert pr.passed_tasks_results["proc_retry"].result == "ok" + assert "proc_retry" in report.successes + assert report.successes["proc_retry"].result == "ok" assert len(calls) == 2 # --- validation ---