DEV Community

Dakota Huang
Dakota Huang

Posted on

Pin Sleep Argument Lists Before One Retry Extract

Messy retry loops fail extracts in a predictable way. Unpinned helpers change attempt counts and delay math. Freeze sleep arguments and retryable error classes first. Then move only the loop into a helper.

Core rule

Do not extract a retry helper from mixed networking code. Pin sleep arguments and retryable exception types first. After those pins pass, move only the loop.

Leave jitter, logging, and HTTP details untouched. Those behaviors need separate goldens later. One extract should change structure, not timing policy.

The usual breakage

A god function loads data and retries timeouts inline. It also maps business errors in the same block. A model asked to clean retries will invent new policy. It often retries Exception instead of two classes.

Wall-clock elapsed time is the wrong golden. Process scheduling noise will fail that suite. Spy time.sleep instead of measuring seconds on a clock. The argument list is the backoff contract.

Observables to freeze

Capture four fields for every fixture row.

  1. Attempt count against the loader mock.
  2. Exact time.sleep argument sequence.
  3. Exception class that escapes after exhaustion.
  4. Sleep count after a mid-loop success.

Use this decision table before writing tests.

Injected loader behavior Attempts Sleep args Escapes as
TimeoutError then success 2 [0.1] no exception
TimeoutError three times 3 [0.1, 0.2] TimeoutError
ConnectionError then success 2 [0.1] no exception
ValueError on first call 1 [] ValueError
success on first call 1 [] no exception

The table is the product behavior. Comments in the messy function are not the contract. If a later extract changes one cell, reject the patch.

Artifact: stdlib characterization harness

The following Python is a proposed local example. It is not a live benchmark and claims no production timings. It uses only the standard library. Drop both files into an empty directory.

File invoice_fetch.py

# Proposed local example. Not production networking code.
from __future__ import annotations

import time
from typing import Callable, TypeVar

T = TypeVar("T")
RETRYABLE = (TimeoutError, ConnectionError)
DELAYS = (0.1, 0.2, 0.4)


def fetch_invoice(load: Callable[[], T]) -> T:
    last_exc: BaseException | None = None
    for index, delay in enumerate(DELAYS):
        try:
            return load()
        except RETRYABLE as exc:
            last_exc = exc
            if index < len(DELAYS) - 1:
                time.sleep(delay)
            continue
    assert last_exc is not None
    raise last_exc
Enter fullscreen mode Exit fullscreen mode

File test_invoice_fetch_chars.py

# Proposed characterization suite. Label unexecuted until you run it.
import unittest
from unittest.mock import patch

import invoice_fetch


class SleepRecorder:
    def __init__(self) -> None:
        self.calls: list[float] = []

    def __call__(self, seconds: float) -> None:
        self.calls.append(seconds)


class Loader:
    def __init__(self, outcomes: list[object]) -> None:
        self.outcomes = list(outcomes)
        self.calls = 0

    def __call__(self):
        self.calls += 1
        item = self.outcomes.pop(0)
        if isinstance(item, BaseException):
            raise item
        return item


class CharacterizeFetchInvoice(unittest.TestCase):
    def _run(self, outcomes):
        recorder = SleepRecorder()
        loader = Loader(outcomes)
        with patch.object(invoice_fetch.time, "sleep", recorder):
            try:
                value = invoice_fetch.fetch_invoice(loader)
                error = None
            except BaseException as exc:
                value = None
                error = exc
        return loader.calls, recorder.calls, value, error

    def test_success_first_try_does_not_sleep(self):
        attempts, sleeps, value, error = self._run(["ok"])
        self.assertEqual(attempts, 1)
        self.assertEqual(sleeps, [])
        self.assertEqual(value, "ok")
        self.assertIsNone(error)

    def test_timeout_then_success_sleeps_once(self):
        attempts, sleeps, value, error = self._run(
            [TimeoutError("t"), "ok"]
        )
        self.assertEqual(attempts, 2)
        self.assertEqual(sleeps, [0.1])
        self.assertEqual(value, "ok")
        self.assertIsNone(error)

    def test_three_timeouts_do_not_sleep_after_last(self):
        attempts, sleeps, value, error = self._run(
            [TimeoutError("a"), TimeoutError("b"), TimeoutError("c")]
        )
        self.assertEqual(attempts, 3)
        self.assertEqual(sleeps, [0.1, 0.2])
        self.assertIsNone(value)
        self.assertIsInstance(error, TimeoutError)

    def test_value_error_is_not_retryable(self):
        attempts, sleeps, value, error = self._run(
            [ValueError("bad invoice")]
        )
        self.assertEqual(attempts, 1)
        self.assertEqual(sleeps, [])
        self.assertIsNone(value)
        self.assertIsInstance(error, ValueError)


if __name__ == "__main__":
    unittest.main()
Enter fullscreen mode Exit fullscreen mode

Commands

python -m unittest test_invoice_fetch_chars.py -v
Enter fullscreen mode Exit fullscreen mode

A passing run means the current loop is pinned. It does not mean the loop is well designed. Do not refactor on a red characterization suite.

Numbered workflow

1. Copy the tangled function into a fixture module

Keep production imports out of the first pin. Reproduce the retry branch with a callable hook. The hook replaces HTTP for the test. Real sockets add noise and flakes.

