[adapters] Have output endpoints report their own checkpoint totals - #6789
Merged
Conversation
mythical-fred
approved these changes
Aug 2, 2026
mythical-fred
left a comment
There was a problem hiding this comment.
Clean fix. Moving the reading from an approximation (transmitted + buffered + queued, which double-counted while the endpoint thread bumped transmitted_records before releasing the batch) to a latch armed by the circuit thread and populated by the endpoint the moment it crosses the threshold is the right shape for this. The idle-endpoint arming-time shortcut and the removed-endpoint no-report path are exactly the edge cases I would have asked about, and the concurrency test that walks the arm point across 16 offsets over 400 batches is convincing. Approving.
One doc-comment typo noted inline.
A checkpoint needs each output endpoint's `transmitted_records` as of the
moment the endpoint has transmitted the output the checkpoint covers, since
that is the state the pipeline resumes from. It cannot read that itself: the
circuit thread takes the checkpoint while the endpoints are still working, and
it must not block waiting for them. So it guessed using this formula:
```rust
transmitted_records: snapshot.transmitted_records
+ snapshot.buffered_records
+ snapshot.queued_records,
```
There are several issues with this:
1. The second and third terms are the number of records output by the circuit,
not the number of records send by the connector. They can have non-unit
multiplicities and canceling out weights, making them an approximation
to what the connector will output by the time the checkpoint completes
2. There was a window of double counting when the same records were accounted
by both `queued_records` and `transmitted_records`. The endpoint thread
raises `transmitted_records` from inside `push_batch_to_encoder`, while
the batch those records came from stays in `queued_records` until `output_batch`
releases it afterwards.
3. This aside, each counter is implemented as a separate atomic variable,
meaning that they can be in an inconsistent state for short periods of time.
The second issue above caused the suspend_barrier4 test to fail.
Having tried several solutions, I concluded that the only correct way to do this
is to have the endpoint report the actual transmitted number of records when
it's finished processing records associated with the checkpoint, which is what
this commit implements.
As a bonums, `transmitted_bytes` can now be tracked as well, and no longer has
to be left out of the checkpoint.
Signed-off-by: Leonid Ryzhyk <ryzhyk@gmail.com>
ryzhyk
force-pushed
the
fix-checkpoint-output-stats-double-count
branch
from
August 2, 2026 16:07
68cef71 to
4cd9584
Compare
blp
approved these changes
Aug 2, 2026
blp
left a comment
Member
There was a problem hiding this comment.
I really don't like how complicated everything is getting. It's at the point where I can't see bugs because things are so complicated.
I don't see a bug in this.
Thanks for the fix.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
A checkpoint needs each output endpoint's
transmitted_recordsas of the moment the endpoint has transmitted the output the checkpoint covers, since that is the state the pipeline resumes from. It cannot read that itself: the circuit thread takes the checkpoint while the endpoints are still working, and it must not block waiting for them. So it guessed using this formula:There are several issues with this:
queued_recordsandtransmitted_records. The endpoint thread raisestransmitted_recordsfrom insidepush_batch_to_encoder, while the batch those records came from stays inqueued_recordsuntiloutput_batchreleases it afterwards.The second issue above caused the suspend_barrier4 test to fail.
Having tried several solutions, I concluded that the only correct way to do this is to have the endpoint report the actual transmitted number of records when it's finished processing records associated with the checkpoint, which is what this commit implements.
As a bonums,
transmitted_bytescan now be tracked as well, and no longer has to be left out of the checkpoint.Describe Manual Test Plan
Checklist
Breaking Changes?
Mark if you think the answer is yes for any of these components:
Describe Incompatible Changes