Подтверждение и гарантии доставки

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

SQS доставляет хотя бы один раз. Полученное сообщение остаётся в очереди, невидимое, пока потребитель его не удалит. Стандартная очередь может также доставить сообщение дважды и не в том порядке, в каком его отправили (AWS: доставка at-least-once).

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

Что делает каждый исход

Каждый исход - один вызов SQS для сообщения, которое получил обработчик:

Исход	Вызов SQS	Что происходит с сообщением
ack	DeleteMessage	оно покидает очередь
drop	DeleteMessage	оно покидает очередь: в SQS нет вызова, который отвергает сообщение и сохраняет его
retry	ChangeMessageVisibility в ноль	оно сразу выдаётся повторно
retry_after	ChangeMessageVisibility на задержку	оно выдаётся повторно, когда задержка заканчивается

Сообщение, которое обработчик так и не завершил, потому что процесс остановился посреди работы, вызова не получает. Оно возвращается, когда истекает его тайм-аут видимости. Поэтому обработчик выполняет работу до исхода, а сообщение удаляется только после того, как работа сделана.

Обработчик, который проводит платёж один раз

Макросы

src/main.rs

/// How long a payment on an account under review waits before the next try.
const REVIEW: Duration = Duration::from_secs(60);

/// The receive after which a payment under review is given up.
const LAST_RECEIVE: u32 = 3;

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "payments")]
struct Payment {
    id: u64,
    account: String,
    amount_cents: u64,
    under_review: bool,
}

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "ledger-entries")]
struct Entry {
    payment: u64,
    account: String,
    amount_cents: u64,
}

#[derive(OutSlot)]
#[publishes(Entry)]
struct Entries;

/// The ids of the payments already booked.
#[derive(Debug, Default, Clone)]
struct Ledger {
    booked: Arc<Mutex<HashSet<u64>>>,
}