Name the fixture after the production function. Do not rename delays during the copy. Copy-paste errors here become false goldens later.

2. Patch time.sleep on the module that calls it

Patch the name used by the messy module. Patching time.sleep on the wrong module yields a false green. Record every seconds argument in a list. Ignore the return value of sleep.

Confirm the spy with a deliberate wrong delay. If the test still passes, the patch target is wrong. Fix the target before writing more rows.

3. Drive one behavior per test

Do not fold five error classes into one case. Mixed assertions hide which table cell drifted. Name tests after the table rows above. Failures then point at one contract cell.

Keep loader outcomes in a short list. Pop one outcome per attempt. That matches how a real retry loop consumes failures.

4. Store goldens as literals

Keep expected sleep tuples in the test file. Do not rebuild delays with ** inside assertions. The test must show numbers a reviewer can read. Hidden formulas drift with the production formula.

Treat 0.1, 0.2 as data, not as a series. If production later uses 0.1, 0.3, the test must fail. Silent series generators hide that change.

5. Extract only the retry loop

Move the for loop into run_with_delays. Pass the retryable tuple and the delay tuple. Do not add jitter in this commit. Do not add logging in this commit.

Do not wrap exceptions in a new type. Callers already catch TimeoutError and ValueError. A new wrapper is a behavior change. Behavior changes need new table rows.

6. Re-run the same four fields

Attempt counts must match the table. Sleep lists must match the table. Escaping types must match the table. Any extra sleep means a failed extract.

If one row fails, revert the helper. Do not tune delays to soothe the test. The table wins over the new structure.

Smallest safe production edit

Proposed extract below. Still local example code. It keeps DELAYS and RETRYABLE unchanged.

from __future__ import annotations

import time
from typing import Callable, Iterable, TypeVar

T = TypeVar("T")
RETRYABLE = (TimeoutError, ConnectionError)
DELAYS = (0.1, 0.2, 0.4)


def run_with_delays(
    load: Callable[[], T],
    delays: Iterable[float],
    retryable: tuple[type[BaseException], ...],
) -> T:
    delay_list = list(delays)
    last_exc: BaseException | None = None
    for index, delay in enumerate(delay_list):
        try:
            return load()
        except retryable as exc:
            last_exc = exc
            if index < len(delay_list) - 1:
                time.sleep(delay)
            continue
    assert last_exc is not None
    raise last_exc


def fetch_invoice(load: Callable[[], T]) -> T:
    return run_with_delays(load, DELAYS, RETRYABLE)
Enter fullscreen mode Exit fullscreen mode

Stop the commit after this single structural edit. Resist extra cleanups in the same diff. Header maps, URL joins, and JSON decoding stay put. Those cuts need their own pins.

False greens to reject

Patching time.sleep in the test module is a common miss. The messy module still sleeps for real. Tests look slow and still pass weak assertions. Always patch invoice_fetch.time.sleep.

Another false green is asserting len(sleeps) > 0 only. Length ignores 0.1 versus 0.5. A helper can double backoff and still pass. Compare the full argument list.

A third false green is catching Exception in the test. That hides a wrapper change. Assert type(error) against the table class. Wrappers are contract breaks.

After the suite is green

Models rewrite retry code with extra policy objects. That help is useful only after pins exist. Feed the unittest file and the messy function together. Ask for a loop extract that keeps every table cell.

Disclosure: This article was prepared as part of MonkeyCode's product outreach.

MonkeyCode provides free model access and a free server option. Those two availability facts are the product surface used here. A remote draft can propose run_with_delays against the unittest file. It cannot invent the sleep contract for you.

Reject any patch that introduces jitter or broader except Exception. If the characterization file already fails, skip the model session. Fix the pin first, or the model will optimize a lie.

Limitations

Sleep spies miss C extensions that delay without Python sleep. They also miss a local alias bound after the patch. Async code needs an asyncio.sleep spy instead. Threaded retries need a clock that is safe across threads.

Characterization freezes bugs as well as features. If attempt three never ran in production, the suite will protect that gap. Schedule a deliberate behavior change as a second commit. Do not hide it inside the extract.

Jitter, deadlines, and retry-after headers stay out of scope. Pin them with new rows before touching them. HTTP status retry rules are a different contract. Record status codes in a later suite.

Deadline loops that poll time.monotonic need a fake clock. This sleep-list method will not catch those. Choose the observable that the production code actually calls.

Who should not use this approach

Skip this method on greenfield services with no callers. Write an explicit retry policy first in that case. Skip it when delay math must change in the same patch. That mix is a behavior change, not an extract.

Skip it for latency checks that require real waits. These tests never sleep on the wall clock. They cannot detect a pause that bypasses time.sleep. Skip it when exception types cannot be frozen yet.

Broad except Exception retries are a product decision. Pins will only fossilize that decision. Do not use this workflow to launder a wider catch. Change the table in public review first.

Checklist before merge

  1. Table cells match the unittest method names.
  2. python -m unittest test_invoice_fetch_chars.py -v is green.
  3. Diff touches the loop and the new helper only.
  4. No jitter, logging, or HTTP edits in the same commit.

The extract is done when the table is unchanged. Anything else is a second change. Keep that second change behind a new golden row.

Top comments (0)