Skip to main content

Checkpoints and delivery guarantees

This guide shows what each outcome of a handler does to a shard's checkpoint, what is read again after a restart, and how a record that can never be handled is retired.

Kinesis keeps no state per record and per reader. A reader remembers how far it got in each shard, its checkpoint, and starts from there next time. In RustStream the handler's outcome moves the checkpoint.

The service​

The example ships orders with a carrier. A shipment without an address can never ship. A shipment for a carrier this release of the service does not know waits for a release that knows it.

src/main.rs
/// The carriers this release of the service books with.
const CARRIERS: [&str; 2] = ["dhl", "ups"];

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "shipments")]
struct Shipment {
    id: u64,
    address: String,
    carrier: String,
}

#[subscriber(KinesisStream::new("shipments").create_if_missing(1))]
async fn ship(shipment: &Shipment) -> HandlerOutcome {
    if shipment.address.is_empty() {
        tracing::warn!(shipment.id, "shipment without an address retired");
        return HandlerOutcome::drop();
    }
    if !CARRIERS.contains(&shipment.carrier.as_str()) {
        tracing::warn!(
            shipment.id,
            shipment.carrier,
            "unknown carrier, shipment held"
        );
        return HandlerOutcome::retry();
    }
    tracing::info!(shipment.id, shipment.carrier, "shipment booked");
    HandlerOutcome::ack()
}

fn service(broker: KinesisBroker) -> RustStream {
    RustStream::new(AppInfo::new("shipping", "0.1.0")).with_broker(broker, |b| {
        b.include(ship);
    })
}

#[ruststream::app]
fn app() -> impl App {
    service(
        KinesisBroker::new()
            .endpoint("http://localhost:4566")
            .region("us-east-1")
            .test_credentials(),
    )
}

What each outcome does​

OutcomeThe recordThe shard's checkpoint
ackhandledmoves past it once every earlier record of the shard is handled
dropretired, never read againas for ack
retry, with no cap at the mount siteleft unhandledstops in front of it

A checkpoint covers every record before it, so it advances only over a run of handled records. The crate writes it to the lease store when it moves. A record handled out of order, by a parallel lane or in a batch, moves nothing until the records before it are handled too.

drop is how ship retires a shipment without an address: another try gets the same answer.

retry asks Kinesis to deliver the record again, and a shard cannot do that for one record: it is a log read in order. The delivery answers AckError::Unsupported, which the runtime logs as a warning, and the record stays unhandled. ship holds a shipment for an unknown carrier this way. The reader goes on: the shipments behind it are delivered and handled, and the checkpoint waits in front of the held one. Only its own shard waits; the checkpoints of the other shards move on. Under a cap declared at the mount site, retry publishes a copy instead (Retries and dead-letter streams).

What is read again​

Delivery is at-least-once. A reader that takes a shard starts after its checkpoint, so everything from the first unhandled record on is delivered again: the held shipment, and every shipment behind it, handled or not. That happens when the service restarts, and when another instance takes the shard over (Several instances). So a handler that can see a record twice is written to be idempotent: booking the same shipment twice books it once.

The checkpoint lives in the lease store, and the store decides what survives a restart. The default MemoryLeaseStore lives in the process and starts empty, so a restarted service opens each shard at its tip and reads only what is published from then on. DynamoLeaseStore keeps the checkpoints in a table, and a restarted service resumes where the last one stopped. A shard with no checkpoint opens where start_at says, and at the tip when nothing says otherwise (Replay and seeking).

A held record does not come back while the same reader holds the shard. To read it again without a restart, a handler seeks back to it.

Batches and closed shards​

A batch handler that returns one outcome settles every record of the batch with it. One that returns Vec<HandlerOutcome> settles each record on its own, and each moves the checkpoint of its own shard.

A shard that a split or a merge closed is read to its end. Its checkpoint becomes SHARD_END only after every delivery from it is settled, and only then do its children start. A held record in a closed shard therefore holds its children back as well.

Testing the outcomes​

The tests in the example file run the app in process and check how each shipment settled. The in-process Kinesis keeps no checkpoints, so a held record does not come back there either. What a restart reads again runs against a stack, in the test of the Settlement example: it restarts the service over a store that outlives the first run and sees the held record first.