Add resolveNamespaceID

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.

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", ""))
+}