mildew

A high-performance, zero-dependency resilience and fault-tolerance framework for Python.

Designed for modern distributed applications, mildew provides production-grade isolation, retry strategies, rate limiting, and circuit breaking for both synchronous (def) and asynchronous (async def) execution flows.

---

Why Mildew Exists

Building reliable distributed systems in Python is hard. Network calls time out, third-party APIs experience transient downtime, and databases get overloaded under heavy traffic spikes.

Without fault-tolerance patterns:

• A single slow downstream API can exhaust your thread pool or asyncio event loop, causing your entire application to crash (Cascading Failure).
• Uncontrolled retries against an unstable database can cause a "thundering herd" effect, preventing the database from recovering.
• Missing rate limits result in HTTP 429 errors, IP blocks, or API key suspensions.

Mildew was built to solve these complex programming challenges cleanly, bringing resilience patterns popularized by tools like Resilience4j and Polly into the Python ecosystem with:

• Zero External Dependencies: Uses only standard Python libraries (asyncio, threading, time, collections).
• Universal Compatibility: Works out of the box with sync functions (def) and async coroutines (async def).
• Low Overhead & Thread-Safe: Optimized sliding-window metrics and locking mechanisms.

---

Real-World Use Cases

1. Flaky Third-Party API Integrations

When calling payment gateways (Stripe, PayPal) or AI services (OpenAI, Anthropic), transient HTTP 5xx errors or timeouts are common.

• Solution: Combine RetryPolicy (exponential backoff + jitter) with CircuitBreaker to retry brief glitches and stop calling the API if it goes down completely.

2. Protecting Database & Connection Pools

If an endpoint triggers heavy database queries, high concurrency can quickly exhaust the connection pool and stall the entire service.

• Solution: Wrap database calls with Bulkhead to cap maximum concurrency for expensive queries.

3. API Pacing & Rate Limit Compliance

External APIs strictly enforce rate limits (e.g., 100 requests per minute).

• Solution: Apply RateLimiter decorators to keep outgoing requests within provider allowances, avoiding HTTP 429 status codes.

4. Background Job Queueing & Processing

Dispatching and executing background tasks without overloading system resources.

• Solution: Use ResilientTaskEngine to manage worker concurrency and track completion statuses asynchronously.

---

Key Features

• Circuit Breaker: Halts execution to failing downstream services to prevent cascading outages.
• Retry Policy: Exponential backoff with jitter and customizable exception predicates.
• Rate Limiter: Token-bucket algorithm for call pacing and rate management.
• Bulkhead: Concurrency isolation to prevent resource exhaustion.
• Resilient Pipeline: Combine multiple protection strategies into a single decorator.
• Task Engine: Async task worker pool execution queue with status tracking.
• Metrics Tracker: Thread-safe rolling window latency and error tracker.

---

Table of Contents

• Installation
• Detailed Usage
  • 1. Circuit Breaker
  • 2. Retry Policy
  • 3. Rate Limiter
  • 4. Bulkhead
  • 5. Resilient Pipeline
  • 6. Resilient Task Engine
  • 7. Metrics Tracking
• Configuration Reference
• Running Tests
• License

---

Installation

Copy the mildew folder directly into your Python project path:

cp -r mildew /path/to/your/project/

---

Detailed Usage

1. Circuit Breaker

The CircuitBreaker monitors call success/failure rates in a sliding window. It operates in three states:

• CLOSED: Normal operation. Calls pass through.
• OPEN: Failure threshold exceeded. Calls immediately raise CircuitOpenError.
• HALF_OPEN: Recovery time elapsed. Probes downstream service with limited test calls.

from mildew import CircuitBreaker, CircuitOpenError

cb = CircuitBreaker(
    fail_threshold=5,     # Min samples before evaluating failure rate
    fail_rate=0.5,        # 50% failure rate triggers OPEN state
    recovery_time=10.0,   # Seconds to stay in OPEN before attempting HALF_OPEN
    half_open_max=3       # Successful test calls needed to reset to CLOSED
)

@cb
async def fetch_user_profile(user_id: str):
    return await http_client.get(f"/users/{user_id}")

2. Retry Policy

RetryPolicy automatically retries failed function calls using exponential backoff (delay * backoff^attempt) and optional randomized jitter.

from mildew import RetryPolicy

def is_transient_error(exc: BaseException) -> bool:
    return isinstance(exc, (TimeoutError, ConnectionResetError))

retry = RetryPolicy(
    max_attempts=4,
    delay=0.1,
    backoff=2.0,
    jitter=True,
    predicate=is_transient_error
)

@retry
def query_database():
    pass

3. Rate Limiter

Thread-safe Token Bucket implementation to limit call throughput over time windows.

from mildew import RateLimiter, RateLimitExceeded

limiter = RateLimiter(rate=50, per=1.0)

@limiter
def send_webhook(payload: dict):
    pass

4. Bulkhead

Isolates concurrency limits to prevent one component from starving shared thread pools or async tasks.

from mildew import Bulkhead, BulkheadFullError

bulkhead = Bulkhead(max_concurrent=10)

@bulkhead
async def process_file(path: str):
    pass

5. Resilient Pipeline

Combine multiple policies into a unified execution wrapper:

from mildew import (
    ResilientPipeline,
    RetryPolicy,
    RateLimiter,
    CircuitBreaker,
    Bulkhead
)

pipeline = (
    ResilientPipeline()
    .add(Bulkhead(max_concurrent=20))
    .add(RateLimiter(rate=100, per=1.0))
    .add(CircuitBreaker(fail_threshold=5, recovery_time=15.0))
    .add(RetryPolicy(max_attempts=3, delay=0.1))
)

@pipeline
async def reliable_external_call():
    pass

6. Resilient Task Engine

An asynchronous task execution queue with worker concurrency limits and result aggregation.

import asyncio
from mildew import ResilientTaskEngine

async def main():
    engine = ResilientTaskEngine(concurrency=4)
    await engine.start()

    await engine.submit("job_1", async_work_fn, arg1, kw=1)
    await engine.submit("job_2", sync_work_fn, arg2)

    await asyncio.sleep(0.5)

    print(engine.results["job_1"])

    await engine.stop()

asyncio.run(main())

7. Metrics Tracking

The Metrics collector records timestamps, success rates, and execution latency in a thread-safe sliding window.

from mildew import Metrics

metrics = Metrics(win_size=60.0)

metrics.mark(success=True, latency=0.042)
metrics.mark(success=False, latency=0.120)

stats = metrics.stats()
print(stats)

---

Configuration Reference

Module	Parameter	Type	Default	Description
CircuitBreaker	fail_threshold	int	5	Minimum total calls before computing failure rate.
	fail_rate	float	0.5	Failure threshold percentage (0.0 to 1.0) to open circuit.
	recovery_time	float	10.0	Cooldown duration (seconds) in OPEN state before testing.
	half_open_max	int	3	Consecutive successes needed in HALF_OPEN to close circuit.
RetryPolicy	max_attempts	int	3	Total call attempts including initial try.
	delay	float	0.1	Base delay (seconds) before first retry.
	backoff	float	2.0	Multiplier for exponential backoff.
	jitter	bool	True	Adds uniform random noise (0 to current delay).
RateLimiter	rate	int	--	Total token capacity allowed in time window.
	per	float	1.0	Window period duration in seconds.
Bulkhead	max_concurrent	int	--	Maximum concurrent execution slots.

---

Running Tests

Execute unit and integration tests using pytest:

pytest tests/

---

License

This project is licensed under the terms of THE BEER-WARE LICENSE (Revision 42). See the LICENSE file for the full license text.