postmark

README

Postmark: a transactional email & notification platform

A production-like example application built on Conveyor: a miniature transactional email and notification service, in the spirit of Postmark, Resend, or Courier. An API accepts requests to notify users, and every piece of downstream work is a Conveyor task. It runs as a real deployment: a Postgres-backed, three-node conveyord cluster on Kubernetes with worker and producer pods, so you can watch it flow, break it, and recover it.

The point is that each Conveyor feature falls out of the product naturally rather than being bolted on. A password reset can't wait behind a million-recipient newsletter, so queues are weighted. A 2FA code must jump the line, so it carries a high priority. A flaky email provider must be retried but a hard bounce must not, so failures map to retry or dead-letter. None of this is contrived: it's how such a platform actually behaves.

The "email provider" is simulated: no real SMTP, no credentials, no network egress. That keeps the example self-contained and, more importantly, makes its failure modes (flakiness, a full outage, hard bounces) controllable, so the retry, circuit-breaker, and dead-letter behaviors are demonstrable on demand rather than left to chance.

What it shows

Conveyor feature	How Postmark uses it
Weighted queues	transactional (10) ≫ default (5) ≫ marketing (1): resets and codes never wait behind a campaign blast.
Push dispatch + concurrency	Workers run 20 slots against a provider that accepts only 8 connections, so back-pressure is real, not theoretical.
Per-task priority	A 2FA code (priority 9) jumps ahead of a welcome email in the same transactional queue.
Delayed / scheduled tasks	The welcome series and trial-ending reminder are enqueued with ProcessIn, sent later, not now.
Cron	A weekly digest is materialized by a cron entry; kill the scheduler's node and it still fires.
Unique tasks	Password resets are keyed user:<id>:password-reset, so a "resend" storm collapses to one mail.
Retries with backoff	The provider fails a fraction of sends transiently; tasks retry and succeed.
Dead-letter / archive	A hard-bounce address returns SkipRetry; the task lands in the archive.
Circuit breaker	When the provider goes fully down, each task type's breaker trips, then recovers.
Pause / resume	The "stop all marketing now" incident button, while transactional keeps flowing.
Per-key concurrency	A campaign tags every send with its tenant, capping each tenant's in-flight sends.
Rate limiting	The marketing queue is capped at 20 sends/second to protect the provider.
Retention	Delivered receipts stay visible for the audit view before they're purged.
Timeouts	A 2FA send is abandoned if it can't complete fast, because a late code is useless.
Crash safety / clustering	Three Postgres-backed nodes; delete a pod under load and nothing is lost.

Run it

One command stands up the whole thing: a Postgres broker, a three-node conveyord cluster, two worker pods, a producer pod, the queue and cron configuration, and the live dashboard:

make postmark-demo

It builds the images, loads them into a throwaway kind cluster, installs the Helm chart, deploys the Postmark workload, configures the queues and the weekly-digest cron, opens the dashboard, and blocks until you press Ctrl-C (which tears the cluster down). Requires docker, kind, kubectl, and helm.

The dashboard opens at http://localhost:8080/, already authenticated: the demo opens it with the API token in the URL, which the dashboard stores client-side and strips from the address bar, so there is nothing to type. Turn on Auto-refresh and watch the three queues fill and drain: a deep, fast transactional queue, a steady default queue, and a slow-draining marketing queue.

Things to try

While the demo runs (in another terminal), the cluster is yours to poke. Every operation is a make target, so there's nothing to copy-paste:

make postmark-stats        # per-queue depth and pause flags
make postmark-pause        # stop all marketing sends (incident button); transactional keeps flowing
make postmark-resume       # resume marketing
make postmark-archived     # the dead-letter queue: hard-bounce campaign mail piling up
make postmark-events       # stream lifecycle events (enqueued, leased, retried, archived...)
make postmark-kill-node    # delete a server pod and watch zero task loss across failover

make postmark-kill-node is the headline: under continuous load it deletes a node, the queues rebalance, in-flight work redelivers elsewhere, and the digest cron survives, and nothing is lost.

