Build a reusable rate-limiting processor that enforces per-tenant message throughput using the ports.Processor interface.
Your bridge routes messages from an MQTT broker to an SQS queue. Multiple tenants share the same ingress topic, and you need to throttle high-volume tenants so they cannot starve others. The built-in processors (filter, transform, circuit breaker, tenant) do not cover rate limiting, so you build a custom one.
The processor:
x-bridge.tenant-id header.ErrThrottled, causing the runtime to retry with backoff.flowchart LR
MQTT["MQTT Broker\nevents/#"]
Q["SQS Queue\nprocessed"]
MQTT -->|Subscribe| R[Receiver\nmqtt-in]
R --> Route[Route\ningest]
Route --> RL["Processor\nrate-limit"]
RL --> S[Sender\nsqs-out]
S -->|SendMessage| Q
style Route fill:#f96,stroke:#333
style RL fill:#fd6,stroke:#333
Every processor implements two methods:
type Processor interface {
Name() string
Process(ctx context.Context, env *domain.Envelope, next ProcessorFunc) error
}
type ProcessorFunc func(ctx context.Context, env *domain.Envelope) error
The next parameter is the continuation – calling it passes the envelope to the next processor in the chain (or to dispatch if this is the last processor). This is the onion model: each processor wraps the next, so cross-cutting concerns execute in order on the way in and in reverse on the way out.
Key rules:
next(ctx, env) unless you intentionally reject the message.nil to indicate success (the runtime acks the source delivery).domain.BridgeError with the appropriate ErrorClass to drive retry or DLQ behavior.package ratelimit
import (
"context"
"sync"
"time"
"github.com/mariotoffia/gobridge/domain"
"github.com/mariotoffia/gobridge/ports"
)
var _ ports.Processor = (*Processor)(nil)
// Config holds per-tenant rate-limit parameters.
type Config struct {
// MaxRequests is the maximum number of messages per window per tenant.
MaxRequests int
// Window is the sliding window duration.
Window time.Duration
}
// Processor enforces per-tenant rate limiting using a sliding window.
type Processor struct {
name string
config Config
mu sync.Mutex
// tenants maps tenant ID to a slice of request timestamps.
tenants map[string][]time.Time
}
// New creates a rate-limit processor with the given config.
func New(name string, cfg Config) *Processor {
if cfg.MaxRequests <= 0 {
cfg.MaxRequests = 100
}
if cfg.Window <= 0 {
cfg.Window = time.Minute
}
return &Processor{
name: name,
config: cfg,
tenants: make(map[string][]time.Time),
}
}
func (p *Processor) Name() string {
if p.name == "" {
return "rate-limit"
}
return p.name
}
func (p *Processor) Process(
ctx context.Context,
env *domain.Envelope,
next ports.ProcessorFunc,
) error {
tenant, _ := domain.GetHeaderString(env.Headers, domain.HeaderTenantID)
if tenant == "" {
tenant = "_default"
}
if !p.allow(tenant) {
return domain.ErrThrottled.With("tenant", tenant)
}
return next(ctx, env)
}
// allow returns true if the tenant has capacity in the current window.
func (p *Processor) allow(tenant string) bool {
now := time.Now()
cutoff := now.Add(-p.config.Window)
p.mu.Lock()
defer p.mu.Unlock()
// Prune expired timestamps.
ts := p.tenants[tenant]
start := 0
for start < len(ts) && ts[start].Before(cutoff) {
start++
}
ts = ts[start:]
if len(ts) >= p.config.MaxRequests {
p.tenants[tenant] = ts
return false
}
p.tenants[tenant] = append(ts, now)
return true
}
| Decision | Rationale |
|---|---|
Return ErrThrottled |
This is a Transient error, so the runtime retries with backoff rather than sending to the DLQ. |
_default tenant fallback |
Messages without a tenant header still get rate-limited under a shared budget. |
sync.Mutex for state |
The runtime calls Process concurrently (up to MaxInFlight). The mutex protects the sliding window map. |
| No goroutines | The processor is stateless from the runtime’s perspective – it does not own lifecycle. The runtime handles start/stop. |
bridge:
id: rate-limited-bridge
sessions:
- id: mqtt-session
transport: mqtt
# direct_hold relies on the broker redelivering what a crashed process never
# acknowledged; only a persistent (or exclusive) session does that.
session_mode: persistent
options:
session:
broker_url: tcp://localhost:1883
client_id: rate-bridge-01
clean_start: false
session_expiry_interval: 3600
stores:
# A persistent session keeps an exact record of the filters it installed on
# the broker (ADR 0003); seed the baseline once, before the first start:
# gobridge -config bridge.yaml -seed-managed-subscriptions mqtt-session
managed_subscriptions:
type: sqlite
options:
path: /var/lib/gobridge/state/managed-subscriptions.db
# Where a message the route gives up on is kept.
dlq:
type: sqlite
options:
path: /var/lib/gobridge/state/dlq.db
receivers:
- id: mqtt-in
transport: mqtt
session_id: mqtt-session
topics:
- topic: "events/#"
qos: 1
senders:
- id: sqs-out
transport: sqs
options:
queue_url: https://sqs.us-west-1.amazonaws.com/123456789/processed
region: us-west-1
bindings:
- id: to-sqs
sender_id: sqs-out
# Naming the session on the binding is what makes the bridge manage it:
# connect, subscribe, reconcile. A session nobody manages never subscribes.
session_id: mqtt-session
address: processed
routes:
- id: ingest
receiver_id: mqtt-in
delivery_mode: direct_hold
dispatch_mode: single
bindings: [to-sqs]
processors: [rate-limit]
policy:
# Exactly one replica consumes this subscription; a second copy of this
# process would double-deliver. See Scenario 8 for fenced ownership.
allow_unfenced: true
max_in_flight: 50
on_permanent_failure: dlq
package main
import (
"context"
"errors"
"log"
"log/slog"
"os/signal"
"syscall"
"time"
cfgparser "github.com/mariotoffia/gobridge/config/parser"
"github.com/mariotoffia/gobridge/ports"
"example.com/ratelimit"
paho "github.com/mariotoffia/gobridge/adapters/mqtt/transport/paho"
sqs "github.com/mariotoffia/gobridge/adapters/aws/transport/sqs"
"github.com/mariotoffia/gobridge/bridge"
)
func main() {
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
defer stop()
logger := slog.Default()
// Register each linked adapter's config decoder; ParseFile requires a
// non-nil registry.
reg := ports.NewRegistry()
if err := errors.Join(paho.Register(reg), sqs.Register(reg)); err != nil {
log.Fatal(err)
}
cfg, err := cfgparser.ParseFile("bridge.yaml", cfgparser.FormatAuto, reg)
if err != nil {
log.Fatal(err)
}
rl := ratelimit.New("rate-limit", ratelimit.Config{
MaxRequests: 200,
Window: time.Minute,
})
rt, err := bridge.NewBuilder(cfg, bridge.WithLogger(logger)).
RegisterTransportFactory("mqtt", paho.NewFactory(logger)).
RegisterTransportFactory("sqs", sqs.NewFactory(logger)).
RegisterProcessor("rate-limit", rl).
Build(ctx)
if err != nil {
log.Fatal(err)
}
if err := rt.Start(ctx); err != nil {
log.Fatal(err)
}
<-ctx.Done()
_ = rt.Stop(context.Background())
}
The processor name in RegisterProcessor must match the name in routes[].processors.
A processor that adds headers without blocking:
func (p *EnrichProcessor) Process(ctx context.Context, env *domain.Envelope, next ports.ProcessorFunc) error {
domain.SetHeader(env.Headers, "x-bridge.processed-at", time.Now().UTC().Format(time.RFC3339))
domain.SetHeader(env.Headers, "x-bridge.source-region", p.region)
return next(ctx, env)
}
Reject invalid payloads permanently (no retry):
func (p *ValidateProcessor) Process(ctx context.Context, env *domain.Envelope, next ports.ProcessorFunc) error {
if len(env.Payload()) == 0 {
return domain.ErrInvalidPayload.With("reason", "empty payload")
}
if len(env.Payload()) > p.maxSize {
return domain.ErrPayloadTooLarge.With("size", strconv.Itoa(len(env.Payload())))
}
return next(ctx, env)
}
ErrInvalidPayload and ErrPayloadTooLarge are Permanent errors, so the runtime routes them to the DLQ (per policy) without retrying.
Drop messages silently (ack without DLQ) using the filter sentinel:
func (p *DropProcessor) Process(ctx context.Context, env *domain.Envelope, next ports.ProcessorFunc) error {
if env.Subject == "internal/heartbeat" {
return domain.ErrMessageFiltered // ack, no DLQ
}
return next(ctx, env)
}
routes:
- id: full-pipeline
receiver_id: mqtt-in
bindings: [to-sqs]
processors: [validate, enrich, rate-limit]
Processors execute left to right. A failure in any processor short-circuits the chain – the remaining processors and dispatch are skipped.