GODRIVER-3810 Update WithTransaction to raise timeout error. #2344
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change | ||||||
|---|---|---|---|---|---|---|---|---|
|
|
@@ -81,38 +81,48 @@ func executeStartTransaction(ctx context.Context, operation *operation) (*operat | |||||||
| return newErrorResult(sess.StartTransaction(opts)), nil | ||||||||
| } | ||||||||
|
|
||||||||
| func executeWithTransaction(ctx context.Context, op *operation, loopDone <-chan struct{}) (*operationResult, error) { | ||||||||
| sess, err := entities(ctx).session(op.Object) | ||||||||
| if err != nil { | ||||||||
| return nil, err | ||||||||
| } | ||||||||
|
|
||||||||
| // Process the "callback" argument. This is an array of operation objects, each of which should be executed inside | ||||||||
| // the transaction. | ||||||||
| callback, err := op.Arguments.LookupErr("callback") | ||||||||
| if err != nil { | ||||||||
| return nil, newMissingArgumentError("callback") | ||||||||
| } | ||||||||
| var operations []*operation | ||||||||
| if err := callback.Unmarshal(&operations); err != nil { | ||||||||
| return nil, fmt.Errorf("error transforming callback option to slice of operations: %v", err) | ||||||||
| } | ||||||||
|
|
||||||||
| // Remove the "callback" field and process the other options. | ||||||||
| var temp transactionOptions | ||||||||
| if err := bson.Unmarshal(removeFieldsFromDocument(op.Arguments, "callback"), &temp); err != nil { | ||||||||
| return nil, fmt.Errorf("error unmarshalling arguments to transactionOptions: %v", err) | ||||||||
| } | ||||||||
|
|
||||||||
| _, withTransErr := sess.WithTransaction(ctx, func(ctx context.Context) (any, error) { | ||||||||
| var cbErr error | ||||||||
| for idx, oper := range operations { | ||||||||
| res, execErr := oper.execute(ctx, loopDone) | ||||||||
| if execErr != nil { | ||||||||
| // Capture the error but continue executing the remaining operations in the callback. | ||||||||
| err = fmt.Errorf("error executing operation %q at index %d: %v", oper.Name, idx, execErr) | ||||||||
| return nil, nil | ||||||||
|
Comment on lines
+113
to
+114
There was a problem hiding this comment. executeWithTransaction callback: on execErr, the callback currently returns (nil, nil), which causes WithTransaction to commit even though the callback did not successfully execute all operations, and relies on mutating the outer
Suggested change
|
||||||||
| } | ||||||||
| if cbErr == nil && res != nil { | ||||||||
| cbErr = res.Err | ||||||||
| } | ||||||||
| } | ||||||||
|
Comment on lines
+108
to
119
There was a problem hiding this comment. executeWithTransaction callback: once an operation inside the callback produces a non-nil operation error (res.Err), the callback should return that error immediately so subsequent operations are not executed. The current logic records the first res.Err in cbErr but continues executing later operations, which can change transactional behavior compared to a real application callback and can cause extra side effects or different errors. |
||||||||
| return nil, cbErr | ||||||||
| }, temp.TransactionOptionsBuilder) | ||||||||
| if err != nil { | ||||||||
| return nil, err | ||||||||
| } | ||||||||
| return &operationResult{Err: withTransErr}, nil | ||||||||
| } | ||||||||
|
|
||||||||
| func executeGetSnapshotTime(ctx context.Context, op *operation) (*operationResult, error) { | ||||||||
|
|
||||||||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -827,6 +827,35 @@ func (bwe ClientBulkWriteException) Error() string { | |
| return "bulk write exception: " + strings.Join(causes, ", ") | ||
| } | ||
|
|
||
| // TimeoutError represents an error that occurred due to a timeout. | ||
| type TimeoutError struct { | ||
| Wrapped error | ||
|
Comment on lines
+831
to
+832
Contributor
There was a problem hiding this comment. Does Additionally, we need to update |
||
| } | ||
|
|
||
| // Error implements the error interface. | ||
| func (e TimeoutError) Error() string { | ||
|
matthewdale marked this conversation as resolved.
|
||
| const timeoutMsg = "operation timed out" | ||
| if e.Wrapped == nil { | ||
| return timeoutMsg | ||
| } | ||
| return fmt.Sprintf("%s: %v", timeoutMsg, e.Wrapped.Error()) | ||
| } | ||
|
|
||
| // Unwrap returns the underlying error. | ||
| func (e TimeoutError) Unwrap() error { | ||
| return e.Wrapped | ||
| } | ||
|
|
||
| // HasErrorLabel returns true if the error contains the specified label. | ||
| func (e TimeoutError) HasErrorLabel(label string) bool { | ||
| if label == "ExceededTimeLimitError" { | ||
| return true | ||
| } else if le := LabeledError(nil); errors.As(e.Wrapped, &le) { | ||
| return le.HasErrorLabel(label) | ||
| } | ||
| return false | ||
| } | ||
|
Comment on lines
+830
to
+857
There was a problem hiding this comment. Consider adding a unit test for TimeoutError to ensure it integrates with existing timeout detection and labeling (e.g., IsTimeout should return true via the ExceededTimeLimitError label, and HasErrorLabel should delegate to the wrapped error for other labels). There are already table-driven tests for IsTimeout in mongo/errors_test.go that can be extended. |
||
|
|
||
| // returnResult is used to determine if a function calling processWriteError should return | ||
| // the result or return nil. Since the processWriteError function is used by many different | ||
| // methods, both *One and *Many, we need a way to differentiate if the method should return | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -148,13 +148,13 @@ func (s *Session) WithTransaction( | |
| } | ||
| backoff := expDur * time.Duration(jitter.Int63n(512)) / 512 | ||
| if time.Since(startTime)+backoff > transTimeout { | ||
| return nil, TimeoutError{Wrapped: err} | ||
| } | ||
| sleep := time.NewTimer(backoff) | ||
| select { | ||
| case <-timeout.C: | ||
| sleep.Stop() | ||
| return nil, TimeoutError{Wrapped: err} | ||
|
Member
There was a problem hiding this comment. The spec says we have to distinguish between CSOT and non-CSOT errors: In the current approach we always return TimeoutErrors which are indistinguishable.
Member
There was a problem hiding this comment. @tadjik1 The non-CSOT path in WithTransaction uses a manual wall-clock check against the 120-second limit, not a context deadline. AFIAK context.DeadlineExceeded is always CSOT in Go Driver.
Contributor
Author
There was a problem hiding this comment. Also, since the original error is wrapped in the new type, the user can always trace back and distinguish the errors.
Contributor
There was a problem hiding this comment. This is probably beyond the scope of this PR, but moving forward we should avoid picking which error to return and instead return all errors joined using E.g. var errs []error
// ...
select {
case <-timeout.C:
errs = append(errs, errors.New("default WithTransaction timeout reached"))
return nil, errutil.Join(errs...)
default:
}
// ... |
||
| case <-sleep.C: | ||
| } | ||
| if expDur < backoffMax { | ||
|
|
@@ -178,7 +178,7 @@ func (s *Session) WithTransaction( | |
|
|
||
| select { | ||
| case <-timeout.C: | ||
| return nil, TimeoutError{Wrapped: err} | ||
| default: | ||
| } | ||
|
Comment on lines
179
to
183
There was a problem hiding this comment. WithTransaction now wraps some timeout paths with TimeoutError, but other uses of the same overall timeout timer still return the raw |
||
|
|
||
|
|
@@ -217,15 +217,14 @@ func (s *Session) WithTransaction( | |
| return res, nil | ||
| } | ||
|
|
||
| var cerr CommandError | ||
| if errors.As(err, &cerr) { | ||
| if cerr.HasErrorLabel(driver.UnknownTransactionCommitResult) && !cerr.IsMaxTimeMSExpiredError() { | ||
| select { | ||
| case <-timeout.C: | ||
| return res, TimeoutError{Wrapped: err} | ||
| default: | ||
|
Comment on lines
+223
to
+226
Contributor
There was a problem hiding this comment. Is the intent of this change to only respect the default WithTransaction timeout if the error is a |
||
| } | ||
| continue | ||
| } | ||
| if cerr.HasErrorLabel(driver.TransientTransactionError) { | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Just to clarify: we capture
errhere because we want to get "3rd type of result" from this function, errors that are not related toWithTransactionlogic, right? So we do want to get error, but we don't want driver to handle this as if it's a transaction error.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Correct, we need to capture the error while still completing all operations in the transaction.