Release v0.1.0-preview.28: transactional account email

Reviewed source export adds verified TLS mail, encrypted outbox, mailbox verification and password reset protocols. Preserve public ancestry; omit local development history and operational queue. Consumer deployment and inbox delivery proof remain separate.

mailsqlite/README.md

@@ -0,0 +1,47 @@
+<!-- SPDX-License-Identifier: MPL-2.0 -->
+
+# Encrypted transactional outbox
+
+`mailsqlite.Queue` stores a single-recipient `mail.Message` as authenticated
+AES-256-GCM ciphertext. Use a separate 32-byte application-managed secret, kept
+out of source control and logs; back it up separately. It may be wrapped in SQLite
+only if the wrapping key stays outside the database and is backed up separately.
+Identity and expiry
+are bound to the ciphertext. Key-derived HMACs support idempotency without storing
+plaintext message hashes. Losing the key loses pending message contents.
+
+Call `CreateSchema` inside the application's explicit, versioned migration.
+`EnqueueTx` joins a caller-owned transaction, allowing account changes, audit and
+mail intent to commit or roll back together. `Enqueue` is a convenience for a
+standalone transaction. A message ID belongs to exactly one message/expiry, even
+after its payload is cleared. Do not use the queue to authorize recipients.
+
+`ProcessOne` commits a one-minute claim before calling the transport; it never
+holds a database writer lock over SMTP. Transports must honor the supplied
+deadline (at most 30 seconds). A stale worker cannot acknowledge a newer lease.
+Retryable failures back off for 1, 2, 4 and 8 minutes, up to five attempts, only
+while the message is valid. Application workers own scheduling and shutdown.
+
+An incorrect key or corrupt payload never reaches SMTP. Such work retains its
+ciphertext and retries decoding after five minutes without consuming a delivery
+attempt. Restoring the correct key before expiry can recover pending messages.
+This is not transparent key rotation: drain the old queue or provide an explicit
+migration before changing keys.
+
+Payloads expire within 24 hours and are cleared after terminal delivery results
+or expiry. Run `Sweep` periodically even when sending is disabled; each call is
+bounded to 100 records. Pending capacity defaults to 1,000 (maximum 10,000).
+Safe metadata/deduplication tombstones remain; applications own any later bounded
+retention policy and must not reuse purged IDs. Never expose `Recent` publicly.
+
+SMTP acceptance is not inbox delivery. A crash or lost acknowledgement can cause
+a retry after the remote server accepted DATA. Stable Message-ID helps diagnose
+duplicates but cannot make SMTP exactly-once. Do not use this queue for payments
+or another external operation requiring an exactly-once commitment.
+
+Local Go/race/vet tests cover encryption and identity binding, domain rollback,
+idempotency, concurrent capacity/claims, lock-free network waits, stale workers,
+cancelled acknowledgements, retry bounds, safe diagnostics, expiry and wrong-key
+recovery. No real SMTP credential, provider delivery or consumer deployment is
+claimed here. [Account verification/reset protocols](../authmail/README.md) are a
+separate optional layer, not behavior inferred by the queue.

mailsqlite/queue.go