The workers cycle a provider outage every 4 minutes for 40 seconds. During each outage every send fails and each task type's circuit breaker trips: the server withholds that type's credit, so the affected queues stop draining and their backlog grows in the dashboard (the trip is also recorded by the conveyor.breaker.open metric). When the provider recovers, the breakers close and the backlog drains. No action needed; just watch.

Tear it down

Pressing Ctrl-C on make postmark-demo already deletes the throwaway cluster. Otherwise:

make postmark-down    # remove just the Postmark workload + its queue/cron config; leave the cluster running
make postmark-clean   # delete the whole throwaway demo cluster

(make postmark-down pauses the weekly-digest cron rather than deleting it: the broker keeps cron entries, and the admin CLI exposes pause/resume, not delete.)

How it maps to the code

The application is broker-agnostic Go; the deployment makes it durable and clustered.

File	Responsibility
postmark.go	The vocabulary: queues, task types, priorities, and the Email payload.
provider.go	The simulated provider: a connection limit, transient flakiness, hard bounces, and an outage switch.
handlers.go	The Mux: every send type, the weekly digest, and a logging middleware that labels each outcome (delivered / archived / retry).
produce.go	The producer: the realistic, transactional-heavy mix of product actions and the per-action Conveyor options that exercise each feature.
cmd/worker	The worker process: serves the three queues and delivers through the provider.
cmd/producer	The producer process: drives a continuous workload.
deploy/	The image and Kubernetes manifests for running it in-cluster.
setup.sh	Configures per-queue concurrency, the marketing rate limit, and the digest cron through the admin CLI.

Deploy it on your own cluster

The make postmark-* targets above drive the throwaway kind cluster and need no hand-typed kubectl. To run the same workload on a cluster you operate, the building blocks are ordinary; adapt these to your namespace, image registry, and API Service:

# 1. Stand up the conveyord cluster itself — the server the workload runs
#    against. This example ships no conveyord manifest; it reuses the production
#    Helm chart, which creates the `conveyor` StatefulSet, the `conveyor` API
#    Service, and the `conveyor-headless` peer Service. See
#    deploy/helm/conveyor/README.md for the Postgres DSN and API-token Secrets
#    it expects.
helm install conveyor deploy/helm/conveyor -n conveyor --create-namespace \
  --set broker.driver=postgres --set replicaCount=3 \
  --set broker.dsnSecret.name=conveyor-broker \
  --set auth.tokensSecret.name=conveyor-auth

# 2. Build the workload image (worker + producer) and push it where your cluster can pull it.
docker build -f examples/postmark/deploy/Dockerfile -t postmark:e2e .

# 3. Deploy the workers and producer (they read the API URL and token from the
#    same conveyor-auth Secret the chart consumes).
kubectl -n conveyor apply -f examples/postmark/deploy/postmark.yaml

# 4. Configure the queues and the weekly-digest cron. Point the CLI at the API,
#    e.g. through a port-forward:
kubectl -n conveyor port-forward svc/conveyor 8080:8080 &
CONVEYOR_ADDR=http://localhost:8080 CONVEYOR_TOKEN=<your-token> \
  ./examples/postmark/setup.sh

Adjust the image reference, imagePullPolicy, and the API Service name in deploy/postmark.yaml for your cluster. To remove the workload, delete the manifest and run teardown.sh (the inverse of setup.sh).

Honest caveats

• The provider is simulated. A real SMTP or API integration would only distract from the queue mechanics, and would make the flaky/outage/bounce demos depend on a third party. The send is local and controllable by design.
• Times are compressed. The welcome follow-up (15s), trial-ending reminder (30s), and weekly digest (every minute) fire fast enough to watch in a demo. A real platform would use days and a Monday-morning cron; the mechanism is identical.

Documentation

Overview

Package postmark is a production-like example application built on Conveyor: a miniature transactional email and notification platform. An API (the Producer) accepts requests to notify users, and every piece of downstream work is a Conveyor task processed by the handlers in this package.

