Skip to content

Docket in Production

This page covers configuring workers, connecting to Redis, managing state and results, monitoring, and operational tools for running Docket in production.

Worker Configuration

Core Settings

Workers have several configuration knobs for different environments:

async with Worker(
    docket,
    name="worker-1",                                    # Unique worker identifier
    concurrency=20,                                     # Parallel task limit
    redelivery_timeout=timedelta(minutes=5),           # When tasks become redeliverable
    reconnection_delay=timedelta(seconds=5),           # Redis reconnection backoff
    minimum_check_interval=timedelta(milliseconds=100), # Polling frequency
    scheduling_resolution=timedelta(milliseconds=250),  # Future task check frequency
    schedule_automatic_tasks=True,                      # Enable perpetual task startup
    message_batch=1000,                                 # Messages per Redis command
) as worker:
    await worker.run_forever()

redelivery_timeout sets when a task becomes eligible for redelivery, not when redelivery happens. Each worker sweeps for eligible tasks on a timer at a quarter of the timeout, jittered 0.75–1.25×. A task therefore waits up to about 31% past the timeout before another worker claims it, and usually less. A sweep walks a very long pending list in bounded steps, one per poll pass, so that walk can add time beyond that.

message_batch caps how many messages one Redis command may claim, for both the delivery read and the redelivery sweep. A larger batch costs fewer round trips. It also makes Redis serialize that many whole messages into one reply, and makes each sweep read about ten times the batch in pending-list entries. A burst larger than one batch still drains in full, because the worker reads again while slots stay free.

Environment Variable Configuration

The worker CLI supports environment variables for its configuration:

# Core docket settings
export DOCKET_NAME=orders
export DOCKET_URL=redis://redis.production.com:6379/0

# Worker settings
export DOCKET_WORKER_NAME=orders-worker-1
export DOCKET_WORKER_CONCURRENCY=50
export DOCKET_WORKER_MESSAGE_BATCH=1000
export DOCKET_WORKER_REDELIVERY_TIMEOUT=10m
export DOCKET_WORKER_RECONNECTION_DELAY=5s
export DOCKET_WORKER_MINIMUM_CHECK_INTERVAL=100ms
export DOCKET_WORKER_SCHEDULING_RESOLUTION=250ms

# Monitoring
export DOCKET_WORKER_HEALTHCHECK_PORT=8080
export DOCKET_WORKER_METRICS_PORT=9090

# Logging
export DOCKET_LOGGING_LEVEL=INFO
export DOCKET_LOGGING_FORMAT=json

# Task modules
export DOCKET_TASKS=myapp.tasks:production_tasks

To install worker-level dependencies, set DOCKET_WORKER_DEPENDENCIES=myapp.dependencies:worker_dependencies, pointing to an exported mapping or sequence of dependency instances. These environment variables configure the CLI; direct Worker.run() calls take explicit arguments.

CLI Usage

Run workers in production using the CLI:

# Basic worker
docket worker --tasks myapp.tasks:all_tasks

# Production worker with full configuration
docket worker \
  --docket orders \
  --url redis://redis.prod.com:6379/0 \
  --name orders-worker-1 \
  --concurrency 50 \
  --redelivery-timeout 10m \
  --healthcheck-port 8080 \
  --metrics-port 9090 \
  --logging-format json \
  --tasks myapp.tasks:production_tasks

Use --dependencies myapp.dependencies:worker_dependencies to load a dependency collection, overriding DOCKET_WORKER_DEPENDENCIES if set. See Standalone workers for the collection definition and the equivalent Python entry point using Worker.run().

Signal Handling

Workers started through the CLI or Worker.run() catch SIGTERM and SIGINT and shut down gracefully — they stop accepting new tasks and wait for in-flight tasks to finish before exiting. On container orchestrators like Kubernetes, set terminationGracePeriodSeconds to be longer than your slowest expected task so the worker has time to drain. Tasks that don't finish before the grace period expires will be redelivered to other workers based on redelivery_timeout.