#[subscriber(SqsQueue::new("payments"))]
async fn book(
    payment: &Payment,
    ctx: &mut Context<'_, (), Ledger>,
    Out(entries): Out<impl Publisher, Entries>,
) -> HandlerOutcome {
    // SQS counts the receives of a message, this one included.
    let receives: u32 = ctx
        .headers()
        .get_str(RECEIVE_COUNT_HEADER)
        .and_then(|count| count.parse().ok())
        .unwrap_or(1);
    if payment.under_review {
        if receives >= LAST_RECEIVE {
            tracing::error!(payment.id, receives, "payment dropped, still under review");
            return HandlerOutcome::drop();
        }
        return HandlerOutcome::retry_after(REVIEW);
    }
    let booked = Arc::clone(&ctx.state().booked);
    if booked
        .lock()
        .unwrap_or_else(PoisonError::into_inner)
        .contains(&payment.id)
    {
        tracing::info!(payment.id, receives, "payment already booked");
        return HandlerOutcome::ack();
    }
    let entry = Entry {
        payment: payment.id,
        account: payment.account.clone(),
        amount_cents: payment.amount_cents,
    };
    if entries.message(&entry).publish().await.is_err() {
        // Nothing is booked yet: the payment comes back and the work runs again.
        return HandlerOutcome::retry();
    }
    booked
        .lock()
        .unwrap_or_else(PoisonError::into_inner)
        .insert(payment.id);
    tracing::info!(payment.id, "payment booked");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

/// How long a payment on an account under review waits before the next try.
const REVIEW: Duration = Duration::from_secs(60);

/// The receive after which a payment under review is given up.
const LAST_RECEIVE: u32 = 3;

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

impl OutgoingDestination for Payment {
    type Form = FixedName;

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

impl MessageHeaders for Payment {
    type Contract = NoHeaders;
}

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

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

impl OutgoingDestination for Entry {
    type Form = FixedName;

    const DESTINATION: &'static str = "ledger-entries";
}

impl MessageHeaders for Entry {
    type Contract = NoHeaders;
}

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

struct Entries;

impl OutSlot for Entries {
    const NAME: &'static str = "Entries";
    type Destination = Reads;
}

impl PublishedThrough<Entries> for Entry {}

/// The ids of the payments already booked.
#[derive(Debug, Default, Clone)]
struct Ledger {
    booked: Arc<Mutex<HashSet<u64>>>,
}

struct Book;

impl<Wire> Handle<Payment, (), Outs<(Wire,)>, (), Ledger> for Book
where
    Wire: OutEntry<Entries, Wire: Publisher>,
{
    async fn handle(
        &self,
        payment: &Payment,
        outs: &Outs<(Wire,)>,
        ctx: &mut Context<'_, (), Ledger>,
    ) -> Result<(), HandlerOutcome> {
        // SQS counts the receives of a message, this one included.
        let receives: u32 = ctx
            .headers()
            .get_str(RECEIVE_COUNT_HEADER)
            .and_then(|count| count.parse().ok())
            .unwrap_or(1);
        if payment.under_review {
            if receives >= LAST_RECEIVE {
                tracing::error!(payment.id, receives, "payment dropped, still under review");
                return Err(HandlerOutcome::drop());
            }
            return Err(HandlerOutcome::retry_after(REVIEW));
        }
        let booked = Arc::clone(&ctx.state().booked);
        if booked
            .lock()
            .unwrap_or_else(PoisonError::into_inner)
            .contains(&payment.id)
        {
            tracing::info!(payment.id, receives, "payment already booked");
            return Ok(());
        }
        let entry = Entry {
            payment: payment.id,
            account: payment.account.clone(),
            amount_cents: payment.amount_cents,
        };
        if outs.get(Entries).message(&entry).publish().await.is_err() {
            // Nothing is booked yet: the payment comes back and the work runs again.
            return Err(HandlerOutcome::retry());
        }
        booked
            .lock()
            .unwrap_or_else(PoisonError::into_inner)
            .insert(payment.id);
        tracing::info!(payment.id, "payment booked");
        Ok(())
    }
}

book записывает в книгу учёта проводку платежа и затем подтверждает его. Платёж, который SQS доставил второй раз, находится среди проведённых идентификаторов и подтверждается без второй проводки. Множество в состоянии приложения заменяет хранилище, в котором сервис держит проведённые идентификаторы; идентификатор - собственный идентификатор платежа, а не идентификатор сообщения SQS, потому что отправитель, который отправил дважды, отправил два сообщения. Когда проводку не удаётся опубликовать, ничего не проводится, и retry возвращает платёж для новой попытки.

Обработчик читает RECEIVE_COUNT_HEADER - собственный ApproximateReceiveCount очереди. Он учитывает и текущее получение, поэтому у первой доставки там 1. Платёж по счёту на проверке ждёт минуту через retry_after, а на третьем получении обработчик сдаётся и отбрасывает его. Как сама очередь ограничивает число получений и сохраняет то, что не удалось обработать, показывает руководство Переобработка и DLQ.

Пакет завершается по одному сообщению

Макросы

src/main.rs

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "refunds")]
struct Refund {
    payment: u64,
    amount_cents: u64,
}

/// The largest refund paid without a person's approval.
const APPROVAL_LIMIT_CENTS: u64 = 50_000;

/// How long a refund waits for approval before it is looked at again.
const APPROVAL: Duration = Duration::from_secs(300);

/// Settles each refund of a batch on its own: a small refund is paid and deleted, a large one
/// waits on the queue for its approval.
#[subscriber(SqsQueue::new("refunds"))]
async fn settle_refunds(refunds: &[Refund]) -> Vec<HandlerOutcome> {
    refunds
        .iter()
        .map(|refund| {
            if refund.amount_cents > APPROVAL_LIMIT_CENTS {
                tracing::info!(refund.payment, "refund waits for approval");
                HandlerOutcome::retry_after(APPROVAL)
            } else {
                tracing::info!(refund.payment, refund.amount_cents, "refund paid");
                HandlerOutcome::ack()
            }
        })
        .collect()
}

