Shield

A Shield wraps one async call to one dependency with resilience built in: a per-attempt timeout, automatic retries, and an adaptive rate limiter that engages only when the dependency starts to struggle. One decorator, one knob.

Why

• Wrap one call to one dependency without composing four primitives by hand.
• Survive transient slow-downs (network blips, brief overloads) without touching the call site.
• Pace the whole client down automatically when a dependency keeps timing out, then ramp back up gradually.
• Stay invisible on the happy path. The adaptive layer turns itself on only after the first slow-down.

Usage

Decorate an async function, tell Shield with when= which errors mean "the dependency is slow", and call it as usual:

import httpx

from grelmicro.resilience import shield


@shield.api(when=(httpx.TimeoutException, httpx.ConnectError))
async def fetch(client: httpx.AsyncClient, url: str) -> bytes:
    response = await client.get(url)
    response.raise_for_status()
    return response.content

when= is required. It takes the same forms as Retry: an exception class, a tuple of classes, a predicate on the exception, or a Match.

Shield is async-only. Use it on coroutine functions.

Choose a profile

A profile sets the starting rate and timeout for the kind of dependency you are calling. Pick one and you are done.

Profile	Use when	Initial rate (per second)	Initial timeout
internal	In-cluster RPC. Healthy services, fast latency, tight budgets.	100	1.0s
api (default)	External HTTP APIs. Moderate latency, occasional outages, third-party SLAs.	2	10.0s
slow	Long-running calls: LLM inference, batch jobs, large queries.	0.5	120.0s

import httpx

from grelmicro.resilience import shield


class MyRpcTimeout(Exception): ...  # noqa: N818


class MyLLMError(Exception): ...


@shield.internal(when=MyRpcTimeout)
async def call_internal_rpc() -> None: ...


@shield.api(when=(httpx.TimeoutException, httpx.ConnectError))
async def call_external_api() -> None: ...


@shield.slow(when=MyLLMError)
async def call_llm(prompt: str) -> str:
    return prompt

Profiles are mutually exclusive: pick one per Shield instance. To pace differently, switch profile instead of tuning individual fields. If none of the three fit, you have outgrown Shield. Tune Retry, Timeout, RateLimiter, and CircuitBreaker yourself and hand them to a Stack, which applies them in the safe order.

Sharing across functions

Build a Shield instance once and decorate several functions with it. They share one retry budget and one adaptive controller, which is what you want when those functions hit the same dependency:

import httpx
from pydantic import BaseModel

from grelmicro.resilience import Shield

github = Shield.api(
    "github",
    when=(httpx.TimeoutException, httpx.ConnectError),
)


class Repo(BaseModel):
    name: str


@github
async def list_repos() -> list[Repo]:
    return []


@github
async def get_repo() -> Repo:
    return Repo(name="grelmicro")

One Shield instance per logical dependency. Two functions hitting GitHub share one budget. Add Stripe and you get a second Shield.

The name= argument is the registration name used in logs, metrics, and give-up notes. It defaults to the wrapped function's __qualname__ with @shield.api(...), and is a required positional argument with Shield.api(...).

Imperative form

For inline calls that span several statements, use Shield.run:

import httpx

from grelmicro.resilience import Shield

github = Shield.api("github", when=httpx.TimeoutException)


async def parse_response(response: httpx.Response) -> bytes:
    return response.content


async def handler(client: httpx.AsyncClient, url: str) -> bytes:
    response = await github.run(client.get, url)
    return await github.run(parse_response, response)

Shield.run(fn, *args, **kwargs) calls fn(*args, **kwargs) under the same retry-budget and adaptive state as the decorator form.

Cache and fallback

When Shield gives up, it tries two recovery paths in order: a cache lookup, then a fallback callable. Set either or both.

Cached fallback

Pass a Cache instance to cache=. Shield writes the return value on every success and reads it back on give-up:

from decimal import Decimal

import httpx

from grelmicro.cache import TTLCache
from grelmicro.resilience import shield


