From 9cabfc935ecbd46473a3b5ab9ef85205d6dce15d Mon Sep 17 00:00:00 2001 From: Tomer Rosenthal <17064840+torosent@users.noreply.github.com> Date: Tue, 6 Oct 2026 13:20:09 -0700 Subject: [PATCH 1/2] align default identity tags with .NET Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- CHANGELOG.md | 1 + README.md | 25 ++ api/context.go | 4 + internal/contextprop/tags.go | 8 +- internal/contextprop/tags_test.go | 133 +++++++++- task/activity.go | 20 +- task/context_test.go | 16 +- task/orchestrator.go | 35 +-- task/tag_identity_bench_test.go | 82 ++++++ task/tag_identity_test.go | 412 ++++++++++++++++++++++++++++++ task/versioning_test.go | 8 +- 11 files changed, 704 insertions(+), 40 deletions(-) create mode 100644 task/tag_identity_bench_test.go create mode 100644 task/tag_identity_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 871dfa56..9aa0db80 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 no longer add owning-orchestration identity tags by default, aligning default automatic tags with .NET. **Migration:** new activity contexts guarantee the owning instance ID, but not the owning orchestration's name, version, or parent instance ID; use `task.WithActivityOrchestrationIdentity()` to persist that richer identity per activity and retry. Historical identity tags and explicit executor context remain readable. Activity identity, full orchestrator identity, 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..bf660c3f 100644 --- a/README.md +++ b/README.md @@ -517,6 +517,31 @@ 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 by default. 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 legacy plain context fields. +Distributed trace context remains a separate protocol field. + +**Migration:** newly scheduled activities always retain their own name, version, +and task ID in `api.ActivityContextInfo`, and their owning instance ID in +`api.OrchestrationContextInfo`. The owning orchestration's name, version, and parent +instance ID are no longer supplied automatically to activity contexts. If an +activity needs that richer Go identity, add the deterministic per-call option: + +```go +ctx.CallActivity("Inspect", task.WithActivityOrchestrationIdentity()) +``` + +This option persists the identity for every attempt, including retries and +recovery. Previously scheduled activities still decode their historical identity +tags, and explicitly supplied executor context remains supported. Orchestrator +contexts retain their complete persisted identity, including after sub-orchestration +starts and ContinueAsNew version changes. No protocol or server change is required: +these identities already have native orchestration fields, whereas activity work +items only carry the owning instance ID. + 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..f40f3434 100644 --- a/api/context.go +++ b/api/context.go @@ -12,6 +12,10 @@ const ReservedContextFieldPrefix = "__durabletask.context." type ContextFields map[string]string // OrchestrationContextInfo identifies the orchestration associated with a task context. +// Orchestrator contexts contain the full persisted identity. In activities, only +// InstanceID is guaranteed by default; Name, Version, and ParentInstanceID require +// task.WithActivityOrchestrationIdentity, historical identity tags, or an explicitly +// supplied executor context. 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..92350d96 100644 --- a/internal/contextprop/tags.go +++ b/internal/contextprop/tags.go @@ -14,9 +14,10 @@ const ( parentInstanceIDTag = api.ReservedContextFieldPrefix + "parent_instance_id" ) -// Encode returns a new tag map containing immutable fields and orchestration identity. +// Encode returns a new tag map containing caller fields and user tags. +// A non-nil info opts into the owning orchestration's identity. func Encode( - info api.OrchestrationContextInfo, + info *api.OrchestrationContextInfo, fields api.ContextFields, userTags ...map[string]string, ) map[string]string { @@ -24,6 +25,9 @@ func Encode( if len(userTags) > 0 { tags = tagcodec.Merge(tags, tagcodec.EncodeUserTags(userTags[0])) } + if info == nil { + return tags + } if tags == nil { tags = make(map[string]string, 5) } diff --git a/internal/contextprop/tags_test.go b/internal/contextprop/tags_test.go index 6ad00751..5b5eb434 100644 --- a/internal/contextprop/tags_test.go +++ b/internal/contextprop/tags_test.go @@ -5,10 +5,11 @@ 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{ + tags := Encode(&api.OrchestrationContextInfo{ InstanceID: "instance", Name: "orchestration", Version: "v1", @@ -30,7 +31,7 @@ func TestEncodeDecode(t *testing.T) { func TestEncodeSeparatesContextFieldsAndUserTags(t *testing.T) { tags := Encode( - api.OrchestrationContextInfo{}, + nil, api.ContextFields{"tenant": "context"}, map[string]string{"team": "tag"}, ) @@ -48,7 +49,7 @@ func TestEncodeSeparatesContextFieldsAndUserTags(t *testing.T) { } func TestEncodeOverwritesReservedCallerFields(t *testing.T) { - tags := Encode(api.OrchestrationContextInfo{}, api.ContextFields{ + tags := Encode(&api.OrchestrationContextInfo{}, api.ContextFields{ api.ReservedContextFieldPrefix + "orchestration_version": "spoofed", }) info, fields := Decode(tags) @@ -59,3 +60,129 @@ func TestEncodeOverwritesReservedCallerFields(t *testing.T) { t.Fatalf("reserved field leaked into caller fields: %v", fields) } } + +func TestEncodeWithoutIdentity(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(nil, test.fields, test.tags) + require.Equal(t, test.want, encoded) + info, fields := Decode(encoded) + require.Equal(t, api.OrchestrationContextInfo{}, info) + 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 TestDecodeRetainedTagFormats(t *testing.T) { + for _, test := range []struct { + name string + tags map[string]string + fields api.ContextFields + user map[string]string + }{ + {name: "absent"}, + {name: "empty", tags: map[string]string{}}, + { + name: "legacy plain fields", + tags: map[string]string{"tenant": "plain", "empty": ""}, + fields: api.ContextFields{"tenant": "plain", "empty": ""}, + user: map[string]string{"tenant": "plain", "empty": ""}, + }, + { + name: "legacy namespaced fields and user tags", + tags: map[string]string{ + tagcodec.ContextFieldPrefix + "tenant": "context", + tagcodec.UserTagPrefix + "tenant": "user", + tagcodec.ContextFieldPrefix + "empty": "", + }, + fields: api.ContextFields{"tenant": "context", "empty": ""}, + user: map[string]string{"tenant": "user"}, + }, + { + name: "encoded plain user tags are not fields", + tags: map[string]string{ + tagcodec.ContextEncodingTag: "1", + "tenant": "user", + "empty": "", + }, + user: map[string]string{"tenant": "user", "empty": ""}, + }, + } { + t.Run(test.name, func(t *testing.T) { + _, fields := Decode(test.tags) + require.Equal(t, test.fields, fields) + require.Equal(t, test.user, tagcodec.DecodeUserTagsOrPlain(test.tags)) + _, missing := fields["missing"] + require.False(t, missing) + if _, present := test.fields["empty"]; present { + value, ok := fields["empty"] + require.True(t, ok) + require.Empty(t, value) + } + }) + } +} + +func TestIdentityEncodingRetainsExplicitEmptyValues(t *testing.T) { + encoded := Encode(&api.OrchestrationContextInfo{}, nil) + require.Equal(t, map[string]string{ + tagcodec.ContextEncodingTag: "1", + instanceIDTag: "", + nameTag: "", + versionTag: "", + parentInstanceIDTag: "", + }, encoded) + _, fields := Decode(encoded) + require.Nil(t, fields) + require.Nil(t, tagcodec.DecodeUserTagsOrPlain(encoded)) +} + +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/activity.go b/task/activity.go index 3f44cd71..5121953a 100644 --- a/task/activity.go +++ b/task/activity.go @@ -16,10 +16,11 @@ import ( type CallActivityOption func(*callActivityOptions, api.DataConverter) error type callActivityOptions struct { - rawInput *wrapperspb.StringValue - version *wrapperspb.StringValue - retryPolicy *RetryPolicy - tags map[string]string + rawInput *wrapperspb.StringValue + version *wrapperspb.StringValue + retryPolicy *RetryPolicy + tags map[string]string + includeOrchestrationIdentity bool } func (options *callActivityOptions) versionOrInherited(inheritedVersion string) *wrapperspb.StringValue { @@ -133,6 +134,17 @@ func WithActivityTags(tags map[string]string) CallActivityOption { } } +// WithActivityOrchestrationIdentity persists the owning orchestration's name, +// version, and parent instance ID for OrchestrationContextInfoFromContext in the +// activity. Without this option, only the owning instance ID is guaranteed. +// Activity identity, caller context fields, user tags, and tracing are unaffected. +func WithActivityOrchestrationIdentity() CallActivityOption { + return func(opt *callActivityOptions, _ api.DataConverter) error { + opt.includeOrchestrationIdentity = true + return nil + } +} + // WithActivityRetryPolicy snapshots policy when this option is created. Later // caller mutations do not affect activity retries. func WithActivityRetryPolicy(policy *RetryPolicy) CallActivityOption { diff --git a/task/context_test.go b/task/context_test.go index 84381cca..77c6907c 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(nil, fields) fields["tenant"] = "mutated" events := []*protos.HistoryEvent{helpers.NewOrchestratorStartedEvent(), started} @@ -180,12 +180,14 @@ 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 = map[string]string{ + "__durabletask.context.encoding": "1", + "__durabletask.context.instance_id": "tagged-instance", + "__durabletask.context.orchestration_name": "tagged-parent", + "__durabletask.context.orchestration_version": "v4", + "__durabletask.context.parent_instance_id": "root", + "__durabletask.context.field.tenant": "tagged", + } response, err := NewTaskExecutor(registry).ExecuteActivity( context.Background(), "tagged-instance", diff --git a/task/orchestrator.go b/task/orchestrator.go index 58bbd870..f06b82cb 100644 --- a/task/orchestrator.go +++ b/task/orchestrator.go @@ -721,12 +721,17 @@ 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)) + var info *api.OrchestrationContextInfo + if options.includeOrchestrationIdentity { + info = &api.OrchestrationContextInfo{ + InstanceID: ctx.ID, + Name: ctx.Name, + Version: ctx.Version, + ParentInstanceID: ctx.parentInstanceID, + } + } + scheduleTaskAction.GetScheduleTask().Tags = contextprop.Encode( + info, ctx.contextFields, mergeStringMaps(ctx.orchestrationTags, options.tags)) ctx.pendingActions[scheduleTaskAction.Id] = scheduleTaskAction @@ -783,12 +788,7 @@ 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, - }, + nil, mergeStringMaps(ctx.contextFields, options.contextFields), mergeStringMaps(ctx.orchestrationTags, options.tags), ) @@ -1779,16 +1779,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(nil, 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..005946ee --- /dev/null +++ b/task/tag_identity_bench_test.go @@ -0,0 +1,82 @@ +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" +) + +func activityIdentityWireMessages(t testing.TB, includeIdentity bool) []proto.Message { + t.Helper() + registry := NewTaskRegistry() + require.NoError(t, registry.AddOrchestratorNVersion("parent", "v1", func(ctx *OrchestrationContext) (any, error) { + options := []CallActivityOption{WithRawActivityInput(`"input"`)} + if includeIdentity { + options = append(options, WithActivityOrchestrationIdentity()) + } + ctx.CallActivity("inspect", options...) + 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() + if includeIdentity { + require.Equal(t, legacyIdentityTags(), action.Tags, "opt-in must reproduce the legacy identity payload") + } else { + 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, + } + return []proto.Message{action, event, request} +} + +func TestActivityIdentitySerializedSize(t *testing.T) { + defaultMessages := activityIdentityWireMessages(t, false) + legacyMessages := activityIdentityWireMessages(t, true) + for index, name := range []string{"action", "history", "request"} { + defaultBytes, err := proto.MarshalOptions{Deterministic: true}.Marshal(defaultMessages[index]) + require.NoError(t, err) + legacyBytes, err := proto.MarshalOptions{Deterministic: true}.Marshal(legacyMessages[index]) + require.NoError(t, err) + require.Less(t, len(defaultBytes), len(legacyBytes)) + t.Logf("%s: default=%d legacy/opt-in=%d delta=%d protobuf bytes", + name, len(defaultBytes), len(legacyBytes), len(legacyBytes)-len(defaultBytes)) + } +} + +func BenchmarkActivityIdentityWireSize(b *testing.B) { + for _, includeIdentity := range []bool{false, true} { + messages := activityIdentityWireMessages(b, includeIdentity) + for index, name := range []string{"action", "history", "request"} { + b.Run(name+"/identity="+boolName(includeIdentity), func(b *testing.B) { + message := messages[index] + 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..8a78f599 --- /dev/null +++ b/task/tag_identity_test.go @@ -0,0 +1,412 @@ +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 TestScheduledActivityIdentityIsOptIn(t *testing.T) { + for _, version := range []string{"", "v1"} { + for _, includeIdentity := range []bool{false, true} { + for _, withCallerData := range []bool{false, true} { + t.Run(version+"/identity="+boolName(includeIdentity)+"/caller="+boolName(withCallerData), func(t *testing.T) { + registry := NewTaskRegistry() + options := []CallActivityOption{ + WithActivityVersion("a2"), + WithRawActivityInput(`"input"`), + } + if includeIdentity { + options = append(options, WithActivityOrchestrationIdentity()) + } + 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(nil, 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()) + info, persistedFields := contextprop.Decode(scheduled.Tags) + wantInfo := api.OrchestrationContextInfo{InstanceID: "instance"} + if includeIdentity { + wantInfo.Name, wantInfo.Version, wantInfo.ParentInstanceID = "parent", version, "root-instance" + require.Equal(t, wantInfo, info) + } else { + require.Equal(t, api.OrchestrationContextInfo{}, info) + 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 legacyIdentityTags() map[string]string { + return 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", + } +} + +func TestLegacyActivityTagsReplayWithNewWriter(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", WithActivityOrchestrationIdentity()) + return nil, ctx.CallActivity("next", WithActivityInput(output)).Await(nil) + })) + started := helpers.NewExecutionStartedEvent("parent", "instance", nil, nil, nil, nil, wrapperspb.String("v1")) + started.GetExecutionStarted().Tags = legacyIdentityTags() + started.GetExecutionStarted().Tags["user"] = "retained" + started.GetExecutionStarted().Tags[tagcodec.ContextFieldPrefix+"tenant"] = "retained" + scheduled := helpers.NewTaskScheduledEvent(0, "first", wrapperspb.String("v1"), wrapperspb.String(`"original"`), nil) + scheduled.GetTaskScheduled().Tags = legacyIdentityTags() + 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 TestActivityIdentityChoiceSurvivesRetryAndRedelivery(t *testing.T) { + for _, includeIdentity := range []bool{false, true} { + t.Run(boolName(includeIdentity), func(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"}), + } + if includeIdentity { + options = append(options, WithActivityOrchestrationIdentity()) + } + 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(nil, 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"} + if includeIdentity { + want.Name, want.Version, want.ParentInstanceID = "parent", "v1", "root-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 = legacyIdentityTags() + started.GetExecutionStarted().Tags["scope"] = "parent" + started.GetExecutionStarted().Tags["user"] = "retained" + started.GetExecutionStarted().Tags[tagcodec.ContextFieldPrefix+"tenant"] = "parent" + 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 TestLegacySubOrchestrationTagsReplayAndRetry(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(nil, 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 = legacyIdentityTags() + 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, {}, legacyIdentityTags()} { + 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 TestActivityExplicitContextOverridesHistoricalIdentity(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: "v1", ParentInstanceID: "root-instance", + }, 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", + }) + event := helpers.NewTaskScheduledEvent(4, "inspect", nil, nil, nil) + event.GetTaskScheduled().Tags = legacyIdentityTags() + event.GetTaskScheduled().Tags[tagcodec.ContextFieldPrefix+"tenant"] = "durable" + event.GetTaskScheduled().Tags["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()) } } From 561424ad5162a0bb42cfc075586ce1af95c19c82 Mon Sep 17 00:00:00 2001 From: Tomer Rosenthal <17064840+torosent@users.noreply.github.com> Date: Tue, 6 Oct 2026 13:38:53 -0700 Subject: [PATCH 2/2] remove identity migration support Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- CHANGELOG.md | 2 +- README.md | 26 +-- api/context.go | 8 +- internal/contextprop/tags.go | 42 +---- internal/contextprop/tags_test.go | 124 +------------ task/activity.go | 20 +- task/context_test.go | 17 +- task/executor.go | 19 +- task/orchestrator.go | 16 +- task/tag_identity_bench_test.go | 68 ++++--- task/tag_identity_test.go | 298 +++++++++++++----------------- 11 files changed, 205 insertions(+), 435 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9aa0db80..a2195b22 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -57,7 +57,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed -- Activities, sub-orchestrations, and ContinueAsNew no longer add owning-orchestration identity tags by default, aligning default automatic tags with .NET. **Migration:** new activity contexts guarantee the owning instance ID, but not the owning orchestration's name, version, or parent instance ID; use `task.WithActivityOrchestrationIdentity()` to persist that richer identity per activity and retry. Historical identity tags and explicit executor context remain readable. Activity identity, full orchestrator identity, user-tag inheritance, immutable context fields, namespace markers, and distributed tracing are preserved. +- 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 bf660c3f..219e77bc 100644 --- a/README.md +++ b/README.md @@ -518,29 +518,17 @@ 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 by default. Without caller tags or context fields, their tag maps +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 legacy plain context fields. +present, the encoding marker separates user tags from context fields. Distributed trace context remains a separate protocol field. -**Migration:** newly scheduled activities always retain their own name, version, -and task ID in `api.ActivityContextInfo`, and their owning instance ID in -`api.OrchestrationContextInfo`. The owning orchestration's name, version, and parent -instance ID are no longer supplied automatically to activity contexts. If an -activity needs that richer Go identity, add the deterministic per-call option: - -```go -ctx.CallActivity("Inspect", task.WithActivityOrchestrationIdentity()) -``` - -This option persists the identity for every attempt, including retries and -recovery. Previously scheduled activities still decode their historical identity -tags, and explicitly supplied executor context remains supported. Orchestrator -contexts retain their complete persisted identity, including after sub-orchestration -starts and ContinueAsNew version changes. No protocol or server change is required: -these identities already have native orchestration fields, whereas activity work -items only carry the owning instance ID. +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. diff --git a/api/context.go b/api/context.go index f40f3434..05086eff 100644 --- a/api/context.go +++ b/api/context.go @@ -5,17 +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. In activities, only -// InstanceID is guaranteed by default; Name, Version, and ParentInstanceID require -// task.WithActivityOrchestrationIdentity, historical identity tags, or an explicitly -// supplied executor context. The activity's own identity is in ActivityContextInfo. +// 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 92350d96..ea36b582 100644 --- a/internal/contextprop/tags.go +++ b/internal/contextprop/tags.go @@ -7,47 +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 caller fields and user tags. -// A non-nil info opts into the owning orchestration's 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 info == nil { - return tags - } - 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)) +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 5b5eb434..a9609967 100644 --- a/internal/contextprop/tags_test.go +++ b/internal/contextprop/tags_test.go @@ -8,60 +8,7 @@ import ( "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( - nil, - 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 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 TestEncodeWithoutIdentity(t *testing.T) { +func TestEncodeCallerData(t *testing.T) { for _, test := range []struct { name string fields api.ContextFields @@ -92,10 +39,9 @@ func TestEncodeWithoutIdentity(t *testing.T) { }, } { t.Run(test.name, func(t *testing.T) { - encoded := Encode(nil, test.fields, test.tags) + encoded := Encode(test.fields, test.tags) require.Equal(t, test.want, encoded) - info, fields := Decode(encoded) - require.Equal(t, api.OrchestrationContextInfo{}, info) + fields := api.ContextFields(tagcodec.DecodeContextFields(encoded)) if len(test.fields) == 0 { require.Nil(t, fields, "user tags must not become context fields") } else { @@ -114,70 +60,6 @@ func TestEncodeWithoutIdentity(t *testing.T) { } } -func TestDecodeRetainedTagFormats(t *testing.T) { - for _, test := range []struct { - name string - tags map[string]string - fields api.ContextFields - user map[string]string - }{ - {name: "absent"}, - {name: "empty", tags: map[string]string{}}, - { - name: "legacy plain fields", - tags: map[string]string{"tenant": "plain", "empty": ""}, - fields: api.ContextFields{"tenant": "plain", "empty": ""}, - user: map[string]string{"tenant": "plain", "empty": ""}, - }, - { - name: "legacy namespaced fields and user tags", - tags: map[string]string{ - tagcodec.ContextFieldPrefix + "tenant": "context", - tagcodec.UserTagPrefix + "tenant": "user", - tagcodec.ContextFieldPrefix + "empty": "", - }, - fields: api.ContextFields{"tenant": "context", "empty": ""}, - user: map[string]string{"tenant": "user"}, - }, - { - name: "encoded plain user tags are not fields", - tags: map[string]string{ - tagcodec.ContextEncodingTag: "1", - "tenant": "user", - "empty": "", - }, - user: map[string]string{"tenant": "user", "empty": ""}, - }, - } { - t.Run(test.name, func(t *testing.T) { - _, fields := Decode(test.tags) - require.Equal(t, test.fields, fields) - require.Equal(t, test.user, tagcodec.DecodeUserTagsOrPlain(test.tags)) - _, missing := fields["missing"] - require.False(t, missing) - if _, present := test.fields["empty"]; present { - value, ok := fields["empty"] - require.True(t, ok) - require.Empty(t, value) - } - }) - } -} - -func TestIdentityEncodingRetainsExplicitEmptyValues(t *testing.T) { - encoded := Encode(&api.OrchestrationContextInfo{}, nil) - require.Equal(t, map[string]string{ - tagcodec.ContextEncodingTag: "1", - instanceIDTag: "", - nameTag: "", - versionTag: "", - parentInstanceIDTag: "", - }, encoded) - _, fields := Decode(encoded) - require.Nil(t, fields) - require.Nil(t, tagcodec.DecodeUserTagsOrPlain(encoded)) -} - func TestCloneDoesNotAliasTags(t *testing.T) { require.Nil(t, Clone[map[string]string](nil)) require.Nil(t, Clone(map[string]string{})) diff --git a/task/activity.go b/task/activity.go index 5121953a..3f44cd71 100644 --- a/task/activity.go +++ b/task/activity.go @@ -16,11 +16,10 @@ import ( type CallActivityOption func(*callActivityOptions, api.DataConverter) error type callActivityOptions struct { - rawInput *wrapperspb.StringValue - version *wrapperspb.StringValue - retryPolicy *RetryPolicy - tags map[string]string - includeOrchestrationIdentity bool + rawInput *wrapperspb.StringValue + version *wrapperspb.StringValue + retryPolicy *RetryPolicy + tags map[string]string } func (options *callActivityOptions) versionOrInherited(inheritedVersion string) *wrapperspb.StringValue { @@ -134,17 +133,6 @@ func WithActivityTags(tags map[string]string) CallActivityOption { } } -// WithActivityOrchestrationIdentity persists the owning orchestration's name, -// version, and parent instance ID for OrchestrationContextInfoFromContext in the -// activity. Without this option, only the owning instance ID is guaranteed. -// Activity identity, caller context fields, user tags, and tracing are unaffected. -func WithActivityOrchestrationIdentity() CallActivityOption { - return func(opt *callActivityOptions, _ api.DataConverter) error { - opt.includeOrchestrationIdentity = true - return nil - } -} - // WithActivityRetryPolicy snapshots policy when this option is created. Later // caller mutations do not affect activity retries. func WithActivityRetryPolicy(policy *RetryPolicy) CallActivityOption { diff --git a/task/context_test.go b/task/context_test.go index 77c6907c..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(nil, 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,14 +180,7 @@ func TestActivityContextDecodesDurableContextTags(t *testing.T) { } event := helpers.NewTaskScheduledEvent(4, "inspect-tags", nil, nil, nil) - event.GetTaskScheduled().Tags = map[string]string{ - "__durabletask.context.encoding": "1", - "__durabletask.context.instance_id": "tagged-instance", - "__durabletask.context.orchestration_name": "tagged-parent", - "__durabletask.context.orchestration_version": "v4", - "__durabletask.context.parent_instance_id": "root", - "__durabletask.context.field.tenant": "tagged", - } + event.GetTaskScheduled().Tags = contextprop.Encode(api.ContextFields{"tenant": "tagged"}, nil) response, err := NewTaskExecutor(registry).ExecuteActivity( context.Background(), "tagged-instance", @@ -204,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 f06b82cb..63f7b9fe 100644 --- a/task/orchestrator.go +++ b/task/orchestrator.go @@ -721,17 +721,8 @@ func (ctx *OrchestrationContext) internalScheduleActivity( helpers.GetTaskFunctionName(activity), options.rawInput, options.versionOrInherited(ctx.Version)) - var info *api.OrchestrationContextInfo - if options.includeOrchestrationIdentity { - info = &api.OrchestrationContextInfo{ - InstanceID: ctx.ID, - Name: ctx.Name, - Version: ctx.Version, - ParentInstanceID: ctx.parentInstanceID, - } - } scheduleTaskAction.GetScheduleTask().Tags = contextprop.Encode( - info, ctx.contextFields, mergeStringMaps(ctx.orchestrationTags, options.tags)) + ctx.contextFields, mergeStringMaps(ctx.orchestrationTags, options.tags)) ctx.pendingActions[scheduleTaskAction.Id] = scheduleTaskAction @@ -788,7 +779,6 @@ func (ctx *OrchestrationContext) internalCallSubOrchestrator( options.versionOrDefault(ctx.defaultVersion), ) createSubOrchestrationAction.GetCreateSubOrchestration().Tags = contextprop.Encode( - nil, 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,7 +1769,7 @@ func (ctx *OrchestrationContext) setCompleteInternal( completed := completedAction.GetCompleteOrchestration() if status == protos.OrchestrationStatus_ORCHESTRATION_STATUS_CONTINUED_AS_NEW { completed.NewVersion = ctx.continuedAsNewVersion - completed.Tags = contextprop.Encode(nil, 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 index 005946ee..8d1549b3 100644 --- a/task/tag_identity_bench_test.go +++ b/task/tag_identity_bench_test.go @@ -13,15 +13,17 @@ import ( "google.golang.org/protobuf/types/known/wrapperspb" ) -func activityIdentityWireMessages(t testing.TB, includeIdentity bool) []proto.Message { +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) { - options := []CallActivityOption{WithRawActivityInput(`"input"`)} - if includeIdentity { - options = append(options, WithActivityOrchestrationIdentity()) - } - ctx.CallActivity("inspect", options...) + ctx.CallActivity("inspect", WithRawActivityInput(`"input"`)) return nil, nil })) started := helpers.NewExecutionStartedEvent("parent", "instance", nil, @@ -30,11 +32,7 @@ func activityIdentityWireMessages(t testing.TB, includeIdentity bool) []proto.Me []*protos.HistoryEvent{started}, nil) require.NoError(t, err) action := scheduledActivityAction(t, result.Response).GetScheduleTask() - if includeIdentity { - require.Equal(t, legacyIdentityTags(), action.Tags, "opt-in must reproduce the legacy identity payload") - } else { - require.Nil(t, action.Tags) - } + 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) @@ -45,29 +43,49 @@ func activityIdentityWireMessages(t testing.TB, includeIdentity bool) []proto.Me OrchestrationInstance: &protos.OrchestrationInstance{InstanceId: "instance"}, ParentTraceContext: trace, Tags: action.Tags, } - return []proto.Message{action, event, request} + 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) { - defaultMessages := activityIdentityWireMessages(t, false) - legacyMessages := activityIdentityWireMessages(t, true) - for index, name := range []string{"action", "history", "request"} { - defaultBytes, err := proto.MarshalOptions{Deterministic: true}.Marshal(defaultMessages[index]) + for _, test := range activityIdentityWireMessages(t) { + currentBytes, err := proto.MarshalOptions{Deterministic: true}.Marshal(test.current) require.NoError(t, err) - legacyBytes, err := proto.MarshalOptions{Deterministic: true}.Marshal(legacyMessages[index]) + previousBytes, err := proto.MarshalOptions{Deterministic: true}.Marshal(test.previous) require.NoError(t, err) - require.Less(t, len(defaultBytes), len(legacyBytes)) - t.Logf("%s: default=%d legacy/opt-in=%d delta=%d protobuf bytes", - name, len(defaultBytes), len(legacyBytes), len(legacyBytes)-len(defaultBytes)) + 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 _, includeIdentity := range []bool{false, true} { - messages := activityIdentityWireMessages(b, includeIdentity) - for index, name := range []string{"action", "history", "request"} { - b.Run(name+"/identity="+boolName(includeIdentity), func(b *testing.B) { - message := messages[index] + 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) diff --git a/task/tag_identity_test.go b/task/tag_identity_test.go index 8a78f599..e6ab67c6 100644 --- a/task/tag_identity_test.go +++ b/task/tag_identity_test.go @@ -28,77 +28,66 @@ func scheduledActivityAction(t testing.TB, response *protos.OrchestratorResponse return nil } -func TestScheduledActivityIdentityIsOptIn(t *testing.T) { +func TestActivityIdentityUsesWorkItem(t *testing.T) { for _, version := range []string{"", "v1"} { - for _, includeIdentity := range []bool{false, true} { - for _, withCallerData := range []bool{false, true} { - t.Run(version+"/identity="+boolName(includeIdentity)+"/caller="+boolName(withCallerData), func(t *testing.T) { - registry := NewTaskRegistry() - options := []CallActivityOption{ - WithActivityVersion("a2"), - WithRawActivityInput(`"input"`), - } - if includeIdentity { - options = append(options, WithActivityOrchestrationIdentity()) - } - 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(nil, 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()) - info, persistedFields := contextprop.Decode(scheduled.Tags) - wantInfo := api.OrchestrationContextInfo{InstanceID: "instance"} - if includeIdentity { - wantInfo.Name, wantInfo.Version, wantInfo.ParentInstanceID = "parent", version, "root-instance" - require.Equal(t, wantInfo, info) - } else { - require.Equal(t, api.OrchestrationContextInfo{}, info) - 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()) - }) - } + 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()) + }) } } } @@ -110,17 +99,7 @@ func boolName(value bool) string { return "no" } -func legacyIdentityTags() map[string]string { - return 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", - } -} - -func TestLegacyActivityTagsReplayWithNewWriter(t *testing.T) { +func TestActivityTagsReplay(t *testing.T) { registry := NewTaskRegistry() require.NoError(t, registry.AddOrchestratorNVersion("parent", "v1", func(ctx *OrchestrationContext) (any, error) { var output string @@ -130,15 +109,14 @@ func TestLegacyActivityTagsReplayWithNewWriter(t *testing.T) { canceled, cancel := ctx.WithCancel() cancel() require.ErrorIs(t, canceled.WaitForSingleEvent("never", -1).Await(nil), ErrTaskCanceled) - canceled.CallActivity("never", WithActivityOrchestrationIdentity()) + 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 = legacyIdentityTags() - started.GetExecutionStarted().Tags["user"] = "retained" - started.GetExecutionStarted().Tags[tagcodec.ContextFieldPrefix+"tenant"] = "retained" + 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 = legacyIdentityTags() + 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) @@ -156,72 +134,62 @@ func TestLegacyActivityTagsReplayWithNewWriter(t *testing.T) { }, next.GetScheduleTask().Tags) } -func TestActivityIdentityChoiceSurvivesRetryAndRedelivery(t *testing.T) { - for _, includeIdentity := range []bool{false, true} { - t.Run(boolName(includeIdentity), func(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"}), - } - if includeIdentity { - options = append(options, WithActivityOrchestrationIdentity()) - } - 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(nil, 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"} - if includeIdentity { - want.Name, want.Version, want.ParentInstanceID = "parent", "v1", "root-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 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) { @@ -249,10 +217,8 @@ func TestSubOrchestrationAndContinueAsNewUseNativeIdentity(t *testing.T) { return state{info, api.ContextFieldsFromContext(ctx.Context())}, nil })) started := helpers.NewExecutionStartedEvent("parent", "instance", nil, nil, nil, nil, wrapperspb.String("v1")) - started.GetExecutionStarted().Tags = legacyIdentityTags() - started.GetExecutionStarted().Tags["scope"] = "parent" - started.GetExecutionStarted().Tags["user"] = "retained" - started.GetExecutionStarted().Tags[tagcodec.ContextFieldPrefix+"tenant"] = "parent" + 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 { @@ -295,7 +261,7 @@ func TestSubOrchestrationAndContinueAsNewUseNativeIdentity(t *testing.T) { }, completionAction(t, next).Tags) } -func TestLegacySubOrchestrationTagsReplayAndRetry(t *testing.T) { +func TestSubOrchestrationTagsReplayAndRetry(t *testing.T) { registry := NewTaskRegistry() require.NoError(t, registry.AddOrchestratorNVersion("parent", "v1", func(ctx *OrchestrationContext) (any, error) { var output string @@ -311,11 +277,12 @@ func TestLegacySubOrchestrationTagsReplayAndRetry(t *testing.T) { 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(nil, api.ContextFields{"tenant": "parent"}, + 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 = legacyIdentityTags() + 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"}, @@ -367,7 +334,7 @@ func TestUntaggedLifecycleActionsHaveNoIdentityTags(t *testing.T) { return nil, nil })) started := helpers.NewExecutionStartedEvent("parent", "instance", nil, nil, nil, nil) - for _, tags := range []map[string]string{nil, {}, legacyIdentityTags()} { + 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) @@ -386,24 +353,23 @@ func TestUntaggedLifecycleActionsHaveNoIdentityTags(t *testing.T) { } } -func TestActivityExplicitContextOverridesHistoricalIdentity(t *testing.T) { +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: "v1", ParentInstanceID: "root-instance", + 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", + InstanceID: "explicit-instance", Name: "explicit-name", Version: "explicit-version", ParentInstanceID: "explicit-parent", }) event := helpers.NewTaskScheduledEvent(4, "inspect", nil, nil, nil) - event.GetTaskScheduled().Tags = legacyIdentityTags() - event.GetTaskScheduled().Tags[tagcodec.ContextFieldPrefix+"tenant"] = "durable" - event.GetTaskScheduled().Tags["user"] = "not a context field" + 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)