fix(queuev2): broaden RTrim guard to catch raise-upper case

The previous guard used Less(prevUpper) which only caught trims that
shrank the upper bound below prevUpper. It missed the case where
RTrimBySize kept some prefetched tasks (raising the bound above
prevUpper) but dropped the newest ones. The re-advance on lines 392-399
then claimed coverage for the dropped tail, causing GetTask to miss
tasks that exist in the DB.

Changed to !Equal(prevUpper): any change means putTasks called
updateExclusiveUpperBound (= a trim occurred), so skip re-advance in
all cases.

Test: TestCachedQueueReader_PrefetchTrimRaisesUpperBoundAbovePrevUpper
simulates inject filling 1 slot during DB fetch, DB returning 3 tasks,
RTrim keeping 3 oldest and dropping the 4th, confirming the guard fires
even though the trimmed bound is above prevUpper.

Signed-off-by: Seva Kaloshin <seva.kaloshin@gmail.com>

Author: arzonus

service/history/queuev2/queue_reader_cached.go

@@ -371,15 +371,14 @@ func (q *cachedQueueReader) prefetch() error {
 	prevUpper := q.exclusiveUpperBound
 	q.putTasks(resp.Tasks)
 
-	// putTasks calls RTrimBySize, which can shrink exclusiveUpperBound if
-	// concurrent Inject calls filled the cache between the capacity snapshot
-	// and the write-lock acquisition above. When a trim occurred, the upper
-	// bound already reflects the actual cache contents. Re-advancing it to
-	// exclusiveMaxKey or NextTaskKey would claim coverage for tasks that were
-	// just evicted, causing GetTask cache hits to return fewer tasks than exist
-	// in the DB for that range.
-	if q.exclusiveUpperBound.Less(prevUpper) {
-		q.logger.Debug("prefetch skipped upper-bound advance: cache trimmed after concurrent inject",
+	// putTasks calls RTrimBySize, which calls updateExclusiveUpperBound when a
+	// trim occurs. Any change to exclusiveUpperBound here means the cache no
+	// longer holds a contiguous prefix up to NextTaskKey or exclusiveMaxKey, so
+	// re-advancing would create false coverage. Use Equal (not Less) to catch
+	// both the shrink case and the case where the trim kept some DB tasks,
+	// raising the bound above prevUpper but below the target ceiling.
+	if !q.exclusiveUpperBound.Equal(prevUpper) {
+		q.logger.Debug("prefetch skipped upper-bound advance: cache trimmed",
 			tag.Dynamic("trimmedUpper", q.exclusiveUpperBound),
 			tag.Dynamic("prevUpper", prevUpper),
 		)

service/history/queuev2/queue_reader_cached_test.go

@@ -1545,6 +1545,61 @@ func TestCachedQueueReader_PrefetchTrimPreventsUpperBoundAdvance(t *testing.T) {
 		"trimmed upper bound should be less than prevUpper")
 }
 
+// TestCachedQueueReader_PrefetchTrimRaisesUpperBoundAdvance verifies the
+// complementary case: when RTrimBySize keeps some prefetched tasks (raising
+// exclusiveUpperBound above prevUpper) but drops the newest ones, the prefetch
+// must NOT re-advance to NextTaskKey or exclusiveMaxKey. Doing so would claim
+// coverage for the dropped tasks, causing GetTask hits to miss them.
+func TestCachedQueueReader_PrefetchTrimRaisesUpperBoundAbovePrevUpper(t *testing.T) {
+	now := time.Date(2025, 1, 1, 12, 0, 0, 0, time.UTC)
+	ts := clock.NewMockedTimeSourceAt(now)
+	ctrl := gomock.NewController(t)
+	opts := defaultTestOptions()
+	// MaxSize=3. prevUpper = now+1min. DB returns 3 tasks (now+2m, now+3m, now+4m).
+	// During the DB round-trip, inject adds 1 inject task (now+30s, below prevUpper).
+	// After putTasks: queue = inject(30s) + db(2m) + db(3m) + db(4m) = 4 tasks > MaxSize=3.
+	// RTrimBySize(3) keeps 3 oldest: inject(30s), db(2m), db(3m); drops db(4m).
+	// newUpper = db(3m).Next() > prevUpper (now+1min) → !Equal fires → skip re-advance.
+	opts.MaxSize = dynamicproperties.GetIntPropertyFn(3)
+
+	injectTask := newTestTask(now.Add(30*time.Second), 0)
+	dbTask1 := newTestTask(now.Add(2*time.Minute), 1)
+	dbTask2 := newTestTask(now.Add(3*time.Minute), 2)
+	dbTask3 := newTestTask(now.Add(4*time.Minute), 3) // will be dropped
+	prevUpper := persistence.NewHistoryTaskKey(now.Add(1*time.Minute), 0)
+
+	base := NewMockQueueReader(ctrl)
+	queue := newInMemQueue()
+	r := newCachedQueueReaderWithOptions(base, queue, opts, ts, testlogger.New(t), metrics.NoopScope)
+	r.exclusiveUpperBound = prevUpper
+
+	base.EXPECT().GetTask(gomock.Any(), gomock.Any()).DoAndReturn(
+		func(_ context.Context, _ *GetTaskRequest) (*GetTaskResponse, error) {
+			r.mu.Lock()
+			r.queue.PutTasks([]persistence.Task{injectTask})
+			r.mu.Unlock()
+			nextKey := persistence.NewHistoryTaskKey(now.Add(5*time.Minute), 0)
+			return &GetTaskResponse{
+				Tasks:    []persistence.Task{dbTask1, dbTask2, dbTask3},
+				Progress: &GetTaskProgress{NextTaskKey: nextKey},
+			}, nil
+		},
+	)
+
+	err := r.prefetch()
+	assert.NoError(t, err)
+
+	// Queue should hold the 3 oldest: injectTask, dbTask1, dbTask2. dbTask3 dropped.
+	assert.Equal(t, 3, r.queue.Len())
+	// Upper bound must be dbTask2.Next() — NOT advanced to NextTaskKey (now+5min).
+	wantUpper := dbTask2.GetTaskKey().Next()
+	assert.Equal(t, wantUpper, r.exclusiveUpperBound,
+		"upper bound must not advance past trimmed point even when trim raises bound above prevUpper")
+	// Confirm it is above prevUpper (the scenario the original Less check missed).
+	assert.True(t, r.exclusiveUpperBound.Greater(prevUpper),
+		"trimmed upper is above prevUpper — the case the Less-only guard missed")
+}
+
 func TestCachedQueueReader_PrefetchGapDetected(t *testing.T) {
 	now := time.Date(2025, 1, 1, 12, 0, 0, 0, time.UTC)
 	ts := clock.NewMockedTimeSourceAt(now)