Running in production

This guide runs a Kinesis service on AWS: where its credentials and region come from, which streams it expects to find, and the IAM permissions it needs.

Credentials and region from the environment

KinesisBroker::new() takes nothing from the code. The AWS SDK's default chain finds the region and the credentials when the service starts: environment variables, the shared ~/.aws files, SSO, and the role of the container or the instance. A service that runs under an IAM role on ECS, EKS or EC2 needs no configuration at all. The broker builds its configuration with the SDK's newest behaviour version.

The SDK reads credentials when it signs its first request. That request is the check of the first subscription's stream, at startup, so a service with wrong credentials stops before it handles a record.

A configuration of your own

Macros

src/main.rs

/// The AWS configuration of the service: a pinned behaviour version, its region, and the
/// credentials of the `orders-service` profile, read when the first request is signed.
fn aws_config() -> SdkConfig {
    let credentials = ProfileFileCredentialsProvider::builder()
        .profile_name("orders-service")
        .build();
    SdkConfig::builder()
        .behavior_version(BehaviorVersion::v2026_01_12())
        .region(Region::new("eu-west-1"))
        .credentials_provider(SharedCredentialsProvider::new(credentials))
        .build()
}

Manual

src/main.rs

/// The AWS configuration of the service: a pinned behaviour version, its region, and the
/// credentials of the `orders-service` profile, read when the first request is signed.
fn aws_config() -> SdkConfig {
    let credentials = ProfileFileCredentialsProvider::builder()
        .profile_name("orders-service")
        .build();
    SdkConfig::builder()
        .behavior_version(BehaviorVersion::v2026_01_12())
        .region(Region::new("eu-west-1"))
        .credentials_provider(SharedCredentialsProvider::new(credentials))
        .build()
}

KinesisBroker::from_config takes an SdkConfig the service builds. aws_config builds one from the environment with an async loader, while the app function is synchronous. So the example builds it with SdkConfig::builder(), which does no I/O. The profile provider reads the orders-service profile on the first request; a profile with a role_arn assumes that role. The pinned behaviour version keeps the SDK's defaults from changing under the service when the SDK is updated. The broker uses the configuration as it is: endpoint, region and test_credentials apply to KinesisBroker::new() only, and a configuration of your own names them with endpoint_url, region and credentials_provider on the builder. A DynamoLeaseStore takes the same configuration (Several instances). The example needs two more dependencies:

Cargo.toml

[dependencies]
aws-config = "1"
aws-sdk-kinesis = "1"

The service

Macros

src/main.rs

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "orders")]
struct Order {
    id: u64,
    amount_cents: u64,
}

// The stream belongs to the infrastructure, so the descriptor does not create it.
#[subscriber(KinesisStream::new("orders"))]
async fn take_order(order: &Order) -> HandlerOutcome {
    if order.amount_cents == 0 {
        tracing::warn!(order.id, "order without an amount dropped");
        return HandlerOutcome::drop();
    }
    tracing::info!(order.id, order.amount_cents, "order taken");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        KinesisBroker::from_config(aws_config()),
        |b| {
            b.include(take_order);
        },
    )
}

Manual

src/main.rs

#[derive(Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
struct Order {
    id: u64,
    amount_cents: u64,
}

impl OutgoingDestination for Order {
    type Form = FixedName;

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

impl MessageHeaders for Order {
    type Contract = NoHeaders;
}

struct TakeOrder;

impl Handle<Order> for TakeOrder {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        if order.amount_cents == 0 {
            tracing::warn!(order.id, "order without an amount dropped");
            return ready(Err(HandlerOutcome::drop()));
        }
        tracing::info!(order.id, order.amount_cents, "order taken");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        KinesisBroker::from_config(aws_config()),
        |b| {
            // The stream belongs to the infrastructure, so the descriptor does not create it.
            b.include(subscriber(KinesisStream::new("orders"), TakeOrder).build());
        },
    )
}

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

The orders stream belongs to the infrastructure: Terraform, CloudFormation or the console create it, with its capacity mode, retention and encryption. A stream is named by its name; the account and the region come from the configuration. The descriptor leaves out create_if_missing, so a missing stream stops the service at startup with an error that names it, instead of creating an empty one. create_if_missing(n) creates a provisioned stream with n shards, which suits a local stack; whether a production stream is provisioned or on demand is the operator's choice.

Requests go over the SDK's default HTTPS client, and the SDK retries a failed request with its own retry policy. A handler on a dedicated thread (threads(n)) sends through an SDK client of its own, built from the same configuration and sharing its credentials.

IAM permissions

The service above needs these permissions on its stream:

policy.json

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "kinesis:DescribeStreamSummary",
        "kinesis:ListShards",
        "kinesis:GetShardIterator",
        "kinesis:GetRecords"
      ],
      "Resource": "arn:aws:kinesis:eu-west-1:123456789012:stream/orders"
    }
  ]
}

Each feature adds its own:

Feature	Permissions
a publish, a reply or a retry copy	kinesis:PutRecord on the stream it goes to
create_if_missing	kinesis:CreateStream and kinesis:DescribeStream
DynamoLeaseStore	dynamodb:GetItem and dynamodb:UpdateItem on the lease table

What the AsyncAPI document shows

With the asyncapi feature, the service's AsyncAPI document names one server: the host of the broker's endpoint, kinesis.<region>.amazonaws.com when region names one, or kinesis.amazonaws.com. The document never holds a credential, so it can be published.

The tests in the example file start the same app in process; the in-process Kinesis signs nothing and reads no profile. Testing a Kinesis service covers the harness.