more checkout

cmd/order/amqp.go

@@ -13,31 +13,59 @@ import (
 	amqp "github.com/rabbitmq/amqp091-go"
 )
 
-// rabbitPublisher implements order.Publisher for the amqp_emit flow hook.
+// rabbitPublisher is the broker-delivery side used by the outbox relay. It dials
+// lazily and reconnects on failure, so the relay can be started even when the
+// broker is down at boot: messages accumulate durably in the outbox and drain
+// once the broker is reachable.
 type rabbitPublisher struct {
+	url  string
+	mu   sync.Mutex
 	conn *amqp.Connection
 	ch   *amqp.Channel
-	mu   sync.Mutex
 }
 
-func newRabbitPublisher(url string) (*rabbitPublisher, error) {
-	conn, err := amqp.Dial(url)
+func newRabbitPublisher(url string) *rabbitPublisher {
+	return &rabbitPublisher{url: url}
+}
+
+func (p *rabbitPublisher) connect() error {
+	conn, err := amqp.Dial(p.url)
 	if err != nil {
-		return nil, err
+		return err
 	}
 	ch, err := conn.Channel()
 	if err != nil {
 		_ = conn.Close()
-		return nil, err
+		return err
 	}
-	return &rabbitPublisher{conn: conn, ch: ch}, nil
+	p.conn, p.ch = conn, ch
+	return nil
+}
+
+func (p *rabbitPublisher) reset() {
+	if p.ch != nil {
+		_ = p.ch.Close()
+	}
+	if p.conn != nil {
+		_ = p.conn.Close()
+	}
+	p.ch, p.conn = nil, nil
 }
 
 func (p *rabbitPublisher) Publish(exchange, routingKey string, body []byte) error {
 	p.mu.Lock()
 	defer p.mu.Unlock()
-	return p.ch.PublishWithContext(context.Background(), exchange, routingKey, false, false,
+	if p.ch == nil {
+		if err := p.connect(); err != nil {
+			return err
+		}
+	}
+	err := p.ch.PublishWithContext(context.Background(), exchange, routingKey, false, false,
 		amqp.Publishing{ContentType: "application/json", Body: body})
+	if err != nil {
+		p.reset() // force a reconnect on the next attempt
+	}
+	return err
 }
 
 // This ingester lets the order service supersede the legacy go-order-manager:

cmd/order/handlers_checkout.go

@@ -4,7 +4,6 @@ import (
 	"encoding/json"
 	"fmt"
 	"net/http"
-	"sync"
 	"time"
 
 	"git.k6n.net/mats/go-cart-actor/pkg/flow"
@@ -38,14 +37,6 @@ type fromCheckoutReq struct {
 	} `json:"payment"`
 }
 
-// idempotencyStore guards against duplicate order creation for the same
-// idempotency key. In-memory for step 1; a durable store (Redis, bolt) would
-// survive restarts. The map is keyed by idempotencyKey → orderId.
-var (
-	idempotencyMu   sync.Mutex
-	idempotencyKeys = map[string]uint64{}
-)
-
 // handleFromCheckout creates an event-sourced order from a settled checkout.
 // It runs the place-and-pay flow with a passthrough provider that records the
 // already-completed authorization and capture, bypassing any external payment
@@ -65,10 +56,13 @@ func (s *server) handleFromCheckout(w http.ResponseWriter, r *http.Request) {
 	}
 
 	// ── Idempotency check ─────────────────────────────────────────────
+	// Lock the key for the whole check→create→record sequence so a concurrent
+	// retry with the same key can't race past the check and create a second
+	// order. The store is durable, so this also holds across a restart.
 	if req.IdempotencyKey != "" {
-		idempotencyMu.Lock()
-		if existingID, ok := idempotencyKeys[req.IdempotencyKey]; ok {
-			idempotencyMu.Unlock()
+		unlock := s.idem.Lock(req.IdempotencyKey)
+		defer unlock()
+		if existingID, ok := s.idem.Get(req.IdempotencyKey); ok {
 			// The order already exists — return it.
 			g, err := s.applier.Get(r.Context(), existingID)
 			if err != nil {
@@ -76,13 +70,12 @@ func (s *server) handleFromCheckout(w http.ResponseWriter, r *http.Request) {
 				return
 			}
 			writeJSON(w, http.StatusConflict, map[string]any{
-				"orderId": order.OrderId(existingID).String(),
-				"order":   g,
+				"orderId":  order.OrderId(existingID).String(),
+				"order":    g,
 				"existing": true,
 			})
 			return
 		}
-		idempotencyMu.Unlock()
 	}
 
 	// ── Create the order grain ────────────────────────────────────────
@@ -141,11 +134,14 @@ func (s *server) handleFromCheckout(w http.ResponseWriter, r *http.Request) {
 		s.logger.Warn("from-checkout flow failed", "orderId", id.String(), "err", runErr)
 	}
 
-	// Record the idempotency key so retries are safe.
+	// Record the idempotency key durably so retries (incl. after a restart) are
+	// safe. We record even on a failed flow: the failed order grain exists, and a
+	// blind retry with the same key should return it rather than place a second
+	// order — the caller is told to retry with a fresh key (402 contract).
 	if req.IdempotencyKey != "" {
-		idempotencyMu.Lock()
-		idempotencyKeys[req.IdempotencyKey] = ordID
-		idempotencyMu.Unlock()
+		if err := s.idem.Put(req.IdempotencyKey, ordID); err != nil {
+			s.logger.Error("idempotency record failed", "key", req.IdempotencyKey, "orderId", id.String(), "err", err)
+		}
 	}
 
 	writeJSON(w, status, map[string]any{

cmd/order/main.go

@@ -21,7 +21,9 @@ import (
 
 	"git.k6n.net/mats/go-cart-actor/pkg/actor"
 	"git.k6n.net/mats/go-cart-actor/pkg/flow"
+	"git.k6n.net/mats/go-cart-actor/pkg/idempotency"
 	"git.k6n.net/mats/go-cart-actor/pkg/order"
+	"git.k6n.net/mats/go-cart-actor/pkg/outbox"
 	messages "git.k6n.net/mats/go-cart-actor/proto/order"
 
 	"google.golang.org/protobuf/encoding/protojson"
@@ -39,6 +41,7 @@ type server struct {
 	provider    order.PaymentProvider
 	defaultFlow string
 	dataDir     string
+	idem        *idempotency.Store
 	logger      *slog.Logger
 }
 
@@ -94,10 +97,15 @@ func main() {
 	amqpURL := os.Getenv("AMQP_URL")
 	var emitPub order.Publisher
 	if amqpURL != "" {
-		if pub, err := newRabbitPublisher(amqpURL); err != nil {
-			logger.Warn("amqp_emit hook: publisher unavailable", "err", err)
+		// The amqp_emit hook writes to a durable outbox rather than the broker
+		// directly; a relay goroutine drains it to RabbitMQ (reconnecting), so an
+		// event survives a broker outage or a crash after the state change.
+		box, err := outbox.Open(filepath.Join(dataDir, "outbox.log"))
+		if err != nil {
+			logger.Warn("outbox disabled: open failed", "err", err)
 		} else {
-			emitPub = pub
+			go box.Run(context.Background(), newRabbitPublisher(amqpURL), 5*time.Second, logger)
+			emitPub = box
 		}
 	}
 	freg := flow.NewRegistry()
@@ -115,10 +123,18 @@ func main() {
 		log.Fatalf("load flows: %v", err)
 	}
 
+	// Durable idempotency store for order creation — survives a restart, so a
+	// client retry after a crash returns the existing order instead of creating
+	// a duplicate.
+	idem, err := idempotency.Open(filepath.Join(dataDir, "idempotency.log"))
+	if err != nil {
+		log.Fatalf("open idempotency store: %v", err)
+	}
+
 	s := &server{
 		pool: pool, applier: applier, storage: storage, reg: reg,
 		engine: engine, freg: freg, flows: flows, provider: provider,
-		defaultFlow: def.Name, dataDir: dataDir, logger: logger,
+		defaultFlow: def.Name, dataDir: dataDir, idem: idem, logger: logger,
 	}
 
 	mux := http.NewServeMux()

pkg/idempotency/store.go

@@ -0,0 +1,122 @@
+// Package idempotency provides a small durable key→id store so a cross-service
+// mutation can be retried safely. It is the durable replacement for the
+// in-memory map that guarded order creation in cmd/order: an in-memory map is
+// lost on restart, so a crash followed by a client retry created a duplicate
+// order. This store is backed by an append-only JSONL file and survives a
+// restart — a recorded key is durable the moment Put returns.
+//
+// It targets the single-node order service: appends are fsync'd under a mutex,
+// and a per-key lock lets a caller make a check→create→record sequence atomic
+// against a concurrent retry carrying the same key.
+package idempotency
+
+import (
+	"bufio"
+	"encoding/json"
+	"errors"
+	"os"
+	"sync"
+)
+
+// Store maps an idempotency key to a result id (e.g. an order id), cached in
+// memory and persisted to an append-only file.
+type Store struct {
+	mu sync.Mutex
+	m  map[string]uint64
+	f  *os.File
+
+	klMu     sync.Mutex
+	keyLocks map[string]*sync.Mutex
+}
+
+type record struct {
+	Key string `json:"k"`
+	ID  uint64 `json:"id"`
+}
+
+// Open loads the store at path (creating the file if absent) and returns a
+// Store ready for use. Corrupt trailing lines (e.g. from a crash mid-write) are
+// skipped rather than failing the load. The caller owns Close.
+func Open(path string) (*Store, error) {
+	m := map[string]uint64{}
+	if f, err := os.Open(path); err == nil {
+		sc := bufio.NewScanner(f)
+		sc.Buffer(make([]byte, 0, 64*1024), 1024*1024)
+		for sc.Scan() {
+			line := sc.Bytes()
+			if len(line) == 0 {
+				continue
+			}
+			var rec record
+			if err := json.Unmarshal(line, &rec); err != nil || rec.Key == "" {
+				continue // skip a torn/corrupt line
+			}
+			m[rec.Key] = rec.ID
+		}
+		_ = f.Close()
+	} else if !errors.Is(err, os.ErrNotExist) {
+		return nil, err
+	}
+
+	f, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644)
+	if err != nil {
+		return nil, err
+	}
+	return &Store{m: m, f: f, keyLocks: map[string]*sync.Mutex{}}, nil
+}
+
+// Get returns the recorded id for key and whether it was present.
+func (s *Store) Get(key string) (uint64, bool) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	id, ok := s.m[key]
+	return id, ok
+}
+
+// Put records key→id durably. A repeated key keeps the first recorded value
+// (first writer wins) and performs no disk write, matching idempotent
+// semantics. The append is fsync'd before Put returns.
+func (s *Store) Put(key string, id uint64) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if _, ok := s.m[key]; ok {
+		return nil
+	}
+	line, err := json.Marshal(record{Key: key, ID: id})
+	if err != nil {
+		return err
+	}
+	if _, err := s.f.Write(append(line, '\n')); err != nil {
+		return err
+	}
+	if err := s.f.Sync(); err != nil {
+		return err
+	}
+	s.m[key] = id
+	return nil
+}
+
+// Lock serialises all work for a single key so a check→create→record sequence
+// is atomic against a concurrent retry carrying the same key. It returns the
+// unlock function. An empty key is not locked (returns a no-op).
+func (s *Store) Lock(key string) func() {
+	if key == "" {
+		return func() {}
+	}
+	s.klMu.Lock()
+	mu, ok := s.keyLocks[key]
+	if !ok {
+		mu = &sync.Mutex{}
+		s.keyLocks[key] = mu
+	}
+	s.klMu.Unlock()
+	mu.Lock()
+	return mu.Unlock
+}
+
+// Close closes the underlying file.
+func (s *Store) Close() error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.f.Close()
+}

pkg/idempotency/store_test.go

@@ -0,0 +1,106 @@
+package idempotency
+
+import (
+	"os"
+	"path/filepath"
+	"sync"
+	"testing"
+)
+
+func TestPutGet(t *testing.T) {
+	s, err := Open(filepath.Join(t.TempDir(), "idem.log"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer s.Close()
+
+	if _, ok := s.Get("a"); ok {
+		t.Fatal("unexpected hit on empty store")
+	}
+	if err := s.Put("a", 42); err != nil {
+		t.Fatal(err)
+	}
+	if id, ok := s.Get("a"); !ok || id != 42 {
+		t.Fatalf("Get(a) = %d,%v; want 42,true", id, ok)
+	}
+}
+
+func TestFirstWriterWins(t *testing.T) {
+	s, _ := Open(filepath.Join(t.TempDir(), "idem.log"))
+	defer s.Close()
+	_ = s.Put("a", 1)
+	_ = s.Put("a", 2) // ignored
+	if id, _ := s.Get("a"); id != 1 {
+		t.Fatalf("got %d, want first value 1", id)
+	}
+}
+
+func TestDurableAcrossReopen(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "idem.log")
+	s, _ := Open(path)
+	_ = s.Put("checkout-x", 1001)
+	_ = s.Put("checkout-y", 1002)
+	_ = s.Close()
+
+	// A fresh process (the restart case the in-memory map failed) must still see
+	// the recorded keys — this is the duplicate-order bug the durable store fixes.
+	s2, err := Open(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer s2.Close()
+	if id, ok := s2.Get("checkout-x"); !ok || id != 1001 {
+		t.Fatalf("after reopen Get(checkout-x) = %d,%v; want 1001,true", id, ok)
+	}
+	if id, ok := s2.Get("checkout-y"); !ok || id != 1002 {
+		t.Fatalf("after reopen Get(checkout-y) = %d,%v; want 1002,true", id, ok)
+	}
+}
+
+func TestCorruptTrailingLineSkipped(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "idem.log")
+	s, _ := Open(path)
+	_ = s.Put("good", 7)
+	_ = s.Close()
+
+	// Simulate a torn write from a crash mid-append.
+	f, _ := os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0o644)
+	_, _ = f.WriteString(`{"k":"bad","id":`)
+	_ = f.Close()
+
+	s2, err := Open(path)
+	if err != nil {
+		t.Fatalf("open with corrupt tail: %v", err)
+	}
+	defer s2.Close()
+	if id, ok := s2.Get("good"); !ok || id != 7 {
+		t.Fatalf("good key lost after corrupt tail: %d,%v", id, ok)
+	}
+	if _, ok := s2.Get("bad"); ok {
+		t.Fatal("corrupt line should not have loaded")
+	}
+}
+
+func TestLockSerialisesKey(t *testing.T) {
+	s, _ := Open(filepath.Join(t.TempDir(), "idem.log"))
+	defer s.Close()
+
+	// Two goroutines contend on the same key; the lock must serialise the
+	// check→create→record so exactly one writes and both agree on the value.
+	var wg sync.WaitGroup
+	for i := 0; i < 2; i++ {
+		wg.Add(1)
+		go func(v uint64) {
+			defer wg.Done()
+			unlock := s.Lock("k")
+			defer unlock()
+			if _, ok := s.Get("k"); !ok {
+				_ = s.Put("k", v)
+			}
+		}(uint64(i + 1))
+	}
+	wg.Wait()
+	if _, ok := s.Get("k"); !ok {
+		t.Fatal("expected a value recorded under contention")
+	}
+}

pkg/outbox/outbox.go

@@ -0,0 +1,255 @@
+// Package outbox implements a durable transactional-outbox relay for AMQP
+// publishing. Today the order service's amqp_emit hook publishes best-effort: a
+// broker outage or a crash after a state change but before the publish loses
+// the event. The outbox closes that gap.
+//
+// Producers call Publish, which appends the message to an append-only JSONL file
+// and fsyncs before returning — so the intent to emit is durable the moment
+// Publish returns, independent of broker availability. A single relay goroutine
+// (Run) drains undelivered entries to the broker and records an ack line on
+// success, retrying on failure. Delivery is therefore at-least-once: a crash
+// between the broker publish and the ack line replays that message on restart,
+// which downstream consumers must tolerate (the commerce events are keyed and
+// already designed to be idempotent).
+//
+// This targets the single-node order service. The file is compacted on Open
+// (acked entries are dropped), which bounds growth across restarts.
+package outbox
+
+import (
+	"bufio"
+	"context"
+	"encoding/json"
+	"errors"
+	"log/slog"
+	"os"
+	"sort"
+	"sync"
+	"time"
+)
+
+// Publisher delivers a message to the broker. The relay retries on error, so an
+// implementation should surface a transient failure (broker down) as an error
+// rather than dropping the message.
+type Publisher interface {
+	Publish(exchange, routingKey string, body []byte) error
+}
+
+// line is one record in the log: either a pending message (Seq>0 with payload)
+// or an ack marker (Ack>0). Body round-trips as base64 via encoding/json.
+type line struct {
+	Seq        uint64 `json:"seq,omitempty"`
+	Exchange   string `json:"exchange,omitempty"`
+	RoutingKey string `json:"rk,omitempty"`
+	Body       []byte `json:"body,omitempty"`
+	Ack        uint64 `json:"ack,omitempty"`
+}
+
+type entry struct {
+	seq        uint64
+	exchange   string
+	routingKey string
+	body       []byte
+}
+
+// Outbox is a durable, file-backed message buffer with an at-least-once relay.
+type Outbox struct {
+	mu      sync.Mutex
+	f       *os.File
+	path    string
+	seq     uint64
+	pending map[uint64]entry
+	wake    chan struct{}
+}
+
+// Open loads the outbox at path (creating it if absent), drops already-acked
+// entries, and rewrites the file with only the still-pending ones (compaction).
+// Returns an Outbox whose pending set is ready for the relay to drain.
+func Open(path string) (*Outbox, error) {
+	pending := map[uint64]entry{}
+	acked := map[uint64]struct{}{}
+	var maxSeq uint64
+
+	if f, err := os.Open(path); err == nil {
+		sc := bufio.NewScanner(f)
+		sc.Buffer(make([]byte, 0, 64*1024), 8*1024*1024)
+		for sc.Scan() {
+			b := sc.Bytes()
+			if len(b) == 0 {
+				continue
+			}
+			var l line
+			if err := json.Unmarshal(b, &l); err != nil {
+				continue // skip a torn/corrupt line
+			}
+			switch {
+			case l.Ack > 0:
+				acked[l.Ack] = struct{}{}
+			case l.Seq > 0:
+				pending[l.Seq] = entry{seq: l.Seq, exchange: l.Exchange, routingKey: l.RoutingKey, body: l.Body}
+				if l.Seq > maxSeq {
+					maxSeq = l.Seq
+				}
+			}
+		}
+		_ = f.Close()
+	} else if !errors.Is(err, os.ErrNotExist) {
+		return nil, err
+	}
+	for seq := range acked {
+		delete(pending, seq)
+	}
+
+	ob := &Outbox{path: path, seq: maxSeq, pending: pending, wake: make(chan struct{}, 1)}
+	if err := ob.compact(); err != nil {
+		return nil, err
+	}
+	return ob, nil
+}
+
+// compact rewrites the file with only the current pending entries, then reopens
+// it for appends. Called on Open so the log can't grow without bound across
+// restarts. Must be called with no concurrent writers (Open, before Run).
+func (o *Outbox) compact() error {
+	tmp := o.path + ".tmp"
+	f, err := os.OpenFile(tmp, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o644)
+	if err != nil {
+		return err
+	}
+	seqs := make([]uint64, 0, len(o.pending))
+	for seq := range o.pending {
+		seqs = append(seqs, seq)
+	}
+	sort.Slice(seqs, func(i, j int) bool { return seqs[i] < seqs[j] })
+	w := bufio.NewWriter(f)
+	for _, seq := range seqs {
+		e := o.pending[seq]
+		b, _ := json.Marshal(line{Seq: e.seq, Exchange: e.exchange, RoutingKey: e.routingKey, Body: e.body})
+		if _, err := w.Write(append(b, '\n')); err != nil {
+			_ = f.Close()
+			return err
+		}
+	}
+	if err := w.Flush(); err != nil {
+		_ = f.Close()
+		return err
+	}
+	if err := f.Sync(); err != nil {
+		_ = f.Close()
+		return err
+	}
+	if err := f.Close(); err != nil {
+		return err
+	}
+	if err := os.Rename(tmp, o.path); err != nil {
+		return err
+	}
+	o.f, err = os.OpenFile(o.path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644)
+	return err
+}
+
+// Publish durably records a message and returns once it is fsync'd. It satisfies
+// the order.Publisher seam, so the amqp_emit hook records to the outbox instead
+// of publishing to the broker directly. The relay delivers it asynchronously.
+func (o *Outbox) Publish(exchange, routingKey string, body []byte) error {
+	o.mu.Lock()
+	o.seq++
+	seq := o.seq
+	b, err := json.Marshal(line{Seq: seq, Exchange: exchange, RoutingKey: routingKey, Body: body})
+	if err != nil {
+		o.mu.Unlock()
+		return err
+	}
+	if _, err := o.f.Write(append(b, '\n')); err != nil {
+		o.mu.Unlock()
+		return err
+	}
+	if err := o.f.Sync(); err != nil {
+		o.mu.Unlock()
+		return err
+	}
+	o.pending[seq] = entry{seq: seq, exchange: exchange, routingKey: routingKey, body: body}
+	o.mu.Unlock()
+
+	select {
+	case o.wake <- struct{}{}: // nudge the relay
+	default:
+	}
+	return nil
+}
+
+// PendingCount reports how many messages are awaiting delivery (for tests/metrics).
+func (o *Outbox) PendingCount() int {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	return len(o.pending)
+}
+
+// Run drains the outbox to pub, retrying failed rounds every retry interval and
+// waking immediately when a new message arrives. It blocks until ctx is done.
+func (o *Outbox) Run(ctx context.Context, pub Publisher, retry time.Duration, logger *slog.Logger) {
+	t := time.NewTicker(retry)
+	defer t.Stop()
+	for {
+		o.drain(pub, logger)
+		select {
+		case <-ctx.Done():
+			return
+		case <-t.C:
+		case <-o.wake:
+		}
+	}
+}
+
+// drain attempts to deliver every pending message in seq order. It stops the
+// round at the first delivery error (likely the broker is down) so messages keep
+// their order and the next tick retries.
+func (o *Outbox) drain(pub Publisher, logger *slog.Logger) {
+	o.mu.Lock()
+	seqs := make([]uint64, 0, len(o.pending))
+	for seq := range o.pending {
+		seqs = append(seqs, seq)
+	}
+	sort.Slice(seqs, func(i, j int) bool { return seqs[i] < seqs[j] })
+	batch := make([]entry, 0, len(seqs))
+	for _, seq := range seqs {
+		batch = append(batch, o.pending[seq])
+	}
+	o.mu.Unlock()
+
+	for _, e := range batch {
+		if err := pub.Publish(e.exchange, e.routingKey, e.body); err != nil {
+			if logger != nil {
+				logger.Warn("outbox: delivery failed, will retry", "seq", e.seq, "err", err)
+			}
+			return // broker likely down; retry the whole round next tick
+		}
+		if err := o.ack(e.seq); err != nil && logger != nil {
+			// The message was delivered; failing to persist the ack only risks a
+			// redelivery on restart (at-least-once), so log and continue.
+			logger.Warn("outbox: ack write failed", "seq", e.seq, "err", err)
+		}
+	}
+}
+
+// ack records that seq was delivered and drops it from the pending set.
+func (o *Outbox) ack(seq uint64) error {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	b, _ := json.Marshal(line{Ack: seq})
+	if _, err := o.f.Write(append(b, '\n')); err != nil {
+		return err
+	}
+	if err := o.f.Sync(); err != nil {
+		return err
+	}
+	delete(o.pending, seq)
+	return nil
+}
+
+// Close closes the underlying file.
+func (o *Outbox) Close() error {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	return o.f.Close()
+}

pkg/outbox/outbox_test.go

@@ -0,0 +1,130 @@
+package outbox
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+)
+
+// fakePublisher records deliveries and can be toggled to fail (broker down).
+type fakePublisher struct {
+	mu    sync.Mutex
+	fail  bool
+	gotKR []string
+}
+
+func (f *fakePublisher) setFail(v bool) {
+	f.mu.Lock()
+	f.fail = v
+	f.mu.Unlock()
+}
+
+func (f *fakePublisher) Publish(_, routingKey string, body []byte) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	if f.fail {
+		return errors.New("broker down")
+	}
+	f.gotKR = append(f.gotKR, routingKey+":"+string(body))
+	return nil
+}
+
+func (f *fakePublisher) delivered() []string {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	out := make([]string, len(f.gotKR))
+	copy(out, f.gotKR)
+	return out
+}
+
+func waitFor(t *testing.T, cond func() bool) {
+	t.Helper()
+	deadline := time.Now().Add(2 * time.Second)
+	for time.Now().Before(deadline) {
+		if cond() {
+			return
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+	t.Fatal("condition not met within deadline")
+}
+
+func TestRelayDelivers(t *testing.T) {
+	ob, err := Open(t.TempDir() + "/outbox.log")
+	if err != nil {
+		t.Fatal(err)
+	}
+	pub := &fakePublisher{}
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go ob.Run(ctx, pub, 20*time.Millisecond, nil)
+
+	_ = ob.Publish("", "order-events", []byte("e1"))
+	_ = ob.Publish("", "order-events", []byte("e2"))
+
+	waitFor(t, func() bool { return len(pub.delivered()) == 2 })
+	waitFor(t, func() bool { return ob.PendingCount() == 0 })
+}
+
+func TestBrokerDownThenRecovers(t *testing.T) {
+	ob, _ := Open(t.TempDir() + "/outbox.log")
+	pub := &fakePublisher{}
+	pub.setFail(true)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go ob.Run(ctx, pub, 20*time.Millisecond, nil)
+
+	_ = ob.Publish("", "k", []byte("held"))
+	// While the broker is down the message stays pending and undelivered.
+	waitFor(t, func() bool { return ob.PendingCount() == 1 })
+	if len(pub.delivered()) != 0 {
+		t.Fatal("nothing should be delivered while broker is down")
+	}
+	// Broker recovers — the relay drains it.
+	pub.setFail(false)
+	waitFor(t, func() bool { return len(pub.delivered()) == 1 && ob.PendingCount() == 0 })
+}
+
+func TestDurableAcrossRestart(t *testing.T) {
+	path := t.TempDir() + "/outbox.log"
+	ob, _ := Open(path)
+	// Enqueue durably but never run a relay — simulates a crash after the state
+	// change recorded the event but before it was delivered.
+	_ = ob.Publish("", "order-events", []byte("survivor"))
+	_ = ob.Close()
+
+	// Restart: the pending event must still be there and then deliver.
+	ob2, err := Open(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if ob2.PendingCount() != 1 {
+		t.Fatalf("pending after restart = %d, want 1", ob2.PendingCount())
+	}
+	pub := &fakePublisher{}
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go ob2.Run(ctx, pub, 20*time.Millisecond, nil)
+	waitFor(t, func() bool { return len(pub.delivered()) == 1 })
+}
+
+func TestAckedNotRedelivered(t *testing.T) {
+	path := t.TempDir() + "/outbox.log"
+	ob, _ := Open(path)
+	pub := &fakePublisher{}
+	ctx, cancel := context.WithCancel(context.Background())
+	go ob.Run(ctx, pub, 20*time.Millisecond, nil)
+	_ = ob.Publish("", "k", []byte("once"))
+	waitFor(t, func() bool { return ob.PendingCount() == 0 })
+	cancel()
+	_ = ob.Close()
+
+	// Reopen (compaction drops the acked entry); nothing should redeliver.
+	ob2, _ := Open(path)
+	defer ob2.Close()
+	if ob2.PendingCount() != 0 {
+		t.Fatalf("acked message redelivered after restart: pending=%d", ob2.PendingCount())
+	}
+}