diff --git a/README.md b/README.md index 378e312..17d7059 100644 --- a/README.md +++ b/README.md @@ -187,7 +187,7 @@ err := ctx.RunStep(Charge, order, &receipt, gush.WithNoRetry("Declined")) `WithNoRetry` matches the type of a `gush.NewApplicationError`. Pass `gush.StepTimeoutErrorType` to stop retrying steps that timed out. A `TemporalTimeout:` entry naming a timeout Temporal does not know is an error when there can be a retry. -Errors from `RunFlow` and `RunStep` work with `errors.As`, `gush.IsApplicationError`, `gush.IsPermanentError` and `gush.IsPanicError`. `gush.IsInvalidConfigError` recognizes a request refused as invalid, such as settings Temporal refuses or an input over its size limit, so sending it again cannot help. Wrapping one with `fmt.Errorf` and `%w` keeps its type and retryability. Its details stay on the original error, so `errors.As` can find an outer error without them. Call `errors.Unwrap` on that one until `Details` succeeds. +Errors from `RunFlow` and `RunStep` work with `errors.As`, `gush.IsApplicationError`, `gush.IsPermanentError` and `gush.IsPanicError`. `gush.IsInvalidConfigError` recognizes a request refused as invalid, such as settings Temporal refuses, an input over its 2MiB blob limit or a request over gRPC's 4MiB message limit, so sending it again cannot help. Wrapping one with `fmt.Errorf` and `%w` keeps its type and retryability. Its details stay on the original error, so `errors.As` can find an outer error without them. Call `errors.Unwrap` on that one until `Details` succeeds. ## Priority and fairness diff --git a/error.go b/error.go index 477cd3d..9f14b4e 100644 --- a/error.go +++ b/error.go @@ -7,6 +7,8 @@ import ( "go.temporal.io/api/serviceerror" "go.temporal.io/sdk/temporal" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" ) // ApplicationError is an error with a type name that retry policies can match. @@ -52,9 +54,11 @@ func IsFlowRunningError(err error) bool { // IsInvalidConfigError reports whether a request was refused as invalid, so sending it again // cannot succeed: Gush found settings Temporal refuses, including in a flow that has since // failed on them, or the server answered InvalidArgument. The server gives that answer for -// more than settings, such as an input over its size limit. Past a flow it sees through Abort, -// NewPermanentError and %w, but not errors.Join or several %w, and not a client's refusal inside -// a step, whose text is all that crosses the step. +// more than settings, such as an input over its blob size limit, 2MiB by default. A request over +// gRPC's message size limit, 4MiB by default, is refused before it reaches Temporal, and that +// counts too. Past a flow it sees through Abort, NewPermanentError and %w, but not errors.Join +// or several %w, and not a client's refusal inside a step, whose text is all that crosses the +// step. func IsInvalidConfigError(err error) bool { if errors.Is(err, errInvalidConfig) { return true @@ -63,6 +67,9 @@ func IsInvalidConfigError(err error) bool { if errors.As(err, &rejected) { return true } + if isMessageTooLarge(err) { + return true + } // A flow can wrap the failure in an error of its own before returning it, so look at every // ApplicationError in the chain rather than the outermost. @@ -76,6 +83,36 @@ func IsInvalidConfigError(err error) bool { return false } +// messageTooLargePrefixes are how gRPC words a refusal of a message over its size limit. The +// code alone, ResourceExhausted, cannot tell, because Temporal also uses it for rate limits. +var messageTooLargePrefixes = []string{ + "grpc: received message larger than max", + "grpc: message after decompression larger than max", + "grpc: received message after decompression larger than max", + "grpc: trying to send message larger than max", + "grpc: message too large", + "trying to send message larger than max", +} + +// isMessageTooLarge reports whether gRPC refused a request for its size. The SDK gives such an +// error its own type and does not retry it, but nothing outside the SDK can name that type. +func isMessageTooLarge(err error) bool { + var withStatus interface{ GRPCStatus() *status.Status } + if !errors.As(err, &withStatus) { + return false + } + grpcStatus := withStatus.GRPCStatus() + if grpcStatus == nil || grpcStatus.Code() != codes.ResourceExhausted { + return false + } + for _, prefix := range messageTooLargePrefixes { + if strings.HasPrefix(grpcStatus.Message(), prefix) { + return true + } + } + return false +} + // IsPanicError reports whether err is a PanicError. func IsPanicError(err error) bool { return temporal.IsPanicError(err) diff --git a/error_test.go b/error_test.go index a7f0647..e6c6e9b 100644 --- a/error_test.go +++ b/error_test.go @@ -10,6 +10,8 @@ import ( "go.temporal.io/api/enums/v1" "go.temporal.io/api/serviceerror" "go.temporal.io/sdk/temporal" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" ) func TestUnwrapError(t *testing.T) { @@ -419,14 +421,19 @@ func TestIsInvalidConfigError(t *testing.T) { sameType := NewApplicationError("settings not loaded yet", invalidConfigErrorType) applicationType := NewPermanentError("settings not loaded yet", "InvalidConfig", nil) - recognized := []error{failedFlow, wrappedByFlow, fmt.Errorf("run: %w", failedFlow), invalid, fmt.Errorf("start: %w", invalid), rejected, fmt.Errorf("start: %w", rejected)} + // What the SDK returns when gRPC refuses a request over its message size limit. + tooLarge := status.Error(codes.ResourceExhausted, "grpc: received message larger than max (5242977 vs. 4194304)") + // A rate limit shares the code and can succeed later. + rateLimited := status.Error(codes.ResourceExhausted, "namespace rate limit exceeded") + + recognized := []error{failedFlow, wrappedByFlow, fmt.Errorf("run: %w", failedFlow), invalid, fmt.Errorf("start: %w", invalid), rejected, fmt.Errorf("start: %w", rejected), tooLarge, fmt.Errorf("start: %w", tooLarge)} for _, err := range recognized { if !IsInvalidConfigError(err) { t.Errorf("IsInvalidConfigError(%v) = false", err) } } - others := []error{sameType, applicationType, nil, errors.New("some other failure"), serviceerror.NewUnavailable("try later")} + others := []error{sameType, applicationType, nil, errors.New("some other failure"), serviceerror.NewUnavailable("try later"), rateLimited, serviceerror.NewResourceExhausted(0, "busy")} for _, err := range others { if IsInvalidConfigError(err) { t.Errorf("IsInvalidConfigError(%v) = true", err) diff --git a/integration_test.go b/integration_test.go index 5b9dc21..8593407 100644 --- a/integration_test.go +++ b/integration_test.go @@ -4,6 +4,7 @@ package gush_test import ( "context" + "crypto/rand" "errors" "fmt" "io" @@ -1959,3 +1960,22 @@ func TestStepInitialIntervalAboveTheDefaultCapRuns(t *testing.T) { require.NoError(t, err) assert.Equal(t, 42, output.Amount) } + +// Temporal refuses an input over its blob size limit, and gRPC refuses one over its message size +// limit before Temporal sees it. Sending either again cannot succeed. +func TestStartFlowRefusesAnOversizedInput(t *testing.T) { + sizes := map[string]int{"over the blob limit": 3 << 20, "over the message limit": 5 << 20} + for name, size := range sizes { + t.Run(name, func(t *testing.T) { + // Random bytes, so compression cannot bring the payload under either limit. + input := make([]byte, size) + _, err := rand.Read(input) + require.NoError(t, err) + + _, err = testClient.StartFlow(testContext(t), flowRequest(Double, input, nil)) + + require.Error(t, err) + assert.True(t, gush.IsInvalidConfigError(err), "%v", err) + }) + } +} diff --git a/retry.go b/retry.go index d142d0b..d38b1a8 100644 --- a/retry.go +++ b/retry.go @@ -122,32 +122,37 @@ func (policy RetryPolicy) intervals() (initial, maximum time.Duration) { } // totalBackoff returns the time spent waiting between attempts if all MaxAttempts are used, -// applying the same defaults Temporal does to zero settings. +// applying the same defaults Temporal does to zero settings. The delay before retry k is +// interval*coefficient^k until it reaches the cap, so this sums that run in closed form, as +// Temporal computes each delay, and rounds up: Temporal truncates each delay, so the total +// never falls short of its own. A loop over the retries ran inside the flow and, with millions +// of attempts and delays too small to grow, held it past the deadlock detector's limit. func (policy RetryPolicy) totalBackoff() time.Duration { + retries := policy.MaxAttempts - 1 + if retries <= 0 { + return 0 + } interval, maxInterval := policy.intervals() coefficient := policy.BackoffCoefficient if coefficient == 0 { coefficient = 2.0 } + if math.IsNaN(coefficient) || coefficient <= 1 || interval >= maxInterval { + return scaleDuration(retries, interval) + } - var total time.Duration - for retries := policy.MaxAttempts - 1; retries > 0; retries-- { - if interval >= maxInterval { - return addDurations(total, scaleDuration(retries, maxInterval)) - } - if coefficient <= 1 { - return addDurations(total, scaleDuration(retries, interval)) - } - total = addDurations(total, interval) + // Retries before the delay reaches the cap; a ratio that is a power of the coefficient + // counts the delay equal to the cap on either side, which gives the same total. + growing := math.Ceil(math.Log(float64(maxInterval)/float64(interval)) / math.Log(coefficient)) + geometric := min(float64(retries), growing) + // Dividing first keeps a huge coefficient from overflowing the product. + sum := float64(interval) * ((math.Pow(coefficient, geometric) - 1) / (coefficient - 1)) - next := float64(interval) * coefficient - if next >= float64(maxInterval) { - interval = maxInterval - continue - } - interval = time.Duration(next) + total := maxDuration + if sum < float64(maxDuration) { + total = time.Duration(math.Ceil(sum)) } - return total + return addDurations(total, scaleDuration(retries-int32(geometric), maxInterval)) } // addDurations returns a+b, stopping at maxDuration rather than overflowing. diff --git a/step.go b/step.go index b070406..e0b3b0b 100644 --- a/step.go +++ b/step.go @@ -69,6 +69,12 @@ type StepInfo struct { // policy, pass a NewApplicationError, which is a request to retry. A canceled error cancels // the step if it is being canceled, and otherwise fails it for good too. An ApplicationError // caused by a cancellation is recorded without that cause or its details. +// +// A result over Temporal's blob size limit, 2MiB by default, fails the step on the server while +// Resume returns nil, because the server accepts the request and then refuses the result. One +// over gRPC's 4MiB message limit makes Resume return an error IsInvalidConfigError recognizes, +// and the step stays suspended with its token still valid: resume it again with a smaller +// result or an error, or it waits until its attempt times out and is retried under its policy. func (c *Client) Resume(ctx context.Context, resumeToken string, result any, stepErr error) error { token, err := decodeResumeToken(resumeToken) if err != nil { diff --git a/temporal_options_test.go b/temporal_options_test.go index b5ba59d..ffe4152 100644 --- a/temporal_options_test.go +++ b/temporal_options_test.go @@ -205,6 +205,31 @@ func TestTotalBackoff(t *testing.T) { policy: RetryPolicy{InitialInterval: time.Second, BackoffCoefficient: 1, MaxInterval: time.Second, MaxAttempts: 1 << 30}, expected: time.Duration(1<<30-1) * time.Second, }, + { + name: "a cap reached exactly", + policy: RetryPolicy{InitialInterval: time.Second, BackoffCoefficient: 2, MaxInterval: 64 * time.Second, MaxAttempts: 10}, + expected: (1 + 2 + 4 + 8 + 16 + 32 + 64 + 64 + 64) * time.Second, + }, + { + name: "the default cap", + policy: RetryPolicy{MaxAttempts: 10}, + expected: (1 + 2 + 4 + 8 + 16 + 32 + 64 + 100 + 100) * time.Second, + }, + { + name: "rounding does not fall short", + policy: RetryPolicy{InitialInterval: 22, BackoffCoefficient: 4.65, MaxAttempts: 2}, + expected: 22, + }, + { + name: "a huge coefficient does not overflow", + policy: RetryPolicy{InitialInterval: time.Second, BackoffCoefficient: 1e300, MaxInterval: time.Minute, MaxAttempts: 3}, + expected: time.Second + time.Minute, + }, + { + name: "a NaN coefficient adds no growth", + policy: RetryPolicy{InitialInterval: time.Second, BackoffCoefficient: math.NaN(), MaxAttempts: 4}, + expected: 3 * time.Second, + }, } for _, tt := range tests { @@ -228,3 +253,67 @@ func TestWithWaitForCancelReachesTemporal(t *testing.T) { t.Error("WithWaitForCancel did not reach the activity options") } } + +// Delays too small to grow once truncated, or a coefficient barely above one, used to run a loop +// for every retry inside the flow and trip the deadlock detector. +// temporalBackoff adds up the delays as Temporal computes them, one retry at a time. +func temporalBackoff(policy RetryPolicy) time.Duration { + interval, maxInterval := policy.intervals() + coefficient := policy.BackoffCoefficient + if coefficient == 0 { + coefficient = 2.0 + } + var total time.Duration + for retry := range policy.MaxAttempts - 1 { + // Cap before converting: a float beyond int64 converts differently on each platform. + delay := min(float64(interval)*math.Pow(coefficient, float64(retry)), float64(maxInterval)) + total += time.Duration(delay) + } + return total +} + +func TestTotalBackoffCoversTemporalsDelays(t *testing.T) { + intervals := []time.Duration{1, 7, 22, time.Millisecond, 3 * time.Second} + coefficients := []float64{1.01, 1.1, 1.5, 2, 4.65, 10} + for _, interval := range intervals { + for _, coefficient := range coefficients { + for attempts := int32(1); attempts <= 40; attempts++ { + policy := RetryPolicy{ + InitialInterval: interval, + BackoffCoefficient: coefficient, + MaxInterval: 1000 * interval, + MaxAttempts: attempts, + } + want := temporalBackoff(policy) + got := policy.totalBackoff() + // Float error stays far below a nanosecond per retry. + slack := time.Duration(attempts) + want/1e9 + if got < want || got > want+slack { + t.Fatalf("%+v: totalBackoff() = %d, Temporal waits %d", policy, got, want) + } + } + } + } +} + +func TestTotalBackoffIsQuickForAnyPolicy(t *testing.T) { + policies := map[string]RetryPolicy{ + "nanosecond delays": {InitialInterval: time.Nanosecond, BackoffCoefficient: 1.1, MaxAttempts: math.MaxInt32}, + "slow growth": {InitialInterval: time.Second, BackoffCoefficient: 1.0000001, MaxAttempts: math.MaxInt32}, + } + for name, policy := range policies { + t.Run(name, func(t *testing.T) { + // A loop over these retries takes seconds, so this bound leaves a loaded machine room. + start := time.Now() + got := policy.totalBackoff() + if elapsed := time.Since(start); elapsed > time.Second { + t.Errorf("totalBackoff took %v", elapsed) + } + + _, maxInterval := policy.intervals() + if lower := scaleDuration(policy.MaxAttempts/2, maxInterval); got < lower { + t.Errorf("totalBackoff() = %v, want at least %v", got, lower) + } + }) + } +}