From 4cbb25de5929943b56092c486e021eb43a1a036c Mon Sep 17 00:00:00 2001 From: Randy <857271370@qq.com> Date: Thu, 5 Nov 2026 21:04:29 +0800 Subject: [PATCH 01/11] =?UTF-8?q?=F0=9F=90=9B=20fix:=20read=20pointer=20ou?= =?UTF-8?q?tputs=20and=20named=20byte=20slices=20as=20raw=20values?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- codec.go | 78 ++++++++++++++++++++++++++---------- codec_test.go | 107 ++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 164 insertions(+), 21 deletions(-) diff --git a/codec.go b/codec.go index 8c70130..ce6a4a3 100644 --- a/codec.go +++ b/codec.go @@ -28,22 +28,33 @@ var conversionMethods = []reflect.Type{ // usedAsIs reports whether values of a type are their own encoding: strings and byte // slices, including types built on them, unless the type chooses its own JSON form. func usedAsIs(valueType reflect.Type) bool { - isBytes := valueType.Kind() == reflect.Slice && valueType.Elem() == reflect.TypeFor[byte]() - if valueType.Kind() != reflect.String && !isBytes { + if valueType.Kind() != reflect.String && !isByteSlice(valueType) { return false } + return !hasConversionMethods(valueType) +} + +// isByteSlice reports whether a type is a slice of bytes. Elements with their own JSON or +// text form make it a list instead, as they do for encoding/json. +func isByteSlice(valueType reflect.Type) bool { + if valueType.Kind() != reflect.Slice || valueType.Elem().Kind() != reflect.Uint8 { + return false + } + return !hasConversionMethods(valueType.Elem()) +} +func hasConversionMethods(valueType reflect.Type) bool { pointer := reflect.PointerTo(valueType) for _, methods := range conversionMethods { if pointer.Implements(methods) { - return false + return true } } - return true + return false } // Encode serializes a value the way flows and steps do for their input and output. -// Strings and byte slices are used as they are, including through a pointer and including +// Strings and byte slices are used as they are, through any number of pointers and including // types built on them, unless the type has its own JSON or text methods. A nil value or nil // pointer encodes to nil, and everything else becomes JSON without HTML escaping. // @@ -55,16 +66,15 @@ func Encode(value any) ([]byte, error) { } reflected := reflect.ValueOf(value) - if reflected.Kind() == reflect.Pointer { + for reflected.Kind() == reflect.Pointer { if reflected.IsNil() { return nil, nil } - typed, ok := value.(*any) - if ok { - return Encode(*typed) - } reflected = reflected.Elem() } + if reflected.Kind() == reflect.Interface { + return Encode(reflected.Interface()) + } if usedAsIs(reflected.Type()) { if reflected.Kind() == reflect.String { @@ -84,8 +94,9 @@ func Encode(value any) ([]byte, error) { } // Decode deserializes data produced by Encode into target. A pointer to anything Encode uses -// as it is receives the bytes as they are, empty data zeroes target, and everything else is -// decoded as JSON. *any receives what the JSON holds, or a string when the data is not JSON. +// as it is, through any number of pointers, receives the bytes as they are, empty data zeroes +// target, and everything else is decoded as JSON. *any receives what the JSON holds, or a +// string when the data is not JSON. func Decode(data []byte, target any) error { if len(data) == 0 { setToZeroValue(target) @@ -98,17 +109,42 @@ func Decode(data []byte, target any) error { return nil } - reflected := reflect.ValueOf(target) - if reflected.Kind() == reflect.Pointer && !reflected.IsNil() && usedAsIs(reflected.Type().Elem()) { - element := reflected.Elem() - if element.Kind() == reflect.String { - element.SetString(string(data)) - return nil - } - element.SetBytes(bytes.Clone(data)) + destination, ok := rawDestination(target) + if !ok { + return json.Unmarshal(data, target) + } + if destination.Kind() == reflect.String { + destination.SetString(string(data)) return nil } - return json.Unmarshal(data, target) + destination.SetBytes(bytes.Clone(data)) + return nil +} + +// rawDestination finds what receives data as it is when target leads to a type Encode uses +// as it is, allocating the pointers on the way, as encoding/json does for a nil pointer. +func rawDestination(target any) (destination reflect.Value, ok bool) { + destination = reflect.ValueOf(target) + if destination.Kind() != reflect.Pointer || destination.IsNil() { + return destination, false + } + + finalType := destination.Type().Elem() + for finalType.Kind() == reflect.Pointer { + finalType = finalType.Elem() + } + if !usedAsIs(finalType) { + return destination, false + } + + destination = destination.Elem() + for destination.Kind() == reflect.Pointer { + if destination.IsNil() { + destination.Set(reflect.New(destination.Type().Elem())) + } + destination = destination.Elem() + } + return destination, true } // setToZeroValue zeroes the value target points to, leaving unexported struct fields alone. diff --git a/codec_test.go b/codec_test.go index 4f8d345..aaad611 100644 --- a/codec_test.go +++ b/codec_test.go @@ -483,3 +483,110 @@ func TestDecodeIntoAny(t *testing.T) { expected := map[string]any{"a": 1.0} require.Equal(t, expected, output) } + +type namedByte byte + +type namedBytes []namedByte + +// jsonByte has its own JSON form, which makes a slice of it a list, as in encoding/json. +type jsonByte byte + +func (b jsonByte) MarshalJSON() ([]byte, error) { + return json.Marshal(int(b) + 1) +} + +func (b *jsonByte) UnmarshalJSON(data []byte) error { + var value int + err := json.Unmarshal(data, &value) + if err != nil { + return err + } + *b = jsonByte(value - 1) + return nil +} + +func TestSliceOfNamedBytesTravelsRaw(t *testing.T) { + encoded, err := Encode([]namedByte("raw")) + require.NoError(t, err) + require.Equal(t, "raw", string(encoded)) + encoded, err = Encode(namedBytes("raw")) + require.NoError(t, err) + require.Equal(t, "raw", string(encoded)) + + var plain []namedByte + require.NoError(t, Decode([]byte("raw"), &plain)) + require.Equal(t, []namedByte("raw"), plain) + var named namedBytes + require.NoError(t, Decode([]byte("raw"), &named)) + require.Equal(t, namedBytes("raw"), named) +} + +func TestSliceOfBytesWithTheirOwnJSONIsAList(t *testing.T) { + encoded, err := Encode([]jsonByte{1, 2}) + require.NoError(t, err) + require.Equal(t, "[2,3]", string(encoded)) + + var decoded []jsonByte + require.NoError(t, Decode(encoded, &decoded)) + require.Equal(t, []jsonByte{1, 2}, decoded) +} + +func TestPointersOfAnyDepthReachRawValues(t *testing.T) { + text := "done" + pointer := &text + encoded, err := Encode(&pointer) + require.NoError(t, err) + require.Equal(t, "done", string(encoded)) + + var output *string + require.NoError(t, Decode([]byte("done"), &output)) + require.NotNil(t, output) + require.Equal(t, "done", *output) + + var deeper **string + require.NoError(t, Decode([]byte("done"), &deeper)) + require.NotNil(t, deeper) + require.Equal(t, "done", **deeper) + + var bytesOutput *[]byte + require.NoError(t, Decode([]byte("raw"), &bytesOutput)) + require.NotNil(t, bytesOutput) + require.Equal(t, []byte("raw"), *bytesOutput) +} + +func TestPointerToNilPointerEncodesToNil(t *testing.T) { + var pointer *string + encoded, err := Encode(&pointer) + require.NoError(t, err) + require.Nil(t, encoded) + + var empty any + encoded, err = Encode(&empty) + require.NoError(t, err) + require.Nil(t, encoded) +} + +func TestEmptyDataZeroesAPointerOutput(t *testing.T) { + text := "old" + output := &text + require.NoError(t, Decode(nil, &output)) + require.Nil(t, output) +} + +func TestPointerToAnyHoldingAPointerEncodesWhatItHolds(t *testing.T) { + text := "done" + var value any = &text + encoded, err := Encode(&value) + require.NoError(t, err) + require.Equal(t, "done", string(encoded)) +} + +func TestPointerOutputOfAStructStillDecodesJSON(t *testing.T) { + type receipt struct { + ID string `json:"id"` + } + var output *receipt + require.NoError(t, Decode([]byte(`{"id":"a"}`), &output)) + require.NotNil(t, output) + require.Equal(t, "a", output.ID) +} From 6045b88847d7ded0f9af564d49cfaa09621c5412 Mon Sep 17 00:00:00 2001 From: Randy <857271370@qq.com> Date: Thu, 5 Nov 2026 21:04:29 +0800 Subject: [PATCH 02/11] =?UTF-8?q?=F0=9F=A6=BA=20feat:=20reject=20step=20an?= =?UTF-8?q?d=20flow=20settings=20Temporal=20cannot=20run?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- error.go | 12 +++++++++ error_test.go | 23 ++++++++++++++++ flow_config.go | 10 +++++++ retry.go | 33 +++++++++++++++++++---- retry_internal_test.go | 59 ++++++++++++++++++++++++++++++++++++++++-- step_config.go | 19 +++++++++++++- 6 files changed, 148 insertions(+), 8 deletions(-) diff --git a/error.go b/error.go index 8b49a69..2f9b8b6 100644 --- a/error.go +++ b/error.go @@ -5,6 +5,7 @@ import ( "regexp" "strings" + "go.temporal.io/api/serviceerror" "go.temporal.io/sdk/temporal" ) @@ -48,6 +49,17 @@ func IsFlowRunningError(err error) bool { return temporal.IsWorkflowExecutionAlreadyStartedError(err) } +// IsInvalidConfigError reports whether a flow or step did not start because of its settings: +// Gush found settings Temporal could never run, or the server rejected the request as +// invalid. Retrying the same settings cannot succeed. +func IsInvalidConfigError(err error) bool { + if errors.Is(err, errInvalidConfig) { + return true + } + var rejected *serviceerror.InvalidArgument + return errors.As(err, &rejected) +} + // 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 24c3531..24ea85e 100644 --- a/error_test.go +++ b/error_test.go @@ -403,3 +403,26 @@ func TestIsFlowRunningError(t *testing.T) { t.Error("IsFlowRunningError(plain) = true") } } + +func TestIsInvalidConfigError(t *testing.T) { + config := DefaultFlowConfig() + config.Timeout = -1 + client := &Client{} + request := FlowRequest{Flow: "AnyFlow"} + _, invalid := client.StartFlow(context.Background(), request, WithConfig(config)) + rejected := serviceerror.NewInvalidArgument("the server said no") + + recognized := []error{invalid, fmt.Errorf("start: %w", invalid), rejected, fmt.Errorf("start: %w", rejected)} + for _, err := range recognized { + if !IsInvalidConfigError(err) { + t.Errorf("IsInvalidConfigError(%v) = false", err) + } + } + + others := []error{nil, errors.New("some other failure"), serviceerror.NewUnavailable("try later")} + for _, err := range others { + if IsInvalidConfigError(err) { + t.Errorf("IsInvalidConfigError(%v) = true", err) + } + } +} diff --git a/flow_config.go b/flow_config.go index 51c72ae..94dd991 100644 --- a/flow_config.go +++ b/flow_config.go @@ -1,6 +1,7 @@ package gush import ( + "fmt" "maps" "slices" "time" @@ -38,6 +39,15 @@ func DefaultFlowConfig() *FlowConfig { return config } +// validate rejects settings Temporal cannot run. On a child flow they would leave the parent +// Running with its tasks failing. +func (config *FlowConfig) validate() error { + if config.Timeout < 0 { + return fmt.Errorf("%w: timeout %v must not be negative", errInvalidConfig, config.Timeout) + } + return config.RetryPolicy.validate() +} + // Apply applies options to config in order. func (config *FlowConfig) Apply(options ...FlowOption) { for _, option := range options { diff --git a/retry.go b/retry.go index 2d3011e..f160d1e 100644 --- a/retry.go +++ b/retry.go @@ -1,9 +1,12 @@ package gush import ( + "errors" "fmt" + "strings" "time" + "go.temporal.io/api/enums/v1" "go.temporal.io/sdk/temporal" ) @@ -16,6 +19,13 @@ const ( defaultStepAttempts int32 = 10 ) +// timeoutErrorTypePrefix marks a NoRetry entry that names a timeout rather than an +// ApplicationError type, and Temporal rejects one naming an unknown timeout. +const timeoutErrorTypePrefix = "TemporalTimeout:" + +// errInvalidConfig marks settings that Temporal could never run. See IsInvalidConfigError. +var errInvalidConfig = errors.New("gush: invalid config") + // RetryPolicy defines how flows and steps are retried after they fail. Every setting left // at zero uses its default. type RetryPolicy struct { @@ -35,7 +45,9 @@ type RetryPolicy struct { // number is an error. MaxAttempts int32 `json:"max_attempts"` - // NoRetry lists the ApplicationError types that fail immediately. + // NoRetry lists the ApplicationError types that fail immediately. An entry starting with + // TemporalTimeout: names a timeout instead, such as StepTimeoutErrorType, and any + // other timeout name is an error. NoRetry []string `json:"no_retry"` } @@ -45,13 +57,24 @@ type RetryPolicy struct { func (policy RetryPolicy) validate() error { switch { case policy.MaxAttempts < UnlimitedAttempts: - return fmt.Errorf("gush: max attempts %d must be positive, zero for the default, or UnlimitedAttempts", policy.MaxAttempts) + return fmt.Errorf("%w: max attempts %d must be positive, zero for the default, or UnlimitedAttempts", errInvalidConfig, policy.MaxAttempts) case policy.BackoffCoefficient != 0 && policy.BackoffCoefficient < 1: - return fmt.Errorf("gush: backoff coefficient %v must be at least 1", policy.BackoffCoefficient) + return fmt.Errorf("%w: backoff coefficient %v must be at least 1", errInvalidConfig, policy.BackoffCoefficient) case policy.InitialInterval < 0: - return fmt.Errorf("gush: initial interval %v must not be negative", policy.InitialInterval) + return fmt.Errorf("%w: initial interval %v must not be negative", errInvalidConfig, policy.InitialInterval) case policy.MaxInterval < 0: - return fmt.Errorf("gush: max interval %v must not be negative", policy.MaxInterval) + return fmt.Errorf("%w: max interval %v must not be negative", errInvalidConfig, policy.MaxInterval) + } + + for _, errorType := range policy.NoRetry { + timeoutName, isTimeout := strings.CutPrefix(errorType, timeoutErrorTypePrefix) + if !isTimeout { + continue + } + timeoutType, err := enums.TimeoutTypeFromString(timeoutName) + if err != nil || timeoutType == enums.TIMEOUT_TYPE_UNSPECIFIED { + return fmt.Errorf("%w: no retry entry %q names an unknown timeout", errInvalidConfig, errorType) + } } return nil } diff --git a/retry_internal_test.go b/retry_internal_test.go index 7ee5175..dcf76ce 100644 --- a/retry_internal_test.go +++ b/retry_internal_test.go @@ -1,6 +1,7 @@ package gush import ( + "errors" "testing" "time" ) @@ -13,10 +14,14 @@ func TestRetryPolicyValidate(t *testing.T) { "negative coefficient": {BackoffCoefficient: -1}, "negative initial": {InitialInterval: -time.Second}, "negative maximum": {MaxInterval: -time.Second}, + "an unknown timeout": {NoRetry: []string{"TemporalTimeout:Soon"}}, + "no timeout name": {NoRetry: []string{"TemporalTimeout:"}}, + "unspecified timeout": {NoRetry: []string{"TemporalTimeout:Unspecified"}}, } for name, policy := range invalid { - if policy.validate() == nil { - t.Errorf("%s: validate() = nil", name) + err := policy.validate() + if !errors.Is(err, errInvalidConfig) { + t.Errorf("%s: validate() = %v, want an invalid config error", name, err) } } @@ -26,6 +31,9 @@ func TestRetryPolicyValidate(t *testing.T) { "a number": {MaxAttempts: 3}, "a coefficient of 1": {BackoffCoefficient: 1}, "a cap below first": {InitialInterval: time.Minute, MaxInterval: time.Second}, + "an error type": {NoRetry: []string{"Declined", ""}}, + "the step timeout": {NoRetry: []string{StepTimeoutErrorType}}, + "another timeout": {NoRetry: []string{"TemporalTimeout:Heartbeat"}}, } for name, policy := range valid { if err := policy.validate(); err != nil { @@ -71,3 +79,50 @@ func TestTotalBackoffWithACapBelowTheFirstDelay(t *testing.T) { t.Errorf("totalBackoff() = %v, want %v", got, expected) } } + +func TestStepConfigValidate(t *testing.T) { + invalid := map[string]func(config *StepConfig){ + "no attempt timeout": func(config *StepConfig) { config.AttemptTimeout = 0 }, + "negative attempt timeout": func(config *StepConfig) { config.AttemptTimeout = -time.Second }, + "negative heartbeat": func(config *StepConfig) { config.HeartbeatTimeout = -time.Second }, + "negative pickup": func(config *StepConfig) { config.PickupTimeout = -time.Second }, + "a bad retry policy": func(config *StepConfig) { config.MaxAttempts = -2 }, + } + for name, change := range invalid { + config := DefaultStepConfig() + change(config) + err := config.validate() + if !errors.Is(err, errInvalidConfig) { + t.Errorf("%s: validate() = %v, want an invalid config error", name, err) + } + } + + config := DefaultStepConfig() + config.HeartbeatTimeout = time.Second + config.PickupTimeout = time.Second + err := config.validate() + if err != nil { + t.Errorf("valid settings: validate() = %v", err) + } +} + +func TestFlowConfigValidate(t *testing.T) { + config := DefaultFlowConfig() + config.Timeout = -time.Second + err := config.validate() + if !errors.Is(err, errInvalidConfig) { + t.Errorf("negative timeout: validate() = %v, want an invalid config error", err) + } + + config = DefaultFlowConfig() + config.MaxAttempts = -2 + err = config.validate() + if !errors.Is(err, errInvalidConfig) { + t.Errorf("a bad retry policy: validate() = %v, want an invalid config error", err) + } + + err = DefaultFlowConfig().validate() + if err != nil { + t.Errorf("defaults: validate() = %v", err) + } +} diff --git a/step_config.go b/step_config.go index 71a1d41..6f0a65d 100644 --- a/step_config.go +++ b/step_config.go @@ -1,6 +1,9 @@ package gush -import "time" +import ( + "fmt" + "time" +) // StepConfig holds the settings of one step run. type StepConfig struct { @@ -51,6 +54,20 @@ func (option stepOption) applyStep(config *StepConfig) { option(config) } +// validate rejects settings Temporal cannot run, which would leave the flow Running with its +// tasks failing. +func (config *StepConfig) validate() error { + switch { + case config.AttemptTimeout <= 0: + return fmt.Errorf("%w: attempt timeout %v must be positive", errInvalidConfig, config.AttemptTimeout) + case config.HeartbeatTimeout < 0: + return fmt.Errorf("%w: heartbeat timeout %v must not be negative", errInvalidConfig, config.HeartbeatTimeout) + case config.PickupTimeout < 0: + return fmt.Errorf("%w: pickup timeout %v must not be negative", errInvalidConfig, config.PickupTimeout) + } + return config.RetryPolicy.validate() +} + // WithQueue runs the step on another task queue. func WithQueue(queue string) StepOption { apply := func(config *StepConfig) { From f4b4c86962fdaff7e668f0e4fe21a031fa51518c Mon Sep 17 00:00:00 2001 From: Randy <857271370@qq.com> Date: Thu, 5 Nov 2026 21:04:29 +0800 Subject: [PATCH 03/11] =?UTF-8?q?=F0=9F=90=9B=20fix:=20keep=20ErrorIfRunni?= =?UTF-8?q?ng=20when=20WithConfig=20replaces=20settings?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- flow_config.go | 6 ++++-- flow_config_test.go | 22 ++++++++++++++++++++++ 2 files changed, 26 insertions(+), 2 deletions(-) diff --git a/flow_config.go b/flow_config.go index 94dd991..9b7a15c 100644 --- a/flow_config.go +++ b/flow_config.go @@ -63,8 +63,9 @@ func (option flowOption) applyFlow(config *FlowConfig) { // WithConfig replaces the settings accumulated so far with a copy of config, so options // after it refine config. An empty Priority or FairKey keeps the current one, and -// LogAttributes merge over the current ones, so a child flow keeps what it inherits. A nil -// config changes nothing. +// LogAttributes merge over the current ones, so a child flow keeps what it inherits. +// ErrorIfRunning stays on once set, because a config cannot switch off that safety check. A +// nil config changes nothing. func WithConfig(config *FlowConfig) FlowOption { apply := func(target *FlowConfig) { if config == nil { @@ -73,6 +74,7 @@ func WithConfig(config *FlowConfig) FlowOption { current := *target *target = *config + target.ErrorIfRunning = config.ErrorIfRunning || current.ErrorIfRunning target.NoRetry = slices.Clone(config.NoRetry) if target.Priority == "" { target.Priority = current.Priority diff --git a/flow_config_test.go b/flow_config_test.go index fd243ab..f1ebe83 100644 --- a/flow_config_test.go +++ b/flow_config_test.go @@ -362,3 +362,25 @@ func TestJSONConfigDecodesOverExistingSettings(t *testing.T) { t.Errorf("existing settings changed: %q, %d, %q", config.ID, config.MaxAttempts, config.Priority) } } + +func TestWithConfigKeepsErrorIfRunningOnceSet(t *testing.T) { + config := gush.DefaultFlowConfig() + config.Apply(gush.WithErrorIfRunning(), gush.WithConfig(gush.DefaultFlowConfig())) + if !config.ErrorIfRunning { + t.Error("WithConfig switched off an earlier WithErrorIfRunning") + } + + queued := gush.DefaultFlowConfig() + queued.ErrorIfRunning = true + config = gush.DefaultFlowConfig() + config.Apply(gush.WithConfig(queued)) + if !config.ErrorIfRunning { + t.Error("WithConfig ignored ErrorIfRunning in the config") + } + + config = gush.DefaultFlowConfig() + config.Apply(gush.WithConfig(gush.DefaultFlowConfig())) + if config.ErrorIfRunning { + t.Error("ErrorIfRunning turned on without being asked for") + } +} From ac550e878c71c07582e1ad15326223658677f4fe Mon Sep 17 00:00:00 2001 From: Randy <857271370@qq.com> Date: Thu, 5 Nov 2026 21:04:29 +0800 Subject: [PATCH 04/11] =?UTF-8?q?=E2=99=BB=EF=B8=8F=20refactor:=20rename?= =?UTF-8?q?=20the=20LogAttributes=20type=20to=20LogAttributeMap?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- log_attribute.go | 6 +++--- options.go | 2 +- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/log_attribute.go b/log_attribute.go index 3131eb5..657c4d9 100644 --- a/log_attribute.go +++ b/log_attribute.go @@ -64,12 +64,12 @@ func unsignedAttr(key string, value uint64) LogAttribute { return attribute.Int64(key, int64(value)) } -// LogAttributes are the fields attached to every log record of a run. -type LogAttributes map[string]any +// LogAttributeMap holds the fields attached to every log record of a run, by key. +type LogAttributeMap map[string]any // UnmarshalJSON adds the attributes in data to these, keeping numbers exact, and null clears // them. Plain JSON decoding would turn every number into a float64, which damages large IDs. -func (attributes *LogAttributes) UnmarshalJSON(data []byte) error { +func (attributes *LogAttributeMap) UnmarshalJSON(data []byte) error { decoded, err := decodeLogAttributes(data) if err != nil { return err diff --git a/options.go b/options.go index 2dd5876..999503b 100644 --- a/options.go +++ b/options.go @@ -17,7 +17,7 @@ type CommonConfig struct { FairKey string `json:"fair_key"` // LogAttributes are attached to every log record and inherited by child flows and steps. - LogAttributes LogAttributes `json:"log_attributes"` + LogAttributes LogAttributeMap `json:"log_attributes"` } // FlowOption customizes a flow run. From 1efe8736b3e8d9d9a2eefab2be53783df0977c89 Mon Sep 17 00:00:00 2001 From: Randy <857271370@qq.com> Date: Thu, 5 Nov 2026 21:04:29 +0800 Subject: [PATCH 05/11] =?UTF-8?q?=E2=9C=A8=20feat:=20export=20NewLogExport?= =?UTF-8?q?er=20for=20hosts=20with=20their=20own=20provider?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- log_export.go | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/log_export.go b/log_export.go index 1cae980..3f6543f 100644 --- a/log_export.go +++ b/log_export.go @@ -85,7 +85,7 @@ func InitLogExport() (shutdown func(), err error) { return func() {}, nil } - exporter, err := newLogExporterFromEnv(context.Background()) + exporter, err := NewLogExporter(context.Background()) if err != nil { return nil, err } @@ -127,10 +127,11 @@ func releaseLogExport() { } } -// newLogExporterFromEnv returns nil when no endpoint is configured. Otherwise the exporter -// reads OpenTelemetry's standard variables itself. It accepts a malformed endpoint, such as -// collector:4318, and then loses every record, so that is an error here. -func newLogExporterFromEnv(ctx context.Context) (sdklog.Exporter, error) { +// NewLogExporter creates the OTLP exporter InitLogExport uses, from the same variables, for a +// process that builds its own LoggerProvider. It returns nil when no endpoint is configured. +// The exporter reads OpenTelemetry's standard variables itself. It accepts a malformed +// endpoint, such as collector:4318, and then loses every record, so that is an error here. +func NewLogExporter(ctx context.Context) (sdklog.Exporter, error) { variable, endpoint := firstEnv(envOTLPLogsEndpoint, envOTLPEndpoint) if endpoint == "" { return nil, nil From 16194631fbb6f6668ab246a7d2643a7ca66e45a0 Mon Sep 17 00:00:00 2001 From: Randy <857271370@qq.com> Date: Thu, 5 Nov 2026 21:04:29 +0800 Subject: [PATCH 06/11] =?UTF-8?q?=E2=9C=85=20test:=20cover=20rejected=20se?= =?UTF-8?q?ttings=20and=20pointer=20outputs=20on=20a=20real=20server?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- integration_test.go | 59 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 59 insertions(+) diff --git a/integration_test.go b/integration_test.go index def55b0..fa026a8 100644 --- a/integration_test.go +++ b/integration_test.go @@ -83,6 +83,7 @@ func runWithTemporal(m *testing.M) int { serveContext, stopServing := context.WithCancel(context.Background()) testWorker := gush.NewWorker(testQueue). + WithFlows(InvalidStepTimeoutFlow, InvalidPickupTimeoutFlow, InvalidNoRetryFlow, InvalidChildTimeoutFlow, PointerOutputFlow). WithFlows(CancelableApprovalFlow, InvalidStepAttemptsFlow, InvalidChildAttemptsFlow, InvalidStartAttemptsFlow, HighInitialIntervalFlow, Double, FailingFlow, ParentOfFailingFlow, FailingStepFlow, PlainFailureFlow, AbortFlow, ApprovalFlow, ExpiringApprovalFlow, TimeoutFlow, HeartbeatFlow, SilentFlow, ReplayFlow, AttributesFlow, AttributesChildFlow, @@ -1791,6 +1792,64 @@ func TestNegativeAttemptsOtherThanUnlimitedAreAnError(t *testing.T) { } } +func InvalidStepTimeoutFlow(ctx *gush.FlowContext) (any, error) { + return nil, ctx.RunStep(DoubleStep, order{ID: "bad-timeout", Amount: 1}, nil, gush.WithHeartbeatTimeout(-time.Second)) +} + +func InvalidPickupTimeoutFlow(ctx *gush.FlowContext) (any, error) { + return nil, ctx.RunStep(DoubleStep, order{ID: "bad-pickup", Amount: 1}, nil, gush.WithPickupTimeout(-time.Second)) +} + +func InvalidNoRetryFlow(ctx *gush.FlowContext) (any, error) { + return nil, ctx.RunStep(DoubleStep, order{ID: "bad-no-retry", Amount: 1}, nil, gush.WithNoRetry("TemporalTimeout:Soon")) +} + +func InvalidChildTimeoutFlow(ctx *gush.FlowContext) (any, error) { + request := gush.FlowRequest{Flow: SmallFlow} + return nil, ctx.RunFlow(request, gush.WithTimeout(-time.Second)) +} + +// Each of these used to leave its flow Running, with flow tasks failing until someone looked. +func TestSettingsTemporalCannotRunFailTheFlow(t *testing.T) { + ctx := testContext(t) + + flows := []any{InvalidStepTimeoutFlow, InvalidPickupTimeoutFlow, InvalidNoRetryFlow, InvalidChildTimeoutFlow} + for _, flow := range flows { + request := flowRequest(flow, nil, nil) + err := testClient.RunFlow(ctx, request) + assert.True(t, gush.IsPermanentError(err), "%v", err) + assert.ErrorContains(t, err, "invalid config") + } + + _, err := testClient.StartFlow(ctx, flowRequest(SmallFlow, nil, nil), gush.WithTimeout(-time.Second)) + assert.True(t, gush.IsInvalidConfigError(err), "%v", err) + _, err = testClient.StartFlow(ctx, flowRequest(SmallFlow, nil, nil), gush.WithNoRetry("TemporalTimeout:Soon")) + assert.True(t, gush.IsInvalidConfigError(err), "%v", err) +} + +func TestServerRejectionIsAnInvalidConfigError(t *testing.T) { + ctx := testContext(t) + + tooLong := strings.Repeat("x", 2000) + _, err := testClient.StartFlow(ctx, flowRequest(SmallFlow, nil, nil), gush.WithID(tooLong)) + assert.True(t, gush.IsInvalidConfigError(err), "%v", err) +} + +func PointerOutputFlow(ctx *gush.FlowContext) (any, error) { + text := "pointer" + return &text, nil +} + +func TestPointerToStringOutputArrivesRaw(t *testing.T) { + ctx := testContext(t) + + var output *string + err := testClient.RunFlow(ctx, flowRequest(PointerOutputFlow, nil, &output)) + require.NoError(t, err) + require.NotNil(t, output) + assert.Equal(t, "pointer", *output) +} + func HighInitialIntervalFlow(ctx *gush.FlowContext) (any, error) { var output order input := order{ID: "high-interval", Amount: 21} From 6b2eeb31e7a455d9c5d8fbfcffccae296f1dfc59 Mon Sep 17 00:00:00 2001 From: Randy <857271370@qq.com> Date: Thu, 5 Nov 2026 21:04:29 +0800 Subject: [PATCH 07/11] =?UTF-8?q?=F0=9F=93=9D=20docs:=20describe=20the=20s?= =?UTF-8?q?tricter=20settings=20and=20raw=20value=20rules?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index d08a023..ca3734f 100644 --- a/README.md +++ b/README.md @@ -106,7 +106,7 @@ err := ctx.RunFlow(request) Flows and steps are registered under their bare Go function names, and can be called by that name as a string. Two functions with the same name in different packages collide, and `Serve` reports it. Use `NewFlowHandler` and `NewStepHandler` to choose a name, which closures and the instantiations of a generic function need too. -Strings and byte slices, including types built on them, travel as raw bytes unless the type has its own JSON or text methods. Everything else travels as JSON, so read a value back as the type it was written. Adding or removing those methods changes how such a type encodes, so let flows carrying it finish first. +Strings and byte slices, through any number of pointers and including types built on them, travel as raw bytes unless the type has its own JSON or text methods. Everything else travels as JSON, so read a value back as the type it was written. Adding or removing those methods changes how such a type encodes, so let flows carrying it finish first. ### Parallel steps and cleanup @@ -157,7 +157,7 @@ A step that is already running only finds out about the cancellation through `ct | `WithID(id)` | flow | Flow ID. Otherwise one is generated, which `StartFlow` returns. A client's `RunFlow` or `StartFlow` joins a flow already running with this ID and ignores this run's flow, input and settings, so concurrent callers share one run. A child flow gets an error that `gush.IsFlowRunningError` recognizes. | | `WithErrorIfRunning()` | flow | A client's `RunFlow` or `StartFlow` returns that same error instead of joining. | | `WithTimeout(d)` | flow | Limit on the whole execution, including every retry. | -| `WithConfig(c)` | flow | Starts from a `FlowConfig`, such as one decoded from a queue. Settings left out of the JSON, or zero, use their defaults, and durations in JSON are integer nanoseconds. Whoever applies it chooses `RunFlow` or `StartFlow`. | +| `WithConfig(c)` | flow | Starts from a `FlowConfig`, such as one decoded from a queue. Settings left out of the JSON, or zero, use their defaults, and durations in JSON are integer nanoseconds. `WithErrorIfRunning` stays on once set. Whoever applies it chooses `RunFlow` or `StartFlow`. | | `WithQueue(q)` | step | Runs on another queue. | | `WithAttemptTimeout(d)` | step | Limit on each attempt, not the step as a whole. | | `WithHeartbeatTimeout(d)` | step | Fails an attempt that stops calling `ctx.Heartbeat()`. | @@ -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. -Errors from `RunFlow` and `RunStep` work with `errors.As`, `gush.IsApplicationError`, `gush.IsPermanentError` and `gush.IsPanicError`. 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 flow or step refused because of settings Temporal could never run, so retrying them 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 @@ -280,7 +280,7 @@ Lines written while Temporal replays a flow are dropped, so each appears once. H ### Export over OTLP -`gush.Dial` starts exporting when `OTEL_EXPORTER_OTLP_LOGS_ENDPOINT` or `OTEL_EXPORTER_OTLP_ENDPOINT` is set, and `Close` flushes. A process that logs without a client calls `gush.InitLogExport()` itself. [OpenTelemetry's other variables](https://opentelemetry.io/docs/specs/otel/protocol/exporter/) set the headers and TLS. The protocol is `http/protobuf` unless they ask for `grpc`, and any other protocol is an error: +`gush.Dial` starts exporting when `OTEL_EXPORTER_OTLP_LOGS_ENDPOINT` or `OTEL_EXPORTER_OTLP_ENDPOINT` is set, and `Close` flushes. A process that logs without a client calls `gush.InitLogExport()` itself, and one that builds its own `LoggerProvider` gets the same exporter from `gush.NewLogExporter(ctx)`. [OpenTelemetry's other variables](https://opentelemetry.io/docs/specs/otel/protocol/exporter/) set the headers and TLS. The protocol is `http/protobuf` unless they ask for `grpc`, and any other protocol is an error: ```sh OTEL_EXPORTER_OTLP_ENDPOINT=https://collector:4317 From c92961e822c2d286582a305d66b9dcbe5c518aba Mon Sep 17 00:00:00 2001 From: Randy <857271370@qq.com> Date: Thu, 5 Nov 2026 22:01:37 +0800 Subject: [PATCH 08/11] =?UTF-8?q?=F0=9F=90=9B=20fix:=20reject=20only=20set?= =?UTF-8?q?tings=20Temporal=20refuses=20and=20recognize=20them=20after=20a?= =?UTF-8?q?=20flow=20fails?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- error.go | 13 +++++--- error_test.go | 4 ++- flow_config.go | 5 +-- flow_context.go | 12 +++++-- integration_test.go | 58 +++++++++++++++++++++++++++----- retry.go | 72 +++++++++++++++++++++++++++++++--------- retry_internal_test.go | 69 +++++++++++++++++++++++++++++++++----- step_config.go | 26 +++++++++------ temporal_options.go | 10 +++--- temporal_options_test.go | 17 ++++++++++ 10 files changed, 230 insertions(+), 56 deletions(-) diff --git a/error.go b/error.go index 2f9b8b6..6c7b53d 100644 --- a/error.go +++ b/error.go @@ -49,15 +49,20 @@ func IsFlowRunningError(err error) bool { return temporal.IsWorkflowExecutionAlreadyStartedError(err) } -// IsInvalidConfigError reports whether a flow or step did not start because of its settings: -// Gush found settings Temporal could never run, or the server rejected the request as -// invalid. Retrying the same settings cannot succeed. +// 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. func IsInvalidConfigError(err error) bool { if errors.Is(err, errInvalidConfig) { return true } var rejected *serviceerror.InvalidArgument - return errors.As(err, &rejected) + if errors.As(err, &rejected) { + return true + } + var applicationError *ApplicationError + return errors.As(err, &applicationError) && applicationError.Type() == invalidConfigErrorType } // IsPanicError reports whether err is a PanicError. diff --git a/error_test.go b/error_test.go index 24ea85e..b0a9c2b 100644 --- a/error_test.go +++ b/error_test.go @@ -411,8 +411,10 @@ func TestIsInvalidConfigError(t *testing.T) { request := FlowRequest{Flow: "AnyFlow"} _, invalid := client.StartFlow(context.Background(), request, WithConfig(config)) rejected := serviceerror.NewInvalidArgument("the server said no") + // What a flow that failed on its settings looks like once it leaves the flow. + failedFlow := NewPermanentError("invalid config", invalidConfigErrorType, nil) - recognized := []error{invalid, fmt.Errorf("start: %w", invalid), rejected, fmt.Errorf("start: %w", rejected)} + recognized := []error{failedFlow, fmt.Errorf("run: %w", failedFlow), invalid, fmt.Errorf("start: %w", invalid), rejected, fmt.Errorf("start: %w", rejected)} for _, err := range recognized { if !IsInvalidConfigError(err) { t.Errorf("IsInvalidConfigError(%v) = false", err) diff --git a/flow_config.go b/flow_config.go index 9b7a15c..3acfb95 100644 --- a/flow_config.go +++ b/flow_config.go @@ -18,7 +18,8 @@ type FlowConfig struct { // ID is the flow ID. If empty, one is generated. ID string `json:"id"` - // Timeout limits the whole execution, including every retry. Zero means unlimited. + // Timeout limits the whole execution, including every retry. Zero means unlimited, and a + // negative one is an error. Timeout time.Duration `json:"timeout"` // ErrorIfRunning makes a client's RunFlow or StartFlow fail rather than join a flow @@ -45,7 +46,7 @@ func (config *FlowConfig) validate() error { if config.Timeout < 0 { return fmt.Errorf("%w: timeout %v must not be negative", errInvalidConfig, config.Timeout) } - return config.RetryPolicy.validate() + return config.RetryPolicy.validate(defaultFlowAttempts) } // Apply applies options to config in order. diff --git a/flow_context.go b/flow_context.go index a2bef17..6eef822 100644 --- a/flow_context.go +++ b/flow_context.go @@ -113,7 +113,7 @@ func (flowCtx *FlowContext) RunStep(step any, input, output any, options ...Step } err = config.validate() if err != nil { - return flowCtx.Abort(err) + return abortInvalidConfig(err) } payload, err := EncodePayload(input) if err != nil { @@ -142,7 +142,7 @@ func (flowCtx *FlowContext) RunFlow(request FlowRequest, options ...FlowOption) config := flowCtx.childFlowConfig(options) err := config.validate() if err != nil { - return flowCtx.Abort(err) + return abortInvalidConfig(err) } childOptions := newChildWorkflowOptions(request.Queue, config) @@ -178,7 +178,7 @@ func (flowCtx *FlowContext) StartFlow(request FlowRequest, options ...FlowOption config := flowCtx.childFlowConfig(options) err = config.validate() if err != nil { - return "", flowCtx.Abort(err) + return "", abortInvalidConfig(err) } // Canceling a context also cancels the children started on it. @@ -271,6 +271,12 @@ func (flowCtx *FlowContext) Decode(target any) error { return nil } +// abortInvalidConfig fails the flow over settings validate rejected. +func abortInvalidConfig(err error) error { + cause := []error{err} + return abortError(invalidConfigErrorType, cause) +} + // Abort returns an error that fails the flow permanently, with no further retries. func (flowCtx *FlowContext) Abort(cause ...error) error { return abortError(abortedFlowErrorType, cause) diff --git a/integration_test.go b/integration_test.go index fa026a8..45ec0f2 100644 --- a/integration_test.go +++ b/integration_test.go @@ -7,6 +7,7 @@ import ( "errors" "fmt" "io" + "math" "net/http" "net/http/httptest" "os" @@ -83,7 +84,8 @@ func runWithTemporal(m *testing.M) int { serveContext, stopServing := context.WithCancel(context.Background()) testWorker := gush.NewWorker(testQueue). - WithFlows(InvalidStepTimeoutFlow, InvalidPickupTimeoutFlow, InvalidNoRetryFlow, InvalidChildTimeoutFlow, PointerOutputFlow). + WithFlows(InvalidStepTimeoutFlow, InvalidNoRetryFlow, InvalidChildTimeoutFlow, InvalidGrandchildFlow, PointerOutputFlow). + WithFlows(NegativePickupFlow, PickupInsteadOfAttemptFlow, OneAttemptNoRetryFlow, ChildNoRetryFlow, TooManyAttemptsFlow). WithFlows(CancelableApprovalFlow, InvalidStepAttemptsFlow, InvalidChildAttemptsFlow, InvalidStartAttemptsFlow, HighInitialIntervalFlow, Double, FailingFlow, ParentOfFailingFlow, FailingStepFlow, PlainFailureFlow, AbortFlow, ApprovalFlow, ExpiringApprovalFlow, TimeoutFlow, HeartbeatFlow, SilentFlow, ReplayFlow, AttributesFlow, AttributesChildFlow, @@ -1796,10 +1798,6 @@ func InvalidStepTimeoutFlow(ctx *gush.FlowContext) (any, error) { return nil, ctx.RunStep(DoubleStep, order{ID: "bad-timeout", Amount: 1}, nil, gush.WithHeartbeatTimeout(-time.Second)) } -func InvalidPickupTimeoutFlow(ctx *gush.FlowContext) (any, error) { - return nil, ctx.RunStep(DoubleStep, order{ID: "bad-pickup", Amount: 1}, nil, gush.WithPickupTimeout(-time.Second)) -} - func InvalidNoRetryFlow(ctx *gush.FlowContext) (any, error) { return nil, ctx.RunStep(DoubleStep, order{ID: "bad-no-retry", Amount: 1}, nil, gush.WithNoRetry("TemporalTimeout:Soon")) } @@ -1809,24 +1807,68 @@ func InvalidChildTimeoutFlow(ctx *gush.FlowContext) (any, error) { return nil, ctx.RunFlow(request, gush.WithTimeout(-time.Second)) } +func InvalidGrandchildFlow(ctx *gush.FlowContext) (any, error) { + request := gush.FlowRequest{Flow: InvalidStepTimeoutFlow} + return nil, ctx.RunFlow(request) +} + // Each of these used to leave its flow Running, with flow tasks failing until someone looked. -func TestSettingsTemporalCannotRunFailTheFlow(t *testing.T) { +func TestSettingsTemporalRefusesFailTheFlow(t *testing.T) { ctx := testContext(t) - flows := []any{InvalidStepTimeoutFlow, InvalidPickupTimeoutFlow, InvalidNoRetryFlow, InvalidChildTimeoutFlow} + flows := []any{InvalidStepTimeoutFlow, InvalidNoRetryFlow, InvalidChildTimeoutFlow, InvalidGrandchildFlow} for _, flow := range flows { request := flowRequest(flow, nil, nil) err := testClient.RunFlow(ctx, request) assert.True(t, gush.IsPermanentError(err), "%v", err) + assert.True(t, gush.IsInvalidConfigError(err), "%v", err) assert.ErrorContains(t, err, "invalid config") } _, err := testClient.StartFlow(ctx, flowRequest(SmallFlow, nil, nil), gush.WithTimeout(-time.Second)) assert.True(t, gush.IsInvalidConfigError(err), "%v", err) - _, err = testClient.StartFlow(ctx, flowRequest(SmallFlow, nil, nil), gush.WithNoRetry("TemporalTimeout:Soon")) + _, err = testClient.StartFlow(ctx, flowRequest(SmallFlow, nil, nil), gush.WithMaxAttempts(2), gush.WithNoRetry("TemporalTimeout:Soon")) assert.True(t, gush.IsInvalidConfigError(err), "%v", err) } +func NegativePickupFlow(ctx *gush.FlowContext) (any, error) { + return nil, ctx.RunStep(DoubleStep, order{ID: "negative-pickup", Amount: 1}, nil, gush.WithPickupTimeout(-time.Second)) +} + +func PickupInsteadOfAttemptFlow(ctx *gush.FlowContext) (any, error) { + return nil, ctx.RunStep(DoubleStep, order{ID: "pickup-only", Amount: 1}, nil, gush.WithAttemptTimeout(0), gush.WithPickupTimeout(30*time.Second)) +} + +func OneAttemptNoRetryFlow(ctx *gush.FlowContext) (any, error) { + return nil, ctx.RunStep(DoubleStep, order{ID: "one-attempt", Amount: 1}, nil, gush.WithMaxAttempts(1), gush.WithNoRetry("TemporalTimeout:Soon")) +} + +func ChildNoRetryFlow(ctx *gush.FlowContext) (any, error) { + request := gush.FlowRequest{Flow: SmallFlow} + return nil, ctx.RunFlow(request, gush.WithNoRetry("TemporalTimeout:Soon")) +} + +func TooManyAttemptsFlow(ctx *gush.FlowContext) (any, error) { + return nil, ctx.RunStep(DoubleStep, order{ID: "too-many", Amount: 1}, nil, gush.WithMaxAttempts(math.MaxInt32), gush.WithPickupTimeout(time.Minute)) +} + +// Temporal accepts these settings, so a check that rejected them would fail flows in flight +// when a worker upgrades, even after the step already ran. +func TestSettingsTemporalAcceptsKeepRunning(t *testing.T) { + ctx := testContext(t) + + flows := []any{NegativePickupFlow, PickupInsteadOfAttemptFlow, OneAttemptNoRetryFlow, ChildNoRetryFlow, TooManyAttemptsFlow} + for _, flow := range flows { + request := flowRequest(flow, nil, nil) + err := testClient.RunFlow(ctx, request) + assert.NoError(t, err) + } + + id, err := testClient.StartFlow(ctx, flowRequest(SmallFlow, nil, nil), gush.WithNoRetry("TemporalTimeout:Soon")) + assert.NoError(t, err) + assert.NotEmpty(t, id) +} + func TestServerRejectionIsAnInvalidConfigError(t *testing.T) { ctx := testContext(t) diff --git a/retry.go b/retry.go index f160d1e..a70b0f2 100644 --- a/retry.go +++ b/retry.go @@ -3,6 +3,7 @@ package gush import ( "errors" "fmt" + "math" "strings" "time" @@ -23,9 +24,16 @@ const ( // ApplicationError type, and Temporal rejects one naming an unknown timeout. const timeoutErrorTypePrefix = "TemporalTimeout:" -// errInvalidConfig marks settings that Temporal could never run. See IsInvalidConfigError. +// invalidConfigErrorType is the ApplicationError type of a flow that failed on settings +// validate rejected, which lets IsInvalidConfigError recognize it after it leaves the flow. +const invalidConfigErrorType = "InvalidConfig" + +// errInvalidConfig marks settings that Temporal refuses. See IsInvalidConfigError. var errInvalidConfig = errors.New("gush: invalid config") +// maxDuration is the longest time a time.Duration can hold. +const maxDuration = time.Duration(math.MaxInt64) + // RetryPolicy defines how flows and steps are retried after they fail. Every setting left // at zero uses its default. type RetryPolicy struct { @@ -46,15 +54,18 @@ type RetryPolicy struct { MaxAttempts int32 `json:"max_attempts"` // NoRetry lists the ApplicationError types that fail immediately. An entry starting with - // TemporalTimeout: names a timeout instead, such as StepTimeoutErrorType, and any - // other timeout name is an error. + // TemporalTimeout: names a timeout instead, such as StepTimeoutErrorType. Naming a timeout + // Temporal does not know is an error, unless there is only one attempt and so no retry. NoRetry []string `json:"no_retry"` } -// validate rejects settings Temporal cannot run, which would leave a flow Running with its -// tasks failing, and an attempt count that only a mistake produces, such as a budget computed -// below zero, which must not turn into retrying forever. -func (policy RetryPolicy) validate() error { +// validate rejects settings Temporal refuses, which would leave a flow Running with its tasks +// failing, and an attempt count that only a mistake produces, such as a budget computed below +// zero, which must not turn into retrying forever. +// +// Flow code calls it, so it must not reject what Temporal accepts: a refused command leaves no +// history for a replay to meet, while a stricter check fails a flow in flight that ran it. +func (policy RetryPolicy) validate(defaultAttempts int32) error { switch { case policy.MaxAttempts < UnlimitedAttempts: return fmt.Errorf("%w: max attempts %d must be positive, zero for the default, or UnlimitedAttempts", errInvalidConfig, policy.MaxAttempts) @@ -66,6 +77,10 @@ func (policy RetryPolicy) validate() error { return fmt.Errorf("%w: max interval %v must not be negative", errInvalidConfig, policy.MaxInterval) } + // With only one attempt Temporal does not look at NoRetry. + if policy.attempts(defaultAttempts) == 1 { + return nil + } for _, errorType := range policy.NoRetry { timeoutName, isTimeout := strings.CutPrefix(errorType, timeoutErrorTypePrefix) if !isTimeout { @@ -79,6 +94,14 @@ func (policy RetryPolicy) validate() error { return nil } +// attempts returns MaxAttempts with zero read as the given default. +func (policy RetryPolicy) attempts(defaultAttempts int32) int32 { + if policy.MaxAttempts == 0 { + return defaultAttempts + } + return policy.MaxAttempts +} + // intervals returns the first delay and the cap, applying the defaults Temporal gives zero // settings. Temporal rejects a cap below the first delay. func (policy RetryPolicy) intervals() (initial, maximum time.Duration) { @@ -105,23 +128,42 @@ func (policy RetryPolicy) totalBackoff() time.Duration { var total time.Duration for retries := policy.MaxAttempts - 1; retries > 0; retries-- { if interval >= maxInterval { - return total + time.Duration(retries)*maxInterval + return addDurations(total, scaleDuration(retries, maxInterval)) } if coefficient <= 1 { - return total + time.Duration(retries)*interval + return addDurations(total, scaleDuration(retries, interval)) + } + total = addDurations(total, interval) + + next := float64(interval) * coefficient + if next >= float64(maxInterval) { + interval = maxInterval + continue } - total += interval - interval = time.Duration(float64(interval) * coefficient) + interval = time.Duration(next) } return total } +// addDurations returns a+b, stopping at maxDuration rather than overflowing. +func addDurations(a, b time.Duration) time.Duration { + if b > 0 && a > maxDuration-b { + return maxDuration + } + return a + b +} + +// scaleDuration returns count*duration, stopping at maxDuration rather than overflowing. +func scaleDuration(count int32, duration time.Duration) time.Duration { + if duration > 0 && time.Duration(count) > maxDuration/duration { + return maxDuration + } + return time.Duration(count) * duration +} + // temporalPolicy returns the policy Temporal runs, which reads zero attempts as unlimited. func (policy RetryPolicy) temporalPolicy(defaultAttempts int32) *temporal.RetryPolicy { - attempts := policy.MaxAttempts - if attempts == 0 { - attempts = defaultAttempts - } + attempts := policy.attempts(defaultAttempts) _, maxInterval := policy.intervals() temporalPolicy := &temporal.RetryPolicy{ diff --git a/retry_internal_test.go b/retry_internal_test.go index dcf76ce..2c3615c 100644 --- a/retry_internal_test.go +++ b/retry_internal_test.go @@ -2,6 +2,7 @@ package gush import ( "errors" + "math" "testing" "time" ) @@ -19,7 +20,7 @@ func TestRetryPolicyValidate(t *testing.T) { "unspecified timeout": {NoRetry: []string{"TemporalTimeout:Unspecified"}}, } for name, policy := range invalid { - err := policy.validate() + err := policy.validate(defaultStepAttempts) if !errors.Is(err, errInvalidConfig) { t.Errorf("%s: validate() = %v, want an invalid config error", name, err) } @@ -36,12 +37,36 @@ func TestRetryPolicyValidate(t *testing.T) { "another timeout": {NoRetry: []string{"TemporalTimeout:Heartbeat"}}, } for name, policy := range valid { - if err := policy.validate(); err != nil { + if err := policy.validate(defaultStepAttempts); err != nil { t.Errorf("%s: validate() = %v", name, err) } } } +// Temporal does not look at NoRetry when there is only one attempt, so flows that do this run +// today, and rejecting it would fail them on replay. +func TestRetryPolicyValidateAcceptsAnUnknownTimeoutWithOneAttempt(t *testing.T) { + unknown := []string{"TemporalTimeout:Soon"} + + explicit := RetryPolicy{MaxAttempts: 1, NoRetry: unknown} + if err := explicit.validate(defaultStepAttempts); err != nil { + t.Errorf("one attempt on a step: validate() = %v", err) + } + + byDefault := RetryPolicy{NoRetry: unknown} + if err := byDefault.validate(defaultFlowAttempts); err != nil { + t.Errorf("a flow's default of one attempt: validate() = %v", err) + } + if err := byDefault.validate(defaultStepAttempts); !errors.Is(err, errInvalidConfig) { + t.Errorf("a step's default of ten attempts: validate() = %v, want an invalid config error", err) + } + + retried := RetryPolicy{MaxAttempts: 2, NoRetry: unknown} + if err := retried.validate(defaultFlowAttempts); !errors.Is(err, errInvalidConfig) { + t.Errorf("two attempts on a flow: validate() = %v, want an invalid config error", err) + } +} + func TestTemporalPolicyAttempts(t *testing.T) { tests := map[int32]int32{0: 10, 3: 3, UnlimitedAttempts: 0} for attempts, expected := range tests { @@ -85,8 +110,16 @@ func TestStepConfigValidate(t *testing.T) { "no attempt timeout": func(config *StepConfig) { config.AttemptTimeout = 0 }, "negative attempt timeout": func(config *StepConfig) { config.AttemptTimeout = -time.Second }, "negative heartbeat": func(config *StepConfig) { config.HeartbeatTimeout = -time.Second }, - "negative pickup": func(config *StepConfig) { config.PickupTimeout = -time.Second }, "a bad retry policy": func(config *StepConfig) { config.MaxAttempts = -2 }, + "negative attempt timeout with a pickup timeout": func(config *StepConfig) { + config.AttemptTimeout = -time.Second + config.PickupTimeout = time.Minute + }, + "no attempt timeout and no overall limit": func(config *StepConfig) { + config.AttemptTimeout = 0 + config.PickupTimeout = time.Minute + config.MaxAttempts = UnlimitedAttempts + }, } for name, change := range invalid { config := DefaultStepConfig() @@ -97,12 +130,30 @@ func TestStepConfigValidate(t *testing.T) { } } - config := DefaultStepConfig() - config.HeartbeatTimeout = time.Second - config.PickupTimeout = time.Second - err := config.validate() - if err != nil { - t.Errorf("valid settings: validate() = %v", err) + valid := map[string]func(config *StepConfig){ + "every timeout": func(config *StepConfig) { + config.HeartbeatTimeout = time.Second + config.PickupTimeout = time.Second + }, + // Gush sends Temporal no pickup timeout for this. + "negative pickup": func(config *StepConfig) { config.PickupTimeout = -time.Second }, + // The overall limit is the timeout Temporal needs. + "a pickup timeout instead of an attempt timeout": func(config *StepConfig) { + config.AttemptTimeout = 0 + config.PickupTimeout = 30 * time.Second + }, + "attempts too many to total": func(config *StepConfig) { + config.MaxAttempts = math.MaxInt32 + config.PickupTimeout = time.Minute + }, + } + for name, change := range valid { + config := DefaultStepConfig() + change(config) + err := config.validate() + if err != nil { + t.Errorf("%s: validate() = %v", name, err) + } } } diff --git a/step_config.go b/step_config.go index 6f0a65d..165931a 100644 --- a/step_config.go +++ b/step_config.go @@ -13,14 +13,16 @@ type StepConfig struct { Queue string `json:"queue"` // AttemptTimeout limits each attempt, not the step as a whole. It also bounds how long - // a suspended step waits for Resume. + // a suspended step waits for Resume. A negative one is an error, and so is zero, unless + // PickupTimeout and a limited number of attempts give the step an overall limit. AttemptTimeout time.Duration `json:"attempt_timeout"` // HeartbeatTimeout fails an attempt that goes this long without calling Heartbeat. - // Zero disables it. + // Zero disables it, and a negative one is an error. HeartbeatTimeout time.Duration `json:"heartbeat_timeout"` - // PickupTimeout fails the step if no worker picks it up in time. Zero means unlimited. + // PickupTimeout fails the step if no worker picks it up in time. Zero or negative means + // unlimited. PickupTimeout time.Duration `json:"pickup_timeout"` // WaitForCancel makes RunStep wait for a canceled step to end. See WithWaitForCancel. @@ -54,18 +56,22 @@ func (option stepOption) applyStep(config *StepConfig) { option(config) } -// validate rejects settings Temporal cannot run, which would leave the flow Running with its -// tasks failing. +// validate rejects settings Temporal refuses, judging the timeouts it would actually receive. +// See RetryPolicy.validate for why it must not go further. func (config *StepConfig) validate() error { + err := config.RetryPolicy.validate(defaultStepAttempts) + if err != nil { + return err + } + + options := newActivityOptions(config) switch { - case config.AttemptTimeout <= 0: + case options.StartToCloseTimeout < 0, options.StartToCloseTimeout == 0 && options.ScheduleToCloseTimeout == 0: return fmt.Errorf("%w: attempt timeout %v must be positive", errInvalidConfig, config.AttemptTimeout) - case config.HeartbeatTimeout < 0: + case options.HeartbeatTimeout < 0: return fmt.Errorf("%w: heartbeat timeout %v must not be negative", errInvalidConfig, config.HeartbeatTimeout) - case config.PickupTimeout < 0: - return fmt.Errorf("%w: pickup timeout %v must not be negative", errInvalidConfig, config.PickupTimeout) } - return config.RetryPolicy.validate() + return nil } // WithQueue runs the step on another task queue. diff --git a/temporal_options.go b/temporal_options.go index 06bd6ba..46a6b04 100644 --- a/temporal_options.go +++ b/temporal_options.go @@ -1,8 +1,6 @@ package gush import ( - "time" - "go.temporal.io/api/enums/v1" "go.temporal.io/sdk/client" "go.temporal.io/sdk/temporal" @@ -79,8 +77,12 @@ func newActivityOptions(config *StepConfig) workflow.ActivityOptions { if retryPolicy.MaximumAttempts > 0 { resolved := config.RetryPolicy resolved.MaxAttempts = retryPolicy.MaximumAttempts - perAttempt := config.PickupTimeout + config.AttemptTimeout - options.ScheduleToCloseTimeout = time.Duration(resolved.MaxAttempts)*perAttempt + resolved.totalBackoff() + perAttempt := addDurations(config.PickupTimeout, config.AttemptTimeout) + total := addDurations(scaleDuration(resolved.MaxAttempts, perAttempt), resolved.totalBackoff()) + // A total too long to hold is as good as unlimited attempts. + if total < maxDuration { + options.ScheduleToCloseTimeout = total + } } } return options diff --git a/temporal_options_test.go b/temporal_options_test.go index 45670af..09fec4f 100644 --- a/temporal_options_test.go +++ b/temporal_options_test.go @@ -2,6 +2,7 @@ package gush import ( + "math" "testing" "time" @@ -118,6 +119,22 @@ func TestPickupTimeoutWithUnlimitedAttemptsHasNoOverallLimit(t *testing.T) { } } +// Temporal refuses a negative timeout, which is what the total would become by overflowing. +func TestPickupTimeoutWithAttemptsTooManyToTotalHasNoOverallLimit(t *testing.T) { + config := DefaultStepConfig() + config.PickupTimeout = time.Minute + config.MaxAttempts = math.MaxInt32 + + options := newActivityOptions(config) + + if options.ScheduleToCloseTimeout != 0 { + t.Errorf("schedule to close = %v", options.ScheduleToCloseTimeout) + } + if options.RetryPolicy.MaximumAttempts != math.MaxInt32 { + t.Errorf("attempts sent to Temporal = %d", options.RetryPolicy.MaximumAttempts) + } +} + func TestZeroAttemptsMeansTheDefaultOnAFlow(t *testing.T) { config := DefaultFlowConfig() config.MaxAttempts = 0 From 938d09a7de47f2e9f4432fbf7b75382d4634d86c Mon Sep 17 00:00:00 2001 From: Randy <857271370@qq.com> Date: Thu, 5 Nov 2026 22:01:37 +0800 Subject: [PATCH 09/11] =?UTF-8?q?=F0=9F=93=9D=20docs:=20describe=20the=20s?= =?UTF-8?q?ettings=20errors=20and=20empty=20pointer=20outputs?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index ca3734f..d2e2948 100644 --- a/README.md +++ b/README.md @@ -106,7 +106,7 @@ err := ctx.RunFlow(request) Flows and steps are registered under their bare Go function names, and can be called by that name as a string. Two functions with the same name in different packages collide, and `Serve` reports it. Use `NewFlowHandler` and `NewStepHandler` to choose a name, which closures and the instantiations of a generic function need too. -Strings and byte slices, through any number of pointers and including types built on them, travel as raw bytes unless the type has its own JSON or text methods. Everything else travels as JSON, so read a value back as the type it was written. Adding or removing those methods changes how such a type encodes, so let flows carrying it finish first. +Strings and byte slices, through any number of pointers and including types built on them, travel as raw bytes unless the type has its own JSON or text methods. Everything else travels as JSON, so read a value back as the type it was written. A pointer output stays nil when the output is empty. Adding or removing those methods changes how such a type encodes, so let flows carrying it finish first, and do the same when release notes say an upgrade changes how values travel. ### Parallel steps and cleanup @@ -156,11 +156,11 @@ A step that is already running only finds out about the cancellation through `ct | `WithLogAttributes(m)` | both | The same, for several attributes. | | `WithID(id)` | flow | Flow ID. Otherwise one is generated, which `StartFlow` returns. A client's `RunFlow` or `StartFlow` joins a flow already running with this ID and ignores this run's flow, input and settings, so concurrent callers share one run. A child flow gets an error that `gush.IsFlowRunningError` recognizes. | | `WithErrorIfRunning()` | flow | A client's `RunFlow` or `StartFlow` returns that same error instead of joining. | -| `WithTimeout(d)` | flow | Limit on the whole execution, including every retry. | +| `WithTimeout(d)` | flow | Limit on the whole execution, including every retry. A negative one is an error. | | `WithConfig(c)` | flow | Starts from a `FlowConfig`, such as one decoded from a queue. Settings left out of the JSON, or zero, use their defaults, and durations in JSON are integer nanoseconds. `WithErrorIfRunning` stays on once set. Whoever applies it chooses `RunFlow` or `StartFlow`. | | `WithQueue(q)` | step | Runs on another queue. | -| `WithAttemptTimeout(d)` | step | Limit on each attempt, not the step as a whole. | -| `WithHeartbeatTimeout(d)` | step | Fails an attempt that stops calling `ctx.Heartbeat()`. | +| `WithAttemptTimeout(d)` | step | Limit on each attempt, not the step as a whole. A negative one is an error, and so is zero without `WithPickupTimeout`. | +| `WithHeartbeatTimeout(d)` | step | Fails an attempt that stops calling `ctx.Heartbeat()`. A negative one is an error. | | `WithPickupTimeout(d)` | step | Fails if no worker picks the step up in time. | | `WithWaitForCancel()` | step | When the flow is canceled, waits for the running step to end instead of returning at once. | @@ -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. -Errors from `RunFlow` and `RunStep` work with `errors.As`, `gush.IsApplicationError`, `gush.IsPermanentError` and `gush.IsPanicError`. `gush.IsInvalidConfigError` recognizes a flow or step refused because of settings Temporal could never run, so retrying them 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 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. ## Priority and fairness From 997bfa0e7fbafd86e3ae9f051a76aeef7ebd40df Mon Sep 17 00:00:00 2001 From: Randy <857271370@qq.com> Date: Thu, 5 Nov 2026 22:15:33 +0800 Subject: [PATCH 10/11] =?UTF-8?q?=F0=9F=90=9B=20fix:=20recognize=20invalid?= =?UTF-8?q?=20settings=20through=20wrappers=20and=20keep=20a=20limit=20on?= =?UTF-8?q?=20overflow?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 2 +- error.go | 11 ++++++++++- error_test.go | 9 +++++++-- flow_config.go | 3 ++- integration_test.go | 17 +++++++++++++---- retry.go | 14 +++++++++----- retry_internal_test.go | 5 +++++ step_config.go | 6 ++++-- temporal_options.go | 6 +----- temporal_options_test.go | 6 +++--- 10 files changed, 55 insertions(+), 24 deletions(-) diff --git a/README.md b/README.md index d2e2948..298d5e9 100644 --- a/README.md +++ b/README.md @@ -159,7 +159,7 @@ A step that is already running only finds out about the cancellation through `ct | `WithTimeout(d)` | flow | Limit on the whole execution, including every retry. A negative one is an error. | | `WithConfig(c)` | flow | Starts from a `FlowConfig`, such as one decoded from a queue. Settings left out of the JSON, or zero, use their defaults, and durations in JSON are integer nanoseconds. `WithErrorIfRunning` stays on once set. Whoever applies it chooses `RunFlow` or `StartFlow`. | | `WithQueue(q)` | step | Runs on another queue. | -| `WithAttemptTimeout(d)` | step | Limit on each attempt, not the step as a whole. A negative one is an error, and so is zero without `WithPickupTimeout`. | +| `WithAttemptTimeout(d)` | step | Limit on each attempt, not the step as a whole. A negative one is an error, and so is zero, unless `WithPickupTimeout` and a limited number of attempts give the step an overall limit. | | `WithHeartbeatTimeout(d)` | step | Fails an attempt that stops calling `ctx.Heartbeat()`. A negative one is an error. | | `WithPickupTimeout(d)` | step | Fails if no worker picks the step up in time. | | `WithWaitForCancel()` | step | When the flow is canceled, waits for the running step to end instead of returning at once. | diff --git a/error.go b/error.go index 6c7b53d..1a225c6 100644 --- a/error.go +++ b/error.go @@ -61,8 +61,17 @@ func IsInvalidConfigError(err error) bool { if errors.As(err, &rejected) { 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. var applicationError *ApplicationError - return errors.As(err, &applicationError) && applicationError.Type() == invalidConfigErrorType + for errors.As(err, &applicationError) { + if applicationError.Type() == invalidConfigErrorType && applicationError.NonRetryable() { + return true + } + err = applicationError.Unwrap() + } + return false } // IsPanicError reports whether err is a PanicError. diff --git a/error_test.go b/error_test.go index b0a9c2b..a7f0647 100644 --- a/error_test.go +++ b/error_test.go @@ -413,15 +413,20 @@ func TestIsInvalidConfigError(t *testing.T) { rejected := serviceerror.NewInvalidArgument("the server said no") // What a flow that failed on its settings looks like once it leaves the flow. failedFlow := NewPermanentError("invalid config", invalidConfigErrorType, nil) + // A flow that wrapped that failure in an error of its own before returning it. + wrappedByFlow := NewPermanentError("order failed", "OrderFailed", failedFlow) + // An application's own errors, one of them retryable and sharing the reserved type. + sameType := NewApplicationError("settings not loaded yet", invalidConfigErrorType) + applicationType := NewPermanentError("settings not loaded yet", "InvalidConfig", nil) - recognized := []error{failedFlow, fmt.Errorf("run: %w", failedFlow), invalid, fmt.Errorf("start: %w", invalid), rejected, fmt.Errorf("start: %w", rejected)} + recognized := []error{failedFlow, wrappedByFlow, fmt.Errorf("run: %w", failedFlow), invalid, fmt.Errorf("start: %w", invalid), rejected, fmt.Errorf("start: %w", rejected)} for _, err := range recognized { if !IsInvalidConfigError(err) { t.Errorf("IsInvalidConfigError(%v) = false", err) } } - others := []error{nil, errors.New("some other failure"), serviceerror.NewUnavailable("try later")} + others := []error{sameType, applicationType, nil, errors.New("some other failure"), serviceerror.NewUnavailable("try later")} for _, err := range others { if IsInvalidConfigError(err) { t.Errorf("IsInvalidConfigError(%v) = true", err) diff --git a/flow_config.go b/flow_config.go index 3acfb95..d933f54 100644 --- a/flow_config.go +++ b/flow_config.go @@ -112,7 +112,8 @@ func WithErrorIfRunning() FlowOption { return flowOption(apply) } -// WithTimeout limits the whole flow execution, including every retry. +// WithTimeout limits the whole flow execution, including every retry. A negative timeout is an +// error. func WithTimeout(timeout time.Duration) FlowOption { apply := func(config *FlowConfig) { config.Timeout = timeout diff --git a/integration_test.go b/integration_test.go index 45ec0f2..c5db13d 100644 --- a/integration_test.go +++ b/integration_test.go @@ -84,8 +84,8 @@ func runWithTemporal(m *testing.M) int { serveContext, stopServing := context.WithCancel(context.Background()) testWorker := gush.NewWorker(testQueue). - WithFlows(InvalidStepTimeoutFlow, InvalidNoRetryFlow, InvalidChildTimeoutFlow, InvalidGrandchildFlow, PointerOutputFlow). - WithFlows(NegativePickupFlow, PickupInsteadOfAttemptFlow, OneAttemptNoRetryFlow, ChildNoRetryFlow, TooManyAttemptsFlow). + WithFlows(InvalidStepTimeoutFlow, InvalidNoRetryFlow, InvalidChildTimeoutFlow, InvalidGrandchildFlow, WrappedInvalidSettingsFlow, PointerOutputFlow). + WithFlows(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, @@ -1807,6 +1807,11 @@ func InvalidChildTimeoutFlow(ctx *gush.FlowContext) (any, error) { return nil, ctx.RunFlow(request, gush.WithTimeout(-time.Second)) } +func WrappedInvalidSettingsFlow(ctx *gush.FlowContext) (any, error) { + err := ctx.RunStep(DoubleStep, order{ID: "wrapped", Amount: 1}, nil, gush.WithHeartbeatTimeout(-time.Second)) + return nil, ctx.Abort(err) +} + func InvalidGrandchildFlow(ctx *gush.FlowContext) (any, error) { request := gush.FlowRequest{Flow: InvalidStepTimeoutFlow} return nil, ctx.RunFlow(request) @@ -1816,7 +1821,7 @@ func InvalidGrandchildFlow(ctx *gush.FlowContext) (any, error) { func TestSettingsTemporalRefusesFailTheFlow(t *testing.T) { ctx := testContext(t) - flows := []any{InvalidStepTimeoutFlow, InvalidNoRetryFlow, InvalidChildTimeoutFlow, InvalidGrandchildFlow} + flows := []any{InvalidStepTimeoutFlow, InvalidNoRetryFlow, InvalidChildTimeoutFlow, InvalidGrandchildFlow, WrappedInvalidSettingsFlow} for _, flow := range flows { request := flowRequest(flow, nil, nil) err := testClient.RunFlow(ctx, request) @@ -1848,6 +1853,10 @@ func ChildNoRetryFlow(ctx *gush.FlowContext) (any, error) { return nil, ctx.RunFlow(request, gush.WithNoRetry("TemporalTimeout:Soon")) } +func TooManyAttemptsOnlyPickupFlow(ctx *gush.FlowContext) (any, error) { + return nil, ctx.RunStep(DoubleStep, order{ID: "too-many-pickup", Amount: 1}, nil, gush.WithAttemptTimeout(0), gush.WithMaxAttempts(math.MaxInt32), gush.WithPickupTimeout(30*time.Minute)) +} + func TooManyAttemptsFlow(ctx *gush.FlowContext) (any, error) { return nil, ctx.RunStep(DoubleStep, order{ID: "too-many", Amount: 1}, nil, gush.WithMaxAttempts(math.MaxInt32), gush.WithPickupTimeout(time.Minute)) } @@ -1857,7 +1866,7 @@ func TooManyAttemptsFlow(ctx *gush.FlowContext) (any, error) { func TestSettingsTemporalAcceptsKeepRunning(t *testing.T) { ctx := testContext(t) - flows := []any{NegativePickupFlow, PickupInsteadOfAttemptFlow, OneAttemptNoRetryFlow, ChildNoRetryFlow, TooManyAttemptsFlow} + flows := []any{NegativePickupFlow, PickupInsteadOfAttemptFlow, OneAttemptNoRetryFlow, ChildNoRetryFlow, TooManyAttemptsFlow, TooManyAttemptsOnlyPickupFlow} for _, flow := range flows { request := flowRequest(flow, nil, nil) err := testClient.RunFlow(ctx, request) diff --git a/retry.go b/retry.go index a70b0f2..62f5277 100644 --- a/retry.go +++ b/retry.go @@ -21,12 +21,14 @@ const ( ) // timeoutErrorTypePrefix marks a NoRetry entry that names a timeout rather than an -// ApplicationError type, and Temporal rejects one naming an unknown timeout. +// ApplicationError type, and with more than one attempt Temporal rejects one naming an unknown +// timeout. const timeoutErrorTypePrefix = "TemporalTimeout:" // invalidConfigErrorType is the ApplicationError type of a flow that failed on settings -// validate rejected, which lets IsInvalidConfigError recognize it after it leaves the flow. -const invalidConfigErrorType = "InvalidConfig" +// validate rejected, which lets IsInvalidConfigError recognize it after it leaves the flow. It +// carries the package's name so that an application's own error type does not match it. +const invalidConfigErrorType = "GushInvalidConfig" // errInvalidConfig marks settings that Temporal refuses. See IsInvalidConfigError. var errInvalidConfig = errors.New("gush: invalid config") @@ -63,8 +65,10 @@ type RetryPolicy struct { // failing, and an attempt count that only a mistake produces, such as a budget computed below // zero, which must not turn into retrying forever. // -// Flow code calls it, so it must not reject what Temporal accepts: a refused command leaves no -// history for a replay to meet, while a stricter check fails a flow in flight that ran it. +// Flow code calls it, so a new check must not reject what Temporal accepts: a refused command +// leaves no history for a replay to meet, while a stricter check fails a flow in flight that +// ran it. The checks on the backoff settings are older than this rule, and loosening them would +// change what replays see. func (policy RetryPolicy) validate(defaultAttempts int32) error { switch { case policy.MaxAttempts < UnlimitedAttempts: diff --git a/retry_internal_test.go b/retry_internal_test.go index 2c3615c..14e35d1 100644 --- a/retry_internal_test.go +++ b/retry_internal_test.go @@ -146,6 +146,11 @@ func TestStepConfigValidate(t *testing.T) { config.MaxAttempts = math.MaxInt32 config.PickupTimeout = time.Minute }, + "attempts too many to total and no attempt timeout": func(config *StepConfig) { + config.AttemptTimeout = 0 + config.MaxAttempts = math.MaxInt32 + config.PickupTimeout = 30 * time.Minute + }, } for name, change := range valid { config := DefaultStepConfig() diff --git a/step_config.go b/step_config.go index 165931a..9443de3 100644 --- a/step_config.go +++ b/step_config.go @@ -93,7 +93,8 @@ func WithWaitForCancel() StepOption { return stepOption(apply) } -// WithAttemptTimeout limits each attempt of the step. +// WithAttemptTimeout limits each attempt of the step. See StepConfig.AttemptTimeout for what is +// an error. func WithAttemptTimeout(timeout time.Duration) StepOption { apply := func(config *StepConfig) { config.AttemptTimeout = timeout @@ -101,7 +102,8 @@ func WithAttemptTimeout(timeout time.Duration) StepOption { return stepOption(apply) } -// WithHeartbeatTimeout fails an attempt that goes this long without calling Heartbeat. +// WithHeartbeatTimeout fails an attempt that goes this long without calling Heartbeat. A negative +// timeout is an error. func WithHeartbeatTimeout(timeout time.Duration) StepOption { apply := func(config *StepConfig) { config.HeartbeatTimeout = timeout diff --git a/temporal_options.go b/temporal_options.go index 46a6b04..4a14438 100644 --- a/temporal_options.go +++ b/temporal_options.go @@ -78,11 +78,7 @@ func newActivityOptions(config *StepConfig) workflow.ActivityOptions { resolved := config.RetryPolicy resolved.MaxAttempts = retryPolicy.MaximumAttempts perAttempt := addDurations(config.PickupTimeout, config.AttemptTimeout) - total := addDurations(scaleDuration(resolved.MaxAttempts, perAttempt), resolved.totalBackoff()) - // A total too long to hold is as good as unlimited attempts. - if total < maxDuration { - options.ScheduleToCloseTimeout = total - } + options.ScheduleToCloseTimeout = addDurations(scaleDuration(resolved.MaxAttempts, perAttempt), resolved.totalBackoff()) } } return options diff --git a/temporal_options_test.go b/temporal_options_test.go index 09fec4f..b5ba59d 100644 --- a/temporal_options_test.go +++ b/temporal_options_test.go @@ -119,15 +119,15 @@ func TestPickupTimeoutWithUnlimitedAttemptsHasNoOverallLimit(t *testing.T) { } } -// Temporal refuses a negative timeout, which is what the total would become by overflowing. -func TestPickupTimeoutWithAttemptsTooManyToTotalHasNoOverallLimit(t *testing.T) { +// Temporal refuses a negative timeout, which is what the total could become by overflowing. +func TestPickupTimeoutWithAttemptsTooManyToTotalStopsAtTheLongestLimit(t *testing.T) { config := DefaultStepConfig() config.PickupTimeout = time.Minute config.MaxAttempts = math.MaxInt32 options := newActivityOptions(config) - if options.ScheduleToCloseTimeout != 0 { + if options.ScheduleToCloseTimeout != maxDuration { t.Errorf("schedule to close = %v", options.ScheduleToCloseTimeout) } if options.RetryPolicy.MaximumAttempts != math.MaxInt32 { From 66f8bb2bd31fb72694761917dc6c0f6acaeae766 Mon Sep 17 00:00:00 2001 From: Randy <857271370@qq.com> Date: Thu, 5 Nov 2026 22:31:02 +0800 Subject: [PATCH 11/11] =?UTF-8?q?=F0=9F=93=9D=20docs:=20note=20the=20limit?= =?UTF-8?q?s=20of=20invalid=20config=20recognition=20and=20timeout=20names?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 2 +- error.go | 4 +++- options.go | 3 ++- retry.go | 1 + 4 files changed, 7 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 298d5e9..536760a 100644 --- a/README.md +++ b/README.md @@ -185,7 +185,7 @@ return nil, gush.NewPermanentError("card declined", "Declined", err) 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. +`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. diff --git a/error.go b/error.go index 1a225c6..477cd3d 100644 --- a/error.go +++ b/error.go @@ -52,7 +52,9 @@ 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. +// 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. func IsInvalidConfigError(err error) bool { if errors.Is(err, errInvalidConfig) { return true diff --git a/options.go b/options.go index 999503b..f90be3f 100644 --- a/options.go +++ b/options.go @@ -89,7 +89,8 @@ func WithMaxAttempts(attempts int32) Option { return commonOption(apply) } -// WithNoRetry adds ApplicationError types that fail immediately instead of retrying. +// WithNoRetry adds ApplicationError types that fail immediately instead of retrying. See +// RetryPolicy.NoRetry for entries that name a timeout, where an unknown name is an error. func WithNoRetry(errorTypes ...string) Option { apply := func(config *CommonConfig) { config.NoRetry = append(config.NoRetry, errorTypes...) diff --git a/retry.go b/retry.go index 62f5277..d142d0b 100644 --- a/retry.go +++ b/retry.go @@ -58,6 +58,7 @@ type RetryPolicy struct { // NoRetry lists the ApplicationError types that fail immediately. An entry starting with // TemporalTimeout: names a timeout instead, such as StepTimeoutErrorType. Naming a timeout // Temporal does not know is an error, unless there is only one attempt and so no retry. + // Use the short name: Temporal also accepts TIMEOUT_TYPE_START_TO_CLOSE, which never matches. NoRetry []string `json:"no_retry"` }