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

[skippaDaBitFlippa]

stop closed deleteCh as soon as batchEmitter.Stop returned. batch.Client.Stop can return at its shutdown timeout with delivery callbacks still running, so a late success callback hit send on closed channel and panicked. #2461 drains the queue on Stop, which makes long shutdowns more likely, so this should land before the root module picks that up.

deleteCh is now closed under a lock, and enqueueDelete sends under the read lock. A callback that arrives after the close skips the delete. The event stays persisted and is retransmitted on the next start, same at-least-once guarantee as before. The inline-delete path (coalescing disabled) already failed harmlessly on the cancelled context, so it's unchanged.

TestDurableEmitter_DeliveryCallbackAfterTimedOutStop runs a queued success callback after Close(). It panics on main and passes 10/10 with -race here. Full pkg/durableemitter suite passes with -race.

Flagged by @pkcll in the #2461 review.

[github-actions]

📊 API Diff Results

No changes detected for module github.com/smartcontractkit/chainlink-common

View full report

[Copilot]

🟢 Approval recommended

The locking correctly prevents sends on the closed channel while preserving at-least-once delivery semantics.

0 open findings

What changed in this PR

Prevents late delivery callbacks from panicking after durable emitter shutdown.

Changes:

• Synchronizes delete-channel closure and callback sends.
• Adds regression coverage for callbacks arriving after timed-out shutdown.

File	Description
pkg/​durableemitter/​durable_emitter.go	Safely handles late delete callbacks.
pkg/​durableemitter/​durable_emitter_test.go	Tests callback execution after close.

🧠 Review effort: Balanced

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[pkcll]

The fix for the shutdown panic looks right. deleteCh is closed under deleteMu after deleteClosed is set, and enqueueDelete sends under the read lock, so a success callback that outlives batchEmitter.Stop() no longer sends on a closed channel. Holding the read lock across the blocking send is safe because the workers keep draining until the close.

Ran locally on c0f89a313:

• the new test -race -count=10 and the full pkg/durableemitter suite with -race both pass;
• 200 held success callbacks fired concurrently with em.Close(): 0 panics here vs 52/200 on main.

The only red CI check is benchmark (BenchmarkKeystore_Sign), which is unrelated.

A few edge cases worth considering:

1. Inline deletes during the final flush are cancelled (DeleteBatchSize <= 0)

The description says the inline path "already failed harmlessly on the cancelled context". That path matters for every callback that acks during batchEmitter.Stop(), not only the late ones:

• stop() closes stopCh before batchEmitter.Stop(), and the inline delete uses cbCtx from d.stopCh.NewCtx().
• Once #2461 drains the queue on Stop, the final flush can deliver everything that was buffered, and each of those deletes is cancelled.
• So every event delivered during shutdown stays in the store and is re-sent on the next start: a duplicate for every event delivered while shutting down.

Reproduced with a store whose BatchDelete behaves like SQL: it takes about 20ms and aborts on ctx.Done(). 3 events acked during Stop leave 3/3 rows on this branch (and on main). That context also never applies PublishTimeout, although the Config doc calls it "the deadline for DB operations in delivery callbacks (BatchDelete)".

It's a non-default config, so not blocking. A small fix that mirrors flushDeletes:

// deliveryCallback, inline path
delCtx, delCancel := d.deleteContext()
defer delCancel()
deleted, deleteErr := d.store.BatchDelete(delCtx, []int64{id})

// deleteContext bounds an inline delete by PublishTimeout. It is not derived from
// stopCh: callbacks also run while stop waits for the batch emitter to flush.
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())
}

Test recipe:

• a BatchEmitter whose Stop() runs the held callbacks with nil;
• cfg.DeleteBatchSize = 0;
• a store that selects on ctx.Done() vs a 20ms timer;
• assert store.Len() == 0 after Close().

2. The concurrent path isn't exercised

The test runs a single callback after Close() has returned, so the "closing under deleteMu waits for in-progress sends" path never runs. A test that fires ~200 held success callbacks concurrently with em.Close() (recovering and counting panics, under -race) would cover it. That's the run that shows 52/200 panics on main and 0 here.

3. Late successes become silent duplicates

A delete skipped after close guarantees that event is re-sent on the next start, but it's logged at Debug. Consider Info or Warn with a count at shutdown, or a counter, so the duplicates can be explained after the fact.

4. Document the late-success policy

Leaving late successes persisted keeps Close free of store side effects, at the cost of one duplicate per late callback. That's a reasonable trade-off. A sentence in the DurableEmitter doc would help, since callers may assume Close leaves no acked rows behind.

Overall: looks good. #1 is the one I'd fix (or explicitly defer); the rest are optional.

[pkcll]

@DylanTinianov @patrickhuie19, could you confirm duplicates are acceptable
see #2465 (review)
" Inline deletes during the final flush are cancelled" ... "So every event delivered during shutdown stays in the store and is re-sent on the next start: a duplicate for every event delivered while shutting down."

[skippaDaBitFlippa]

@pkcll fixed #1 and #2 in 0ef48c5.

#1: inline deletes and their deliverComplete metric now use deleteContext(), which is context.Background() bounded by PublishTimeout, the same pattern as flushDeletes. TestDurableEmitter_InlineDeletesSurviveShutdownFlush follows your recipe: Stop acks the held events, coalescing is off, and the store aborts a 20ms BatchDelete on ctx.Done(). It leaves 3/3 rows on the previous commit and 0 now.

#2: TestDurableEmitter_DeliveryCallbacksRacingStop fires 200 held success callbacks concurrently with Close() and counts recovered panics. It sees 194-200/200 on main and 0 here, 10/10 with -race. Insert coalescing is off in that test so it runs in ~10ms instead of ~10s.

#3 and #4 are not in this PR yet. I can add a one-time Warn on the first late success plus the doc sentence if you think they belong here.

[patrickhuie19]

@DylanTinianov @patrickhuie19, could you confirm duplicates are acceptable see #2465 (review) " Inline deletes during the final flush are cancelled" ... "So every event delivered during shutdown stays in the store and is re-sent on the next start: a duplicate for every event delivered while shutting down."

This is fine, we do not expect exactly once delivery. we do expect at least once