diff --git a/pkg/durableemitter/durable_emitter.go b/pkg/durableemitter/durable_emitter.go index 225adb6c10..ff9bc260ae 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 @@ -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 } diff --git a/pkg/durableemitter/durable_emitter_test.go b/pkg/durableemitter/durable_emitter_test.go index 9d1af319be..462f1d77c7 100644 --- a/pkg/durableemitter/durable_emitter_test.go +++ b/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()