From 03243f29fd9bf47b31e100c58f3ee4e32efb6d2f Mon Sep 17 00:00:00 2001 From: Alex Mazzeo Date: Wed, 2 Sep 2026 14:19:52 -0700 Subject: [PATCH 1/7] Add generated spec based on Python, .NET, and Java prototype behavior for collaborative review --- features/transfer_types/spec.md | 172 ++++++++++++++++++++++++++++++++ 1 file changed, 172 insertions(+) create mode 100644 features/transfer_types/spec.md diff --git a/features/transfer_types/spec.md b/features/transfer_types/spec.md new file mode 100644 index 00000000..3ea246a1 --- /dev/null +++ b/features/transfer_types/spec.md @@ -0,0 +1,172 @@ +# Transfer Type Converter Behavioral Specification + +Status: Draft + +Last updated: 2026-09-02 + +## Purpose + +This document defines the portable behavior of Temporal transfer type converters. It specifies observable conversion semantics without prescribing SDK-specific APIs, generic type mechanisms, or converter construction strategies. + +## Terminology + +- **Model type:** The user-facing type associated with a transfer type converter. +- **Transfer type:** The serializer-facing type produced from a model value. +- **Transfer value:** A value of the transfer type. +- **Payload conversion:** Conversion between an in-memory value and a Temporal `Payload`. +- **Payload codec:** A byte-to-byte transformation applied to a payload, such as encryption or compression. +- **Failure details:** Typed payloads stored in application failures, cancellations, timeout heartbeat details, reset failures, or nested failure causes. + +## Normative behavior + +### Conversion order + +For outbound values, an SDK MUST apply operations in this order: + +1. Locate the transfer converter from the top-level model value's runtime type. +2. Invoke the transfer converter when one is present. +3. Pass the resulting transfer value to the configured payload converter. +4. Apply external storage processing and payload codecs according to the SDK's existing data-converter contract. + +For inbound values, an SDK MUST apply the inverse order: + +1. Apply payload codecs and external storage retrieval according to the SDK's existing data-converter contract. +2. Locate the transfer converter from the requested model type. +3. Ask the configured payload converter to decode the payload as the transfer type. +4. Invoke the transfer converter to reconstruct the requested model value. + +The transfer layer MUST NOT replace or bypass the configured payload converter, failure converter, payload codecs, external storage provider, or serialization context. + +### Top-level values only + +Transfer conversion MUST apply only to values directly handed to the SDK payload-conversion boundary. It MUST NOT recursively inspect fields, properties, collection elements, or other nested values. + +Nested serialization remains the responsibility of the configured payload converter. + +### Exactly one transfer step + +An SDK MUST perform at most one transfer conversion for a top-level value. + +If model type `A` converts to transfer value `B`, and `B` also has a transfer converter, the SDK MUST pass `B` directly to the configured payload converter. It MUST NOT invoke the converter associated with `B`. + +On decode, the payload converter MUST decode directly to the selected transfer type. Only the converter associated with the requested model type may reconstruct the final value. + +### Encode and decode type selection + +Encoding MUST select a converter using the top-level value's runtime type. A null outbound value has no runtime model type and MUST pass directly to the configured payload converter. + +Decoding MUST select a converter using the requested user-facing type. If the SDK has no requested type information, it MUST return the configured payload converter's normal result without applying transfer reconstruction. + +The SDK MUST provide the requested model type information available in its type system when selecting the transfer type and reconstructing the model value. This includes generic arguments when the SDK's conversion API preserves them. + +### Converter declaration and lookup + +A transfer converter declaration applies only to the exact model type on which it is declared. Declarations MUST NOT be inherited from a base type. + +A subclass or other derived type MAY declare its own converter independently of its base type. + +Runtime encoding MUST look only for a converter declared by the exact runtime type. Inbound decoding MUST look only for a converter declared by the exact requested type. + +Every selected converter MUST provide a concrete, non-null transfer type for inbound payload conversion. The SDK MUST fail clearly if the converter does not provide a valid transfer type. It MUST NOT fall back to payload metadata inference after selecting a converter. + +### Transfer value nullability + +When a converter has been selected for an inbound model type, the SDK MUST invoke its reconstruction hook even when the decoded transfer value is null. + +This permits a non-null model value to use null as its transfer representation. + +An outbound null model value MUST pass through without converter lookup because no runtime model type is available. + +### Failure conversion + +Failure semantics MUST remain owned by the configured failure converter. + +The failure converter MUST use the same transfer-aware payload-conversion path used for ordinary arguments and results when encoding or decoding typed failure payloads. This includes: + +- Application failure details. +- Cancellation details. +- Timeout last-heartbeat details. +- Reset workflow failure details. +- Details contained in nested failure causes. + +Payload codecs and external storage transformations MUST be applied exactly once to failure payloads and in their normal order. + +Transfer conversion does not apply to untyped failure metadata such as messages, stack traces, failure types, retry state, or encoded failure attributes. Existing payload-codec behavior for encoded failure attributes remains unchanged. + +### Raw payloads + +A raw payload wrapper MUST preserve its existing pass-through behavior. A transfer converter MUST NOT reinterpret a raw payload wrapper as a model value. + +### Configured converter authority + +The configured payload converter remains authoritative over the transfer representation's wire format. Transfer conversion MUST NOT add transfer-specific payload metadata or choose a wire encoding independently. + +The configured failure converter remains authoritative over the mapping between language exceptions and Temporal failure protos. + +### Serialization context + +Transfer conversion MUST preserve serialization context propagation to payload converters, failure converters, codecs, and external storage components. + +### Converter lifetime and concurrency + +Converter construction, lookup, and invocation MUST be safe under concurrent serialization. An SDK MAY reuse converter instances, so converter implementations MUST satisfy the concurrency requirements documented by that SDK. + +The number of converter instances and the time at which they are constructed are not part of the portable behavior. + +### Invalid declarations and callback failures + +An invalid converter declaration MUST fail clearly before conversion completes. Examples include an incompatible converter, a converter that cannot be constructed according to the SDK's API, or an invalid transfer type. + +Exceptions raised by transfer callbacks MUST remain observable as conversion failures. The SDK MUST NOT silently fall back to ordinary model serialization after selecting a converter. + +## Conformance requirements + +An implementation conforms to this specification when all of the following hold: + +- Ordinary arguments and results satisfy the normative conversion order. +- Failure details use transfer conversion without bypassing a configured failure converter. +- Payload codecs and external storage transformations are applied exactly once. +- A selected converter reconstructs from a null transfer value. +- Top-level-only and one-step behavior have explicit regression tests. +- Raw payload pass-through remains intact. +- Context propagation remains intact. +- Exact converter lookup is implemented and documented as non-inherited behavior. +- Workflow history containing transfer-represented values can be replayed. + +## Required conformance tests + +### Payload conversion + +- Annotated argument and result round trip. +- Mixed annotated and ordinary values preserve position and type. +- Missing payloads still produce language-default values. +- Available requested type information reaches converter selection and reconstruction. +- Null outbound model passes through. +- Non-null model represented by null reconstructs through its hook. +- Raw payload wrapper passes through unchanged. + +### Conversion boundaries + +- Annotated `A` converts to annotated `B` without invoking `B`'s converter. +- Annotated nested values do not invoke their converters. +- An unannotated subclass does not inherit an annotated base converter. +- A subclass can declare its own converter independently. + +### Failures + +- Application failure details round trip. +- Cancellation details round trip. +- Timeout last-heartbeat details round trip. +- Nested failure causes round trip their details. +- Custom failure converter remains authoritative. +- Payload codecs transform each failure payload exactly once. + +### Integration surfaces + +- Workflow client and worker. +- Workflow history replay. +- Activity stubs and standalone activity client. +- Local activities and activity heartbeats. +- Nexus client and worker. +- Schedule client create and describe operations. +- Test activity environment. From 4cb382ccbb36d5b642f61f4da91c6da940858da4 Mon Sep 17 00:00:00 2001 From: Alex Mazzeo Date: Wed, 2 Sep 2026 17:30:56 -0700 Subject: [PATCH 2/7] More WIP spec --- .../transfer_types/spec.md | 149 +++++++++++++++--- 1 file changed, 130 insertions(+), 19 deletions(-) rename features/{ => data_converter}/transfer_types/spec.md (60%) diff --git a/features/transfer_types/spec.md b/features/data_converter/transfer_types/spec.md similarity index 60% rename from features/transfer_types/spec.md rename to features/data_converter/transfer_types/spec.md index 3ea246a1..3415ea73 100644 --- a/features/transfer_types/spec.md +++ b/features/data_converter/transfer_types/spec.md @@ -1,55 +1,166 @@ # Transfer Type Converter Behavioral Specification -Status: Draft - Last updated: 2026-09-02 -## Purpose +## Motivation + +Transfer Type Conversion provides a way to translate to and from a Payload serialization friendly format, e.g. a protobuf object, to a more ergonomic or language idiomatic type. +TransferTypeConverters are added to the API facing type and specify the target "transfer type" which is then used by the configured Payload Converter serialize to/from a Payload. + +For example, given the protobuf definition + +```proto3 + message WorkflowExecution { + string workflow_id = 1; + string run_id = 2; + } + + message PauseRequest { + WorkflowExecution execution = 1; + string reason = 2; + } +``` + +Using a generated protobuf object in Python may force an awkward interface: + +```python +request = pause_pb2.PauseRequest(reason="maintenance") + +# This does not test presence. Protobuf returns an empty message. +assert request.execution is not None +assert request.execution.workflow_id == "" + +# You must use the protobuf specific inspection for presence. +if request.HasField("execution"): + print(request.execution.workflow_id) +else: + print("No workflow execution provided") +``` -This document defines the portable behavior of Temporal transfer type converters. It specifies observable conversion semantics without prescribing SDK-specific APIs, generic type mechanisms, or converter construction strategies. +Adding a TransferTypeConverter allows you to construct a more idiomatic type: + +```python +@dataclass(frozen=True, slots=True) +class WorkflowExecution: + workflow_id: str + run_id: str + + +# Add the PauseRequestConverter that converts to and from pause_pb2.PauseRequest +@transfer_type_convertible(PauseRequestConverter) +@dataclass(frozen=True, slots=True) +class PauseRequest: + reason: str + execution: WorkflowExecution | None = None + + +request = PauseRequest(reason="maintenance") + +# Can use normal Python optional-value semantics: +if request.execution is not None: + pause_workflow(request.execution.workflow_id) +else: + print("No workflow execution provided") +``` ## Terminology - **Model type:** The user-facing type associated with a transfer type converter. +- **Model value:** A value of the model type. - **Transfer type:** The serializer-facing type produced from a model value. - **Transfer value:** A value of the transfer type. -- **Payload conversion:** Conversion between an in-memory value and a Temporal `Payload`. -- **Payload codec:** A byte-to-byte transformation applied to a payload, such as encryption or compression. -- **Failure details:** Typed payloads stored in application failures, cancellations, timeout heartbeat details, reset failures, or nested failure causes. +- **Transfer type converter:** The conversion logic to map between the model type and the transfer type. + +## Behaviors + +### Conversion wraps configured payload converters -## Normative behavior +#### Encoding -### Conversion order +When encoding a value, the transfer type converter is applied to the model type **before** the payload converter so the payload converter receives the transfer type. The external storage processing and codec are applied as normal to the resulting payload. -For outbound values, an SDK MUST apply operations in this order: +```mermaid +sequenceDiagram + actor App as Application + participant Client + participant Converter as Transfer type converter + participant Serializer as Payload converter + participant Codec as Codec & ExtStore + participant Server as Temporal Server + + App->>+Client: Send request containing model value + + Note over Client,Converter: The SDK finds the converter
associated with the model type + + Client->>+Converter: to_transfer_type(model value) + Converter-->>-Client: Transfer value + + Client->>+Serializer: to_payload(transfer value) + Serializer-->>-Client: payload + + Client-->+Codec: process(payloads) + Codec-->-Client: encoded payloads + + Client->>-Server: Send outgoing request + +``` 1. Locate the transfer converter from the top-level model value's runtime type. 2. Invoke the transfer converter when one is present. 3. Pass the resulting transfer value to the configured payload converter. 4. Apply external storage processing and payload codecs according to the SDK's existing data-converter contract. -For inbound values, an SDK MUST apply the inverse order: +#### Decoding + +When decoding a value, external storage processing and codec are applied as normal to the payload. Then the transfer type converter is applied to the transfer type **after** the payload converter so. + +```mermaid +sequenceDiagram + actor App as Application + participant Client + participant Converter as Transfer type converter + participant Serializer as Payload converter + participant Codec as ExtStore & Codec + participant Server as Temporal Server -1. Apply payload codecs and external storage retrieval according to the SDK's existing data-converter contract. + Server->>+Client: Send incoming response + + Client->>+Codec: process(encoded payloads) + Codec-->>-Client: payloads + + Note over Client,Converter: The SDK finds the converter
associated with the model type + + Client->>+Serializer: from_payload(payload, transfer type) + Serializer-->>-Client: Transfer value + + Client->>+Converter: from_transfer_type(transfer value, model type) + Converter-->>-Client: Model value + + Client-->>-App: Return response containing model value +``` + +1. Apply payload codecs and external storage retrieval. 2. Locate the transfer converter from the requested model type. 3. Ask the configured payload converter to decode the payload as the transfer type. 4. Invoke the transfer converter to reconstruct the requested model value. -The transfer layer MUST NOT replace or bypass the configured payload converter, failure converter, payload codecs, external storage provider, or serialization context. - ### Top-level values only -Transfer conversion MUST apply only to values directly handed to the SDK payload-conversion boundary. It MUST NOT recursively inspect fields, properties, collection elements, or other nested values. +Transfer conversion only applies to top-level values and does not recursively inspect fields for transfer types. -Nested serialization remains the responsibility of the configured payload converter. +Nested serialization is the responsibility of the configured payload converter. ### Exactly one transfer step -An SDK MUST perform at most one transfer conversion for a top-level value. +Transfer type converters are applied at most once. For example, given: + +- type `A` has transfer type converter `converterA` +- `converterA` produces the transfer type `B` +- type `B` has transfer type converter `converterB` -If model type `A` converts to transfer value `B`, and `B` also has a transfer converter, the SDK MUST pass `B` directly to the configured payload converter. It MUST NOT invoke the converter associated with `B`. +When encoding type `A`, the SDK does not inspect the output type `B` for a transfer type converter and thus does not apply `converterB`. -On decode, the payload converter MUST decode directly to the selected transfer type. Only the converter associated with the requested model type may reconstruct the final value. +Similarly when decoding, the SDK will convert using only the transfer type encoder registered on the requested model type. ### Encode and decode type selection From 2aae7cf74a0629211b7f5195a13191f405022ba4 Mon Sep 17 00:00:00 2001 From: Alex Mazzeo Date: Thu, 3 Sep 2026 16:05:20 -0700 Subject: [PATCH 3/7] Edit the transfer_types spec --- .../data_converter/transfer_types/spec.md | 133 ++++-------------- 1 file changed, 24 insertions(+), 109 deletions(-) diff --git a/features/data_converter/transfer_types/spec.md b/features/data_converter/transfer_types/spec.md index 3415ea73..3d96042d 100644 --- a/features/data_converter/transfer_types/spec.md +++ b/features/data_converter/transfer_types/spec.md @@ -105,7 +105,7 @@ sequenceDiagram ``` -1. Locate the transfer converter from the top-level model value's runtime type. +1. Locate the transfer converter from the declared model type. 2. Invoke the transfer converter when one is present. 3. Pass the resulting transfer value to the configured payload converter. 4. Apply external storage processing and payload codecs according to the SDK's existing data-converter contract. @@ -150,7 +150,7 @@ Transfer conversion only applies to top-level values and does not recursively in Nested serialization is the responsibility of the configured payload converter. -### Exactly one transfer step +### At most one transfer step Transfer type converters are applied at most once. For example, given: @@ -162,122 +162,37 @@ When encoding type `A`, the SDK does not inspect the output type `B` for a trans Similarly when decoding, the SDK will convert using only the transfer type encoder registered on the requested model type. -### Encode and decode type selection +### TransferTypeConverter Selection -Encoding MUST select a converter using the top-level value's runtime type. A null outbound value has no runtime model type and MUST pass directly to the configured payload converter. +Converting to the transfer type should use the type hint for the target execution rather than the value's runtime type. For example: -Decoding MUST select a converter using the requested user-facing type. If the SDK has no requested type information, it MUST return the configured payload converter's normal result without applying transfer reconstruction. +```java + // Given this workflow + @WorkflowMethod + void run(Animal animal); -The SDK MUST provide the requested model type information available in its type system when selecting the transfer type and reconstructing the model value. This includes generic arguments when the SDK's conversion API preserves them. - -### Converter declaration and lookup - -A transfer converter declaration applies only to the exact model type on which it is declared. Declarations MUST NOT be inherited from a base type. - -A subclass or other derived type MAY declare its own converter independently of its base type. + // And this invocation + Animal value = new Dog(); + workflow.run(value); +``` -Runtime encoding MUST look only for a converter declared by the exact runtime type. Inbound decoding MUST look only for a converter declared by the exact requested type. +Here the transfer type converter associated with `Animal` should be used to +even if `Dog` also has a transfer type converter ensure decoding can match. +If the type hint does not have an associated transfer type converter, the +value is passed directly to the configured payload converter. -Every selected converter MUST provide a concrete, non-null transfer type for inbound payload conversion. The SDK MUST fail clearly if the converter does not provide a valid transfer type. It MUST NOT fall back to payload metadata inference after selecting a converter. +Converting from the transfer type should also use the type hint for the target execution. If the type hint does not have an associated transfer type converter, +the result of the configured payload converter should be used directly. -### Transfer value nullability +A transfer converter declaration applies only to the exact model type on which it is declared. Declarations are not inherited from a base type. -When a converter has been selected for an inbound model type, the SDK MUST invoke its reconstruction hook even when the decoded transfer value is null. +A subclass or other derived type can declare its own converter independently of its base type. -This permits a non-null model value to use null as its transfer representation. +### TransferType Nullability Converter -An outbound null model value MUST pass through without converter lookup because no runtime model type is available. +TransferTypeConverters should specify a non-null transfer type. ### Failure conversion -Failure semantics MUST remain owned by the configured failure converter. - -The failure converter MUST use the same transfer-aware payload-conversion path used for ordinary arguments and results when encoding or decoding typed failure payloads. This includes: - -- Application failure details. -- Cancellation details. -- Timeout last-heartbeat details. -- Reset workflow failure details. -- Details contained in nested failure causes. - -Payload codecs and external storage transformations MUST be applied exactly once to failure payloads and in their normal order. - -Transfer conversion does not apply to untyped failure metadata such as messages, stack traces, failure types, retry state, or encoded failure attributes. Existing payload-codec behavior for encoded failure attributes remains unchanged. - -### Raw payloads - -A raw payload wrapper MUST preserve its existing pass-through behavior. A transfer converter MUST NOT reinterpret a raw payload wrapper as a model value. - -### Configured converter authority - -The configured payload converter remains authoritative over the transfer representation's wire format. Transfer conversion MUST NOT add transfer-specific payload metadata or choose a wire encoding independently. - -The configured failure converter remains authoritative over the mapping between language exceptions and Temporal failure protos. - -### Serialization context - -Transfer conversion MUST preserve serialization context propagation to payload converters, failure converters, codecs, and external storage components. - -### Converter lifetime and concurrency - -Converter construction, lookup, and invocation MUST be safe under concurrent serialization. An SDK MAY reuse converter instances, so converter implementations MUST satisfy the concurrency requirements documented by that SDK. - -The number of converter instances and the time at which they are constructed are not part of the portable behavior. - -### Invalid declarations and callback failures - -An invalid converter declaration MUST fail clearly before conversion completes. Examples include an incompatible converter, a converter that cannot be constructed according to the SDK's API, or an invalid transfer type. - -Exceptions raised by transfer callbacks MUST remain observable as conversion failures. The SDK MUST NOT silently fall back to ordinary model serialization after selecting a converter. - -## Conformance requirements - -An implementation conforms to this specification when all of the following hold: - -- Ordinary arguments and results satisfy the normative conversion order. -- Failure details use transfer conversion without bypassing a configured failure converter. -- Payload codecs and external storage transformations are applied exactly once. -- A selected converter reconstructs from a null transfer value. -- Top-level-only and one-step behavior have explicit regression tests. -- Raw payload pass-through remains intact. -- Context propagation remains intact. -- Exact converter lookup is implemented and documented as non-inherited behavior. -- Workflow history containing transfer-represented values can be replayed. - -## Required conformance tests - -### Payload conversion - -- Annotated argument and result round trip. -- Mixed annotated and ordinary values preserve position and type. -- Missing payloads still produce language-default values. -- Available requested type information reaches converter selection and reconstruction. -- Null outbound model passes through. -- Non-null model represented by null reconstructs through its hook. -- Raw payload wrapper passes through unchanged. - -### Conversion boundaries - -- Annotated `A` converts to annotated `B` without invoking `B`'s converter. -- Annotated nested values do not invoke their converters. -- An unannotated subclass does not inherit an annotated base converter. -- A subclass can declare its own converter independently. - -### Failures - -- Application failure details round trip. -- Cancellation details round trip. -- Timeout last-heartbeat details round trip. -- Nested failure causes round trip their details. -- Custom failure converter remains authoritative. -- Payload codecs transform each failure payload exactly once. - -### Integration surfaces - -- Workflow client and worker. -- Workflow history replay. -- Activity stubs and standalone activity client. -- Local activities and activity heartbeats. -- Nexus client and worker. -- Schedule client create and describe operations. -- Test activity environment. +Failure conversion should use the transfer converter aware payload converter before +applying the configured payload converter to failure details. From c97b6bd2c26094e708b649338df5d9975438bfa8 Mon Sep 17 00:00:00 2001 From: Alex Mazzeo Date: Fri, 4 Sep 2026 09:19:12 -0700 Subject: [PATCH 4/7] Update features/data_converter/transfer_types/spec.md Co-authored-by: Dan Plyukhin --- features/data_converter/transfer_types/spec.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/features/data_converter/transfer_types/spec.md b/features/data_converter/transfer_types/spec.md index 3d96042d..34fb06e2 100644 --- a/features/data_converter/transfer_types/spec.md +++ b/features/data_converter/transfer_types/spec.md @@ -4,7 +4,7 @@ Last updated: 2026-09-02 ## Motivation -Transfer Type Conversion provides a way to translate to and from a Payload serialization friendly format, e.g. a protobuf object, to a more ergonomic or language idiomatic type. +Transfer Type Conversion provides a way to translate between a Payload serialization friendly format, e.g. a protobuf object, and a more ergonomic or language-idiomatic type. TransferTypeConverters are added to the API facing type and specify the target "transfer type" which is then used by the configured Payload Converter serialize to/from a Payload. For example, given the protobuf definition From 8bc17beddad8d852cd352c44c413bc2282da71c6 Mon Sep 17 00:00:00 2001 From: Alex Mazzeo Date: Fri, 4 Sep 2026 11:18:11 -0700 Subject: [PATCH 5/7] Update features/data_converter/transfer_types/spec.md Co-authored-by: Dan Plyukhin --- features/data_converter/transfer_types/spec.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/features/data_converter/transfer_types/spec.md b/features/data_converter/transfer_types/spec.md index 34fb06e2..fade2ff6 100644 --- a/features/data_converter/transfer_types/spec.md +++ b/features/data_converter/transfer_types/spec.md @@ -194,5 +194,5 @@ TransferTypeConverters should specify a non-null transfer type. ### Failure conversion -Failure conversion should use the transfer converter aware payload converter before +Failure conversion should use the transfer converter before applying the configured payload converter to failure details. From cfe48f90ec0cec37a8f84f5e9f5844257025ec30 Mon Sep 17 00:00:00 2001 From: Alex Mazzeo Date: Fri, 4 Sep 2026 13:31:30 -0700 Subject: [PATCH 6/7] Update formatting --- features/data_converter/transfer_types/spec.md | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/features/data_converter/transfer_types/spec.md b/features/data_converter/transfer_types/spec.md index fade2ff6..11feaf2a 100644 --- a/features/data_converter/transfer_types/spec.md +++ b/features/data_converter/transfer_types/spec.md @@ -73,7 +73,7 @@ else: ## Behaviors -### Conversion wraps configured payload converters +### Conversion Flow #### Encoding @@ -144,13 +144,15 @@ sequenceDiagram 3. Ask the configured payload converter to decode the payload as the transfer type. 4. Invoke the transfer converter to reconstruct the requested model value. -### Top-level values only +### SDK Behavior + +#### Top-level values only Transfer conversion only applies to top-level values and does not recursively inspect fields for transfer types. Nested serialization is the responsibility of the configured payload converter. -### At most one transfer step +#### At most one transfer step Transfer type converters are applied at most once. For example, given: @@ -162,7 +164,7 @@ When encoding type `A`, the SDK does not inspect the output type `B` for a trans Similarly when decoding, the SDK will convert using only the transfer type encoder registered on the requested model type. -### TransferTypeConverter Selection +#### TransferTypeConverter Selection Converting to the transfer type should use the type hint for the target execution rather than the value's runtime type. For example: @@ -188,11 +190,11 @@ A transfer converter declaration applies only to the exact model type on which i A subclass or other derived type can declare its own converter independently of its base type. -### TransferType Nullability Converter +#### Transfer Type Nullability TransferTypeConverters should specify a non-null transfer type. -### Failure conversion +#### Failure conversion Failure conversion should use the transfer converter before applying the configured payload converter to failure details. From db571fd720cd656ab95089bfbd0c77cd6f3e01af Mon Sep 17 00:00:00 2001 From: Dan Plyukhin Date: Fri, 4 Sep 2026 17:58:56 -0500 Subject: [PATCH 7/7] Features tests for transfer types --- cmd/run.go | 18 +- .../data_converter/transfer_types/config.json | 3 + .../data_converter/transfer_types/feature.cs | 432 +++++++++++++++++ .../data_converter/transfer_types/feature.py | 444 ++++++++++++++++++ .../data_converter/transfer_types/spec.md | 3 +- .../dotnet/Temporalio.Features.Harness/App.cs | 17 +- .../Temporalio.Features.Harness/Runner.cs | 15 +- harness/go/cmd/run.go | 3 +- harness/python/feature.py | 28 +- harness/python/main.py | 15 +- 10 files changed, 947 insertions(+), 31 deletions(-) create mode 100644 features/data_converter/transfer_types/config.json create mode 100644 features/data_converter/transfer_types/feature.cs create mode 100644 features/data_converter/transfer_types/feature.py diff --git a/cmd/run.go b/cmd/run.go index 7bd3898a..093f6ae7 100644 --- a/cmd/run.go +++ b/cmd/run.go @@ -36,7 +36,7 @@ import ( "gopkg.in/yaml.v3" ) -// nexusFeatureDirPrefix marks features that require a per-test Nexus endpoint. +// nexusFeatureDirPrefix preserves automatic endpoint creation for Nexus features. const nexusFeatureDirPrefix = "nexus/" const ( @@ -457,9 +457,8 @@ func (r *Runner) runBatch(ctx context.Context, batch runBatch) error { return err } - // Create a Nexus endpoint per feature under features/nexus/ targeting that feature's task - // queue. Endpoint names are passed to the lang harness through RunFeature.NexusEndpoint and - // the endpoints are deleted once the lang harness completes. + // Create a Nexus endpoint for Nexus features and other features that request one. Endpoint + // names are passed to the language harness and deleted once it completes. deleteEndpoints, err := r.createNexusEndpoints(ctx, config, batch.Run) if err != nil { return err @@ -853,14 +852,14 @@ func rootDir() string { return filepath.Dir(filepath.Dir(currFile)) } -// createNexusEndpoints creates a Nexus endpoint per RunFeature whose Dir is under -// features/nexus, populating RunFeature.NexusEndpoint. It returns a cleanup function that -// deletes the created endpoints. The cleanup function is always safe to call. +// createNexusEndpoints creates an endpoint for Nexus features and features that request one. +// It returns a cleanup function that is always safe to call. func (r *Runner) createNexusEndpoints(ctx context.Context, config RunConfig, run *cmd.Run) (func(), error) { noop := func() {} var nexusFeatures []*cmd.RunFeature for i := range run.Features { - if strings.HasPrefix(run.Features[i].Dir, nexusFeatureDirPrefix) { + if strings.HasPrefix(run.Features[i].Dir, nexusFeatureDirPrefix) || + run.Features[i].Config.NeedsNexusEndpoint { nexusFeatures = append(nexusFeatures, &run.Features[i]) } } @@ -930,6 +929,9 @@ func (r *Runner) createNexusEndpoints(ctx context.Context, config RunConfig, run r.log.Warn("Skipping Nexus features: server does not support Nexus endpoint creation", "Feature", feature.Dir, "Error", err) cleanup() + for i := range run.Features { + run.Features[i].NexusEndpoint = "" + } kept := run.Features[:0] for _, f := range run.Features { if !strings.HasPrefix(f.Dir, nexusFeatureDirPrefix) { diff --git a/features/data_converter/transfer_types/config.json b/features/data_converter/transfer_types/config.json new file mode 100644 index 00000000..3531cec3 --- /dev/null +++ b/features/data_converter/transfer_types/config.json @@ -0,0 +1,3 @@ +{ + "needsNexusEndpoint": true +} diff --git a/features/data_converter/transfer_types/feature.cs b/features/data_converter/transfer_types/feature.cs new file mode 100644 index 00000000..ec252ba0 --- /dev/null +++ b/features/data_converter/transfer_types/feature.cs @@ -0,0 +1,432 @@ +namespace data_converter.transfer_types; + +using System.Text.Json; +using NexusRpc; +using NexusRpc.Handlers; +using Temporalio.Activities; +using Temporalio.Client; +using Temporalio.Converters; +using Temporalio.Exceptions; +using Temporalio.Features.Harness; +using Temporalio.Worker; +using Temporalio.Workflows; +using ApiWorkflowExecution = Temporalio.Api.Common.V1.WorkflowExecution; +using Payload = Temporalio.Api.Common.V1.Payload; +using TemporalRetryPolicy = Temporalio.Common.RetryPolicy; + +class Feature : IFeature +{ + public void ConfigureWorker(Runner runner, TemporalWorkerOptions options) => + options.AddWorkflow(). + AddWorkflow(). + AddAllActivities(new TransferActivities()). + AddNexusService(new TransferServiceHandler()); + + public async Task ExecuteAsync(Runner runner) + { + var failedWorkflowId = $"{runner.PreparedFeature.Dir}-failing-{Guid.NewGuid()}"; + var failedOptions = runner.NewWorkflowOptions(); + failedOptions.Id = failedWorkflowId; + var exception = await Assert.ThrowsAsync(() => + runner.Client.StartWorkflowAsync( + (ThrowingWorkflow wf) => wf.RunAsync( + new ThrowingValue("expected transfer conversion failure")), + failedOptions)); + Assert.Equal("expected transfer conversion failure", exception.Message); + + var rpcException = await Assert.ThrowsAsync(() => + runner.Client.GetWorkflowHandle(failedWorkflowId).DescribeAsync()); + Assert.Equal(RpcException.StatusCode.NotFound, rpcException.Code); + + await ExecuteStandaloneActivityAsync(runner); + await ExecuteStandaloneNexusAsync(runner); + + return await runner.Client.StartWorkflowAsync( + (TransferWorkflow wf) => wf.RunAsync( + new NonGenericValue("non-generic", "client-extra"), + new Box(123, "client-extra"), + new DerivedFromConvertedBase( + "converted-base", "client-extra", "client-derived-extra"), + new DerivedFromConvertedBase( + "unconverted-derived", "plain-extra", "derived-extra"), + new PlainValue( + "plain", + "plain-extra", + new NonGenericValue("nested", "nested-extra")), + new ConvertedDerived( + "plain-base", "plain-extra", "ignored-derived-extra"), + new ConvertedDerived( + "converted-derived", "client-extra", "client-derived-extra")), + runner.NewWorkflowOptions()); + } + + public async Task CheckResultAsync(Runner runner, WorkflowHandle handle) + { + var result = await handle.GetResultAsync(); + Assert.True( + result == new NonGenericValue( + "workflow-result", TransferModels.TransferredMarker), + result.Value); + + var history = await handle.FetchHistoryAsync(); + var started = history.Events.Single( + evt => evt.WorkflowExecutionStartedEventAttributes != null). + WorkflowExecutionStartedEventAttributes; + var inputs = started.Input.Payloads_; + Assert.Equal(7, inputs.Count); + AssertProtobufPayload(inputs[0], "non-generic"); + AssertProtobufPayload(inputs[1], "box"); + AssertJsonPayload(inputs[2], TransferModels.TransferredMarker); + AssertJsonPayload(inputs[3]); + AssertJsonPayload(inputs[4]); + AssertJsonPayload(inputs[5], "plain-extra"); + AssertJsonPayload(inputs[6], TransferModels.TransferredMarker); + + var activityFailed = history.Events.Single( + evt => evt.ActivityTaskFailedEventAttributes != null). + ActivityTaskFailedEventAttributes; + var detailPayload = activityFailed.Failure.ApplicationFailureInfo.Details.Payloads_.Single(); + AssertProtobufPayload(detailPayload, "non-generic"); + + var completed = history.Events.Single( + evt => evt.WorkflowExecutionCompletedEventAttributes != null). + WorkflowExecutionCompletedEventAttributes; + AssertProtobufPayload(completed.Result.Payloads_.Single(), "non-generic"); + } + + private static async Task ExecuteStandaloneActivityAsync(Runner runner) + { + var result = await runner.Client.ExecuteActivityAsync( + () => TransferActivities.TransferAsync( + new NonGenericValue("activity-input", "client-extra")), + new StartActivityOptions( + $"{runner.PreparedFeature.Dir}-activity-{Guid.NewGuid()}", + runner.WorkerOptions.TaskQueue!) + { + StartToCloseTimeout = TimeSpan.FromSeconds(10), + RetryPolicy = new TemporalRetryPolicy { MaximumAttempts = 1 }, + }); + Assert.Equal( + new NonGenericValue("activity-result", TransferModels.TransferredMarker), + result); + } + + private static async Task ExecuteStandaloneNexusAsync(Runner runner) + { + if (runner.NexusEndpoint == null) + { + runner.Logger.LogInformation( + "Skipping Standalone Nexus check because no endpoint is available"); + return; + } + + var client = runner.Client.CreateNexusClient(runner.NexusEndpoint); + try + { + var result = await client.ExecuteNexusOperationAsync( + service => service.Transfer( + new NonGenericValue("nexus-input", "client-extra")), + new($"{runner.PreparedFeature.Dir}-nexus-{Guid.NewGuid()}") + { + ScheduleToCloseTimeout = TimeSpan.FromSeconds(10), + }); + Assert.Equal( + new NonGenericValue("nexus-result", TransferModels.TransferredMarker), + result); + } + catch (RpcException e) when (e.Code == RpcException.StatusCode.Unimplemented) + { + runner.Logger.LogInformation( + "Skipping Standalone Nexus check because the server does not support it"); + } + } + + private static void AssertProtobufPayload(Payload payload, string expectedRunId) + { + Assert.Equal("json/protobuf", payload.Metadata["encoding"].ToStringUtf8()); + Assert.Equal( + "temporal.api.common.v1.WorkflowExecution", + payload.Metadata["messageType"].ToStringUtf8()); + var value = Assert.IsType( + new JsonProtoConverter().ToValue(payload, typeof(ApiWorkflowExecution))); + Assert.Equal(expectedRunId, value.RunId); + } + + private static void AssertJsonPayload(Payload payload, string? expectedExtra = null) + { + Assert.Equal("json/plain", payload.Metadata["encoding"].ToStringUtf8()); + if (expectedExtra != null) + { + using var json = JsonDocument.Parse(payload.Data.ToStringUtf8()); + Assert.Equal(expectedExtra, json.RootElement.GetProperty("Extra").GetString()); + } + } + + [Workflow] + class TransferWorkflow + { + [WorkflowRun] + public async Task RunAsync( + NonGenericValue nonGeneric, + Box box, + ConvertedBase convertedBase, + DerivedFromConvertedBase unconvertedDerived, + PlainValue plain, + PlainBase plainBase, + ConvertedDerived convertedDerived) + { + var failures = new List(); + Check( + nonGeneric == new NonGenericValue( + "non-generic", TransferModels.TransferredMarker), + "non-generic"); + Check(box == new Box(123, TransferModels.TransferredMarker), "generic"); + Check(convertedBase.GetType() == typeof(ConvertedBase), "base-type"); + Check( + convertedBase == new ConvertedBase( + "converted-base", TransferModels.TransferredMarker), + "base-converter"); + Check( + unconvertedDerived.GetType() == typeof(DerivedFromConvertedBase), + "exact-declaration-type"); + Check( + unconvertedDerived == new DerivedFromConvertedBase( + "unconverted-derived", "plain-extra", "derived-extra"), + "exact-declaration-value"); + Check(plainBase.GetType() == typeof(PlainBase), "declared-plain-base-type"); + Check( + plainBase == new PlainBase("plain-base", "plain-extra"), + "declared-plain-base-value"); + Check( + convertedDerived == new ConvertedDerived( + "converted-derived", + TransferModels.TransferredMarker, + TransferModels.TransferredMarker), + "converted-derived"); + Check( + plain == new PlainValue( + "plain", + "plain-extra", + new NonGenericValue("nested", "nested-extra")), + "top-level-only"); + + try + { + await Workflow.ExecuteActivityAsync( + (TransferActivities act) => act.FailWithTransferDetailAsync(), + new ActivityOptions + { + StartToCloseTimeout = TimeSpan.FromSeconds(10), + RetryPolicy = new TemporalRetryPolicy { MaximumAttempts = 1 }, + }); + failures.Add("failure-detail-missing"); + } + catch (ActivityFailureException e) + { + if (e.InnerException is not ApplicationFailureException appFailure) + { + failures.Add("failure-detail-error"); + } + else + { + Check(appFailure.Details.Count == 1, "failure-detail-count"); + if (appFailure.Details.Count == 1) + { + var detail = appFailure.Details.ElementAt(0); + Check( + detail.WorkflowId == "failure-detail" && + detail.RunId == "non-generic", + "failure-detail-value"); + } + } + } + + var resultValue = failures.Count == 0 ? + "workflow-result" : $"failed:{string.Join(',', failures)}"; + return new NonGenericValue(resultValue, "must-not-be-serialized"); + + void Check(bool condition, string name) + { + if (!condition) + { + failures.Add(name); + } + } + } + } + + [Workflow] + class ThrowingWorkflow + { + [WorkflowRun] + public Task RunAsync(ThrowingValue value) => + throw new InvalidOperationException("A converter failure must prevent workflow start"); + } + + class TransferActivities + { + [Activity] + public static Task TransferAsync(NonGenericValue value) + { + Assert.Equal( + new NonGenericValue("activity-input", TransferModels.TransferredMarker), + value); + return Task.FromResult( + new NonGenericValue("activity-result", "must-not-be-serialized")); + } + + [Activity] + public Task FailWithTransferDetailAsync() => + throw new ApplicationFailureException( + "intentional transfer detail failure", + nonRetryable: true, + details: new object?[] + { + new NonGenericValue("failure-detail", "must-not-be-serialized"), + }); + } + + [NexusService] + interface ITransferService + { + [NexusOperation] + NonGenericValue Transfer(NonGenericValue input); + } + + [NexusServiceHandler(typeof(ITransferService))] + class TransferServiceHandler + { + [NexusOperationHandler] + public IOperationHandler Transfer() => + OperationHandler.Sync((context, value) => + { + Assert.Equal( + new NonGenericValue( + "nexus-input", TransferModels.TransferredMarker), + value); + return new NonGenericValue("nexus-result", "must-not-be-serialized"); + }); + } +} + +static class TransferModels +{ + public const string TransferredMarker = "created-from-transfer-type"; +} + +[TemporalTransferTypeConverter(typeof(NonGenericValueConverter))] +record NonGenericValue(string Value, string Extra); + +sealed class NonGenericValueConverter : ITemporalTransferTypeConverter +{ + public Type TransferType => typeof(ApiWorkflowExecution); + + public object ToTransferType(object? value) => new ApiWorkflowExecution + { + WorkflowId = ((NonGenericValue)value!).Value, + RunId = "non-generic", + }; + + public object FromTransferType(object? transferType) + { + var value = (ApiWorkflowExecution)transferType!; + Assert.Equal("non-generic", value.RunId); + return new NonGenericValue(value.WorkflowId, TransferModels.TransferredMarker); + } +} + +[TemporalTransferTypeConverter(typeof(BoxConverter<>))] +record Box(T Value, string Extra); + +sealed class BoxConverter : ITemporalTransferTypeConverter +{ + public Type TransferType => typeof(ApiWorkflowExecution); + + public object ToTransferType(object? value) => new ApiWorkflowExecution + { + WorkflowId = Convert.ToString(((Box)value!).Value)!, + RunId = "box", + }; + + public object FromTransferType(object? transferType) + { + var value = (ApiWorkflowExecution)transferType!; + Assert.Equal("box", value.RunId); + return new Box( + (T)Convert.ChangeType(value.WorkflowId, typeof(T)), + TransferModels.TransferredMarker); + } +} + +[TemporalTransferTypeConverter(typeof(ConvertedBaseConverter))] +record ConvertedBase(string Value, string Extra); + +record DerivedFromConvertedBase(string Value, string Extra, string DerivedExtra) : + ConvertedBase(Value, Extra); + +sealed class ConvertedBaseConverter : ITemporalTransferTypeConverter +{ + public Type TransferType => typeof(PlainBase); + + public object ToTransferType(object? value) => + new PlainBase( + ((ConvertedBase)value!).Value, + TransferModels.TransferredMarker); + + public object FromTransferType(object? transferType) + { + var value = (PlainBase)transferType!; + return new ConvertedBase(value.Value, value.Extra); + } +} + +record PlainBase(string Value, string Extra); + +[TemporalTransferTypeConverter(typeof(ConvertedDerivedConverter))] +record ConvertedDerived(string Value, string Extra, string DerivedExtra) : + PlainBase(Value, Extra); + +sealed class ConvertedDerivedConverter : ITemporalTransferTypeConverter +{ + public Type TransferType => typeof(PlainBase); + + public object ToTransferType(object? value) + { + var model = (ConvertedDerived)value!; + return new PlainBase(model.Value, TransferModels.TransferredMarker); + } + + public object FromTransferType(object? transferType) + { + var value = (PlainBase)transferType!; + return new ConvertedDerived( + value.Value, + TransferModels.TransferredMarker, + TransferModels.TransferredMarker); + } +} + +record PlainValue(string Value, string Extra, NonGenericValue Nested); + +[TemporalTransferTypeConverter(typeof(ThrowingValueConverter))] +record ThrowingValue(string Value); + +sealed class ThrowingValueConverter : ITemporalTransferTypeConverter +{ + public Type TransferType => typeof(ApiWorkflowExecution); + + public object? ToTransferType(object? value) => + throw new TransferConversionException(((ThrowingValue)value!).Value); + + public object? FromTransferType(object? transferType) => + throw new TransferConversionException( + ((ApiWorkflowExecution)transferType!).WorkflowId); +} + +sealed class TransferConversionException : Exception +{ + public TransferConversionException(string message) + : base(message) + { + } +} diff --git a/features/data_converter/transfer_types/feature.py b/features/data_converter/transfer_types/feature.py new file mode 100644 index 00000000..00781e0c --- /dev/null +++ b/features/data_converter/transfer_types/feature.py @@ -0,0 +1,444 @@ +from __future__ import annotations + +import logging +import uuid +from dataclasses import dataclass +from datetime import timedelta +from typing import Generic, TypeVar, cast, get_args + +import nexusrpc +from temporalio import activity, workflow +from temporalio.api.common.v1 import Payload, WorkflowExecution +from temporalio.api.enums.v1 import EventType +from temporalio.client import RPCError, RPCStatusCode, WorkflowHandle +from temporalio.common import RetryPolicy +from temporalio.converter import ( + JSONProtoPayloadConverter, + TransferTypeConverter, + transfer_type_convertible, +) +from temporalio.exceptions import ActivityError, ApplicationError + +from harness.python.feature import Runner, register_feature + +logger = logging.getLogger(__name__) +TRANSFERRED_MARKER = "created-from-transfer-type" + + +class NonGenericValueConverter( + TransferTypeConverter["NonGenericValue", WorkflowExecution] +): + transfer_type = WorkflowExecution + + def to_transfer_type(self, value: NonGenericValue) -> WorkflowExecution: + return WorkflowExecution(workflow_id=value.value, run_id="non-generic") + + def from_transfer_type( + self, value: WorkflowExecution, type_hint: type[NonGenericValue] + ) -> NonGenericValue: + assert type_hint is NonGenericValue + assert value.run_id == "non-generic" + return NonGenericValue(value=value.workflow_id, extra=TRANSFERRED_MARKER) + + +@transfer_type_convertible(NonGenericValueConverter) +@dataclass +class NonGenericValue: + value: str + extra: str + + +T = TypeVar("T") + + +@dataclass +class Box(Generic[T]): + value: T + extra: str + + +class BoxConverter(TransferTypeConverter[Box[T], WorkflowExecution]): + transfer_type = WorkflowExecution + + def to_transfer_type(self, value: Box[T]) -> WorkflowExecution: + return WorkflowExecution(workflow_id=str(value.value), run_id="box") + + def from_transfer_type( + self, value: WorkflowExecution, type_hint: type[Box[T]] + ) -> Box[T]: + assert value.run_id == "box" + type_args = get_args(type_hint) + item_type = type_args[0] if type_args else str + item: str | int = value.workflow_id + if item_type is int: + item = int(item) + return Box(value=cast(T, item), extra=TRANSFERRED_MARKER) + + +transfer_type_convertible(BoxConverter)(Box) + + +class ConvertedBaseConverter(TransferTypeConverter["ConvertedBase", WorkflowExecution]): + transfer_type = WorkflowExecution + + def to_transfer_type(self, value: ConvertedBase) -> WorkflowExecution: + return WorkflowExecution(workflow_id=value.value, run_id="converted-base") + + def from_transfer_type( + self, value: WorkflowExecution, type_hint: type[ConvertedBase] + ) -> ConvertedBase: + assert value.run_id == "converted-base" + return ConvertedBase(value=value.workflow_id, extra=TRANSFERRED_MARKER) + + +@transfer_type_convertible(ConvertedBaseConverter) +@dataclass +class ConvertedBase: + value: str + extra: str + + +@dataclass +class DerivedFromConvertedBase(ConvertedBase): + derived_extra: str + + +@dataclass +class PlainBase: + value: str + extra: str + + +class ConvertedDerivedConverter( + TransferTypeConverter["ConvertedDerived", WorkflowExecution] +): + transfer_type = WorkflowExecution + + def to_transfer_type(self, value: ConvertedDerived) -> WorkflowExecution: + return WorkflowExecution(workflow_id=value.value, run_id="converted-derived") + + def from_transfer_type( + self, value: WorkflowExecution, type_hint: type[ConvertedDerived] + ) -> ConvertedDerived: + assert type_hint is ConvertedDerived + assert value.run_id == "converted-derived" + return ConvertedDerived( + value=value.workflow_id, + extra=TRANSFERRED_MARKER, + derived_extra=TRANSFERRED_MARKER, + ) + + +@transfer_type_convertible(ConvertedDerivedConverter) +@dataclass +class ConvertedDerived(PlainBase): + derived_extra: str + + +@dataclass +class PlainValue: + value: str + extra: str + nested: NonGenericValue + + +class TransferConversionError(RuntimeError): + pass + + +class ThrowingValueConverter(TransferTypeConverter["ThrowingValue", WorkflowExecution]): + transfer_type = WorkflowExecution + + def to_transfer_type(self, value: ThrowingValue) -> WorkflowExecution: + raise TransferConversionError(value.value) + + def from_transfer_type( + self, value: WorkflowExecution, type_hint: type[ThrowingValue] + ) -> ThrowingValue: + raise TransferConversionError(value.workflow_id) + + +@transfer_type_convertible(ThrowingValueConverter) +@dataclass +class ThrowingValue: + value: str + + +@activity.defn +async def fail_with_transfer_detail() -> None: + raise ApplicationError( + "intentional transfer detail failure", + NonGenericValue("failure-detail", "must-not-be-serialized"), + non_retryable=True, + ) + + +@activity.defn +async def standalone_transfer_activity(value: NonGenericValue) -> NonGenericValue: + assert value == NonGenericValue("activity-input", TRANSFERRED_MARKER) + return NonGenericValue("activity-result", "must-not-be-serialized") + + +@nexusrpc.service +class TransferService: + transfer: nexusrpc.Operation[NonGenericValue, NonGenericValue] + + +@nexusrpc.handler.service_handler(service=TransferService) +class TransferServiceHandler: + @nexusrpc.handler.sync_operation + async def transfer( + self, ctx: nexusrpc.handler.StartOperationContext, value: NonGenericValue + ) -> NonGenericValue: + assert value == NonGenericValue("nexus-input", TRANSFERRED_MARKER) + return NonGenericValue("nexus-result", "must-not-be-serialized") + + +@workflow.defn +class Workflow: + @workflow.run + async def run( + self, + non_generic: NonGenericValue, + box: Box[int], + converted_base: ConvertedBase, + unconverted_derived: DerivedFromConvertedBase, + plain: PlainValue, + plain_base: PlainBase, + converted_derived: ConvertedDerived, + ) -> NonGenericValue: + failures: list[str] = [] + + def check(condition: bool, name: str) -> None: + if not condition: + failures.append(name) + + check( + non_generic == NonGenericValue("non-generic", TRANSFERRED_MARKER), + "non-generic", + ) + check(box == Box(123, TRANSFERRED_MARKER), "generic") + check(type(converted_base) is ConvertedBase, "base-type") + check( + converted_base == ConvertedBase("converted-base", TRANSFERRED_MARKER), + "base-converter", + ) + check( + type(unconverted_derived) is DerivedFromConvertedBase, + "exact-declaration-type", + ) + check( + unconverted_derived + == DerivedFromConvertedBase( + "unconverted-derived", "plain-extra", "derived-extra" + ), + "exact-declaration-value", + ) + check(type(plain_base) is PlainBase, "declared-plain-base-type") + check( + plain_base == PlainBase("plain-base", "plain-extra"), + "declared-plain-base-value", + ) + check( + converted_derived + == ConvertedDerived( + "converted-derived", TRANSFERRED_MARKER, TRANSFERRED_MARKER + ), + "converted-derived", + ) + check( + plain + == PlainValue( + "plain", + "plain-extra", + NonGenericValue("nested", "nested-extra"), + ), + "top-level-only", + ) + + try: + await workflow.execute_activity( + fail_with_transfer_detail, + start_to_close_timeout=timedelta(seconds=10), + ) + failures.append("failure-detail-missing") + except ActivityError as err: + check(isinstance(err.cause, ApplicationError), "failure-detail-error") + details = ( + err.cause.details if isinstance(err.cause, ApplicationError) else () + ) + check(len(details) == 1, "failure-detail-count") + detail = details[0] if details else None + check(isinstance(detail, WorkflowExecution), "failure-detail-type") + check( + detail + == WorkflowExecution( + workflow_id="failure-detail", run_id="non-generic" + ), + "failure-detail-value", + ) + + result_value = "workflow-result" + if failures: + result_value = "failed:" + ",".join(failures) + return NonGenericValue(result_value, "must-not-be-serialized") + + +@workflow.defn +class ThrowingWorkflow: + @workflow.run + async def run(self, value: ThrowingValue) -> None: + raise AssertionError("a converter failure must prevent workflow start") + + +async def exercise_standalone_activity(runner: Runner) -> None: + result = await runner.client.execute_activity( + standalone_transfer_activity, + NonGenericValue("activity-input", "client-extra"), + id=f"{runner.feature.rel_dir}-activity-{uuid.uuid4()}", + task_queue=runner.task_queue, + start_to_close_timeout=timedelta(seconds=10), + retry_policy=RetryPolicy(maximum_attempts=1), + ) + assert result == NonGenericValue("activity-result", TRANSFERRED_MARKER) + + +async def exercise_standalone_nexus(runner: Runner) -> None: + if runner.nexus_endpoint is None: + logger.info("Skipping Standalone Nexus check because no endpoint is available") + return + client = runner.client.create_nexus_client( + service=TransferService, + endpoint=runner.nexus_endpoint, + ) + try: + result = await client.execute_operation( + TransferService.transfer, + NonGenericValue("nexus-input", "client-extra"), + id=f"{runner.feature.rel_dir}-nexus-{uuid.uuid4()}", + schedule_to_close_timeout=timedelta(seconds=10), + ) + except RPCError as err: + if err.status == RPCStatusCode.UNIMPLEMENTED: + logger.info( + "Skipping Standalone Nexus check because the server does not support it" + ) + return + raise + assert result == NonGenericValue("nexus-result", TRANSFERRED_MARKER) + + +async def start(runner: Runner) -> WorkflowHandle: + failed_workflow_id = f"{runner.feature.rel_dir}-failing-{uuid.uuid4()}" + try: + await runner.client.start_workflow( + ThrowingWorkflow.run, + ThrowingValue("expected transfer conversion failure"), + id=failed_workflow_id, + task_queue=runner.task_queue, + execution_timeout=timedelta(minutes=1), + ) + raise AssertionError("workflow start should fail during transfer conversion") + except TransferConversionError as err: + assert str(err) == "expected transfer conversion failure" + + try: + await runner.client.get_workflow_handle(failed_workflow_id).describe() + raise AssertionError("converter failure should prevent the workflow RPC") + except RPCError as err: + assert err.status == RPCStatusCode.NOT_FOUND + + await exercise_standalone_activity(runner) + await exercise_standalone_nexus(runner) + + return await runner.client.start_workflow( + Workflow.run, + args=[ + NonGenericValue("non-generic", "client-extra"), + Box(123, "client-extra"), + DerivedFromConvertedBase( + "converted-base", "client-extra", "client-derived-extra" + ), + DerivedFromConvertedBase( + "unconverted-derived", "plain-extra", "derived-extra" + ), + PlainValue( + "plain", + "plain-extra", + NonGenericValue("nested", "nested-extra"), + ), + ConvertedDerived("plain-base", "plain-extra", "ignored-derived-extra"), + ConvertedDerived( + "converted-derived", "client-extra", "client-derived-extra" + ), + ], + id=f"{runner.feature.rel_dir}-{uuid.uuid4()}", + task_queue=runner.task_queue, + execution_timeout=timedelta(minutes=1), + ) + + +async def check_result(runner: Runner, handle: WorkflowHandle) -> None: + result = await handle.result() + assert result == NonGenericValue("workflow-result", TRANSFERRED_MARKER), ( + result.value + ) + + events = [event async for event in handle.fetch_history_events()] + started = next( + event + for event in events + if event.event_type == EventType.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED + ) + inputs = started.workflow_execution_started_event_attributes.input.payloads + assert len(inputs) == 7 + assert_protobuf_payload(inputs[0], "non-generic") + assert_protobuf_payload(inputs[1], "box") + assert_protobuf_payload(inputs[2], "converted-base") + assert_json_payload(inputs[3]) + assert_json_payload(inputs[4]) + assert_json_payload(inputs[5]) + assert_protobuf_payload(inputs[6], "converted-derived") + + activity_failed = next( + event + for event in events + if event.event_type == EventType.EVENT_TYPE_ACTIVITY_TASK_FAILED + ) + detail_payload = activity_failed.activity_task_failed_event_attributes.failure.application_failure_info.details.payloads[ + 0 + ] + assert_protobuf_payload(detail_payload, "non-generic") + + completed = next( + event + for event in events + if event.event_type == EventType.EVENT_TYPE_WORKFLOW_EXECUTION_COMPLETED + ) + assert_protobuf_payload( + completed.workflow_execution_completed_event_attributes.result.payloads[0], + "non-generic", + ) + + +def assert_protobuf_payload(payload: Payload, expected_run_id: str) -> None: + assert payload.metadata["encoding"] == b"json/protobuf" + assert ( + payload.metadata["messageType"] == b"temporal.api.common.v1.WorkflowExecution" + ) + value = JSONProtoPayloadConverter().from_payload(payload, WorkflowExecution) + assert isinstance(value, WorkflowExecution) + assert value.run_id == expected_run_id + + +def assert_json_payload(payload: Payload) -> None: + assert payload.metadata["encoding"] == b"json/plain" + + +register_feature( + workflows=[Workflow, ThrowingWorkflow], + activities=[fail_with_transfer_detail, standalone_transfer_activity], + nexus_service_handlers=[TransferServiceHandler()], + start=start, + check_result=check_result, +) diff --git a/features/data_converter/transfer_types/spec.md b/features/data_converter/transfer_types/spec.md index 11feaf2a..d8f33ae0 100644 --- a/features/data_converter/transfer_types/spec.md +++ b/features/data_converter/transfer_types/spec.md @@ -196,5 +196,4 @@ TransferTypeConverters should specify a non-null transfer type. #### Failure conversion -Failure conversion should use the transfer converter before -applying the configured payload converter to failure details. +Failure conversion should NOT use the transfer converter. diff --git a/harness/dotnet/Temporalio.Features.Harness/App.cs b/harness/dotnet/Temporalio.Features.Harness/App.cs index 70a912d2..c897197a 100644 --- a/harness/dotnet/Temporalio.Features.Harness/App.cs +++ b/harness/dotnet/Temporalio.Features.Harness/App.cs @@ -39,19 +39,20 @@ public static class App name: "--tls-server-name", description: "TLS server name to use for verification"); - private static readonly Argument> featuresArgument = new( + private static readonly Argument< + List<(string Dir, string TaskQueue, string? NexusEndpoint)>> featuresArgument = new( name: "features", parse: result => result.Tokens.Select(token => { - var pieces = token.Value.Split(':', 2); - if (pieces.Length != 2) + var pieces = token.Value.Split(':', 3); + if (pieces.Length < 2) { throw new ArgumentException("Feature must be dir + ':' + task queue"); } - return (pieces[0], pieces[1]); + return (pieces[0], pieces[1], pieces.Length == 3 ? pieces[2] : null); }).ToList(), - description: "Features as dir + ':' + task queue") + description: "Features as dir + ':' + task queue + optional ':' + Nexus endpoint") { Arity = ArgumentArity.OneOrMore }; /// @@ -119,7 +120,8 @@ private static async Task RunCommandAsync(InvocationContext ctx) // Go over each feature, calling the runner for it var failures = new List(); - foreach (var (dir, taskQueue) in ctx.ParseResult.GetValueForArgument(featuresArgument)) + foreach (var (dir, taskQueue, nexusEndpoint) in + ctx.ParseResult.GetValueForArgument(featuresArgument)) { var feature = PreparedFeature.AllFeatures.SingleOrDefault(feature => feature.Dir == dir) ?? @@ -129,6 +131,7 @@ private static async Task RunCommandAsync(InvocationContext ctx) await new Runner( clientOptions, taskQueue, + nexusEndpoint, feature, loggerFactory, ctx.ParseResult.GetValueForOption(httpProxyUrlOption) @@ -159,4 +162,4 @@ private static async Task RunCommandAsync(InvocationContext ctx) logger.LogInformation("All features passed"); } } -} \ No newline at end of file +} diff --git a/harness/dotnet/Temporalio.Features.Harness/Runner.cs b/harness/dotnet/Temporalio.Features.Harness/Runner.cs index 1498ceac..03043e72 100644 --- a/harness/dotnet/Temporalio.Features.Harness/Runner.cs +++ b/harness/dotnet/Temporalio.Features.Harness/Runner.cs @@ -43,10 +43,10 @@ public async Task StopAndWait() } } - internal Runner( TemporalClientConnectOptions clientConnectOptions, string taskQueue, + string? nexusEndpoint, PreparedFeature feature, ILoggerFactory loggerFactory, string? httpProxyUrl) @@ -55,6 +55,7 @@ internal Runner( Logger = loggerFactory.CreateLogger(PreparedFeature.FeatureType); Feature = (IFeature)Activator.CreateInstance(PreparedFeature.FeatureType, true)!; HttpProxyUrl = httpProxyUrl; + NexusEndpoint = nexusEndpoint; ClientOptions = (TemporalClientConnectOptions)clientConnectOptions.Clone(); Feature.ConfigureClient(this, ClientOptions); @@ -75,6 +76,8 @@ internal Runner( public string? HttpProxyUrl { get; private init; } + public string? NexusEndpoint { get; private init; } + /// /// Run the feature with the given cancellation token. /// @@ -282,7 +285,7 @@ public async Task WaitForEventAsync(WorkflowHandle handle, FuncTask for completion. public async Task WaitForActivityTaskScheduledAsync(WorkflowHandle handle, TimeSpan? timeout = null) { - await WaitForEventAsync(handle, - e => e.EventType == Temporalio.Api.Enums.V1.EventType.ActivityTaskScheduled, + await WaitForEventAsync(handle, + e => e.EventType == Temporalio.Api.Enums.V1.EventType.ActivityTaskScheduled, timeout); } -} \ No newline at end of file +} diff --git a/harness/go/cmd/run.go b/harness/go/cmd/run.go index b56e8d42..baef4e77 100644 --- a/harness/go/cmd/run.go +++ b/harness/go/cmd/run.go @@ -86,7 +86,7 @@ type RunFeature struct { Dir string TaskQueue string // NexusEndpoint is the pre-created Nexus endpoint name targeting this feature's namespace - // and task queue. Set by the top-level runner for features under features/nexus. + // and task queue. The top-level runner sets it when the feature requests an endpoint. NexusEndpoint string Config RunFeatureConfig VariantName string @@ -102,6 +102,7 @@ func (r RunFeature) SummaryName() string { // RunFeatureConfig is config from config.json. type RunFeatureConfig struct { NoWorkflow bool `json:"noWorkflow"` + NeedsNexusEndpoint bool `json:"needsNexusEndpoint"` Go RunFeatureConfigGo `json:"go"` ExpectUnauthedProxyCount int `json:"expectUnauthedProxyCount"` ExpectAuthedProxyCount int `json:"expectAuthedProxyCount"` diff --git a/harness/python/feature.py b/harness/python/feature.py index 990bf60a..25826250 100644 --- a/harness/python/feature.py +++ b/harness/python/feature.py @@ -7,7 +7,18 @@ from dataclasses import dataclass from datetime import timedelta from pathlib import Path -from typing import Any, Awaitable, Callable, Dict, List, Mapping, Optional, Type, Union +from typing import ( + Any, + Awaitable, + Callable, + Dict, + List, + Mapping, + Optional, + Sequence, + Type, + Union, +) from temporalio import workflow from temporalio.api.common.v1 import Payload @@ -34,6 +45,7 @@ def register_feature( *, workflows: List[Type], activities: List[Callable] = [], + nexus_service_handlers: Sequence[Any] = (), expect_activity_error: Optional[str] = None, expect_run_result: Optional[Any] = None, file: Optional[str] = None, @@ -60,6 +72,7 @@ def register_feature( rel_dir=rel_dir, workflows=workflows, activities=activities, + nexus_service_handlers=nexus_service_handlers, expect_activity_error=expect_activity_error, expect_run_result=expect_run_result, start=start, @@ -95,6 +108,7 @@ class Feature: rel_dir: str # Always relative to feature dir and uses forward slashes workflows: List[Type] activities: List[Callable] + nexus_service_handlers: Sequence[Any] expect_activity_error: Optional[str] expect_run_result: Optional[Any] start: Optional[Callable[[Runner], Awaitable[WorkflowHandle]]] @@ -112,6 +126,7 @@ def __init__( address: str, namespace: str, task_queue: str, + nexus_endpoint: Optional[str], feature: Feature, tls_config: Optional[TLSConfig], http_proxy_url: Optional[str], @@ -119,6 +134,7 @@ def __init__( self.address = address self.namespace = namespace self.task_queue = task_queue + self.nexus_endpoint = nexus_endpoint self.feature = feature self.worker: Optional[Worker] = None self._worker_task: Optional[asyncio.Task] = None @@ -205,16 +221,20 @@ async def check_result(self, handle: WorkflowHandle) -> None: raise err def start_worker(self): - """Creates and starts worker with the task queue, workflows, and - activities set.""" + """Create and start the worker for this feature.""" if self.worker is not None: raise RuntimeError("Worker already started") + worker_config: Dict[str, Any] = dict(self.feature.worker_config or {}) + if self.feature.nexus_service_handlers: + worker_config["nexus_service_handlers"] = ( + self.feature.nexus_service_handlers + ) self.worker = Worker( self.client, task_queue=self.task_queue, workflows=self.feature.workflows, activities=self.feature.activities, - **self.feature.worker_config, + **worker_config, ) self._worker_task = asyncio.create_task(self.worker.run()) diff --git a/harness/python/main.py b/harness/python/main.py index ed4fe4bd..d27122de 100644 --- a/harness/python/main.py +++ b/harness/python/main.py @@ -30,7 +30,9 @@ async def run(): "--tls-server-name", help="TLS server name to use for verification (optional)" ) parser.add_argument( - "features", help="Features as dir + ':' + task queue", nargs="+" + "features", + help="Features as dir + ':' + task queue + optional ':' + Nexus endpoint", + nargs="+", ) args = parser.parse_args() @@ -67,8 +69,14 @@ async def run(): # Run each feature failed_features = [] for rel_dir_and_task_queue in cast(List[str], args.features): - # Split rel dir and task queue - rel_dir, _, task_queue = rel_dir_and_task_queue.partition(":") + # Features that request Nexus include a pre-created endpoint after the task queue. + pieces = rel_dir_and_task_queue.split(":", 2) + if len(pieces) < 2: + raise ValueError( + f"Feature argument is missing its task queue: {rel_dir_and_task_queue}" + ) + rel_dir, task_queue = pieces[:2] + nexus_endpoint = pieces[2] if len(pieces) == 3 else None if rel_dir not in rel_dirs: raise ValueError(f"Cannot find feature file in {rel_dir}") # Import @@ -82,6 +90,7 @@ async def run(): address=args.server, namespace=args.namespace, task_queue=task_queue, + nexus_endpoint=nexus_endpoint, feature=features[rel_dir], tls_config=tls_config, http_proxy_url=args.http_proxy_url if args.http_proxy_url else None,