Pub/Sub

Publishers & Subscribers (Pub/Sub) let you build systems that communicate by broadcasting events asynchronously. This is a great way to decouple services for better reliability and responsiveness.

Encore.rs lets you declare topics and subscriptions in code. Encore provisions them for each environment: a managed Pub/Sub service in the cloud, and a local broker under encore run, with no setup.

Here is the whole flow: a topic, a subscription that handles its messages, and a publish.

use encore::pubsub::{Message, Subscription, Topic};
use serde::{Deserialize, Serialize};

#[derive(Clone, Serialize, Deserialize)]
pub struct SignupEvent {
    pub user_id: String,
}

pub static SIGNUPS: Topic<SignupEvent> = Topic::new("signups");

#[encore::pubsub::register]
static WELCOME: Subscription<SignupEvent> =
    Subscription::new(&SIGNUPS, "send-welcome", send_welcome);

async fn send_welcome(msg: Message<SignupEvent>) -> Result<(), encore::Error> {
    println!("welcome, {}", msg.data.user_id);
    Ok(())
}

async fn signed_up(user_id: String) -> Result<(), encore::Error> {
    SIGNUPS.publish(&SignupEvent { user_id }).await?;
    Ok(())
}

Creating a topic

The core of Pub/Sub is the topic, a named channel on which you publish messages. A topic is typed by the message it carries, and declared as a module-level static:

use encore::pubsub::Topic;
use serde::{Deserialize, Serialize};

#[derive(Clone, Serialize, Deserialize)]
pub struct SignupEvent {
    pub user_id: String,
}

pub static SIGNUPS: Topic<SignupEvent> = Topic::new("signups");

Encore reads the declaration from your source code to provision the topic, so:

• It must be a module-level static, not a local variable or a value built inside a function.
• The name must be a string literal: lowercase letters, digits, dashes and periods (user.signed-up), starting with a letter.

At-least-once delivery

By default, a topic delivers each message to each subscription at least once: if Encore believes a message wasn't handled, it delivers it again. So every subscription handler should be idempotent: handling a message twice should have the same effect as handling it once. Track what you've already done in a database, or make the action itself idempotent.

Exactly-once delivery

Declare a topic .exactly_once() to have the infrastructure minimize redelivery:

static PAYMENTS: Topic<PaymentEvent> = Topic::new("payments").exactly_once();

A message may still be redelivered in rare cases, such as when the acknowledgement that it was handled is lost on the network, so handlers should still be idempotent where correctness matters. Exactly-once delivery doesn't deduplicate publishes either: publish the same message twice and it's delivered twice.

Cloud providers limit the throughput of exactly-once topics:

• AWS: 300 messages per second for the topic (see AWS SQS Quotas).
• GCP: at least 3,000 messages per second across all topics in the region (see GCP Pub/Sub Quotas).

.exactly_once() is part of the declaration: it takes the topic by value, so it can't be called on the static later.

Ordered topics

A topic delivers messages in no particular order, which lets them be handled in parallel. When the messages about one entity must be handled in the order they were published, declare an OrderedTopic. Its publish takes an ordering key: messages with the same key are delivered in the order they were published, and messages with different keys in no particular order.

use encore::pubsub::OrderedTopic;

#[derive(Serialize, Deserialize)]
pub struct CartEvent {
    pub cart_id: u64,
    pub event: String,
}

pub static CART_EVENTS: OrderedTopic<CartEvent> = OrderedTopic::new("cart-events");

async fn checkout(cart_id: u64) -> Result<(), encore::Error> {
    let key = cart_id.to_string();
    // Delivered in this order, as they share a key.
    CART_EVENTS.publish(&CartEvent { cart_id, event: "checkout_started".into() }, &key).await?;
    CART_EVENTS.publish(&CartEvent { cart_id, event: "checkout_completed".into() }, &key).await?;
    Ok(())
}

A message with a given key isn't delivered until the one before it has been handled or given up on, so one message that keeps failing holds up the rest of its key (head-of-line blocking). Log and alert on failures, and choose the subscription's retry policy with that in mind.

Cloud providers limit the throughput of ordered topics:

• AWS: 300 messages per second for the topic (see AWS SQS Quotas).
• GCP: 1 MBps for each ordering key (see GCP Pub/Sub Resource Limits).

An ordered topic can also be .exactly_once().

Message encodings

A topic's second type argument says how its messages are encoded:

use encore::pubsub::{Raw, Topic};

static BLOBS: Topic<Vec<u8>, Raw> = Topic::new("blobs");

Protobuf needs the SDK's protobuf feature:

Cargo.toml
[dependencies]
encore-dev = { path = "/path/to/encore-dev", features = ["protobuf"] }

The encoding is part of the topic's type, so publishers and subscribers can't disagree about it.

Publishing messages

Call publish on the topic. It returns the message's id once the message has been published:

let id = SIGNUPS.publish(&SignupEvent { user_id: "ada".into() }).await?;

Once publish returns, every subscription to the topic will receive the message. Subscribers see the same id.

To send attributes alongside a message, use publish_with:

use encore::pubsub::PublishOptions;