Вручную

src/main.rs

#[derive(Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Refund {
    payment: u64,
    amount_cents: u64,
}

impl OutgoingDestination for Refund {
    type Form = FixedName;

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

impl MessageHeaders for Refund {
    type Contract = NoHeaders;
}

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

/// The largest refund paid without a person's approval.
const APPROVAL_LIMIT_CENTS: u64 = 50_000;

/// How long a refund waits for approval before it is looked at again.
const APPROVAL: Duration = Duration::from_secs(300);

/// Settles each refund of a batch on its own: a small refund is paid and deleted, a large one
/// waits on the queue for its approval.
struct SettleRefunds;

impl Handle<[Refund], (), (), (), Ledger> for SettleRefunds {
    fn handle(
        &self,
        refunds: &[Refund],
        _outs: &(),
        _ctx: &mut Context<'_, (), Ledger>,
    ) -> impl Future<Output = Result<(), Vec<HandlerOutcome>>> {
        let outcomes: Vec<HandlerOutcome> = refunds
            .iter()
            .map(|refund| {
                if refund.amount_cents > APPROVAL_LIMIT_CENTS {
                    tracing::info!(refund.payment, "refund waits for approval");
                    HandlerOutcome::retry_after(APPROVAL)
                } else {
                    tracing::info!(refund.payment, refund.amount_cents, "refund paid");
                    HandlerOutcome::ack()
                }
            })
            .collect();
        ready(Err(outcomes))
    }
}

Обработчик пакетов возвращает по одному исходу на сообщение, в порядке среза. Каждый исход - отдельный вызов для своего сообщения. Здесь мелкие возвраты удаляются, а крупный остаётся в очереди и через пять минут возвращается один. Обработчик пакетов, который возвращает один исход, завершает им каждое сообщение пакета.

Приложение

Макросы

src/main.rs

#[ruststream::app]
fn app() -> impl App<State = Ledger> {
    RustStream::new(AppInfo::new("payments", "0.1.0"))
        .on_startup(async move |()| Ok::<_, Infallible>(Ledger::default()))
        .with_broker(
            SqsBroker::new()
                .endpoint("http://localhost:4566")
                .region("us-east-1")
                .test_credentials(),
            |b| {
                b.include(book).out(Entries, Publish::default()).build();
                b.include(settle_refunds.batch(nonzero!(10)));
            },
        )
}

Вручную

src/main.rs

fn app() -> impl App<State = Ledger> {
    RustStream::new(AppInfo::new("payments", "0.1.0"))
        .on_startup(async move |()| Ok::<_, Infallible>(Ledger::default()))
        .with_broker(
            SqsBroker::new()
                .endpoint("http://localhost:4566")
                .region("us-east-1")
                .test_credentials(),
            |b| {
                b.include(
                    subscriber("payments", Book)
                        .map_source(|_| SqsQueue::new("payments"))
                        .build(),
                )
                .out(Entries, Publish::default())
                .build();
                b.include(
                    subscriber("refunds", SettleRefunds)
                        .batch(nonzero!(10))
                        .map_source(|_| SqsQueue::new("refunds"))
                        .build(),
                );
            },
        )
}

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

book публикует проводки через слот Entries, привязанный к издателю очереди в точке монтирования. Очередь FIFO в течение пяти минут отбрасывает повторную отправку с тем же идентификатором дедупликации, и это не даёт отправителю отправить одно сообщение дважды; это показывает руководство Очереди FIFO. Обработчик на стандартной очереди всё равно ведёт собственный учёт того, что уже сделал.

Тесты в файле примера доставляют платёж дважды, на остановленных часах дают платежу на проверке исчерпать получения и отправляют три возврата одним пакетом. Обвязку разбирает руководство Тестирование сервиса на SQS.