feat(core): retry a failed operation in place

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",

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,

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)

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]:

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.
 

packages/interloper-core/src/interloper/retry/__init__.py

@@ -0,0 +1,5 @@
+"""Retry policies."""
+
+from interloper.retry.base import RetryPolicy
+
+__all__ = ["RetryPolicy"]

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))

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.

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: