Be more deffensive with the StreamSubscriber internal state
#3545
+21
−22
Conversation
This file contains 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
BalmungSan
force-pushed
the
fix-flow-stream-subscriber
branch
from
ba25fd6 to
bdb2835
Compare
BalmungSan
commented
Mar 10, 2025
Comment on lines
-69
to
+71
| private var inOnNextLoop: Boolean = _ | ||
| private var inOnNextLoop: Boolean = false | ||
| private var buffer: Array[Any] = null | ||
| private var index: Int = _ | ||
| private var index: Int = 0 |
There was a problem hiding this comment.
AFAIK, these are equivalent, but just in case, I made sure inOnNextLoop starts as false, the index shouldn't matter but just for consistency; I also thought of using -1 but that feels confusing.
Comment on lines
-191
to
+194
| cb.apply(Right(Some(Chunk.array(buffer)))) | ||
| val chunk = Chunk.array(buffer) | ||
| inOnNextLoop = false | ||
| buffer = null | ||
| cb.apply(Right(Some(chunk))) |
There was a problem hiding this comment.
Reset the state BEFORE publishing the Chunk.
Comment on lines
-217
to
+222
| cb.apply(Left(ex)) | ||
| // We do the updates here, | ||
| // to ensure they happen after we have secured the state. | ||
| inOnNextLoop = false | ||
| buffer = null | ||
| cb.apply(Left(ex)) |
There was a problem hiding this comment.
Reset the state BEFORE publishing the Exception.
Comment on lines
248
to
263
| // We do the updates here, | ||
| // to ensure they happen after we have secured the state. | ||
| if (canceled) { | ||
| s.cancel() | ||
| inOnNextLoop = false | ||
| buffer = null | ||
| cb.apply(Right(None)) | ||
| } else if (index == 0) { | ||
| inOnNextLoop = false | ||
| buffer = null | ||
| cb.apply(Right(None)) | ||
| } else { | ||
| cb.apply(Right(Some(Chunk.array(buffer, offset = 0, length = index)))) | ||
| // We do the updates here, | ||
| // to ensure they happen after we have secured the state. | ||
| val chunk = Chunk.array(buffer, offset = 0, length = index) | ||
| inOnNextLoop = false | ||
| buffer = null | ||
| cb.apply(Right(Some(chunk))) |
There was a problem hiding this comment.
Make sure we reset the state BEFORE anything else.
Comment on lines
-286
to
+307
| s.cancel() | ||
| val ex = Left(new InvalidStateException(operation = "Received request", state)) | ||
| otherCB.apply(ex) | ||
| cb.apply(ex) | ||
| // We do the updates here, | ||
| // to ensure they happen after we have secured the state. | ||
| inOnNextLoop = false | ||
| buffer = null | ||
| s.cancel() | ||
| val ex = Left(new InvalidStateException(operation = "Received request", state)) | ||
| otherCB.apply(ex) | ||
| cb.apply(ex) |
There was a problem hiding this comment.
Reset the state BEFORE publishing the Exception.
BalmungSan
commented
Mar 12, 2025
Comment on lines
-209
to
-212
| // We do the updates here, | ||
| // to ensure they happen after we have secured the state. | ||
| inOnNextLoop = false | ||
| buffer = null |
Comment on lines
+242
to
+256
| // We do the updates here, | ||
| // to ensure they happen after we have secured the state. | ||
| if (canceled) { | ||
| s.cancel() | ||
| inOnNextLoop = false | ||
| buffer = null | ||
| cb.apply(Right(None)) | ||
| } else if (index == 0) { | ||
| } else if (buffer eq null) { | ||
| inOnNextLoop = false | ||
| cb.apply(Right(None)) | ||
| } else { | ||
| cb.apply(Right(Some(Chunk.array(buffer, offset = 0, length = index)))) | ||
| // We do the updates here, | ||
| // to ensure they happen after we have secured the state. | ||
| val chunk = Chunk.array(buffer, offset = 0, length = index) | ||
| inOnNextLoop = false | ||
| buffer = null | ||
| cb.apply(Right(Some(chunk))) |
There was a problem hiding this comment.
Reset the state BEFORE publishing the Chunk (or lack of).
Comment on lines
-246
to
+249
| } else if (index == 0) { | ||
| } else if (buffer eq null) { |
There was a problem hiding this comment.
Rather than relying on the index, we simply check if the buffer hasn't been initialized.
Comment on lines
-271
to
-274
| // We do the updates here, | ||
| // to ensure they happen after we have secured the state. | ||
| inOnNextLoop = false | ||
| index = 0 |
There was a problem hiding this comment.
Redundant.
The initial state is this, and finishing each loop leaves the state like this as well.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
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.
I think this fixes the flaky flow interop tests: #3541 #3542
Related to #3449
I think the main change is ensuring we reset the
StreamSubscriberstate BEFORE completing the callback with theChunk.But, I also made sure we reset the state in any place where we complete a callback, even with exceptions, just in case.
c.c. @armanbilge