When embedding a worker in another framework, use worker.run_forever() or worker.run_until_finished() within the worker's async context manager. These methods leave signal handling to the host framework.

Connection Management

Redis Connection Pools

Docket builds and manages its own Redis connection pool from the url. Tune the pool with standard redis-py options in the URL's query string — for example, setting max_connections to match or exceed your worker concurrency:

async with Docket(
    name="orders",
    url="redis://redis.prod.com:6379/0?max_connections=50&health_check_interval=30",
) as docket:
    pass

Docket defaults the connection read timeout (socket_timeout) to None so its blocking reads — worker polling, the strike-stream monitor, and execution state/progress streams — aren't cut short by redis-py 8's 5-second default. If you set socket_timeout in the URL it takes precedence, so keep it comfortably above your longest blocking read (the strike monitor blocks for 60 seconds) or leave it unset; redis-py's TCP keepalive still detects dead connections either way.

Redis Cluster Support

Docket supports Redis Cluster using the redis+cluster:// URL scheme:

# Connect to Redis Cluster
async with Docket(
    name="orders",
    url="redis+cluster://cluster-node-1:6379/0"
) as docket:
    pass

# With authentication
async with Docket(
    name="orders",
    url="redis+cluster://user:password@cluster-node-1:6379/0"
) as docket:
    pass

# TLS connection to cluster
async with Docket(
    name="orders",
    url="rediss+cluster://cluster-node-1:6379/0"
) as docket:
    pass

When using cluster mode, Docket automatically:

  • Uses hash-tagged keys ({docket_name}:*) to ensure all keys hash to the same slot
  • Manages cluster client lifecycle and connection distribution
  • Handles pub/sub through a dedicated node connection (cluster pub/sub limitation)

Note: All Docket data for a single docket name will be stored on the same cluster shard. This ensures atomicity for Lua scripts and simplifies data management, but means individual dockets don't benefit from cluster data distribution.

Redis Sentinel Support

Docket supports Redis Sentinel master discovery using the redis+sentinel:// URL scheme. The URL lists the Sentinel daemons (defaulting to port 26379) and names the monitored master group; Docket resolves the current master through the Sentinels and follows failover automatically:

# Discover the master named "mymaster" through two Sentinels
async with Docket(
    name="orders",
    url="redis+sentinel://sentinel-a:26379,sentinel-b:26379/mymaster/0"
) as docket:
    pass

# With authentication: the URL credentials apply to the data nodes, and the
# sentinel_username/sentinel_password query parameters to the Sentinels
async with Docket(
    name="orders",
    url=(
        "redis+sentinel://user:password@sentinel-a:26379,sentinel-b:26379"
        "/mymaster/0?sentinel_password=sentinelsecret"
    ),
) as docket:
    pass

# TLS for both the data nodes and the Sentinels
async with Docket(
    name="orders",
    url="rediss+sentinel://sentinel-a:26379,sentinel-b:26379/mymaster/0"
) as docket:
    pass

Standard redis-py connection options in the query string (see Redis Connection Pools above) apply to the data-node pool just as they do for standalone URLs — for example redis+sentinel://sentinel-a:26379/mymaster/0?max_connections=50. The database index comes from the path (/mymaster/0) and the data-node credentials from the URL userinfo.

Because Docket disables the read timeout for its long blocking reads, the Sentinel data-node pool defaults to tight TCP keepalive so that a master which dies silently — a network partition or frozen host that never closes the connection — is noticed within seconds rather than waiting on the OS-default keepalive, letting the worker follow the Sentinel failover promptly.

Redis Outages

A worker keeps running through Redis trouble. When Redis drops the connection, times out, or refuses a command (NOREPLICAS or MISCONF when it cannot persist, READONLY after a failover), the worker logs a warning, increments docket_redis_disruptions, lets its in-flight tasks finish, waits reconnection_delay, and reconnects. It repeats that for as long as the outage lasts and never exits because of Redis. A task whose acknowledgement failed during the outage stays pending and runs again after redelivery_timeout, the same at-least-once delivery you get when a worker crashes.

