FIFO queues

This guide sends to an SQS FIFO queue in message groups, drops duplicate sends with a deduplication id, and handles several groups at once, each in order.

A FIFO queue keeps the order of the messages within a message group and hands out a group's next message only after the one before it is deleted. Each send also carries a deduplication id, and the queue drops a second send with the same id for five minutes (AWS: deduplication). A queue whose name ends in .fifo is a FIFO queue.

The example is a shipping service. It turns each order into a shipment, and the shipments of one customer go out in the order the customer placed them.

Sending in a group

Macros
Manual

src/main.rs
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "orders")]
struct Order {
    id: u64,
    customer: String,
}

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "shipments.fifo")]
struct Shipment {
    order: u64,
    customer: String,
}

#[derive(OutSlot)]
#[publishes(Shipment)]
struct Shipments;

#[subscriber(SqsQueue::new("orders").create_if_missing())]
async fn dispatch(
    order: &Order,
    Out(shipments): Out<impl Publisher<Options = SqsPublishOptions>, Shipments>,
) -> HandlerOutcome {
    let shipment = Shipment {
        order: order.id,
        customer: order.customer.clone(),
    };
    // A customer's shipments form one group, so they are handled in the order they were sent.
    // The order id is the deduplication id, so sending the same order again ships nothing more.
    let sent = shipments
        .message(&shipment)
        .group_id(&order.customer)
        .deduplication_id(order.id.to_string())
        .publish()
        .await;
    if sent.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

src/main.rs
#[derive(Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Order {
    id: u64,
    customer: String,
}

impl OutgoingDestination for Order {
    type Form = FixedName;

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

impl MessageHeaders for Order {
    type Contract = NoHeaders;
}

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

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

impl OutgoingDestination for Shipment {
    type Form = FixedName;

    const DESTINATION: &'static str = "shipments.fifo";
}

impl MessageHeaders for Shipment {
    type Contract = NoHeaders;
}

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

struct Shipments;

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

impl PublishedThrough<Shipments> for Shipment {}

struct Dispatch;

impl<Wire> Handle<Order, (), Outs<(Wire,)>> for Dispatch
where
    Wire: OutEntry<Shipments, Wire: Publisher<Options = SqsPublishOptions>>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Wire,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let shipment = Shipment {
            order: order.id,
            customer: order.customer.clone(),
        };
        // A customer's shipments form one group, so they are handled in the order they were
        // sent. The order id is the deduplication id, so sending the same order again ships
        // nothing more.
        let sent = outs
            .get(Shipments)
            .message(&shipment)
            .group_id(&order.customer)
            .deduplication_id(order.id.to_string())
            .publish()
            .await;
        if sent.is_err() {
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}

dispatch publishes a shipment to shipments.fifo through the Shipments slot. The slot is bounded on SqsPublishOptions, so the publish takes the steps of SqsPublishSteps. A handler that names these steps imports this crate's prelude.

group_id puts the shipment in the customer's group. When a publish names no group, the group comes from the message's PARTITION_KEY_HEADER, then from the group the policy fixed with Publish::default().group_id(..), and last it is "default". The header is the spelling every RustStream broker reads, so a producer that sets it needs no SQS step.

deduplication_id is the order's id. When orders delivers an order twice, the second shipment is a send the queue drops. A publish that names no deduplication id gets an id unique to the process, so two equal payloads sent on purpose both arrive. A policy carries no deduplication id: one fixed id would drop every send after the first.

A group or a deduplication id on a standard queue is refused with SqsError::NotFifo.

Several groups at once

Macros
Manual

src/main.rs
/// Ships one shipment. Four groups are handled at once, each group in its own order.
#[subscriber(SqsQueue::new("shipments.fifo").create_if_missing(), workers(4, by_key))]
async fn ship(shipment: &Shipment, ctx: &mut Context<'_>) -> HandlerOutcome {
    let group = ctx
        .headers()
        .get_str(PARTITION_KEY_HEADER)
        .unwrap_or_default();
    tracing::info!(shipment.order, group, "shipment sent");
    HandlerOutcome::ack()
}

src/main.rs
/// Ships one shipment.
struct Ship;

impl Handle<Shipment> for Ship {
    fn handle(
        &self,
        shipment: &Shipment,
        _outs: &(),
        ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        let group = ctx
            .headers()
            .get_str(PARTITION_KEY_HEADER)
            .unwrap_or_default();
        tracing::info!(shipment.order, group, "shipment sent");
        ready(Ok(()))
    }
}

A delivery from a FIFO queue carries its group in PARTITION_KEY_HEADER. workers(4, by_key) runs four lanes and sends every message of one group to the same lane. So four customers ship at once, and the shipments of one customer go out one after another. One receive can return several messages of a group, and plain workers(4) would run them side by side.

The app

Macros
Manual

src/main.rs
#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("shipping", "0.1.0")).with_broker(
        SqsBroker::new()
            .endpoint("http://localhost:4566")
            .region("us-east-1")
            .test_credentials(),
        |b| {
            b.include(dispatch)
                .out(Shipments, Publish::default())
                .build();
            b.include(ship);
        },
    )
}

src/main.rs
fn app() -> impl App {
    RustStream::new(AppInfo::new("shipping", "0.1.0")).with_broker(
        SqsBroker::new()
            .endpoint("http://localhost:4566")
            .region("us-east-1")
            .test_credentials(),
        |b| {
            b.include(
                subscriber("orders", Dispatch)
                    .map_source(|_| SqsQueue::new("orders").create_if_missing())
                    .build(),
            )
            .out(Shipments, Publish::default())
            .build();
            // Four groups at once, each group in its own order.
            b.include(
                subscriber("shipments.fifo", Ship)
                    .map_source(|_| SqsQueue::new("shipments.fifo").create_if_missing())
                    .workers_by_key(nonzero!(4))
                    .build(),
            );
        },
    )
}

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

create_if_missing creates shipments.fifo as a FIFO queue with content-based deduplication. A FIFO queue takes a FIFO dead-letter queue; Retries and dead-letter queues covers the cap. SQS limits the throughput of one group, so a service spreads its load over many groups.

The tests in the example file check the group a shipment travels in, send one order twice, and send the orders of two customers at once. Testing an SQS service covers the harness.

Sending in a group​

Several groups at once​

The app​

Sending in a group

Several groups at once

The app