Merge cockroachdb#155591

155591: rangefeed: extract TxnPushNotifier r=wenyihu6 a=stevendanna

This addresses a testing TODO by extracting the txn push notification goroutine into a struct that can be tested without a store.

This perhaps makes it a bit synthetic of a test since we don't have a test that ensures that the store starts the push notifier. But, we weren't testing anything at all before so this still feels like an improvement.

Epic: none
Release note: None

Co-authored-by: Steven Danna <[email protected]>

Author: craig[bot]

pkg/kv/kvserver/rangefeed/BUILD.bazel

@@ -20,6 +20,7 @@ go_library(
         "stream_manager.go",
         "task.go",
         "test_helpers.go",
+        "txn_push_notifier.go",
         "unbuffered_registration.go",
         "unbuffered_sender.go",
     ],

pkg/kv/kvserver/rangefeed/processor_helpers_test.go

@@ -211,6 +211,7 @@ type processorTestHelper struct {
 	rts                      *resolvedTimestamp
 	syncEventC               func()
 	sendSpanSync             func(*roachpb.Span)
+	rawScheduler             *Scheduler
 	scheduler                *ClientScheduler
 	toBufferedStreamIfNeeded func(s Stream) Stream
 }
@@ -348,7 +349,7 @@ func withSettings(st *cluster.Settings) option {
 	}
 }
 
