Fail flows on settings Temporal refuses and close raw value gaps

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. 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. |
-| `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`. |
+| `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, 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. |
 
@@ -185,9 +185,9 @@ 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`. 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
 
@@ -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

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.

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

error.go

@@ -5,6 +5,7 @@ import (
 	"regexp"
 	"strings"
 
+	"go.temporal.io/api/serviceerror"
 	"go.temporal.io/sdk/temporal"
 )
 
@@ -48,6 +49,33 @@ func IsFlowRunningError(err error) bool {
 	return temporal.IsWorkflowExecutionAlreadyStartedError(err)
 }
 
+// IsInvalidConfigError reports whether a request was refused as invalid, so sending it again
+// cannot succeed: Gush found settings Temporal refuses, including in a flow that has since
+// failed on them, or the server answered InvalidArgument. The server gives that answer for
+// more than settings, such as an input over its size limit. Past a flow it sees through Abort,
+// NewPermanentError and %w, but not errors.Join or several %w, and not a client's refusal inside
+// a step, whose text is all that crosses the step.
+func IsInvalidConfigError(err error) bool {
+	if errors.Is(err, errInvalidConfig) {
+		return true
+	}
+	var rejected *serviceerror.InvalidArgument
+	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
+	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.
 func IsPanicError(err error) bool {
 	return temporal.IsPanicError(err)

error_test.go

@@ -403,3 +403,33 @@ 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")
+	// 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, 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{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)
+		}
+	}
+}