The point of the example is that each Conveyor feature falls out of the product naturally rather than being bolted on. Password resets and 2FA codes ride a heavily weighted transactional queue; campaign blasts ride a lightly weighted marketing queue; a fake email provider with a fixed connection limit and an occasional outage exercises concurrency limits, retries with backoff, dead-lettering, and the per-task-type circuit breaker.

The worker and producer run as separate processes (cmd/worker and cmd/producer) against a Postgres-backed conveyord cluster; the deploy directory runs them on Kubernetes, which is how the example showcases Conveyor's durability and high-availability story. The send is always simulated, never real SMTP, so the example needs no secrets and the flaky and outage behaviors stay controllable for the retry, circuit-breaker, and dead-letter demos.

Index

• Constants
• Variables
• func BounceAddress(userID int) string
• func DeliverableAddress(userID int) string
• func IsHardBounce(addr string) bool
• func NewMux(provider *Provider, logger *slog.Logger) *conveyor.Mux
• func WorkerQueues() map[string]int
• type Email
• type Producer
  • func NewProducer(client *conveyor.Client, logger *slog.Logger) *Producer
  • func (p *Producer) Campaign(ctx context.Context, tenant string) error
  • func (p *Producer) PasswordReset(ctx context.Context, userID int) error
  • func (p *Producer) Receipt(ctx context.Context, userID int) error
  • func (p *Producer) ResendStorm(ctx context.Context, userID int) error
  • func (p *Producer) Run(ctx context.Context, interval time.Duration) error
  • func (p *Producer) TwoFactor(ctx context.Context, userID int) error
  • func (p *Producer) Welcome(ctx context.Context, userID int) error
• type Provider
  • func NewProvider(config ProviderConfig) *Provider
  • func (p *Provider) Down() bool
  • func (p *Provider) InFlight() int
  • func (p *Provider) Send(ctx context.Context, email Email) error
  • func (p *Provider) SetDown(down bool)
  • func (p *Provider) Stats() ProviderStats
• type ProviderConfig
  • func DefaultProviderConfig() ProviderConfig
• type ProviderStats

Constants

const (
	// QueueTransactional carries must-send-now mail: password resets, 2FA
	// codes, and receipts. It is weighted the heaviest.
	QueueTransactional = "transactional"
	// QueueDefault carries ordinary notifications: welcome mail and trial
	// reminders. It is weighted in between.
	QueueDefault = "default"
	// QueueMarketing carries campaign blasts. It is weighted the lightest so a
	// large send drains slowly without starving the other tiers.
	QueueMarketing = "marketing"
)

The platform's queue tiers. A worker declares a relative weight per queue and the server hands out each queue's work in proportion, so a password reset never waits behind a million-recipient newsletter.

const (
	// TaskWelcome is a welcome email sent immediately after signup.
	TaskWelcome = "email:welcome"
	// TaskPasswordReset is a password-reset email, deduplicated per user so a
	// burst of "resend" clicks does not send ten mails.
	TaskPasswordReset = "email:password-reset"
	// TaskTwoFactor is a 2FA code: the most urgent mail, jumping ahead of
	// everything else in the transactional queue and bounded by a tight timeout.
	TaskTwoFactor = "email:2fa"
	// TaskReceipt is a purchase receipt, kept visible after completion for the
	// audit view.
	TaskReceipt = "email:receipt"
	// TaskTrialEnding is a "your trial ends soon" reminder, scheduled for a
	// future time rather than sent now.
	TaskTrialEnding = "email:trial-ending"
	// TaskCampaign is one recipient of a marketing campaign blast.
	TaskCampaign = "email:campaign"
	// TaskDigest builds and sends every user's weekly activity summary; it is
	// materialized by a cron entry, not enqueued by the producer.
	TaskDigest = "digest:weekly"
)

Task types routed to the handlers registered by NewMux.

