durableemitter: don't panic when a delivery callback runs after stop

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
@@ -540,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 {
@@ -552,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
@@ -574,10 +591,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 +664,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
 }

pkg/durableemitter/durable_emitter_test.go

@@ -264,6 +264,133 @@ func TestDurableEmitter_MarkCoalescingFlushesPendingOnClose(t *testing.T) {
 	assert.Equal(t, 0, store.Len(), "Close must flush buffered marks before returning")
 }
 
+// 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 *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 *heldBatchEmitter) Start(context.Context) {}
+
+func (b *heldBatchEmitter) Stop() {
+	if b.flushOnStop {
+		for _, cb := range b.queued() {
+			cb(nil)
+		}
+	}
+}
+
+func (b *heldBatchEmitter) 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 := &heldBatchEmitter{}
+
+	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_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()