Переобработка и DLQ

Это руководство обрабатывает сообщение SQS заново сразу или позже, ограничивает число его доставок и поручает SQS переносить в DLQ то, что раз за разом не обрабатывается.

Обе задачи SQS решает сам. Переобработка - это то же сообщение, снова сделанное видимым, с сохранённым счётчиком получений. Предел и DLQ - это политика перенаправления очереди, и объявляет её точка монтирования.

Пример - сервис счетов, который списывает оплату счетов с клиентов.

Переобработка сейчас и позже

Макросы

src/main.rs

/// The largest invoice charged without a credit check.
const CREDIT_LIMIT_CENTS: u64 = 100_000;

/// How long an invoice waits for the credit check before it comes back.
const CREDIT_CHECK: Duration = Duration::from_secs(30);

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "invoices")]
struct Invoice {
    id: u64,
    customer: String,
    amount_cents: u64,
}

#[subscriber(SqsQueue::new("invoices").create_if_missing())]
async fn charge(invoice: &Invoice) -> HandlerOutcome {
    if invoice.customer.is_empty() {
        // Another try gets the same answer.
        tracing::warn!(invoice.id, "invoice without a customer dropped");
        return HandlerOutcome::drop();
    }
    if invoice.amount_cents > CREDIT_LIMIT_CENTS {
        tracing::info!(invoice.id, "invoice waits for the credit check");
        return HandlerOutcome::retry_after(CREDIT_CHECK);
    }
    tracing::info!(invoice.id, invoice.amount_cents, "invoice charged");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

/// The largest invoice charged without a credit check.
const CREDIT_LIMIT_CENTS: u64 = 100_000;

/// How long an invoice waits for the credit check before it comes back.
const CREDIT_CHECK: Duration = Duration::from_secs(30);

#[derive(Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Invoice {
    id: u64,
    customer: String,
    amount_cents: u64,
}

impl OutgoingDestination for Invoice {
    type Form = FixedName;

    const DESTINATION: &'static str = "invoices";
}

impl MessageHeaders for Invoice {
    type Contract = NoHeaders;
}

impl MessageInfo for Invoice {
    const NAME: &'static str = "Invoice";
}

struct Charge;

impl Handle<Invoice> for Charge {
    fn handle(
        &self,
        invoice: &Invoice,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        if invoice.customer.is_empty() {
            // Another try gets the same answer.
            tracing::warn!(invoice.id, "invoice without a customer dropped");
            return ready(Err(HandlerOutcome::drop()));
        }
        if invoice.amount_cents > CREDIT_LIMIT_CENTS {
            tracing::info!(invoice.id, "invoice waits for the credit check");
            return ready(Err(HandlerOutcome::retry_after(CREDIT_CHECK)));
        }
        tracing::info!(invoice.id, invoice.amount_cents, "invoice charged");
        ready(Ok(()))
    }
}

Счёт выше кредитного лимита ждёт проверки кредита. retry_after задаёт тайм-аут видимости сообщения 30 секунд, и по их истечении SQS выдаёт его повторно. SQS считает целые секунды, поэтому доля секунды округляется вверх. SQS скрывает сообщение не дольше двенадцати часов с момента получения, поэтому задержка заканчивается в их пределах. Более долгая задержка сокращается до конца этих двенадцати часов, и крейт пишет в журнал предупреждение с именем очереди. Всё это время сообщение остаётся в своей очереди, поэтому отложенная переобработка переживает перезапуск сервиса.

retry задаёт нулевую видимость, и сообщение сразу выдаётся повторно. Счёт без клиента при каждой попытке получает один и тот же ответ, поэтому charge его отбрасывает.

Предел и DLQ

Макросы

src/main.rs

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("invoices", "0.1.0")).with_broker(
        SqsBroker::new()
            .endpoint("http://localhost:4566")
            .region("us-east-1")
            .test_credentials(),
        |b| {
            // Four deliveries of one invoice, the first included; then SQS moves it.
            b.include(charge)
                .max_attempts(nonzero!(4u32))
                .dead_letter("invoices-dead");
            b.include(review);
        },
    )
}

Вручную

src/main.rs

fn app() -> impl App {
    RustStream::new(AppInfo::new("invoices", "0.1.0")).with_broker(
        SqsBroker::new()
            .endpoint("http://localhost:4566")
            .region("us-east-1")
            .test_credentials(),
        |b| {
            // Four deliveries of one invoice, the first included; then SQS moves it.
            b.include(
                subscriber("invoices", Charge)
                    .map_source(|_| SqsQueue::new("invoices").create_if_missing())
                    .build(),
            )
            .max_attempts(nonzero!(4u32))
            .dead_letter("invoices-dead");
            b.include(
                subscriber("invoices-dead", Review)
                    .map_source(|_| SqsQueue::new("invoices-dead").create_if_missing())
                    .build(),
            );
        },
    )
}

fn main() -> ExitCode {
    run_main(app)
}

max_attempts и dead_letter вместе образуют политику перенаправления очереди. Предел становится её maxReceiveCount, а DLQ - её целью. Подписка записывает политику в очередь при открытии. С этого момента SQS сам считает получения и переносит сообщение, у которого они кончились; сервис ничего не публикует. Здесь счёт доставляется четыре раза, а на пятом получении SQS переносит его в invoices-dead.

Два шага задаются вместе. Предел без DLQ или DLQ без предела останавливают сервис при старте с ошибкой SqsError::IncompleteRedrive, которая называет недостающую половину. SQS принимает maxReceiveCount не больше 1000.

DLQ должна существовать, когда открывается подписка, если только её не создаёт дескриптор приложения с create_if_missing. Она должна быть того же вида, что и её очередь: очереди FIFO нужна DLQ типа FIFO. Для записи политики нужны sqs:SetQueueAttributes на очереди и sqs:GetQueueAttributes на DLQ.

drop удаляет сообщение. На очереди с пределом drop доставки, которая израсходовала последнее получение, вместо удаления возвращает сообщение, и SQS переносит его в DLQ, а не теряет.

Чтение DLQ

Макросы

src/main.rs

/// Reads the invoices SQS moved away after their last delivery.
#[subscriber(SqsQueue::new("invoices-dead").create_if_missing())]
async fn review(invoice: &Invoice) -> HandlerOutcome {
    tracing::error!(invoice.id, invoice.customer, "invoice needs a review");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

/// Reads the invoices SQS moved away after their last delivery.
struct Review;

impl Handle<Invoice> for Review {
    fn handle(
        &self,
        invoice: &Invoice,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::error!(invoice.id, invoice.customer, "invoice needs a review");
        ready(Ok(()))
    }
}

DLQ - обычная очередь, и review читает её, как любую другую. Сообщение приходит таким, каким его отправили, со своими атрибутами. Очередь без читателя хранит сообщения до конца срока хранения, поэтому человек может посмотреть их и в консоли.

Тестирование задержки

Тесты в файле примера запускают приложение внутри процесса на остановленных часах, поэтому tb.advance(CREDIT_CHECK) пропускает 30 секунд мгновенно. Один тест проводит счёт через четыре доставки в invoices-dead, другой проверяет, что предел без DLQ не запускается. Обвязку разбирает руководство Тестирование сервиса на SQS.