-func withPushTxnsIntervalAge(interval, age time.Duration) option {
+func withPushTxnsIntervalAge(age time.Duration) option {
 	return func(config *testConfig) {
 		config.PushTxnsAge = age
 	}
@@ -451,6 +452,7 @@ func newTestProcessor(
 		h.sendSpanSync = func(span *roachpb.Span) {
 			p.syncSendAndWait(&syncEvent{c: make(chan struct{}), testRegCatchupSpan: span})
 		}
+		h.rawScheduler = sch
 		h.scheduler = &p.scheduler
 		switch cfg.feedType {
 		case scheduledProcessorWithUnbufferedSender:

pkg/kv/kvserver/rangefeed/processor_test.go

@@ -927,8 +927,8 @@ func TestProcessorTxnPushAttempt(t *testing.T) {
 	})
 }
 
-// TestProcessorTxnPushDisabled tests that processors don't attempt txn pushes
-// when disabled.
+// TestProcessorTxnPushDisabled tests that the TxnPushNotifier doesn't send txn
+// push notifications when disabled.
 func TestProcessorTxnPushDisabled(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 
@@ -951,12 +951,6 @@ func TestProcessorTxnPushDisabled(t *testing.T) {
 	PushTxnsEnabled.Override(ctx, &st.SV, false)
 
 	// Set up a txn pusher and processor that errors on any pushes.
-	//
-	// TODO(kv): We don't test the scheduled processor here, since the setting
-	// instead controls the Store.startRangefeedTxnPushNotifier() loop which sits
-	// outside of the processor and can't be tested with this test harness. Write
-	// a new test when the legacy processor is removed and the scheduled processor
-	// is used by default.
 	var tp testTxnPusher
 	tp.mockPushTxns(func(ctx context.Context, txns []enginepb.TxnMeta, ts hlc.Timestamp) ([]*roachpb.Transaction, bool, error) {
 		err := errors.Errorf("unexpected txn push for txns=%v ts=%s", txns, ts)
@@ -965,9 +959,16 @@ func TestProcessorTxnPushDisabled(t *testing.T) {
 	})
 
 	p, h, stopper := newTestProcessor(t, withSettings(st), withPusher(&tp),
-		withPushTxnsIntervalAge(pushInterval, time.Millisecond))
+		withPushTxnsIntervalAge(time.Millisecond))
 	defer stopper.Stop(ctx)
 
+	notifier := NewTxnPushNotifier(
+		pushInterval,
+		st, h.rawScheduler,
+		func(f func(i int64)) { f(p.ID()) },
+	)
+	require.NoError(t, notifier.Start(ctx, stopper))
+
 	// Move the resolved ts forward to just before the txn timestamp.
 	rts := ts.Add(-1, 0)
 	require.True(t, p.ForwardClosedTS(ctx, rts))

pkg/kv/kvserver/rangefeed/txn_push_notifier.go

@@ -0,0 +1,74 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package rangefeed
+
+import (
+	"context"
+	"time"
+
+	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
+	"github.com/cockroachdb/cockroach/pkg/util/stop"
+)
+
+// TxnPushNotifier enqueues a PushTxnEvent for all processors yielded by
+// visitProcessors every interval.
+type TxnPushNotifier struct {
+	interval        time.Duration
+	settings        *cluster.Settings
+	scheduler       *Scheduler
+	visitProcessors func(func(int64))
+}
+
+func NewTxnPushNotifier(
+	interval time.Duration,
+	settings *cluster.Settings,
+	scheduler *Scheduler,
+	visitProcessors func(func(int64)),
+) *TxnPushNotifier {
+	return &TxnPushNotifier{
+		interval:        interval,
+		settings:        settings,
+		scheduler:       scheduler,
+		visitProcessors: visitProcessors,
+	}
+}
+
+func (t *TxnPushNotifier) Start(ctx context.Context, stopper *stop.Stopper) error {
+	taskOpts := stop.TaskOpts{
+		TaskName: "rangefeed-txn-push-notifier",
+		SpanOpt:  stop.SterileRootSpan,
+	}
+	ctx, hdl, err := stopper.GetHandle(ctx, taskOpts)
+	if err != nil {
+		return err
+	}
+	go func(ctx context.Context, hdl *stop.Handle) {
+		defer hdl.Activate(ctx).Release(ctx)
+
+		ctx, cancel := stopper.WithCancelOnQuiesce(ctx)
+		defer cancel()
+
+		ticker := time.NewTicker(t.interval)
+		defer ticker.Stop()
+
+		for {
+			select {
+			case <-ticker.C:
+				if !PushTxnsEnabled.Get(&t.settings.SV) {
+					continue
+				}
+
+				batch := t.scheduler.NewEnqueueBatch()
+				t.visitProcessors(batch.Add)
+				t.scheduler.EnqueueBatch(batch, PushTxnQueued)
+				batch.Close()
+			case <-ctx.Done():
+				return
+			}
+		}
+	}(ctx, hdl)
+	return nil
+}

pkg/kv/kvserver/store.go

@@ -2454,7 +2454,9 @@ func (s *Store) Start(ctx context.Context, stopper *stop.Stopper) error {
 	// Connect rangefeeds to closed timestamp updates.
 	s.startRangefeedUpdater(ctx)
 
-	s.startRangefeedTxnPushNotifier(ctx)
+	if err := s.startRangefeedTxnPushNotifier(ctx); err != nil {
+		return err
+	}
 
 	if s.replicateQueue != nil {
 		s.storeRebalancer = NewStoreRebalancer(
@@ -2662,48 +2664,27 @@ func (s *Store) startRangefeedUpdater(ctx context.Context) {
 // startRangefeedTxnPushNotifier starts a worker that would periodically
 // enqueue txn push event for rangefeed processors to let them push lagging
 // transactions.
-func (s *Store) startRangefeedTxnPushNotifier(ctx context.Context) {
+func (s *Store) startRangefeedTxnPushNotifier(ctx context.Context) error {
 	interval := rangefeed.DefaultPushTxnsInterval
 	if i := s.TestingKnobs().RangeFeedPushTxnsInterval; i > 0 {
 		interval = i
 	}
-
-	_ /* err */ = s.stopper.RunAsyncTaskEx(ctx, stop.TaskOpts{
-		TaskName: "transaction-rangefeed-push-notifier",
-		SpanOpt:  stop.SterileRootSpan,
-	}, func(ctx context.Context) {
-		ctx, cancel := s.stopper.WithCancelOnQuiesce(ctx)
-		defer cancel()
-
-		makeSchedulerBatch := func() *rangefeed.SchedulerBatch {
-			batch := s.rangefeedScheduler.NewEnqueueBatch()
+	tpn := rangefeed.NewTxnPushNotifier(
+		interval,
+		s.ClusterSettings(),
+		s.rangefeedScheduler,
+		func(f func(int64)) {
 			s.rangefeedReplicas.Lock()
-			for _, id := range s.rangefeedReplicas.m {
-				if id != 0 {
-					// Only process ranges that use scheduler.
-					batch.Add(id)
+			for _, procID := range s.rangefeedReplicas.m {
+				// Only process ranges that use scheduler.
+				if procID != 0 {
+					f(procID)
 				}
 			}
 			s.rangefeedReplicas.Unlock()
-			return batch
-		}
-
-		ticker := time.NewTicker(interval)
-		for {
-			select {
-			case <-ticker.C:
-				if !rangefeed.PushTxnsEnabled.Get(&s.ClusterSettings().SV) {
-					continue
-				}
-				batch := makeSchedulerBatch()
-				s.rangefeedScheduler.EnqueueBatch(batch, rangefeed.PushTxnQueued)
-				batch.Close()
-			case <-ctx.Done():
-				ticker.Stop()
-				return
-			}
-		}
-	})
+		},
+	)
+	return tpn.Start(ctx, s.stopper)
 }
 
 func (s *Store) addReplicaWithRangefeed(rangeID roachpb.RangeID, schedulerID int64) {