From 4764db77e94b23b70b52a5fcf3eb25340ec5a7a8 Mon Sep 17 00:00:00 2001 From: mchain0 Date: Mon, 27 Jul 2026 11:44:41 +0200 Subject: [PATCH 1/3] cre-5665: durable emitter regression test --- pkg/durableemitter/durable_emitter_test.go | 99 ++++++++++++++++++++++ 1 file changed, 99 insertions(+) diff --git a/pkg/durableemitter/durable_emitter_test.go b/pkg/durableemitter/durable_emitter_test.go index 3bda950a1e..f1980a60ca 100644 --- a/pkg/durableemitter/durable_emitter_test.go +++ b/pkg/durableemitter/durable_emitter_test.go @@ -1508,3 +1508,102 @@ 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 demonstrates the root cause of CRE-5665: +// when DurableEmitterEnabled defaults to true (v2.56+), every GlobalEmit call +// blocks on the insert coalescer's flush interval + DB insert latency. When +// the emitter is not initialized (v2.54/v2.55 default false), GlobalEmit is +// an instant no-op. +// +// The test simulates a workflow execution that emits 5 telemetry events +// (e.g. capability started, user metric, capability finished, execution +// finished, metering report) and compares the total wall-clock time in both +// modes. +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) --- + // GlobalEmit calls DurableEmitter.Emit which blocks on the insert coalescer. + // Each call blocks for at least flushInterval (the coalescer waits that long + // to collect a batch before flushing) plus the simulated 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 (blocked on insert coalescer)", numEmissions, enabledDuration) + + // Each emission blocks for at least flushInterval because the batch (size 500) + // never fills up with a single concurrent caller — the coalescer always waits + // the full linger period before flushing. With 5 sequential emissions that's + // at least 5 * flushInterval = 500ms. + minExpected := time.Duration(numEmissions) * flushInterval + require.GreaterOrEqual(t, enabledDuration, minExpected, + "with durable emitter enabled, %d sequential emissions must take at least %v (each blocks on flush interval)", + numEmissions, minExpected) + + // The enabled path must be dramatically slower than the disabled path. + // This is the regression: the same workflow takes 50x+ longer just because + // the durable emitter default flipped from false to true. + ratio := float64(enabledDuration) / float64(disabledDuration) + t.Logf("Regression ratio: %.0fx slower with durable emitter enabled", ratio) + require.Greater(t, ratio, 50.0, + "enabled path must be at least 50x slower than disabled path") +} From 904e53e0a6dd36b722398e8a9bf2b66eaca7fdb2 Mon Sep 17 00:00:00 2001 From: mchain0 Date: Mon, 27 Jul 2026 12:27:55 +0200 Subject: [PATCH 2/3] cre-5665: durable emitter regression fix and test --- pkg/durableemitter/durable_emitter.go | 40 ++++++++++++ pkg/durableemitter/durable_emitter_test.go | 73 +++++++++++++--------- pkg/durableemitter/setup.go | 18 +++++- 3 files changed, 97 insertions(+), 34 deletions(-) diff --git a/pkg/durableemitter/durable_emitter.go b/pkg/durableemitter/durable_emitter.go index 0e2cdf3db9..e836875f9a 100644 --- a/pkg/durableemitter/durable_emitter.go +++ b/pkg/durableemitter/durable_emitter.go @@ -144,6 +144,15 @@ 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. +type asyncEmitRequest struct { + body []byte + attrKVs []any +} + // insertRequest is a single Emit() caller waiting for a coalesced batch INSERT. type insertRequest struct { payload []byte @@ -192,6 +201,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 @@ -252,6 +267,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", @@ -319,6 +335,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 } @@ -504,6 +524,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 { @@ -525,6 +552,19 @@ 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 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() { diff --git a/pkg/durableemitter/durable_emitter_test.go b/pkg/durableemitter/durable_emitter_test.go index f1980a60ca..de6cbe567a 100644 --- a/pkg/durableemitter/durable_emitter_test.go +++ b/pkg/durableemitter/durable_emitter_test.go @@ -1525,20 +1525,22 @@ func (s *slowBatchStore) InsertBatch(ctx context.Context, payloads [][]byte) ([] return s.MemDurableEventStore.InsertBatch(ctx, payloads) } -// TestGlobalEmit_BlockingBehavior demonstrates the root cause of CRE-5665: -// when DurableEmitterEnabled defaults to true (v2.56+), every GlobalEmit call -// blocks on the insert coalescer's flush interval + DB insert latency. When -// the emitter is not initialized (v2.54/v2.55 default false), GlobalEmit is -// an instant no-op. +// TestGlobalEmit_BlockingBehavior is a regression guard for CRE-5665. // -// The test simulates a workflow execution that emits 5 telemetry events -// (e.g. capability started, user metric, capability finished, execution -// finished, metering report) and compares the total wall-clock time in both -// modes. +// 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 + 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. @@ -1557,10 +1559,11 @@ func TestGlobalEmit_BlockingBehavior(t *testing.T) { 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) --- - // GlobalEmit calls DurableEmitter.Emit which blocks on the insert coalescer. - // Each call blocks for at least flushInterval (the coalescer waits that long - // to collect a batch before flushing) plus the simulated insert latency. + // --- 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, @@ -1588,22 +1591,30 @@ func TestGlobalEmit_BlockingBehavior(t *testing.T) { require.NoError(t, err, "GlobalEmit must succeed when emitter is initialized") } enabledDuration := time.Since(start) - t.Logf("Durable emitter ENABLED: %d emissions took %v (blocked on insert coalescer)", numEmissions, enabledDuration) - - // Each emission blocks for at least flushInterval because the batch (size 500) - // never fills up with a single concurrent caller — the coalescer always waits - // the full linger period before flushing. With 5 sequential emissions that's - // at least 5 * flushInterval = 500ms. - minExpected := time.Duration(numEmissions) * flushInterval - require.GreaterOrEqual(t, enabledDuration, minExpected, - "with durable emitter enabled, %d sequential emissions must take at least %v (each blocks on flush interval)", - numEmissions, minExpected) - - // The enabled path must be dramatically slower than the disabled path. - // This is the regression: the same workflow takes 50x+ longer just because - // the durable emitter default flipped from false to true. + 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("Regression ratio: %.0fx slower with durable emitter enabled", ratio) - require.Greater(t, ratio, 50.0, - "enabled path must be at least 50x slower than disabled path") + 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) } diff --git a/pkg/durableemitter/setup.go b/pkg/durableemitter/setup.go index 6fa0c58f10..6c2ef5c2a5 100644 --- a/pkg/durableemitter/setup.go +++ b/pkg/durableemitter/setup.go @@ -33,15 +33,27 @@ 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) + select { + case d.asyncEmitCh <- &asyncEmitRequest{body: body, attrKVs: attrKVs}: + 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. + return nil } - return nil } // SetupConfig holds all configuration required to create and start a From e61dd3a959c873d858c39f66c66d125385a607d3 Mon Sep 17 00:00:00 2001 From: mchain0 Date: Mon, 27 Jul 2026 13:05:25 +0200 Subject: [PATCH 3/3] cre-5665: metrics --- pkg/durableemitter/durable_emitter.go | 17 ++++++- pkg/durableemitter/durable_emitter_metrics.go | 47 +++++++++++++++++++ pkg/durableemitter/setup.go | 7 ++- 3 files changed, 68 insertions(+), 3 deletions(-) diff --git a/pkg/durableemitter/durable_emitter.go b/pkg/durableemitter/durable_emitter.go index 959bb8c277..e557252ffd 100644 --- a/pkg/durableemitter/durable_emitter.go +++ b/pkg/durableemitter/durable_emitter.go @@ -149,9 +149,12 @@ func DefaultConfig() Config { // 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 + body []byte + attrKVs []any + enqueuedAt time.Time } // insertRequest is a single Emit() caller waiting for a coalesced batch INSERT. @@ -560,6 +563,9 @@ func (d *DurableEmitter) stop() error { // 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) } @@ -855,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) } } diff --git a/pkg/durableemitter/durable_emitter_metrics.go b/pkg/durableemitter/durable_emitter_metrics.go index 71c931a4a6..e9e62b67c9 100644 --- a/pkg/durableemitter/durable_emitter_metrics.go +++ b/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 } diff --git a/pkg/durableemitter/setup.go b/pkg/durableemitter/setup.go index 5e91d1ec49..2d321e4b82 100644 --- a/pkg/durableemitter/setup.go +++ b/pkg/durableemitter/setup.go @@ -45,13 +45,18 @@ func GlobalEmit(ctx context.Context, body []byte, attrKVs ...any) error { if d == nil { return ErrNotInitialized } + req := &asyncEmitRequest{body: body, attrKVs: attrKVs, enqueuedAt: time.Now()} select { - case d.asyncEmitCh <- &asyncEmitRequest{body: body, attrKVs: attrKVs}: + 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 } }