soev for developers
WarrenReference

warren.retry_management

Retry management: the worker that stores soft-failed messages and republishes them after their delay. A lease (for example MongoRetryLease) lets one instance schedule at a time.

Classes

MongoRetryLease

classMongoRetryLease(client, *, database_name, collection_name)
paramclientAsyncMongoClient
paramdatabase_namestr
paramcollection_namestr
methodacquire_or_renewasync (holder_id, now, expires_at) -> bool
paramholder_idstr
paramnowfloat
paramexpires_atfloat

RetryWorker

classRetryWorker(worker_name, *, retry_store, republish_publisher, message_key_func=None, scoping_enabled=False, lease=None, lease_ttl_seconds=30, clock=time.time, sleep=asyncio.sleep)

Worker that manages retry scheduling for soft-failed messages.

Observes the observer exchange and acts only on soft-failure envelopes. It ignores all other messages. For each envelope:

  1. Persists to store (durability).
  2. Schedules asyncio.call_later for delayed republishing.
  3. On timer: fetches message, republishes, deletes from store.

On startup (via schedule_pending), loads scheduling metadata for persisted messages and reschedules remaining delays. With a lease, call start to acquire ownership and poll for takeover. Standbys only persist. Omitting the lease is intended for single-process use.

paramworker_namestr

Worker identifier.

paramretry_storeDocumentStoreInterface

Persistent store for messages awaiting retry. Must use "retry_key" as its doc_id_field.

paramrepublish_publisherPublisherInterface

Publisher targeting the processing exchange.

parammessage_key_funcCallable[[dict], str] | None
= None

Function to extract a composite key from a message dict. Defaults to building key from job_id/doc_id/part_idx using build_message_key.

paramscoping_enabledbool
= False

Prefix the retry key with the scope of the message (s:<scope>:) when the message has a valid scope.

paramleaseRetryLeaseInterface | None
= None

Lease that lets one instance schedule and republish at a time. None means single-process use: this instance always schedules.

paramlease_ttl_secondsint
= 30

Lifetime in seconds of the lease.

paramclockCallable[[], float]
= time.time

Current time in seconds. Replace it in tests.

paramsleepCallable[[float], Awaitable[None]]
= asyncio.sleep

Delay function for the lease loop. Replace it in tests.

Raises:

  • ValueError: If retry_store's doc_id_field is not "retry_key", or if lease_ttl_seconds is 0 or less.
attributeREQUIRED_DOC_ID_FIELDstr
= 'retry_key'
method__call__async (message) -> dict | None

Unwrap soft-failure envelope, persist, and schedule.

parammessagedict

Soft-failure wrapper with data_type: "soft-failure" and the failed message in data.

Returns

dict | None

None -- no downstream publish.

methodschedule_pendingasync () -> None

Load persisted retries and schedule timers.

Called on startup after setup. Queries the retry store for all persisted messages, computes remaining delay, and schedules timers. Messages past their fire time are republished immediately.

methodstartasync () -> None

Start lease polling and recover persisted retries.

methodshutdownasync () -> None

Stop lease polling and cancel retries, leaving persisted work intact.

RetryWorkerRunner

classRetryWorkerRunner(config, worker_name, *, exchange, republish_exchange=None, retry_store=None, republish_publisher=None, consumer_manager_factory=None, message_key_func=None, infra=None)

Runs a RetryWorker with its lifecycle hooks.

Creates infrastructure from RuntimeConfig and builds default components in setup(). Inject custom components to override defaults (e.g. for testing).

paramconfigRuntimeConfig

runtime infrastructure configuration.

paramworker_namestr

unique identifier for this retry worker.

paramexchangeRMQExchangeConfig

The observer exchange (observer_exchange(pipeline.exchange)).

paramrepublish_exchangeRMQExchangeConfig | None
= None

The data exchange (pipeline.exchange). Defaults to exchange, which is correct for fanout and topic only.

paramretry_storeDocumentStoreInterface | None
= None

optional override for the retry envelope store. Default: CachedDocumentStore wrapping MongoDB + Redis.

paramrepublish_publisherPublisherInterface | None
= None

optional override for the exchange publisher. Default: a publisher of config.backend (from create_publisher) on republish_exchange, with ReplayRouter.

paramconsumer_manager_factoryConsumerManagerFactory | None
= None

optional override for the consumer manager factory. Default: the consumer manager of config.backend (from create_consumer_manager) on the queue of the retry worker, bound to exchange.

parammessage_key_funcCallable[[dict], str] | None
= None

optional function to extract a composite key from a message dict. Passed to RetryWorker.

paraminfraRuntimeInfra | None
= None

optional shared infrastructure. When given, the runner uses it and does not close it; the caller owns its lifetime. Needed when several runners share one process (backend: memory).

methodsetupasync () -> None

Create infrastructure, build defaults, wire the worker.

  1. Create infrastructure (MongoDB, Redis, pubsub connection)
  2. Build default retry store, publisher, consumer factory for any components not injected
  3. Set up the republish publisher
  4. Create the RetryWorker
  5. Create and set up the consumer manager
  6. Schedule pending retries from the store

Protocols

RetryLeaseInterface

protocolRetryLeaseInterface
methodacquire_or_renewasync (holder_id, now, expires_at) -> bool

Claim an expired lease or extend this holder's lease.

paramholder_idstr
paramnowfloat
paramexpires_atfloat

Constants

RETRY_WORKER_TYPE

constantRETRY_WORKER_TYPE: str = 'retry_worker'

warren.processors

Framework processor protocol.

warren.runtime

Runtime layer for the distributed processing framework.

On this page

ClassesMongoRetryLeaseRetryWorkerRetryWorkerRunnerProtocolsRetryLeaseInterfaceConstantsRETRY_WORKER_TYPE