Documentation
¶
Overview ¶
Package outbox implements the transactional-outbox pattern: events are written into an outbox table within the same database transaction as the business data, and a background Relay publishes pending events at-least-once.
Index ¶
Examples ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func Enqueue ¶
Enqueue JSON-encodes payload and inserts a pending Event using db, which may be a *bun.DB or a bun.Tx. Pass your business transaction so the event commits atomically with your writes.
func Migrate ¶
Migrate creates the outbox table (if it does not exist) with an index on (published_at, id) so pending-event scans stay cheap, and the outbox_dead and outbox_attempts tables used by WithMaxAttempts.
func Prune ¶
Prune deletes events published more than olderThan ago and returns how many were removed. Published rows are never deleted otherwise, so call it periodically (e.g. from a scheduled job) to keep the table small. It also drops attempt counters left behind by events that are no longer pending.
Types ¶
type Attempt ¶ added in v2.1.0
type Attempt struct {
bun.BaseModel `bun:"table:outbox_attempts,alias:oa"`
EventID int64 `bun:"event_id,pk"`
Attempts int `bun:"attempts,notnull"`
}
Attempt counts the failed publishes of one pending event. The relay keeps it in the database (see WithMaxAttempts), so the count survives restarts and is shared by every replica.
type DeadEvent ¶
type DeadEvent struct {
bun.BaseModel `bun:"table:outbox_dead,alias:od"`
ID int64 `bun:"id,pk"`
Topic string `bun:"topic,notnull"`
Payload []byte `bun:"payload,notnull"`
CreatedAt time.Time `bun:"created_at,notnull"`
FailedAt time.Time `bun:"failed_at,notnull"`
Error string `bun:"error,notnull"`
}
DeadEvent is an event the relay gave up on (see WithMaxAttempts). It keeps the original ID, topic, payload, and creation time, plus when and why publishing was abandoned. Re-enqueue it with Enqueue to retry.
type Event ¶
type Event struct {
bun.BaseModel `bun:"table:outbox,alias:o"`
ID int64 `bun:"id,pk,autoincrement"`
Topic string `bun:"topic,notnull"`
Payload []byte `bun:"payload,notnull"`
CreatedAt time.Time `bun:"created_at,notnull"`
PublishedAt *time.Time `bun:"published_at"`
}
Event is one row of the outbox. Payload holds the JSON-encoded event body. PublishedAt is nil while the event is pending and set once the relay has published it.
type Publisher ¶
Publisher publishes a single event's raw payload bytes to its topic.
func NewRedisPublisher ¶
func NewRedisPublisher(client redis.UniversalClient) Publisher
NewRedisPublisher returns a Publisher that publishes the raw payload bytes on the topic channel via client.Publish, without re-encoding. The channel message is therefore exactly the stored JSON, so redis.Subscribe[T] on the consumer side decodes it directly.
type Relay ¶
type Relay struct {
// contains filtered or unexported fields
}
Relay reads pending events from the outbox and publishes them.
func NewRelay ¶
func NewRelay(db *bun.DB, pub Publisher, opts ...RelayOption) *Relay
NewRelay returns a Relay that publishes outbox events from db via pub.
func (*Relay) Process ¶
Process publishes one batch of pending events, oldest first, up to the configured batch size. A Publish error stops the batch and is returned, leaving the failed event and the rest pending (unless WithMaxAttempts dead-letters it, in which case the batch continues). It returns the number of events published and marked. When a Locker is configured and another replica holds the lock, Process returns (0, nil). Process must not be called concurrently on the same Relay.
The events published in a batch are marked with a single UPDATE when the batch ends, on every exit path including a cancelled context. Delivery is at-least-once: if the process dies mid-batch, the events it had already published are published again on the next run, so consumers must be idempotent (see the inbox package).
Example ¶
Write an event in the same transaction as the business data, then let the relay publish it. If the transaction rolls back, no event is ever published.
package main
import (
"context"
"database/sql"
"fmt"
"github.com/uptrace/bun"
"github.com/uptrace/bun/dialect/sqlitedialect"
"github.com/uptrace/bun/driver/sqliteshim"
"github.com/rahmadafandi/fibr/v2/outbox"
)
// recordingPublisher prints each event it would publish.
type recordingPublisher struct{}
func (recordingPublisher) Publish(_ context.Context, topic string, payload []byte) error {
fmt.Printf("publish %s %s\n", topic, payload)
return nil
}
// Write an event in the same transaction as the business data, then let the
// relay publish it. If the transaction rolls back, no event is ever published.
func main() {
sqldb, _ := sql.Open(sqliteshim.ShimName, "file::memory:?cache=shared")
db := bun.NewDB(sqldb, sqlitedialect.New())
defer db.Close()
ctx := context.Background()
_ = outbox.Migrate(ctx, db)
// Business write + event, atomically.
tx, _ := db.BeginTx(ctx, nil)
_ = outbox.Enqueue(ctx, tx, "order.created", map[string]int{"order_id": 42})
_ = tx.Commit()
// Background relay publishes committed events at-least-once.
relay := outbox.NewRelay(db, recordingPublisher{})
n, _ := relay.Process(ctx)
fmt.Println("published:", n)
}
Output: publish order.created {"order_id":42} published: 1
type RelayOption ¶
type RelayOption func(*Relay)
RelayOption configures a Relay.
func WithBatchSize ¶
func WithBatchSize(n int) RelayOption
WithBatchSize sets the maximum number of events published per Process call. The default is 100.
func WithInterval ¶
func WithInterval(d time.Duration) RelayOption
WithInterval sets how often Run calls Process. The default is one second.
func WithLock ¶
WithLock makes Process run each batch under a distributed lock, so only one replica relays at a time. The lock is held for ttl and renewed during the batch whenever a third of ttl has passed, so ttl needs to exceed roughly three times the time to publish a single event, not a whole batch.
func WithLogger ¶
func WithLogger(l *logger.Logger) RelayOption
WithLogger sets the logger Run reports Process errors to. The default is logger.Default().
func WithMaxAttempts ¶
func WithMaxAttempts(n int) RelayOption
WithMaxAttempts moves an event to the outbox_dead table (see DeadEvent) once its publish has failed n times, so one poison event cannot block the queue forever. The count is stored in the outbox_attempts table, so it survives restarts and is shared by every replica; run Migrate to create that table. The default, 0, retries the head event indefinitely.
Source Files