From c0f89a313b70376a12a69e4c5dcfa174c47af17b Mon Sep 17 00:00:00 2001 From: Chet Sharma Date: Tue, 13 Oct 2026 14:44:35 -0700 Subject: [PATCH 1/2] durableemitter: don't panic when a delivery callback runs after stop stop closed deleteCh as soon as batchEmitter.Stop returned, but Stop can return at its shutdown timeout with callbacks still running, so a late success callback panicked on send. The close now happens under a lock and late callbacks skip the delete, so the event is retransmitted on the next start. --- pkg/durableemitter/durable_emitter.go | 35 ++++++++++++----- pkg/durableemitter/durable_emitter_test.go | 44 ++++++++++++++++++++++ 2 files changed, 70 insertions(+), 9 deletions(-) diff --git a/pkg/durableemitter/durable_emitter.go b/pkg/durableemitter/durable_emitter.go index 225adb6c10..8f71243670 100644 --- a/pkg/durableemitter/durable_emitter.go +++ b/pkg/durableemitter/durable_emitter.go @@ -196,12 +196,17 @@ type DurableEmitter struct { // deleteCh funnels delivered event ids to the delete coalescer. Nil when delete // coalescing is disabled (DeleteBatchSize <= 0). Its only producers are the - // delivery callbacks, so it is closed during shutdown only after they have - // quiesced (see stop). deleteWg tracks - // the deleteBatchLoop workers, which must outlive d.wg so callbacks running - // during the batch emitter's final flush can still enqueue deletes. - deleteCh chan int64 - deleteWg sync.WaitGroup + // delivery callbacks. deleteWg tracks the deleteBatchLoop workers, which must + // outlive d.wg so callbacks running during the batch emitter's final flush can + // still enqueue deletes. + // + // batchEmitter.Stop can return at its shutdown timeout while callbacks are still + // running, so deleteCh is closed under deleteMu and sends happen under its read + // lock; a callback that arrives after deleteClosed leaves the event persisted. + deleteCh chan int64 + deleteWg sync.WaitGroup + deleteMu sync.RWMutex + deleteClosed bool // stopCh signals background loops to exit. stopCh services.StopChan @@ -574,10 +579,15 @@ func (d *DurableEmitter) stop() error { // Stop the batch emitter: flushes remaining queued events, waits for all // in-flight PublishBatch RPCs, and waits for all delivery callbacks. d.batchEmitter.Stop() - // The delivery callbacks (drained by batchEmitter.Stop) are the only producers - // of deleteCh, so it is now safe to close it and let the workers flush any buffered deletes. + // Delivery callbacks are deleteCh's only producers. Most have finished once + // batchEmitter.Stop returns, but it can return at its shutdown timeout with some + // still running; closing under deleteMu waits for in-progress sends (the workers + // keep draining) and turns any later send into a no-op. if d.deleteCh != nil { + d.deleteMu.Lock() + d.deleteClosed = true close(d.deleteCh) + d.deleteMu.Unlock() d.deleteWg.Wait() } return nil @@ -642,11 +652,18 @@ func (d *DurableEmitter) insertBatchLoop() { // enqueueDelete hands a delivered event id to the delete coalescer. It returns // false when coalescing is disabled so the caller can delete inline. // Send is a blocking hand-off; back-pressure here slows the delivery callback -// rather than dropping a delete. +// rather than dropping a delete. After shutdown it skips the delete: the event +// stays persisted and is retransmitted on the next start (at-least-once). func (d *DurableEmitter) enqueueDelete(id int64) bool { if d.deleteCh == nil { return false } + d.deleteMu.RLock() + defer d.deleteMu.RUnlock() + if d.deleteClosed { + d.eng.Debugw("DurableEmitter: delivery confirmed after shutdown; event left for retransmit", "id", id) + return true + } d.deleteCh <- id return true } diff --git a/pkg/durableemitter/durable_emitter_test.go b/pkg/durableemitter/durable_emitter_test.go index 9d1af319be..64a35a9ce8 100644 --- a/pkg/durableemitter/durable_emitter_test.go +++ b/pkg/durableemitter/durable_emitter_test.go @@ -264,6 +264,50 @@ func TestDurableEmitter_MarkCoalescingFlushesPendingOnClose(t *testing.T) { assert.Equal(t, 0, store.Len(), "Close must flush buffered marks before returning") } +// timedOutBatchEmitter models batch.Client.Stop returning at its shutdown timeout: +// Stop returns without waiting, and the queued callbacks run afterwards. +type timedOutBatchEmitter struct { + mu sync.Mutex + callbacks []func(error) +} + +func (b *timedOutBatchEmitter) QueueMessage(_ *chipingress.CloudEventPb, cb func(error)) error { + b.mu.Lock() + defer b.mu.Unlock() + b.callbacks = append(b.callbacks, cb) + return nil +} + +func (b *timedOutBatchEmitter) Start(context.Context) {} +func (b *timedOutBatchEmitter) Stop() {} + +func (b *timedOutBatchEmitter) queued() []func(error) { + b.mu.Lock() + defer b.mu.Unlock() + return append([]func(error){}, b.callbacks...) +} + +func TestDurableEmitter_DeliveryCallbackAfterTimedOutStop(t *testing.T) { + store := NewMemDurableEventStore() + be := &timedOutBatchEmitter{} + + cfg := DefaultConfig() + cfg.DisablePruning = true + cfg.DeleteBatchSize = 100 + cfg.DeleteBatchWorkers = 1 + + em := newTestDurableEmitter(t, store, be, &cfg) + require.NoError(t, em.Start(t.Context())) + require.NoError(t, em.Emit(t.Context(), []byte("late"), testEmitAttrs()...)) + require.Eventually(t, func() bool { return len(be.queued()) == 1 }, 2*time.Second, 10*time.Millisecond) + + require.NoError(t, em.Close()) + + // The batch emitter reports success after DurableEmitter has shut down its delete coalescer. + assert.NotPanics(t, func() { be.queued()[0](nil) }, "late success callback must not send on the closed delete channel") + assert.Equal(t, 1, store.Len(), "the event stays persisted, so the next start retransmits it") +} + func TestDurableEmitter_MarkCoalescingDisabledMarksInline(t *testing.T) { store := &deleteRecordingStore{MemDurableEventStore: NewMemDurableEventStore()} be := newTestBatchEmitter() From 0ef48c536d34d123ab167a72a4c5e67b5620af20 Mon Sep 17 00:00:00 2001 From: Chet Sharma Date: Tue, 13 Oct 2026 14:59:49 -0700 Subject: [PATCH 2/2] durableemitter: bound inline deletes by PublishTimeout instead of stopCh stop closes stopCh before the batch emitter's final flush, so with coalescing off every event acked during shutdown had its delete cancelled and was re-sent on the next start. Inline deletes now use PublishTimeout, as the Config doc already describes. Also adds a test for success callbacks racing Close. --- pkg/durableemitter/durable_emitter.go | 16 +++- pkg/durableemitter/durable_emitter_test.go | 103 +++++++++++++++++++-- 2 files changed, 107 insertions(+), 12 deletions(-) diff --git a/pkg/durableemitter/durable_emitter.go b/pkg/durableemitter/durable_emitter.go index 8f71243670..ff9bc260ae 100644 --- a/pkg/durableemitter/durable_emitter.go +++ b/pkg/durableemitter/durable_emitter.go @@ -545,8 +545,12 @@ func (d *DurableEmitter) deliveryCallback(id int64, eventPb *chipingress.CloudEv return } + // Not derived from stopCh: stop closes it before the batch emitter's final + // flush, which would cancel the delete for every event acked during shutdown. + delCtx, delCancel := d.deleteContext() + defer delCancel() tDelete := time.Now() - deleted, deleteErr := d.store.BatchDelete(cbCtx, []int64{id}) + deleted, deleteErr := d.store.BatchDelete(delCtx, []int64{id}) deleteElapsed := time.Since(tDelete) if h := d.cfg.Hooks; h != nil && h.OnBatchDelete != nil { @@ -557,11 +561,19 @@ func (d *DurableEmitter) deliveryCallback(id int64, eventPb *chipingress.CloudEv return } if d.metrics != nil { - d.metrics.deliverComplete.Add(cbCtx, deleted) + d.metrics.deliverComplete.Add(delCtx, deleted) } } } +// deleteContext bounds an inline delivery delete by PublishTimeout. +func (d *DurableEmitter) deleteContext() (context.Context, context.CancelFunc) { + if d.cfg.PublishTimeout > 0 { + return context.WithTimeout(context.Background(), d.cfg.PublishTimeout) + } + return context.WithCancel(context.Background()) +} + // stop signals background loops to stop and waits for them to finish, then // stops the batch emitter (which flushes any queued events and waits for all // in-flight callbacks). It is invoked by the services.Engine when the embedded diff --git a/pkg/durableemitter/durable_emitter_test.go b/pkg/durableemitter/durable_emitter_test.go index 64a35a9ce8..462f1d77c7 100644 --- a/pkg/durableemitter/durable_emitter_test.go +++ b/pkg/durableemitter/durable_emitter_test.go @@ -264,24 +264,33 @@ func TestDurableEmitter_MarkCoalescingFlushesPendingOnClose(t *testing.T) { assert.Equal(t, 0, store.Len(), "Close must flush buffered marks before returning") } -// timedOutBatchEmitter models batch.Client.Stop returning at its shutdown timeout: -// Stop returns without waiting, and the queued callbacks run afterwards. -type timedOutBatchEmitter struct { - mu sync.Mutex - callbacks []func(error) +// heldBatchEmitter holds every callback until the test runs it. Stop either returns +// at once, like batch.Client.Stop hitting its shutdown timeout, or with flushOnStop +// runs the held callbacks with success first, like its final flush. +type heldBatchEmitter struct { + mu sync.Mutex + callbacks []func(error) + flushOnStop bool } -func (b *timedOutBatchEmitter) QueueMessage(_ *chipingress.CloudEventPb, cb func(error)) error { +func (b *heldBatchEmitter) QueueMessage(_ *chipingress.CloudEventPb, cb func(error)) error { b.mu.Lock() defer b.mu.Unlock() b.callbacks = append(b.callbacks, cb) return nil } -func (b *timedOutBatchEmitter) Start(context.Context) {} -func (b *timedOutBatchEmitter) Stop() {} +func (b *heldBatchEmitter) Start(context.Context) {} + +func (b *heldBatchEmitter) Stop() { + if b.flushOnStop { + for _, cb := range b.queued() { + cb(nil) + } + } +} -func (b *timedOutBatchEmitter) queued() []func(error) { +func (b *heldBatchEmitter) queued() []func(error) { b.mu.Lock() defer b.mu.Unlock() return append([]func(error){}, b.callbacks...) @@ -289,7 +298,7 @@ func (b *timedOutBatchEmitter) queued() []func(error) { func TestDurableEmitter_DeliveryCallbackAfterTimedOutStop(t *testing.T) { store := NewMemDurableEventStore() - be := &timedOutBatchEmitter{} + be := &heldBatchEmitter{} cfg := DefaultConfig() cfg.DisablePruning = true @@ -308,6 +317,80 @@ func TestDurableEmitter_DeliveryCallbackAfterTimedOutStop(t *testing.T) { assert.Equal(t, 1, store.Len(), "the event stays persisted, so the next start retransmits it") } +func TestDurableEmitter_DeliveryCallbacksRacingStop(t *testing.T) { + const n = 200 + store := NewMemDurableEventStore() + be := &heldBatchEmitter{} + + cfg := DefaultConfig() + cfg.DisablePruning = true + cfg.DeleteBatchSize = 100 + cfg.DeleteBatchWorkers = 1 + cfg.InsertBatchSize = 0 // insert inline so 200 Emits don't each wait out the coalescing linger + + em := newTestDurableEmitter(t, store, be, &cfg) + require.NoError(t, em.Start(t.Context())) + for range n { + require.NoError(t, em.Emit(t.Context(), []byte("race"), testEmitAttrs()...)) + } + require.Eventually(t, func() bool { return len(be.queued()) == n }, 5*time.Second, 10*time.Millisecond) + + var panics atomic.Int64 + var callbacks sync.WaitGroup + start := make(chan struct{}) + for _, cb := range be.queued() { + callbacks.Go(func() { + defer func() { + if recover() != nil { + panics.Add(1) + } + }() + <-start + cb(nil) + }) + } + close(start) + require.NoError(t, em.Close()) + callbacks.Wait() + + assert.Zero(t, panics.Load(), "success callbacks racing Close must not send on the closed delete channel") +} + +// slowDeleteStore makes BatchDelete take time and abort on cancellation, like a SQL store. +type slowDeleteStore struct { + *MemDurableEventStore +} + +func (s *slowDeleteStore) BatchDelete(ctx context.Context, ids []int64) (int64, error) { + select { + case <-ctx.Done(): + return 0, ctx.Err() + case <-time.After(20 * time.Millisecond): + } + return s.MemDurableEventStore.BatchDelete(ctx, ids) +} + +func TestDurableEmitter_InlineDeletesSurviveShutdownFlush(t *testing.T) { + const n = 3 + store := &slowDeleteStore{MemDurableEventStore: NewMemDurableEventStore()} + be := &heldBatchEmitter{flushOnStop: true} + + cfg := DefaultConfig() + cfg.DisablePruning = true + cfg.DeleteBatchSize = 0 // inline deletes + + em := newTestDurableEmitter(t, store, be, &cfg) + require.NoError(t, em.Start(t.Context())) + for range n { + require.NoError(t, em.Emit(t.Context(), []byte("ack-during-stop"), testEmitAttrs()...)) + } + require.Eventually(t, func() bool { return len(be.queued()) == n }, 2*time.Second, 10*time.Millisecond) + + require.NoError(t, em.Close()) // the final flush acks every held event + + assert.Zero(t, store.Len(), "events acked during shutdown must be deleted, not re-sent on the next start") +} + func TestDurableEmitter_MarkCoalescingDisabledMarksInline(t *testing.T) { store := &deleteRecordingStore{MemDurableEventStore: NewMemDurableEventStore()} be := newTestBatchEmitter()