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 new file mode 100644 index 00000000..d8f33ae0 --- /dev/null +++ b/features/data_converter/transfer_types/spec.md @@ -0,0 +1,199 @@ +# Transfer Type Converter Behavioral Specification + +Last updated: 2026-09-02 + +## Motivation + +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 + +```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") +``` + +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. +- **Transfer type converter:** The conversion logic to map between the model type and the transfer type. + +## Behaviors + +### Conversion Flow + +#### Encoding + +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. + +```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 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. + +#### 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 + + 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. + +### 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 + +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` + +When encoding type `A`, the SDK does not inspect the output type `B` for a transfer type converter and thus does not apply `converterB`. + +Similarly when decoding, the SDK will convert using only the transfer type encoder registered on the requested model type. + +#### 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: + +```java + // Given this workflow + @WorkflowMethod + void run(Animal animal); + + // And this invocation + Animal value = new Dog(); + workflow.run(value); +``` + +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. + +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. + +A transfer converter declaration applies only to the exact model type on which it is declared. Declarations are not inherited from a base type. + +A subclass or other derived type can declare its own converter independently of its base type. + +#### Transfer Type Nullability + +TransferTypeConverters should specify a non-null transfer type. + +#### Failure conversion + +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,