retry

retry adds per-attempt timeout, retry/backoff, and optional semaphore-based concurrency control around async and sync callables. It works for ordinary functions without creating an EventBus. Sync callables stay sync and block synchronously for retry sleeps and semaphore waits. Async callables stay async and await retry sleeps and semaphore waits. Supported callable forms:

Python: sync and async functions, methods, and lambdas via @retry(...) or retry(...)(fn).
TypeScript: sync and async functions and arrow functions via retry(...)(fn), plus sync and async class methods via @retry(...).
Rust: sync and async free functions and methods via abxbus::retry!; call those retried functions from Rust closures such as EventBus handlers.
Go: retry is not implemented yet.

Signature

Python
TypeScript
Rust
Go

def retry(
    retry_after: float = 0,
    max_attempts: int = 1,
    timeout: float | None = None,
    slow_timeout: float | None = None,
    retry_on_errors: list[type[Exception] | re.Pattern[str]] | tuple[type[Exception] | re.Pattern[str], ...] | None = None,
    retry_backoff_factor: float = 1.0,
    semaphore_limit: int | None = None,
    semaphore_name: str | Callable[..., str] | None = None,
    semaphore_lax: bool = True,
    semaphore_scope: Literal['multiprocess', 'global', 'class', 'instance'] = 'global',
    semaphore_timeout: float | None = None,
) -> Callable[[Callable[P, T]], Callable[P, T]]

retry({
  max_attempts?: number,                                                // default: 1
  retry_after?: number,                                                 // default: 0 (seconds)
  retry_backoff_factor?: number,                                        // default: 1.0
  retry_on_errors?: Array<(new (...args) => Error) | RegExp | string>,  // default: retry any error
  timeout?: number | null,                                              // default: no per-attempt timeout
  slow_timeout?: number | null,                                         // default: disabled
  semaphore_limit?: number | null,                                      // default: no semaphore limit
  semaphore_name?: string | ((...args: any[]) => string) | null,        // default: function name
  semaphore_lax?: boolean,                                              // default: true
  semaphore_scope?: 'multiprocess' | 'global' | 'class' | 'instance',   // default: 'global'
  semaphore_timeout?: number | null,                                    // default: derived when timeout + limit are set
})

abxbus::retry! {
    max_attempts = 3,
    retry_after = 1.0,
    retry_backoff_factor = 2.0,
    timeout = 5.0,
    slow_timeout = 20.0,
    retry_if = should_retry,
    semaphore_limit = 2,
    semaphore_name = "api",
    semaphore_lax = true,
    semaphore_scope = "global",
    semaphore_timeout = 10.0;

    async fn run(&self) -> Result<T, E> {
        /* body */
    }
}

// Not implemented in abxbus-go yet.
//
// Use bus/event/handler timeout fields for outer execution budgets today.
// Per-attempt retry/backoff/semaphore behavior should be wrapped inside
// your Go handler until retry parity is implemented.

Options

Option	Description
`max_attempts`	Total attempts including the first call (`1` disables retries).
`retry_after`	Base delay between retries, in seconds.
`retry_backoff_factor`	Delay multiplier applied after each failed attempt.
`retry_on_errors`	Optional matcher list to restrict which errors are retried.
`timeout`	Per-attempt timeout in seconds (`None`/`undefined` means no per-attempt timeout).
`slow_timeout`	Warning threshold for a decorated call, in seconds. Warnings are throttled to at most one every 2 seconds per decorated method/function.
`semaphore_limit`	Max concurrent executions sharing the same semaphore.
`semaphore_name`	Semaphore key (string or function deriving a key from call args).
`semaphore_scope`	Semaphore sharing scope (`multiprocess`, `global`, `class`, `instance`).
`semaphore_timeout`	Max wait time for semaphore acquisition before timeout/lax fallback.
`semaphore_lax`	If true, continue execution without semaphore limit when acquisition times out.

Example: Standalone function

Python
TypeScript
Rust

from abxbus.retry import retry

@retry(max_attempts=3, retry_after=1, timeout=5)
async def fetch_with_retry(url: str) -> dict:
    return await fetch_json(url)

import { retry } from 'abxbus/retry'

async function fetchJsonWithRetry(url: string): Promise<Record<string, unknown>> {
  return await fetchJson(url)
}

const fetchWithRetry = retry({ max_attempts: 3, retry_after: 1, timeout: 5 })(fetchJsonWithRetry)

#[derive(Debug)]
enum ApiError {
    Retry(abxbus::retry::RetryError),
    Transport(String),
}

impl From<abxbus::retry::RetryError> for ApiError {
    fn from(error: abxbus::retry::RetryError) -> Self {
        Self::Retry(error)
    }
}

abxbus::retry! {
    max_attempts = 3, retry_after = 1.0, timeout = 5.0;
    async fn fetch_with_retry(url: String) -> Result<serde_json::Value, ApiError> {
        fetch_json(url).await.map_err(ApiError::Transport)
    }
}

Example: Inline wrapper

