Simplify system retry logic: Part 1 #3172
Conversation
| } | ||
| ) | ||
|
|
||
| var _ grpc.UnaryServerInterceptor = (*RetryableInterceptor)(nil).Intercept |
There was a problem hiding this comment.
This is great.. are you planning to consolidate the "retryable rpc clients" (client/*/retryable_client*.go) with an client interceptor also?
There was a problem hiding this comment.
Oh that's a good idea. I wasn't planning for it, but yeah I can replace them with a client interceptor.
Metrics client probably also need to be converted into a client interceptor as currently retryable client is on top of metrics client.
There was a problem hiding this comment.
Created a task for now #3183 as the replacement is not the main purpose of this PR.
| case *serviceerror.Unavailable: | ||
| return true |
There was a problem hiding this comment.
under what condition does persistence layer return Unavailable error?
There was a problem hiding this comment.
I couldn't find any usage for cassandra, but for sql, it's returned here: https://github.com/temporalio/temporal/blob/master/common/persistence/sql/common.go#L66
I removed the resource exhausted error from the existing IsPersistenceTransientError implementation, so only Unavailable error is remained here.
| currentRunID, err := c.getCurrentRunID( | ||
| ctx, | ||
| shardOwnershipAsserted, | ||
| namespaceID, | ||
| workflowID, | ||
| ) |
There was a problem hiding this comment.
why we need this get again? It is done at line 243 already
There was a problem hiding this comment.
If workflow is closed, we can't know if there's a newer run created after getting the runID on L243. So need to get runID again to validate it's still the latest run. And since workflow is locked before the second check, there's won't be a newer run if the second check is successful.
| // ErrMaxAttemptsExceeded is exported temporarily for integration test | ||
| ErrMaxAttemptsExceeded = errors.New("maximum attempts exceeded to update history") | ||
| // ErrLocateCurrentWorkflowExecution is the error returned when current workflow execution can't be located | ||
| ErrLocateCurrentWorkflowExecution = serviceerror.NewUnavailable("unable to locate current workflow execution") |
There was a problem hiding this comment.
This only happens when workflow creates a newer run between getting current runID and loading the mutable state for that runID in getCurrentWorkflowContext(). So it's due to a race condition of two concurrent user requests, which can likely be resolved by a retry.
| return false | ||
| } | ||
|
|
||
| func IsServiceHandlerRetryableError(err error) bool { | ||
| switch err.(type) { | ||
| case *serviceerror.Internal, |
There was a problem hiding this comment.
I am surprised to see Internal here. It shouldn't be retryable.
There was a problem hiding this comment.
I agree. But I am not confident enough that we follow this requirement throughout our codebase so it can be removed from here. 😞
What changed?
conditionalRetryCount) for ErrConflict, ErrStaleState and when unable to locate current workflow run. Instead when ^ error happens, history interceptor will attempt a retry.NOTE:
Why?
How did you test it?
Potential risks
Is hotfix candidate?