Merge #153740 #153965 #154027

153740: kvserver/rangefeed: add capacity to node level buffered sender r=stevendanna a=wenyihu6

Resolves: #152505
Release note: none

---
**kvserver/rangefeed: add capacity to node level buffered sender**

This patch adds capacity to node level buffered sender which will shut down all
registrations if the node level buffer had overflowed.

Part of: #129813
Release note: none

---
**kvserver/rangefeed: rename overflow to overflowed**

This commit renames overflow to overflowed in BufferedSender for clarity.

---
**kvserver/rangefeed: dynamically adjust capacity per buffered sender**

Previously, buffered sender capacity was configured via a cluster setting. This
commit removes that setting and makes capacity scale dynamically with the number
of active registrations. The minimum (and default) is 4096 * 20, and capacity
never drops below that. Above the minimum, it grows as 4096 * <active
registrations>. Events already buffered will not dropped when capacity
decreases, and capacity is not adjusted during shutdown (0).

---
**rangefeed: move back to static (bug configurable) send queue limit**

The interactions here probably require more thought before doing
anything more complicated than a simple static limit.

Epic: none
Release note: None

---
**rangefeed: fix incorrect comment**

Epic: none
Release note: None

---
**rangefeed: deflake TestUnbufferedRegWithStreamManager**

Under stress, this test would fail because attempting to publish 20
events in a tight loop would overwhelm the processor's event buffer
(which has a capacity of 16).

Epic: none
Release note: None



153965: sql/inspect: support checking multiple indexes in one INSPECT job r=spilchen a=spilchen

Previously, there was an assert that made sure when building the INSPECT job that we only had one check. This removes that assert so that we can have mulitple checks. When you run INSPECT against many indexes on a table or against a database, we will have multiple checks, so it was necessary for that case.

Closes #148300

Release note: none
Epic: none

154027: activity: add fallback isolation level for statements not in transaction r=jasonlmfong a=jasonlmfong

The change is achieved by retrieving the session default isolation level

