带抖动与退避的重试

from collections import Counter
import random


class Clock:
    def __init__(self):
        self.now = 0.0

    def advance(self, seconds):
        self.now += seconds


class ScriptedCall:
    def __init__(self, clock, steps):
        self.clock, self.steps, self.timeouts = clock, iter(steps), []

    def __call__(self, timeout):
        self.timeouts.append(timeout)
        status, duration = next(self.steps)
        self.clock.advance(min(duration, timeout))
        if duration >= timeout:
            raise TimeoutError("the cooperative fixture used the remaining budget")
        return status


def retry(call, clock, budget, sample, method="GET", max_attempts=3):
    deadline = clock.now + budget
    for attempt in range(1, max_attempts + 1):
        remaining = deadline - clock.now
        if remaining <= 0:
            return "deadline", attempt - 1
        try:
            status = call(remaining)
        except TimeoutError:
            status = None
        if clock.now >= deadline:
            return "deadline", attempt
        if status == 200:
            return "success", attempt
        if status not in (None, 503) or method not in ("GET", "HEAD"):
            return "not-retryable", attempt
        if attempt == max_attempts:
            return "attempt-limit", attempt
        ceiling = 2.0 ** min(attempt - 1, 3)
        delay = ceiling * sample()
        assert 0 <= delay <= ceiling
        if delay >= deadline - clock.now:
            return "wait-exceeds-budget", attempt
        clock.advance(delay)
    raise AssertionError("positive attempt limit required")


clock = Clock()
call = ScriptedCall(clock, [(503, 0.25), (503, 0.25), (200, 0.25)])
assert retry(call, clock, 5, lambda: 0.5) == ("success", 3)
assert clock.now == 2.25 and call.timeouts == [5, 4.25, 3]
print("three attempts finish at 2.25; propagated timeouts shrink from 5 to 4.25 to 3")

class EarlyTimeout:
    def __init__(self, clock):
        self.clock, self.timeouts = clock, []

    def __call__(self, timeout):
        self.timeouts.append(timeout)
        self.clock.advance(0.25)
        if len(self.timeouts) == 1:
            raise TimeoutError("a transient failure before the deadline")
        return 200


clock = Clock()
call = EarlyTimeout(clock)
assert retry(call, clock, 5, lambda: 0.5) == ("success", 2)
assert clock.now == 1.0 and call.timeouts == [5, 4.25]
print("an early timeout can recover without resetting the shared deadline")

for method, status in (("POST", 503), ("GET", 400)):
    clock = Clock()
    call = ScriptedCall(clock, [(status, 0.25)])
    assert retry(call, clock, 5, lambda: 0.5, method) == ("not-retryable", 1)
clock = Clock()
call = ScriptedCall(clock, [(503, 0.25)] * 4)
assert retry(call, clock, 5, lambda: 0.5) == ("attempt-limit", 3)
assert len(call.timeouts) == 3
print("POST, permanent failure, and the total-attempt cap stop further work")

clock = Clock()
call = ScriptedCall(clock, [(503, 0.25)] * 7)
assert retry(call, clock, 100, lambda: 1.0, max_attempts=7) == ("attempt-limit", 7)
assert clock.now == 32.75  # Seven 0.25 calls plus waits 1, 2, 4, 8, 8, 8.
print("the exponential ceiling stops at 8 rather than growing without bound")

clock = Clock()
call = ScriptedCall(clock, [(503, 0.1)])
assert retry(call, clock, 0.5, lambda: 1.0) == ("wait-exceeds-budget", 1)
assert clock.now == 0.1 and len(call.timeouts) == 1
clock = Clock()
call = ScriptedCall(clock, [(503, 99)])
assert retry(call, clock, 0.5, lambda: 0.5) == ("deadline", 1)
assert clock.now == 0.5
clock = Clock()
call = ScriptedCall(clock, [])
assert retry(call, clock, 0, lambda: 0.5) == ("deadline", 0) and call.timeouts == []
print("a wait that cannot fit is abandoned; cooperative execution cannot consume a fresh budget per attempt")

rng = random.Random(7)  # Repeatable samples for this experiment, not a production seed.
delays = [rng.random() for _ in range(100)]
bins = Counter(int(delay * 10) for delay in delays)
fixed_bins = Counter(int(1.0 * 10) for _ in range(100))
assert all(0 <= delay < 1 for delay in delays) and sum(bins.values()) == 100
assert max(bins.values()) < 100
assert max(fixed_bins.values()) == 100
print("first-retry 0.1-second-bin peak: fixed delay", max(fixed_bins.values()), "; full-jitter sample", max(bins.values()))

leaf_calls = 0


class Unavailable(Exception):
    pass


def nested(depth):
    global leaf_calls
    if depth == 0:
        leaf_calls += 1
        raise Unavailable
    for _ in range(3):
        try:
            return nested(depth - 1)
        except Unavailable:
            pass
    raise Unavailable


try:
    nested(3)
except Unavailable:
    pass
assert leaf_calls == 27
print("three retrying layers, each allowing three total attempts, invoke the failing leaf 27 times")