From a1f0bad972d602a67a4f319f6de42a48b6823f4b Mon Sep 17 00:00:00 2001 From: matst80 Date: Sun, 21 Jun 2026 23:47:53 +0200 Subject: [PATCH] more --- cmd/order/amqp.go | 7 ++++++- cmd/order/handlers_checkout.go | 11 ++++++----- cmd/order/main.go | 9 +++++---- pkg/idempotency/store.go | 28 ++++++++++++++-------------- pkg/outbox/outbox.go | 7 +++++++ 5 files changed, 38 insertions(+), 24 deletions(-) diff --git a/cmd/order/amqp.go b/cmd/order/amqp.go index 67adac4..4205772 100644 --- a/cmd/order/amqp.go +++ b/cmd/order/amqp.go @@ -60,7 +60,12 @@ func (p *rabbitPublisher) Publish(exchange, routingKey string, body []byte) erro return err } } - err := p.ch.PublishWithContext(context.Background(), exchange, routingKey, false, false, + // Bound the publish so a half-open/black-holed connection can't wedge the + // outbox relay indefinitely; on timeout the error path resets and the relay + // retries on its next tick. + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + err := p.ch.PublishWithContext(ctx, exchange, routingKey, false, false, amqp.Publishing{ContentType: "application/json", Body: body}) if err != nil { p.reset() // force a reconnect on the next attempt diff --git a/cmd/order/handlers_checkout.go b/cmd/order/handlers_checkout.go index c7e5e8e..cd86d9f 100644 --- a/cmd/order/handlers_checkout.go +++ b/cmd/order/handlers_checkout.go @@ -134,11 +134,12 @@ 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 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 != "" { + // Record the idempotency key durably only on success, so retries (incl. after + // a restart) return the captured order. A failed flow (402) is deliberately + // NOT recorded: the failed grain is terminal, and a retry with the same key + // should re-attempt rather than be handed back a dead order as a 409 hit + // (which the checkout client treats as a successful, existing order). + if req.IdempotencyKey != "" && runErr == nil { if err := s.idem.Put(req.IdempotencyKey, ordID); err != nil { s.logger.Error("idempotency record failed", "key", req.IdempotencyKey, "orderId", id.String(), "err", err) } diff --git a/cmd/order/main.go b/cmd/order/main.go index e587a4f..703c8c9 100644 --- a/cmd/order/main.go +++ b/cmd/order/main.go @@ -102,11 +102,12 @@ func main() { // 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 { - go box.Run(context.Background(), newRabbitPublisher(amqpURL), 5*time.Second, logger) - emitPub = box + // Durability-critical (same as the idempotency store): fail fast rather + // than silently publish events nowhere for the whole process lifetime. + log.Fatalf("open outbox: %v", err) } + go box.Run(context.Background(), newRabbitPublisher(amqpURL), 5*time.Second, logger) + emitPub = box } freg := flow.NewRegistry() flow.RegisterBuiltinHooks(freg) diff --git a/pkg/idempotency/store.go b/pkg/idempotency/store.go index bd234d0..3999da2 100644 --- a/pkg/idempotency/store.go +++ b/pkg/idempotency/store.go @@ -14,19 +14,23 @@ import ( "bufio" "encoding/json" "errors" + "hash/fnv" "os" "sync" ) +// lockShards bounds the per-key lock memory: keys hash into a fixed set of +// mutexes rather than a map that grows with every distinct key ever seen. Two +// unrelated keys that collide on a shard serialise occasionally — harmless. +const lockShards = 256 + // 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 + mu sync.Mutex + m map[string]uint64 + f *os.File + locks [lockShards]sync.Mutex } type record struct { @@ -62,7 +66,7 @@ func Open(path string) (*Store, error) { if err != nil { return nil, err } - return &Store{m: m, f: f, keyLocks: map[string]*sync.Mutex{}}, nil + return &Store{m: m, f: f}, nil } // Get returns the recorded id for key and whether it was present. @@ -103,13 +107,9 @@ 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() + h := fnv.New32a() + _, _ = h.Write([]byte(key)) + mu := &s.locks[h.Sum32()%lockShards] mu.Lock() return mu.Unlock } diff --git a/pkg/outbox/outbox.go b/pkg/outbox/outbox.go index 5386536..dc36f3b 100644 --- a/pkg/outbox/outbox.go +++ b/pkg/outbox/outbox.go @@ -23,6 +23,7 @@ import ( "errors" "log/slog" "os" + "path/filepath" "sort" "sync" "time" @@ -144,6 +145,12 @@ func (o *Outbox) compact() error { if err := os.Rename(tmp, o.path); err != nil { return err } + // fsync the directory so the rename (the file swap) is itself durable — a + // crash right after must not leave the pending log lost or a stale .tmp. + if d, derr := os.Open(filepath.Dir(o.path)); derr == nil { + _ = d.Sync() + _ = d.Close() + } o.f, err = os.OpenFile(o.path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644) return err }