From 9d420c43faddfc64a5d11b20d07e91f5e65883f2 Mon Sep 17 00:00:00 2001 From: matst80 Date: Sun, 21 Jun 2026 23:43:48 +0200 Subject: [PATCH] more checkout --- cmd/order/amqp.go | 44 ++++-- cmd/order/handlers_checkout.go | 34 ++--- cmd/order/main.go | 24 +++- pkg/idempotency/store.go | 122 ++++++++++++++++ pkg/idempotency/store_test.go | 106 ++++++++++++++ pkg/outbox/outbox.go | 255 +++++++++++++++++++++++++++++++++ pkg/outbox/outbox_test.go | 130 +++++++++++++++++ 7 files changed, 684 insertions(+), 31 deletions(-) create mode 100644 pkg/idempotency/store.go create mode 100644 pkg/idempotency/store_test.go create mode 100644 pkg/outbox/outbox.go create mode 100644 pkg/outbox/outbox_test.go diff --git a/cmd/order/amqp.go b/cmd/order/amqp.go index 33d8cb4..67adac4 100644 --- a/cmd/order/amqp.go +++ b/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: diff --git a/cmd/order/handlers_checkout.go b/cmd/order/handlers_checkout.go index e9e7706..c7e5e8e 100644 --- a/cmd/order/handlers_checkout.go +++ b/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{ diff --git a/cmd/order/main.go b/cmd/order/main.go index 15f341b..e587a4f 100644 --- a/cmd/order/main.go +++ b/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() diff --git a/pkg/idempotency/store.go b/pkg/idempotency/store.go new file mode 100644 index 0000000..bd234d0 --- /dev/null +++ b/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() +} diff --git a/pkg/idempotency/store_test.go b/pkg/idempotency/store_test.go new file mode 100644 index 0000000..ec3fb30 --- /dev/null +++ b/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") + } +} diff --git a/pkg/outbox/outbox.go b/pkg/outbox/outbox.go new file mode 100644 index 0000000..5386536 --- /dev/null +++ b/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() +} diff --git a/pkg/outbox/outbox_test.go b/pkg/outbox/outbox_test.go new file mode 100644 index 0000000..2e95c34 --- /dev/null +++ b/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()) + } +}