From 2019fdecd928271af79a28d9063f8458ed0eda23 Mon Sep 17 00:00:00 2001 From: JH Date: Mon, 24 Aug 2026 19:44:58 -0700 Subject: [PATCH] Add resolveNamespaceID Given a field somewhere in a replication message, resolveNamespaceID walks up the parent chain and returns the namespace id of the nearest message that owns a namespace. Nothing calls it yet. Only four message types count as owners, matched by a type switch. Other messages carry a NamespaceId that names a different namespace, so matching on the field name would pick the wrong one. The clearest example is StartChildWorkflowExecutionInitiatedEventAttributes, which holds the child's NamespaceId right beside the parent's SearchAttributes. Walking up rather than remembering the last NamespaceId seen on the way down keeps the answer independent of traversal order, which the visit library does not define. Co-Authored-By: Claude Opus 5 --- interceptor/reflection.go | 43 +++++++++ interceptor/resolve_namespace_test.go | 120 ++++++++++++++++++++++++++ 2 files changed, 163 insertions(+) create mode 100644 interceptor/resolve_namespace_test.go diff --git a/interceptor/reflection.go b/interceptor/reflection.go index 9f836fa4..d0eda81d 100644 --- a/interceptor/reflection.go +++ b/interceptor/reflection.go @@ -10,6 +10,8 @@ import ( "go.temporal.io/api/history/v1" "go.temporal.io/api/namespace/v1" "go.temporal.io/api/workflowservice/v1" + persistencespb "go.temporal.io/server/api/persistence/v1" + replicationspb "go.temporal.io/server/api/replication/v1" "go.temporal.io/server/common/log" "go.temporal.io/server/common/log/tag" "go.temporal.io/server/common/persistence/serialization" @@ -137,6 +139,47 @@ type stringMatcher func(name string) (string, bool) // It returns whether anything was matched and any error it encountered. type visitor func(logger log.Logger, obj any, match stringMatcher) (bool, error) +// resolveNamespaceID walks UP from vwp to the nearest enclosing message that owns a namespace, and +// returns that message's NamespaceId. +// +// Only these four message types count as owners. Other messages carry a NamespaceId naming a +// DIFFERENT namespace, so matching on the field name would pick the wrong one. The clearest example +// is history.StartChildWorkflowExecutionInitiatedEventAttributes, which holds the *child's* +// NamespaceId right beside the parent's SearchAttributes. +// +// Walking upward, rather than remembering the last NamespaceId seen on the way down, keeps the +// answer independent of traversal order. Order is unspecified: visit.ValuesUnsafe pops the front of +// its worklist but swaps in the last element, so it is neither breadth first nor depth first. +// +// fallback is returned when the walk reaches no owner. It exists to carry the namespace across the +// two boundaries the parent chain cannot cross: a data blob, whose events are visited in a fresh +// traversal, and a response whose namespace is only on the request it was paired with. +func resolveNamespaceID(vwp visit.ValueWithParent, fallback string) string { + for p := vwp.Parent; p != nil; p = p.Parent { + // Protobuf messages are always reached through a pointer, and the walk queues the + // dereferenced struct, so an owner is always addressable here. If that ever stops being + // true the owner is skipped and the fallback is returned, which the caller can see. + if p.Kind() != reflect.Struct || !p.CanAddr() { + continue + } + var nsID string + switch owner := p.Addr().Interface().(type) { + case *replicationspb.HistoryTaskAttributes: + nsID = owner.NamespaceId + case *replicationspb.BackfillHistoryTaskAttributes: + nsID = owner.NamespaceId + case *replicationspb.SyncVersionedTransitionTaskAttributes: + nsID = owner.NamespaceId + case *persistencespb.WorkflowExecutionInfo: + nsID = owner.NamespaceId + } + if nsID != "" { + return nsID + } + } + return fallback +} + // visitNamespace uses reflection to recursively visit all fields // in the given object. When it finds namespace string fields, it invokes // the provided match function. diff --git a/interceptor/resolve_namespace_test.go b/interceptor/resolve_namespace_test.go new file mode 100644 index 00000000..9976a541 --- /dev/null +++ b/interceptor/resolve_namespace_test.go @@ -0,0 +1,120 @@ +package interceptor + +import ( + "reflect" + "testing" + + "github.com/keilerkonzept/visit" + "github.com/stretchr/testify/require" + "go.temporal.io/api/common/v1" + "go.temporal.io/api/history/v1" + persistencespb "go.temporal.io/server/api/persistence/v1" + replicationspb "go.temporal.io/server/api/replication/v1" +) + +// resolveAt walks obj and returns what resolveNamespaceID answers at every field named fieldName. +// Walking for real is the point: it exercises the same parent chain the translator sees. +func resolveAt(t *testing.T, obj any, fieldName, fallback string) []string { + t.Helper() + var got []string + err := visit.Values(obj, func(vwp visit.ValueWithParent) (visit.Action, error) { + if vwp.Kind() == reflect.Ptr && vwp.IsNil() { + return visit.Skip, nil + } + fieldType, action := getParentFieldType(vwp) + if action != "" { + return action, nil + } + if fieldType.Name == fieldName { + got = append(got, resolveNamespaceID(vwp, fallback)) + } + return visit.Continue, nil + }) + require.NoError(t, err) + return got +} + +func testSearchAttributes() *common.SearchAttributes { + return &common.SearchAttributes{ + IndexedFields: map[string]*common.Payload{"TestSA": {Data: []byte("v")}}, + } +} + +func TestResolveNamespaceIDOneHop(t *testing.T) { + // The owner is the immediate parent. This is the common case. + obj := &persistencespb.WorkflowExecutionInfo{ + NamespaceId: "ns-a", + SearchAttributes: map[string]*common.Payload{"TestSA": {Data: []byte("v")}}, + } + require.Equal(t, []string{"ns-a"}, resolveAt(t, obj, "SearchAttributes", "unused")) +} + +func TestResolveNamespaceIDMultipleHops(t *testing.T) { + // The blob sits two levels below its owner, so checking only the immediate parent is not + // enough: VersionedTransitionArtifact has no NamespaceId of its own. + obj := &replicationspb.SyncVersionedTransitionTaskAttributes{ + NamespaceId: "ns-a", + VersionedTransitionArtifact: &replicationspb.VersionedTransitionArtifact{ + EventBatches: []*common.DataBlob{{Data: []byte("x")}}, + NewRunInfo: &replicationspb.NewRunInfo{ + EventBatch: &common.DataBlob{Data: []byte("y")}, + }, + }, + } + require.Equal(t, []string{"ns-a"}, resolveAt(t, obj, "EventBatches", "unused")) + // NewRunInfo puts a blob one level deeper still. + require.Equal(t, []string{"ns-a"}, resolveAt(t, obj, "EventBatch", "unused")) +} + +func TestResolveNamespaceIDIgnoresChildNamespaceID(t *testing.T) { + // StartChildWorkflowExecutionInitiatedEventAttributes holds the CHILD's NamespaceId right next + // to the PARENT's SearchAttributes. It is not an owner, so the walk must step over it. This + // test fails if anyone swaps the type switch for a NamespaceId field name match. + obj := &history.HistoryEvent{ + Attributes: &history.HistoryEvent_StartChildWorkflowExecutionInitiatedEventAttributes{ + StartChildWorkflowExecutionInitiatedEventAttributes: &history.StartChildWorkflowExecutionInitiatedEventAttributes{ + NamespaceId: "ns-child", + SearchAttributes: testSearchAttributes(), + }, + }, + } + require.Equal(t, []string{"ns-parent"}, resolveAt(t, obj, "SearchAttributes", "ns-parent")) +} + +func TestResolveNamespaceIDIgnoresParentNamespaceID(t *testing.T) { + // WorkflowExecutionInfo has both. Only NamespaceId is the workflow's own. + obj := &persistencespb.WorkflowExecutionInfo{ + NamespaceId: "ns-a", + ParentNamespaceId: "ns-parent", + SearchAttributes: map[string]*common.Payload{"TestSA": {Data: []byte("v")}}, + } + require.Equal(t, []string{"ns-a"}, resolveAt(t, obj, "SearchAttributes", "unused")) +} + +func TestResolveNamespaceIDEmptyOwnerKeepsWalking(t *testing.T) { + // An owner whose NamespaceId is empty tells us nothing, so the walk carries on outward. + obj := &replicationspb.SyncVersionedTransitionTaskAttributes{ + NamespaceId: "ns-a", + VersionedTransitionArtifact: &replicationspb.VersionedTransitionArtifact{ + StateAttributes: &replicationspb.VersionedTransitionArtifact_SyncWorkflowStateSnapshotAttributes{ + SyncWorkflowStateSnapshotAttributes: &replicationspb.SyncWorkflowStateSnapshotAttributes{ + State: &persistencespb.WorkflowMutableState{ + ExecutionInfo: &persistencespb.WorkflowExecutionInfo{ + NamespaceId: "", + SearchAttributes: map[string]*common.Payload{"TestSA": {Data: []byte("v")}}, + }, + }, + }, + }, + }, + } + require.Equal(t, []string{"ns-a"}, resolveAt(t, obj, "SearchAttributes", "unused")) +} + +func TestResolveNamespaceIDFallsBackWhenNoOwner(t *testing.T) { + // Nothing above the attribute owns a namespace, which is what happens inside a data blob and + // on the raw history responses. + obj := &history.WorkflowExecutionStartedEventAttributes{SearchAttributes: testSearchAttributes()} + require.Equal(t, []string{"ns-fallback"}, resolveAt(t, obj, "SearchAttributes", "ns-fallback")) + require.Equal(t, []string{""}, resolveAt(t, obj, "SearchAttributes", "")) +}