Add reliable message queues

README.md

@@ -82,7 +82,8 @@ pip install pydocket
 ```
 
 Docket requires a [Redis](http://redis.io/) server with Streams support (which was
-introduced in Redis 5.0.0). Docket is tested with:
+introduced in Redis 5.0.0). Reliable message queues require Redis 6.2 or newer.
+Docket is tested with:
 
 - Redis 6.2, 7.4, and 8.6 (standalone and cluster modes)
 - [Valkey](https://valkey.io/) 8.1

docs/api-reference.md

@@ -1,3 +1,5 @@
 # API Reference
 
 ::: docket
+
+::: docket.queue

docs/queues.md

@@ -0,0 +1,101 @@
+# Reliable Message Queues
+
+Docket queues provide durable, at-least-once message delivery for systems that
+own their own execution and result model. Unlike Docket tasks, queue messages
+are opaque bytes: Docket does not import a function, execute the payload, or
+store a result.
+
+This is useful when Docket is the delivery layer beneath another runtime:
+
+```python
+from datetime import timedelta
+
+from docket import Docket
+
+async with Docket(name="orders") as docket:
+    queue = docket.queue("commands")
+    await queue.put(
+        "scheduled",
+        b'{"order_id": "123"}',
+        key="order:123:charge",
+    )
+```
+
+A subscription competes with other subscriptions in the same consumer group.
+Only one receives a given delivery:
+
+```python
+async with Docket(name="orders") as docket:
+    queue = docket.queue("commands")
+    async with queue.subscribe(
+        {"retry": 0, "scheduled": 1},
+        visibility_timeout=timedelta(minutes=5),
+    ) as subscription:
+        while True:
+            message = await subscription.receive()
+            try:
+                await execute_in_my_runtime(message.data)
+            except RetryableError:
+                await message.release("retry")
+            else:
+                await message.acknowledge()
+```
+
+Lower numeric topic priorities are returned first when multiple claimed
+messages are ready. Each topic is FIFO. `release()` atomically moves a message
+to another topic, which supports an immediate retry lane without losing the
+original delivery.
+
+## Delivery guarantees
+
+Queue delivery is at least once:
+
+- A claimed message remains in Redis until it is acknowledged.
+- The subscription renews visibility while the message is outstanding.
+- If the subscriber exits or loses its Redis connection, another subscriber
+  can reclaim the message after `visibility_timeout`.
+- `acknowledge()` removes the message only after the downstream runtime has
+  accepted it.
+
+Choose a visibility timeout longer than normal processing stalls and Redis
+failovers. Consumers must still be idempotent because a process can finish its
+side effect and fail before acknowledging the message.
+
+Message keys are deduplicated across every topic in a queue while the message
+is queued or in flight. For repair loops that may rediscover accepted work,
+retain a short acknowledgement tombstone:
+
+```python
+queue = docket.queue(
+    "commands",
+    acknowledgement_ttl=timedelta(hours=1),
+)
+```
+
+Publishing the same key during that period returns `False`; a new publication
+returns `True`.
+
+## Backpressure
+
+Set `max_size` on `put()` to bound the number of queued and in-flight messages
+in a topic:
+
+```python
+await queue.put("scheduled", payload, max_size=1_000)
+```
+
+The publisher waits until capacity is available. The same option on
+`release()` prevents an immediate-retry lane from exceeding its bound while
+keeping the source delivery claimable until the atomic move succeeds.
+
+## Operations
+
+Queues use Redis Streams consumer groups and require Redis 6.2 or newer.
+Subscriptions retry transient Redis errors, recreate expired consumer groups,
+renew claims, and reclaim abandoned deliveries. They use the Docket's existing
+standalone, cluster, Sentinel, authentication, and connection-pool
+configuration.
+
+Use the same Docket name, queue name, and consumer group for replicas that
+should share work. A queue supports one logical consumer group: acknowledged
+messages are deleted rather than broadcast to independent groups.

mkdocs.yml

@@ -76,6 +76,7 @@ nav:
   - Dependency Injection: dependency-injection.md
   - Task Design Patterns: task-patterns.md
   - Task Observability: observability.md
+  - Reliable Message Queues: queues.md
   - Testing with Docket: testing.md
   - Docket in Production: production.md
   - API Reference: api-reference.md

pyproject.toml

@@ -95,6 +95,9 @@ docket = "docket.__main__:app"
 [tool.hatch.version]
 source = "vcs"
 
+[tool.hatch.version.raw-options]
+fallback_version = "0.0.0"
+
 [tool.hatch.metadata]
 allow-direct-references = true
 

src/docket/__init__.py

@@ -39,6 +39,7 @@
     TaskCall,
 )
 from .strikelist import StrikeList
+from .queue import Queue, QueueMessage, QueueSubscription
 from .worker import Worker
 from . import testing
 
@@ -63,6 +64,9 @@
     "Logged",
     "Perpetual",
     "Progress",
+    "Queue",
+    "QueueMessage",
+    "QueueSubscription",
     "Retry",
     "Shared",
     "StrikeList",

src/docket/_queue_scripts.py

@@ -0,0 +1,99 @@
+"""Atomic Redis operations used by reliable message queues."""
+
+from ._lua import Arg, Key, redis_script
+from ._redis import RedisClient
+
+
+@redis_script
+async def put_message(
+    redis: RedisClient,
+    *,
+    stream_key: Key[str],
+    deduplication_key: Key[str],
+    message_key: Arg[str],
+    data: Arg[bytes],
+    max_size: Arg[int],
+    now_timestamp: Arg[float],
+) -> bytes:
+    """
+    redis.call('ZREMRANGEBYSCORE', deduplication_key, 1, now_timestamp)
+    if redis.call('ZSCORE', deduplication_key, message_key) then
+        return 'DUPLICATE'
+    end
+    if max_size > 0 and redis.call('XLEN', stream_key) >= max_size then
+        return 'FULL'
+    end
+
+    local message_id = redis.call(
+        'XADD', stream_key, '*', 'key', message_key, 'data', data
+    )
+    redis.call('EXPIRE', stream_key, 2147483647)
+    redis.call('ZADD', deduplication_key, 0, message_key)
+    redis.call('EXPIRE', deduplication_key, 2147483647)
+    return message_id
+    """
+    ...
+
+
+@redis_script
+async def acknowledge_message(
+    redis: RedisClient,
+    *,
+    stream_key: Key[str],
+    deduplication_key: Key[str],
+    group_name: Arg[str],
+    message_id: Arg[bytes],
+    message_key: Arg[str],
+    idle_ttl_seconds: Arg[int],
+    acknowledged_until: Arg[float],
+) -> int:
+    """
+    redis.call('XACK', stream_key, group_name, message_id)
+    redis.call('XDEL', stream_key, message_id)
+    if acknowledged_until > 0 then
+        redis.call('ZADD', deduplication_key, acknowledged_until, message_key)
+        redis.call('EXPIRE', deduplication_key, 2147483647)
+    else
+        redis.call('ZREM', deduplication_key, message_key)
+    end
+    if redis.call('XLEN', stream_key) == 0 then
+        redis.call('EXPIRE', stream_key, idle_ttl_seconds)
+    end
+    return 1
+    """
+    ...
+
+
+@redis_script
+async def release_message(
+    redis: RedisClient,
+    *,
+    source_stream_key: Key[str],
+    destination_stream_key: Key[str],
+    group_name: Arg[str],
+    message_id: Arg[bytes],
+    message_key: Arg[str],
+    data: Arg[bytes],
+    max_size: Arg[int],
+    idle_ttl_seconds: Arg[int],
+) -> bytes:
+    """
+    if source_stream_key ~= destination_stream_key
+        and max_size > 0
+        and redis.call('XLEN', destination_stream_key) >= max_size
+    then
+        return 'FULL'
+    end
+
+    redis.call('XACK', source_stream_key, group_name, message_id)
+    redis.call('XDEL', source_stream_key, message_id)
+    local new_message_id = redis.call(
+        'XADD', destination_stream_key, '*', 'key', message_key, 'data', data
+    )
+    redis.call('EXPIRE', destination_stream_key, 2147483647)
+    if redis.call('XLEN', source_stream_key) == 0 then
+        redis.call('EXPIRE', source_stream_key, idle_ttl_seconds)
+    end
+    return new_message_id
+    """
+    ...

src/docket/docket.py

@@ -60,6 +60,7 @@
     Strike,
     StrikeList,
 )
+from .queue import DocketQueueMixin
 
 logger: logging.Logger = logging.getLogger(__name__)
 tracer: trace.Tracer = trace.get_tracer(__name__)
@@ -126,7 +127,7 @@ async def _cancel_task(
 TaskCollection = Iterable[TaskFunction]
 
 
-class Docket(DocketSnapshotMixin):
+class Docket(DocketQueueMixin, DocketSnapshotMixin):
     """A Docket represents a collection of tasks that may be scheduled for later
     execution.  With a Docket, you can add, replace, and cancel tasks.
     Example:
@@ -199,7 +200,6 @@ def worker_group_name(self) -> str:
     @property
     def prefix(self) -> str:
         """Return the key prefix for this docket.
-
         All Redis keys for this docket are prefixed with this value.
 
         For Redis Cluster mode, returns a hash-tagged prefix like "{myapp}"