@shield.api(
    "prices",
    when=httpx.TimeoutException,
    cache=TTLCache(ttl=300),
)
async def fetch_price(symbol: str) -> Decimal:
    return Decimal(0)

• On success: await cache.set(key, return_value) runs fire-and-forget. A write failure is logged at debug and never propagates.
• On give-up: await cache.get(key) runs. A hit returns the cached value. A miss continues to the next recovery path.
• Key: f"{shield.name}:{stable_hash(args, kwargs)}" by default. Override it with cache_key=:

from decimal import Decimal

import httpx

from grelmicro.cache import TTLCache
from grelmicro.resilience import shield


@shield.api(
    "prices",
    when=httpx.TimeoutException,
    cache=TTLCache(ttl=300),
    cache_key=lambda symbol, *_: f"price:{symbol}",
)
async def fetch_price(symbol: str) -> Decimal:
    return Decimal(0)

Non-hashable arguments (Pydantic models, dataclasses) are hashed via a stable repr().

Custom fallback

Pass a callable to fallback= for when the cache misses (or no cache is set). It receives the exception that escaped Shield:

from decimal import Decimal

import httpx

from grelmicro.cache import TTLCache
from grelmicro.resilience import shield


async def last_known_price(exc: Exception) -> Decimal:
    return Decimal(0)


@shield.api(
    "prices",
    when=httpx.TimeoutException,
    cache=TTLCache(ttl=300),
    fallback=last_known_price,
)
async def fetch_price(symbol: str) -> Decimal:
    return Decimal(0)

Recovery order on give-up

Step	Condition	Outcome
1	cache= set and cache hit	return the cached value
2	cache= unset or cache miss, fallback= set	call fallback(exc), return its result
3	neither path returns a value	re-raise the original exception with a PEP 678 note

Use cache= alone when stale data is acceptable. Add fallback= as a safety net for a cold cache. Use fallback= alone for synthesized defaults.

Per-call rate ceiling

If the dependency has a contractual quota (an SLA limit, a third-party rate limit), set max_rate= to cap the adaptive limiter's growth:

from grelmicro.resilience import shield


class APIConnectionError(Exception): ...


@shield.api(
    "stripe",
    when=APIConnectionError,
    max_rate=10.0,
)
async def charge_card(amount: int) -> None:
    return None

The rate still grows back after recovery, but never past max_rate. Without it, the ceiling grows unbounded while the dependency stays healthy.

Which exceptions are retried

Shield decides what to do by the type of exception the wrapped call raises.

Raised by the wrapped call	Retried?	Slows the rate?	Consumes retry budget?	Cache / fallback recovery?
An error when= matches, or a TimeoutError	yes	yes	yes	yes, on give-up
Any other Exception	no	no	no	yes, on give-up
ResilienceError subclass	no	no	no	no, propagates immediately
BaseException outside Exception (KeyboardInterrupt, CancelledError, SystemExit)	no	no	no	no, propagates immediately

You declare what "transient" means with when=. Any other Exception skips the retry loop and goes straight to recovery: if a cache or fallback returns a value, that value is returned, otherwise the original exception is re-raised. ResilienceError and BaseException-outside-Exception always propagate unchanged.

TimeoutError is always retried, whatever when= says. Shield's own per-attempt timeout raises it, so a slow attempt always counts as transient. Shield sees raised errors only, so a Match.result(...) arm never matches.

Behavior on giving up

When Shield gives up and no recovery path returns a value, the original exception is re-raised with a PEP 678 note attached:

import asyncio

import httpx

from grelmicro.resilience import shield


@shield.api(when=httpx.TimeoutException)
async def fetch(url: str) -> bytes:
    raise httpx.TimeoutException("dependency stalled")


async def main() -> None:
    try:
        await fetch("https://example.com/recs")
    except httpx.TimeoutException as exc:
        print(exc.__notes__)


if __name__ == "__main__":
    asyncio.run(main())

output

['shield: attempts exhausted after 4/4 attempts in 3.82s (api profile)']

The note records the give-up reason (budget exhausted, attempts exhausted, non-retryable exception), the attempt count, the elapsed time, and the profile name. Callers catch the original exception type, unchanged. There is no ShieldError wrapper.

