Сообщение снова падает: проверяем dead-letter queue и безопасный возврат
На изолированной SQLite-модели проверяем лимит повторов, перенос постоянной ошибки в dead-letter queue, откат при сбое между вставкой и удалением и однократный возврат исправленного сообщения.
В этой статье
Если одно сообщение снова и снова ломает обработчик, не оставляйте его в бесконечном цикле повторов. После постоянной ошибки или исчерпания лимита попыток перенесите сообщение в отдельную очередь ошибок — dead-letter queue, или DLQ. Возвращать его в рабочую очередь стоит только после исправления причины и с контролем повторной доставки.
Для владельца магазина последствие вполне земное: один неверный SKU или неожиданное поле может занять обработчики, задержать корректные заказы и превратить журнал ошибок в шум. DLQ отделяет неисправимое сейчас сообщение от основного потока, но сама по себе не исправляет данные и не гарантирует однократную обработку.
Где заканчивается повтор и начинается очередь ошибок
Состояние | Действие | Критерий результата |
|---|---|---|
Временная ошибка | Повторить с паузой и лимитом | Сообщение обработано либо достигло лимита |
Постоянная ошибка | Сразу изолировать | Основная очередь продолжает работу |
Лимит попыток исчерпан | Перенести в DLQ | Сообщение больше не крутится в рабочем потоке |
Причина исправлена | Контролируемо вернуть | В DLQ записи нет, в рабочей очереди одна запись |
Повторный возврат | Не создавать копию | Операция сообщает, что возвращать уже нечего |
Временной считаем ошибку, которая может исчезнуть без изменения сообщения: например, HTTP 503 или короткий сетевой тайм-аут. Постоянной — ошибку данных или бизнес-правила, которую тот же запрос повторит с тем же результатом: неизвестный SKU, недопустимое количество, отсутствующий обязательный идентификатор. Если классификация сомнительна, нужен ограниченный повтор, а не вечность.
Что проверяем на маленьком примере
Учебная модель использует только стандартную библиотеку Python и временный файл SQLite. Она не подключается к сети и не работает с реальным брокером. После завершения удаляется временная директория файловой системы вместе с тестовой базой. Проверка выполнена 28 октября 2026 года на CPython 3.12.14 и SQLite 3.53.1.
временная ошибка дважды возвращает сообщение на повтор, после успеха оно исчезает из очереди и не попадает в DLQ;
постоянная ошибка переносит сообщение в DLQ после первой неудачи;
три временных сбоя исчерпывают лимит и изолируют сообщение;
искусственный сбой между записью в DLQ и удалением из рабочей очереди откатывает всю транзакцию;
исправленное сообщение возвращается один раз, а повторный redrive — контролируемый возврат — не создаёт копию.
Сценарий запускается обычным интерпретатором Python 3.12. Все идентификаторы, пути и данные заданы внутри примера; база создаётся в изолированном временном каталоге. Операций с рабочими файлами и внешними очередями нет.
#!/usr/bin/env python3
import sqlite3
import tempfile
from pathlib import Path
MAX_ATTEMPTS = 3
def create_schema(conn):
conn.executescript("""
CREATE TABLE queue (
message_id TEXT PRIMARY KEY,
payload TEXT NOT NULL,
attempts INTEGER NOT NULL DEFAULT 0,
last_error TEXT
);
CREATE TABLE dead_letter (
message_id TEXT PRIMARY KEY,
payload TEXT NOT NULL,
attempts INTEGER NOT NULL,
reason TEXT NOT NULL
);
""")
def enqueue(conn, message_id, payload):
with conn:
conn.execute(
"INSERT INTO queue(message_id, payload) VALUES (?, ?)",
(message_id, payload),
)
def record_failure(conn, message_id, error, *, permanent, inject_crash=False):
with conn:
row = conn.execute(
"SELECT payload, attempts FROM queue WHERE message_id = ?",
(message_id,),
).fetchone()
if row is None:
raise LookupError(message_id)
payload, attempts = row
next_attempt = attempts + 1
if permanent or next_attempt >= MAX_ATTEMPTS:
reason = "permanent" if permanent else "attempt_limit"
conn.execute(
"""INSERT INTO dead_letter(message_id, payload, attempts, reason)
VALUES (?, ?, ?, ?)""",
(message_id, payload, next_attempt, reason),
)
if inject_crash:
raise RuntimeError("simulated crash before source delete")
conn.execute("DELETE FROM queue WHERE message_id = ?", (message_id,))
return "dead_letter"
conn.execute(
"""UPDATE queue
SET attempts = ?, last_error = ?
WHERE message_id = ?""",
(next_attempt, error, message_id),
)
return "retry"
def acknowledge(conn, message_id):
with conn:
conn.execute("DELETE FROM queue WHERE message_id = ?", (message_id,))
def redrive(conn, message_id, corrected_payload):
with conn:
row = conn.execute(
"SELECT 1 FROM dead_letter WHERE message_id = ?", (message_id,)
).fetchone()
if row is None:
return False
conn.execute(
"""INSERT INTO queue(message_id, payload, attempts, last_error)
VALUES (?, ?, 0, NULL)""",
(message_id, corrected_payload),
)
conn.execute(
"DELETE FROM dead_letter WHERE message_id = ?", (message_id,)
)
return True
def count(conn, table, message_id):
if table not in {"queue", "dead_letter"}:
raise ValueError(table)
return conn.execute(
f"SELECT COUNT(*) FROM {table} WHERE message_id = ?", (message_id,)
).fetchone()[0]
with tempfile.TemporaryDirectory(prefix="dlq-lab-") as tmpdir:
db_path = Path(tmpdir) / "queue.sqlite3"
conn = sqlite3.connect(db_path)
create_schema(conn)
enqueue(conn, "transient-ok", "sku=GOOD&qty=1")
assert record_failure(conn, "transient-ok", "HTTP 503", permanent=False) == "retry"
assert record_failure(conn, "transient-ok", "HTTP 503", permanent=False) == "retry"
acknowledge(conn, "transient-ok")
assert count(conn, "queue", "transient-ok") == 0
assert count(conn, "dead_letter", "transient-ok") == 0
print("transient_then_success: queue=0 dead_letter=0")
enqueue(conn, "bad-sku", "sku=UNKNOWN&qty=1")
assert record_failure(conn, "bad-sku", "unknown SKU", permanent=True) == "dead_letter"
assert conn.execute(
"SELECT attempts, reason FROM dead_letter WHERE message_id = ?",
("bad-sku",),
).fetchone() == (1, "permanent")
print("permanent_error: queue=0 dead_letter=1 attempts=1 reason=permanent")
enqueue(conn, "retry-limit", "sku=GOOD&qty=2")
assert record_failure(conn, "retry-limit", "timeout", permanent=False) == "retry"
assert record_failure(conn, "retry-limit", "timeout", permanent=False) == "retry"
assert record_failure(conn, "retry-limit", "timeout", permanent=False) == "dead_letter"
assert conn.execute(
"SELECT attempts, reason FROM dead_letter WHERE message_id = ?",
("retry-limit",),
).fetchone() == (3, "attempt_limit")
print("attempt_limit: queue=0 dead_letter=1 attempts=3 reason=attempt_limit")
enqueue(conn, "rollback", "sku=BROKEN&qty=1")
try:
record_failure(
conn, "rollback", "invalid payload",
permanent=True, inject_crash=True,
)
except RuntimeError as exc:
assert str(exc) == "simulated crash before source delete"
else:
raise AssertionError("injected failure did not happen")
assert count(conn, "queue", "rollback") == 1
assert count(conn, "dead_letter", "rollback") == 0
print("transaction_rollback: queue=1 dead_letter=0")
assert redrive(conn, "bad-sku", "sku=FIXED&qty=1") is True
assert redrive(conn, "bad-sku", "sku=FIXED&qty=1") is False
assert conn.execute(
"SELECT payload, attempts FROM queue WHERE message_id = ?",
("bad-sku",),
).fetchone() == ("sku=FIXED&qty=1", 0)
acknowledge(conn, "bad-sku")
assert count(conn, "queue", "bad-sku") == 0
print("corrected_redrive: first=true second=false final_queue=0 dead_letter=0")
conn.close()
assert not db_path.exists()
Наблюдаемый результат
transient_then_success: queue=0 dead_letter=0
permanent_error: queue=0 dead_letter=1 attempts=1 reason=permanent
attempt_limit: queue=0 dead_letter=1 attempts=3 reason=attempt_limit
transaction_rollback: queue=1 dead_letter=0
corrected_redrive: first=true second=false final_queue=0 dead_letter=0
Все проверки assert завершились успешно с кодом 0. После выхода из TemporaryDirectory файл базы отсутствовал. Это подтверждает поведение именно учебной модели; на RabbitMQ, Amazon SQS или собственном брокере нужно отдельно проверить их гарантии доставки, конкуренцию обработчиков, удержание сообщений и права доступа.
Почему перенос должен быть атомарным
Операция «записать в DLQ, затем удалить из основной очереди» состоит из двух изменений. Если между ними завершится процесс, без транзакции можно получить копию в обеих очередях или потерять сообщение. В модели оба изменения выполняются внутри одного контекста транзакции SQLite: исключение до удаления откатывает вставку, поэтому исходное сообщение остаётся на месте.
Такое поведение контекстного менеджера соединения — фиксация при успехе и откат при исключении — описано в документации sqlite3 для Python 3.12. Соединение он автоматически не закрывает, поэтому в примере вызов close указан явно.
Возврат после исправления: тут есть подвох
Redrive не должен означать «вернуть всё и посмотреть». Сначала исправляют причину: справочник SKU, схему сообщения, права или код обработчика. Затем выбирают ограниченную партию, сбрасывают счётчик попыток по явному правилу и наблюдают за скоростью ошибок, длиной очереди и временем обработки.
В модели метод redrive в одной транзакции вставляет исправленное сообщение в рабочую очередь и удаляет запись из DLQ. Первое выполнение возвращает true, второе — false. Это защищает только от повторного запуска именно этой функции в однопроцессной модели; распределённая система всё равно должна иметь идемпотентный обработчик.
В Amazon SQS очередь ошибок связывается с исходной очередью политикой redrive: после заданного числа получений сообщение попадает в DLQ. Руководство SQS по DLQ отдельно предупреждает о влиянии на порядок FIFO-очередей. При возврате AWS рекомендует начинать с небольшой скорости и наблюдать за системой; штатный redrive не изменяет и не фильтрует сообщения. Это описано в руководстве по redrive.
В RabbitMQ сообщения могут перейти в dead-letter exchange после reject или nack без requeue, истечения TTL, превышения длины очереди либо delivery limit для quorum queue. Условия и ограничения перечислены в официальной документации RabbitMQ по dead lettering. Механизм отличается от SQS, поэтому переносить значения лимитов и гарантии между продуктами нельзя.
Что заложить в промышленную реализацию
стабильный message_id и идемпотентность потребителя: повторная доставка не должна создавать второй заказ или платёж;
разделение временных и постоянных ошибок с явным кодом причины, а не только текстом исключения;
лимит попыток, пауза с разбросом и общий бюджет повторов, чтобы несколько уровней не умножали нагрузку;
атомарный перенос или документированная гарантия брокера; при невозможности — сверка по message_id и процедура устранения дублей;
метрики длины рабочей очереди и DLQ, возраста старейшего сообщения, доли ошибок и числа возвратов;
ограниченный redrive после исправления: небольшая партия, критерий остановки, журнал оператора и возможность вернуть сообщение в карантин;
срок хранения, шифрование, доступ к полезной нагрузке и маскирование персональных данных;
отдельная проверка последовательности обработки событий: DLQ и повторный ввод могут изменить хронологический порядок связанных сообщений.
Как DLQ сочетается с другими защитами
Transactional outbox отвечает за другое место отказа: событие должно появиться после сохранения заказа, даже если отправитель временно недоступен. Практический пример есть в разборе transactional outbox для webhook.
Лимит повторов не заменяет идемпотентность: одно сообщение может прийти повторно из-за тайм-аута подтверждения, перезапуска потребителя или redrive. Отдельно это проверено в материале о повторном webhook и идемпотентности.
Критерий приёмки простой: ядовитое сообщение перестаёт блокировать основной поток, его причина и число попыток видны, сбой во время переноса не теряет исходные данные. Возврат исправленных сообщений выполняется небольшими партиями, с заданной скоростью и критерием остановки; повторная доставка не создаёт второй бизнес-результат. Если хотя бы один пункт нельзя проверить, DLQ пока остаётся складом проблем, а не механизмом восстановления.
If a single message repeatedly breaks the handler, do not leave it in an infinite retry loop. After a persistent error or exhaustion of the attempt limit, move the message to a separate error queue—the dead-letter queue (DLQ). Return it to the working queue only after fixing the root cause and with controls for re-delivery.
For a store owner, the consequence is quite tangible: one incorrect SKU or an unexpected field can stall handlers, delay valid orders, and turn the error log into noise. The DLQ separates a currently unfixable message from the main flow, but by itself it does not correct the data nor guarantee single processing.
Where the retry ends and the error queue begins
State | Action | Result criterion |
|---|---|---|
Temporary error | Retry with pause and limit | Message processed or limit reached |
Permanent error | Isolate immediately | The main queue continues to operate |
Retry limit exhausted | Move to DLQ | The message no longer cycles in the workflow |
Cause resolved | Return in a controlled manner | No record in DLQ; one record in the working queue |
Repeat Return | Do Not Create a Copy | The operation reports that there is nothing left to return |
We consider a transient error one that may disappear without changing the message: for example, HTTP 503 or a short network timeout. A permanent error is a data or business rule error that the same request will reproduce with the same result: unknown SKU, invalid quantity, missing required identifier. If the classification is doubtful, a limited retry is needed, not an infinite one.
What we check with a small example
The training model uses only the standard Python library and a temporary SQLite file. It does not connect to the network or work with a real broker. After completion, the temporary file system directory is removed along with the test database. The check was performed on October 28, 2026, on CPython 3.12.14 and SQLite 3.53.1.
A transient error twice returns the message for retry; after success, it disappears from the queue and does not enter the DLQ;
A permanent error moves the message to the DLQ after the first failure;
Three transient failures exhaust the limit and isolate the message;
An artificial failure between writing to the DLQ and removing from the working queue rolls back the entire transaction;
The corrected message is returned once, and a subsequent redrive—a controlled return—does not create a copy.
The scenario runs with the standard Python 3.12 interpreter. All identifiers, paths, and data are defined within the example; the database is created in an isolated temporary directory. There are no operations on working files or external queues.
#!/usr/bin/env python3
import sqlite3
import tempfile
from pathlib import Path
MAX_ATTEMPTS = 3
def create_schema(conn):
conn.executescript("""
CREATE TABLE queue (
message_id TEXT PRIMARY KEY,
payload TEXT NOT NULL,
attempts INTEGER NOT NULL DEFAULT 0,
last_error TEXT
);
CREATE TABLE dead_letter (
message_id TEXT PRIMARY KEY,
payload TEXT NOT NULL,
attempts INTEGER NOT NULL,
reason TEXT NOT NULL
);
""")
def enqueue(conn, message_id, payload):
with conn:
conn.execute(
"INSERT INTO queue(message_id, payload) VALUES (?, ?)",
(message_id, payload),
)
def record_failure(conn, message_id, error, *, permanent, inject_crash=False):
with conn:
row = conn.execute(
"SELECT payload, attempts FROM queue WHERE message_id = ?",
(message_id,),
).fetchone()
if row is None:
raise LookupError(message_id)
payload, attempts = row
next_attempt = attempts + 1
if permanent or next_attempt >= MAX_ATTEMPTS:
reason = "permanent" if permanent else "attempt_limit"
conn.execute(
"""INSERT INTO dead_letter(message_id, payload, attempts, reason)
VALUES (?, ?, ?, ?)""",
(message_id, payload, next_attempt, reason),
)
if inject_crash:
raise RuntimeError("simulated crash before source delete")
conn.execute("DELETE FROM queue WHERE message_id = ?", (message_id,))
return "dead_letter"
conn.execute(
"""UPDATE queue
SET attempts = ?, last_error = ?
WHERE message_id = ?""",
(next_attempt, error, message_id),
)
return "retry"
def acknowledge(conn, message_id):
with conn:
conn.execute("DELETE FROM queue WHERE message_id = ?", (message_id,))
def redrive(conn, message_id, corrected_payload):
with conn:
row = conn.execute(
"SELECT 1 FROM dead_letter WHERE message_id = ?", (message_id,)
).fetchone()
if row is None:
return False
conn.execute(
"""INSERT INTO queue(message_id, payload, attempts, last_error)
VALUES (?, ?, 0, NULL)""",
(message_id, corrected_payload),
)
conn.execute(
"DELETE FROM dead_letter WHERE message_id = ?", (message_id,)
)
return True
def count(conn, table, message_id):
if table not in {"queue", "dead_letter"}:
raise ValueError(table)
return conn.execute(
f"SELECT COUNT(*) FROM {table} WHERE message_id = ?", (message_id,)
).fetchone()[0]
with tempfile.TemporaryDirectory(prefix="dlq-lab-") as tmpdir:
db_path = Path(tmpdir) / "queue.sqlite3"
conn = sqlite3.connect(db_path)
create_schema(conn)
enqueue(conn, "transient-ok", "sku=GOOD&qty=1")
assert record_failure(conn, "transient-ok", "HTTP 503", permanent=False) == "retry"
assert record_failure(conn, "transient-ok", "HTTP 503", permanent=False) == "retry"
acknowledge(conn, "transient-ok")
assert count(conn, "queue", "transient-ok") == 0
assert count(conn, "dead_letter", "transient-ok") == 0
print("transient_then_success: queue=0 dead_letter=0")
enqueue(conn, "bad-sku", "sku=UNKNOWN&qty=1")
assert record_failure(conn, "bad-sku", "unknown SKU", permanent=True) == "dead_letter"
assert conn.execute(
"SELECT attempts, reason FROM dead_letter WHERE message_id = ?",
("bad-sku",),
).fetchone() == (1, "permanent")
print("permanent_error: queue=0 dead_letter=1 attempts=1 reason=permanent")
enqueue(conn, "retry-limit", "sku=GOOD&qty=2")
assert record_failure(conn, "retry-limit", "timeout", permanent=False) == "retry"
assert record_failure(conn, "retry-limit", "timeout", permanent=False) == "retry"
assert record_failure(conn, "retry-limit", "timeout", permanent=False) == "dead_letter"
assert conn.execute(
"SELECT attempts, reason FROM dead_letter WHERE message_id = ?",
("retry-limit",),
).fetchone() == (3, "attempt_limit")
print("attempt_limit: queue=0 dead_letter=1 attempts=3 reason=attempt_limit")
enqueue(conn, "rollback", "sku=BROKEN&qty=1")
try:
record_failure(
conn, "rollback", "invalid payload",
permanent=True, inject_crash=True,
)
except RuntimeError as exc:
assert str(exc) == "simulated crash before source delete"
else:
raise AssertionError("injected failure did not happen")
assert count(conn, "queue", "rollback") == 1
assert count(conn, "dead_letter", "rollback") == 0
print("transaction_rollback: queue=1 dead_letter=0")
assert redrive(conn, "bad-sku", "sku=FIXED&qty=1") is True
assert redrive(conn, "bad-sku", "sku=FIXED&qty=1") is False
assert conn.execute(
"SELECT payload, attempts FROM queue WHERE message_id = ?",
("bad-sku",),
).fetchone() == ("sku=FIXED&qty=1", 0)
acknowledge(conn, "bad-sku")
assert count(conn, "queue", "bad-sku") == 0
print("corrected_redrive: first=true second=false final_queue=0 dead_letter=0")
conn.close()
assert not db_path.exists()
Observed result
transient_then_success: queue=0 dead_letter=0
permanent_error: queue=0 dead_letter=1 attempts=1 reason=permanent
attempt_limit: queue=0 dead_letter=1 attempts=3 reason=attempt_limit
transaction_rollback: queue=1 dead_letter=0
corrected_redrive: first=true second=false final_queue=0 dead_letter=0
All assert checks completed successfully with exit code 0. After exiting TemporaryDirectory, the database file was absent. This confirms the behavior of the training model specifically; on RabbitMQ, Amazon SQS, or a custom broker, you must separately verify their delivery guarantees, consumer competition, message retention, and access rights.
Why the transfer must be atomic
The operation 'write to DLQ, then remove from the main queue' consists of two changes. If the process terminates between them without a transaction, you may end up with a copy in both queues or lose the message entirely. In the model, both changes occur within a single SQLite transaction context: an exception before removal rolls back the insertion, leaving the original message in place.
This behavior of the connection context manager — committing on success and rolling back on exception — is described in the sqlite3 documentation for Python 3.12. The manager does not automatically close the connection, so the example explicitly calls close.
Return after fix: there is a catch
Redrive must not mean 'return everything and look'. First, fix the root cause: the SKU reference, message schema, permissions, or handler code. Then select a limited batch, reset the attempt counter according to an explicit rule, and monitor error rates, queue length, and processing time.
In the model, the redrive method inserts the corrected message into the working queue and removes the record from the DLQ within a single transaction. The first execution returns true; the second returns false. This protects only against re-running this specific function in a single-process model; a distributed system must still have an idempotent handler.
In Amazon SQS, an error queue is linked to the source queue via a redrive policy: after a specified number of receives, the message moves to the DLQ. The SQS guide on DLQ separately warns about the impact on FIFO queue ordering. AWS recommends starting with a low rate and monitoring the system upon return; standard redrive does not modify or filter messages. This is described in the redrive guide.
In RabbitMQ, messages can move to a dead-letter exchange after reject or nack without requeue, upon TTL expiration, exceeding the queue length, or reaching the delivery limit for a quorum queue. Conditions and constraints are listed in the official RabbitMQ documentation on dead lettering. The mechanism differs from SQS, so limit values and guarantees cannot be transferred between products.
What to include in a production implementation
stable message_id and consumer idempotency: re-delivery must not create a second order or payment;
separation of transient and permanent errors with an explicit error code, not just an exception message;
attempt limit, backoff with jitter, and a total retry budget to prevent multiple layers from multiplying the load;
atomic transfer or a documented broker guarantee; if impossible, reconciliation by message_id and a duplicate-removal procedure;
metrics for queue length and DLQ size, age of the oldest message, error rates, and return counts;
limited redrive after remediation: a small batch, stop criteria, operator logs, and the ability to return the message to quarantine;
retention period, encryption, access to the payload, and masking of personal data;
separate validation of event processing order: DLQ and re-injection can alter the chronological order of related messages.
How DLQ integrates with other safeguards
The transactional outbox handles a different failure point: the event must appear after the order is saved, even if the sender is temporarily unavailable. A practical example is in the analysis of transactional outbox for webhook.
The retry limit does not replace idempotency: a single message may arrive again due to an acknowledgment timeout, consumer restart, or redrive. This is separately verified in the material about repeated webhooks and idempotency.
The acceptance criterion is simple: a toxic message must no longer block the main flow; its cause and attempt count must be visible; a failure during transfer must not lose the original data. Corrected messages are returned in small batches, at a specified rate and with a stop criterion; re-delivery must not create a second business result. If even one item cannot be verified, the DLQ remains a repository of problems rather than a recovery mechanism.
Обсуждение 0
Делись опытом и задавай вопросы. Комментарии без ссылок появляются после проверки редактором.
Пока никто не написал. Начни обсуждение.