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