Composing with client-side retries

Shield is the outer layer of resilience. Many client libraries ship their own retries, tuned for protocol-level transience (Retry-After headers, idempotency keys, modeled status codes). Shield does not replace that work. It adds a slower layer on top.

You do not need to disable the client's retries. Pass the client's terminal exception types to when=:

import httpx

from grelmicro.resilience import shield

retrying_httpx_client = httpx.AsyncClient()


@shield.api(
    "github",
    when=(httpx.TimeoutException, httpx.ConnectError),
)
async def fetch(url: str) -> httpx.Response:
    return await retrying_httpx_client.get(url)

When the inner layer exhausts its own attempts and surfaces an exception, Shield sees it. The adaptive layer engages only when the outer exception escapes, never on per-attempt blips the inner layer already handled.

What Shield does not do

• HTTP status sniffing. No Retry-After parsing, no 429/503 awareness. Call response.raise_for_status() inside the wrapped function and pass the resulting exception type to when=.
• Sync callables. Async-only. Wrap sync code in asyncio.to_thread(...) if needed.
• Hedged requests. Firing a backup attempt and taking the first success is a different primitive. It is on the roadmap, not shipped.
• Distributed retry budget. Each Shield is per-process. There is no cross-replica coordination. Fleet-wide budgets are on the roadmap.
• Total deadline propagation. Shield enforces per-attempt timeouts. Wrap the whole call in asyncio.timeout(...) for a total deadline. A deadline that propagates on its own is on the roadmap.

How the adaptive layer works

You never tune these numbers. They are fixed by profile choice and shown here so the behavior is not a black box.

• Per-attempt timeout. Tracks the p95 of the last 32 successful latencies, multiplied by 2.5, clamped to the profile range. A dependency that gets slower raises its own timeout, up to the profile ceiling.
• Retry gate. A consecutive-failure budget with refund-on-recovery. Each give-up consumes budget, each success refunds it. When the budget is empty, Shield stops retrying instead of hammering a dead dependency.
• Adaptive rate limiter. A CUBIC-style controller from RFC 9438 with C = 0.4 and β = 0.7. It shrinks the allowed rate when slow-downs persist, then grows it back along the CUBIC curve as the dependency recovers. It stays at the profile's full rate until the first sustained slow-down, so the happy path pays nothing.

Configuration

Pass the profile and knobs as keyword arguments.

import httpx

from grelmicro.resilience import Shield

github = Shield.api(
    "github",
    when=(httpx.TimeoutException, httpx.ConnectError),
    max_rate=20.0,
)

Environment variables

Prefix: GREL_SHIELD_{NAME_UPPER}_. The default instance drops the name segment and reads GREL_SHIELD_*.

Environment variables are read by default

A GREL_* variable that fills a component field is read unless GREL_ENV_LOAD=0. With GREL_ENV_LOAD=0, a variable that is set is ignored, the default applies, and startup warns with its name. Passing env_load=False to a component turns its reads off and stays quiet. See How a value is resolved for the three ways to configure a component, including a local .env.

Env var	Field	Type	Default
GREL_SHIELD_{NAME_UPPER}_WHEN	when	CSV or JSON list of FQN strings (e.g. httpx.TimeoutException,httpx.ConnectError). Coerced to Match.exception(...). Predicate forms cannot come from env.	required
GREL_SHIELD_{NAME_UPPER}_MAX_RATE	max_rate	float or empty	unset

The profile, cache, cache_key, and fallback cannot come from env. Pick the profile with Shield.internal, Shield.api or Shield.slow, and pass the rest as keyword arguments.

from grelmicro.resilience import Shield

github = Shield("github", env_load=True)

Advanced

For the from_config declarative path and pydantic-settings composition, see Declarative configuration.

Live reconfiguration

Shield inherits Reconfigurable[ShieldConfig]. Calling shield.reconfigure(new_config) swaps the snapshot for future calls. An in-flight await shield.run(...) keeps its snapshot until it completes. See Live reconfiguration.

Reference

See the API reference for every option.