GitHub

Schedules and retries

PreviousNext

Retry failures and repeat work with composable schedules.

A schedule decides whether to try again and how long to wait. retry uses one to re-run an effect after a failure; repeat re-runs it after a success.

retry.py
from dataclasses import dataclass

import pyfect as fx
from pyfect import schedule


@dataclass
class Unavailable(fx.TaggedError):
    attempt: int


def flaky_service(failures: int) -> fx.Effect[str, Unavailable]:
    """Fails `failures` times, then succeeds."""
    attempts: list[int] = []

    def attempt() -> fx.Effect[str, Unavailable]:
        attempts.append(len(attempts) + 1)
        if len(attempts) <= failures:
            return fx.fail(Unavailable(len(attempts)))
        return fx.succeed("ok")

    # suspend: each retry calls attempt() again
    return fx.suspend(attempt)


# Wait 10ms, 20ms, 40ms... between attempts, at most 5 retries.
policy = schedule.exponential(0.01).up_to(times=5)
assert fx.run_sync_exit(flaky_service(2).retry(policy)) == fx.Success("ok")

# The keyword form: retry while the error matches, at most twice.
again = flaky_service(2).retry(times=2, while_=lambda e: e.attempt < 10)
assert fx.run_sync_exit(again) == fx.Success("ok")

# When the retries run out, the effect fails with the last error.
exhausted = flaky_service(5).retry(times=2)
assert fx.run_sync_exit(exhausted) == fx.exit.fail(Unavailable(3))

The waits of policy above, before each of its five retries:

schedule.exponential(0.01).up_to(times=5)02040608010012014016018012345Wait before it (ms)Retry
schedule.exponential(0.01).up_to(times=5)05010015012345Wait before it (ms)Retry

retry and repeat take a schedule, or the keywords schedule=, times=, while_= and until=. When retries run out, the effect fails with the last error.

repeat.py
import pyfect as fx
from pyfect import schedule

ticks: list[int] = []


def tick() -> fx.Effect[None]:
    return fx.sync(lambda: ticks.append(len(ticks)))


# Run once, then repeat every 10ms, three more times.
fx.run_sync(tick().repeat(schedule.spaced(0.01).up_to(times=3)))
assert ticks == [0, 1, 2, 3]

# Schedules combine: jittered exponential backoff, at most one second per wait.
backoff = schedule.min(schedule.exponential(0.1).jittered(), schedule.spaced(1))
assert fx.run_sync(fx.succeed("first try").retry(backoff)) == "first try"

Constructors: spaced, fixed, exponential, fibonacci, windowed, recurs, cron. Combinators: up_to(times=, duration=), jittered, while_, max / min, concat, and more. Schedules use the Clock service, so tests can run them on virtual time (Testing).

Reference: pyfect.schedule.