closes: [#153931](#153931)
Epic: None
Release note: None

Co-authored-by: wenyihu6 <[email protected]>
Co-authored-by: Steven Danna <[email protected]>
Co-authored-by: Matt Spilchen <[email protected]>
Co-authored-by: Jason Fong <[email protected]>

Author: 5 people

pkg/kv/kvserver/kvserverbase/base.go

@@ -383,3 +383,16 @@ var MaxCommandSize = settings.RegisterByteSizeSetting(
 	MaxCommandSizeDefault,
 	settings.ByteSizeWithMinimum(MaxCommandSizeFloor),
 )
+
+// DefaultRangefeedEventCap is the channel capacity of the rangefeed processor
+// and each registration. It is also used to calculate the default capacity
+// limit for the buffered sender.
+//
+// The size of an event is 72 bytes, so this will result in an allocation on the
+// order of ~300KB per RangeFeed. That's probably ok given the number of ranges
+// on a node that we'd like to support with active rangefeeds, but it's
+// certainly on the upper end of the range.
+//
+// Note that processors also must reserve memory from one of two memory monitors
+// for each event.
+const DefaultRangefeedEventCap = 4096

pkg/kv/kvserver/rangefeed/BUILD.bazel

@@ -30,6 +30,7 @@ go_library(
         "//pkg/kv/kvpb",
         "//pkg/kv/kvserver/concurrency/isolation",
         "//pkg/kv/kvserver/concurrency/lock",
+        "//pkg/kv/kvserver/kvserverbase",
         "//pkg/roachpb",
         "//pkg/settings",
         "//pkg/settings/cluster",

pkg/kv/kvserver/rangefeed/buffered_sender.go

@@ -10,6 +10,9 @@ import (
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvserverbase"
+	"github.com/cockroachdb/cockroach/pkg/settings"
+	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/util/retry"
 	"github.com/cockroachdb/cockroach/pkg/util/stop"
 	"github.com/cockroachdb/cockroach/pkg/util/syncutil"
@@ -45,6 +48,38 @@ import (
 //               BufferedPerRangeEventSink.Send    BufferedPerRangeEventSink.SendError
 //
 
+// RangefeedSingleBufferedSenderQueueMaxSize is the maximum number of events
+// that the buffered sender will buffer before it starts returning capacity
+// exceeded errors. Updates to this setting are only applied to need
+// MuxRangefeedCalls, existing streams will use the previous value until
+// restarted.
+//
+// # The main goal of this limit is to provide a backstop against the
+//
+// The default here has been somewhat arbitrarily chosen considering that:
+//
+//   - We want to avoid capacity exceeded errors that wouldn't have occurred
+//     when the buffered registrations were in use.
+//
+//   - We don't want to drastically increase the amount of queueing allowed for a
+//     single registration.
+//
+//   - One buffered sender is feeding a single gRPC client.
+//
+//   - Events emitted during catchup scans have their own per-registration buffer
+//     still.
+//
+// TODO(ssd): This is a bit of a stop-gap so that we have a knob to turn if we
+// need to. We probably want each buffered sender (or each consumerID) to be
+// able to hold up to some fraction of the total rangefeed budget. But we are
+// starting here for now.
+var RangefeedSingleBufferedSenderQueueMaxSize = settings.RegisterIntSetting(
+	settings.SystemOnly,
+	"kv.rangefeed.buffered_sender.queue_max_size",
+	"max size of a buffered senders event queue (0 for no max)",
+	kvserverbase.DefaultRangefeedEventCap*8,
+)
+
 // BufferedSender is embedded in every rangefeed.BufferedPerRangeEventSink,
 // serving as a helper which buffers events before forwarding events to the
 // underlying gRPC stream.
@@ -58,6 +93,9 @@ type BufferedSender struct {
 		syncutil.Mutex
 		stopped bool
 		buffer  *eventQueue
+		// capacity is the maximum number of events that can be buffered.
+		capacity   int64
+		overflowed bool
 	}
 
 	// notifyDataC is used to notify the BufferedSender.run goroutine that there
@@ -72,14 +110,16 @@ type BufferedSender struct {
 }
 
 func NewBufferedSender(
-	sender ServerStreamSender, bsMetrics *BufferedSenderMetrics,
+	sender ServerStreamSender, settings *cluster.Settings, bsMetrics *BufferedSenderMetrics,
 ) *BufferedSender {
 	bs := &BufferedSender{
 		sender:  sender,
 		metrics: bsMetrics,
 	}
 	bs.queueMu.buffer = newEventQueue()
 	bs.notifyDataC = make(chan struct{}, 1)
+	bs.queueMu.buffer = newEventQueue()
+	bs.queueMu.capacity = RangefeedSingleBufferedSenderQueueMaxSize.Get(&settings.SV)
 	return bs
 }
 
@@ -95,6 +135,13 @@ func (bs *BufferedSender) sendBuffered(
 	if bs.queueMu.stopped {
 		return errors.New("stream sender is stopped")
 	}
+	if bs.queueMu.overflowed {
+		return newRetryErrBufferCapacityExceeded()
+	}
+	if bs.queueMu.capacity > 0 && bs.queueMu.buffer.len() >= bs.queueMu.capacity {
+		bs.queueMu.overflowed = true
+		return newRetryErrBufferCapacityExceeded()
+	}
 	// TODO(wenyihu6): pass an actual context here
 	alloc.Use(context.Background())
 	bs.queueMu.buffer.pushBack(sharedMuxEvent{ev, alloc})
@@ -130,7 +177,7 @@ func (bs *BufferedSender) run(
 			return nil
 		case <-bs.notifyDataC:
 			for {
-				e, success := bs.popFront()
+				e, success, overflowed, remains := bs.popFront()
 				if !success {
 					break
 				}
@@ -143,18 +190,26 @@ func (bs *BufferedSender) run(
 				if err != nil {
 					return err
 				}
+				if overflowed && remains == int64(0) {
+					return newRetryErrBufferCapacityExceeded()
+				}
 			}
 		}
 	}
 }
 
 // popFront pops the front event from the buffer queue. It returns the event and
 // a boolean indicating if the event was successfully popped.
-func (bs *BufferedSender) popFront() (e sharedMuxEvent, success bool) {
+func (bs *BufferedSender) popFront() (
+	e sharedMuxEvent,
+	success bool,
+	overflowed bool,
+	remains int64,
+) {
 	bs.queueMu.Lock()
 	defer bs.queueMu.Unlock()
 	event, ok := bs.queueMu.buffer.popFront()
-	return event, ok
+	return event, ok, bs.queueMu.overflowed, bs.queueMu.buffer.len()
 }
 
 // cleanup is called when the sender is stopped. It is expected to free up
@@ -168,13 +223,18 @@ func (bs *BufferedSender) cleanup(ctx context.Context) {
 	bs.metrics.BufferedSenderQueueSize.Dec(remaining)
 }
 
-// Used for testing only.
 func (bs *BufferedSender) len() int {
 	bs.queueMu.Lock()
 	defer bs.queueMu.Unlock()
 	return int(bs.queueMu.buffer.len())
 }
 
+func (bs *BufferedSender) overflowed() bool {
+	bs.queueMu.Lock()
+	defer bs.queueMu.Unlock()
+	return bs.queueMu.overflowed
+}
+
 // Used for testing only.
 func (bs *BufferedSender) waitForEmptyBuffer(ctx context.Context) error {
 	opts := retry.Options{

pkg/kv/kvserver/rangefeed/buffered_sender_test.go

@@ -13,6 +13,8 @@ import (
 
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
+	"github.com/cockroachdb/cockroach/pkg/testutils"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
@@ -34,7 +36,8 @@ func TestBufferedSenderDisconnectStream(t *testing.T) {
 	defer stopper.Stop(ctx)
 	testServerStream := newTestServerStream()
 	smMetrics := NewStreamManagerMetrics()
-	bs := NewBufferedSender(testServerStream, NewBufferedSenderMetrics())
+	st := cluster.MakeTestingClusterSettings()
+	bs := NewBufferedSender(testServerStream, st, NewBufferedSenderMetrics())
 	sm := NewStreamManager(bs, smMetrics)
 	require.NoError(t, sm.Start(ctx, stopper))
 	defer sm.Stop(ctx)
@@ -87,7 +90,8 @@ func TestBufferedSenderChaosWithStop(t *testing.T) {
 	testServerStream := newTestServerStream()
 
 	smMetrics := NewStreamManagerMetrics()
-	bs := NewBufferedSender(testServerStream, NewBufferedSenderMetrics())
+	st := cluster.MakeTestingClusterSettings()
+	bs := NewBufferedSender(testServerStream, st, NewBufferedSenderMetrics())
 	sm := NewStreamManager(bs, smMetrics)
 	require.NoError(t, sm.Start(ctx, stopper))
 
@@ -164,3 +168,125 @@ func TestBufferedSenderChaosWithStop(t *testing.T) {
 		require.Equal(t, 0, bs.len())
 	})
 }
+
+// TestBufferedSenderOnOverflow tests that BufferedSender handles overflow
+// properly. It does not test the shutdown flow with stream manager.
+func TestBufferedSenderOnOverflow(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+	ctx := context.Background()
+
+	stopper := stop.NewStopper()
+	defer stopper.Stop(ctx)
+	testServerStream := newTestServerStream()
+	st := cluster.MakeTestingClusterSettings()
+
+	queueCap := int64(24)
+	RangefeedSingleBufferedSenderQueueMaxSize.Override(ctx, &st.SV, queueCap)
+	bs := NewBufferedSender(testServerStream, st, NewBufferedSenderMetrics())
+	require.Equal(t, queueCap, bs.queueMu.capacity)
+
+	val1 := roachpb.Value{RawBytes: []byte("val"), Timestamp: hlc.Timestamp{WallTime: 1}}
+	ev1 := new(kvpb.RangeFeedEvent)
+	ev1.MustSetValue(&kvpb.RangeFeedValue{Key: keyA, Value: val1})
+	muxEv := &kvpb.MuxRangeFeedEvent{RangeFeedEvent: *ev1, RangeID: 0, StreamID: 1}
+
+	for range queueCap {
+		require.NoError(t, bs.sendBuffered(muxEv, nil))
+	}
+	require.Equal(t, queueCap, int64(bs.len()))
+	e, success, overflowed, remains := bs.popFront()
+	require.Equal(t, sharedMuxEvent{
+		ev:    muxEv,
+		alloc: nil,
+	}, e)
+	require.True(t, success)
+	require.False(t, overflowed)
+	require.Equal(t, queueCap-1, remains)
+	require.Equal(t, queueCap-1, int64(bs.len()))
+	require.NoError(t, bs.sendBuffered(muxEv, nil))
+	require.Equal(t, queueCap, int64(bs.len()))
+
+	// Overflow now.
+	require.Equal(t, bs.sendBuffered(muxEv, nil).Error(),
+		newRetryErrBufferCapacityExceeded().Error())
+}
+
+// TestBufferedSenderOnStreamShutdown tests that BufferedSender and
+// StreamManager handle overflow and shutdown properly.
+func TestBufferedSenderOnStreamShutdown(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+	ctx := context.Background()
+
+	stopper := stop.NewStopper()
+	defer stopper.Stop(ctx)
+	testServerStream := newTestServerStream()
+	smMetrics := NewStreamManagerMetrics()
+	st := cluster.MakeTestingClusterSettings()
+
+	queueCap := int64(24)
+	RangefeedSingleBufferedSenderQueueMaxSize.Override(ctx, &st.SV, queueCap)
+	bs := NewBufferedSender(testServerStream, st, NewBufferedSenderMetrics())
+	require.Equal(t, queueCap, bs.queueMu.capacity)
+
+	sm := NewStreamManager(bs, smMetrics)
+	require.NoError(t, sm.Start(ctx, stopper))
+	defer sm.Stop(ctx)
+
+	p, h, pStopper := newTestProcessor(t, withRangefeedTestType(scheduledProcessorWithBufferedSender))
+	defer pStopper.Stop(ctx)
+
+	streamID := int64(42)
+
+	val1 := roachpb.Value{RawBytes: []byte("val"), Timestamp: hlc.Timestamp{WallTime: 1}}
+	ev1 := new(kvpb.RangeFeedEvent)
+	ev1.MustSetValue(&kvpb.RangeFeedValue{Key: keyA, Value: val1})
+	muxEv := &kvpb.MuxRangeFeedEvent{RangeFeedEvent: *ev1, RangeID: 0, StreamID: streamID}
+
+	// Block the stream so that we can overflow later.
+	unblock := testServerStream.BlockSend()
+	defer unblock()
+
+	waitForQueueLen := func(len int) {
+		testutils.SucceedsSoon(t, func() error {
+			if bs.len() == len {
+				return nil
+			}
+			return errors.Newf("expected %d events, found %d", len, bs.len())
+		})
+	}
+
+	// Add our stream to the stream manager.
+	registered, d, _ := p.Register(ctx, h.span, hlc.Timestamp{}, nil, /* catchUpIter */
+		false /* withDiff */, false /* withFiltering */, false /* withOmitRemote */, noBulkDelivery,
+		sm.NewStream(streamID, 1 /*rangeID*/))
+	require.True(t, registered)
+	sm.AddStream(streamID, d)
+
+	require.NoError(t, sm.sender.sendBuffered(muxEv, nil))
+	// At this point we actually have sent 2 events. 1 checkpoint event sent by
+	// register and 1 event sent on the line above. We wait for 1 of these events
+	// to be pulled off the queue and block in the sender, leaving 1 in the queue.
+	waitForQueueLen(1)
+	// Now fill the rest of the queue.
+	for range queueCap - 1 {
+		require.NoError(t, sm.sender.sendBuffered(muxEv, nil))
+	}
+
+	// The next write should overflow.
+	capExceededErrStr := newRetryErrBufferCapacityExceeded().Error()
+	err := sm.sender.sendBuffered(muxEv, nil)
+	require.EqualError(t, err, capExceededErrStr)
+	require.True(t, bs.overflowed())
+
+	unblock()
+	waitForQueueLen(0)
+	// Overflow cleanup.
+	err = <-sm.Error()
+	require.EqualError(t, err, capExceededErrStr)
+	// Note that we expect the stream manager to shut down here, but no actual
+	// error events would be sent during the shutdown.
+	err = sm.sender.sendBuffered(muxEv, nil)
+	require.EqualError(t, err, capExceededErrStr)
+}

pkg/kv/kvserver/rangefeed/sender_helper_test.go

@@ -35,6 +35,9 @@ type testServerStream struct {
 	eventsSent int
 	// streamEvents is a map of streamID to a list of events sent to that stream.
 	streamEvents map[int64][]*kvpb.MuxRangeFeedEvent
+
+	// t can optionally set for additional logging.
+	t *testing.T
 }
 
 var _ ServerStreamSender = &testServerStream{}
@@ -121,6 +124,9 @@ func (s *testServerStream) SendIsThreadSafe() {}
 // Send mocks grpc.ServerStream Send method. It only counts events and stores
 // events by streamID in streamEvents.
 func (s *testServerStream) Send(e *kvpb.MuxRangeFeedEvent) error {
+	if s.t != nil {
+		s.t.Logf("Sending event for StreamID %d: %v", e.StreamID, e)
+	}
 	s.Lock()
 	defer s.Unlock()
 	s.eventsSent++

pkg/kv/kvserver/rangefeed/stream_manager_test.go

@@ -13,6 +13,7 @@ import (
 	"testing"
 
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
+	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/testutils"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
@@ -175,12 +176,13 @@ func TestStreamManagerErrorHandling(t *testing.T) {
 	testutils.RunValues(t, "feed type", testTypes, func(t *testing.T, rt rangefeedTestType) {
 		testServerStream := newTestServerStream()
 		smMetrics := NewStreamManagerMetrics()
+		st := cluster.MakeTestingClusterSettings()
 		var s sender
 		switch rt {
 		case scheduledProcessorWithUnbufferedSender:
 			s = NewUnbufferedSender(testServerStream)
 		case scheduledProcessorWithBufferedSender:
-			s = NewBufferedSender(testServerStream, NewBufferedSenderMetrics())
+			s = NewBufferedSender(testServerStream, st, NewBufferedSenderMetrics())
 		default:
 			t.Fatalf("unknown rangefeed test type %v", rt)
 		}