SIGNUPS
    .publish_with(
        &event,
        PublishOptions {
            attributes: [("source".to_string(), "import".to_string())].into(),
        },
    )
    .await?;

On an OrderedTopic, publish and publish_with take the ordering key after the message.

Encore reads which services publish to which topics from the calls to publish, and grants each of them permission to. So a topic can only be used by calling its methods on the static directly: it can't be stored, returned, or passed to a function. If library code needs to publish, give it a function that does.

Subscribing to messages

A subscription delivers every message published to its topic to a handler. Declare it as a module-level static with Subscription::new, taking the topic, a subscription name, and the handler:

use encore::pubsub::{Message, Subscription};

#[encore::pubsub::register]
static WELCOME: Subscription<SignupEvent> =
    Subscription::new(&SIGNUPS, "send-welcome", send_welcome);

async fn send_welcome(msg: Message<SignupEvent>) -> Result<(), encore::Error> {
    // Send the email…
    Ok(())
}

• The name must be a string literal, unique for the topic: lowercase letters, digits and dashes.
• The handler is an async fn taking a Message<T> and returning Result<(), encore::Error>. Its future must be Send.
• The subscription runs in the service that declares it, which can be the topic's service or any other.

Each subscription receives every message, independently of the topic's other subscriptions: if one handler is slow, its subscription builds up a backlog while the others keep up.

The message

A handler receives a Message<T>, with these fields:

Starting subscriptions

A static does nothing until something uses it, so something has to start a subscription. #[encore::pubsub::register] registers one to start with the app. Registered subscriptions start when your main calls encore::initialize():

#[tokio::main]
async fn main() {
    // Connects to Encore's runtime and starts the registered subscriptions.
    encore::initialize();

    // Serve HTTP with your framework of choice…
}

A service that only handles messages, and serves no requests of its own, can call encore::run_blocking() from a plain (non-async) main instead. It starts the registered subscriptions and blocks until the process is told to shut down.

fn main() {
    encore::run_blocking();
}

A subscription declared without the attribute is started by hand with WELCOME.start(). Without the attribute, the handler must be written as a closure that boxes the future: |msg| Box::pin(send_welcome(msg)). The attribute is what lets you give the plain name of an async fn.

If a service declares a subscription and never starts it, Encore logs a warning shortly after the service starts.

When the process is told to shut down, Encore stops taking new messages and waits for in-flight handlers to finish. See Graceful shutdown.

Subscription options

Options follow the handler, each a method of its own:

use std::time::Duration;

#[encore::pubsub::register]
static WELCOME: Subscription<SignupEvent> =
    Subscription::new(&SIGNUPS, "send-welcome", send_welcome)
        .ack_deadline(Duration::from_secs(60))
        .max_retries(5)
        .max_concurrency(4);

Encore reads the options from your source code to configure the infrastructure, so write them where the subscription is declared, with literal values: durations as Duration::from_secs(…) or Duration::from_millis(…), and numbers as literals. Each option can be set once.

Error handling

If a handler returns an error, or a message can't be decoded as the topic's type, the message is delivered again later, with a backoff between min_backoff and max_backoff. After max_retries failed retries it's moved to the subscription's dead-letter queue, so the subscription can keep handling other messages until the bug is fixed. You can then release dead-lettered messages to be processed again.

Locally, use the encore pubsub commands to see and manage what happened to each message:

$ encore pubsub ls                                         # topics, subscriptions and their queues
$ encore pubsub queue user-added greet --state dead
$ encore pubsub show <message-id>                          # a message and its delivery attempts
$ encore pubsub release user-added greet                   # deliver dead messages again

Using a topic from another service

Any service can publish to a topic and subscribe to it. In a service other than the one that declares it, bind it by name with Topic::named (or OrderedTopic::named):

use encore::pubsub::Topic;

static SIGNUPS: Topic<SignupEvent> = Topic::named("signups");

The name must be a string literal. A named topic takes no options: it's configured where it's declared, which may be a service in another language. Its message type and encoding are your claim: across languages Encore can't check them, so make sure they match what's published.

Testing

Under encore test, each test publishes to topics of its own, and no subscription runs: nothing handles what a test publishes. That lets you test publishers without the side effects of subscribers.

To check what was published, encore::testing::published returns what the current test, and the code it called, published to a topic, in order. To test a handler, call it with what was published:

#[cfg(test)]
mod tests {
    use super::*;

    #[encore::test]
    async fn signing_up_sends_a_welcome() -> Result<(), encore::Error> {
        signed_up("ada".to_string()).await?;

        let published = encore::testing::published(&SIGNUPS)?;
        assert_eq!(published.len(), 1);
        assert_eq!(published[0].data.user_id, "ada");

        // Run the subscription's handler on it.
        send_welcome(published[0].clone()).await?;
        Ok(())
    }
}

Message<T> is Clone when T is. See Testing for more.

Ensuring consistency between services

Database writes and publishes aren't transactional: a service can write to its database and then fail to publish, or the other way around. If that matters, consider the transactional outbox pattern: write the message to an outbox table in the same transaction as the change, and publish from the table.