REP-5358 Fix hangs when reading change stream channels. #77
Conversation
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
tdq45gj
requested changes
Dec 9, 2024
Collaborator
tdq45gj
left a comment
tdq45gj
left a comment
There was a problem hiding this comment.
Looks good to me in general. I wonder if there's any point calling Get before Eventual is ready.
internal/util/eventual.go
Outdated
|
|
||
| // Get returns an option that contains the Eventual’s value, or | ||
| // empty if the value isn’t ready yet. | ||
| func (e *Eventual[T]) Get() option.Option[T] { |
Collaborator
There was a problem hiding this comment.
Is there any case that we want to call Get before it's ready? Looks like we're calling Get().MustGet() everywhere, can we make Get return T or fail if it's not ready?
Collaborator
Author
There was a problem hiding this comment.
I thought about that. I was/am a bit averse to proliferating panics, but you’re right that there’s little point in calling Get() before the value is ready. (It can be checked anyway by doing a select on Ready() with a default.)
I’ll change it.
| sess, err := suite.srcMongoClient.StartSession() | ||
| suite.Require().NoError(err) | ||
| sctx := mongo.NewSessionContext(ctx, sess) | ||
| _, err = suite.srcMongoClient.Database("testDb").Collection("testColl").InsertOne( |
Collaborator
There was a problem hiding this comment.
Can we use suite.DBNameForTest()?
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.
This fixes hangs from the change stream and document reader when contexts are canceled.
StartChangeEventHandler doesn’t need to send to the error channel. It now just returns its error.
The other channels—writes-off, error, and done—now allow infinite (non-blocking) reads once populated. This works via a new struct, Eventual, that supplies the same semantics as Context’s Done() and Err() methods. The change stream struct’s writes-off and error channels are written to be Eventual. The doneChan remains a channel but now is just closed rather than written to.
This also refactors a couple functions for general reuse and fixes TestStartAtTimeNoChanges, which has flapped occasionally.