const (
	// PriorityUrgent puts 2FA codes ahead of everything in their queue.
	PriorityUrgent = 9
	// PriorityHigh puts password resets ahead of ordinary transactional mail.
	PriorityHigh = 7
	// PriorityBulk sinks campaign mail below interactive notifications.
	PriorityBulk = 2
)

Dispatch priorities within a queue (1 lowest, 9 highest; the unset default is 4). A 2FA code outranks a welcome email sharing the transactional queue.

Variables

var (
	// ErrTransient is a temporary failure (the provider's flaky "500"): the send
	// should be retried.
	ErrTransient = errors.New("postmark: provider returned a transient error")
	// ErrHardBounce is a permanent failure: the address is undeliverable and the
	// task must be archived rather than retried.
	ErrHardBounce = errors.New("postmark: recipient address hard-bounced")
	// ErrProviderDown is returned by every call while the provider is in an
	// outage. A task type that sees it repeatedly trips its circuit breaker.
	ErrProviderDown = errors.New("postmark: provider is unavailable")
)

Provider errors. A handler maps each to a Conveyor outcome: a hard bounce is archived (SkipRetry), while a transient failure or a full outage is retried.

Functions

func BounceAddress

func BounceAddress(userID int) string

BounceAddress returns a permanently undeliverable address for a user, used to demonstrate hard bounces landing in the archive.

func DeliverableAddress

func DeliverableAddress(userID int) string

DeliverableAddress returns the ordinary, deliverable address for a user.

func IsHardBounce

func IsHardBounce(addr string) bool

IsHardBounce reports whether addr is permanently undeliverable. A handler that gets a hard bounce wraps the failure in conveyor.SkipRetry so the task is archived immediately instead of retried against an address that never works.

func NewMux

func NewMux(provider *Provider, logger *slog.Logger) *conveyor.Mux

NewMux builds the task router for a Postmark worker: every send task type is served by the same send handler (so each gets its own circuit breaker keyed by type), the weekly digest gets its own, and a logging middleware records the outcome of every task. The provider is the simulated email backend the handlers deliver through.

func WorkerQueues

func WorkerQueues() map[string]int

WorkerQueues returns the queue-to-weight map a Postmark worker serves: transactional far above default, marketing far below, so dispatch favors the mail that must go now. Pass it to conveyor.WithQueues.

Types

type Email

type Email struct {
	// UserID identifies the recipient account; it keys per-user uniqueness.
	UserID int `json:"user_id"`
	// To is the destination address. An address at the hard-bounce domain is
	// permanently undeliverable; see IsHardBounce.
	To string `json:"to"`
	// Subject is the human-readable subject line, used only for log output.
	Subject string `json:"subject"`
	// Tenant identifies the sending customer; it keys per-tenant send
	// concurrency on campaign blasts. Empty for system mail.
	Tenant string `json:"tenant,omitempty"`
}

Email is the payload of every send task: who to mail, on whose behalf, and what about. A single shape serves every task type because, to the platform, they are all "send this user an email".

type Producer

type Producer struct {
	// contains filtered or unexported fields
}

Producer simulates the customer apps hitting the platform's API. Each method enqueues the Conveyor tasks one product action generates; Run drives a believable continuous mix of them.

func NewProducer

func NewProducer(client *conveyor.Client, logger *slog.Logger) *Producer

NewProducer builds a Producer over an enqueueing client.

func (*Producer) Campaign

func (p *Producer) Campaign(ctx context.Context, tenant string) error

Campaign enqueues one marketing blast for a tenant: campaignRecipients recipients on the lightly weighted marketing queue at bulk priority, all sharing the tenant's concurrency key so one campaign cannot stampede the provider. A small fraction of recipients are at a hard-bounce address and will dead-letter.

func (*Producer) PasswordReset

func (p *Producer) PasswordReset(ctx context.Context, userID int) error

PasswordReset enqueues a password-reset email on the transactional queue, deduplicated per user: a second reset for the same user while the first is still pending is rejected as a duplicate, not sent twice.

func (*Producer) Receipt

