Skip to content

fix(processor): export the last partial batch on shutdown - #924

Merged
samcm merged 3 commits into
masterfrom
fix/batch-processor-shutdown-flush
Oct 20, 2026
Merged

samcm merged 3 commits into
masterfrom
fix/batch-processor-shutdown-flush

Conversation

@samcm

@samcm samcm commented Oct 20, 2026 •

Copy link
Copy Markdown
Member

BatchItemProcessor.Shutdown lost the last partial batch, so every component using it dropped whatever it held since the last flush on each shutdown (sinks flush every 5s by default). Shutdown moved the queue into the batch builder's in-memory batch, then closed the queue and the workers' stop channel back to back. The builder's final send raced the workers exiting, and that batch was never exported.

Fix

  • Shutdown closes the queue under a lock that Write holds while sending. A write can no longer race the close, so the "send on closed channel" recover is gone.
  • Once the queue is closed, the batch builder sends its last batch and closes the batch channel. Workers keep exporting until that channel is drained, then exit. Shutdown waits for them and then shuts the exporter down.
  • Exports ignore the start context's cancellation, so a component whose context ends before Stop still flushes. ExportTimeout bounds each export, and the Shutdown context bounds the wait.

Callers. These components use the processor through their sinks, plus the discovery coordinator client, the server's node-record coordinator and mimicry's execution peers:

