flow

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
 }

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
 }

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)
+		exchange, routingKey, body, err := BuildOrderCreatedPayload(o, source)
-		if len(created.Lines) == 0 {
+		if err != nil {
+			return err
+		}
+		if body == nil {
 			return nil // nothing for a reactor to act on
 		}
-		payload, err := created.Encode()
+		return pub.Publish(exchange, routingKey, body)
-		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)
 	}, 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.