From 388b8cd51c62766cfe839c6afc08e1e6b26f227c Mon Sep 17 00:00:00 2001 From: Arseniy Popov Date: Wed, 17 Jun 2026 01:43:58 +0300 Subject: [PATCH 1/2] docs: expand docs on acks --- docs/docs/sqla/design.md | 464 ------------------ docs/docs/sqlbroker/design.md | 6 +- docs/docs/sqlbroker/tutorial.md | 68 ++- .../sqlbroker/broker/registrator.py | 4 +- faststream_sqlbroker/sqlbroker/message.py | 2 +- 5 files changed, 49 insertions(+), 495 deletions(-) delete mode 100644 docs/docs/sqla/design.md diff --git a/docs/docs/sqla/design.md b/docs/docs/sqla/design.md deleted file mode 100644 index 393b1ba..0000000 --- a/docs/docs/sqla/design.md +++ /dev/null @@ -1,464 +0,0 @@ ---- -# 0.5 - API -# 2 - Release -# 3 - Contributing -# 5 - Template Page -# 10 - Default -search: - boost: 10 ---- - -!!! warning "Alpha" - `faststream-sqlbroker` is currently in alpha. - -# Design and Features - -## Message Lifecycle - -A published message starts out as `PENDING`. When it is acquired by a worker, it is marked as `PROCESSING`, which prevents it from being acquired by other worker processes. If the maximum allowed number of deliveries is configured and exceeded, the message is marked as `FAILED`. If not, the message is processed. If processing didn't raise an exception or if the message was manually Ack'ed in the handler, it is marked as `COMPLETED`. If the message was manually Nack'ed or Reject'ed or if processing raised an exception and [`AckPolicy`](../getting-started/acknowledgement.md){.internal-link} was set to `REJECT_ON_ERROR` or `NACK_ON_ERROR`, the message is Nack'ed or Reject'ed. Reject'ed messages are marked as `FAILED`. For Nack'ed messages, the retry policy determines if the message is allowed to be retried. If retry is allowed, the message is marked as `RETRYABLE`. If not, the message is marked as `FAILED`. - -`PENDING`, `PROCESSING`, and `RETRYABLE` messages reside in the main table, while `COMPLETED` and `FAILED` messages are moved to the archive table on status change. - -```mermaid -flowchart TD - PENDING -->|acquired by worker| PROCESSING - PROCESSING --> MaxDel{max_deliveries
configured
and exceeded?} - MaxDel -->|yes| FAILED - MaxDel -->|no| Result{Process message} - Result -->|Ack| COMPLETED - Result -->|Reject| FAILED - Result -->|Nack| Retry{Retry allowed?} - Retry -->|yes| RETRYABLE - Retry -->|no| FAILED - RETRYABLE -->|acquired by worker| PROCESSING -``` - -## Subscriber Internals - -On start, the subscriber spawns four types of concurrent loops: - -**1. Fetch loop** — Periodically fetches batches of `PENDING` or `RETRYABLE` messages from the database, simultaneously updating them: marking as `PROCESSING`, setting `acquired_at` to now, and incrementing `deliveries_count`. Only messages with `next_attempt_at <= now` are fetched, ordered by `next_attempt_at`. The fetched messages are placed into an internal queue. The fetch limit is the minimum of `fetch_batch_size` and the free buffer capacity (`fetch_batch_size * overfetch_factor` minus currently queued messages). If the last fetch was "full" (returned as many messages as the limit), the next fetch happens after `min_fetch_interval`; otherwise after `max_fetch_interval`. - -**2. Worker loops** (`max_workers` concurrent instances) — Each worker takes a message from the internal queue and first checks if `max_deliveries` has been exceeded; if so, the message is Rejected without processing. Otherwise, processing proceeds. Depending on the processing result, [`AckPolicy`](../getting-started/acknowledgement.md){.internal-link}, and manual Ack/Nack/Reject, the message is Ack'ed, Nack'ed, or Rejected. For Nack'ed messages, the `retry_strategy` is consulted to determine if and when the message might be retried. If allowed to be retried, the message is marked as `RETRYABLE`; otherwise as `FAILED`. Ack'ed messages are marked as `COMPLETED` and rejected messages are marked as `FAILED`. The message is then buffered for flushing. - -**3. Flush loop** — Periodically flushes the buffered message state changes to the database. `COMPLETED` and `FAILED` messages are moved from the primary table to the archive table. The state of `RETRYABLE` messages is updated in the primary table. - -**4. Release stuck loop** — Periodically releases messages that have been stuck in `PROCESSING` state for longer than `release_stuck_timeout` since `acquired_at`. These messages are marked back as `PENDING`. - -On stop, all loops are gracefully stopped. Messages that have been acquired but are not yet being processed are drained from the internal queue and marked back as `PENDING`. The subscriber waits for all tasks to complete within `graceful_timeout`, then performs a final flush. - -## Features - -### Supported Databases - -PostgreSQL, MySQL, and SQLite are currently supported. - -### Processing Guarantees - -This design adheres to the **"at least once"** processing guarantee because flushing changes to the database happens only after a processing attempt. A flush might not happen due to e.g. a crash. This might lead to messages being processed more times than allowed by the `retry_strategy`, and to the database state being inconsistent with the true number of attempts. - -### Work Sharing - -Multiple subscriber instances (in different processes or on different machines) can safely consume from the same queue without double-processing because `SELECT ... FOR UPDATE SKIP LOCKED` is utilized - -### Short-Lived Transactions - -This design opts for separate short-lived transactions instead of a single one. This first one fetches messages with `SELECT ... FOR UPDATE SKIP LOCKED` and sets their state to `PROCESSING`, which prevents them from being fetched by other processes/nodes. The other two transactions flush message state updates to the database. - -### Poison Message Protection - -Setting `max_deliveries` to a non-`None` value provides protection from the [poison message problem](https://www.rabbitmq.com/docs/quorum-queues#poison-message-handling){.external-link target="_blank"} (messages that crash the worker without the ability to catch the exception, e.g. due to OOM terminations) because `deliveries_count` is incremented and `max_deliveries` is checked prior to a processing attempt. However, this comes at the expense of potentially over-counting deliveries, especially for messages that are being processed concurrently with the poison message (a crash would leave them with incremented `deliveries_count` despite possibly not having been processed), and violating the at-most-once processing semantics. - -### Ordered Processing - -As of now, to achieve processing strictly in the order of `next_attempt_at` (or in the order of publishing if no `next_attempt_at` was provided), only one process should be consuming from the same queue and its concurrency should be set to 1 with `max_workers=1`. - -### Delayed Delivery and Retries - -[Delayed delivery](../sqlbroker/tutorial.md#delayed-delivery){.internal-link} is supported with the use of `next_attempt_at`, and the same field is set by retry strategies for [delayed retries](../sqlbroker/tutorial.md#delayed-retries){.internal-link}. - -### No Fanout - -As of now, fanout, where each of the consumer groups processes every message, is not supported. - -### No LISTEN/NOTIFY - -As of now, `LISTEN/NOTIFY` is not supported. - -## SQL Statements - -### Acquire Messages - -=== "PostgreSQL" - ```sql linenums="1" - WITH - ready AS ( - SELECT - message.id AS id, - message.queue AS queue, - message.headers AS headers, - message.payload AS payload, - message.state AS state, - message.attempts_count AS attempts_count, - message.deliveries_count AS deliveries_count, - message.created_at AS created_at, - message.first_attempt_at AS first_attempt_at, - message.next_attempt_at AS next_attempt_at, - message.last_attempt_at AS last_attempt_at, - message.acquired_at AS acquired_at - FROM - message - WHERE - ( - message.state = $4::sqlbrokermessagestate - OR message.state = $5::sqlbrokermessagestate - ) - AND message.next_attempt_at <= $6::TIMESTAMP WITHOUT TIME ZONE - AND ( - message.queue = $7::VARCHAR - OR message.queue = $8::VARCHAR - ) - ORDER BY - message.next_attempt_at - LIMIT - $9::INTEGER - FOR UPDATE - SKIP LOCKED - ), - updated AS ( - UPDATE message - SET - state = $1::sqlbrokermessagestate, - deliveries_count = (message.deliveries_count + $2::BIGINT), - acquired_at = $3::TIMESTAMP WITHOUT TIME ZONE - WHERE - message.id IN ( - SELECT - ready.id - FROM - ready - ) - RETURNING - message.id, - message.queue, - message.headers, - message.payload, - message.state, - message.attempts_count, - message.deliveries_count, - message.created_at, - message.first_attempt_at, - message.next_attempt_at, - message.last_attempt_at, - message.acquired_at - ) - SELECT - updated.id, - updated.queue, - updated.headers, - updated.payload, - updated.state, - updated.attempts_count, - updated.deliveries_count, - updated.created_at, - updated.first_attempt_at, - updated.next_attempt_at, - updated.last_attempt_at, - updated.acquired_at - FROM - updated - ORDER BY - updated.next_attempt_at - ``` - -=== "MySQL" - ```sql linenums="1" - BEGIN; - - SELECT - message.id AS id - FROM - message - WHERE - ( - message.state = % s - OR message.state = % s - ) - AND message.next_attempt_at <= % s - AND ( - message.queue = % s - OR message.queue = % s - ) - ORDER BY - message.next_attempt_at - LIMIT - % s FOR - UPDATE SKIP LOCKED; - - UPDATE message - SET - state = % s, - deliveries_count = (message.deliveries_count + % s), - acquired_at = % s - WHERE - message.id IN (% s); - - COMMIT; - ``` - -=== "SQLite" - ```sql linenums="1" - WITH ready AS ( - SELECT - message.id AS id - FROM - message - WHERE - (message.state = ? OR message.state = ?) - AND message.next_attempt_at <= ? - AND message.queue = ? - ORDER BY - message.next_attempt_at - LIMIT ? OFFSET ? - ) - UPDATE message - SET - state = ?, - deliveries_count = (message.deliveries_count + ?), - acquired_at = ? - WHERE - message.id IN (SELECT ready.id FROM ready) - RETURNING - id AS id, - queue AS queue, - headers AS headers, - payload AS payload, - state AS state, - attempts_count AS attempts_count, - deliveries_count AS deliveries_count, - created_at AS created_at, - first_attempt_at AS first_attempt_at, - next_attempt_at AS next_attempt_at, - last_attempt_at AS last_attempt_at, - acquired_at AS acquired_at - ``` - -### Archive Messages - -=== "PostgreSQL" - ```sql linenums="1" - BEGIN; - - INSERT INTO - message_archive ( - id, - queue, - headers, - payload, - state, - attempts_count, - deliveries_count, - created_at, - first_attempt_at, - last_attempt_at, - archived_at - ) - VALUES - ( - $1::BIGINT, - $2::VARCHAR, - $3::JSON, - $4::BYTEA, - $5::sqlbrokermessagestate, - $6::BIGINT, - $7::BIGINT, - $8::TIMESTAMP WITHOUT TIME ZONE, - $9::TIMESTAMP WITHOUT TIME ZONE, - $10::TIMESTAMP WITHOUT TIME ZONE, - $11::TIMESTAMP WITHOUT TIME ZONE - ); - - DELETE FROM message - WHERE - message.id IN ($1::BIGINT); - - COMMIT; - ``` - -=== "MySQL" - ```sql linenums="1" - BEGIN; - - INSERT INTO - message_archive ( - id, - queue, - headers, - payload, - state, - attempts_count, - deliveries_count, - created_at, - first_attempt_at, - last_attempt_at, - archived_at - ) - VALUES - ( - % s, - % s, - % s, - % s, - % s, - % s, - % s, - % s, - % s, - % s, - % s - ); - - DELETE FROM message - WHERE - message.id IN (% s); - ``` - -=== "SQLite" - ```sql linenums="1" - BEGIN; - - INSERT INTO message_archive ( - id, - queue, - headers, - payload, - state, - attempts_count, - deliveries_count, - created_at, - first_attempt_at, - last_attempt_at, - archived_at - ) - VALUES - (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?); - - DELETE FROM message - WHERE - message.id IN (?); - - COMMIT; - ``` - -### Requeue Messages - -=== "PostgreSQL" - ```sql linenums="1" - UPDATE message - SET - state = $1::sqlbrokermessagestate, - attempts_count = $2::BIGINT, - deliveries_count = $3::BIGINT, - first_attempt_at = $4::TIMESTAMP WITHOUT TIME ZONE, - next_attempt_at = $5::TIMESTAMP WITHOUT TIME ZONE, - last_attempt_at = $6::TIMESTAMP WITHOUT TIME ZONE, - acquired_at = $7::TIMESTAMP WITHOUT TIME ZONE - WHERE - message.id = $8::BIGINT - ``` - -=== "MySQL" - ```sql linenums="1" - UPDATE message - SET - state = % s, - attempts_count = % s, - deliveries_count = % s, - first_attempt_at = % s, - next_attempt_at = % s, - last_attempt_at = % s, - acquired_at = % s - WHERE - message.id = % s - ``` - -=== "SQLite" - ```sql linenums="1" - UPDATE message - SET - state = ?, - attempts_count = ?, - deliveries_count = ?, - first_attempt_at = ?, - next_attempt_at = ?, - last_attempt_at = ?, - acquired_at = ? - WHERE - message.id = ? - ``` - -Note that for bulk updates the arguments are sent in a batch in a single network call using `execute_many()`. - -### Requeue Stuck Messages - -=== "PostgreSQL" - ```sql linenums="1" - UPDATE message - SET - state = $1::sqlbrokermessagestate, - next_attempt_at = $2::TIMESTAMP WITHOUT TIME ZONE, - acquired_at = $3::TIMESTAMP WITHOUT TIME ZONE - WHERE - message.id IN ( - SELECT - message.id - FROM - message - WHERE - message.state = $4::sqlbrokermessagestate - AND message.acquired_at < $5::TIMESTAMP WITHOUT TIME ZONE - ) - ``` - -=== "MySQL" - ```sql linenums="1" - UPDATE message - SET - state = % s, - next_attempt_at = % s, - acquired_at = % s - WHERE - message.id IN ( - SELECT - anon_1.id - FROM - ( - SELECT - message.id AS id - FROM - message - WHERE - message.state = % s - AND message.acquired_at < % s - ) AS anon_1 - ) - ``` - -=== "SQLite" - ```sql linenums="1" - UPDATE message - SET - state = ?, - next_attempt_at = ?, - acquired_at = ? - WHERE - message.id IN ( - SELECT - message.id - FROM - message - WHERE - message.state = ? - AND message.acquired_at < ? - ) - ``` diff --git a/docs/docs/sqlbroker/design.md b/docs/docs/sqlbroker/design.md index f40d721..1d92786 100644 --- a/docs/docs/sqlbroker/design.md +++ b/docs/docs/sqlbroker/design.md @@ -15,7 +15,7 @@ search: ## Message Lifecycle -A published message starts out as `PENDING`. When it is acquired by a worker, it is marked as `PROCESSING`, which prevents it from being acquired by other worker processes. If the maximum allowed number of deliveries is configured and exceeded, the message is marked as `FAILED`. If not, the message is processed. If processing didn't raise an exception or if the message was manually Ack'ed in the handler, it is marked as `COMPLETED`. If the message was manually Nack'ed or Reject'ed or if processing raised an exception and [`AckPolicy`](../getting-started/acknowledgement.md){.internal-link} was set to `REJECT_ON_ERROR` or `NACK_ON_ERROR`, the message is Nack'ed or Reject'ed. Reject'ed messages are marked as `FAILED`. For Nack'ed messages, the retry policy determines if the message is allowed to be retried. If retry is allowed, the message is marked as `RETRYABLE`. If not, the message is marked as `FAILED`. +A published message starts out as `PENDING`. When it is acquired by a worker, it is marked as `PROCESSING`, which prevents it from being acquired by other worker processes. If the maximum allowed number of deliveries is configured and exceeded, the message is marked as `FAILED`. If not, the message is processed. If processing didn't raise an exception or if the message was manually [Acked](../sqlbroker/tutorial.md#ack){.internal-link} in the handler, it is marked as `COMPLETED`. If the message was manually [Nacked](../sqlbroker/tutorial.md#nack){.internal-link} or [Rejected](../sqlbroker/tutorial.md#reject){.internal-link} or if processing raised an exception and [`AckPolicy`](../getting-started/acknowledgement.md){.internal-link} was set to `REJECT_ON_ERROR` or `NACK_ON_ERROR`, the message is [Nacked](../sqlbroker/tutorial.md#nack){.internal-link} or [Rejected](../sqlbroker/tutorial.md#reject){.internal-link}. [Rejected](../sqlbroker/tutorial.md#reject){.internal-link} messages are marked as `FAILED`. For [Nacked](../sqlbroker/tutorial.md#nack){.internal-link} messages, the retry policy determines if the message is allowed to be retried. If retry is allowed, the message is marked as `RETRYABLE`. If not, the message is marked as `FAILED`. `PENDING`, `PROCESSING`, and `RETRYABLE` messages reside in the main table. On status change, `COMPLETED` and `FAILED` messages are removed from the main table and, depending on `retain_in_archive_on_ack` and `retain_in_archive_on_reject`, copied to the archive table. @@ -39,7 +39,7 @@ On start, the subscriber spawns four types of concurrent loops: **1. Fetch loop** — Periodically fetches batches of `PENDING` or `RETRYABLE` messages from the database, simultaneously updating them: marking as `PROCESSING`, setting `acquired_at` to now, and incrementing `deliveries_count`. Only messages with `next_attempt_at <= now` are fetched, ordered by `next_attempt_at`. The fetched messages are placed into an internal queue. The fetch limit is the minimum of `fetch_batch_size` and the free buffer capacity (`fetch_batch_size * overfetch_factor` minus currently queued messages). If the last fetch was "full" (returned as many messages as the limit), the next fetch happens after `min_fetch_interval`; otherwise after `max_fetch_interval`. -**2. Worker loops** (`max_workers` concurrent instances) — Each worker takes a message from the internal queue and first checks if `max_deliveries` has been exceeded; if so, the message is Rejected without processing. Otherwise, processing proceeds. Depending on the processing result, [`AckPolicy`](../getting-started/acknowledgement.md){.internal-link}, and manual Ack/Nack/Reject, the message is Ack'ed, Nack'ed, or Rejected. For Nack'ed messages, the `retry_strategy` is consulted to determine if and when the message might be retried. If allowed to be retried, the message is marked as `RETRYABLE`; otherwise as `FAILED`. Ack'ed messages are marked as `COMPLETED` and rejected messages are marked as `FAILED`. The message is then buffered for flushing. +**2. Worker loops** (`max_workers` concurrent instances) — Each worker takes a message from the internal queue and first checks if `max_deliveries` has been exceeded; if so, the message is [Rejected](../sqlbroker/tutorial.md#reject){.internal-link} without processing. Otherwise, processing proceeds. Depending on the processing result, [`AckPolicy`](../getting-started/acknowledgement.md){.internal-link}, and manual [Ack](../sqlbroker/tutorial.md#ack){.internal-link}/[Nack](../sqlbroker/tutorial.md#nack){.internal-link}/[Reject](../sqlbroker/tutorial.md#reject){.internal-link}, the message is [Acked](../sqlbroker/tutorial.md#ack){.internal-link}, [Nacked](../sqlbroker/tutorial.md#nack){.internal-link}, or [Rejected](../sqlbroker/tutorial.md#reject){.internal-link}. For [Nacked](../sqlbroker/tutorial.md#nack){.internal-link} messages, the `retry_strategy` is consulted to determine if and when the message might be retried. If allowed to be retried, the message is marked as `RETRYABLE`; otherwise as `FAILED`. [Acked](../sqlbroker/tutorial.md#ack){.internal-link} messages are marked as `COMPLETED` and rejected messages are marked as `FAILED`. The message is then buffered for flushing. **3. Flush loop** — Periodically flushes the buffered message state changes to the database. `COMPLETED` and `FAILED` messages are removed from the primary table and, depending on `retain_in_archive_on_ack` and `retain_in_archive_on_reject`, copied to the archive table. The state of `RETRYABLE` messages is updated in the primary table. @@ -71,7 +71,7 @@ Setting `max_deliveries` to a non-`None` value provides protection from the [poi ### Dead Letter Queue -The archive table doubles as a dead-letter queue: if the broker defines an archive table (`message_archive_table_name`) and `retain_in_archive_on_reject` is `True` (default), `FAILED` messages (Reject'ed, or Nack'ed with retries exhausted) are archived there. +The archive table doubles as a dead-letter queue: if the broker defines an archive table (`message_archive_table_name`) and `retain_in_archive_on_reject` is `True` (default), `FAILED` messages ([Rejected](../sqlbroker/tutorial.md#reject){.internal-link}, or [Nacked](../sqlbroker/tutorial.md#nack){.internal-link} with retries exhausted) are archived there. ### Ordered Processing diff --git a/docs/docs/sqlbroker/tutorial.md b/docs/docs/sqlbroker/tutorial.md index cbbf4a8..901682d 100644 --- a/docs/docs/sqlbroker/tutorial.md +++ b/docs/docs/sqlbroker/tutorial.md @@ -44,13 +44,13 @@ Currently the only supported DB schema variant is `WORK_QUEUE`, at version `1`, #### Broker parameters - **`engine`** — SQLAlchemy `AsyncEngine` to use for requests to the database. -- **`schema`** — `SqlBrokerSchemaConfig` describing the broker tables and schema selection. -- **`schema.message_table_name`** — Name of the table containing active messages. Defaults to `message`. -- **`schema.message_archive_table_name`** — Name of the table containing completed/failed messages. Defaults to `message_archive`. Set to `None` to run without an archive table, in which case subscribers must set both `retain_in_archive_on_ack` and `retain_in_archive_on_reject` to `False`. -- **`schema.variant`** — Schema variant to use. Defaults to `WORK_QUEUE`. -- **`schema.version`** — Variant-specific schema version enum. For `WORK_QUEUE`, use `SqlBrokerWorkQueueSchemaVersion`, defaulting to `V1`. -- **`validate_schema_on_start`** — If `True` (default), validates that the configured tables exist and conform to the expected schema. -- **`graceful_timeout`** — Seconds to wait for in-flight messages to finish processing during shutdown. +- **`schema`** (default: `SqlBrokerSchemaConfig()`) — `SqlBrokerSchemaConfig` describing the broker tables and schema selection. +- **`schema.message_table_name`** (default: `message`) — Name of the table containing active messages. +- **`schema.message_archive_table_name`** (default: `message_archive`) — Name of the table containing completed/failed messages. Set to `None` to run without an archive table, in which case subscribers must set both `retain_in_archive_on_ack` and `retain_in_archive_on_reject` to `False`. +- **`schema.variant`** (default: `WORK_QUEUE`) — Schema variant to use. +- **`schema.version`** (default: `V1`) — Variant-specific schema version enum. For `WORK_QUEUE`, use `SqlBrokerWorkQueueSchemaVersion`. +- **`validate_schema_on_start`** (default: `True`) — Validates that the configured tables exist and conform to the expected schema. +- **`graceful_timeout`** (default: `15.0`) — Seconds to wait for in-flight messages to finish processing during shutdown. ## Publishing @@ -61,10 +61,10 @@ Currently the only supported DB schema variant is `WORK_QUEUE`, at version `1`, The broker's and publisher's (see [publishing](../getting-started/publishing/index.md){.internal-link}) `.publish()` methods accept: - **`message`** — The message body. -- **`queue`** — The target queue name. -- **`headers`** — Optional `dict[str, str]` of message headers. -- **`next_attempt_at`** — Optional `datetime` (with timezone) for delayed delivery. -- **`connection`** — Optional SQLAlchemy `AsyncConnection` for transactional publishing. +- **`queue`** (default: `""`) — The target queue name. +- **`headers`** (default: `None`) — Optional `dict[str, str]` of message headers. +- **`next_attempt_at`** (default: `None`) — Optional `datetime` (with timezone) for delayed delivery. +- **`connection`** (default: `None`) — Optional SQLAlchemy `AsyncConnection` for transactional publishing. ### Delayed delivery @@ -90,7 +90,7 @@ When `connection` is provided, the message insert participates in the same datab - **`queues`** — List of queue names to consume from. - **`max_workers`** (default: `1`) — Number of concurrent handler coroutines. -- **`retry_strategy`** (default: `NoRetryStrategy()`) — Called to determine if and how soon a Nack'ed message is retried. +- **`retry_strategy`** (default: `NoRetryStrategy()`) — Called to determine if and how soon a [Nacked](#nack){.internal-link} message is retried. - **`fetch_batch_size`** — Maximum number of messages to fetch in a single batch. A fetch's actual limit might be lower if the free capacity of the acquired-but-not-yet-processed messages set is smaller. - **`overfetch_factor`** (default: `1.5`) — Multiplier for `fetch_batch_size` to size the maximum size of the set of acquired-but-not-yet-processed messages. - **`min_fetch_interval`** — Minimum interval between consecutive fetches. If the last fetch was full (returned as many messages as the fetch's limit), the next fetch happens after both (i) minimum fetch interval has passed, and (ii) capacity equal to the fetch batch size has freed up in the set of acquired-but-not-yet-processed messages. @@ -98,40 +98,58 @@ When `connection` is provided, the message insert participates in the same datab - **`flush_interval`** — Interval between flushes of processed message state to the database. - **`release_stuck_interval`** (default: `60`) — Interval between checks for stuck [`PROCESSING`](../sqlbroker/design.md#message-lifecycle){.internal-link} messages. - **`release_stuck_timeout`** (default: `60 * 10`) — Interval since `acquired_at` after which a [`PROCESSING`](../sqlbroker/design.md#message-lifecycle){.internal-link} message is considered stuck and is released back to [`PENDING`](../sqlbroker/design.md#message-lifecycle){.internal-link}. -- **`max_deliveries`** (default: `None`) — Maximum number of deliveries allowed for a message for [poison message protection](../sqlbroker/design.md#poison-message-protection){.internal-link}. If set, messages that have reached this limit are Reject'ed to [`FAILED`](../sqlbroker/design.md#message-lifecycle){.internal-link} without processing. Note that this might violate the at-least-once processing semantics. +- **`max_deliveries`** (default: `None`) — Maximum number of deliveries allowed for a message for [poison message protection](../sqlbroker/design.md#poison-message-protection){.internal-link}. If set, messages that have reached this limit are [Rejected](#reject){.internal-link} to [`FAILED`](../sqlbroker/design.md#message-lifecycle){.internal-link} without processing. Note that this might violate the at-least-once processing semantics. - **`ack_policy`** (default: `REJECT_ON_ERROR`) — [`AckPolicy`](../getting-started/acknowledgement.md){.internal-link} that controls acknowledgement behavior. -- **`retain_in_archive_on_ack`** — If `True` (default), [`COMPLETED`](../sqlbroker/design.md#message-lifecycle){.internal-link} (Ack'ed) messages, in addition to being removed from the primary table, are also persisted in the archive table. Requires the broker to define an archive table (`message_archive_table_name`). -- **`retain_in_archive_on_reject`** — If `True` (default), [`FAILED`](../sqlbroker/design.md#message-lifecycle){.internal-link} (Reject'ed) messages, in addition to being removed from the primary table, are also persisted in the archive table, where they serve as a [dead-letter queue](../sqlbroker/design.md#dead-letter-queue){.internal-link}. Requires the broker to define an archive table (`message_archive_table_name`). +- **`retain_in_archive_on_ack`** (default: `True`) — [`COMPLETED`](../sqlbroker/design.md#message-lifecycle){.internal-link} ([Acked](#ack){.internal-link}) messages, in addition to being removed from the primary table, are also persisted in the archive table. Requires the broker to define an archive table (`message_archive_table_name`). +- **`retain_in_archive_on_reject`** (default: `True`) — [`FAILED`](../sqlbroker/design.md#message-lifecycle){.internal-link} ([Rejected](#reject){.internal-link}) messages, in addition to being removed from the primary table, are also persisted in the archive table, where they serve as a [dead-letter queue](../sqlbroker/design.md#dead-letter-queue){.internal-link}. Requires the broker to define an archive table (`message_archive_table_name`). + + +### Message Lifecycle + +A published message starts out in the `message` table with a [`PENDING`](../sqlbroker/design.md#message-lifecycle){.internal-link} status. Once a subscriber acquires it, the row is marked as [`PROCESSING`](../sqlbroker/design.md#message-lifecycle){.internal-link} and the message is processed. + +From there, the following acknowledgment actions are possible: + +#### **Ack** +The message is marked as [`COMPLETED`](../sqlbroker/design.md#message-lifecycle){.internal-link} and is moved from the `message` table to the `message_archive` table. +#### **Nack** +The `retry_policy` is called to determine if the message is allowed to be retried and when it will be retried. If allowed to be retried, the message is marked as [`RETRYABLE`](../sqlbroker/design.md#message-lifecycle){.internal-link} in the `message` table. If not, the message is [Rejected](#reject){.internal-link}. +#### **Reject** +The message is marked as [`FAILED`](../sqlbroker/design.md#message-lifecycle){.internal-link} and is moved from the `message` table to the `message_archive` table. ### Acknowledgements +One of the [acknowledgement actions](#message-lifecycle){.internal-link} can be performed automatically according to the `ack_policy` or manually in the handler. + #### Automatic via AckPolicy -Set `ack_policy` on the subscriber to control what happens after handler execution. +Set `ack_policy` on the subscriber to control what happens after handler execution depending on whether the handler raised an exception or returned. ```python linenums="1" {!> docs_src/sqlbroker/acknowledgements.py [ln:1-26]!} ``` -- `AckPolicy.ACK` — Marks the message as [`COMPLETED`](../sqlbroker/design.md#message-lifecycle){.internal-link} after the handler attempt, even if the handler raises an exception. -- `AckPolicy.ACK_FIRST` — Has the same effect as `AckPolicy.ACK` for this broker. -- `AckPolicy.REJECT_ON_ERROR` — On success, acknowledges the message. If the handler raises, the message is Reject'ed and marked as [`FAILED`](../sqlbroker/design.md#message-lifecycle){.internal-link}. `retry_strategy` is ignored. -- `AckPolicy.NACK_ON_ERROR` — On success, acknowledges the message. If the handler raises, the message is Nack'ed. The `retry_strategy` is then called to determine whether the message should be scheduled for retry as [`RETRYABLE`](../sqlbroker/design.md#message-lifecycle){.internal-link} or Reject'ed and marked as [`FAILED`](../sqlbroker/design.md#message-lifecycle){.internal-link}. With `NoRetryStrategy()` or `None` in `retry_strategy`, this has the same effect as `REJECT_ON_ERROR`. -- `AckPolicy.MANUAL` — Requires explicit `msg.ack()`, `msg.nack()`, or `msg.reject()` in the handler. In the absence of explicit action, the message is Reject'ed as a safety precaution. +- `AckPolicy.ACK` — [Acks](#ack){.internal-link} the message after the handler attempt, even if the handler raised an exception. +- `AckPolicy.ACK_FIRST` — Same as `AckPolicy.ACK` for this broker. +- `AckPolicy.REJECT_ON_ERROR` — On success, [Acks](#ack){.internal-link} the message. On exception, [Rejects](#reject){.internal-link} the message. `retry_strategy` is ignored. +- `AckPolicy.NACK_ON_ERROR` — On success, [Acks](#ack){.internal-link} the message. On exception, [Nacks](#nack){.internal-link} the message. With `NoRetryStrategy()` or `None` in `retry_strategy`, this has the same effect as `REJECT_ON_ERROR`. +- `AckPolicy.MANUAL` — Requires explicit `msg.ack()`, `msg.nack()`, or `msg.reject()` in the handler. In the absence of explicit action, the message is [Rejected](#reject){.internal-link} as a safety precaution. + +Automatic acknowledgement applies only if the handler did not already call one of the manual acknowledgement methods. #### Manual -Use `AckPolicy.MANUAL` when the handler should decide the outcome explicitly. +Use `AckPolicy.MANUAL` when the handler should decide the outcome explicitly with `msg.ack()`, `msg.nack()`, or `msg.reject()` ```python linenums="1" {!> docs_src/sqlbroker/acknowledgements.py [ln:29-39]!} ``` -For any `ack_policy` manual acknowledgement overrides the automatic policy decision. If the handler calls `msg.ack()`, `msg.nack()`, or `msg.reject()`, that explicit action is used`. +Manual acknowledgement can also be used with any other `ack_policy`, not just `AckPolicy.MANUAL`. The override the `ack_policy` driven action. ### Retry strategies -When a message is Nack'ed (either manually or by `AckPolicy.NACK_ON_ERROR`), the `retry_strategy` determines if and when the message should be retried. By default, `NoRetryStrategy()` disables retries. All strategies accept `max_attempts` and `max_total_delay_seconds` as common parameters — if either limit is reached, the message is marked as [`FAILED`](../sqlbroker/design.md#message-lifecycle){.internal-link} instead of [`RETRYABLE`](../sqlbroker/design.md#message-lifecycle){.internal-link}. Otherwise, the strategy schedules the message for a retry determined by the returned `next_attempt_at`. +When a message is [Nacked](#nack){.internal-link} (either manually or by `AckPolicy.NACK_ON_ERROR`), the `retry_strategy` determines if and when the message should be retried. By default, `NoRetryStrategy()` disables retries. All strategies accept `max_attempts` and `max_total_delay_seconds` as common parameters — if either limit is reached, the message is marked as [`FAILED`](../sqlbroker/design.md#message-lifecycle){.internal-link} instead of [`RETRYABLE`](../sqlbroker/design.md#message-lifecycle){.internal-link}. Otherwise, the strategy schedules the message for a retry determined by the returned `next_attempt_at`. #### `ConstantRetryStrategy` @@ -175,7 +193,7 @@ Retries after `base_delay_seconds` plus random jitter in the range `[-jitter_sec #### `NoRetryStrategy` -No retries — the message is marked as [`FAILED`](../sqlbroker/design.md#message-lifecycle){.internal-link} on the first Nack. +No retries — the message is marked as [`FAILED`](../sqlbroker/design.md#message-lifecycle){.internal-link} on the first [Nack](#nack){.internal-link}. ```python linenums="1" {!> docs_src/sqlbroker/retry.py [ln:45]!} diff --git a/faststream_sqlbroker/sqlbroker/broker/registrator.py b/faststream_sqlbroker/sqlbroker/broker/registrator.py index 76d977d..28155c1 100644 --- a/faststream_sqlbroker/sqlbroker/broker/registrator.py +++ b/faststream_sqlbroker/sqlbroker/broker/registrator.py @@ -56,7 +56,7 @@ def subscriber( # type: ignore[override] max_workers: Number of concurrent handler coroutines. retry_strategy: - Called to determine if and how soon a Nack'ed message is retried. + Called to determine if and how soon a Nacked message is retried. Defaults to `NoRetryStrategy()`. min_fetch_interval: Minimum interval between consecutive fetches. If the last fetch was @@ -84,7 +84,7 @@ def subscriber( # type: ignore[override] `600`. max_deliveries: Maximum number of deliveries allowed for a message. If set, messages - that have reached this limit are Reject'ed to `FAILED` without + that have reached this limit are Rejected to `FAILED` without processing. Useful for poison message protection. Note that this might violate the at-least-once processing semantics. ack_policy: diff --git a/faststream_sqlbroker/sqlbroker/message.py b/faststream_sqlbroker/sqlbroker/message.py index bafe747..3a7e6a6 100644 --- a/faststream_sqlbroker/sqlbroker/message.py +++ b/faststream_sqlbroker/sqlbroker/message.py @@ -146,7 +146,7 @@ async def _assert_state_updated(self, logger: "LoggerProto | None") -> None: f"State of message {self} was not updated after processing, " f"perhaps due to the AckPolicy.MANUAL policy and lack of manual " f"acknowledgement in the handler. As a precaution, the message " - f"was Reject'ed.", + f"was Rejected.", ) await self.reject() From 412291fae7a5209e62e29853b7bcfce122831ccf Mon Sep 17 00:00:00 2001 From: Arseniy Popov Date: Wed, 17 Jun 2026 17:37:01 +0300 Subject: [PATCH 2/2] docs: expand docs on acks --- docs/docs/sqlbroker/tutorial.md | 43 ++++++++++++++++++------------- docs/docs_src/sqlbroker/broker.py | 4 --- docs/docs_src/sqlbroker/retry.py | 14 ++++++---- 3 files changed, 34 insertions(+), 27 deletions(-) diff --git a/docs/docs/sqlbroker/tutorial.md b/docs/docs/sqlbroker/tutorial.md index 901682d..5713969 100644 --- a/docs/docs/sqlbroker/tutorial.md +++ b/docs/docs/sqlbroker/tutorial.md @@ -29,7 +29,7 @@ pip install "faststream-sqlbroker" ## Database Tables -Currently the only supported DB schema variant is `WORK_QUEUE`, at version `1`, which uses up to two tables — `message` (active messages) and `message_archive` (completed/failed messages). These settings are grouped under the broker's `schema` parameter via `SqlBrokerSchemaConfig`. Set `message_archive_table_name=None` there to omit using archiving on success and [DLQ](../sqlbroker/design.md#dead-letter-queue){.internal-link}. You can customize the tables to your liking (partition them, add indices, specify more specific data types like `JSONB`, etc.) as long as they generally conform to the selected schema definition. Schema check is done on startup if the brokers's `validate_schema_on_start` is `True`. +Currently the only supported DB schema variant is `WORK_QUEUE`, at version `1`, which uses up to two tables — `message` (active messages) and `message_archive` (completed/failed messages). These settings are grouped under the broker's `schema` parameter via `SqlBrokerSchemaConfig`. Set `message_archive_table_name=None` there to omit using archiving on success and [DLQ](../sqlbroker/design.md#dead-letter-queue){.internal-link}. You can customize the tables to your liking (partition them, add indices, specify more specific data types like `JSONB`, etc.) as long as they generally conform to the selected schema definition. Schema check is done on startup if the broker's `validate_schema_on_start` is `True`. ```python linenums="1" {!> docs_src/sqlbroker/tables.py !} @@ -96,26 +96,28 @@ When `connection` is provided, the message insert participates in the same datab - **`min_fetch_interval`** — Minimum interval between consecutive fetches. If the last fetch was full (returned as many messages as the fetch's limit), the next fetch happens after both (i) minimum fetch interval has passed, and (ii) capacity equal to the fetch batch size has freed up in the set of acquired-but-not-yet-processed messages. - **`max_fetch_interval`** — Maximum interval between consecutive fetches. - **`flush_interval`** — Interval between flushes of processed message state to the database. -- **`release_stuck_interval`** (default: `60`) — Interval between checks for stuck [`PROCESSING`](../sqlbroker/design.md#message-lifecycle){.internal-link} messages. -- **`release_stuck_timeout`** (default: `60 * 10`) — Interval since `acquired_at` after which a [`PROCESSING`](../sqlbroker/design.md#message-lifecycle){.internal-link} message is considered stuck and is released back to [`PENDING`](../sqlbroker/design.md#message-lifecycle){.internal-link}. -- **`max_deliveries`** (default: `None`) — Maximum number of deliveries allowed for a message for [poison message protection](../sqlbroker/design.md#poison-message-protection){.internal-link}. If set, messages that have reached this limit are [Rejected](#reject){.internal-link} to [`FAILED`](../sqlbroker/design.md#message-lifecycle){.internal-link} without processing. Note that this might violate the at-least-once processing semantics. -- **`ack_policy`** (default: `REJECT_ON_ERROR`) — [`AckPolicy`](../getting-started/acknowledgement.md){.internal-link} that controls acknowledgement behavior. -- **`retain_in_archive_on_ack`** (default: `True`) — [`COMPLETED`](../sqlbroker/design.md#message-lifecycle){.internal-link} ([Acked](#ack){.internal-link}) messages, in addition to being removed from the primary table, are also persisted in the archive table. Requires the broker to define an archive table (`message_archive_table_name`). -- **`retain_in_archive_on_reject`** (default: `True`) — [`FAILED`](../sqlbroker/design.md#message-lifecycle){.internal-link} ([Rejected](#reject){.internal-link}) messages, in addition to being removed from the primary table, are also persisted in the archive table, where they serve as a [dead-letter queue](../sqlbroker/design.md#dead-letter-queue){.internal-link}. Requires the broker to define an archive table (`message_archive_table_name`). +- **`release_stuck_interval`** (default: `60`) — Interval between checks for stuck [`PROCESSING`](#message-lifecycle){.internal-link} messages. +- **`release_stuck_timeout`** (default: `60 * 10`) — Interval since `acquired_at` after which a [`PROCESSING`](#message-lifecycle){.internal-link} message is considered stuck and is released back to [`PENDING`](#message-lifecycle){.internal-link}. +- **`max_deliveries`** (default: `None`) — Maximum number of deliveries allowed for a message for [poison message protection](../sqlbroker/design.md#poison-message-protection){.internal-link}. If set, messages that have reached this limit are [Rejected](#reject){.internal-link} without processing. Note that this might violate the at-least-once processing semantics. +- **`ack_policy`** (default: `REJECT_ON_ERROR`) — [`AckPolicy`](#automatic-via-ackpolicy){.internal-link} that controls acknowledgement behavior. +- **`retain_in_archive_on_ack`** (default: `True`) — [Acked](#ack){.internal-link} messages, in addition to being removed from the primary table, are also persisted in the archive table. Requires the broker to define an archive table (`message_archive_table_name`). +- **`retain_in_archive_on_reject`** (default: `True`) — [Rejected](#reject){.internal-link} messages, in addition to being removed from the primary table, are also persisted in the archive table, where they serve as a [dead-letter queue](../sqlbroker/design.md#dead-letter-queue){.internal-link}. Requires the broker to define an archive table (`message_archive_table_name`). ### Message Lifecycle -A published message starts out in the `message` table with a [`PENDING`](../sqlbroker/design.md#message-lifecycle){.internal-link} status. Once a subscriber acquires it, the row is marked as [`PROCESSING`](../sqlbroker/design.md#message-lifecycle){.internal-link} and the message is processed. +A published message starts out in the `message` table with a `PENDING` status. Once a subscriber acquires it, the row is marked as `PROCESSING` and the message is processed. -From there, the following acknowledgment actions are possible: +From there, the following acknowledgement actions are possible: #### **Ack** -The message is marked as [`COMPLETED`](../sqlbroker/design.md#message-lifecycle){.internal-link} and is moved from the `message` table to the `message_archive` table. +The message is marked as `COMPLETED` and is moved from the `message` table to the `message_archive` table. #### **Nack** -The `retry_policy` is called to determine if the message is allowed to be retried and when it will be retried. If allowed to be retried, the message is marked as [`RETRYABLE`](../sqlbroker/design.md#message-lifecycle){.internal-link} in the `message` table. If not, the message is [Rejected](#reject){.internal-link}. +The `retry_strategy` is called to determine if the message is allowed to be retried and when it will be retried. If allowed to be retried, the message is marked as `RETRYABLE` in the `message` table. If not, the message is [Rejected](#reject){.internal-link}. #### **Reject** -The message is marked as [`FAILED`](../sqlbroker/design.md#message-lifecycle){.internal-link} and is moved from the `message` table to the `message_archive` table. +The message is marked as `FAILED` and is moved from the `message` table to the `message_archive` table. + +These actions are performed by [Acknowledgements](#acknowledgements){.internal-link} after a processing attempt. Also, if `max_deliveries` was set on a subscriber and exceeded, the message is [Rejected](#reject){.internal-link} prior to a processing attempt. ### Acknowledgements @@ -145,11 +147,16 @@ Use `AckPolicy.MANUAL` when the handler should decide the outcome explicitly wit {!> docs_src/sqlbroker/acknowledgements.py [ln:29-39]!} ``` -Manual acknowledgement can also be used with any other `ack_policy`, not just `AckPolicy.MANUAL`. The override the `ack_policy` driven action. +Manual acknowledgements can also be used with any other `ack_policy`, not just `AckPolicy.MANUAL`. They override the `ack_policy` driven action. ### Retry strategies -When a message is [Nacked](#nack){.internal-link} (either manually or by `AckPolicy.NACK_ON_ERROR`), the `retry_strategy` determines if and when the message should be retried. By default, `NoRetryStrategy()` disables retries. All strategies accept `max_attempts` and `max_total_delay_seconds` as common parameters — if either limit is reached, the message is marked as [`FAILED`](../sqlbroker/design.md#message-lifecycle){.internal-link} instead of [`RETRYABLE`](../sqlbroker/design.md#message-lifecycle){.internal-link}. Otherwise, the strategy schedules the message for a retry determined by the returned `next_attempt_at`. +When a message is [Nacked](#nack){.internal-link} (either manually with `msg.nack()` or by `AckPolicy.NACK_ON_ERROR`), the `retry_strategy` determines if and when the message should be retried. By default, `NoRetryStrategy()` disables retries. All strategies accept common parameters: + +- `max_attempts` - Maximum number of processing attempts. +- `max_total_delay_seconds` - Maximum delay between the first and last attempt. + +If either limit is reached, the message is marked as [Rejected](#reject){.internal-link}. Otherwise, `next_attempt_at` is set on the message to signify a scheduled retry. #### `ConstantRetryStrategy` @@ -180,7 +187,7 @@ Delay starts at `initial_delay_seconds` and is multiplied by `multiplier` on eac Same as exponential backoff, but adds random jitter (up to `delay * jitter_factor`) to spread out retries and avoid thundering herds. ```python linenums="1" -{!> docs_src/sqlbroker/retry.py [ln:30-36]!} +{!> docs_src/sqlbroker/retry.py [ln:31-38]!} ``` #### `ConstantWithJitterRetryStrategy` @@ -188,15 +195,15 @@ Same as exponential backoff, but adds random jitter (up to `delay * jitter_facto Retries after `base_delay_seconds` plus random jitter in the range `[-jitter_seconds, +jitter_seconds]`. ```python linenums="1" -{!> docs_src/sqlbroker/retry.py [ln:38-43]!} +{!> docs_src/sqlbroker/retry.py [ln:40-45]!} ``` #### `NoRetryStrategy` -No retries — the message is marked as [`FAILED`](../sqlbroker/design.md#message-lifecycle){.internal-link} on the first [Nack](#nack){.internal-link}. +No retries — the message is marked as [Rejected](#reject){.internal-link} on the first [Nack](#nack){.internal-link}. ```python linenums="1" -{!> docs_src/sqlbroker/retry.py [ln:45]!} +{!> docs_src/sqlbroker/retry.py [ln:47-49]!} ``` ## Transactional outbox diff --git a/docs/docs_src/sqlbroker/broker.py b/docs/docs_src/sqlbroker/broker.py index 0ef9f67..464a7e2 100644 --- a/docs/docs_src/sqlbroker/broker.py +++ b/docs/docs_src/sqlbroker/broker.py @@ -3,8 +3,6 @@ from faststream_sqlbroker import ( SqlBroker, SqlBrokerSchemaConfig, - SqlBrokerSchemaVariant, - SqlBrokerWorkQueueSchemaVersion, ) engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/mydb") @@ -13,7 +11,5 @@ schema=SqlBrokerSchemaConfig( message_table_name="message", message_archive_table_name="message_archive", - variant=SqlBrokerSchemaVariant.WORK_QUEUE, - version=SqlBrokerWorkQueueSchemaVersion.V1, ), ) diff --git a/docs/docs_src/sqlbroker/retry.py b/docs/docs_src/sqlbroker/retry.py index bd72c58..8451884 100644 --- a/docs/docs_src/sqlbroker/retry.py +++ b/docs/docs_src/sqlbroker/retry.py @@ -16,22 +16,24 @@ linear = LinearRetryStrategy( initial_delay_seconds=1, step_seconds=2, - max_attempts=5, + max_attempts=3, max_total_delay_seconds=60, ) exponential = ExponentialBackoffRetryStrategy( initial_delay_seconds=1, + multiplier=2.0, # default max_delay_seconds=60, - max_attempts=8, + max_attempts=3, max_total_delay_seconds=300, ) exponential_jitter = ExponentialBackoffWithJitterRetryStrategy( initial_delay_seconds=1, + multiplier=2.0, # default max_delay_seconds=60, - jitter_factor=0.5, - max_attempts=8, + jitter_factor=0.5, # default + max_attempts=3, max_total_delay_seconds=300, ) @@ -42,4 +44,6 @@ max_total_delay_seconds=None, ) -no_retry = NoRetryStrategy() +no_retry = NoRetryStrategy( + max_attempts=1, # default +)