Tell a stale resume token from one that was never valid

README.md

@@ -230,6 +230,8 @@ err = client.Resume(ctx, resumeToken, decision, nil)
 
 An error passed to `Resume` fails the step for good, so a rejection does not send the request again. To run the code again under the step's retry policy, pass a `gush.NewApplicationError`. A canceled error cancels the step if it is being canceled, and otherwise fails it for good too.
 
+A token belongs to one attempt. `gush.IsStaleResumeTokenError` reports that the attempt is over: the step ended, or it is retrying under a new token. The server cannot tell this from a well-formed token that matches no step, such as a corrupted one. `gush.IsResumeTokenError` also covers a token the server cannot read. If a flow has moved on from a canceled step that has not yet ended on the server, `Resume` succeeds and the result is discarded.
+
 ## Changing running flows
 
 Flows can run for a long time, and a replay must take the same path it took originally. `Version` keeps old executions on the old code.

integration_test.go

@@ -85,7 +85,7 @@ func runWithTemporal(m *testing.M) int {
 	serveContext, stopServing := context.WithCancel(context.Background())
 	testWorker := gush.NewWorker(testQueue).
 		WithFlows(InvalidStepTimeoutFlow, InvalidNoRetryFlow, InvalidChildTimeoutFlow, InvalidGrandchildFlow, WrappedInvalidSettingsFlow, PointerOutputFlow).
-		WithFlows(NegativePickupFlow, PickupInsteadOfAttemptFlow, OneAttemptNoRetryFlow, ChildNoRetryFlow, TooManyAttemptsFlow, TooManyAttemptsOnlyPickupFlow).
+		WithFlows(RetriedApprovalFlow, NegativePickupFlow, PickupInsteadOfAttemptFlow, OneAttemptNoRetryFlow, ChildNoRetryFlow, TooManyAttemptsFlow, TooManyAttemptsOnlyPickupFlow).
 		WithFlows(CancelableApprovalFlow, InvalidStepAttemptsFlow, InvalidChildAttemptsFlow, InvalidStartAttemptsFlow,
 			HighInitialIntervalFlow, Double, FailingFlow, ParentOfFailingFlow, FailingStepFlow, PlainFailureFlow, AbortFlow, ApprovalFlow,
 			ExpiringApprovalFlow, TimeoutFlow, HeartbeatFlow, SilentFlow, ReplayFlow, AttributesFlow, AttributesChildFlow,
@@ -418,6 +418,49 @@ func TestSuspendedStepFinishesWithResume(t *testing.T) {
 	err := testClient.Resume(ctx, token, "again", nil)
 	require.Error(t, err)
 	assert.True(t, gush.IsResumeTokenError(err), "%v", err)
+	assert.True(t, gush.IsStaleResumeTokenError(err), "%v", err)
+}
+
+func RetriedApprovalFlow(ctx *gush.FlowContext) (any, error) {
+	var decision string
+	err := ctx.RunStep(ApprovalStep, string(ctx.Input), &decision,
+		gush.WithAttemptTimeout(3*time.Second),
+		gush.WithMaxAttempts(2),
+		gush.WithInitialInterval(retryBackoff),
+	)
+	return decision, err
+}
+
+// The step is still alive here, waiting on its second attempt, and only that token finishes it.
+func TestFirstAttemptsTokenIsStaleWhileTheSecondAttemptWaits(t *testing.T) {
+	ctx := testContext(t)
+	key := uniqueKey(t)
+
+	result := make(chan error, 1)
+	var decision string
+	go func() { result <- testClient.RunFlow(ctx, flowRequest(RetriedApprovalFlow, key, &decision)) }()
+
+	first := <-resumeTokens
+	second := <-resumeTokens
+
+	err := testClient.Resume(ctx, first, "late", nil)
+	require.Error(t, err)
+	assert.True(t, gush.IsStaleResumeTokenError(err), "%v", err)
+
+	require.NoError(t, testClient.Resume(ctx, second, "approved", nil))
+	require.NoError(t, <-result)
+	assert.Equal(t, "approved", decision)
+}
+
+// The server rejects these as unreadable, which is not the same as a token whose attempt is over.
+func TestUnreadableResumeTokenIsNotStale(t *testing.T) {
+	ctx := testContext(t)
+
+	err := testClient.Resume(ctx, "bnVsbA==", "result", nil)
+
+	require.Error(t, err)
+	assert.True(t, gush.IsResumeTokenError(err), "%v", err)
+	assert.False(t, gush.IsStaleResumeTokenError(err), "%v", err)
 }
 
 func TestSuspendedStepFailsWithResumeError(t *testing.T) {

step.go

@@ -53,7 +53,8 @@ type StepInfo struct {
 	// FairKey is this step's fairness group.
 	FairKey string
 
-	// ResumeToken is what Resume needs to finish this step after it suspends.
+	// ResumeToken is what Resume needs to finish this step after it suspends. It belongs to
+	// this attempt, and a retry gets a new one.
 	ResumeToken string
 
 	// Attempt counts from 1.
@@ -132,8 +133,8 @@ func withoutCanceledCause(err error) error {
 	return temporal.NewApplicationErrorWithOptions(appErr.Message(), appErr.Type(), options)
 }
 
-// IsResumeTokenError reports whether Resume failed because its token is invalid or the
-// step no longer exists, such as after it timed out or was already resumed.
+// IsResumeTokenError reports whether Resume failed because its token is invalid, or stale as
+// IsStaleResumeTokenError describes.
 func IsResumeTokenError(err error) bool {
 	if errors.Is(err, errInvalidResumeToken) {
 		return true
@@ -149,6 +150,14 @@ func IsResumeTokenError(err error) bool {
 		return true
 	}
 
+	return IsStaleResumeTokenError(err)
+}
+
+// IsStaleResumeTokenError reports whether Resume failed because its token's attempt is over:
+// the step ended, or it is retrying under a new token. That makes it a late callback rather
+// than a bad one, though the server cannot tell it from a well-formed token that matches no
+// step, such as a corrupted one. Every such error is also an IsResumeTokenError.
+func IsStaleResumeTokenError(err error) bool {
 	var notFound *serviceerror.NotFound
 	return errors.As(err, &notFound)
 }

step_context_test.go

@@ -3,12 +3,14 @@ package gush_test
 import (
 	"context"
 	"errors"
+	"fmt"
 	"testing"
 	"time"
 
 	"github.com/batchstream/gush"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.temporal.io/api/serviceerror"
 	"go.temporal.io/sdk/activity"
 )
 
@@ -96,3 +98,22 @@ func TestMalformedResumeTokenIsAResumeTokenError(t *testing.T) {
 	require.Error(t, err)
 	assert.True(t, gush.IsResumeTokenError(err), "%v", err)
 }
+
+func TestStaleResumeTokenIsAlsoAResumeTokenError(t *testing.T) {
+	err := fmt.Errorf("resume: %w", serviceerror.NewNotFound("activity task not found"))
+
+	assert.True(t, gush.IsStaleResumeTokenError(err), "%v", err)
+	assert.True(t, gush.IsResumeTokenError(err), "%v", err)
+}
+
+func TestBadResumeTokenIsNotStale(t *testing.T) {
+	client := &gush.Client{}
+	malformed := client.Resume(context.Background(), "!!!not-base64!!!", nil, nil)
+	rejected := serviceerror.NewInvalidArgument("Error deserializing task token.")
+	noNamespace := serviceerror.NewNamespaceNotFound("namespace")
+
+	for _, err := range []error{malformed, rejected, noNamespace} {
+		assert.True(t, gush.IsResumeTokenError(err), "%v", err)
+		assert.False(t, gush.IsStaleResumeTokenError(err), "%v", err)
+	}
+}