From 97fbe0883ea8f6b753786e43c6ad5d140aefaede Mon Sep 17 00:00:00 2001 From: Guillaume Onfroy Date: Thu, 17 Sep 2026 18:05:46 +0200 Subject: [PATCH 1/3] feat(core): retry a failed operation in place An operation carrying a `RetryPolicy` gets another attempt whenever it raises an error it calls retryable and its budget allows one: the failure is recorded as retried, the backoff is slept, and the same node executes again. An operation carrying none is attempted once, so nothing changes until something declares a policy. The policy is declared on the component whose unit it governs, which is what keeps one budget from meaning three things: an operation's retries its own execution, a source's is the default its assets inherit through `_resolve` alongside `dataset` and `normalizer`, and a job's is for its runs, which the platform will read in phase 2. Numbers live in the policy; whether an error is worth another attempt is behaviour, and lives on `Operation.retryable`. Only an exhausted or declined failure marks a node failed, so a node still working through its attempts cancels no dependents and does not trip `fail_fast`. `OPERATION_RETRIED` records the attempts that were retried, and `OPERATION_FAILED` keeps meaning exhausted, which is the same rule the hooks will follow one level up. Attempts are their own event rows: the deterministic id gains the attempt, left out when it is 1 so every id written before retries existed is unchanged. The multi-process worker owns its loop because only it sees the failures, and none of the reporting: the errors it retried travel back with the outcome and the parent replays them, so events and attempt counters match what happened in the child. By Digitl --- .../src/interloper/__init__.py | 2 + .../src/interloper/events/console.py | 1 + .../src/interloper/events/types.py | 1 + .../src/interloper/job/base.py | 2 + .../src/interloper/operation/base.py | 19 +++ .../src/interloper/retry/__init__.py | 5 + .../src/interloper/retry/base.py | 85 ++++++++++++++ .../src/interloper/runner/async_runner.py | 44 +++++-- .../src/interloper/runner/multi_process.py | 59 +++++++--- .../src/interloper/runner/state.py | 57 +++++++-- .../src/interloper/source/base.py | 5 + .../interloper-core/tests/job/test_base.py | 21 +++- .../tests/operation/test_base.py | 17 +++ .../interloper-core/tests/retry/__init__.py | 0 .../interloper-core/tests/retry/test_base.py | 45 +++++++ .../tests/runner/test_async_runner.py | 110 ++++++++++++++++++ .../tests/runner/test_multi_process.py | 89 ++++++++++++-- .../tests/runner/test_state.py | 48 ++++++++ .../tests/runner/test_state_event_ids.py | 16 +++ .../interloper-core/tests/source/test_base.py | 33 ++++++ 20 files changed, 611 insertions(+), 48 deletions(-) create mode 100644 packages/interloper-core/src/interloper/retry/__init__.py create mode 100644 packages/interloper-core/src/interloper/retry/base.py create mode 100644 packages/interloper-core/tests/retry/__init__.py create mode 100644 packages/interloper-core/tests/retry/test_base.py diff --git a/packages/interloper-core/src/interloper/__init__.py b/packages/interloper-core/src/interloper/__init__.py index 4abb56dc..c073028b 100644 --- a/packages/interloper-core/src/interloper/__init__.py +++ b/packages/interloper-core/src/interloper/__init__.py @@ -72,6 +72,7 @@ RESTClient, SinglePagePaginator, ) +from interloper.retry import RetryPolicy from interloper.runner import AsyncRunner, MultiProcessRunner, Runner, RunResult, SerialRunner from interloper.schema import Schema, schema from interloper.serializable import Serializable, Spec @@ -143,6 +144,7 @@ "Renewal", "Resource", "ResourceDefinition", + "RetryPolicy", "RunResult", "Runner", "Schema", diff --git a/packages/interloper-core/src/interloper/events/console.py b/packages/interloper-core/src/interloper/events/console.py index b0e279ed..292cfb5e 100644 --- a/packages/interloper-core/src/interloper/events/console.py +++ b/packages/interloper-core/src/interloper/events/console.py @@ -23,6 +23,7 @@ _EVENT_LEVELS: dict[EventType, int] = { EventType.RUN_FAILED: logging.ERROR, EventType.OPERATION_FAILED: logging.ERROR, + EventType.OPERATION_RETRIED: logging.WARNING, EventType.BACKFILL_FAILED: logging.ERROR, EventType.OPERATION_CANCELED: logging.WARNING, EventType.OPERATION_QUEUED: logging.DEBUG, diff --git a/packages/interloper-core/src/interloper/events/types.py b/packages/interloper-core/src/interloper/events/types.py index 9bc491b8..8b67c2ca 100644 --- a/packages/interloper-core/src/interloper/events/types.py +++ b/packages/interloper-core/src/interloper/events/types.py @@ -26,6 +26,7 @@ class EventType(Enum): OPERATION_STARTED = "operation_started" OPERATION_COMPLETED = "operation_completed" OPERATION_FAILED = "operation_failed" + OPERATION_RETRIED = "operation_retried" OPERATION_CANCELED = "operation_canceled" # Asset data (the data() call) diff --git a/packages/interloper-core/src/interloper/job/base.py b/packages/interloper-core/src/interloper/job/base.py index 58a8442d..0f96d367 100644 --- a/packages/interloper-core/src/interloper/job/base.py +++ b/packages/interloper-core/src/interloper/job/base.py @@ -8,6 +8,7 @@ from interloper.component import Component, Relation from interloper.operation import Operation, Workload +from interloper.retry import RetryPolicy if TYPE_CHECKING: from interloper.asset.base import Asset @@ -52,6 +53,7 @@ class Job(Component, Workload): destinations: list[Destination] = Relation("destination", many=True, optional=True) enabled: bool = Field(default=True, description="Job will run on the configured schedule") + retry: RetryPolicy | None = Field(default=None, description="Attempt budget for this job's runs") tags: list[str] = Field(default_factory=list) def operations(self) -> list[Operation]: diff --git a/packages/interloper-core/src/interloper/operation/base.py b/packages/interloper-core/src/interloper/operation/base.py index 49d2547a..630045ca 100644 --- a/packages/interloper-core/src/interloper/operation/base.py +++ b/packages/interloper-core/src/interloper/operation/base.py @@ -29,6 +29,7 @@ from interloper.component.base import Component from interloper.errors import format_exception from interloper.partitioning.base import PartitionConfig +from interloper.retry import RetryPolicy if TYPE_CHECKING: from interloper.component.relation import Relation @@ -119,6 +120,7 @@ class Operation(Component, Workload): partitioning: ClassVar[PartitionConfig | None] = None enabled: bool = Field(default=True, description="Operation will execute") + retry: RetryPolicy | None = Field(default=None, description="Attempt budget for this operation's execution") def operations(self) -> list[Operation]: """An operation is trivially its own workload. @@ -202,6 +204,23 @@ async def execute(self, context: OperationContext) -> OperationResult: The effects to persist (often none). """ + def retryable(self, error: Exception) -> bool: + """Whether another attempt at this operation is worth making. + + Consulted by the runner before it spends an attempt from + :attr:`retry`. Override to recognise a permanent error, such as a + vendor rejecting a request it will reject identically every time. The + default is permissive: the budget, not the classifier, is what bounds + waste. + + Args: + error: The exception :meth:`execute` raised. + + Returns: + ``True`` when the failure may be transient. + """ + return True + def failure(self, error: Exception) -> OperationResult: """Describe a failed execution in terms the platform can persist. diff --git a/packages/interloper-core/src/interloper/retry/__init__.py b/packages/interloper-core/src/interloper/retry/__init__.py new file mode 100644 index 00000000..5a819814 --- /dev/null +++ b/packages/interloper-core/src/interloper/retry/__init__.py @@ -0,0 +1,5 @@ +"""Retry policies.""" + +from interloper.retry.base import RetryPolicy + +__all__ = ["RetryPolicy"] diff --git a/packages/interloper-core/src/interloper/retry/base.py b/packages/interloper-core/src/interloper/retry/base.py new file mode 100644 index 00000000..c5a23c12 --- /dev/null +++ b/packages/interloper-core/src/interloper/retry/base.py @@ -0,0 +1,85 @@ +"""Retry: an attempt budget for a unit of work.""" + +from __future__ import annotations + +import random + +from pydantic import BaseModel, Field + + +class RetryPolicy(BaseModel): + """How many times a unit of work is attempted, and how long between attempts. + + Declared on the component whose unit it governs: an operation's policy + retries that operation's execution, a source's is the default for its + assets, a job's retries that job's runs. A component declaring none is + attempted once; there is no instance-wide default. + + The policy carries numbers only. Whether a given error is worth another + attempt is behaviour rather than configuration, and lives on + :meth:`~interloper.operation.base.Operation.retryable`. + """ + + max_attempts: int = Field( + default=3, + ge=1, + title="Max attempts", + description="Total attempts, including the first", + ) + delay: float = Field( + default=5.0, + ge=0, + title="Delay", + description="Seconds to wait before the second attempt", + ) + backoff: float = Field( + default=2.0, + ge=1, + title="Backoff", + description="Multiplier applied to the delay for each further attempt", + ) + max_delay: float = Field( + default=3600.0, + ge=0, + title="Max delay", + description="Upper bound on any single delay, in seconds", + ) + jitter: float = Field( + default=0.1, + ge=0, + le=1, + title="Jitter", + description="Fraction of the delay spread randomly around it, so attempts do not align", + ) + + def allows(self, attempt: int) -> bool: + """Whether the budget covers an attempt. + + Args: + attempt: The attempt number, counting the first execution as 1. + + Returns: + ``True`` while the attempt is within the budget. + """ + return attempt <= self.max_attempts + + def delay_before(self, attempt: int) -> float: + """How long to wait before an attempt. + + The growth is geometric from the second attempt (the first retry), + capped at ``max_delay``, then spread by ``jitter`` so that operations + failing together do not retry in lockstep. + + Args: + attempt: The attempt number, counting the first execution as 1. + + Returns: + The delay in seconds, never negative; ``0.0`` for the first attempt. + """ + if attempt <= 1: + return 0.0 + delay = min(self.delay * self.backoff ** (attempt - 2), self.max_delay) + if not self.jitter: + return delay + spread = delay * self.jitter + return max(0.0, random.uniform(delay - spread, delay + spread)) diff --git a/packages/interloper-core/src/interloper/runner/async_runner.py b/packages/interloper-core/src/interloper/runner/async_runner.py index b90fe21d..c1dba88b 100644 --- a/packages/interloper-core/src/interloper/runner/async_runner.py +++ b/packages/interloper-core/src/interloper/runner/async_runner.py @@ -168,7 +168,18 @@ async def _execute_operation( operation: Operation, partition_or_window: Partition | PartitionWindow | None = None, ) -> Any: - """Execute a single operation with state tracking. + """Execute a single operation with state tracking, retrying in place. + + An operation carrying a :class:`~interloper.retry.base.RetryPolicy` + gets another attempt whenever it raises an error it calls retryable + and its budget allows one: the failure is recorded as retried, the + backoff is slept, and the same node executes again. An operation + carrying no policy is attempted once. + + Only an exhausted or declined failure marks the node failed, which is + what cancels its dependents and, under ``fail_fast``, stops the walk. + A node still working through its attempts has not failed, so nothing + downstream reacts to it. On failure the operation's own :meth:`~Operation.failure` hook curates the recorded message and effects, and the traceback is @@ -183,8 +194,6 @@ async def _execute_operation( Returns: The execution's effects, or ``None`` if the operation failed. """ - self.state.mark_running(operation) - effective_partition = operation.effective_partition(partition_or_window) span_attrs = attributes.from_metadata( operation._event_metadata(self.state.metadata, effective_partition) @@ -194,16 +203,27 @@ async def _execute_operation( dag=self.state.dag, metadata=self.state.metadata, ) - try: - with tracer().start_as_current_span("interloper.operation.execute", attributes=span_attrs): - result = await operation.execute(context) + policy = operation.retry + + while True: + self.state.mark_running(operation) + try: + with tracer().start_as_current_span("interloper.operation.execute", attributes=span_attrs): + result = await operation.execute(context) + except Exception as e: # noqa: BLE001 — every failure becomes the node's record + attempt = self.state.attempts[operation.id] + if policy is not None and policy.allows(attempt + 1) and operation.retryable(e): + self.state.mark_retried(operation, format_exception(e)) + await asyncio.sleep(policy.delay_before(attempt + 1)) + continue + failed = operation.failure(e) + tb = traceback.format_exc() if type(operation).capture_traceback else None + self.state.mark_failed( + operation, failed.error or format_exception(e), tb=tb, effects=failed, exception=e + ) + return None self.state.mark_completed(operation, effects=result) - except Exception as e: # noqa: BLE001 — every failure becomes the node's record - failed = operation.failure(e) - tb = traceback.format_exc() if type(operation).capture_traceback else None - self.state.mark_failed(operation, failed.error or format_exception(e), tb=tb, effects=failed, exception=e) - return None - return result + return result async def _flush(self, inflight: dict[asyncio.Task[Any], Operation]) -> None: """Let the in-flight tasks finish when the walk ends. diff --git a/packages/interloper-core/src/interloper/runner/multi_process.py b/packages/interloper-core/src/interloper/runner/multi_process.py index 21fa8a13..5c383cdd 100644 --- a/packages/interloper-core/src/interloper/runner/multi_process.py +++ b/packages/interloper-core/src/interloper/runner/multi_process.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio +import time import traceback from concurrent.futures import Future, ProcessPoolExecutor from typing import Any @@ -20,7 +21,7 @@ def _worker( dag_spec: dict[str, Any], partition_or_window: Partition | PartitionWindow | None, metadata: dict[str, Any], -) -> tuple[str, bool, str | None, str | None, dict[str, Any]]: +) -> tuple[str, bool, str | None, str | None, dict[str, Any], list[str]]: """Execute a single operation in a worker process. Reconstructs the DAG from its serialized spec, looks up the target @@ -34,8 +35,14 @@ def _worker( to the node's own effective partition. metadata: Run metadata, also carrying the parent span context. + The worker owns the attempt loop because only it sees the failures, and + none of the reporting: the parent holds the ``RunState`` and emits every + event, so the errors of the attempts that were retried travel back with + the outcome for it to replay. + Returns: - Tuple of ``(id, success, error_message, formatted_traceback, effects)``. + Tuple of ``(id, success, error_message, formatted_traceback, effects, + retried_errors)``. """ from opentelemetry import context as otel_context @@ -50,31 +57,40 @@ def _worker( token = otel_context.attach(context) if context is not None else None operation: Operation | None = None + retried: list[str] = [] try: dag = DAGSpec(**dag_spec).reconstruct() operation = dag.operation_map[operation_id] - result = asyncio.run( - operation.execute( - OperationContext( - partition_or_window=operation.effective_partition(partition_or_window), - dag=dag, - metadata=metadata, - ) - ) + context = OperationContext( + partition_or_window=operation.effective_partition(partition_or_window), + dag=dag, + metadata=metadata, ) + policy = operation.retry + attempt = 1 + while True: + try: + result = asyncio.run(operation.execute(context)) + break + except Exception as e: + if policy is None or not policy.allows(attempt + 1) or not operation.retryable(e): + raise + retried.append(format_exception(e)) + time.sleep(policy.delay_before(attempt + 1)) + attempt += 1 except Exception as e: # noqa: BLE001 if operation is None: - return (operation_id, False, format_exception(e), traceback.format_exc(), {}) + return (operation_id, False, format_exception(e), traceback.format_exc(), {}, retried) failed = operation.failure(e) tb = traceback.format_exc() if type(operation).capture_traceback else None effects = {"config": failed.config, "state": failed.state} - return (operation_id, False, failed.error or format_exception(e), tb, effects) + return (operation_id, False, failed.error or format_exception(e), tb, effects, retried) finally: if token is not None: otel_context.detach(token) # Pool workers are reused; exit hooks may never run. force_flush() - return (operation_id, True, None, None, {"config": result.config, "state": result.state}) + return (operation_id, True, None, None, {"config": result.config, "state": result.state}, retried) class MultiProcessRunner(SyncRunner): @@ -158,8 +174,9 @@ def _handle_completed(self, future: Future[Any], operation: Operation) -> None: """Process a completed future from a worker process. Unlike the base ``_handle_completed``, this interprets the - ``(id, success, error_msg, tb, effects)`` tuple returned by - ``_worker``. + ``(id, success, error_msg, tb, effects, retried_errors)`` tuple + returned by ``_worker``, replaying the attempts the worker retried so + the run's events and attempt counters match what actually happened. Args: future: The finished future returned by ``_submit_operation``. @@ -168,11 +185,14 @@ def _handle_completed(self, future: Future[Any], operation: Operation) -> None: self._futures.pop(future, None) try: - _key, success, error_message, tb, effects = future.result() + _key, success, error_message, tb, effects, retried = future.result() except Exception as e: # noqa: BLE001 — every failure becomes the node's record self.state.mark_failed(operation, format_exception(e), tb=traceback.format_exc(), exception=e) return + for error in retried: + self.state.mark_retried(operation, error) + if success: self.state.mark_completed(operation, effects=OperationResult(**effects)) else: @@ -184,18 +204,21 @@ def _handle_completed(self, future: Future[Any], operation: Operation) -> None: ) def _handle_flushed(self, future: Future[Any], operation: Operation) -> None: - """Interpret the worker ``(id, success, error_message, tb, effects)`` tuple during flush. + """Interpret the worker ``(id, success, error_message, tb, effects, retried_errors)`` tuple during flush. Args: future: The finished future to interpret. operation: The operation the future was submitted for. """ try: - _key, success, error_message, tb, effects = future.result() + _key, success, error_message, tb, effects, retried = future.result() except Exception as e: # noqa: BLE001 self.state.mark_failed(operation, format_exception(e), tb=traceback.format_exc()) return + for error in retried: + self.state.mark_retried(operation, error) + if success: self.state.mark_completed(operation, effects=OperationResult(**effects)) else: diff --git a/packages/interloper-core/src/interloper/runner/state.py b/packages/interloper-core/src/interloper/runner/state.py index 6fb31efb..0d63293a 100644 --- a/packages/interloper-core/src/interloper/runner/state.py +++ b/packages/interloper-core/src/interloper/runner/state.py @@ -53,6 +53,7 @@ def __init__( self.metadata["run_id"] = str(uuid.uuid4()) self.executions: dict[str, ExecutionInfo] = {} + self.attempts: dict[str, int] = {operation.id: 1 for operation in dag.operations} self.partition_or_window: Partition | PartitionWindow | None = None self.start_time: dt.datetime | None = None self.end_time: dt.datetime | None = None @@ -317,6 +318,32 @@ def mark_failed( }, ) + def mark_retried(self, operation: Operation, error: str, *, emit: bool = True) -> None: + """Record a failed attempt that will be retried, and open the next one. + + A retried attempt is not a verdict: the execution keeps its current + status, nothing downstream is canceled, and ``OPERATION_FAILED`` stays + reserved for an exhausted budget. Advancing the counter is what makes + the next attempt's events distinct from this one's. + + Args: + operation: The operation whose attempt failed. + error: Error message describing the failed attempt. + emit: Emit ``OPERATION_RETRIED`` on the EventBus. Set to ``False`` + for cross-process runners where the child emits its own events. + """ + attempt = self.attempts[operation.id] + if emit: + self._emit_operation_event( + EventType.OPERATION_RETRIED, + { + **self._operation_event_metadata(operation), + "error": error, + "message": f"Operation '{operation.key}' failed on attempt {attempt}, retrying: {error}", + }, + ) + self.attempts[operation.id] = attempt + 1 + # -- Internals ------------------------------------------------------------- def _initialize_operations(self) -> None: @@ -364,13 +391,14 @@ def _operation_event_metadata(self, operation: Operation) -> dict[str, Any]: "component_kind": operation.kind, "component_key": operation.key, "partition_or_window": str(self.partition_or_window) if self.partition_or_window else None, + "attempt": self.attempts[operation.id], } if operation.parent is not None: meta["parent_id"] = operation.parent.id return meta @staticmethod - def _operation_event_id(run_id: str, component_id: str, event_type: EventType) -> str: + def _operation_event_id(run_id: str, component_id: str, event_type: EventType, attempt: int = 1) -> str: """Derive a deterministic event id from a run/component/type triple. Both the host and the in-container child run this same code with the @@ -382,29 +410,42 @@ def _operation_event_id(run_id: str, component_id: str, event_type: EventType) - run_id: Id of the run the event belongs to. component_id: Id of the operation the event is about. event_type: The operation-lifecycle event type. + attempt: The attempt the event belongs to, so a retried node's + events are their own rows rather than a dedup of the first + attempt's. Left out of the key when it is 1, which keeps every + id written before retries existed unchanged. Returns: A stable UUID5 string for the event. """ - return str(uuid.uuid5(RunState._OPERATION_EVENT_NS, f"{run_id}:{component_id}:{event_type.value}")) + key = f"{run_id}:{component_id}:{event_type.value}" + if attempt > 1: + key = f"{key}:{attempt}" + return str(uuid.uuid5(RunState._OPERATION_EVENT_NS, key)) def _emit_operation_event(self, event_type: EventType, metadata: dict[str, Any]) -> None: """Emit an operation-lifecycle event with a deterministic id. - The id is derived from ``(run_id, component_id, event_type)`` so the - same logical event dedups across producers (host fallback vs child, or - the duplicate ``operation_queued``). ``metadata`` must carry - ``component_id``. + The id is derived from ``(run_id, component_id, event_type, attempt)`` + so the same logical event dedups across producers (host fallback vs + child, or the duplicate ``operation_queued``) while a retried node's + attempts stay distinct. ``metadata`` must carry ``component_id``. Args: event_type: The operation-lifecycle event type to emit. metadata: Event metadata, as built by ``_operation_event_metadata``; - must carry a ``component_id`` key. + must carry a ``component_id`` key, and an ``attempt`` for any + node past its first. """ event = Event( type=event_type, metadata=metadata, - id=self._operation_event_id(self.run_id, str(metadata["component_id"]), event_type), + id=self._operation_event_id( + self.run_id, + str(metadata["component_id"]), + event_type, + attempt=int(metadata.get("attempt", 1)), + ), ) EventBus.emit_event(event) diff --git a/packages/interloper-core/src/interloper/source/base.py b/packages/interloper-core/src/interloper/source/base.py index e676362e..25638b61 100644 --- a/packages/interloper-core/src/interloper/source/base.py +++ b/packages/interloper-core/src/interloper/source/base.py @@ -14,6 +14,7 @@ from interloper.normalizer import MaterializationStrategy, Normalizer from interloper.operation import Operation, Workload from interloper.resource.fields import InputField, SelectField +from interloper.retry import RetryPolicy from interloper.serializable import IgnoredDescriptor from interloper.utils.text import validate_key @@ -115,6 +116,8 @@ class MySource(Source): tags: ClassVar[list[str]] = [] internal_fields: ClassVar[frozenset[str]] = frozenset({"assets", "normalizer", "select"}) + retry: RetryPolicy | None = Field(default=None, description="Default attempt budget for this source's assets") + destinations: list[Destination] = Relation("destination", many=True, optional=True) # State @@ -559,6 +562,8 @@ def _resolve(self) -> None: asset.default_destination_key = self.default_destination_key if asset.normalizer is None and self.normalizer is not None: asset.normalizer = self.normalizer + if asset.retry is None and self.retry is not None: + asset.retry = self.retry if ( self.materialization_strategy is not None and asset.materialization_strategy is MaterializationStrategy.RECONCILE diff --git a/packages/interloper-core/tests/job/test_base.py b/packages/interloper-core/tests/job/test_base.py index 83bad9ab..bfd05efe 100644 --- a/packages/interloper-core/tests/job/test_base.py +++ b/packages/interloper-core/tests/job/test_base.py @@ -36,6 +36,15 @@ class FakeOtherJobDestination(FakeJobDestination): """Second destination class, to tell a cascaded binding from an own one.""" +class TestRetry: + def test_a_job_carries_a_retry_policy(self): + policy = il.RetryPolicy(max_attempts=2) + assert il.Job(retry=policy).retry == policy + + def test_a_job_declares_no_policy_by_default(self): + assert il.Job().retry is None + + class TestDefinition: """Class-level identity and defaults.""" @@ -45,14 +54,22 @@ def test_kind_and_key(self): def test_definition_self_describes(self): defn = il.CronJob.definition() - assert set(defn.config_schema["properties"]) == {"cron", "timezone", "enabled", "tags", "lookback", "offset"} + assert set(defn.config_schema["properties"]) == { + "cron", + "timezone", + "enabled", + "retry", + "tags", + "lookback", + "offset", + } assert "cron" in defn.config_schema.get("required", []) assert defn.relations["targets"].kinds == ["source", "asset"] assert defn.relations["targets"].many is True def test_anchor_carries_the_workload_only(self): defn = il.Job.definition() - assert set(defn.config_schema["properties"]) == {"enabled", "tags"} + assert set(defn.config_schema["properties"]) == {"enabled", "retry", "tags"} assert il.CronJob.kind == "job" def test_defaults(self): diff --git a/packages/interloper-core/tests/operation/test_base.py b/packages/interloper-core/tests/operation/test_base.py index 79d92b6a..7854f6b3 100644 --- a/packages/interloper-core/tests/operation/test_base.py +++ b/packages/interloper-core/tests/operation/test_base.py @@ -57,6 +57,23 @@ def test_enabled_is_a_field(self): assert _NoopOperation().enabled is True assert _NoopOperation(enabled=False).enabled is False + def test_no_retry_policy_by_default(self): + assert _NoopOperation().retry is None + + def test_every_error_is_retryable_by_default(self): + assert _NoopOperation().retryable(ValueError("boom")) is True + + def test_retryable_can_be_narrowed(self): + class Picky(Operation): + async def execute(self, context: OperationContext) -> OperationResult: + return OperationResult() + + def retryable(self, error: Exception) -> bool: + return not isinstance(error, ValueError) + + assert Picky().retryable(TypeError()) is True + assert Picky().retryable(ValueError()) is False + def test_node_protocol_defaults(self): operation = _NoopOperation() assert operation.enabled is True diff --git a/packages/interloper-core/tests/retry/__init__.py b/packages/interloper-core/tests/retry/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/packages/interloper-core/tests/retry/test_base.py b/packages/interloper-core/tests/retry/test_base.py new file mode 100644 index 00000000..120259b0 --- /dev/null +++ b/packages/interloper-core/tests/retry/test_base.py @@ -0,0 +1,45 @@ +"""Tests for ``interloper.retry.base``.""" + +from __future__ import annotations + +import pytest +from pydantic import ValidationError + +from interloper.retry import RetryPolicy + + +class TestBudget: + def test_allows_up_to_max_attempts(self): + policy = RetryPolicy(max_attempts=3) + assert policy.allows(1) + assert policy.allows(3) + assert not policy.allows(4) + + def test_max_attempts_is_at_least_one(self): + with pytest.raises(ValidationError): + RetryPolicy(max_attempts=0) + + +class TestDelay: + def test_first_attempt_has_no_delay(self): + assert RetryPolicy(delay=10.0).delay_before(1) == 0.0 + + def test_delay_grows_geometrically(self): + policy = RetryPolicy(delay=10.0, backoff=2.0, jitter=0.0) + assert policy.delay_before(2) == 10.0 + assert policy.delay_before(3) == 20.0 + assert policy.delay_before(4) == 40.0 + + def test_delay_is_capped(self): + policy = RetryPolicy(delay=10.0, backoff=10.0, max_delay=50.0, jitter=0.0) + assert policy.delay_before(4) == 50.0 + + def test_jitter_stays_within_its_fraction(self): + policy = RetryPolicy(delay=10.0, backoff=1.0, jitter=0.5) + delays = [policy.delay_before(2) for _ in range(200)] + assert all(5.0 <= delay <= 15.0 for delay in delays) + assert len(set(delays)) > 1 + + def test_jitter_never_yields_a_negative_delay(self): + policy = RetryPolicy(delay=1.0, backoff=1.0, jitter=1.0) + assert all(RetryPolicy.delay_before(policy, 2) >= 0.0 for _ in range(200)) diff --git a/packages/interloper-core/tests/runner/test_async_runner.py b/packages/interloper-core/tests/runner/test_async_runner.py index 02d85a17..cf7345f3 100644 --- a/packages/interloper-core/tests/runner/test_async_runner.py +++ b/packages/interloper-core/tests/runner/test_async_runner.py @@ -80,6 +80,116 @@ def fine() -> list[dict[str, Any]]: assert result.executions["fine-2"].status is ExecutionStatus.COMPLETED +class TestRetry: + """An operation retries in place, within its declared budget.""" + + async def test_an_operation_that_heals_on_the_second_attempt_succeeds(self): + il.MemoryDestination.clear() + calls: list[int] = [] + + @il.asset(retry=il.RetryPolicy(max_attempts=3, delay=0.0, jitter=0.0)) + def flaky() -> list[dict[str, Any]]: + calls.append(1) + if len(calls) == 1: + raise ValueError("transient") + return [{"x": 1}] + + dag = il.DAG(flaky(id="flaky", destinations=[il.MemoryDestination()])) + events: list[Event] = [] + + result = await AsyncRunner(on_event=events.append).run(dag) + + assert result.status is ExecutionStatus.COMPLETED + assert result.executions["flaky"].status is ExecutionStatus.COMPLETED + assert len(calls) == 2 + assert [e.type for e in events if e.type is il.EventType.OPERATION_RETRIED] + assert not [e.type for e in events if e.type is il.EventType.OPERATION_FAILED] + + async def test_an_exhausted_budget_fails_once(self): + il.MemoryDestination.clear() + calls: list[int] = [] + + @il.asset(retry=il.RetryPolicy(max_attempts=2, delay=0.0, jitter=0.0)) + def broken() -> list[dict[str, Any]]: + calls.append(1) + raise ValueError("permanent") + + dag = il.DAG(broken(id="broken", destinations=[il.MemoryDestination()])) + events: list[Event] = [] + + result = await AsyncRunner(on_event=events.append).run(dag) + + assert result.status is ExecutionStatus.FAILED + assert len(calls) == 2 + assert len([e for e in events if e.type is il.EventType.OPERATION_RETRIED]) == 1 + assert len([e for e in events if e.type is il.EventType.OPERATION_FAILED]) == 1 + + async def test_an_operation_without_a_budget_is_attempted_once(self): + il.MemoryDestination.clear() + calls: list[int] = [] + + @il.asset() + def broken() -> list[dict[str, Any]]: + calls.append(1) + raise ValueError("permanent") + + dag = il.DAG(broken(id="broken", destinations=[il.MemoryDestination()])) + + result = await AsyncRunner().run(dag) + + assert result.status is ExecutionStatus.FAILED + assert len(calls) == 1 + + async def test_an_error_the_operation_declines_is_not_retried(self): + il.MemoryDestination.clear() + calls: list[int] = [] + + class Picky(il.Asset): + """Asset that knows a ValueError will never heal.""" + + retry: il.RetryPolicy | None = il.RetryPolicy(max_attempts=3, delay=0.0, jitter=0.0) + + def data(self) -> list[dict[str, Any]]: + calls.append(1) + raise ValueError("permanent") + + def retryable(self, error: Exception) -> bool: + return not isinstance(error, ValueError) + + dag = il.DAG(Picky(id="picky", destinations=[il.MemoryDestination()])) + + result = await AsyncRunner().run(dag) + + assert result.status is ExecutionStatus.FAILED + assert len(calls) == 1 + + async def test_a_retrying_node_does_not_trip_fail_fast(self): + il.MemoryDestination.clear() + calls: list[int] = [] + + @il.asset(retry=il.RetryPolicy(max_attempts=2, delay=0.0, jitter=0.0)) + def flaky() -> list[dict[str, Any]]: + calls.append(1) + if len(calls) == 1: + raise ValueError("transient") + return [{"x": 1}] + + @il.asset() + def independent() -> list[dict[str, Any]]: + return [{"x": 2}] + + dag = il.DAG( + flaky(id="flaky", destinations=[il.MemoryDestination()]), + independent(id="independent", destinations=[il.MemoryDestination()]), + ) + + result = await AsyncRunner(max_workers=1, fail_fast=True).run(dag) + + assert result.status is ExecutionStatus.COMPLETED + assert result.executions["flaky"].status is ExecutionStatus.COMPLETED + assert result.executions["independent"].status is ExecutionStatus.COMPLETED + + class TestMachineryErrors: """Failures of the walk itself, as opposed to an operation's.""" diff --git a/packages/interloper-core/tests/runner/test_multi_process.py b/packages/interloper-core/tests/runner/test_multi_process.py index 94dddf9c..3d7e4938 100644 --- a/packages/interloper-core/tests/runner/test_multi_process.py +++ b/packages/interloper-core/tests/runner/test_multi_process.py @@ -12,6 +12,7 @@ import interloper as il from interloper.errors import RunnerError +from interloper.events import Event from interloper.runner.multi_process import MultiProcessRunner, _worker from interloper.runner.results import ExecutionStatus from interloper.settings import RunnerSettings @@ -56,6 +57,36 @@ def data(self) -> Any: return [{"x": 2}] +_FLAKY_ATTEMPTS: list[int] = [] + + +class RetrySource(il.Source): + """Assets exercising the worker's own attempt loop.""" + + class Flaky(il.Asset): + """Fails its first attempt, then succeeds.""" + + retry: il.RetryPolicy | None = il.RetryPolicy(max_attempts=3, delay=0.0, jitter=0.0) + + def data(self) -> Any: + _FLAKY_ATTEMPTS.append(1) + if len(_FLAKY_ATTEMPTS) == 1: + raise ValueError("transient") + return [{"x": 1}] + + class AlwaysBroken(il.Asset): + """Fails every attempt. + + Raises: + ValueError: Always. + """ + + retry: il.RetryPolicy | None = il.RetryPolicy(max_attempts=2, delay=0.0, jitter=0.0) + + def data(self) -> Any: + raise ValueError("permanent") + + class ChainSource(il.Source): """A failing asset plus the downstream that depends on it.""" @@ -131,7 +162,7 @@ def test_a_successful_operation_returns_its_effects(self) -> None: spec, dag = _spec(WorkerSource(destinations=[il.MemoryDestination()])) operation = next(o for o in dag.operations if o.key == "ok") - operation_id, success, error, tb, effects = _worker(operation.id, spec, None, {}) + operation_id, success, error, tb, effects, _retries = _worker(operation.id, spec, None, {}) assert operation_id == operation.id assert success is True @@ -143,7 +174,7 @@ def test_a_failing_operation_reports_its_error_and_traceback(self) -> None: spec, dag = _spec(WorkerSource(destinations=[il.MemoryDestination()])) operation = next(o for o in dag.operations if o.key == "boom") - operation_id, success, error, tb, effects = _worker(operation.id, spec, None, {}) + operation_id, success, error, tb, effects, _retries = _worker(operation.id, spec, None, {}) assert operation_id == operation.id assert success is False @@ -151,12 +182,37 @@ def test_a_failing_operation_reports_its_error_and_traceback(self) -> None: assert tb is not None and "ValueError" in tb assert set(effects) == {"config", "state"} + def test_a_flaky_operation_is_retried_inside_the_worker(self) -> None: + il.MemoryDestination.clear() + _FLAKY_ATTEMPTS.clear() + spec, dag = _spec(RetrySource(destinations=[il.MemoryDestination()], select=["flaky"])) + operation = next(o for o in dag.operations if o.key == "flaky") + + _id, success, error, tb, _effects, retries = _worker(operation.id, spec, None, {}) + + assert success is True + assert (error, tb) == (None, None) + assert len(_FLAKY_ATTEMPTS) == 2 + assert len(retries) == 1 + assert "transient" in retries[0] + + def test_an_exhausted_budget_reports_its_retried_attempts(self) -> None: + il.MemoryDestination.clear() + spec, dag = _spec(RetrySource(destinations=[il.MemoryDestination()], select=["always_broken"])) + operation = next(o for o in dag.operations if o.key == "always_broken") + + _id, success, error, _tb, _effects, retries = _worker(operation.id, spec, None, {}) + + assert success is False + assert "permanent" in (error or "") + assert len(retries) == 1 + def test_an_unresolvable_operation_id_still_reports_cleanly(self) -> None: # The node is looked up before any operation exists, so the worker has # nothing to build a failure result from. spec, _dag = _spec(WorkerSource(destinations=[il.MemoryDestination()])) - operation_id, success, error, tb, effects = _worker("not-in-this-dag", spec, None, {}) + operation_id, success, error, tb, effects, _retries = _worker("not-in-this-dag", spec, None, {}) assert (operation_id, success) == ("not-in-this-dag", False) assert error is not None @@ -164,7 +220,7 @@ def test_an_unresolvable_operation_id_still_reports_cleanly(self) -> None: assert effects == {} def test_a_malformed_spec_is_reported_not_raised(self) -> None: - operation_id, success, error, tb, effects = _worker("anything", {"nodes": "nonsense"}, None, {}) + operation_id, success, error, tb, effects, _retries = _worker("anything", {"nodes": "nonsense"}, None, {}) assert (operation_id, success, effects) == ("anything", False, {}) assert error is not None @@ -182,6 +238,23 @@ def test_materializes_every_operation(self, importable_in_children: None) -> Non assert result.status is ExecutionStatus.COMPLETED assert len(result.completed_ids) == 2 + def test_a_child_retry_reaches_the_parent_as_events(self, importable_in_children: None) -> None: + il.MemoryDestination.clear() + runner = MultiProcessRunner(max_workers=1) + events: list[Event] = [] + + result = il.run( + runner.model_copy(update={"on_event": events.append}).run( + il.DAG(RetrySource(destinations=[il.MemoryDestination()], select=["flaky"])) + ) + ) + + # The worker owns the loop; the parent replays what it retried, so the + # attempt is visible here even though it happened in another process. + assert result.status is ExecutionStatus.COMPLETED + assert len([e for e in events if e.type is il.EventType.OPERATION_RETRIED]) == 1 + assert not [e for e in events if e.type is il.EventType.OPERATION_FAILED] + def test_a_child_failure_lands_on_the_node(self, importable_in_children: None) -> None: runner = MultiProcessRunner(max_workers=2, fail_fast=False) @@ -259,7 +332,7 @@ def test_a_success_tuple_records_the_effects( ) -> None: runner, operation = prepared future: Future[Any] = Future() - future.set_result((operation.id, True, None, None, {"config": {"cursor": "z"}, "state": {}})) + future.set_result((operation.id, True, None, None, {"config": {"cursor": "z"}, "state": {}}, [])) getattr(runner, handler)(future, operation) @@ -274,7 +347,7 @@ def test_a_failure_tuple_records_the_error_on_the_effects( ) -> None: runner, operation = prepared future: Future[Any] = Future() - future.set_result((operation.id, False, "child exploded", "Traceback...", {"config": {}, "state": {}})) + future.set_result((operation.id, False, "child exploded", "Traceback...", {"config": {}, "state": {}}, [])) getattr(runner, handler)(future, operation) @@ -291,7 +364,7 @@ def test_a_failure_without_a_message_gets_a_placeholder( ) -> None: runner, operation = prepared future: Future[Any] = Future() - future.set_result((operation.id, False, None, None, {"config": {}, "state": {}})) + future.set_result((operation.id, False, None, None, {"config": {}, "state": {}}, [])) getattr(runner, handler)(future, operation) @@ -339,6 +412,6 @@ def test_worker_adopts_and_releases_the_parent_span_context() -> None: inject_metadata(metadata) assert "traceparent" in metadata - _operation_id, success, _error, _tb, _effects = _worker(operation.id, spec, None, metadata) + _operation_id, success, _error, _tb, _effects, _retries = _worker(operation.id, spec, None, metadata) assert success is True diff --git a/packages/interloper-core/tests/runner/test_state.py b/packages/interloper-core/tests/runner/test_state.py index bf335bd8..3538a77f 100644 --- a/packages/interloper-core/tests/runner/test_state.py +++ b/packages/interloper-core/tests/runner/test_state.py @@ -140,6 +140,54 @@ def test_a_dependent_of_only_skipped_operations_is_promoted(self, monkeypatch: A assert state.executions[middle.id].status is ExecutionStatus.READY +class TestRetriedAttempts: + """A retried attempt is recorded without becoming the node's verdict.""" + + def test_mark_retried_emits_and_advances_the_attempt(self, chain: tuple[RunState, dict[str, Any]]) -> None: + state, operations = chain + root = operations["root"] + events: list[il.Event] = [] + il.EventBus.subscribe(events.append) + try: + state.mark_retried(root, "boom") + il.EventBus.flush(timeout=5.0) + finally: + il.EventBus.unsubscribe(events.append) + + assert state.attempts[root.id] == 2 + retried = [event for event in events if event.type is il.EventType.OPERATION_RETRIED] + assert len(retried) == 1 + assert retried[0].metadata["attempt"] == 1 + assert retried[0].metadata["error"] == "boom" + + def test_a_retried_attempt_is_not_a_verdict(self, chain: tuple[RunState, dict[str, Any]]) -> None: + state, operations = chain + root, middle = operations["root"], operations["middle"] + before = (state.executions[root.id].status, state.executions[middle.id].status) + + state.mark_retried(root, "boom") + + # Neither a terminal status for the node nor a cancellation downstream: + # only an exhausted budget is a failure, and mark_failed is what says so. + assert (state.executions[root.id].status, state.executions[middle.id].status) == before + assert state.executions[middle.id].status is not ExecutionStatus.CANCELED + + def test_events_after_a_retry_carry_the_new_attempt(self, chain: tuple[RunState, dict[str, Any]]) -> None: + state, operations = chain + root = operations["root"] + events: list[il.Event] = [] + il.EventBus.subscribe(events.append) + try: + state.mark_retried(root, "boom") + state.mark_failed(root, "boom again") + il.EventBus.flush(timeout=5.0) + finally: + il.EventBus.unsubscribe(events.append) + + failed = [event for event in events if event.type is il.EventType.OPERATION_FAILED] + assert failed[0].metadata["attempt"] == 2 + + class TestStatusBuckets: """The per-status operation views the runners schedule from.""" diff --git a/packages/interloper-core/tests/runner/test_state_event_ids.py b/packages/interloper-core/tests/runner/test_state_event_ids.py index f3d6be5e..dfb753fa 100644 --- a/packages/interloper-core/tests/runner/test_state_event_ids.py +++ b/packages/interloper-core/tests/runner/test_state_event_ids.py @@ -55,6 +55,22 @@ def test_asset_event_id_is_deterministic() -> None: assert base != RunState._operation_event_id("run-1", "asset-1", EventType.OPERATION_COMPLETED) +def test_asset_event_id_is_unchanged_for_a_first_attempt() -> None: + """Ids written before retries existed keep their value.""" + implicit = RunState._operation_event_id("run-1", "asset-1", EventType.OPERATION_FAILED) + explicit = RunState._operation_event_id("run-1", "asset-1", EventType.OPERATION_FAILED, attempt=1) + + assert implicit == explicit + + +def test_asset_event_id_differs_per_attempt() -> None: + """Each attempt's events are their own rows, not a dedup of the first's.""" + first = RunState._operation_event_id("run-1", "asset-1", EventType.OPERATION_STARTED, attempt=1) + second = RunState._operation_event_id("run-1", "asset-1", EventType.OPERATION_STARTED, attempt=2) + + assert first != second + + # -- RunState stamps the deterministic id on what it emits --------------------- diff --git a/packages/interloper-core/tests/source/test_base.py b/packages/interloper-core/tests/source/test_base.py index 073ac4c7..80cb264b 100644 --- a/packages/interloper-core/tests/source/test_base.py +++ b/packages/interloper-core/tests/source/test_base.py @@ -133,6 +133,39 @@ def data(self, orders: il.Upstream) -> Any: # pragma: no cover # -- Identity and class metadata ----------------------------------------------- +class TestRetryInheritance: + def test_a_source_policy_fills_its_assets(self): + policy = il.RetryPolicy(max_attempts=5) + + class Shop(il.Source): + class Orders(il.Asset): + def data(self) -> Any: # pragma: no cover + return [] + + assert Shop(retry=policy).assets[0].retry == policy + + def test_an_assets_own_policy_wins(self): + source_policy = il.RetryPolicy(max_attempts=5) + asset_policy = il.RetryPolicy(max_attempts=2) + + class Shop(il.Source): + class Orders(il.Asset): + retry: il.RetryPolicy | None = asset_policy + + def data(self) -> Any: # pragma: no cover + return [] + + assert Shop(retry=source_policy).assets[0].retry == asset_policy + + def test_no_policy_anywhere_leaves_assets_unset(self): + class Shop(il.Source): + class Orders(il.Asset): + def data(self) -> Any: # pragma: no cover + return [] + + assert Shop().assets[0].retry is None + + class TestIdentity: def test_key_auto_derived_from_class_name(self): assert FakeSource.key == "fake_source" From 3d1d3db8e7e036efc6ad226ec5979d3a973bce12 Mon Sep 17 00:00:00 2001 From: Guillaume Onfroy Date: Thu, 17 Sep 2026 18:22:29 +0200 Subject: [PATCH 2/3] test(core): make the cross-process retry test start-method agnostic The end-to-end pool test read this module's attempt counter as an earlier test had left it whenever the pool forks, which is the default on Linux: the child inherited a populated list, the asset succeeded on its first attempt, and no retry was recorded. macOS spawns, re-imports the module fresh and hid it. Clearing the counter in the test makes it read the same under both. By Digitl --- packages/interloper-core/tests/runner/test_multi_process.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/packages/interloper-core/tests/runner/test_multi_process.py b/packages/interloper-core/tests/runner/test_multi_process.py index 3d7e4938..e55111c7 100644 --- a/packages/interloper-core/tests/runner/test_multi_process.py +++ b/packages/interloper-core/tests/runner/test_multi_process.py @@ -240,6 +240,11 @@ def test_materializes_every_operation(self, importable_in_children: None) -> Non def test_a_child_retry_reaches_the_parent_as_events(self, importable_in_children: None) -> None: il.MemoryDestination.clear() + # A pool that forks (Linux) hands the child this module's state as the + # parent left it, so an earlier test's attempts would make the asset + # succeed first try here. A pool that spawns (macOS) re-imports and + # would not. Clearing makes the test read the same under both. + _FLAKY_ATTEMPTS.clear() runner = MultiProcessRunner(max_workers=1) events: list[Event] = [] From 03da03b634b26985c0c48aed3836a66c39ac23d9 Mon Sep 17 00:00:00 2001 From: Guillaume Onfroy Date: Thu, 17 Sep 2026 18:29:48 +0200 Subject: [PATCH 3/3] test(core): cover replaying retried attempts on the flush path `_handle_flushed` replayed the worker's retried attempts but nothing exercised it with a non-empty list, so a retry landing while the walk drains its in-flight work could have stopped emitting events unnoticed. The existing parametrized case covers both paths back into the parent at once. By Digitl --- .../tests/runner/test_multi_process.py | 24 +++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/packages/interloper-core/tests/runner/test_multi_process.py b/packages/interloper-core/tests/runner/test_multi_process.py index e55111c7..0d7b2d9e 100644 --- a/packages/interloper-core/tests/runner/test_multi_process.py +++ b/packages/interloper-core/tests/runner/test_multi_process.py @@ -331,6 +331,30 @@ def prepared(self) -> tuple[MultiProcessRunner, Any]: runner.state.mark_running(operation) return runner, operation + @pytest.mark.parametrize("handler", ["_handle_completed", "_handle_flushed"]) + def test_the_attempts_the_worker_retried_are_replayed( + self, handler: str, prepared: tuple[MultiProcessRunner, Any] + ) -> None: + # The child owns the loop but emits nothing; both paths back into the + # parent have to replay what it retried, including the flush path a + # fail-fast break or a natural end drains through. + runner, operation = prepared + future: Future[Any] = Future() + future.set_result((operation.id, True, None, None, {"config": {}, "state": {}}, ["boom", "boom again"])) + events: list[Event] = [] + il.EventBus.subscribe(events.append) + try: + getattr(runner, handler)(future, operation) + il.EventBus.flush(timeout=5.0) + finally: + il.EventBus.unsubscribe(events.append) + + assert runner.state.attempts[operation.id] == 3 + retried = [event for event in events if event.type is il.EventType.OPERATION_RETRIED] + assert [event.metadata["error"] for event in retried] == ["boom", "boom again"] + assert [event.metadata["attempt"] for event in retried] == [1, 2] + assert runner.state.executions[operation.id].status is ExecutionStatus.COMPLETED + @pytest.mark.parametrize("handler", ["_handle_completed", "_handle_flushed"]) def test_a_success_tuple_records_the_effects( self, handler: str, prepared: tuple[MultiProcessRunner, Any]