Validate history pagination branch token against mutable state (#11723)

## What changed? `GetWorkflowExecutionHistory` and `GetWorkflowExecutionHistoryReverse` now check `branch_token` in the page token against the token in mutable state. ## Why? To confirm if it is still the correct branch after conflict resolution. ## How did you test it? - [x] built - [x] run locally and tested manually - [x] covered by existing tests - [x] added new unit test(s) - [x] added new functional test(s) (cherry picked from commit ceb1cc1071)
2026-08-31 02:51:51 -07:00 · 2026-08-21 13:41:54 -07:00
parent 4abfeecde6
commit c26eb9faca
10 changed files with 638 additions and 3 deletions
--- a/common/dynamicconfig/constants.go
+++ b/common/dynamicconfig/constants.go
@@ -1714,6 +1714,20 @@ leaves the membership ring, giving in-flight long-polls time to drain before the

 	// keys for history

+	EnablePaginationTokenBranchValidation = NewGlobalBoolSetting(
+		"history.enablePaginationTokenBranchValidation",
+		true,
+		`EnablePaginationTokenBranchValidation enables checking that the branch token in a
+GetWorkflowExecutionHistory(Reverse) page token is the current branch token of the requested
+execution.`,
+	)
+	EnablePaginationTokenBranchValidationShadowMode = NewGlobalBoolSetting(
+		"history.enablePaginationTokenBranchValidationShadowMode",
+		false,
+		`EnablePaginationTokenBranchValidationShadowMode logs and emits metrics for a page token whose
+branch token is not the execution's current one, but still serves the read.`,
+	)
+
 	EnableReplicationStream = NewGlobalBoolSetting(
 		"history.enableReplicationStream",
 		true,
--- a/common/log/tag/tags.go
+++ b/common/log/tag/tags.go
@@ -259,6 +259,11 @@ func WorkflowBranchToken(branchToken []byte) ZapTag {
 	return NewBinaryTag("wf-branch-token", branchToken)
 }

+// WorkflowRequestBranchToken returns tag for a branch token supplied by the caller
+func WorkflowRequestBranchToken(branchToken []byte) ZapTag {
+	return NewBinaryTag("wf-request-branch-token", branchToken)
+}
+
 // WorkflowTreeID returns tag for WorkflowTreeID
 func WorkflowTreeID(treeID string) ZapTag {
 	return NewStringTag("wf-tree-id", treeID)
--- a/common/metrics/metric_defs.go
+++ b/common/metrics/metric_defs.go
@@ -1012,6 +1012,7 @@ var (
 	FailedWorkflowTasksCounter                    = NewCounterDef("failed_workflow_tasks")
 	WorkflowTaskAttempt                           = NewDimensionlessHistogramDef("workflow_task_attempt")
 	StaleMutableStateCounter                      = NewCounterDef("stale_mutable_state")
+	PaginationTokenBranchMismatchCounter          = NewCounterDef("pagination_token_branch_mismatch")
 	AutoResetPointsLimitExceededCounter           = NewCounterDef("auto_reset_points_exceed_limit")
 	AutoResetPointCorruptionCounter               = NewCounterDef("auto_reset_point_corruption")
 	BatchableTaskBatchCount                       = NewGaugeDef("batchable_task_batch_count")
--- a/service/history/api/getworkflowexecutionhistory/api.go
+++ b/service/history/api/getworkflowexecutionhistory/api.go
@@ -256,6 +256,19 @@ func Invoke(
 			continuationToken.FirstEventId = continuationToken.GetNextEventId()
 			continuationToken.NextEventId = nextEventID
 			continuationToken.IsWorkflowRunning = isWorkflowRunning
+		} else {
+			if err = api.ValidateBranchTokenForExecution(
+				ctx,
+				shardContext,
+				workflowConsistencyChecker,
+				eventNotifier,
+				namespaceName,
+				namespaceID,
+				execution,
+				continuationToken.BranchToken,
+			); err != nil {
+				return nil, err
+			}
 		}
 	} else {
 		continuationToken = &tokenspb.HistoryContinuation{}
--- a/service/history/api/getworkflowexecutionhistoryreverse/api.go
+++ b/service/history/api/getworkflowexecutionhistoryreverse/api.go
@@ -103,6 +103,23 @@ func Invoke(
 		}

 		execution.RunId = continuationToken.GetRunId()
+
+		var namespaceName namespace.Name
+		if entry, err := shardContext.GetNamespaceRegistry().GetNamespaceByID(namespaceID); err == nil {
+			namespaceName = entry.Name()
+		}
+		if err = api.ValidateBranchTokenForExecution(
+			ctx,
+			shardContext,
+			workflowConsistencyChecker,
+			eventNotifier,
+			namespaceName,
+			namespaceID,
+			execution,
+			continuationToken.BranchToken,
+		); err != nil {
+			return nil, err
+		}
 	}

 	// TODO below is a temporal solution to guard against invalid event batch
--- a/service/history/api/token.go
+++ b/service/history/api/token.go
@@ -1,10 +1,20 @@
 package api

 import (
+	"bytes"
+	"context"
+
+	commonpb "go.temporal.io/api/common/v1"
 	historyspb "go.temporal.io/server/api/history/v1"
 	"go.temporal.io/server/api/historyservice/v1"
 	tokenspb "go.temporal.io/server/api/token/v1"
+	"go.temporal.io/server/common/log/tag"
+	"go.temporal.io/server/common/metrics"
+	"go.temporal.io/server/common/namespace"
+	serviceerrors "go.temporal.io/server/common/serviceerror"
 	"go.temporal.io/server/service/history/consts"
+	"go.temporal.io/server/service/history/events"
+	historyi "go.temporal.io/server/service/history/interfaces"
 )

 // NOTE: DO NOT MODIFY UNLESS ALSO APPLIED TO ./service/frontend/token_deprecated.go
@@ -116,3 +126,118 @@ func ValidatePaginationToken(
 	}
 	return nil
 }
+
+const (
+	// branchTokenMismatchReasonNonCurrent is a branch this execution records but no longer reads from.
+	branchTokenMismatchReasonNonCurrent metrics.ReasonString = "non_current_branch"
+	branchTokenMismatchReasonForeign    metrics.ReasonString = "foreign_branch"
+)
+
+// maxLoggedBranchTokenLen bounds caller-supplied bytes reaching the log.
+const maxLoggedBranchTokenLen = 4096
+
+func branchTokenMismatchReason(
+	currentBranchToken []byte,
+	requestBranchToken []byte,
+	versionHistories *historyspb.VersionHistories,
+) metrics.ReasonString {
+	if bytes.Equal(requestBranchToken, currentBranchToken) {
+		return ""
+	}
+	for _, versionHistory := range versionHistories.GetHistories() {
+		if bytes.Equal(versionHistory.GetBranchToken(), requestBranchToken) {
+			return branchTokenMismatchReasonNonCurrent
+		}
+	}
+	return branchTokenMismatchReasonForeign
+}
+
+func reportBranchTokenMismatch(
+	shardContext historyi.ShardContext,
+	namespaceName string,
+	execution *commonpb.WorkflowExecution,
+	reason metrics.ReasonString,
+	currentBranchToken []byte,
+	requestBranchToken []byte,
+) {
+	if namespaceName == "" {
+		namespaceName = metrics.NamespaceUnknownTag().Value
+	}
+	metrics.PaginationTokenBranchMismatchCounter.With(shardContext.GetMetricsHandler()).Record(
+		1,
+		metrics.NamespaceTag(namespaceName),
+		metrics.ReasonTag(reason),
+	)
+	loggedRequestToken := requestBranchToken[:min(len(requestBranchToken), maxLoggedBranchTokenLen)]
+	shardContext.GetLogger().Warn("Pagination branch token is not the execution's current branch token",
+		tag.WorkflowNamespace(namespaceName),
+		tag.WorkflowID(execution.GetWorkflowId()),
+		tag.WorkflowRunID(execution.GetRunId()),
+		tag.NewStringTag("reason", string(reason)),
+		tag.WorkflowBranchToken(currentBranchToken),
+		tag.WorkflowRequestBranchToken(loggedRequestToken),
+	)
+}
+
+// ValidateBranchTokenForExecution rejects a paging branch token that is not the execution's current
+// one.
+func ValidateBranchTokenForExecution(
+	ctx context.Context,
+	shardContext historyi.ShardContext,
+	workflowConsistencyChecker WorkflowConsistencyChecker,
+	eventNotifier events.Notifier,
+	namespaceName namespace.Name,
+	namespaceID namespace.ID,
+	execution *commonpb.WorkflowExecution,
+	requestBranchToken []byte,
+) error {
+	config := shardContext.GetConfig()
+	if !config.EnablePaginationTokenBranchValidation() {
+		return nil
+	}
+	if len(requestBranchToken) == 0 {
+		return consts.ErrInvalidNextPageToken
+	}
+
+	response, err := GetOrPollWorkflowMutableState(
+		ctx,
+		shardContext,
+		&historyservice.GetMutableStateRequest{
+			NamespaceId: namespaceID.String(),
+			Execution:   execution,
+		},
+		workflowConsistencyChecker,
+		eventNotifier,
+	)
+	if err != nil {
+		return err
+	}
+
+	currentBranchToken := response.GetCurrentBranchToken()
+	mismatchReason := branchTokenMismatchReason(
+		currentBranchToken,
+		requestBranchToken,
+		response.GetVersionHistories(),
+	)
+	if mismatchReason == "" {
+		return nil
+	}
+
+	reportBranchTokenMismatch(
+		shardContext,
+		namespaceName.String(),
+		execution,
+		mismatchReason,
+		currentBranchToken,
+		requestBranchToken,
+	)
+	if config.EnablePaginationTokenBranchValidationShadowMode() {
+		return nil
+	}
+	return serviceerrors.NewCurrentBranchChanged(
+		currentBranchToken,
+		requestBranchToken,
+		nil,
+		nil,
+	)
+}
--- a/service/history/api/token_test.go
+++ b/service/history/api/token_test.go
@@ -0,0 +1,117 @@
+package api
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	historyspb "go.temporal.io/server/api/history/v1"
+	persistencespb "go.temporal.io/server/api/persistence/v1"
+	"go.temporal.io/server/common/dynamicconfig"
+	"go.temporal.io/server/common/persistence"
+	"go.temporal.io/server/common/persistence/serialization"
+	"go.temporal.io/server/common/primitives"
+	"go.temporal.io/server/service/history/configs"
+	"go.temporal.io/server/service/history/consts"
+	historyi "go.temporal.io/server/service/history/interfaces"
+	"go.uber.org/mock/gomock"
+)
+
+func newTestBranchToken(
+	t *testing.T,
+	treeID string,
+	branchID string,
+	ancestors []*persistencespb.HistoryBranchRange,
+) []byte {
+	t.Helper()
+	branchUtil := persistence.NewHistoryBranchUtil(serialization.NewSerializer())
+	token, err := branchUtil.NewHistoryBranch(
+		"namespace-id", "workflow-id", "run-id", treeID, &branchID, ancestors, 0, 0, 0,
+	)
+	require.NoError(t, err)
+	return token
+}
+
+func testVersionHistories(tokens ...[]byte) *historyspb.VersionHistories {
+	versionHistories := &historyspb.VersionHistories{}
+	for _, token := range tokens {
+		versionHistories.Histories = append(
+			versionHistories.Histories,
+			&historyspb.VersionHistory{BranchToken: token},
+		)
+	}
+	return versionHistories
+}
+
+func TestBranchTokenMismatchReason(t *testing.T) {
+	treeID := primitives.NewUUID().String()
+	branchID := primitives.NewUUID().String()
+	otherTreeID := primitives.NewUUID().String()
+	otherBranchID := primitives.NewUUID().String()
+
+	current := newTestBranchToken(t, treeID, branchID, nil)
+	nonCurrent := newTestBranchToken(t, otherTreeID, otherBranchID, nil)
+
+	t.Run("matches the current branch token", func(t *testing.T) {
+		got := branchTokenMismatchReason(current, current, testVersionHistories(current))
+		require.Empty(t, got)
+	})
+
+	t.Run("matches the current token when an identical token is recorded twice", func(t *testing.T) {
+		got := branchTokenMismatchReason(current, current, testVersionHistories(current, current))
+		require.Empty(t, got)
+	})
+
+	t.Run("matches opaque tokens the branch parser cannot read", func(t *testing.T) {
+		opaque := []byte{1, 2, 3}
+		got := branchTokenMismatchReason(opaque, opaque, testVersionHistories(opaque))
+		require.Empty(t, got)
+	})
+
+	t.Run("reports a non-current branch when the token names an older history", func(t *testing.T) {
+		histories := testVersionHistories(nonCurrent, current)
+		got := branchTokenMismatchReason(current, nonCurrent, histories)
+		require.Equal(t, branchTokenMismatchReasonNonCurrent, got)
+	})
+
+	t.Run("reports foreign for a different branch in the same tree", func(t *testing.T) {
+		request := newTestBranchToken(t, treeID, otherBranchID, nil)
+		got := branchTokenMismatchReason(current, request, testVersionHistories(current))
+		require.Equal(t, branchTokenMismatchReasonForeign, got)
+	})
+
+	t.Run("reports foreign for the current branch carrying injected ancestors", func(t *testing.T) {
+		request := newTestBranchToken(t, treeID, branchID, []*persistencespb.HistoryBranchRange{
+			{BranchId: otherBranchID, BeginNodeId: 1, EndNodeId: 1000},
+		})
+		got := branchTokenMismatchReason(current, request, testVersionHistories(current))
+		require.Equal(t, branchTokenMismatchReasonForeign, got)
+	})
+
+	t.Run("reports foreign when there are no version histories", func(t *testing.T) {
+		got := branchTokenMismatchReason(current, nonCurrent, nil)
+		require.Equal(t, branchTokenMismatchReasonForeign, got)
+	})
+}
+
+func TestValidateBranchTokenForExecution_EmptyRequestToken(t *testing.T) {
+	for _, tc := range []struct {
+		name       string
+		validation bool
+		wantErr    error
+	}{
+		{name: "rejected while validating", validation: true, wantErr: consts.ErrInvalidNextPageToken},
+		{name: "served once validation is disabled", validation: false},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			shardContext := historyi.NewMockShardContext(gomock.NewController(t))
+			shardContext.EXPECT().GetConfig().Return(&configs.Config{
+				EnablePaginationTokenBranchValidation: dynamicconfig.GetBoolPropertyFn(tc.validation),
+			}).AnyTimes()
+
+			err := ValidateBranchTokenForExecution(
+				context.Background(), shardContext, nil, nil, "", "", nil, nil)
+			require.ErrorIs(t, err, tc.wantErr)
+		})
+	}
+}
--- a/service/history/configs/config.go
+++ b/service/history/configs/config.go
@@ -413,6 +413,9 @@ type Config struct {
 	SendRawHistoryBytesToMatchingService  dynamicconfig.BoolPropertyFn
 	SendRawWorkflowHistory                dynamicconfig.BoolPropertyFnWithNamespaceFilter

+	EnablePaginationTokenBranchValidation           dynamicconfig.BoolPropertyFn
+	EnablePaginationTokenBranchValidationShadowMode dynamicconfig.BoolPropertyFn
+
 	WorkflowIdReuseMinimalInterval           dynamicconfig.DurationPropertyFnWithNamespaceFilter
 	EnableWorkflowIdReuseStartTimeValidation dynamicconfig.BoolPropertyFnWithNamespaceFilter
 	BusinessIDReuseRate                      dynamicconfig.IntPropertyFnWithNamespaceFilter
@@ -805,9 +808,13 @@ func NewConfig(
 		EnableUpdateWithStartRetryOnClosedWorkflowAbort:               dynamicconfig.EnableUpdateWithStartRetryOnClosedWorkflowAbort.Get(dc),
 		EnableUpdateWithStartRetryableErrorOnClosedWorkflowAbort:      dynamicconfig.EnableUpdateWithStartRetryableErrorOnClosedWorkflowAbort.Get(dc),

-		SendRawHistoryBetweenInternalServices:    dynamicconfig.SendRawHistoryBetweenInternalServices.Get(dc),
-		SendRawHistoryBytesToMatchingService:     dynamicconfig.SendRawHistoryBytesToMatchingService.Get(dc),
-		SendRawWorkflowHistory:                   dynamicconfig.SendRawWorkflowHistory.Get(dc),
+		SendRawHistoryBetweenInternalServices: dynamicconfig.SendRawHistoryBetweenInternalServices.Get(dc),
+		SendRawHistoryBytesToMatchingService:  dynamicconfig.SendRawHistoryBytesToMatchingService.Get(dc),
+		SendRawWorkflowHistory:                dynamicconfig.SendRawWorkflowHistory.Get(dc),
+
+		EnablePaginationTokenBranchValidation:           dynamicconfig.EnablePaginationTokenBranchValidation.Get(dc),
+		EnablePaginationTokenBranchValidationShadowMode: dynamicconfig.EnablePaginationTokenBranchValidationShadowMode.Get(dc),
+
 		WorkflowIdReuseMinimalInterval:           dynamicconfig.WorkflowIdReuseMinimalInterval.Get(dc),
 		EnableWorkflowIdReuseStartTimeValidation: dynamicconfig.EnableWorkflowIdReuseStartTimeValidation.Get(dc),
 		BusinessIDReuseRate:                      dynamicconfig.BusinessIDReuseRate.Get(dc),
--- a/service/history/history_engine_test.go
+++ b/service/history/history_engine_test.go
@@ -49,12 +49,14 @@ import (
 	"go.temporal.io/server/common/payload"
 	"go.temporal.io/server/common/payloads"
 	"go.temporal.io/server/common/persistence"
+	"go.temporal.io/server/common/persistence/serialization"
 	"go.temporal.io/server/common/persistence/versionhistory"
 	"go.temporal.io/server/common/persistence/visibility/manager"
 	"go.temporal.io/server/common/primitives/timestamp"
 	"go.temporal.io/server/common/rpc/interceptor"
 	"go.temporal.io/server/common/searchattribute"
 	"go.temporal.io/server/common/searchattribute/sadefs"
+	serviceerrors "go.temporal.io/server/common/serviceerror"
 	"go.temporal.io/server/common/tasktoken"
 	"go.temporal.io/server/common/testing/protorequire"
 	"go.temporal.io/server/service/history/api"
@@ -7028,3 +7030,208 @@ func addFailWorkflowEvent(
 	)
 	return event
 }
+
+func (s *engineSuite) mockExecutionWithForeignBranchToken(
+	we *commonpb.WorkflowExecution,
+) (ownedToken []byte, foreignToken []byte) {
+	branchUtil := persistence.NewHistoryBranchUtil(serialization.NewSerializer())
+	treeID := uuid.NewString()
+	ownedBranchID := uuid.NewString()
+	foreignBranchID := uuid.NewString()
+
+	ownedBranchToken, err := branchUtil.NewHistoryBranch(
+		tests.NamespaceID.String(), we.WorkflowId, we.RunId, treeID, &ownedBranchID, nil, 0, 0, 0)
+	s.NoError(err)
+	foreignBranchToken, err := branchUtil.NewHistoryBranch(
+		tests.NamespaceID.String(), we.WorkflowId, we.RunId, treeID, &foreignBranchID, nil, 0, 0, 0)
+	s.NoError(err)
+
+	s.mockExecutionMgr.EXPECT().GetWorkflowExecution(gomock.Any(), gomock.Any()).Return(&persistence.GetWorkflowExecutionResponse{
+		State: &persistencespb.WorkflowMutableState{
+			ExecutionState: &persistencespb.WorkflowExecutionState{
+				State:  enumsspb.WORKFLOW_EXECUTION_STATE_RUNNING,
+				Status: enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING,
+				RunId:  we.RunId,
+			},
+			NextEventId: 5,
+			ExecutionInfo: &persistencespb.WorkflowExecutionInfo{
+				NamespaceId: tests.NamespaceID.String(),
+				WorkflowId:  we.WorkflowId,
+				VersionHistories: &historyspb.VersionHistories{
+					CurrentVersionHistoryIndex: 0,
+					Histories: []*historyspb.VersionHistory{
+						{
+							BranchToken: ownedBranchToken,
+							Items: []*historyspb.VersionHistoryItem{
+								{EventId: 4, Version: 0},
+							},
+						},
+					},
+				},
+			},
+		},
+		MutableStateStats: persistence.MutableStateStatistics{},
+	}, nil).AnyTimes()
+
+	return ownedBranchToken, foreignBranchToken
+}
+
+func (s *engineSuite) getHistoryRequestWithPageToken(
+	we *commonpb.WorkflowExecution,
+	continuation *tokenspb.HistoryContinuation,
+	waitNewEvent bool,
+) *historyservice.GetWorkflowExecutionHistoryRequest {
+	nextPageToken, err := api.SerializeHistoryToken(continuation)
+	s.NoError(err)
+	return &historyservice.GetWorkflowExecutionHistoryRequest{
+		NamespaceId: tests.NamespaceID.String(),
+		Request: &workflowservice.GetWorkflowExecutionHistoryRequest{
+			Execution:              we,
+			MaximumPageSize:        10,
+			NextPageToken:          nextPageToken,
+			WaitNewEvent:           waitNewEvent,
+			HistoryEventFilterType: enumspb.HISTORY_EVENT_FILTER_TYPE_ALL_EVENT,
+			SkipArchival:           true,
+		},
+	}
+}
+
+// expectHistoryReadWithBranchToken allows the history read and pins the branch token it is issued with.
+func (s *engineSuite) expectHistoryReadWithBranchToken(branchToken []byte) {
+	s.mockSearchAttributesProvider.EXPECT().GetSearchAttributes(gomock.Any(), false).
+		Return(searchattribute.TestNameTypeMap(), nil).AnyTimes()
+	s.mockVisibilityMgr.EXPECT().GetIndexName().Return(esIndexName).AnyTimes()
+	s.mockExecutionMgr.EXPECT().ReadHistoryBranch(gomock.Any(), gomock.Any()).DoAndReturn(
+		func(_ context.Context, request *persistence.ReadHistoryBranchRequest) (*persistence.ReadHistoryBranchResponse, error) {
+			s.Equal(branchToken, request.BranchToken)
+			return &persistence.ReadHistoryBranchResponse{HistoryEvents: []*historypb.HistoryEvent{}}, nil
+		},
+	).MinTimes(1)
+}
+
+func (s *engineSuite) TestGetWorkflowExecutionHistory_BranchTokenNotOwnedByExecution() {
+	we := commonpb.WorkflowExecution{WorkflowId: "wid-foreign-branch", RunId: uuid.NewString()}
+
+	engine, err := s.historyEngine.shardContext.GetEngine(context.Background())
+	s.NoError(err)
+
+	s.config.EnablePaginationTokenBranchValidationShadowMode = dynamicconfig.GetBoolPropertyFn(false)
+	_, foreignBranchToken := s.mockExecutionWithForeignBranchToken(&we)
+	req := s.getHistoryRequestWithPageToken(&we, &tokenspb.HistoryContinuation{
+		RunId:             we.GetRunId(),
+		FirstEventId:      common.FirstEventID,
+		NextEventId:       5,
+		PersistenceToken:  []byte("some random persistence token"),
+		BranchToken:       foreignBranchToken,
+		IsWorkflowRunning: true,
+	}, false)
+
+	// The history read must never be attempted, on either the decoded or the raw path.
+	s.mockExecutionMgr.EXPECT().ReadHistoryBranch(gomock.Any(), gomock.Any()).Times(0)
+	s.mockExecutionMgr.EXPECT().ReadRawHistoryBranch(gomock.Any(), gomock.Any()).Times(0)
+
+	for _, sendRawHistory := range []bool{false, true} {
+		s.config.SendRawWorkflowHistory = func(string) bool { return sendRawHistory }
+		_, err = engine.GetWorkflowExecutionHistory(context.Background(), req)
+		var branchErr *serviceerrors.CurrentBranchChanged
+		s.ErrorAs(err, &branchErr, "sendRawHistory=%v", sendRawHistory)
+	}
+}
+
+func (s *engineSuite) TestGetWorkflowExecutionHistory_ForeignBranchTokenServedWhenNotEnforcing() {
+	we := commonpb.WorkflowExecution{WorkflowId: "wid-foreign-branch-served", RunId: uuid.NewString()}
+
+	engine, err := s.historyEngine.shardContext.GetEngine(context.Background())
+	s.NoError(err)
+
+	s.config.SendRawWorkflowHistory = func(string) bool { return false }
+	_, foreignBranchToken := s.mockExecutionWithForeignBranchToken(&we)
+	// Not enforcing reads with the caller's token, exactly as before validation existed.
+	s.expectHistoryReadWithBranchToken(foreignBranchToken)
+
+	for _, tc := range []struct {
+		name       string
+		validation bool
+		shadow     bool
+	}{
+		{name: "shadow mode", validation: true, shadow: true},
+		{name: "validation disabled", validation: false, shadow: false},
+	} {
+		s.config.EnablePaginationTokenBranchValidation = dynamicconfig.GetBoolPropertyFn(tc.validation)
+		s.config.EnablePaginationTokenBranchValidationShadowMode = dynamicconfig.GetBoolPropertyFn(tc.shadow)
+		_, err = engine.GetWorkflowExecutionHistory(
+			context.Background(),
+			s.getHistoryRequestWithPageToken(&we, &tokenspb.HistoryContinuation{
+				RunId:             we.GetRunId(),
+				FirstEventId:      common.FirstEventID,
+				NextEventId:       5,
+				PersistenceToken:  []byte("some random persistence token"),
+				BranchToken:       foreignBranchToken,
+				IsWorkflowRunning: true,
+			}, false),
+		)
+		s.NoError(err, tc.name)
+	}
+}
+
+func (s *engineSuite) TestGetWorkflowExecutionHistory_LongPollDiscardsRequestBranchToken() {
+	we := commonpb.WorkflowExecution{WorkflowId: "wid-longpoll-overwrite", RunId: uuid.NewString()}
+
+	engine, err := s.historyEngine.shardContext.GetEngine(context.Background())
+	s.NoError(err)
+
+	s.config.EnablePaginationTokenBranchValidationShadowMode = dynamicconfig.GetBoolPropertyFn(false)
+	s.config.SendRawWorkflowHistory = func(string) bool { return false }
+	ownedBranchToken, foreignBranchToken := s.mockExecutionWithForeignBranchToken(&we)
+	s.expectHistoryReadWithBranchToken(ownedBranchToken)
+
+	// An empty persistence token and a next event ID behind mutable state select the long poll
+	// refresh, which replaces the caller's branch token instead of validating it.
+	req := s.getHistoryRequestWithPageToken(&we, &tokenspb.HistoryContinuation{
+		RunId:             we.GetRunId(),
+		FirstEventId:      common.FirstEventID,
+		NextEventId:       2,
+		PersistenceToken:  nil,
+		BranchToken:       foreignBranchToken,
+		IsWorkflowRunning: true,
+	}, true)
+
+	_, err = engine.GetWorkflowExecutionHistory(context.Background(), req)
+	s.NoError(err)
+}
+
+func (s *engineSuite) TestGetWorkflowExecutionHistoryReverse_BranchTokenNotOwnedByExecution() {
+	we := commonpb.WorkflowExecution{WorkflowId: "wid-foreign-branch-reverse", RunId: uuid.NewString()}
+
+	engine, err := s.historyEngine.shardContext.GetEngine(context.Background())
+	s.NoError(err)
+
+	s.config.EnablePaginationTokenBranchValidationShadowMode = dynamicconfig.GetBoolPropertyFn(false)
+	_, foreignBranchToken := s.mockExecutionWithForeignBranchToken(&we)
+
+	nextPageToken, err := api.SerializeHistoryToken(&tokenspb.HistoryContinuation{
+		RunId:            we.GetRunId(),
+		FirstEventId:     common.FirstEventID,
+		NextEventId:      5,
+		PersistenceToken: []byte("some random persistence token"),
+		BranchToken:      foreignBranchToken,
+	})
+	s.NoError(err)
+
+	// The history read must never be attempted.
+	s.mockExecutionMgr.EXPECT().ReadHistoryBranchReverse(gomock.Any(), gomock.Any()).Times(0)
+
+	_, err = engine.GetWorkflowExecutionHistoryReverse(
+		context.Background(),
+		&historyservice.GetWorkflowExecutionHistoryReverseRequest{
+			NamespaceId: tests.NamespaceID.String(),
+			Request: &workflowservice.GetWorkflowExecutionHistoryReverseRequest{
+				Execution:       &we,
+				MaximumPageSize: 10,
+				NextPageToken:   nextPageToken,
+			},
+		},
+	)
+	var branchErr *serviceerrors.CurrentBranchChanged
+	s.ErrorAs(err, &branchErr)
+}
--- a/tests/gethistory_test.go
+++ b/tests/gethistory_test.go
@@ -2,21 +2,25 @@ package tests

 import (
 	"context"
+	"errors"
 	"slices"
 	"testing"
 	"time"

 	"github.com/google/uuid"
+	"github.com/stretchr/testify/require"
 	commandpb "go.temporal.io/api/command/v1"
 	commonpb "go.temporal.io/api/common/v1"
 	enumspb "go.temporal.io/api/enums/v1"
 	historypb "go.temporal.io/api/history/v1"
+	"go.temporal.io/api/serviceerror"
 	"go.temporal.io/api/workflowservice/v1"
 	sdkclient "go.temporal.io/sdk/client"
 	"go.temporal.io/sdk/temporal"
 	"go.temporal.io/sdk/workflow"
 	"go.temporal.io/server/common/dynamicconfig"
 	"go.temporal.io/server/common/log/tag"
+	"go.temporal.io/server/common/namespace"
 	"go.temporal.io/server/common/payloads"
 	"go.temporal.io/server/common/persistence/serialization"
 	"go.temporal.io/server/common/testing/parallelsuite"
@@ -747,3 +751,128 @@ func (s *GetHistorySuite) getHistory(

 	return responseInner.History.Events, responseInner.NextPageToken
 }
+
+// startMultiBatchWorkflow starts a workflow and signals it repeatedly. Each signal is its own
+// transaction and therefore its own history node, so a page size of 1 yields many pages and a
+// continuation taken from an early page is far from the last page.
+func startMultiBatchWorkflow(
+	ctx context.Context,
+	assertions *require.Assertions,
+	env *testcore.TestEnv,
+) {
+	_, err := env.FrontendClient().StartWorkflowExecution(ctx, &workflowservice.StartWorkflowExecutionRequest{
+		RequestId:           uuid.NewString(),
+		Namespace:           env.Namespace().String(),
+		WorkflowId:          env.Tv().WorkflowID(),
+		WorkflowType:        env.Tv().WorkflowType(),
+		TaskQueue:           env.Tv().TaskQueue(),
+		WorkflowRunTimeout:  durationpb.New(100 * time.Second),
+		WorkflowTaskTimeout: durationpb.New(10 * time.Second),
+		Identity:            env.Tv().WorkerIdentity(),
+	})
+	assertions.NoError(err)
+
+	for range 6 {
+		_, err = env.FrontendClient().SignalWorkflowExecution(ctx, &workflowservice.SignalWorkflowExecutionRequest{
+			RequestId: uuid.NewString(),
+			Namespace: env.Namespace().String(),
+			WorkflowExecution: &commonpb.WorkflowExecution{
+				WorkflowId: env.Tv().WorkflowID(),
+			},
+			SignalName: "signal",
+			Identity:   env.Tv().WorkerIdentity(),
+		})
+		assertions.NoError(err)
+	}
+}
+
+func (s *GetHistorySuite) TestGetWorkflowExecutionHistory_ContinuationFromAnotherNamespace(
+	enableTransitionHistory bool,
+) {
+	// A single-shard cluster so both namespaces map to the same history shard. On SQL backends
+	// shard_id is part of the history_node primary key, so without this the replay would be
+	// rejected for the wrong reason; on Cassandra it is not part of the key at all.
+	env := s.newTestEnv(
+		enableTransitionHistory,
+		testcore.WithHistoryShardCount(1),
+		// Shadow mode reports the mismatch but still serves the page.
+		testcore.WithDynamicConfig(dynamicconfig.EnablePaginationTokenBranchValidationShadowMode, false),
+	)
+
+	otherNamespace := namespace.Name(testcore.RandomizeStr("other-namespace"))
+	_, err := env.RegisterNamespace(otherNamespace, 1, enumspb.ARCHIVAL_STATE_DISABLED, "", "")
+	s.Require().NoError(err)
+
+	startMultiBatchWorkflow(s.Context(), s.Require(), env)
+
+	// A genuine, non-final continuation obtained with legitimate access to the first namespace.
+	firstPage, err := env.FrontendClient().GetWorkflowExecutionHistory(
+		s.Context(),
+		&workflowservice.GetWorkflowExecutionHistoryRequest{
+			Namespace:              env.Namespace().String(),
+			Execution:              &commonpb.WorkflowExecution{WorkflowId: env.Tv().WorkflowID()},
+			MaximumPageSize:        1,
+			HistoryEventFilterType: enumspb.HISTORY_EVENT_FILTER_TYPE_ALL_EVENT,
+		},
+	)
+	s.Require().NoError(err)
+	s.Require().NotEmpty(firstPage.NextPageToken, "need a non-final continuation for this test")
+
+	// Replayed against the other namespace. No run ID is sent, so nothing in the request itself
+	// refers to the workflow whose history the token points at.
+	resp, err := env.FrontendClient().GetWorkflowExecutionHistory(
+		s.Context(),
+		&workflowservice.GetWorkflowExecutionHistoryRequest{
+			Namespace:              otherNamespace.String(),
+			Execution:              &commonpb.WorkflowExecution{WorkflowId: env.Tv().WorkflowID()},
+			MaximumPageSize:        1,
+			NextPageToken:          firstPage.NextPageToken,
+			HistoryEventFilterType: enumspb.HISTORY_EVENT_FILTER_TYPE_ALL_EVENT,
+		},
+	)
+	s.Error(err)
+	s.Empty(resp.GetHistory().GetEvents())
+	var invalidArgument *serviceerror.InvalidArgument
+	var notFound *serviceerror.NotFound
+	s.True(
+		errors.As(err, &invalidArgument) || errors.As(err, &notFound),
+		"expected InvalidArgument or NotFound, got %T: %v", err, err,
+	)
+}
+
+func (s *RawHistorySuite) TestGetWorkflowExecutionHistoryReverse_ContinuationFromAnotherNamespace() {
+	env := s.newTestEnv(
+		testcore.WithHistoryShardCount(1),
+		// Shadow mode reports the mismatch but still serves the page.
+		testcore.WithDynamicConfig(dynamicconfig.EnablePaginationTokenBranchValidationShadowMode, false),
+	)
+
+	otherNamespace := namespace.Name(testcore.RandomizeStr("other-namespace"))
+	_, err := env.RegisterNamespace(otherNamespace, 1, enumspb.ARCHIVAL_STATE_DISABLED, "", "")
+	s.Require().NoError(err)
+
+	startMultiBatchWorkflow(s.Context(), s.Require(), env)
+
+	firstPage, err := env.FrontendClient().GetWorkflowExecutionHistoryReverse(
+		s.Context(),
+		&workflowservice.GetWorkflowExecutionHistoryReverseRequest{
+			Namespace:       env.Namespace().String(),
+			Execution:       &commonpb.WorkflowExecution{WorkflowId: env.Tv().WorkflowID()},
+			MaximumPageSize: 1,
+		},
+	)
+	s.Require().NoError(err)
+	s.Require().NotEmpty(firstPage.NextPageToken, "need a non-final continuation for this test")
+
+	resp, err := env.FrontendClient().GetWorkflowExecutionHistoryReverse(
+		s.Context(),
+		&workflowservice.GetWorkflowExecutionHistoryReverseRequest{
+			Namespace:       otherNamespace.String(),
+			Execution:       &commonpb.WorkflowExecution{WorkflowId: env.Tv().WorkflowID()},
+			MaximumPageSize: 1,
+			NextPageToken:   firstPage.NextPageToken,
+		},
+	)
+	s.Error(err)
+	s.Empty(resp.GetHistory().GetEvents())
+}