Python
TypeScript
Rust
Go

from abxbus import EventBus, BaseEvent
from abxbus.retry import retry

class FetchEvent(BaseEvent[dict]):
    url: str

bus = EventBus('AppBus')

async def fetch_with_retry(event: FetchEvent) -> dict:
    return await fetch_json(event.url)

bus.on(
    FetchEvent,
    retry(max_attempts=3, retry_after=1, timeout=5)(fetch_with_retry),
)

import { BaseEvent, EventBus, retry } from 'abxbus'
import { z } from 'zod'

const FetchEvent = BaseEvent.extend('FetchEvent', {
  url: z.string(),
  event_result_type: z.record(z.string(), z.unknown()),
})

const bus = new EventBus('AppBus')

bus.on(
  FetchEvent,
  retry({ max_attempts: 3, retry_after: 1, timeout: 5 })(async (event) => {
    return await fetchJson(event.url)
  })
)

use abxbus::{event, event_bus::EventBus};

struct Api;

impl Api {
    abxbus::retry! {
        max_attempts = 3, retry_after = 1.0, timeout = 5.0;
        async fn fetch_with_retry(&self, event: FetchEvent) -> Result<serde_json::Value, ApiError> {
            fetch_json(event.url).await.map_err(ApiError::Transport)
        }
    }
}

let api = Api;
let bus = EventBus::new(Some("AppBus".to_string()));
bus.on(FetchEvent, move |event: FetchEvent| async move {
    api.fetch_with_retry(event).await
});

// retry decorators/method wrappers are not implemented in abxbus-go yet.
//
// Put retry/backoff code in a helper function or inside the method body,
// then register that method as a normal EventBus handler.

Example: Decorated class method

Python
TypeScript
Rust
Go

from abxbus.retry import retry

class ApiService:
    @retry(max_attempts=4, retry_after=1, timeout=10, semaphore_limit=2, semaphore_scope='class')
    async def get_user(self, user_id: str) -> dict:
        return await call_remote_api(user_id)

import { retry } from 'abxbus/retry'

class ApiService {
  @retry({ max_attempts: 4, retry_after: 1, timeout: 10, semaphore_limit: 2, semaphore_scope: 'class' })
  async getUser(userId: string): Promise<Record<string, unknown>> {
    return await callRemoteApi(userId)
  }
}

Without decorators, use the same closure-based wrapper on a normal function:

import { retry } from 'abxbus/retry'

async function getUser(userId: string): Promise<Record<string, unknown>> {
  return await callRemoteApi(userId)
}

const getUserWithRetry = retry({ max_attempts: 4, retry_after: 1, timeout: 10, semaphore_limit: 2 })(getUser)

The same @retry(...) method syntax works with TypeScript’s standard decorators and with legacy experimentalDecorators method decorators. For legacy decorators, enable "experimentalDecorators": true in tsconfig.json; the method keeps its original sync or async return type.

struct ApiService;

impl ApiService {
    abxbus::retry! {
        max_attempts = 4,
        retry_after = 1.0,
        timeout = 10.0,
        semaphore_limit = 2,
        semaphore_scope = "class";

        async fn get_user(&self, user_id: String) -> Result<serde_json::Value, ApiError> {
            call_remote_api(user_id).await.map_err(ApiError::Transport)
        }
    }
}

// retry decorators/method wrappers are not implemented in abxbus-go yet.
//
// Put retry/backoff code in a helper function or inside the method body,
// then register that method as a normal EventBus handler.

Behavior

Semaphore acquisition happens once per call, then all retry attempts run within that acquired slot.
Backoff delay per retry is: retry_after * retry_backoff_factor^(attempt - 1).
Retries stop immediately when the thrown error does not match retry_on_errors.
Bus/event timeouts act as outer execution budgets; retry.timeout is per-attempt.
Sync wrappers return synchronously and block synchronously for retry sleeps and semaphore waits.
Rust retry functions return Result<T, E> where E: From<abxbus::retry::RetryError>.

Runtime differences

Python and TypeScript both support multiprocess, global, class, and instance.
Rust supports multiprocess, global, class, and instance through abxbus::retry!.
Go does not include retry yet.
TypeScript uses async-context re-entrancy tracking in Node/Bun to avoid same-semaphore nested deadlocks.
retry_on_errors matching differs slightly:
- Python: exception classes or compiled regex patterns (matched against "ErrorClass: message").
- TypeScript: error constructors, error-name strings, or regex patterns.
- Rust: typed predicate function via retry_if = should_retry.

Getting Started

Features

API Reference

Concurrency Control

Integrations

Further Reading

Signature

Options

Example: Standalone function

Example: Inline wrapper

Example: Decorated class method

Behavior

Runtime differences

​Signature

​Options

​Example: Standalone function

​Example: Inline wrapper

​Example: Decorated class method

​Behavior

​Runtime differences

Signature

Options

Example: Standalone function

Example: Inline wrapper

Example: Decorated class method

Behavior

Runtime differences