docs/go/primitives/pubsub.md
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's Backend Framework lets you use Pub/Sub in a cloud-agnostic declarative fashion. At deployment, Encore automatically provisions the required infrastructure.
The core of Pub/Sub is the Topic, a named channel on which you publish events. Topics must be declared as package level variables, and cannot be created inside functions. Regardless of where you create a topic, it can be published to from any service, and subscribed to from any service.
When creating a topic, it must be given an event type, a unique name, and a configuration to define its behaviour. See the complete specification in the package documentation.
For example, to create a topic with events about user signups:
package user
import "encore.dev/pubsub"
type SignupEvent struct{ UserID int }
var Signups = pubsub.NewTopic[*SignupEvent]("signups", pubsub.TopicConfig{
DeliveryGuarantee: pubsub.AtLeastOnce,
})
<GitHubLink href="https://github.com/encoredev/examples/tree/main/uptime" desc="Event-driven example application using Pub/Sub." />
The above example configures the topic to ensure that, for each subscription, events will be delivered at least once.
This means that if the topic believes the event was not processed, it will attempt to deliver the message again. Therefore, all subscription handlers should be idempotent. This helps ensure that if the handler is called two or more times, from the outside there's no difference compared to calling it once.
This can be achieved using a database to track if you have already performed the action that the event is meant to trigger, or ensuring that the action being performed is also idempotent in nature.
Topics can also be configured to deliver events exactly once by setting the DeliveryGuarantee field to
pubsub.ExactlyOnce. This enables stronger guarantees on the infrastructure level to minimize the likelihood of
message re-delivery.
However, there are still some rare circumstances when a message might be redelivered. For example, if a networking issue causes the acknowledgement of successful processing the message to be lost before the cloud provider receives it (the Two Generals' Problem). As such, if correctness is critical under all circumstances, it's still advisable to design your subscription handlers to be idempotent.
By enabling exactly-once delivery on a topic the cloud provider enforces certain throughput limitations:
Exactly-once delivery does not perform message deduplication on the publishing side. If Publish is called twice with
the same message, the message will be delivered twice.
Topics are unordered by default, meaning that messages can be delivered in any order. This allows for better throughput on the topic as messages can be processed in parallel. However, in some cases, messages must be delivered in the order they were published for a given entity.
To create an ordered topic, configure the topic's OrderingAttribute to match the pubsub-attr tag on one of the top-level fields of the event type. This field ensures that messages delivered to the same subscriber are delivered in the order of publishing for that specific field value. Messages with a different value on the ordering attribute are delivered in an unspecified order.
To maintain topic order, messages with the same ordering key aren't delivered until the earliest message is processed or dead-lettered, potentially causing delays due to head-of-line blocking. Mitigate processing issues by ensuring robust logging and alerts, and appropriate subscription retry policies.
<Callout type="info">The OrderingAttribute currently has no effect in local environments.
Each cloud provider enforces certain throughput limitations for ordered topics:
package example
import (
"context"
"encore.dev/pubsub"
)
type CartEvent struct {
ShoppingCartID int `pubsub-attr:"cart_id"`
Event string
}
var CartEvents = pubsub.NewTopic[*CartEvent]("cart-events", pubsub.TopicConfig{
DeliveryGuarantee: pubsub.AtLeastOnce,
OrderingAttribute: "cart_id",
})
func Example(ctx context.Context) error {
// These are delivered in order as they all have the same shopping cart ID
CartEvents.Publish(ctx, &CartEvent{ShoppingCartID: 1, Event: "item_added"})
CartEvents.Publish(ctx, &CartEvent{ShoppingCartID: 1, Event: "checkout_started"})
CartEvents.Publish(ctx, &CartEvent{ShoppingCartID: 1, Event: "checkout_completed"})
// This event may be delivered at any point as it has a different shopping cart ID
CartEvents.Publish(ctx, &CartEvent{ShoppingCartID: 2, Event: "item_added"})
}
To publish an Event, call Publish on the topic passing in the event object (which is the type specified in the pubsub.NewTopic[Type] constructor).
For example:
messageID, err := Signups.Publish(ctx, &SignupEvent{UserID: id})
if err != nil {
return err
}
// If we get here the event has been successfully published,
// and all registered subscribers will receive the event.
// The messageID variable contains the unique id of the message,
// which is also provided to the subscribers when processing the event.
By defining the Signups topic variable as an exported variable
you can also publish to the topic from other services in the same way.
Encore uses static analysis to determine which services are publishing messages to what topics. That information is used to provision infrastructure correctly, render architecture diagrams, and configure IAM permissions.
This means that *pubsub.Topic variables can't be passed around however you'd like,
as it makes static analysis impossible in many cases. To work around these restrictions
Encore allows you to get a reference to a topic that can be passed around any way you want.
It looks like this (using the Signups topic above):
signupRef := pubsub.TopicRef[pubsub.Publisher[*SignupEvent]](Signups)
// signupRef is of type pubsub.Publisher[*SignupEvent], which allows publishing.
The difference between a TopicRef and a Topic is that topic references need to pre-declare what permissions are needed. Encore then assumes that all the permissions you declare are used.
For example, if you declare a TopicRef with the pubsub.Publisher permission (as seen above)
Encore assumes that the service will publish messages to the topic and provisions the infrastructure
to support that.
Note that a TopicRef must be declared within a service, but the reference itself can be freely passed around to library code, be dependency injected into service structs, and so on.
To Subscribe to events, you create a Subscription as a package level variable by calling the
pubsub.NewSubscription function.
Each subscription needs:
Handler function to process the eventsHere's an example of how you create a subscription to a topic:
package email
import (
"encore.dev/pubsub"
"user"
)
var _ = pubsub.NewSubscription(
user.Signups, "send-welcome-email",
pubsub.SubscriptionConfig[*SignupEvent]{
Handler: SendWelcomeEmail,
},
)
func SendWelcomeEmail(ctx context.Context, event *SignupEvent) error {
// send email...
return nil
}
Subscriptions can be in the same service as the topic is declared, or in any other service of your application. Each subscription to a single topic receives the events independently of any other subscriptions to the same topic. This means that if one subscription is running very slowly, it will grow a backlog of unprocessed events. However, any other subscriptions will still be processing events in real-time as they are published.
The ctx passed to the handler function is cancelled when the AckDeadline for the subscription is reached.
This is the time when the message is considered to have timed out and can be redelivered to another subscriber.
The timeout defaults to 30 seconds if you don't explicitly configure AckDeadline.
When using service structs for dependency injection
it's common to want to define the subscription handler as a method on the service struct, to be able to access the
injected dependencies. The pubsub package provides the pubsub.MethodHandler function for this purpose:
//encore:service
type Service struct { /* ... */ }
func (s *Service) SendWelcomeEmail(ctx context.Context, event *SignupEvent) error {
// ...
}
var _ = pubsub.NewSubscription(
user.Signups, "send-welcome-email",
pubsub.SubscriptionConfig[*SignupEvent]{
Handler: pubsub.MethodHandler((*Service).SendWelcomeEmail),
},
)
Note that pubsub.MethodHandler only allows referencing methods on the service struct type, not any other type.
When creating a subscription you can configure behavior such as message retention and retry policy, using the SubscriptionConfig type. See the package documentation for the complete configuration options.
The SubscriptionConfig struct fields must be defined as compile-time constants, and cannot be defined in
terms of function calls. This is necessary for Encore to understand the exact requirements of the subscription, in order to provision the correct infrastructure upon deployment.
If a subscription function returns an error, the event being processed will be retried, based on the retry policy
configured on that subscription. After the MaxRetries is hit,
the event will be placed into a dead-letter queue (DLQ) for that subscriber. This allows the subscription to continue
processing events until the bug which caused the event to fail can be fixed. Once fixed, the messages on the dead-letter queue can be manually released to be processed again by the subscriber.
Encore uses a special testing implementation of Pub/Sub topics. When running tests, topics are aware of which test is running. This gives you the following guarantees:
Encore provides a helper function, et.Topic, to access the testing topic. You
can use this object to extract the events that have been published to it during a test.
Here's an example implementation:
package user
import (
"testing"
"encore.dev/et"
"github.com/stretchr/testify/assert"
)
func Test_Register(t *testing.T) {
t.Parallel()
... Call Register() and assert changes to the database ...
// Get all published messages on the Signups topic from this test.
msgs := et.Topic(Signups).PublishedMessages()
assert.Len(t, msgs, 1)
}
Ensuring consistency between services in event-driven applications can be challenging, especially when database writes and Pub/Sub publishing are not transactional. This can lead to inconsistencies between services.
To address this issue without adding excessive complexity, consider using a transactional outbox pattern. For more information on implementing this pattern with Encore, see the Pub/Sub Outbox guide.
Pub/Sub is a powerful building block in a backend application. It can be used to improve app reliability by reducing the blast radius of faulty components and bottlenecks. It can also be used to increase the speed of response to the user, and even helps reduce cognitive overhead for developers by inverting the dependencies between services.
For those not familiar with Pub/Sub, lets take a look at an example API in a user registration service. The behavior we want to implement is that upon registration, we send a welcome email to the user and create a record of the signup in our analytics system. Now let's see how we could implement this only using APIs, compared to how a Pub/Sub implementation might look.
Using API calls between services, we might design a system which looks like this when the user registers:
<div className="grid grid-cols-3 mobile:grid-cols-1 grid-flow-row"> <div className="col-span-2">user service starts a database transaction and records the user in its database.user service makes a call to the email service to send a welcome email.email service then calls an email provider to actually send the email.email service replies to the user service that the request was processed.user service then calls the analytics service to record the signup.analytics service the writes to the data warehouse to record the information.analytics service then replies to the user service that the request was processed.user service commits the database transaction.user service then can reply to the user to say the registration was successful.Notice how we have to wait for everything to complete before we can reply to the user to tell then we've registered them. This means that if our email provider takes 3 seconds to send the email, we've now taken 3 seconds to respond to the user, when in reality once the user was written to the database, we could have responded to the user instantly at that point to confirm the registration.
Another downside to this approach is if our data warehouse is currently broken and reporting errors, our system will also report errors whenever anybody tries to signup! Given analytics is purely internal and doesn't impact users, why should the analytics system being down impact user signup?
A more ideal solution would be if we could decouple the behaviour of emailing the user and recording our analytics, such that the user service only has to record the user in its own database and let the user know they are registered - without worrying about the downstream impacts. Thankfully, this is exactly what Pub/Sub topics allow us to do.
<div className="grid grid-cols-3 mobile:grid-cols-1 grid-flow-row"> <div className="col-span-2">In this example, when a user registers we:
user service starts a database transaction and records the user in its database.signups topic.At this point the user is free to continue interacting with the application and we've isolated the registration behaviour from the rest of the application.
In parallel, the email and analytics services will receive the signup event from the signups topic and will then
perform their respective tasks. If either service returns an error, the event will automatically be backed off and retried
until the service is able to process the event successfully, or reaches the maximum number of attempts and is placed
into the deadletter queue (DLQ).
Notice how in this version, the processing time of the two other services did not impact the end user and in fact the user
service is not even aware of the email and analytics services. This means that new systems which need to know about
new users signing up can be added to the application, without the need to change the user service or impacting its
performance.