Repository navigation
Producer: Split produce function into fetch/process & Proper Handling for Incomplete Error - #2636
Merged
Merged
Conversation
Change Producer API to split the existing method produce_next() into two methods: * Fetch to read data from data source which mush be cancel safe * Process to parse the data which doesn't require cancel safety The changes still preserve the existing method as it's used in multiple other places.
`write_from()` truncates silently when the buffer is full, but the process and serial sources reported the length of the received line instead. On a line longer than the 8192-byte buffer that makes the producer believe bytes arrived which never did.
* Parsers hand back the bytes they still hold through `parse_remaining()`, so a stalled source no longer erodes a pending item one byte per parse call, and text lines stop splitting at reader buffer boundaries. * `can_buffer_more()` tells a full buffer from a merely paused source, which is the only case where a remainder must be resolved right away. Fixes for additionally discovered bugs: * The CLI socket session rewrote every message it had ever collected on each round. * The export read-to-end loop never drained its records and counted rounds instead of messages.
Fixes came from new rust version and not from the changes in this branch
AmmarAbouZor
force-pushed
the
split-producer-and-fix
branch
from
September 1, 2026 07:49
0256289 to
8be26af
Compare
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
1 participant
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.
This PR closes #2413
This PR closes #2635
It splits
Producer::produce_next()into a cancel-safefetch()and aprocess()step, keeping the original method forexisting call sites.
And also it fixes silent data loss when a source (process/serial) receives a line longer than the internal buffer, which led to corruption of pending items and incorrect line splitting when a source stalls mid-read.
producer believed more data had arrived than write_from()'s silent truncation allowed.
item byte by byte and text lines no longer split at reader buffer boundaries.
actually needed.
It also includes additional fixes for multiple issues found in the refactoring: