diff --git a/CHANGELOG.md b/CHANGELOG.md index 871dfa56..a2195b22 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -57,6 +57,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- Activities, sub-orchestrations, and ContinueAsNew do not add owning-orchestration identity tags, aligning automatic tags with .NET. Activity work items supply the owning instance ID and the activity's own identity; orchestrator contexts use native history fields. User-tag inheritance, immutable context fields, namespace markers, and distributed tracing are preserved. - DTS token audiences now default per options/client/worker instance to `https://durabletask.azure.us` when `REGION_NAME` starts with `usgov` or `usdod` (case-insensitively), otherwise `https://durabletask.io`. Explicit `Options.ResourceID` or connection-string `ResourceId` overrides the default without changing the endpoint or credential authority. Resource audiences normalize surrounding whitespace, trailing slashes, and one existing `/.default` suffix before token requests; values that normalize to empty are rejected. Audiences remain stable across token refreshes and reconnects. Government-region applications requiring the old audience must explicitly set `https://durabletask.io`. - Restore DTS-provided activity trace parents on `ActivityContext.Context()` without emitting duplicate SDK durable spans, preserving trace continuity for application instrumentation. - Missing-instance orchestration waits now return `api.ErrInstanceNotFound` immediately instead of retrying `NotFound` until the caller deadline. diff --git a/README.md b/README.md index 9a436781..219e77bc 100644 --- a/README.md +++ b/README.md @@ -517,6 +517,19 @@ The SDK does not save the identity of the converter. A new converter must contin Use `api.WithTags`, `task.WithActivityTags`, and `task.WithSubOrchestrationTags` to attach user tags. An activity and a sub-orchestration inherit the tags of the parent orchestration. A tag on the action has priority over an inherited tag. The completion actions carry the current tags, so ContinueAsNew keeps them. +Activities, sub-orchestrations, and ContinueAsNew do not add owning-orchestration +identity tags. Without caller tags or context fields, their tag maps +are absent on the wire, matching .NET's default automatic-tag behavior. Go keeps +its user-tag inheritance and immutable context-field propagation; when either is +present, the encoding marker separates user tags from context fields. +Distributed trace context remains a separate protocol field. + +Activity contexts expose the activity's own name, version, and task ID in +`api.ActivityContextInfo`, and the owning instance ID in +`api.OrchestrationContextInfo`. Orchestrator contexts obtain their full identity +from native history fields, including after sub-orchestration starts and +ContinueAsNew version changes. + The client sends the sampled caller trace context when it schedules an orchestration or signals an entity. The worker adds separate action trace contexts for the service-owned activity and sub-orchestration spans. The worker does not emit duplicate local Durable Task spans. V2 entity requests do not carry per-operation trace context, so entity-emitted actions cannot inherit it. Use `task.OrchestrationOptions.MaxEventsPerTurn` to limit the new events in one turn. If the worker uses only part of a batch, it sets `numEventsProcessed`. DTS then keeps the remaining events for the next replay. This count obeys the DTS work-item rules. The orchestration control markers do not count against the limit. diff --git a/api/context.go b/api/context.go index ca9882f0..05086eff 100644 --- a/api/context.go +++ b/api/context.go @@ -5,13 +5,15 @@ import ( "maps" ) -// ReservedContextFieldPrefix is reserved for Durable Task runtime identity tags. +// ReservedContextFieldPrefix is reserved for Durable Task context metadata. const ReservedContextFieldPrefix = "__durabletask.context." // ContextFields are immutable caller-supplied values propagated into task contexts. type ContextFields map[string]string // OrchestrationContextInfo identifies the orchestration associated with a task context. +// Orchestrator contexts contain the full persisted identity. Activity work items +// supply only InstanceID; the activity's own identity is in ActivityContextInfo. type OrchestrationContextInfo struct { InstanceID InstanceID Name string diff --git a/internal/contextprop/tags.go b/internal/contextprop/tags.go index d8ebb595..ea36b582 100644 --- a/internal/contextprop/tags.go +++ b/internal/contextprop/tags.go @@ -7,43 +7,9 @@ import ( "github.com/microsoft/durabletask-go/internal/tagcodec" ) -const ( - instanceIDTag = api.ReservedContextFieldPrefix + "instance_id" - nameTag = api.ReservedContextFieldPrefix + "orchestration_name" - versionTag = api.ReservedContextFieldPrefix + "orchestration_version" - parentInstanceIDTag = api.ReservedContextFieldPrefix + "parent_instance_id" -) - -// Encode returns a new tag map containing immutable fields and orchestration identity. -func Encode( - info api.OrchestrationContextInfo, - fields api.ContextFields, - userTags ...map[string]string, -) map[string]string { - tags := tagcodec.EncodeContextFields(fields) - if len(userTags) > 0 { - tags = tagcodec.Merge(tags, tagcodec.EncodeUserTags(userTags[0])) - } - if tags == nil { - tags = make(map[string]string, 5) - } - tags[tagcodec.ContextEncodingTag] = "1" - tags[instanceIDTag] = string(info.InstanceID) - tags[nameTag] = info.Name - tags[versionTag] = info.Version - tags[parentInstanceIDTag] = string(info.ParentInstanceID) - return tags -} - -// Decode separates orchestration identity from caller-supplied immutable fields. -func Decode(tags map[string]string) (api.OrchestrationContextInfo, api.ContextFields) { - info := api.OrchestrationContextInfo{ - InstanceID: api.InstanceID(tags[instanceIDTag]), - Name: tags[nameTag], - Version: tags[versionTag], - ParentInstanceID: api.InstanceID(tags[parentInstanceIDTag]), - } - return info, api.ContextFields(tagcodec.DecodeContextFields(tags)) +// Encode returns a new tag map containing caller fields and user tags. +func Encode(fields api.ContextFields, userTags map[string]string) map[string]string { + return tagcodec.Merge(tagcodec.EncodeContextFields(fields), tagcodec.EncodeUserTags(userTags)) } // Clone returns a defensive copy of tags, or nil when there is nothing to copy. diff --git a/internal/contextprop/tags_test.go b/internal/contextprop/tags_test.go index 6ad00751..a9609967 100644 --- a/internal/contextprop/tags_test.go +++ b/internal/contextprop/tags_test.go @@ -5,57 +5,66 @@ import ( "github.com/microsoft/durabletask-go/api" "github.com/microsoft/durabletask-go/internal/tagcodec" + "github.com/stretchr/testify/require" ) -func TestEncodeDecode(t *testing.T) { - tags := Encode(api.OrchestrationContextInfo{ - InstanceID: "instance", - Name: "orchestration", - Version: "v1", - ParentInstanceID: "parent", - }, api.ContextFields{"tenant": "alpha"}) - - info, fields := Decode(tags) - if info.InstanceID != "instance" || - info.Name != "orchestration" || - info.Version != "v1" || - info.ParentInstanceID != "parent" { - t.Fatalf("unexpected info: %+v", info) - } - - if fields["tenant"] != "alpha" { - t.Fatalf("tenant = %q, want alpha", fields["tenant"]) - } -} - -func TestEncodeSeparatesContextFieldsAndUserTags(t *testing.T) { - tags := Encode( - api.OrchestrationContextInfo{}, - api.ContextFields{"tenant": "context"}, - map[string]string{"team": "tag"}, - ) - _, fields := Decode(tags) - if fields["tenant"] != "context" { - t.Fatalf("tenant = %q, want context", fields["tenant"]) - } - userTags := tagcodec.DecodeUserTags(tags) - if userTags["team"] != "tag" { - t.Fatalf("team = %q, want tag", userTags["team"]) - } - if _, ok := fields["team"]; ok { - t.Fatalf("user tag leaked into context fields: %v", fields) +func TestEncodeCallerData(t *testing.T) { + for _, test := range []struct { + name string + fields api.ContextFields + tags map[string]string + want map[string]string + }{ + {name: "absent"}, + {name: "empty", fields: api.ContextFields{}, tags: map[string]string{}}, + { + name: "user tags only", + tags: map[string]string{"tenant": ""}, + want: map[string]string{tagcodec.ContextEncodingTag: "1", "tenant": ""}, + }, + { + name: "fields only", + fields: api.ContextFields{"tenant": ""}, + want: map[string]string{tagcodec.ContextEncodingTag: "1", tagcodec.ContextFieldPrefix + "tenant": ""}, + }, + { + name: "same key in separate namespaces", + fields: api.ContextFields{"tenant": "context"}, + tags: map[string]string{"tenant": "user"}, + want: map[string]string{ + tagcodec.ContextEncodingTag: "1", + tagcodec.ContextFieldPrefix + "tenant": "context", + "tenant": "user", + }, + }, + } { + t.Run(test.name, func(t *testing.T) { + encoded := Encode(test.fields, test.tags) + require.Equal(t, test.want, encoded) + fields := api.ContextFields(tagcodec.DecodeContextFields(encoded)) + if len(test.fields) == 0 { + require.Nil(t, fields, "user tags must not become context fields") + } else { + require.Equal(t, test.fields, fields) + fields["tenant"] = "decoded mutation" + require.Equal(t, test.want, encoded) + } + if test.fields != nil { + test.fields["tenant"] = "caller mutation" + } + if test.tags != nil { + test.tags["tenant"] = "caller mutation" + } + require.Equal(t, test.want, encoded) + }) } } -func TestEncodeOverwritesReservedCallerFields(t *testing.T) { - tags := Encode(api.OrchestrationContextInfo{}, api.ContextFields{ - api.ReservedContextFieldPrefix + "orchestration_version": "spoofed", - }) - info, fields := Decode(tags) - if info.Version != "" { - t.Fatalf("version = %q, want empty", info.Version) - } - if fields != nil { - t.Fatalf("reserved field leaked into caller fields: %v", fields) - } +func TestCloneDoesNotAliasTags(t *testing.T) { + require.Nil(t, Clone[map[string]string](nil)) + require.Nil(t, Clone(map[string]string{})) + original := map[string]string{"empty": ""} + cloned := Clone(original) + cloned["empty"] = "changed" + require.Equal(t, map[string]string{"empty": ""}, original) } diff --git a/task/context_test.go b/task/context_test.go index 84381cca..85c5678b 100644 --- a/task/context_test.go +++ b/task/context_test.go @@ -67,7 +67,7 @@ func TestOrchestrationContextPropagatesOnlyPersistedIdentityAndFields(t *testing nil, wrapperspb.String("v2"), ) - started.GetExecutionStarted().Tags = contextprop.Encode(api.OrchestrationContextInfo{}, fields) + started.GetExecutionStarted().Tags = contextprop.Encode(fields, nil) fields["tenant"] = "mutated" events := []*protos.HistoryEvent{helpers.NewOrchestratorStartedEvent(), started} @@ -164,7 +164,7 @@ func TestActivityContextPropagatesIdentityFieldsAndLogger(t *testing.T) { } } -func TestActivityContextDecodesDurableContextTags(t *testing.T) { +func TestActivityContextDecodesDurableContextFields(t *testing.T) { registry := NewTaskRegistry() if err := registry.AddActivityN("inspect-tags", func(ctx ActivityContext) (any, error) { orchestration, _ := api.OrchestrationContextInfoFromContext(ctx.Context()) @@ -180,12 +180,7 @@ func TestActivityContextDecodesDurableContextTags(t *testing.T) { } event := helpers.NewTaskScheduledEvent(4, "inspect-tags", nil, nil, nil) - event.GetTaskScheduled().Tags = contextprop.Encode(api.OrchestrationContextInfo{ - InstanceID: "tagged-instance", - Name: "tagged-parent", - Version: "v4", - ParentInstanceID: "root", - }, api.ContextFields{"tenant": "tagged"}) + event.GetTaskScheduled().Tags = contextprop.Encode(api.ContextFields{"tenant": "tagged"}, nil) response, err := NewTaskExecutor(registry).ExecuteActivity( context.Background(), "tagged-instance", @@ -202,9 +197,7 @@ func TestActivityContextDecodesDurableContextTags(t *testing.T) { if err := json.Unmarshal([]byte(response.GetTaskCompleted().GetResult().GetValue()), &output); err != nil { t.Fatal(err) } - if output.Orchestration.Name != "tagged-parent" || - output.Orchestration.Version != "v4" || - output.Orchestration.ParentInstanceID != "root" { + if output.Orchestration != (api.OrchestrationContextInfo{InstanceID: "tagged-instance"}) { t.Fatalf("unexpected orchestration identity: %+v", output.Orchestration) } if output.Fields["tenant"] != "tagged" { diff --git a/task/executor.go b/task/executor.go index cc26ec2d..95480291 100644 --- a/task/executor.go +++ b/task/executor.go @@ -10,10 +10,10 @@ import ( "time" "github.com/microsoft/durabletask-go/api" - "github.com/microsoft/durabletask-go/internal/contextprop" "github.com/microsoft/durabletask-go/internal/failure" "github.com/microsoft/durabletask-go/internal/helpers" "github.com/microsoft/durabletask-go/internal/protos" + "github.com/microsoft/durabletask-go/internal/tagcodec" "google.golang.org/protobuf/types/known/timestamppb" "google.golang.org/protobuf/types/known/wrapperspb" ) @@ -187,23 +187,10 @@ func (te *taskExecutor) ExecuteActivity(ctx context.Context, id api.InstanceID, } ctx = api.ContextWithFields(ctx, te.contextFields) ctx = helpers.ContextWithTraceContext(ctx, ts.GetParentTraceContext()) - tagInfo, tagFields := contextprop.Decode(ts.GetTags()) - ctx = api.ContextWithFields(ctx, tagFields) + ctx = api.ContextWithFields(ctx, api.ContextFields(tagcodec.DecodeContextFields(ts.GetTags()))) orchestrationInfo, _ := api.OrchestrationContextInfoFromContext(ctx) - if orchestrationInfo.Name == "" { - orchestrationInfo.Name = tagInfo.Name - } - if orchestrationInfo.Version == "" { - orchestrationInfo.Version = tagInfo.Version - } - if orchestrationInfo.ParentInstanceID == "" { - orchestrationInfo.ParentInstanceID = tagInfo.ParentInstanceID - } if orchestrationInfo.InstanceID == "" { - orchestrationInfo.InstanceID = tagInfo.InstanceID - if orchestrationInfo.InstanceID == "" { - orchestrationInfo.InstanceID = id - } + orchestrationInfo.InstanceID = id } ctx = api.WithOrchestrationContextInfo(ctx, orchestrationInfo) ctx = api.WithActivityContextInfo(ctx, api.ActivityContextInfo{ diff --git a/task/orchestrator.go b/task/orchestrator.go index 58bbd870..63f7b9fe 100644 --- a/task/orchestrator.go +++ b/task/orchestrator.go @@ -721,12 +721,8 @@ func (ctx *OrchestrationContext) internalScheduleActivity( helpers.GetTaskFunctionName(activity), options.rawInput, options.versionOrInherited(ctx.Version)) - scheduleTaskAction.GetScheduleTask().Tags = contextprop.Encode(api.OrchestrationContextInfo{ - InstanceID: ctx.ID, - Name: ctx.Name, - Version: ctx.Version, - ParentInstanceID: ctx.parentInstanceID, - }, ctx.contextFields, mergeStringMaps(ctx.orchestrationTags, options.tags)) + scheduleTaskAction.GetScheduleTask().Tags = contextprop.Encode( + ctx.contextFields, mergeStringMaps(ctx.orchestrationTags, options.tags)) ctx.pendingActions[scheduleTaskAction.Id] = scheduleTaskAction @@ -783,12 +779,6 @@ func (ctx *OrchestrationContext) internalCallSubOrchestrator( options.versionOrDefault(ctx.defaultVersion), ) createSubOrchestrationAction.GetCreateSubOrchestration().Tags = contextprop.Encode( - api.OrchestrationContextInfo{ - InstanceID: ctx.ID, - Name: ctx.Name, - Version: ctx.Version, - ParentInstanceID: ctx.parentInstanceID, - }, mergeStringMaps(ctx.contextFields, options.contextFields), mergeStringMaps(ctx.orchestrationTags, options.tags), ) @@ -1277,7 +1267,7 @@ func (ctx *OrchestrationContext) onExecutionStarted(es *protos.ExecutionStartedE ctx.Name = es.Name ctx.Version = es.GetVersion().GetValue() ctx.executionID = es.GetOrchestrationInstance().GetExecutionId().GetValue() - _, fields := contextprop.Decode(es.GetTags()) + fields := api.ContextFields(tagcodec.DecodeContextFields(es.GetTags())) ctx.contextFields = mergeStringMaps(ctx.contextFields, fields) ctx.orchestrationTags = tagcodec.DecodeUserTagsOrPlain(es.GetTags()) if parent := es.GetParentInstance(); parent != nil { @@ -1779,16 +1769,7 @@ func (ctx *OrchestrationContext) setCompleteInternal( completed := completedAction.GetCompleteOrchestration() if status == protos.OrchestrationStatus_ORCHESTRATION_STATUS_CONTINUED_AS_NEW { completed.NewVersion = ctx.continuedAsNewVersion - version := ctx.Version - if ctx.continuedAsNewVersion != nil { - version = ctx.continuedAsNewVersion.GetValue() - } - completed.Tags = contextprop.Encode(api.OrchestrationContextInfo{ - InstanceID: ctx.ID, - Name: ctx.Name, - Version: version, - ParentInstanceID: ctx.parentInstanceID, - }, ctx.contextFields, ctx.orchestrationTags) + completed.Tags = contextprop.Encode(ctx.contextFields, ctx.orchestrationTags) } else { completed.Tags = tagcodec.EncodeUserTags(ctx.orchestrationTags) } diff --git a/task/tag_identity_bench_test.go b/task/tag_identity_bench_test.go new file mode 100644 index 00000000..8d1549b3 --- /dev/null +++ b/task/tag_identity_bench_test.go @@ -0,0 +1,100 @@ +package task + +import ( + "context" + "testing" + "time" + + "github.com/microsoft/durabletask-go/internal/helpers" + "github.com/microsoft/durabletask-go/internal/protos" + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/types/known/timestamppb" + "google.golang.org/protobuf/types/known/wrapperspb" +) + +type activityWireSizeCase struct { + name string + current proto.Message + previous proto.Message +} + +func activityIdentityWireMessages(t testing.TB) []activityWireSizeCase { + t.Helper() + registry := NewTaskRegistry() + require.NoError(t, registry.AddOrchestratorNVersion("parent", "v1", func(ctx *OrchestrationContext) (any, error) { + ctx.CallActivity("inspect", WithRawActivityInput(`"input"`)) + return nil, nil + })) + started := helpers.NewExecutionStartedEvent("parent", "instance", nil, + helpers.NewParentInfo(7, "root", "root-instance"), nil, nil, wrapperspb.String("v1")) + result, err := NewTaskExecutor(registry).ExecuteOrchestrator(context.Background(), "instance", nil, + []*protos.HistoryEvent{started}, nil) + require.NoError(t, err) + action := scheduledActivityAction(t, result.Response).GetScheduleTask() + require.Nil(t, action.Tags) + trace := &protos.TraceContext{TraceParent: "00-0123456789abcdef0123456789abcdef-0123456789abcdef-01"} + action.ParentTraceContext = trace + event := helpers.NewTaskScheduledEvent(0, action.Name, action.Version, action.Input, trace) + event.Timestamp = timestamppb.New(time.Unix(1_700_000_000, 0).UTC()) + event.GetTaskScheduled().Tags = action.Tags + request := &protos.ActivityRequest{ + Name: action.Name, Version: action.Version, Input: action.Input, + OrchestrationInstance: &protos.OrchestrationInstance{InstanceId: "instance"}, + ParentTraceContext: trace, Tags: action.Tags, + } + previousTags := map[string]string{ + "__durabletask.context.encoding": "1", + "__durabletask.context.instance_id": "instance", + "__durabletask.context.orchestration_name": "parent", + "__durabletask.context.orchestration_version": "v1", + "__durabletask.context.parent_instance_id": "root-instance", + } + previousAction := proto.Clone(action).(*protos.ScheduleTaskAction) + previousAction.Tags = previousTags + previousEvent := proto.Clone(event).(*protos.HistoryEvent) + previousEvent.GetTaskScheduled().Tags = previousTags + previousRequest := proto.Clone(request).(*protos.ActivityRequest) + previousRequest.Tags = previousTags + return []activityWireSizeCase{ + {"action", action, previousAction}, + {"history", event, previousEvent}, + {"request", request, previousRequest}, + } +} + +func TestActivityIdentitySerializedSize(t *testing.T) { + for _, test := range activityIdentityWireMessages(t) { + currentBytes, err := proto.MarshalOptions{Deterministic: true}.Marshal(test.current) + require.NoError(t, err) + previousBytes, err := proto.MarshalOptions{Deterministic: true}.Marshal(test.previous) + require.NoError(t, err) + require.Less(t, len(currentBytes), len(previousBytes)) + t.Logf("%s: current=%d previous=%d delta=%d protobuf bytes", + test.name, len(currentBytes), len(previousBytes), len(previousBytes)-len(currentBytes)) + } +} + +func BenchmarkActivityIdentityWireSize(b *testing.B) { + for _, test := range activityIdentityWireMessages(b) { + for _, variant := range []struct { + name string + message proto.Message + }{ + {"current", test.current}, + {"previous", test.previous}, + } { + b.Run(test.name+"/"+variant.name, func(b *testing.B) { + message := variant.message + b.ReportAllocs() + for b.Loop() { + _, err := proto.MarshalOptions{Deterministic: true}.Marshal(message) + if err != nil { + b.Fatal(err) + } + } + b.ReportMetric(float64(proto.Size(message)), "wire-bytes") + }) + } + } +} diff --git a/task/tag_identity_test.go b/task/tag_identity_test.go new file mode 100644 index 00000000..e6ab67c6 --- /dev/null +++ b/task/tag_identity_test.go @@ -0,0 +1,378 @@ +package task + +import ( + "context" + "encoding/json" + "testing" + "time" + + "github.com/microsoft/durabletask-go/api" + "github.com/microsoft/durabletask-go/internal/contextprop" + "github.com/microsoft/durabletask-go/internal/helpers" + "github.com/microsoft/durabletask-go/internal/protos" + "github.com/microsoft/durabletask-go/internal/tagcodec" + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/types/known/timestamppb" + "google.golang.org/protobuf/types/known/wrapperspb" +) + +func scheduledActivityAction(t testing.TB, response *protos.OrchestratorResponse) *protos.OrchestratorAction { + t.Helper() + for _, action := range response.Actions { + if action.GetScheduleTask() != nil { + return action + } + } + t.Fatal("missing scheduled activity") + return nil +} + +func TestActivityIdentityUsesWorkItem(t *testing.T) { + for _, version := range []string{"", "v1"} { + for _, withCallerData := range []bool{false, true} { + t.Run(version+"/caller="+boolName(withCallerData), func(t *testing.T) { + registry := NewTaskRegistry() + options := []CallActivityOption{ + WithActivityVersion("a2"), + WithRawActivityInput(`"input"`), + } + var fields api.ContextFields + var tags map[string]string + if withCallerData { + fields = api.ContextFields{"tenant": "persisted", "empty": ""} + tags = map[string]string{"tenant": "user", "scope": "parent", "empty": ""} + options = append(options, WithActivityTags(map[string]string{"scope": "activity"})) + } + require.NoError(t, registry.AddOrchestratorNVersion("parent", version, func(ctx *OrchestrationContext) (any, error) { + ctx.CallActivity("inspect", options...) + return nil, nil + })) + started := helpers.NewExecutionStartedEvent("parent", "instance", nil, + helpers.NewParentInfo(7, "root", "root-instance"), nil, nil, wrapperspb.String(version)) + started.GetExecutionStarted().Tags = contextprop.Encode(fields, tags) + response := executeOrchestrationTurn(t, registry, "instance", nil, []*protos.HistoryEvent{started}) + action := scheduledActivityAction(t, response) + scheduled := action.GetScheduleTask() + require.EqualValues(t, 0, action.Id) + require.Equal(t, `"input"`, scheduled.GetInput().GetValue()) + require.Equal(t, "a2", scheduled.GetVersion().GetValue()) + persistedFields := api.ContextFields(tagcodec.DecodeContextFields(scheduled.Tags)) + wantInfo := api.OrchestrationContextInfo{InstanceID: "instance"} + if !withCallerData { + require.Nil(t, scheduled.Tags) + } + require.Equal(t, fields, persistedFields) + if withCallerData { + require.Equal(t, map[string]string{ + "tenant": "user", "scope": "activity", "empty": "", + }, tagcodec.DecodeUserTagsOrPlain(scheduled.Tags)) + } + require.NoError(t, registry.AddActivityNVersion("inspect", "a2", func(ctx ActivityContext) (any, error) { + orchestration, ok := api.OrchestrationContextInfoFromContext(ctx.Context()) + require.True(t, ok) + require.Equal(t, wantInfo, orchestration) + activity, ok := api.ActivityContextInfoFromContext(ctx.Context()) + require.True(t, ok) + require.Equal(t, api.ActivityContextInfo{ + InstanceID: "instance", Name: "inspect", Version: "a2", TaskID: 0, + }, activity) + require.Equal(t, fields, api.ContextFieldsFromContext(ctx.Context())) + var input string + require.NoError(t, ctx.GetInput(&input)) + return input, nil + })) + event := helpers.NewTaskScheduledEvent(action.Id, scheduled.Name, scheduled.Version, scheduled.Input, nil) + event.GetTaskScheduled().Tags = contextprop.Clone(scheduled.Tags) + result, err := NewTaskExecutor(registry).ExecuteActivity(context.Background(), "instance", event) + require.NoError(t, err) + require.Equal(t, `"input"`, result.GetTaskCompleted().GetResult().GetValue()) + }) + } + } +} + +func boolName(value bool) string { + if value { + return "yes" + } + return "no" +} + +func TestActivityTagsReplay(t *testing.T) { + registry := NewTaskRegistry() + require.NoError(t, registry.AddOrchestratorNVersion("parent", "v1", func(ctx *OrchestrationContext) (any, error) { + var output string + if err := ctx.CallActivity("first", WithRawActivityInput(`"original"`)).Await(&output); err != nil { + return nil, err + } + canceled, cancel := ctx.WithCancel() + cancel() + require.ErrorIs(t, canceled.WaitForSingleEvent("never", -1).Await(nil), ErrTaskCanceled) + canceled.CallActivity("never") + return nil, ctx.CallActivity("next", WithActivityInput(output)).Await(nil) + })) + started := helpers.NewExecutionStartedEvent("parent", "instance", nil, nil, nil, nil, wrapperspb.String("v1")) + started.GetExecutionStarted().Tags = contextprop.Encode(api.ContextFields{"tenant": "retained"}, + map[string]string{"user": "retained"}) + scheduled := helpers.NewTaskScheduledEvent(0, "first", wrapperspb.String("v1"), wrapperspb.String(`"original"`), nil) + scheduled.GetTaskScheduled().Tags = contextprop.Clone(started.GetExecutionStarted().Tags) + old := []*protos.HistoryEvent{helpers.NewOrchestratorStartedEvent(), started, scheduled} + delivered := []*protos.HistoryEvent{helpers.NewTaskCompletedEvent(0, wrapperspb.String(`"output"`))} + first := executeOrchestrationTurn(t, registry, "instance", old, delivered) + replayed := executeOrchestrationTurn(t, registry, "instance", append(old, delivered...), nil) + require.True(t, proto.Equal(first, replayed), "redelivery changed the actions") + next := scheduledActivityAction(t, first) + require.EqualValues(t, 1, next.Id, "canceled work must not consume a sequence number") + require.Equal(t, "next", next.GetScheduleTask().Name) + require.Equal(t, "v1", next.GetScheduleTask().GetVersion().GetValue()) + require.Equal(t, `"output"`, next.GetScheduleTask().GetInput().GetValue()) + require.Equal(t, map[string]string{ + tagcodec.ContextEncodingTag: "1", + tagcodec.ContextFieldPrefix + "tenant": "retained", + "user": "retained", + }, next.GetScheduleTask().Tags) +} + +func TestActivityTagsSurviveRetryAndRedelivery(t *testing.T) { + registry := NewTaskRegistry() + require.NoError(t, registry.AddOrchestratorNVersion("parent", "v1", func(ctx *OrchestrationContext) (any, error) { + options := []CallActivityOption{ + WithActivityRetryPolicy(&RetryPolicy{MaxAttempts: 2, InitialRetryInterval: time.Second}), + WithRawActivityInput(`"retry"`), + WithActivityTags(map[string]string{"scope": "activity"}), + } + return nil, ctx.CallActivity("flaky", options...).Await(nil) + })) + now := time.Unix(1_700_000_000, 0).UTC() + turn := helpers.NewOrchestratorStartedEvent() + turn.Timestamp = timestamppb.New(now) + started := helpers.NewExecutionStartedEvent("parent", "instance", nil, + helpers.NewParentInfo(7, "root", "root-instance"), nil, nil, wrapperspb.String("v1")) + started.GetExecutionStarted().Tags = contextprop.Encode(api.ContextFields{"tenant": "persisted"}, + map[string]string{"scope": "parent", "user": "retained"}) + initial := executeOrchestrationTurn(t, registry, "instance", nil, []*protos.HistoryEvent{turn, started}) + firstAction := scheduledActivityAction(t, initial) + scheduled := helpers.NewTaskScheduledEvent(0, "flaky", wrapperspb.String("v1"), wrapperspb.String(`"retry"`), nil) + scheduled.GetTaskScheduled().Tags = contextprop.Clone(firstAction.GetScheduleTask().Tags) + failure := helpers.NewTaskFailedEvent(0, &protos.TaskFailureDetails{ErrorType: "Transient", ErrorMessage: "retry"}) + history := []*protos.HistoryEvent{turn, started, scheduled, failure} + timerResponse := executeOrchestrationTurn(t, registry, "instance", history, nil) + require.Len(t, timerResponse.Actions, 1) + timer := timerResponse.Actions[0] + require.EqualValues(t, 1, timer.Id) + require.NotNil(t, timer.GetCreateTimer()) + history = append(history, helpers.NewTimerCreatedEvent(timer.Id, timer.GetCreateTimer().FireAt)) + retryTurn := helpers.NewOrchestratorStartedEvent() + retryTurn.Timestamp = timer.GetCreateTimer().FireAt + delivered := []*protos.HistoryEvent{retryTurn, helpers.NewTimerFiredEvent(timer.Id, timer.GetCreateTimer().FireAt, nil)} + retryResponse := executeOrchestrationTurn(t, registry, "instance", history, delivered, + WithContextFields(api.ContextFields{"host": "first"})) + replayed := executeOrchestrationTurn(t, registry, "instance", append(history, delivered...), nil, + WithContextFields(api.ContextFields{"host": "second"})) + require.True(t, proto.Equal(retryResponse, replayed)) + retry := scheduledActivityAction(t, retryResponse) + require.EqualValues(t, 2, retry.Id) + require.True(t, proto.Equal(firstAction.GetScheduleTask(), retry.GetScheduleTask())) + retryEvent := helpers.NewTaskScheduledEvent(retry.Id, "flaky", wrapperspb.String("v1"), wrapperspb.String(`"retry"`), nil) + retryEvent.GetTaskScheduled().Tags = contextprop.Clone(retry.GetScheduleTask().Tags) + require.NoError(t, registry.AddActivityNVersion("flaky", "v1", func(ctx ActivityContext) (any, error) { + info, _ := api.OrchestrationContextInfoFromContext(ctx.Context()) + want := api.OrchestrationContextInfo{InstanceID: "instance"} + require.Equal(t, want, info) + require.Equal(t, api.ContextFields{"tenant": "persisted"}, api.ContextFieldsFromContext(ctx.Context())) + return "recovered", nil + })) + completed, err := NewTaskExecutor(registry).ExecuteActivity(context.Background(), "instance", retryEvent) + require.NoError(t, err) + require.Equal(t, `"recovered"`, completed.GetTaskCompleted().GetResult().GetValue()) + finished := executeOrchestrationTurn(t, registry, "instance", + append(append(history, delivered...), retryEvent), []*protos.HistoryEvent{completed}) + require.Equal(t, protos.OrchestrationStatus_ORCHESTRATION_STATUS_COMPLETED, + completionAction(t, finished).OrchestrationStatus) +} + +func TestSubOrchestrationAndContinueAsNewUseNativeIdentity(t *testing.T) { + registry := NewTaskRegistry() + require.NoError(t, registry.AddOrchestratorNVersion("parent", "v1", func(ctx *OrchestrationContext) (any, error) { + ctx.CallSubOrchestrator("child", WithSubOrchestrationVersion("c1"), + WithSubOrchestrationTags(map[string]string{"scope": "child"}), + WithSubOrchestrationContextFields(api.ContextFields{"tenant": "child"})) + return nil, nil + })) + type state struct { + Info api.OrchestrationContextInfo + Fields api.ContextFields + } + require.NoError(t, registry.AddOrchestratorNVersion("child", "c1", func(ctx *OrchestrationContext) (any, error) { + info, _ := api.OrchestrationContextInfoFromContext(ctx.Context()) + require.Equal(t, api.OrchestrationContextInfo{ + InstanceID: ctx.ID, Name: "child", Version: "c1", ParentInstanceID: "instance", + }, info) + ctx.ContinueAsNew("next", WithContinueAsNewVersion("c2")) + return nil, nil + })) + require.NoError(t, registry.AddOrchestratorNVersion("child", "c2", func(ctx *OrchestrationContext) (any, error) { + info, _ := api.OrchestrationContextInfoFromContext(ctx.Context()) + return state{info, api.ContextFieldsFromContext(ctx.Context())}, nil + })) + started := helpers.NewExecutionStartedEvent("parent", "instance", nil, nil, nil, nil, wrapperspb.String("v1")) + started.GetExecutionStarted().Tags = contextprop.Encode(api.ContextFields{"tenant": "parent"}, + map[string]string{"scope": "parent", "user": "retained"}) + response := executeOrchestrationTurn(t, registry, "instance", nil, []*protos.HistoryEvent{started}) + var child *protos.CreateSubOrchestrationAction + for _, action := range response.Actions { + if action.GetCreateSubOrchestration() != nil { + child = action.GetCreateSubOrchestration() + } + } + require.NotNil(t, child) + wantTags := map[string]string{ + tagcodec.ContextEncodingTag: "1", + tagcodec.ContextFieldPrefix + "tenant": "child", + "scope": "child", + "user": "retained", + } + require.Equal(t, wantTags, child.Tags) + require.Equal(t, "instance:0000", child.InstanceId) + parent := helpers.NewParentInfo(0, "parent", "instance") + childStarted := helpers.NewExecutionStartedEvent(child.Name, child.InstanceId, child.Input, + parent, nil, nil, child.Version) + childStarted.GetExecutionStarted().Tags = contextprop.Clone(child.Tags) + childResponse := executeOrchestrationTurn(t, registry, api.InstanceID(child.InstanceId), nil, + []*protos.HistoryEvent{childStarted}) + continued := completionAction(t, childResponse) + require.Equal(t, protos.OrchestrationStatus_ORCHESTRATION_STATUS_CONTINUED_AS_NEW, continued.OrchestrationStatus) + require.Equal(t, wantTags, continued.Tags) + require.Equal(t, "c2", continued.GetNewVersion().GetValue()) + nextStarted := helpers.NewExecutionStartedEvent(child.Name, child.InstanceId, continued.Result, + parent, nil, nil, continued.NewVersion) + nextStarted.GetExecutionStarted().Tags = contextprop.Clone(continued.Tags) + continued.Tags["scope"] = "mutated" + next := executeOrchestrationTurn(t, registry, api.InstanceID(child.InstanceId), nil, []*protos.HistoryEvent{nextStarted}) + var output state + require.NoError(t, json.Unmarshal([]byte(completionResult(t, next)), &output)) + require.Equal(t, api.OrchestrationContextInfo{ + InstanceID: api.InstanceID(child.InstanceId), Name: "child", Version: "c2", ParentInstanceID: "instance", + }, output.Info) + require.Equal(t, api.ContextFields{"tenant": "child"}, output.Fields) + require.Equal(t, map[string]string{ + tagcodec.ContextEncodingTag: "1", "scope": "child", "user": "retained", + }, completionAction(t, next).Tags) +} + +func TestSubOrchestrationTagsReplayAndRetry(t *testing.T) { + registry := NewTaskRegistry() + require.NoError(t, registry.AddOrchestratorNVersion("parent", "v1", func(ctx *OrchestrationContext) (any, error) { + var output string + err := ctx.CallSubOrchestrator("child", WithSubOrchestrationVersion("c1"), + WithRawSubOrchestratorInput(`"input"`), + WithSubOrchestrationTags(map[string]string{"scope": "child"}), + WithSubOrchestrationContextFields(api.ContextFields{"tenant": "child"}), + WithSubOrchestrationRetryPolicy(&RetryPolicy{ + MaxAttempts: 2, InitialRetryInterval: time.Second, + })).Await(&output) + return output, err + })) + turn := helpers.NewOrchestratorStartedEvent() + turn.Timestamp = timestamppb.New(time.Unix(1_700_000_000, 0).UTC()) + started := helpers.NewExecutionStartedEvent("parent", "instance", nil, nil, nil, nil, wrapperspb.String("v1")) + started.GetExecutionStarted().Tags = contextprop.Encode(api.ContextFields{"tenant": "parent"}, + map[string]string{"scope": "parent", "user": ""}) + created := helpers.NewSubOrchestrationCreatedEvent(0, "child", wrapperspb.String("c1"), + wrapperspb.String(`"input"`), "instance:0000", nil) + created.GetSubOrchestrationInstanceCreated().Tags = contextprop.Encode(api.ContextFields{"tenant": "child"}, + map[string]string{"scope": "child", "user": ""}) + failed := &protos.HistoryEvent{EventType: &protos.HistoryEvent_SubOrchestrationInstanceFailed{ + SubOrchestrationInstanceFailed: &protos.SubOrchestrationInstanceFailedEvent{ + TaskScheduledId: 0, FailureDetails: &protos.TaskFailureDetails{ErrorType: "Transient", ErrorMessage: "retry"}, + }, + }} + history := []*protos.HistoryEvent{turn, started, created, failed} + timer := singleRetryTimer(t, executeOrchestrationTurn(t, registry, "instance", history, nil)) + history = append(history, helpers.NewTimerCreatedEvent(1, timer.FireAt)) + retryTurn := helpers.NewOrchestratorStartedEvent() + retryTurn.Timestamp = timer.FireAt + delivered := []*protos.HistoryEvent{retryTurn, helpers.NewTimerFiredEvent(1, timer.FireAt, nil)} + response := executeOrchestrationTurn(t, registry, "instance", history, delivered) + replayed := executeOrchestrationTurn(t, registry, "instance", append(history, delivered...), nil) + require.True(t, proto.Equal(response, replayed)) + require.Len(t, response.Actions, 1) + retryAction := response.Actions[0] + require.EqualValues(t, 2, retryAction.Id) + retry := retryAction.GetCreateSubOrchestration() + require.NotNil(t, retry) + require.Equal(t, "instance:0002", retry.InstanceId) + require.Equal(t, "child", retry.Name) + require.Equal(t, "c1", retry.GetVersion().GetValue()) + require.Equal(t, `"input"`, retry.GetInput().GetValue()) + require.Equal(t, map[string]string{ + tagcodec.ContextEncodingTag: "1", + tagcodec.ContextFieldPrefix + "tenant": "child", + "scope": "child", + "user": "", + }, retry.Tags) + retryEvent := helpers.NewSubOrchestrationCreatedEvent(retryAction.Id, retry.Name, retry.Version, + retry.Input, retry.InstanceId, nil) + retryEvent.GetSubOrchestrationInstanceCreated().Tags = contextprop.Clone(retry.Tags) + completed := &protos.HistoryEvent{EventType: &protos.HistoryEvent_SubOrchestrationInstanceCompleted{ + SubOrchestrationInstanceCompleted: &protos.SubOrchestrationInstanceCompletedEvent{ + TaskScheduledId: retryAction.Id, Result: wrapperspb.String(`"completed"`), + }, + }} + finished := executeOrchestrationTurn(t, registry, "instance", + append(append(history, delivered...), retryEvent), []*protos.HistoryEvent{completed}) + require.Equal(t, `"completed"`, completionResult(t, finished)) +} + +func TestUntaggedLifecycleActionsHaveNoIdentityTags(t *testing.T) { + registry := NewTaskRegistry() + require.NoError(t, registry.AddOrchestratorN("parent", func(ctx *OrchestrationContext) (any, error) { + ctx.CallActivity("activity") + ctx.CallSubOrchestrator("child") + ctx.ContinueAsNew(nil) + return nil, nil + })) + started := helpers.NewExecutionStartedEvent("parent", "instance", nil, nil, nil, nil) + for _, tags := range []map[string]string{nil, {}} { + started.GetExecutionStarted().Tags = tags + response := executeOrchestrationTurn(t, registry, "instance", nil, []*protos.HistoryEvent{started}) + require.Len(t, response.Actions, 3) + for _, action := range response.Actions { + switch { + case action.GetScheduleTask() != nil: + require.Nil(t, action.GetScheduleTask().Tags) + case action.GetCreateSubOrchestration() != nil: + require.Nil(t, action.GetCreateSubOrchestration().Tags) + case action.GetCompleteOrchestration() != nil: + require.Nil(t, action.GetCompleteOrchestration().Tags) + default: + t.Fatalf("unexpected action: %v", action) + } + } + } +} + +func TestActivityExplicitContextIdentityIsPreserved(t *testing.T) { + registry := NewTaskRegistry() + require.NoError(t, registry.AddActivityN("inspect", func(ctx ActivityContext) (any, error) { + info, _ := api.OrchestrationContextInfoFromContext(ctx.Context()) + require.Equal(t, api.OrchestrationContextInfo{ + InstanceID: "explicit-instance", Name: "explicit-name", Version: "explicit-version", ParentInstanceID: "explicit-parent", + }, info) + require.Equal(t, api.ContextFields{"tenant": "durable", "worker": "local"}, + api.ContextFieldsFromContext(ctx.Context())) + return "retained", nil + })) + base := api.WithOrchestrationContextInfo(context.Background(), api.OrchestrationContextInfo{ + InstanceID: "explicit-instance", Name: "explicit-name", Version: "explicit-version", ParentInstanceID: "explicit-parent", + }) + event := helpers.NewTaskScheduledEvent(4, "inspect", nil, nil, nil) + event.GetTaskScheduled().Tags = contextprop.Encode(api.ContextFields{"tenant": "durable"}, + map[string]string{"user": "not a context field"}) + response, err := NewTaskExecutor(registry, WithContextFields(api.ContextFields{ + "tenant": "worker", "worker": "local", + })).ExecuteActivity(base, "instance", event) + require.NoError(t, err) + require.Equal(t, `"retained"`, response.GetTaskCompleted().GetResult().GetValue()) +} diff --git a/task/versioning_test.go b/task/versioning_test.go index d07c8d4e..f009096e 100644 --- a/task/versioning_test.go +++ b/task/versioning_test.go @@ -350,6 +350,7 @@ func TestActivityTagsRejectReservedAndEmptyKeys(t *testing.T) { for _, tags := range []map[string]string{ {"": "value"}, {api.ReservedContextFieldPrefix + "tenant": "value"}, + {"__durabletask.tags.tenant": "value"}, } { if err := WithActivityTags(tags)(new(callActivityOptions), api.DefaultDataConverter()); err == nil { t.Fatalf("WithActivityTags(%#v) succeeded", tags) @@ -407,8 +408,11 @@ func TestContinueAsNewCarriesTagsAndNextVersion(t *testing.T) { if completed.GetTags()["__durabletask.context.field.correlation_id"] != "42" { t.Fatalf("continue-as-new context tags = %#v", completed.GetTags()) } - if completed.GetTags()["__durabletask.context.orchestration_version"] != "v2" { - t.Fatalf("continue-as-new version tag = %#v", completed.GetTags()) + if completed.GetNewVersion().GetValue() != "v2" { + t.Fatalf("continue-as-new version = %q", completed.GetNewVersion().GetValue()) + } + if _, ok := completed.GetTags()["__durabletask.context.orchestration_version"]; ok { + t.Fatalf("continue-as-new identity tags = %#v", completed.GetTags()) } }