Component Calls Stop on shutdown
server yes, after a gRPC graceful stop
discovery yes
relay monitor yes
mimicry yes (its execution peers' transaction-hash queues are not drained, but those hold hashes still to be fetched, not events)
sentry fixed in #922
cannon no: fixed here

Cannon had the same bug as the sentry. BeaconNode.Start blocks until the beacon node fails, so the signal handler after it was never registered: SIGTERM killed the cannon without flushing its sinks. Cannon now watches the signal alongside the beacon node. On shutdown it cancels the beacon node's context to stop the derivers, stops the scheduler, then flushes the sinks. It waits up to 10s for the beacon node to return first and reports its error, and a deriver cannot start once shutdown has begun. Failures are logged and joined instead of returning early, and the whole sequence is bounded below the chart's 90s terminationGracePeriodSeconds.

Tests. New tests cover:

  • a queued partial batch exported by Shutdown, 20 runs;
  • a flush after the start context is cancelled;
  • every accepted write exported while concurrent writers race Shutdown;
  • a queued sync item completed by Shutdown;
  • the HTTP sink delivering three queued events on Stop, with a 1h batch timeout.

The new processor and HTTP sink tests fail on master and pass here, with -race.

Release note: sinks now export everything still queued when a component stops. Before, sentry, cannon, server, mimicry, relay monitor and discovery dropped up to one batch timeout of events on every shutdown. The cannon now shuts down cleanly on SIGTERM.

@redpandabot redpandabot Bot left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Summary

Found 1 issue. Review scope is incomplete; these findings do not constitute a clean verdict.

Issues

  • 🟡 pkg/cannon/cannon.go:191 — Signal path discards the beacon node's error after stopBeacon(), making beacon shutdown failures invisible to the caller — see the thread on that line

Reviewed @ a9872585
"If it's stupid but it works, it isn't stupid." — Murphy's Laws of Combat

Comment thread pkg/cannon/cannon.go
beaconErr := make(chan error, 1)

go func() {
beaconErr <- c.beacon.Start(beaconCtx)

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 Signal path discards the beacon node's error after stopBeacon(), making beacon shutdown failures invisible to the caller

Start launches c.beacon.Start(beaconCtx) in a goroutine writing to a 1-buffered beaconErr channel, then selects once. On the signal branch it calls stopBeacon() and immediately returns c.Shutdown(ctx) without ever reading beaconErr. Because BeaconNode.Start returns ctx.Err() on cancellation, cancelling beaconCtx always produces a value the buffered send silently absorbs; any real beacon-side error during the shutdown window is equally lost.

SIGTERM/SIGINT reaches the signals case. stopBeacon() cancels beaconCtx; the beacon goroutine returns (ctx.Err() from BeaconNode.Start's case <-ctx.Done(): return ctx.Err(), or an error from an in-flight service start / onReady callback) and sends it into the buffered beaconErr. Start returns c.Shutdown(ctx) and never reads beaconErr.

The beacon node's reported error never reaches cmd/cannon.go, which does if err := cannon.Start(...); err != nil { log.Fatal(err) }. A beacon-node failure that surfaces during shutdown (including a genuine startup/subscription/onReady-callback failure racing the signal) yields a nil-ish Start result and, if Shutdown succeeds, a clean exit. Additionally, Start is not joined before Shutdown cancels and tears down derivers/sinks while the beacon goroutine may still be running, so ordering between beacon teardown and deriver/sink teardown is unmanaged.

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed in 8d02879: after a signal, Start cancels the beacon node and waits for its result (bounded to 10s) before calling Shutdown, so deriver and sink teardown no longer overlaps it. A non-cancellation beacon error is joined with Shutdown's error and reaches cmd/cannon.go.

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Confirmed — addressed by 8d02879, which is the PR head. Reading the code now at pkg/cannon/cannon.go:201-219:

  • After the signal branch calls stopBeacon() (line 201), Start no longer goes straight to Shutdown. It first joins the beacon goroutine: select { case err := <-beaconErr: ... case <-time.After(beaconStopTimeout): ... } (lines 206-212).
  • A non-cancellation beacon error is wrapped (fmt.Errorf("beacon node: %w", err), line 209) and joined with Shutdown's error via errors.Join(errs...) (line 219). The context.Canceled filter is correct — BeaconNode.Start returns bare ctx.Err() on cancellation (pkg/cannon/ethereum/beacon.go:302-305), while real failures (service start, network detection, onReady callbacks) are wrapped errors that still surface.
  • The join happens before c.Shutdown(ctx) (line 214), so deriver/sink teardown no longer overlaps the beacon goroutine in the normal path.
  • The error does reach the caller: cmd/cannon.go:156-158 does log.Fatal(err) on a non-nil Start result.

One residual nuance, not blocking: if the beacon goroutine doesn't return within beaconStopTimeout (10s, line 224), Start proceeds to Shutdown anyway, so teardown overlap is still possible on the timeout path — but that's the deliberate bounded wait (comment at lines 223-224 ties the 10s + 75s budget to the 90s terminationGracePeriodSeconds), and the buffered channel absorbs the late send without leaking the goroutine.

So the original finding no longer holds at this head.


@ 8d02879d

samcm added 2 commits October 20, 2026 21:20
Shutdown drained the queue into the batch builder's in-memory batch, then
closed the queue and the workers' stop channel back to back, so the final
partial batch was never exported: every sink lost what it held since its
last flush on each shutdown.

- Shutdown closes the queue under a lock that writers hold while sending, so
  no write can race the close; the batch builder sends its last batch and
  closes the batch channel, and workers export every batch before exiting.
- Exports ignore the start context's cancellation, so a component whose
  context ends first still flushes; ExportTimeout bounds each export and the
  Shutdown context bounds the wait.
- cannon: BeaconNode.Start blocks until the beacon node fails, so SIGTERM was
  never caught and the sinks were never flushed. Watch the signal alongside
  it, cancel the beacon node to stop the derivers, then flush the sinks,
  logging failures and bounded below the chart's 90s grace period.
On a signal, Start cancelled the beacon node and went straight to Shutdown
without reading its result, so a real beacon failure racing the signal was
lost and deriver/sink teardown could overlap the beacon goroutine. Start now
waits (bounded to 10s) for the beacon node to return, joins any
non-cancellation error with Shutdown's, and only then tears down.
@samcm
samcm force-pushed the fix/batch-processor-shutdown-flush branch from a987258 to 8d02879 Compare October 20, 2026 11:23

@redpandabot redpandabot Bot left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Summary

The PR makes BatchItemProcessor.Shutdown flush the last partial batch: the queue close is serialized against writers via queueMu with a stopCh check before every send, the builder drains and closes batchCh, and workers export until it's empty — the removed send-on-closed-channel recover is genuinely obsolete. The cannon now joins the beacon node and surfaces its error, fixing the previously reported discarded-error finding. One narrow shutdown-during-startup race in the cannon remains worth a look.

Issues

  • 🟡 pkg/cannon/cannon.go:207 — Shutdown can race the beacon's OnReady callback goroutine — see the thread on that line

Reviewed @ 8d02879d
"If it's stupid but it works, it isn't stupid." — Murphy's Laws of Combat

Comment thread pkg/cannon/cannon.go
c.log.WithContext(ctx).Printf("Caught signal: %v", sig)
// Join the beacon node before the derivers and sinks it feeds are torn down.
select {
case err := <-beaconErr:

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 Shutdown can race the beacon's OnReady callback goroutine

beaconErr is delivered when BeaconNode.Start returns, but the goroutine inside it that runs the OnReady callbacks is never joined: it assigns c.eventDerivers (cannon.go:1219) and spawns startDeriverWhenReady goroutines that log.Fatal on failure (cannon.go:1240). If the inner go-eth2-client Start doesn't return promptly on cancellation (or the 10s timeout fires), a SIGTERM during beacon startup has Shutdown reading c.eventDerivers concurrently with the callback writing it (data race) and can start derivers into sinks being stopped — a deriver start failure would Fatal-exit the process before the flush. Untraced: go-eth2-client's Start behavior on cancellation (module cache unavailable); the window requires shutdown during the beacon startup window, which is why this is a concern rather than a blocker.

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed in 0704280: a mutex guards eventDerivers, Shutdown marks the cannon as stopping so a late OnReady callback starts nothing, and a deriver start that fails because the beacon context was cancelled returns instead of calling Fatal.

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

...


@ 07042800

The beacon's OnReady callback assigns eventDerivers and starts them while a
SIGTERM during beacon startup can already have Shutdown reading that slice.
A mutex now guards the slice, Shutdown marks the cannon as stopping so a late
callback starts nothing, and a deriver start that fails because the beacon
context was cancelled no longer exits the process before the sinks flush.

@redpandabot redpandabot Bot left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The PR makes BatchItemProcessor.Shutdown flush the last partial batch by closing the queue under a lock that Write holds, draining batchCh via the workers, and exporting on a cancellation-immune context; cannon now watches SIGTERM alongside the beacon node and flushes sinks on shutdown. The latest commit fixes both previously open cannon findings: the beacon's stop error is now joined into the returned error, and the OnReady/Shutdown race is guarded by deriversMu plus a stopping flag. The locking invariants hold on every exit path I traced; no new problems found.


Reviewed 4 changed file(s) @ 07042800 — no blocking issues found.
"If it's stupid but it works, it isn't stupid." — Murphy's Laws of Combat

@samcm
samcm merged commit 719b24d into master Oct 20, 2026
11 checks passed
@samcm
samcm deleted the fix/batch-processor-shutdown-flush branch October 20, 2026 11:54
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant