Treat oversized requests as invalid and stop the backoff loop in flows

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
 

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)

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)

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)
+		})
+	}
+}

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.

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 {

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)
+			}
+		})
+	}
+}