func (p *Producer) Receipt(ctx context.Context, userID int) error

Receipt enqueues a purchase receipt on the transactional queue and keeps the completed task visible for the audit view via retention.

func (*Producer) ResendStorm

func (p *Producer) ResendStorm(ctx context.Context, userID int) error

ResendStorm simulates an impatient user mashing "resend" on the reset form. Every click after the first collides with the per-user uniqueness key and is dropped, so the user gets one mail instead of resendStormClicks.

func (*Producer) Run

func (p *Producer) Run(ctx context.Context, interval time.Duration) error

Run drives a continuous, transactional-heavy workload, enqueuing one product action every interval until ctx is canceled. It returns ctx.Err on exit.

func (*Producer) TwoFactor

func (p *Producer) TwoFactor(ctx context.Context, userID int) error

TwoFactor enqueues a 2FA code: the highest priority in the transactional queue, so it jumps ahead of welcome and receipt mail, and bounded by a tight timeout because a late code is useless.

func (*Producer) Welcome

func (p *Producer) Welcome(ctx context.Context, userID int) error

Welcome enqueues a signup's mail: a welcome email now, a follow-up tips email scheduled for later, and a trial-ending reminder scheduled further out. The scheduled tasks exercise delayed dispatch.

type Provider

type Provider struct {
	// contains filtered or unexported fields
}

Provider is a fake email/SMTP provider. It accepts a bounded number of concurrent sends, fails a fraction of them transiently, permanently rejects hard-bounce addresses, and can be switched into a total outage. It holds no real connection and sends no real mail, so the example runs offline and its failure modes stay fully controllable.

func NewProvider

func NewProvider(config ProviderConfig) *Provider

NewProvider builds a simulated provider from config, applying defaults for any unset field.

func (*Provider) Down

func (p *Provider) Down() bool

Down reports whether the provider is currently in an outage.

func (*Provider) InFlight

func (p *Provider) InFlight() int

InFlight reports how many sends currently hold a connection. It never exceeds the configured connection limit; surplus callers block in Send until a connection frees up.

func (*Provider) Send

func (p *Provider) Send(ctx context.Context, email Email) error

Send simulates delivering one email, honoring ctx cancellation throughout. It first claims one of the provider's limited connections — blocking if all are in use — then waits out the round-trip latency before deciding the outcome: an outage rejects every call, a hard-bounce address fails permanently, and an otherwise healthy send fails transiently at the configured rate.

func (*Provider) SetDown

func (p *Provider) SetDown(down bool)

SetDown switches the provider into (down=true) or out of (down=false) a total outage. While down, every Send returns ErrProviderDown, which trips the per-task-type circuit breaker; clearing it lets the breaker recover.

func (*Provider) Stats

func (p *Provider) Stats() ProviderStats

Stats snapshots the running send-outcome counters.

type ProviderConfig

type ProviderConfig struct {
	// MaxConnections is the concurrent-send limit; zero selects the default.
	MaxConnections int
	// Latency is the simulated per-send round trip; zero selects the default.
	Latency time.Duration
	// FailureRate is the transient-failure probability in [0,1] while healthy;
	// the zero value never fails transiently. Use DefaultProviderConfig for the
	// realistic default that exercises retries.
	FailureRate float64
}

ProviderConfig configures a simulated provider. The zero value is usable: it applies the package defaults.

func DefaultProviderConfig

func DefaultProviderConfig() ProviderConfig

DefaultProviderConfig returns the realistic provider settings the example's commands run with: a tight connection limit and a fraction of sends failing transiently, so retries with backoff are exercised. Tests build providers directly with a zero FailureRate when they need deterministic delivery.

type ProviderStats

type ProviderStats struct {
	// Sent is the number of successful deliveries.
	Sent int64
	// Transient is the number of transient failures (including outage rejections).
	Transient int64
	// Bounced is the number of permanent hard-bounce rejections.
	Bounced int64
}

ProviderStats is a point-in-time snapshot of a provider's send outcomes.