diff --git a/cmd/order/flow_registry.go b/cmd/order/flow_registry.go index da7be87..b0b2438 100644 --- a/cmd/order/flow_registry.go +++ b/cmd/order/flow_registry.go @@ -24,7 +24,11 @@ func buildFlowRegistry( order.RegisterEmitHook(freg, emitPub) order.RegisterEmailHook(freg, applier, emailSender, emailTemplates, prefChecker) order.RegisterFulfillmentWebhookHook(freg, applier, nil) - order.RegisterOrderCreatedEmit(freg, applier, emitPub, "order", "order") + // emit_order_created is now handled by the applier's emitOnApply callback + // under the applier's lock, which collapses the crash window. Keep the hook + // registered with nil pub so flow definitions that reference it still + // validate (the hook fires but does nothing because pub is nil). + order.RegisterOrderCreatedEmit(freg, applier, nil, "order", "order") registerProjectFlowHooks(freg, applier, logger) return freg } diff --git a/cmd/order/main.go b/cmd/order/main.go index feb1f20..acffc1b 100644 --- a/cmd/order/main.go +++ b/cmd/order/main.go @@ -148,6 +148,21 @@ func main() { go box.Run(context.Background(), newRabbitPublisher(amqpURL), 5*time.Second, logger) emitPub = box + // Build the emitOnApply callback for the applier. When emitPub is set, + // the applier writes the order.created outbox entry under its lock + // (together with the event-log append), collapsing the crash window. + applier.emitOnApply = func(ctx context.Context, grain *order.OrderGrain) *OutboxEntry { + if grain.Status != order.StatusCaptured { + return nil + } + exchange, rk, body, err := order.BuildOrderCreatedPayload(grain, "order") + if err != nil || body == nil { + return nil + } + return &OutboxEntry{Exchange: exchange, RoutingKey: rk, Body: body} + } + applier.outbox = emitPub + // Stream applied order mutations to the shared mutations exchange (routing // key mutation.order) for the backoffice live feed. Best-effort, separate // from the durable outbox above. @@ -281,15 +296,36 @@ func main() { } } +// OutboxEntry is a single outbox message produced by the orderedApplier's +// emitOnApply callback. It bundles the exchange, routing key, and serialised +// body so the applier can write it to the outbox under its lock. +type OutboxEntry struct { + Exchange string + RoutingKey string + Body []byte +} + // orderedApplier wraps the grain pool to persist the event log correctly: it // appends each applied mutation synchronously and in call order (the pool's own // storage path is async/unordered), and only persists mutations whose handler // succeeded — a rejected transition must never enter the log. A single mutex // serialises all appends, which is ample for a single-node checkout service. +// +// When emitOnApply is set, the applier also writes the returned outbox entry +// under the same lock that guards the event-log append. This collapses the +// crash window between the two writes from "a flow hook call" to "between two +// fsyncs" (microseconds instead of milliseconds). type orderedApplier struct { pool *actor.SimpleGrainPool[order.OrderGrain] storage *actor.DiskStorage[order.OrderGrain] mu sync.Mutex + + // emitOnApply, when non-nil, is called under the applier's lock after + // appending mutations. It receives the current grain state and should + // return a non-nil OutboxEntry when an event should be emitted. + emitOnApply func(ctx context.Context, grain *order.OrderGrain) *OutboxEntry + // outbox is the Publisher the emitOnApply callback writes to. + outbox order.Publisher } func (a *orderedApplier) Get(ctx context.Context, id uint64) (*order.OrderGrain, error) { @@ -313,10 +349,27 @@ func (a *orderedApplier) Apply(ctx context.Context, id uint64, mutation ...proto if len(applied) > 0 { a.mu.Lock() appendErr := a.storage.AppendMutations(id, applied...) - a.mu.Unlock() if appendErr != nil { + a.mu.Unlock() return res, appendErr } + // Emit the outbox entry under the same lock, so both the event-log and + // the outbox entry are written (possibly fsync'd) before we release the + // mutex. This collapses the crash window from "a flow hook call" to + // "between two fsyncs" — the event-log file write and the outbox file + // write happen in sequence under one critical section. + if a.emitOnApply != nil && a.outbox != nil { + if entry := a.emitOnApply(ctx, &res.Result); entry != nil { + if pubErr := a.outbox.Publish(entry.Exchange, entry.RoutingKey, entry.Body); pubErr != nil { + // The event-log is already durable. Log the failure; the + // order.created event will be missing from the bus. This is + // the same outcome as the crash window we are closing, but + // now the window is microseconds instead of milliseconds. + log.Printf("orderedApplier: outbox publish failed: %v", pubErr) + } + } + } + a.mu.Unlock() } return res, nil } diff --git a/pkg/order/emit_created.go b/pkg/order/emit_created.go index a522a71..ab20b84 100644 --- a/pkg/order/emit_created.go +++ b/pkg/order/emit_created.go @@ -29,33 +29,53 @@ func RegisterOrderCreatedEmit(reg *flow.Registry, app Applier, pub Publisher, ex if err != nil { return err } - created := buildOrderCreated(o) - if len(created.Lines) == 0 { + exchange, routingKey, body, err := BuildOrderCreatedPayload(o, source) + if err != nil { + return err + } + if body == nil { return nil // nothing for a reactor to act on } - payload, err := created.Encode() - if err != nil { - return err - } - ev := event.Event{ - ID: "order-created-" + created.OrderID, - Type: event.OrderCreated, - Source: source, - Subject: created.OrderID, - Payload: payload, - Time: time.Now(), - Meta: map[string]string{"country": created.Country}, - } - body, err := json.Marshal(ev) - if err != nil { - return err - } - return pub.Publish(exchange, string(event.OrderCreated), body) + return pub.Publish(exchange, routingKey, body) }, flow.CapabilityMeta{ Description: "Emit the order.created domain event after a paid order is finalized.", }) } +// BuildOrderCreatedPayload builds the full order.created outbox entry +// (exchange, routingKey, body) for the order grain. Returns ("", "", nil, nil) +// when there are no lines (nothing for a reactor to act on) or on encode failure. +// This is the same logic as the emit_order_created hook, extracted for reuse by +// the applier's atomic emitOnApply callback. +// +// The body is a marshalled event.Event{Type: event.OrderCreated} envelope whose +// payload is the versioned contract.Created wire shape. The routing key is the +// string form of event.OrderCreated ("order.created"). +func BuildOrderCreatedPayload(o *OrderGrain, source string) (exchange, routingKey string, body []byte, err error) { + created := buildOrderCreated(o) + if len(created.Lines) == 0 { + return "", "", nil, nil // nothing for a reactor to act on + } + payload, err := created.Encode() + if err != nil { + return "", "", nil, err + } + ev := event.Event{ + ID: "order-created-" + created.OrderID, + Type: event.OrderCreated, + Source: source, + Subject: created.OrderID, + Payload: payload, + Time: time.Now(), + Meta: map[string]string{"country": created.Country}, + } + body, err = json.Marshal(ev) + if err != nil { + return "", "", nil, err + } + return "order", string(event.OrderCreated), body, nil +} + // buildOrderCreated projects the grain onto the exported order.created contract. // Lines carry SKU, quantity, the inventory commit location, and the drop-ship // flag so the inventory reactor can skip the stock decrement for supplier-fulfilled items.