kvserver: add COCKROACH_RANGEFEED_SEND_TIMEOUT

This patch adds `COCKROACH_RANGEFEED_SEND_TIMEOUT` controlling the
rangefeed channel send timeout. This is primarily intended for disabling
the timeout entirely in benchmarks, instead backpressuring writers.

Epic: none
Release note: None

Author: erikgrinaker

pkg/kv/kvserver/replica_rangefeed.go

@@ -28,6 +28,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/storage"
 	"github.com/cockroachdb/cockroach/pkg/storage/enginepb"
 	"github.com/cockroachdb/cockroach/pkg/util/admission"
+	"github.com/cockroachdb/cockroach/pkg/util/envutil"
 	"github.com/cockroachdb/cockroach/pkg/util/future"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
@@ -73,6 +74,27 @@ var RangeFeedSmearInterval = settings.RegisterDurationSetting(
 	settings.NonNegativeDuration,
 )
 
+// defaultEventChanCap is the channel capacity of the rangefeed processor and
+// each registration.
+//
+// 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.
+//
+// TODO(dan): Everyone seems to agree that this memory limit would be better set
+// at a store-wide level, but there doesn't seem to be an easy way to accomplish
+// that.
+const defaultEventChanCap = 4096
+
+// defaultEventChanTimeout is the send timeout for events published to a
+// rangefeed processor or rangefeed client channels. When exceeded, the
+// rangefeed or client is disconnected to prevent blocking foreground traffic
+// for longer than this timeout. When set to 0, clients are never disconnected,
+// and slow consumers will backpressure writers up through Raft.
+var defaultEventChanTimeout = envutil.EnvOrDefaultDuration(
+	"COCKROACH_RANGEFEED_SEND_TIMEOUT", 50*time.Millisecond)
+
 // lockedRangefeedStream is an implementation of rangefeed.Stream which provides
 // support for concurrent calls to Send. Note that the default implementation of
 // grpc.Stream is not safe for concurrent calls to Send.
@@ -293,16 +315,6 @@ func (r *Replica) updateRangefeedFilterLocked() bool {
 	return false
 }
 
-// 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.
-//
-// TODO(dan): Everyone seems to agree that this memory limit would be better set
-// at a store-wide level, but there doesn't seem to be an easy way to accomplish
-// that.
-const defaultEventChanCap = 4096
-
 // Rangefeed registration takes place under the raftMu, so log if we ever hold
 // the mutex for too long, as this could affect foreground traffic.
 //
@@ -372,7 +384,7 @@ func (r *Replica) registerWithRangefeedRaftMuLocked(
 		PushTxnsInterval: r.store.TestingKnobs().RangeFeedPushTxnsInterval,
 		PushTxnsAge:      r.store.TestingKnobs().RangeFeedPushTxnsAge,
 		EventChanCap:     defaultEventChanCap,
-		EventChanTimeout: 50 * time.Millisecond,
+		EventChanTimeout: defaultEventChanTimeout,
 		Metrics:          r.store.metrics.RangeFeedMetrics,
 		MemBudget:        feedBudget,
 	}