@@ -0,0 +1,334 @@
+// SPDX-License-Identifier: MPL-2.0
+
+// Package mailsqlite provides an encrypted transactional mail outbox. Callers
+// own schema-version journals, authorization, rate limits and connection opening.
+package mailsqlite
+
+import (
+	"context"
+	"crypto/aes"
+	"crypto/cipher"
+	"crypto/hmac"
+	"crypto/rand"
+	"crypto/sha256"
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"time"
+
+	"gamertan.com/web/mail"
+)
+
+var (
+	ErrInvalid  = errors.New("mailsqlite: invalid outbox operation")
+	ErrConflict = errors.New("mailsqlite: message identity or lease conflict")
+	ErrFull     = errors.New("mailsqlite: pending outbox limit reached")
+)
+
+const maxAttempts = 5
+const leaseDuration = time.Minute
+
+type Options struct {
+	// EncryptionKey is a separate 32-byte application-managed secret. Back it up
+	// separately from SQLite. Losing/changing it makes pending messages unreadable.
+	EncryptionKey []byte
+	MaxPending    int
+	Now           func() time.Time
+}
+
+type Queue struct {
+	db         *sql.DB
+	aead       cipher.AEAD
+	digestKey  []byte
+	maxPending int
+	now        func() time.Time
+}
+
+func (*Queue) String() string         { return "mailsqlite.Queue{keys:redacted}" }
+func (queue *Queue) GoString() string { return queue.String() }
+
+func derive(key []byte, label string) []byte {
+	mac := hmac.New(sha256.New, key)
+	mac.Write([]byte(label))
+	return mac.Sum(nil)
+}
+
+func New(db *sql.DB, options Options) (*Queue, error) {
+	if db == nil || len(options.EncryptionKey) != 32 {
+		return nil, ErrInvalid
+	}
+	if options.MaxPending == 0 {
+		options.MaxPending = 1000
+	}
+	if options.MaxPending < 1 || options.MaxPending > 10000 {
+		return nil, ErrInvalid
+	}
+	if options.Now == nil {
+		options.Now = time.Now
+	}
+	block, err := aes.NewCipher(derive(options.EncryptionKey, "gwf.mail.encryption.v1"))
+	if err != nil {
+		return nil, err
+	}
+	aead, err := cipher.NewGCM(block)
+	if err != nil {
+		return nil, err
+	}
+	return &Queue{db: db, aead: aead, digestKey: derive(options.EncryptionKey, "gwf.mail.deduplication.v1"), maxPending: options.MaxPending, now: options.Now}, nil
+}
+
+// CreateSchema is called inside the application's explicit migration, never
+// request startup. A caller-owned transaction also makes schema creation atomic.
+func CreateSchema(ctx context.Context, tx *sql.Tx) error {
+	if tx == nil {
+		return ErrInvalid
+	}
+	for _, statement := range []string{
+		`CREATE TABLE IF NOT EXISTS gwf_mail_outbox (
+ id TEXT PRIMARY KEY,state TEXT NOT NULL CHECK(state IN ('queued','sending','accepted','failed','expired')),
+ payload BLOB,digest BLOB NOT NULL CHECK(length(digest)=32),created_at INTEGER NOT NULL,expires_at INTEGER NOT NULL,
+ next_attempt_at INTEGER NOT NULL,attempts INTEGER NOT NULL DEFAULT 0,lease_hash BLOB,lease_until INTEGER NOT NULL DEFAULT 0,
+ last_stage TEXT NOT NULL DEFAULT '',smtp_code INTEGER NOT NULL DEFAULT 0,updated_at INTEGER NOT NULL,
+ CHECK(expires_at>created_at),CHECK(attempts>=0),CHECK(lease_hash IS NULL OR length(lease_hash)=32))`,
+		`CREATE INDEX IF NOT EXISTS gwf_mail_outbox_ready ON gwf_mail_outbox(state,next_attempt_at,created_at,id)`,
+		`CREATE INDEX IF NOT EXISTS gwf_mail_outbox_expiry ON gwf_mail_outbox(expires_at)`,
+	} {
+		if _, err := tx.ExecContext(ctx, statement); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+type payload struct {
+	Version   int
+	Message   mail.Message
+	ExpiresAt time.Time
+}
+
+// EnqueueTx commits mail intent alongside the caller's domain change and audit.
+// It never commits, sends SMTP or logs plaintext. Reusing an ID is idempotent only
+// for the same message/expiry; a completed ID cannot send a second time.
+func (queue *Queue) EnqueueTx(ctx context.Context, tx *sql.Tx, message mail.Message, expiresAt time.Time) error {
+	if tx == nil || message.Validate() != nil {
+		return ErrInvalid
+	}
+	now := queue.now().UTC()
+	expiresAt = expiresAt.UTC().Truncate(time.Second)
+	message.CreatedAt = message.CreatedAt.UTC().Truncate(time.Second)
+	if !expiresAt.After(now) || expiresAt.Sub(now) > 24*time.Hour || !expiresAt.After(message.CreatedAt) || message.CreatedAt.After(now.Add(time.Minute)) {
+		return ErrInvalid
+	}
+	raw, err := json.Marshal(payload{Version: 1, Message: message, ExpiresAt: expiresAt})
+	if err != nil {
+		return err
+	}
+	mac := hmac.New(sha256.New, queue.digestKey)
+	mac.Write(raw)
+	digest := mac.Sum(nil)
+	nonce := make([]byte, queue.aead.NonceSize())
+	if _, err = rand.Read(nonce); err != nil {
+		return err
+	}
+	sealed := append([]byte{1}, nonce...)
+	sealed = queue.aead.Seal(sealed, nonce, raw, []byte("gwf.mail.v1:"+message.ID))
+	// First statement acquires SQLite's writer lock, avoiding read/write upgrade
+	// races when multiple requests enqueue or compete for the bounded capacity.
+	var inserted string
+	err = tx.QueryRowContext(ctx, `INSERT INTO gwf_mail_outbox(id,state,payload,digest,created_at,expires_at,next_attempt_at,updated_at)
+ SELECT ?,'queued',?,?,?,?,?,? WHERE (SELECT COUNT(*) FROM gwf_mail_outbox WHERE payload IS NOT NULL)<?
+ ON CONFLICT(id) DO NOTHING RETURNING id`, message.ID, sealed, digest, message.CreatedAt.Unix(), expiresAt.Unix(), now.Unix(), now.Unix(), queue.maxPending).Scan(&inserted)
+	if err == nil {
+		return nil
+	}
+	if !errors.Is(err, sql.ErrNoRows) {
+		return err
+	}
+	var existing []byte
+	if err = tx.QueryRowContext(ctx, `SELECT digest FROM gwf_mail_outbox WHERE id=?`, message.ID).Scan(&existing); errors.Is(err, sql.ErrNoRows) {
+		return ErrFull
+	} else if err != nil {
+		return err
+	}
+	if !hmac.Equal(existing, digest) {
+		return ErrConflict
+	}
+	return nil
+}
+
+func (queue *Queue) Enqueue(ctx context.Context, message mail.Message, expiresAt time.Time) error {
+	tx, err := queue.db.BeginTx(ctx, nil)
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+	if err = queue.EnqueueTx(ctx, tx, message, expiresAt); err != nil {
+		return err
+	}
+	return tx.Commit()
+}
+
+// Record is safe operational metadata; it contains no recipient/body/token.
+// Expose it only through the application's current operator authority.
+type Record struct {
+	ID, State, FailureStage             string
+	SMTPCode, Attempts                  int
+	CreatedAt, ExpiresAt, NextAttemptAt time.Time
+}
+
+func (queue *Queue) Recent(ctx context.Context, limit int) ([]Record, error) {
+	if limit < 1 || limit > 100 {
+		return nil, ErrInvalid
+	}
+	rows, err := queue.db.QueryContext(ctx, `SELECT id,state,last_stage,smtp_code,attempts,created_at,expires_at,next_attempt_at FROM gwf_mail_outbox ORDER BY created_at DESC,id DESC LIMIT ?`, limit)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var records []Record
+	for rows.Next() {
+		var record Record
+		var created, expires, next int64
+		if err = rows.Scan(&record.ID, &record.State, &record.FailureStage, &record.SMTPCode, &record.Attempts, &created, &expires, &next); err != nil {
+			return nil, err
+		}
+		record.CreatedAt = time.Unix(created, 0).UTC()
+		record.ExpiresAt = time.Unix(expires, 0).UTC()
+		if next > 0 && record.State == "queued" {
+			record.NextAttemptAt = time.Unix(next, 0).UTC()
+		}
+		records = append(records, record)
+	}
+	return records, rows.Err()
+}
+
+// Sweep clears sensitive payloads for expired/exhausted work in batches of 100.
+// A running, unexpired lease is left alone. Metadata/deduplication tombstones stay.
+func (queue *Queue) Sweep(ctx context.Context) error {
+	now := queue.now().UTC().Unix()
+	_, err := queue.db.ExecContext(ctx, `UPDATE gwf_mail_outbox SET state=CASE WHEN expires_at<=? THEN 'expired' ELSE 'failed' END,
+ payload=NULL,lease_hash=NULL,lease_until=0,next_attempt_at=0,updated_at=?,last_stage=CASE WHEN expires_at<=? THEN 'expired' ELSE 'attempt-limit' END
+ WHERE id IN (SELECT id FROM gwf_mail_outbox WHERE payload IS NOT NULL AND (expires_at<=? OR attempts>=?) AND (state<>'sending' OR lease_until<=?) ORDER BY expires_at,id LIMIT 100)`, now, now, now, now, maxAttempts, now)
+	return err
+}
+
+func (queue *Queue) open(id string, sealed []byte, expires int64) (mail.Message, error) {
+	n := queue.aead.NonceSize()
+	if len(sealed) < 1+n+queue.aead.Overhead() || sealed[0] != 1 {
+		return mail.Message{}, ErrInvalid
+	}
+	raw, err := queue.aead.Open(nil, sealed[1:1+n], sealed[1+n:], []byte("gwf.mail.v1:"+id))
+	if err != nil {
+		return mail.Message{}, ErrInvalid
+	}
+	var body payload
+	if json.Unmarshal(raw, &body) != nil || body.Version != 1 || body.Message.ID != id || body.ExpiresAt.Unix() != expires || body.Message.Validate() != nil {
+		return mail.Message{}, ErrInvalid
+	}
+	return body.Message, nil
+}
+
+// ProcessOne takes one bounded lease, commits it, then contacts SMTP without a
+// database write lock. It records safe status and clears terminal payloads. A
+// crash/uncertain acceptance can duplicate delivery; stable IDs aid diagnosis.
+// false means no ready work. A delivery failure is a Record, not a raw SMTP error.
+func (queue *Queue) ProcessOne(ctx context.Context, transport mail.Transport) (Record, bool, error) {
+	if transport == nil {
+		return Record{}, false, ErrInvalid
+	}
+	if err := queue.Sweep(ctx); err != nil {
+		return Record{}, false, err
+	}
+	now := queue.now().UTC()
+	lease := make([]byte, 32)
+	if _, err := rand.Read(lease); err != nil {
+		return Record{}, false, err
+	}
+	var record Record
+	var sealed []byte
+	var created, expires int64
+	err := queue.db.QueryRowContext(ctx, `UPDATE gwf_mail_outbox SET state='sending',attempts=attempts+1,lease_hash=?,lease_until=?,updated_at=?
+ WHERE id=(SELECT id FROM gwf_mail_outbox WHERE payload IS NOT NULL AND expires_at>? AND next_attempt_at<=? AND attempts<?
+ AND (state='queued' OR (state='sending' AND lease_until<=?)) ORDER BY created_at,id LIMIT 1)
+ RETURNING id,payload,attempts,created_at,expires_at`, lease, now.Add(leaseDuration).Unix(), now.Unix(), now.Unix(), now.Unix(), maxAttempts, now.Unix()).Scan(&record.ID, &sealed, &record.Attempts, &created, &expires)
+	if errors.Is(err, sql.ErrNoRows) {
+		return Record{}, false, nil
+	}
+	if err != nil {
+		return Record{}, false, err
+	}
+	record.CreatedAt = time.Unix(created, 0).UTC()
+	record.ExpiresAt = time.Unix(expires, 0).UTC()
+	message, sendErr := queue.open(record.ID, sealed, expires)
+	payloadInvalid := sendErr != nil
+	if sendErr == nil {
+		// End transmission before the lease expires. Also stop when the message's
+		// own deadline expires; no stale verification link starts after expiry.
+		duration := min(30*time.Second, record.ExpiresAt.Sub(queue.now().UTC()))
+		if duration <= 0 {
+			sendErr = &mail.Error{Stage: "expired"}
+		} else {
+			sendContext, cancel := context.WithTimeout(ctx, duration)
+			sendErr = transport.Send(sendContext, message)
+			cancel()
+		}
+	}
+	finished := queue.now().UTC()
+	record.State = "accepted"
+	if payloadInvalid {
+		// A wrong key is an operational fault, not a delivery attempt. Preserve
+		// ciphertext until expiry so fixing configuration can recover pending
+		// messages. The same bounded delay also covers corrupt/invalid payloads;
+		// never pass them to SMTP or persist the decryption error.
+		record.Attempts--
+		record.State, record.FailureStage = "queued", "payload"
+		record.NextAttemptAt = finished.Add(5 * time.Minute)
+		if !record.ExpiresAt.After(finished) {
+			record.State = "expired"
+		}
+	} else if sendErr != nil {
+		stage, code, retry := "transport", 0, true
+		var failure *mail.Error
+		if errors.As(sendErr, &failure) {
+			stage, code, retry = failure.Stage, failure.Code, failure.Retryable
+		}
+		// Only bounded known stages may enter durable diagnostics, even for a
+		// caller-provided Transport. Never persist arbitrary error strings.
+		switch stage {
+		case "connect", "tls", "greeting", "starttls-required", "authentication", "sender", "recipient", "data", "acceptance", "payload", "expired":
+		default:
+			stage = "transport"
+		}
+		if code < 0 || code > 599 {
+			code = 0
+		}
+		record.FailureStage, record.SMTPCode = stage, code
+		record.State = "failed"
+		record.NextAttemptAt = finished.Add(time.Minute * time.Duration(1<<(record.Attempts-1)))
+		if !record.ExpiresAt.After(finished) || stage == "expired" {
+			record.State = "expired"
+		} else if retry && record.Attempts < maxAttempts && record.NextAttemptAt.Before(record.ExpiresAt) {
+			record.State = "queued"
+		}
+	}
+	var retain []byte
+	var nextAttempt int64
+	if record.State == "queued" {
+		retain = sealed
+		nextAttempt = record.NextAttemptAt.Unix()
+	} else {
+		record.NextAttemptAt = time.Time{}
+	}
+	result, err := queue.db.ExecContext(ctx, `UPDATE gwf_mail_outbox SET state=?,payload=?,last_stage=?,smtp_code=?,attempts=?,next_attempt_at=?,lease_hash=NULL,lease_until=0,updated_at=? WHERE id=? AND state='sending' AND lease_hash=? AND lease_until>?`, record.State, retain, record.FailureStage, record.SMTPCode, record.Attempts, nextAttempt, finished.Unix(), record.ID, lease, finished.Unix())
+	if err != nil {
+		return record, true, err
+	}
+	changed, err := result.RowsAffected()
+	if err != nil {
+		return record, true, err
+	}
+	if changed != 1 {
+		return record, true, ErrConflict
+	}
+	return record, true, nil
+}

mailsqlite/queue_test.go

@@ -0,0 +1,378 @@
+// SPDX-License-Identifier: MPL-2.0
+
+package mailsqlite
+
+import (
+	"bytes"
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+	"path/filepath"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"gamertan.com/web/mail"
+	_ "modernc.org/sqlite"
+)
+
+type testClock struct{ seconds atomic.Int64 }
+
+func (clock *testClock) now() time.Time { return time.Unix(clock.seconds.Load(), 0).UTC() }
+func (clock *testClock) advance(duration time.Duration) {
+	clock.seconds.Add(int64(duration / time.Second))
+}
+
+type transportFunc func(context.Context, mail.Message) error
+
+func (send transportFunc) Send(ctx context.Context, message mail.Message) error {
+	return send(ctx, message)
+}
+
+func fixture(t *testing.T, capacity int) (*Queue, *sql.DB, *testClock) {
+	t.Helper()
+	db, err := sql.Open("sqlite", "file:"+filepath.ToSlash(filepath.Join(t.TempDir(), "mail.sqlite"))+"?_pragma=busy_timeout(5000)&_pragma=foreign_keys(1)&_pragma=journal_mode(WAL)")
+	if err != nil {
+		t.Fatal(err)
+	}
+	db.SetMaxOpenConns(8)
+	t.Cleanup(func() { db.Close() })
+	tx, err := db.Begin()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err = CreateSchema(context.Background(), tx); err != nil {
+		t.Fatal(err)
+	}
+	if err = CreateSchema(context.Background(), tx); err != nil {
+		t.Fatal(err)
+	}
+	if err = tx.Commit(); err != nil {
+		t.Fatal(err)
+	}
+	clock := &testClock{}
+	clock.seconds.Store(time.Date(2026, 9, 11, 6, 0, 0, 0, time.UTC).Unix())
+	queue, err := New(db, Options{EncryptionKey: bytes.Repeat([]byte{17}, 32), MaxPending: capacity, Now: clock.now})
+	if err != nil {
+		t.Fatal(err)
+	}
+	return queue, db, clock
+}
+
+func messageAt(clock *testClock, number int) mail.Message {
+	return mail.Message{ID: fmt.Sprintf("mail_message_%016d", number), To: "recipient@example.test", Subject: "Confirm your email", Text: "Private recovery token: only-in-ciphertext-123", CreatedAt: clock.now()}
+}
+
+func enqueue(t *testing.T, queue *Queue, clock *testClock, number int) mail.Message {
+	t.Helper()
+	message := messageAt(clock, number)
+	if err := queue.Enqueue(context.Background(), message, clock.now().Add(time.Hour)); err != nil {
+		t.Fatal(err)
+	}
+	return message
+}
+
+func storedPayload(t *testing.T, db *sql.DB, id string) []byte {
+	t.Helper()
+	var sealed []byte
+	if err := db.QueryRow(`SELECT payload FROM gwf_mail_outbox WHERE id=?`, id).Scan(&sealed); err != nil {
+		t.Fatal(err)
+	}
+	return sealed
+}
+
+func TestTransactionalEncryptionAndDeduplication(t *testing.T) {
+	queue, db, clock := fixture(t, 1)
+	ctx := context.Background()
+	message := messageAt(clock, 1)
+	expires := clock.now().Add(time.Hour)
+	tx, err := db.Begin()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err = tx.Exec(`CREATE TABLE domain_change(id TEXT)`); err != nil {
+		t.Fatal(err)
+	}
+	if _, err = tx.Exec(`INSERT INTO domain_change VALUES ('changed')`); err != nil {
+		t.Fatal(err)
+	}
+	if err = queue.EnqueueTx(ctx, tx, message, expires); err != nil {
+		t.Fatal(err)
+	}
+	if err = tx.Rollback(); err != nil {
+		t.Fatal(err)
+	}
+	var count int
+	if err = db.QueryRow(`SELECT COUNT(*) FROM gwf_mail_outbox`).Scan(&count); err != nil || count != 0 {
+		t.Fatalf("rollback count=%d err=%v", count, err)
+	}
+	if err = db.QueryRow(`SELECT COUNT(*) FROM sqlite_master WHERE name='domain_change'`).Scan(&count); err != nil || count != 0 {
+		t.Fatalf("domain rollback count=%d err=%v", count, err)
+	}
+	if err = queue.Enqueue(ctx, message, expires); err != nil {
+		t.Fatal(err)
+	}
+	sealed := storedPayload(t, db, message.ID)
+	if len(sealed) == 0 || bytes.Contains(sealed, []byte(message.To)) || bytes.Contains(sealed, []byte(message.Text)) {
+		t.Fatal("payload is not encrypted")
+	}
+	if err = queue.Enqueue(ctx, message, expires); err != nil {
+		t.Fatalf("idempotent at capacity: %v", err)
+	}
+	changed := message
+	changed.Text = "Different content"
+	if err = queue.Enqueue(ctx, changed, expires); !errors.Is(err, ErrConflict) {
+		t.Fatalf("conflicting content: %v", err)
+	}
+	if err = queue.Enqueue(ctx, message, expires.Add(time.Minute)); !errors.Is(err, ErrConflict) {
+		t.Fatalf("conflicting expiry: %v", err)
+	}
+	if err = queue.Enqueue(ctx, messageAt(clock, 2), expires); !errors.Is(err, ErrFull) {
+		t.Fatalf("capacity: %v", err)
+	}
+	var calls int
+	send := transportFunc(func(_ context.Context, got mail.Message) error {
+		calls++
+		if got.ID != message.ID || got.To != message.To || got.Text != message.Text {
+			t.Fatal("message changed")
+		}
+		return nil
+	})
+	record, work, err := queue.ProcessOne(ctx, send)
+	if err != nil || !work || record.State != "accepted" || record.Attempts != 1 || !record.NextAttemptAt.IsZero() {
+		t.Fatalf("accept: %+v work=%v err=%v", record, work, err)
+	}
+	if storedPayload(t, db, message.ID) != nil {
+		t.Fatal("accepted payload retained")
+	}
+	if err = queue.Enqueue(ctx, message, expires); err != nil {
+		t.Fatal(err)
+	}
+	if _, work, err = queue.ProcessOne(ctx, send); err != nil || work || calls != 1 {
+		t.Fatalf("completed ID sent again: calls=%d work=%v err=%v", calls, work, err)
+	}
+	records, err := queue.Recent(ctx, 10)
+	if err != nil || len(records) != 1 || records[0].State != "accepted" || !records[0].NextAttemptAt.IsZero() {
+		t.Fatalf("metadata: %+v %v", records, err)
+	}
+	if err = queue.Enqueue(ctx, messageAt(clock, 2), expires); err != nil {
+		t.Fatalf("released capacity: %v", err)
+	}
+}
+
+func TestConcurrentCapacity(t *testing.T) {
+	queue, _, clock := fixture(t, 1)
+	var group sync.WaitGroup
+	var wins, full atomic.Int32
+	for number := range 8 {
+		group.Go(func() {
+			err := queue.Enqueue(context.Background(), messageAt(clock, number), clock.now().Add(time.Hour))
+			if err == nil {
+				wins.Add(1)
+			} else if errors.Is(err, ErrFull) {
+				full.Add(1)
+			} else {
+				t.Errorf("enqueue: %v", err)
+			}
+		})
+	}
+	group.Wait()
+	if wins.Load() != 1 || full.Load() != 7 {
+		t.Fatalf("wins=%d full=%d", wins.Load(), full.Load())
+	}
+}
+
+func TestRetryBoundsAndSafeDiagnostics(t *testing.T) {
+	queue, db, clock := fixture(t, 0)
+	message := enqueue(t, queue, clock, 1)
+	send := transportFunc(func(context.Context, mail.Message) error {
+		return &mail.Error{Stage: "recipient", Code: 450, Retryable: true}
+	})
+	for attempt := 1; attempt <= maxAttempts; attempt++ {
+		record, work, err := queue.ProcessOne(context.Background(), send)
+		if err != nil || !work || record.Attempts != attempt || record.FailureStage != "recipient" || record.SMTPCode != 450 {
+			t.Fatalf("attempt %d: %+v %v %v", attempt, record, work, err)
+		}
+		if attempt == maxAttempts {
+			if record.State != "failed" || storedPayload(t, db, message.ID) != nil {
+				t.Fatal("exhausted payload not cleared")
+			}
+			break
+		}
+		delay := time.Minute * time.Duration(1<<(attempt-1))
+		if record.State != "queued" || !record.NextAttemptAt.Equal(clock.now().Add(delay)) {
+			t.Fatalf("retry schedule: %+v", record)
+		}
+		if _, work, err = queue.ProcessOne(context.Background(), send); err != nil || work {
+			t.Fatalf("early retry: %v %v", work, err)
+		}
+		clock.advance(delay)
+	}
+	enqueue(t, queue, clock, 2)
+	record, _, err := queue.ProcessOne(context.Background(), transportFunc(func(context.Context, mail.Message) error {
+		return &mail.Error{Stage: "secret-recipient@example.test", Code: 9999, Retryable: false}
+	}))
+	if err != nil || record.State != "failed" || record.FailureStage != "transport" || record.SMTPCode != 0 {
+		t.Fatalf("unsafe diagnostic: %+v %v", record, err)
+	}
+	enqueue(t, queue, clock, 3)
+	record, _, err = queue.ProcessOne(context.Background(), transportFunc(func(context.Context, mail.Message) error { return errors.New("contains private token and address") }))
+	if err != nil || record.State != "queued" || record.FailureStage != "transport" {
+		t.Fatalf("raw transport error: %+v %v", record, err)
+	}
+}
+
+func TestExpiryAndCiphertextRecovery(t *testing.T) {
+	queue, db, clock := fixture(t, 0)
+	message := enqueue(t, queue, clock, 1)
+	sealed := storedPayload(t, db, message.ID)
+	wrong, err := New(db, Options{EncryptionKey: bytes.Repeat([]byte{18}, 32), Now: clock.now})
+	if err != nil {
+		t.Fatal(err)
+	}
+	var calls int
+	send := transportFunc(func(context.Context, mail.Message) error { calls++; return nil })
+	record, work, err := wrong.ProcessOne(context.Background(), send)
+	if err != nil || !work || record.State != "queued" || record.FailureStage != "payload" || record.Attempts != 0 || calls != 0 {
+		t.Fatalf("wrong key: %+v %v %v calls=%d", record, work, err, calls)
+	}
+	if !bytes.Equal(sealed, storedPayload(t, db, message.ID)) {
+		t.Fatal("key error destroyed pending ciphertext")
+	}
+	clock.advance(5 * time.Minute)
+	record, work, err = queue.ProcessOne(context.Background(), send)
+	if err != nil || !work || record.State != "accepted" || record.Attempts != 1 || calls != 1 {
+		t.Fatalf("key recovery: %+v %v %v calls=%d", record, work, err, calls)
+	}
+	message = enqueue(t, queue, clock, 2)
+	sealed = storedPayload(t, db, message.ID)
+	sealed[len(sealed)-1] ^= 1
+	if _, err = db.Exec(`UPDATE gwf_mail_outbox SET payload=? WHERE id=?`, sealed, message.ID); err != nil {
+		t.Fatal(err)
+	}
+	record, _, err = queue.ProcessOne(context.Background(), send)
+	if err != nil || record.FailureStage != "payload" || calls != 1 {
+		t.Fatalf("tamper sent: %+v %v calls=%d", record, err, calls)
+	}
+	clock.advance(time.Hour)
+	if _, work, err = queue.ProcessOne(context.Background(), send); err != nil || work || calls != 1 {
+		t.Fatalf("expired sent: work=%v err=%v calls=%d", work, err, calls)
+	}
+	if storedPayload(t, db, message.ID) != nil {
+		t.Fatal("expired corrupt payload retained")
+	}
+	if _, err = queue.open(message.ID, sealed[:3], clock.now().Unix()); !errors.Is(err, ErrInvalid) {
+		t.Fatal("short ciphertext accepted")
+	}
+	valid := enqueue(t, queue, clock, 3)
+	if _, err = queue.open("another_message_identity", storedPayload(t, db, valid.ID), clock.now().Add(time.Hour).Unix()); !errors.Is(err, ErrInvalid) {
+		t.Fatal("AAD identity not bound")
+	}
+	if _, err = queue.open(valid.ID, storedPayload(t, db, valid.ID), clock.now().Add(2*time.Hour).Unix()); !errors.Is(err, ErrInvalid) {
+		t.Fatal("expiry not bound")
+	}
+}
+
+func TestLeaseIsolationAndLateAcknowledgement(t *testing.T) {
+	queue, db, clock := fixture(t, 0)
+	message := enqueue(t, queue, clock, 1)
+	entered, release := make(chan struct{}), make(chan struct{})
+	type result struct {
+		record Record
+		work   bool
+		err    error
+	}
+	done := make(chan result, 1)
+	go func() {
+		record, work, err := queue.ProcessOne(context.Background(), transportFunc(func(ctx context.Context, _ mail.Message) error {
+			close(entered)
+			select {
+			case <-release:
+				return nil
+			case <-ctx.Done():
+				return ctx.Err()
+			}
+		}))
+		done <- result{record, work, err}
+	}()
+	<-entered
+	if _, work, err := queue.ProcessOne(context.Background(), transportFunc(func(context.Context, mail.Message) error { t.Error("duplicate active lease"); return nil })); err != nil || work {
+		t.Fatalf("leased work claimed: %v %v", work, err)
+	}
+	// SMTP cannot hold the SQLite writer lock while waiting on the network.
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	if _, err := db.ExecContext(ctx, `CREATE TABLE independent_domain_write(id TEXT)`); err != nil {
+		t.Fatal(err)
+	}
+	clock.advance(leaseDuration + time.Second)
+	record, work, err := queue.ProcessOne(context.Background(), transportFunc(func(context.Context, mail.Message) error { return nil }))
+	if err != nil || !work || record.State != "accepted" || record.Attempts != 2 {
+		t.Fatalf("takeover: %+v %v %v", record, work, err)
+	}
+	close(release)
+	previous := <-done
+	if !previous.work || !errors.Is(previous.err, ErrConflict) {
+		t.Fatalf("late acknowledgement: %+v", previous)
+	}
+	if storedPayload(t, db, message.ID) != nil {
+		t.Fatal("late worker restored payload")
+	}
+}
+
+func TestCancellationLeavesRecoverableLease(t *testing.T) {
+	queue, db, clock := fixture(t, 0)
+	message := enqueue(t, queue, clock, 1)
+	ctx, cancel := context.WithCancel(context.Background())
+	_, work, err := queue.ProcessOne(ctx, transportFunc(func(context.Context, mail.Message) error { cancel(); return nil }))
+	if !work || !errors.Is(err, context.Canceled) {
+		t.Fatalf("cancelled acknowledgement: %v %v", work, err)
+	}
+	if storedPayload(t, db, message.ID) == nil {
+		t.Fatal("uncertain delivery destroyed recovery payload")
+	}
+	clock.advance(leaseDuration + time.Second)
+	record, work, err := queue.ProcessOne(context.Background(), transportFunc(func(context.Context, mail.Message) error { return nil }))
+	if err != nil || !work || record.State != "accepted" || record.Attempts != 2 {
+		t.Fatalf("recovery: %+v %v %v", record, work, err)
+	}
+}
+
+func TestValidation(t *testing.T) {
+	queue, db, clock := fixture(t, 0)
+	ctx := context.Background()
+	for _, options := range []Options{{}, {EncryptionKey: make([]byte, 31)}, {EncryptionKey: make([]byte, 32), MaxPending: -1}, {EncryptionKey: make([]byte, 32), MaxPending: 10001}} {
+		if _, err := New(db, options); !errors.Is(err, ErrInvalid) {
+			t.Fatalf("invalid options: %v", err)
+		}
+	}
+	if _, err := New(nil, Options{EncryptionKey: make([]byte, 32)}); !errors.Is(err, ErrInvalid) {
+		t.Fatal("nil DB accepted")
+	}
+	if err := CreateSchema(ctx, nil); !errors.Is(err, ErrInvalid) {
+		t.Fatal("nil migration accepted")
+	}
+	if err := queue.EnqueueTx(ctx, nil, messageAt(clock, 1), clock.now().Add(time.Hour)); !errors.Is(err, ErrInvalid) {
+		t.Fatal("nil tx accepted")
+	}
+	for _, expiry := range []time.Time{clock.now(), clock.now().Add(-time.Minute), clock.now().Add(25 * time.Hour)} {
+		if err := queue.Enqueue(ctx, messageAt(clock, 1), expiry); !errors.Is(err, ErrInvalid) {
+			t.Fatalf("invalid expiry: %v", err)
+		}
+	}
+	message := messageAt(clock, 1)
+	message.CreatedAt = clock.now().Add(2 * time.Minute)
+	if err := queue.Enqueue(ctx, message, clock.now().Add(time.Hour)); !errors.Is(err, ErrInvalid) {
+		t.Fatal("future message accepted")
+	}
+	for _, limit := range []int{0, 101} {
+		if _, err := queue.Recent(ctx, limit); !errors.Is(err, ErrInvalid) {
+			t.Fatal("invalid metadata limit")
+		}
+	}
+	if _, _, err := queue.ProcessOne(ctx, nil); !errors.Is(err, ErrInvalid) {
+		t.Fatal("nil transport accepted")
+	}
+}