Repository navigation
chipingress/batch: publish queued messages on Stop instead of discarding them - #2461
Conversation
…ing them On Stop the batcher flushed only the batch it was filling. Messages still queued on the buffer channel, typically when sends are saturated, were abandoned: never published and their callbacks never ran, so callers could not count them as dropped or failed. The batcher now drains everything already queued, in batches of at most batchSize, before returning. Stop's shutdown timeout still bounds the wait; batches drained after it expires fail against the closed client and report the error to their callbacks.
|
👋 skippaDaBitFlippa, thanks for creating this pull request! To help reviewers, please consider creating future PRs as drafts first. This allows you to self-review and make any final changes before notifying the team. Once you're ready, you can mark it as "Ready for review" to request feedback. Thanks! |
With the pinned chainlink-common batch client, Stop flushes only the batch being filled; messages queued behind in-flight sends are discarded without a callback (found in local load testing). Correct the Close and ChipIngressDrainTimeout docs. Fix upstream: smartcontractkit/chainlink-common#2461.
📊 API Diff Results
|
There was a problem hiding this comment.
🟡 Changes recommended
Concurrent enqueues can still be accepted after draining finishes, leaving messages and callbacks abandoned.
2 open findings
What changed in this PR
Updates batch shutdown to drain queued messages instead of silently abandoning them.
Changes:
- Drains buffered messages in size-limited batches during shutdown.
- Documents shutdown and timeout behavior.
- Adds a regression test for queued messages behind an in-flight send.
| File | Description |
|---|---|
pkg/chipingress/batch/client.go |
Documents the expanded shutdown guarantees. |
pkg/chipingress/batch/client_test.go |
Tests successful queued-message draining. |
pkg/chipingress/batch/batcher.go |
Implements draining on cancellation. |
🧠 Review effort: Balanced
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
| for len(input) > 0 { | ||
| batch = append(batch, <-input) |
There was a problem hiding this comment.
Fixed in d8a7031. QueueMessage now does its final stop check and the enqueue under an admission RWMutex read lock, and Stop closes stopCh under the write lock. Every message QueueMessage accepted is therefore on the buffer before the batcher drains it, and anything arriving afterwards gets ErrClientShutdown. New test: TestStop/every_message_accepted_while_Stop_races_producers_gets_a_callback.
| }, time.Second, time.Millisecond) | ||
| close(release) |
There was a problem hiding this comment.
Added TestStop/Stop_returns_at_the_shutdown_timeout_and_late_batches_report_errors in d8a7031. It holds the only send slot past a 100ms shutdown timeout, asserts Stop returns within the bound and closes the client, and asserts every callback fires: the 2 messages in flight before the timeout succeed and the other 18 get the closed-client error.
QueueMessage checked stopCh and enqueued later, so a producer could add a message after the batcher had drained and returned; that message was never published and its callback never ran. An admission RWMutex now makes the check-and-enqueue atomic with Stop closing stopCh. Add tests for the shutdown-timeout path (Stop returns at the bound, closes the client, late batches report errors to every callback) and for producers racing Stop (every accepted message gets a callback).
|
End-to-end check in the real consumer path: the chainlink node's legacy telemetry client (smartcontractkit/chainlink#23949) sending to a local chip-ingress + redpanda (atlas chip-e2e dev stack). Chainlink was pointed at this branch with a temporary
|
pkcll
left a comment
pkcll
left a comment
There was a problem hiding this comment.
PR Review: #2461 — chipingress/batch: publish queued messages on Stop instead of discarding them
Summary
This fixes a real data-loss bug. Before, batchWithInterval handled ctx.Done() by flushing only the batch it was building, so everything still in messageBuffer was dropped. Those messages had already been accepted by QueueMessage with a nil return, and their callbacks never ran. I reproduced this independently: 2,000 queued, ~200 sent, ~1,800 dropped without callbacks.
The fix is small and well placed:
drain()empties the buffer in batches of at mostbatchSize.- An admission
RWMutexmakes "checkstopCh, then enqueue" atomic with respect toStopclosingstopCh, so nothing can be accepted after the drain has run. - Behavior after the shutdown timeout is documented and tested.
Ran locally on d8a7031d5:
go test ./batch/ -race -count=30 -run TestStop✅- full batch suite
-race -count=5✅
Jira Context
N/A. Context: chainlink#23949 (legacy-telemetry load test).
Files Reviewed
pkg/chipingress/batch/batcher.go, pkg/chipingress/batch/client.go, pkg/chipingress/batch/client_test.go
Findings
Critical
None.
Major
None.
Minor
m1. No test fails if the admission lock is removed.
I reverted the drain while keeping the lock, and all three new tests fail (3/3), so the drain is well guarded. I then removed the lock and its re-check while keeping the drain. every message accepted while Stop races producers gets a callback still passed 100/100: the window between the fast-path stopCh check and the buffer send is too narrow to hit by chance.
If you're fine with a test seam, a deterministic version is possible: a test-only hook, nil in production, called right after the fast-path check:
// client.go
var testHookAfterFastPath func() // nil in production
// QueueMessage, after the fast-path stopCh check:
if testHookAfterFastPath != nil {
testHookAfterFastPath()
}
// test: hold one producer at the hook until Stop has returned, then let it continue.
// With the fix it gets ErrClientShutdown. Without the re-check it gets nil and its
// callback never runs.If you'd rather not add a seam, a sentence in the PR saying the lock is covered by reasoning rather than a test is fine too.
Nitpicks
drainhas no direct unit test.drain()is now a small pure generic function, so a direct test would pin down chunking (≤batchSize) and ordering, which the client-level tests don't assert. For example, pre-filledchan int(7 items), initial batch[-1],batchSize3 ⇒[[-1 0 1] [2 3 4] [5 6]].len(input) > 0relies on the batcher beinginput's only reader. The comment says so, and it's true today. A non-blockingselect { case msg := <-input: … default: return }would hold even if a second reader is ever added, at no extra cost.- Log noise after the shutdown timeout. Every batch drained after
Close()logsfailed to publish batchat Error level. With the durable emitter's defaults (10k buffer, batch 50), that's up to ~200 error lines per slow shutdown. Consider one summary line, or Warn level, oncestopChis closed. - Exactly-once is checked in aggregate only. The tests compare total callbacks with accepted messages, so a double callback for one message plus a missing one for another would cancel out. A per-message counter would catch both.
Questions
None blocking.
Heads-ups (outside this PR)
- Durable emitter shutdown.
DurableEmitter.stop()closesdeleteChright afterbatchEmitter.Stop()returns (durable_emitter.go~L576-582), and a success callback does a blockingd.deleteCh <- id. IfStopreturns on the shutdown timeout while success callbacks are still running, or blocked on a fulldeleteCh, a late callback panics with send on closed channel. In-flight RPCs are cancelled byClose(), which narrows the window but doesn't close it. Draining makes long shutdowns more likely. The durable emitter guard should land with or before the root-module bump that picks this PR up. No change is needed in this PR. - #2436 (draft) adds an
onDequeuehook tobatchWithInterval. Whichever merges second needs to call it fromdrain()as well, or live-byte accounting misses drained messages. TestCallbacks/callbacks invoked on stop(still skipped onmain) has atime.Sleep(10ms)beforeStop()that worked around this race. It can go once that test is re-enabled.
Positive Notes
- Tight diff. The drain sits exactly where the bug was, and the admission lock is minimal: it's held only across a non-blocking check and send.
- Behavior after the shutdown timeout is defined: late batches fail against the closed client and still report to their callbacks. It's documented on
Stopand tested. - The tests use per-event
PublishResults that match the strict results contract, and each has a clear before/after story in the description.
Recommendation
Approve with minor changes. m1 is optional but recommended; the nits are optional. Fine to merge as is if m1 is consciously deferred.
The producers-vs-Stop stress test passed with the admission lock removed: the window between the fast-path stop check and the enqueue is too narrow to hit by chance. Replace it with a test that parks a producer in that window through a per-client test hook (nil in production), runs Stop, then resumes the producer and requires ErrClientShutdown. It fails without the admission re-check.
|
|
|
||
| // drain processes batch plus everything already buffered on input, without waiting for new items. | ||
| // The batcher is input's only reader, so a non-zero len guarantees the receive won't block. | ||
| func drain[T any](input <-chan T, batch []T, batchSize int, processFn func([]T)) { |
There was a problem hiding this comment.
nit: maybe this alternative is a bit clearer:
for {
select {
case msg := <-input:
batch = append(batch, msg)
if len(batch) >= batchSize {
processFn(batch)
batch = nil
}
default:
if len(batch) > 0 {
processFn(batch)
}
return
}
}
batch.Client.Stopflushed only the batch it was filling. Anything still queued on the buffer channel, typically when sends are saturated, was abandoned: never published, and its callback never ran, so callers couldn't count it as dropped or failed.The batcher now drains everything already queued, in batches of at most
batchSize, before returning.WithShutdownTimeoutstill boundsStop. Batches drained after the timeout fail against the closed client and report the error to their callbacks.Found while load-testing the node's legacy-telemetry path (smartcontractkit/chainlink#23949) against a local chip-ingress: a 100k-message burst lost exactly the buffer's contents on shutdown, with no metric recording it. Beholder's chip batch emitter uses this client too.
New test
TestStop/Stop_publishes_messages_still_queued_behind_in-flight_sendsholds the only send slot while 100 messages are queued, then calls Stop:-race)