cre-5665: durable emitter regression fix and test

pkg/durableemitter/durable_emitter.go

@@ -145,6 +145,18 @@ func DefaultConfig() Config {
 // A separate expiry loop garbage-collects events older than EventTTL to bound
 // table growth.
 
+// asyncEmitRequest is a single GlobalEmit() caller's payload enqueued for
+// background processing by the async emit loop. Unlike insertRequest, the
+// caller never waits for a result — the event is persisted and published
+// asynchronously so the workflow execution path is not blocked.
+// enqueuedAt is set at enqueue time so asyncEmitLoop can record the
+// end-to-end async emit duration.
+type asyncEmitRequest struct {
+	body       []byte
+	attrKVs    []any
+	enqueuedAt time.Time
+}
+
 // insertRequest is a single Emit() caller waiting for a coalesced batch INSERT.
 type insertRequest struct {
 	payload []byte
@@ -193,6 +205,12 @@ type DurableEmitter struct {
 	stopCh services.StopChan
 	wg     sync.WaitGroup
 
+	// asyncEmitCh buffers events for the non-blocking GlobalEmit path. GlobalEmit
+	// does a non-blocking send to this channel; a background goroutine
+	// (asyncEmitLoop) drains it and calls Emit synchronously. This decouples
+	// workflow metric emission from DB insert latency.
+	asyncEmitCh chan *asyncEmitRequest
+
 	// retransmit paging cursor. The retransmit loop pages through the pending
 	// backlog in (created_at, id) order, advancing the cursor each tick and
 	// wrapping to zero at the end, so a persistently-failing ("poison") event
@@ -253,6 +271,7 @@ func NewDurableEmitter(
 		cfg:               cfg,
 		metrics:           m,
 		stopCh:            make(chan struct{}),
+		asyncEmitCh:       make(chan *asyncEmitRequest, 50_000),
 	}
 	d.Service, d.eng = services.Config{
 		Name:  "DurableEmitter",
@@ -320,6 +339,10 @@ func (d *DurableEmitter) start(ctx context.Context) error {
 	if d.metrics != nil && d.cfg.Metrics != nil {
 		d.wg.Go(d.metricsLoop)
 	}
+
+	// asyncEmitLoop processes GlobalEmit events in the background so the
+	// caller never blocks on DB insert latency.
+	d.wg.Go(d.asyncEmitLoop)
 	return nil
 }
 
@@ -505,6 +528,13 @@ func (d *DurableEmitter) deliveryCallback(id int64, eventPb *chipingress.CloudEv
 // in-flight callbacks). It is invoked by the services.Engine when the embedded
 // Service is closed.
 func (d *DurableEmitter) stop() error {
+	// Close asyncEmitCh first so the async emit goroutine finishes processing
+	// any queued events before we tear down the insert coalescer and batch
+	// emitter. The goroutine is tracked by d.wg, so d.wg.Wait() below will
+	// wait for it.
+	if d.asyncEmitCh != nil {
+		close(d.asyncEmitCh)
+	}
 	if d.insertCh != nil {
 		d.insertShutdown.Store(true)
 		for d.insertInFlight.Load() > 0 {
@@ -526,6 +556,22 @@ func (d *DurableEmitter) stop() error {
 	return nil
 }
 
+// asyncEmitLoop drains the asyncEmitCh and calls Emit for each request.
+// This is the background worker that makes GlobalEmit non-blocking: events
+// are enqueued by GlobalEmit and processed here, so the caller never waits
+// for DB insert latency. On channel close (during stop) the loop drains any
+// remaining events before exiting.
+func (d *DurableEmitter) asyncEmitLoop() {
+	for req := range d.asyncEmitCh {
+		if d.metrics != nil {
+			d.metrics.asyncEmitDuration.Record(context.Background(), time.Since(req.enqueuedAt).Seconds())
+		}
+		if err := d.Emit(context.Background(), req.body, req.attrKVs...); err != nil {
+			d.eng.Warnw("DurableEmitter: async emit failed", "err", err)
+		}
+	}
+}
+
 // insertBatchLoop collects insertRequest items from insertCh and flushes them
 // as multi-row INSERTs via BatchInserter.InsertBatch.
 func (d *DurableEmitter) insertBatchLoop() {
@@ -815,6 +861,13 @@ func (d *DurableEmitter) metricsLoop() {
 			} else {
 				d.metrics.deleteCoalescerFill.Record(ctx, 0)
 			}
+			if d.asyncEmitCh != nil {
+				if c := cap(d.asyncEmitCh); c > 0 {
+					depth := int64(len(d.asyncEmitCh))
+					d.metrics.asyncQueueDepth.Record(ctx, depth)
+					d.metrics.asyncQueueFill.Record(ctx, float64(depth)/float64(c))
+				}
+			}
 			d.metrics.pollProcessGauges(ctx)
 		}
 	}

pkg/durableemitter/durable_emitter_metrics.go

@@ -79,6 +79,24 @@ type durableEmitterMetrics struct {
 	// deleteCoalescerFill reports the delete-coalescer channel fill ratio
 	// (len/cap). Only meaningful when DeleteBatchSize > 0; otherwise 0.
 	deleteCoalescerFill metric.Float64Gauge
+	// asyncQueueDepth reports the current number of events waiting in the
+	// async emit channel (GlobalEmit → asyncEmitLoop). A rising value means
+	// the background loop is not keeping up with incoming GlobalEmit calls.
+	asyncQueueDepth metric.Int64Gauge
+	// asyncQueueFill reports the fill ratio (len/cap) of the async emit
+	// channel. Saturating toward 1.0 indicates the async loop is a bottleneck
+	// and events may be dropped (fail-open) on the next GlobalEmit call.
+	asyncQueueFill metric.Float64Gauge
+	// asyncEmitDuration measures the wall time from when GlobalEmit enqueues
+	// a request to when asyncEmitLoop finishes calling Emit for it. This
+	// captures the real end-to-end latency that was previously blocking the
+	// workflow caller.
+	asyncEmitDuration metric.Float64Histogram
+	// asyncDropped counts events that were dropped because the async emit
+	// channel was full (fail-open in GlobalEmit). Non-zero means the durable
+	// emitter is overwhelmed — events were still delivered via the real-time
+	// beholder emitter, but the durable copy was lost.
+	asyncDropped metric.Int64Counter
 }
 
 // durationBuckets provides histogram boundaries (in seconds) tuned for
@@ -294,6 +312,35 @@ func newDurableEmitterMetrics(meter metric.Meter, clientName string) (*durableEm
 	); err != nil {
 		return nil, err
 	}
+	if m.asyncQueueDepth, err = meter.Int64Gauge(
+		"durable_emitter.async_queue.depth",
+		metric.WithUnit("{event}"),
+		metric.WithDescription("Number of events waiting in the async emit channel (GlobalEmit → asyncEmitLoop). Rising values indicate the background loop is not keeping up."),
+	); err != nil {
+		return nil, err
+	}
+	if m.asyncQueueFill, err = meter.Float64Gauge(
+		"durable_emitter.async_queue.fill_ratio",
+		metric.WithUnit("1"),
+		metric.WithDescription("Async emit channel fill ratio (len/cap). Saturating toward 1.0 means events may be dropped (fail-open) on the next GlobalEmit call."),
+	); err != nil {
+		return nil, err
+	}
+	if m.asyncEmitDuration, err = meter.Float64Histogram(
+		"durable_emitter.async_emit.duration",
+		metric.WithUnit("s"),
+		metric.WithDescription("Wall time from GlobalEmit enqueue to asyncEmitLoop Emit completion. This is the end-to-end latency that was previously blocking the workflow caller (CRE-5665)."),
+		durationBuckets,
+	); err != nil {
+		return nil, err
+	}
+	if m.asyncDropped, err = meter.Int64Counter(
+		"durable_emitter.async_queue.dropped",
+		metric.WithUnit("{event}"),
+		metric.WithDescription("Events dropped because the async emit channel was full (fail-open). The real-time beholder emitter already delivered them; only the durable copy was lost."),
+	); err != nil {
+		return nil, err
+	}
 	return m, nil
 }
 

pkg/durableemitter/durable_emitter_test.go

@@ -1554,3 +1554,113 @@ func (m *MemDurableEventStore) ObserveDurableQueue(_ context.Context, eventTTL t
 	st.TTLBudget = eventTTL - st.OldestPendingAge
 	return st, nil
 }
+
+// slowBatchStore wraps MemDurableEventStore and adds a configurable delay to
+// InsertBatch to simulate Postgres INSERT latency.
+type slowBatchStore struct {
+	*MemDurableEventStore
+	insertDelay time.Duration
+}
+
+func (s *slowBatchStore) InsertBatch(ctx context.Context, payloads [][]byte) ([]int64, error) {
+	select {
+	case <-time.After(s.insertDelay):
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
+	return s.MemDurableEventStore.InsertBatch(ctx, payloads)
+}
+
+// TestGlobalEmit_BlockingBehavior is a regression guard for CRE-5665.
+//
+// The original bug: when DurableEmitterEnabled defaulted to true (v2.56+),
+// every GlobalEmit call blocked on the insert coalescer's flush interval
+// (100ms) + DB insert latency, causing workflow execution times to regress
+// from ~10s to ~60s. When the emitter was not initialized (v2.54/v2.55
+// default false), GlobalEmit was an instant no-op.
+//
+// The fix: GlobalEmit is now non-blocking — it enqueues to a buffered
+// channel and a background goroutine calls Emit. This test verifies that
+// both the disabled and enabled paths complete quickly, and that events
+// are still eventually delivered in the enabled path.
+func TestGlobalEmit_BlockingBehavior(t *testing.T) {
+	const numEmissions = 5
+	const flushInterval = 100 * time.Millisecond // matches v2.56 application.go config
+	const insertDelay = 5 * time.Millisecond     // simulated Postgres INSERT latency
+
+	// --- Phase 1: Durable emitter NOT initialized (v2.54/v2.55 behavior) ---
+	// GlobalEmit returns ErrNotInitialized immediately — zero blocking.
+	prevEmitter := globalEmitter.Load()
+	globalEmitter.Store(nil)
+	t.Cleanup(func() { globalEmitter.Store(prevEmitter) })
+
+	ctx := t.Context()
+	start := time.Now()
+	for i := 0; i < numEmissions; i++ {
+		err := GlobalEmit(ctx, []byte("metric-event"), "source", "platform", "type", "test")
+		require.ErrorIs(t, err, ErrNotInitialized, "GlobalEmit must return ErrNotInitialized when no emitter is set")
+	}
+	disabledDuration := time.Since(start)
+	t.Logf("Durable emitter DISABLED: %d emissions took %v (no-op, instant)", numEmissions, disabledDuration)
+	require.Less(t, disabledDuration, 10*time.Millisecond,
+		"with durable emitter disabled, %d emissions must complete in under 10ms", numEmissions)
+
+	// --- Phase 2: Durable emitter initialized (v2.56+ behavior, post-fix) ---
+	// GlobalEmit must now be non-blocking: it enqueues to the async channel
+	// and returns immediately. A background goroutine persists and publishes
+	// the event. The wall-clock time for the caller must not be affected by
+	// the insert coalescer's flush interval or DB insert latency.
+	store := &slowBatchStore{
+		MemDurableEventStore: NewMemDurableEventStore(),
+		insertDelay:          insertDelay,
+	}
+	be := newTestBatchEmitter()
+	cfg := Config{
+		InsertBatchSize:          500, // matches v2.56 application.go config
+		InsertBatchWorkers:       1,
+		InsertBatchFlushInterval: flushInterval,
+		DeleteBatchSize:          100,
+		DeleteBatchWorkers:       1,
+		DisablePruning:           true,
+		PublishTimeout:           5 * time.Second,
+		EventTTL:                 1 * time.Hour,
+	}
+	em, err := NewDurableEmitter(store, be, false, cfg, logger.Test(t), nil)
+	require.NoError(t, err)
+	servicetest.Run(t, em)
+
+	globalEmitter.Store(em)
+
+	start = time.Now()
+	for i := 0; i < numEmissions; i++ {
+		err := GlobalEmit(ctx, []byte("metric-event"), "source", "platform", "type", "test")
+		require.NoError(t, err, "GlobalEmit must succeed when emitter is initialized")
+	}
+	enabledDuration := time.Since(start)
+	t.Logf("Durable emitter ENABLED:  %d emissions took %v (non-blocking, async)", numEmissions, enabledDuration)
+
+	// REGRESSION GUARD: the enabled path must NOT block on the insert coalescer.
+	// Before the fix, 5 sequential emissions took >= 5 * flushInterval = 500ms
+	// because each call blocked until the coalescer flushed. After the fix,
+	// GlobalEmit enqueues to a buffered channel and returns immediately.
+	maxAllowed := time.Duration(numEmissions) * flushInterval
+	require.Less(t, enabledDuration, maxAllowed,
+		"with durable emitter enabled, %d emissions must NOT take >= %v — GlobalEmit must be non-blocking",
+		numEmissions, maxAllowed)
+
+	// Both paths should be in the same order of magnitude — the enabled path
+	// must not be dramatically slower than the disabled path.
+	ratio := float64(enabledDuration) / float64(disabledDuration)
+	t.Logf("Ratio: %.1fx (enabled/disabled)", ratio)
+	require.Less(t, ratio, 50.0,
+		"enabled path must not be more than 50x slower than disabled path (was 4000x+ before fix)")
+
+	// Verify events are eventually delivered despite the non-blocking caller.
+	// The async goroutine processes events through the same Emit path (insert
+	// coalescer + batch emitter), so delivery is delayed by the flush interval
+	// but must complete.
+	require.Eventually(t, func() bool {
+		return be.callCount.Load() == int64(numEmissions)
+	}, 5*time.Second, 50*time.Millisecond,
+		"all %d events must eventually be published by the batch emitter", numEmissions)
+}

pkg/durableemitter/setup.go

@@ -33,15 +33,32 @@ func GetGlobalEmitter() *DurableEmitter {
 }
 
 // GlobalEmit emits an event via the global DurableEmitter.
+//
+// This function is non-blocking: it enqueues the event to the emitter's
+// async channel and returns immediately. A background goroutine persists
+// and publishes the event. If the async channel is full (backpressure),
+// the event is dropped (fail-open) — the beholder emitter already delivered
+// it in real-time, and the durable emitter's value is persistence for
+// retransmit, not inline delivery.
 func GlobalEmit(ctx context.Context, body []byte, attrKVs ...any) error {
 	d := globalEmitter.Load()
 	if d == nil {
 		return ErrNotInitialized
 	}
-	if err := d.Emit(ctx, body, attrKVs...); err != nil {
-		return fmt.Errorf("%w: %w", ErrEmitFailed, err)
+	req := &asyncEmitRequest{body: body, attrKVs: attrKVs, enqueuedAt: time.Now()}
+	select {
+	case d.asyncEmitCh <- req:
+		return nil
+	default:
+		// Channel full — fail open. The event was already emitted via the
+		// beholder emitter; dropping the durable copy is preferable to
+		// blocking the workflow execution path.
+		if d.metrics != nil {
+			d.metrics.asyncDropped.Add(ctx, 1)
+		}
+		d.eng.Warnw("DurableEmitter: async emit channel full, dropping durable copy (fail-open)")
+		return nil
 	}
-	return nil
 }
 
 // SetupConfig holds all configuration required to create and start a