This covers every error redis-py raises, including ones that will not clear on their own, such as a wrong password or host in the URL, or a key of the wrong type under the docket's prefix. Those show up as the same warning with a traceback on every retry, from the first poll onward, so alert on docket_redis_disruptions to catch them. Only an error from outside redis-py, a bug in docket or in a dependency, stops a running worker.

Authentication

Docket supports Redis authentication via URL credentials:

# Password-only authentication (Redis default user)
docket_url = "redis://:password@redis.prod.com:6379/0"

# Username and password authentication (Redis 6.0+ ACL)
docket_url = "redis://myuser:mypassword@redis.prod.com:6379/0"

Credential providers

When the credentials rotate, as with Azure Entra ID tokens, a URL can't carry them. Pass a redis-py CredentialProvider instead, and leave the credentials out of the URL:

from redis_entraid.cred_provider import create_from_default_azure_credential

from docket import Docket

credential_provider = create_from_default_azure_credential(
    ("https://redis.azure.com/.default",)
)

async with Docket(
    name="orders",
    url="rediss://my-cache.redis.azure.net:10000/0",
    credential_provider=credential_provider,
) as docket:
    ...

Docket hands the provider to every connection it opens: the data and pub/sub connections, the result store, and the strike list. A URL with credentials and a provider together raise a ValueError. With a streaming provider, like redis-entraid's, docket re-authenticates its open connections when the token is refreshed. A pub/sub connection can't re-authenticate while it is subscribed, so it keeps the token it connected with. If the server drops it when that token expires, docket sees the same lost connection as when Redis restarts.

Give docket a provider of its own. A streaming provider like redis-entraid's keeps one refresh callback, so a redis-py client of yours that shares the provider takes the refreshes away from docket.

For Sentinel URLs the provider authenticates the data nodes, and the Sentinels keep the sentinel_username/sentinel_password from the URL. memory:// ignores it, and the docket CLI doesn't expose it, so run workers from Python (see Worker) when you need it.

ACL Configuration

When using Redis ACLs with a restricted user, grant access to the key pattern matching your docket name.

Standalone Redis: Keys follow the pattern {docket_name}:*

# Create a user with restricted permissions for a docket named "orders"
ACL SETUSER docket-user on >secure-password ~orders:* &orders:* +@all

Redis Cluster: Keys are hash-tagged with curly braces {docket_name}:*

# For cluster mode, the pattern includes the hash tag braces
ACL SETUSER docket-user on >secure-password ~{orders}:* &{orders}:* +@all

The required permissions are:

  • Key pattern: ~{docket_name}:* (standalone) or ~\{docket_name\}:* (cluster) - matches all Redis keys used by Docket
  • Channel pattern: &{docket_name}:* (standalone) or &\{docket_name\}:* (cluster) - required for task cancellation pub/sub
  • Commands: +@all or the specific command categories Docket uses

For production deployments, you may restrict to only the required command categories:

# More restrictive command permissions (standalone)
ACL SETUSER docket-user on >secure-password \
  ~orders:* &orders:* \
  +@read +@write +@set +@sortedset +@hash +@stream +@pubsub +@scripting +@connection

# More restrictive command permissions (cluster)
ACL SETUSER docket-user on >secure-password \
  ~{orders}:* &{orders}:* \
  +@read +@write +@set +@sortedset +@hash +@stream +@pubsub +@scripting +@connection

Valkey Support

Docket also works with Valkey (Redis fork). Valkey uses the Redis protocol, so connect with a redis:// or rediss:// URL:

export DOCKET_URL=redis://valkey.prod.com:6379/0

State and Result Storage

Docket tracks execution state and stores task results by default. You can configure or disable this behavior based on your observability and throughput requirements.

Execution State Tracking

By default, Docket stores execution state (SCHEDULED, QUEUED, RUNNING, COMPLETED, FAILED) in Redis with a 15-minute TTL:

async with Docket(
    name="orders",
    url="redis://localhost:6379/0",
    execution_ttl=timedelta(minutes=15)  # Default
) as docket:
    # State records expire 15 minutes after task completion
    pass

The execution_ttl controls:

  • How long state records persist in Redis after task completion
  • How long result data is retained (see Result Storage below)

A task added under the key of one that has ended starts a record of its own, which expires only once that task ends.

It does not affect progress information, which Docket deletes when the task ends.

Fire-and-Forget Mode

For maximum throughput in high-volume scenarios, disable state tracking entirely:

async with Docket(
    name="high-throughput",
    url="redis://localhost:6379/0",
    execution_ttl=timedelta(0)  # Disable state persistence
) as docket:
    # State records are deleted when each task ends, and results are not stored
    for event in events:
        await docket.add(process_event)(event)

With execution_ttl=0:

  • No state records after a task ends: Docket writes a task's state record while it is scheduled, queued, or running, and deletes the record when the task ends
  • No result storage: Task return values are not persisted
  • Progress tracking still works: Docket writes progress updates while the task runs and deletes them when the task ends, the same as with any execution_ttl
  • Maximum throughput: Minimizes Redis operations per task
  • get_result() unavailable: Cannot retrieve task results

Result Storage Configuration

Task results are stored using the py-key-value-aio library. By default, Docket uses RedisStore but you can provide a custom storage backend:

from key_value.stores.redis import RedisStore
from key_value.stores.memory import MemoryStore

# Default: Redis-backed result storage
async with Docket(
    name="production",
    url="redis://localhost:6379/0",
    result_storage=RedisStore(url="redis://localhost:6379/1")  # Separate DB
) as docket:
    pass

# Alternative: In-memory storage for testing
async with Docket(
    name="test",
    url="redis://localhost:6379/0",
    result_storage=MemoryStore()
) as docket:
    pass

Custom storage backends must implement the KeyValueStore protocol from py-key-value-aio.

Result Storage Tips

Separate Redis Database: Store results in a different Redis database than task queues:

result_storage = RedisStore(url="redis://localhost:6379/1")  # DB 1 for results
docket = Docket(
    name="orders",
    url="redis://localhost:6379/0",  # DB 0 for queues
    result_storage=result_storage
)

TTL Management: Results inherit the execution_ttl setting. Adjust based on how long you need results:

# Keep results for 1 hour
async with Docket(execution_ttl=timedelta(hours=1)) as docket:
    execution = await docket.add(generate_report)()
    # Result available for 1 hour after completion

None Returns: Tasks that return None don't write to result storage, saving space:

async def send_notification(user_id: str) -> None:
    await send_email(user_id)
    # No result stored - task returns None

Large Results: For large result data, consider storing references instead:

# Instead of returning large data
async def process_large_dataset(dataset_id: str) -> dict:
    data = await expensive_computation()
    return data  # Large object stored in Redis

# Store a reference
async def process_large_dataset(dataset_id: str) -> str:
    data = await expensive_computation()
    s3_key = await store_in_s3(data)
    return s3_key  # Small string stored in Redis

Monitoring State and Result Storage

Track Redis memory usage for state and result data:

# Check memory usage
redis-cli info memory

# Count state records
redis-cli --scan --pattern "docket:runs:*" | wc -l

# Count result keys
redis-cli --scan --pattern "docket:results:*" | wc -l

If memory usage is high:

  • Reduce execution_ttl to expire data sooner
  • Use execution_ttl=0 for fire-and-forget tasks
  • Store large results externally (S3, database) and return references
  • Separate result storage to a different Redis instance

Monitoring and Observability

Prometheus Metrics

Enable Prometheus metrics with the --metrics-port option:

docket worker --metrics-port 9090

Available metrics include:

Task Counters

  • docket_tasks_added - Tasks scheduled
  • docket_tasks_started - Tasks begun execution
  • docket_tasks_succeeded - Successfully completed tasks
  • docket_tasks_failed - Failed tasks
  • docket_tasks_retried - Retry attempts
  • docket_tasks_stricken - Tasks blocked by strikes

Task Timing

  • docket_task_duration - Histogram of task execution times
  • docket_task_punctuality - How close tasks run to their scheduled time

System Health

  • docket_queue_depth - Tasks ready for immediate execution
  • docket_schedule_depth - Tasks scheduled for future execution
  • docket_tasks_running - Currently executing tasks
  • docket_redis_disruptions - Times Redis dropped, timed out, or refused a command that the worker then retried
  • docket_strikes_in_effect - Active strike rules

All metrics include labels for docket name, worker name, and task function name.

Health Checks

Enable health check endpoints:

docket worker --healthcheck-port 8080

The health check endpoint (/) returns 200 OK when the worker is healthy and able to process tasks.

OpenTelemetry Traces

Docket automatically creates OpenTelemetry spans for task execution:

  • Span name: docket.task.{function_name}
  • Attributes: docket name, worker name, task key, attempt number
  • Status: Success/failure with error details
  • Duration: Complete task execution time

Configure your OpenTelemetry exporter to send traces to your observability platform. See the OpenTelemetry Python documentation for configuration examples with various backends like Jaeger, Zipkin, or cloud providers.

Structured Logging

Configure structured logging for production:

# JSON logs for log aggregation
docket worker --logging-format json --logging-level info

# Plain logs for simple deployments
docket worker --logging-format plain --logging-level warning

Log entries include:

  • Task execution start/completion
  • Error details with stack traces
  • Worker lifecycle events
  • Redis connection status
  • Strike/restore operations

Striking and Restoring Tasks

Striking temporarily disables tasks without redeploying code.

Striking Entire Task Types

Disable all instances of a specific task:

# Disable all order processing during maintenance
await docket.strike(process_order)

# Orders added during this time won't be processed
await docket.add(process_order)(order_id=12345)  # Won't run
await docket.add(process_order)(order_id=67890)  # Won't run

# Re-enable when ready
await docket.restore(process_order)

Striking by Parameter Values

Disable tasks based on their arguments using comparison operators:

# Block all tasks for a problematic customer
await docket.strike(None, "customer_id", "==", "12345")

# Block low-priority work during high load
await docket.strike(process_order, "priority", "<=", "low")

# Block all orders above a certain value during fraud investigation
await docket.strike(process_payment, "amount", ">", 10000)

# Later, restore them
await docket.restore(None, "customer_id", "==", "12345")
await docket.restore(process_order, "priority", "<=", "low")

Supported operators include ==, !=, <, <=, >, >=.

Striking Specific Task-Parameter Combinations

Target very specific scenarios:

# Block one customer's orders, and every high-value order
await docket.strike(process_order, "customer_id", "==", "12345")
await docket.strike(process_order, "amount", ">", 1000)

# This order won't run (blocked customer)
await docket.add(process_order)(customer_id="12345", amount=500)

# This order won't run (high amount, from any customer)
await docket.add(process_order)(customer_id="67890", amount=2000)

# This order WILL run (different customer, low amount)
await docket.add(process_order)(customer_id="67890", amount=500)

Any one strike blocks a task. Strikes do not combine into a single condition, so you cannot strike only the orders that match both the customer and the amount.

Striking from the CLI

You can also strike and restore tasks from the command line:

# Block problematic tasks immediately
docket strike problematic_function

# Block tasks for specific customers
docket strike process_order customer_id == "problematic-customer"

# Restore when issues are resolved
docket restore problematic_function

Built-in Utility Tasks

Docket provides utility tasks for smoke-testing and debugging:

from docket import tasks

# Simple trace logging
await docket.add(tasks.trace)("System startup completed")
await docket.add(tasks.trace)("Processing batch 123")

# Intentional failures for testing error handling
await docket.add(tasks.fail)("Testing error notification system")