From cc90064ed98fb9a1cd5e596ddccada40fb5acaef Mon Sep 17 00:00:00 2001 From: Randy <857271370@qq.com> Date: Thu, 5 Nov 2026 21:32:32 +0800 Subject: [PATCH 1/2] =?UTF-8?q?=E2=9C=A8=20feat:=20tell=20a=20step=20that?= =?UTF-8?q?=20has=20ended=20from=20a=20token=20that=20was=20never=20valid?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 2 ++ integration_test.go | 12 ++++++++++++ step.go | 7 +++++++ step_context_test.go | 20 ++++++++++++++++++++ 4 files changed, 41 insertions(+) diff --git a/README.md b/README.md index 536760a..a832ef7 100644 --- a/README.md +++ b/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 callback that arrives after the step ended fails with an error `gush.IsStepGoneError` reports. `gush.IsResumeTokenError` also covers a token that was never valid. + ## 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. diff --git a/integration_test.go b/integration_test.go index c5db13d..9c04ea2 100644 --- a/integration_test.go +++ b/integration_test.go @@ -418,6 +418,18 @@ 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.IsStepGoneError(err), "%v", err) +} + +// The server rejects these as unreadable, which is not the same as a step that has ended. +func TestUnreadableResumeTokenIsNotAGoneStep(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.IsStepGoneError(err), "%v", err) } func TestSuspendedStepFailsWithResumeError(t *testing.T) { diff --git a/step.go b/step.go index 5c39d6e..a3f9383 100644 --- a/step.go +++ b/step.go @@ -149,6 +149,13 @@ func IsResumeTokenError(err error) bool { return true } + return IsStepGoneError(err) +} + +// IsStepGoneError reports whether Resume failed because the step no longer exists, such as +// after it timed out or was already resumed. It tells a late callback from a token that was +// never good: every such error is also an IsResumeTokenError, and not the other way around. +func IsStepGoneError(err error) bool { var notFound *serviceerror.NotFound return errors.As(err, ¬Found) } diff --git a/step_context_test.go b/step_context_test.go index d7a6ea0..58db1d7 100644 --- a/step_context_test.go +++ b/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,21 @@ func TestMalformedResumeTokenIsAResumeTokenError(t *testing.T) { require.Error(t, err) assert.True(t, gush.IsResumeTokenError(err), "%v", err) } + +func TestGoneStepIsAlsoAResumeTokenError(t *testing.T) { + err := fmt.Errorf("resume: %w", serviceerror.NewNotFound("activity task not found")) + + assert.True(t, gush.IsStepGoneError(err), "%v", err) + assert.True(t, gush.IsResumeTokenError(err), "%v", err) +} + +func TestBadResumeTokenIsNotAGoneStep(t *testing.T) { + client := &gush.Client{} + malformed := client.Resume(context.Background(), "!!!not-base64!!!", nil, nil) + rejected := serviceerror.NewInvalidArgument("Error deserializing task token.") + + for _, err := range []error{malformed, rejected} { + assert.True(t, gush.IsResumeTokenError(err), "%v", err) + assert.False(t, gush.IsStepGoneError(err), "%v", err) + } +} From a0669a349d477e39518197c74a8f9d76471e9288 Mon Sep 17 00:00:00 2001 From: Randy <857271370@qq.com> Date: Thu, 5 Nov 2026 22:57:43 +0800 Subject: [PATCH 2/2] =?UTF-8?q?=E2=99=BB=EF=B8=8F=20refactor:=20name=20the?= =?UTF-8?q?=20check=20for=20what=20it=20reports,=20a=20stale=20resume=20to?= =?UTF-8?q?ken?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 2 +- integration_test.go | 41 ++++++++++++++++++++++++++++++++++++----- step.go | 18 ++++++++++-------- step_context_test.go | 11 ++++++----- 4 files changed, 53 insertions(+), 19 deletions(-) diff --git a/README.md b/README.md index a832ef7..378e312 100644 --- a/README.md +++ b/README.md @@ -230,7 +230,7 @@ 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 callback that arrives after the step ended fails with an error `gush.IsStepGoneError` reports. `gush.IsResumeTokenError` also covers a token that was never valid. +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 diff --git a/integration_test.go b/integration_test.go index 9c04ea2..5b9dc21 100644 --- a/integration_test.go +++ b/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,18 +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.IsStepGoneError(err), "%v", err) + assert.True(t, gush.IsStaleResumeTokenError(err), "%v", err) } -// The server rejects these as unreadable, which is not the same as a step that has ended. -func TestUnreadableResumeTokenIsNotAGoneStep(t *testing.T) { +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.IsStepGoneError(err), "%v", err) + assert.False(t, gush.IsStaleResumeTokenError(err), "%v", err) } func TestSuspendedStepFailsWithResumeError(t *testing.T) { diff --git a/step.go b/step.go index a3f9383..b070406 100644 --- a/step.go +++ b/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,13 +150,14 @@ func IsResumeTokenError(err error) bool { return true } - return IsStepGoneError(err) + return IsStaleResumeTokenError(err) } -// IsStepGoneError reports whether Resume failed because the step no longer exists, such as -// after it timed out or was already resumed. It tells a late callback from a token that was -// never good: every such error is also an IsResumeTokenError, and not the other way around. -func IsStepGoneError(err error) bool { +// 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, ¬Found) } diff --git a/step_context_test.go b/step_context_test.go index 58db1d7..e699c97 100644 --- a/step_context_test.go +++ b/step_context_test.go @@ -99,20 +99,21 @@ func TestMalformedResumeTokenIsAResumeTokenError(t *testing.T) { assert.True(t, gush.IsResumeTokenError(err), "%v", err) } -func TestGoneStepIsAlsoAResumeTokenError(t *testing.T) { +func TestStaleResumeTokenIsAlsoAResumeTokenError(t *testing.T) { err := fmt.Errorf("resume: %w", serviceerror.NewNotFound("activity task not found")) - assert.True(t, gush.IsStepGoneError(err), "%v", err) + assert.True(t, gush.IsStaleResumeTokenError(err), "%v", err) assert.True(t, gush.IsResumeTokenError(err), "%v", err) } -func TestBadResumeTokenIsNotAGoneStep(t *testing.T) { +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} { + for _, err := range []error{malformed, rejected, noNamespace} { assert.True(t, gush.IsResumeTokenError(err), "%v", err) - assert.False(t, gush.IsStepGoneError(err), "%v", err) + assert.False(t, gush.IsStaleResumeTokenError(err), "%v", err) } }