rangefeed: dedup TestBufferedSenderDisconnectStream

This tests duplicated TestStreamManagerDisconnectStream, which now
supports both sender types.

Epic: none
Release note: None

Author: stevendanna

pkg/kv/kvserver/rangefeed/buffered_sender_test.go

@@ -7,7 +7,6 @@ package rangefeed
 
 import (
 	"context"
-	"sync/atomic"
 	"testing"
 
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
@@ -22,73 +21,6 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-// TestBufferedSenderWithSendBufferedError tests that BufferedSender can handle stream
-// disconnects properly including context canceled, metrics updates, rangefeed
-// cleanup.
-func TestBufferedSenderDisconnectStream(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()
-	bs := NewBufferedSender(testServerStream, st, NewBufferedSenderMetrics())
-	sm := NewStreamManager(bs, smMetrics)
-	require.NoError(t, sm.Start(ctx, stopper))
-	defer sm.Stop(ctx)
-
-	var sid int64
-	nextStreamID := func() int64 {
-		sid++
-		return sid
-	}
-
-	err := kvpb.NewError(kvpb.NewRangeFeedRetryError(kvpb.RangeFeedRetryError_REASON_NO_LEASEHOLDER))
-	errEvent := func(streamID int64) *kvpb.MuxRangeFeedEvent {
-		return makeMuxRangefeedErrorEvent(streamID, 1, err)
-	}
-
-	t.Run("basic operation", func(t *testing.T) {
-		var num atomic.Int32
-		streamID := nextStreamID()
-		sm.RegisteringStream(streamID)
-		sm.AddStream(streamID, &cancelCtxDisconnector{
-			cancel: func() {
-				num.Add(1)
-				require.NoError(t, sm.sender.sendBuffered(errEvent(streamID), nil))
-			},
-		})
-		require.Equal(t, int64(1), smMetrics.ActiveMuxRangeFeed.Value())
-		require.Equal(t, 0, bs.len())
-		sm.DisconnectStream(streamID, err)
-		testServerStream.waitForEvent(t, errEvent(streamID))
-		require.Equal(t, int32(1), num.Load())
-		require.Equal(t, 1, testServerStream.totalEventsSent())
-		waitForRangefeedCount(t, smMetrics, 0)
-		testServerStream.reset()
-	})
-	t.Run("disconnect stream on the same stream is idempotent", func(t *testing.T) {
-		streamID := nextStreamID()
-		sm.RegisteringStream(streamID)
-		sm.AddStream(streamID, &cancelCtxDisconnector{
-			cancel: func() {
-				require.NoError(t, sm.sender.sendBuffered(errEvent(streamID), nil))
-			},
-		})
-		require.Equal(t, int64(1), smMetrics.ActiveMuxRangeFeed.Value())
-		sm.DisconnectStream(streamID, err)
-		testServerStream.waitForEvent(t, errEvent(streamID))
-		sm.DisconnectStream(streamID, err)
-		require.NoError(t, bs.waitForEmptyBuffer(ctx))
-		require.Equalf(t, 1, testServerStream.totalEventsSent(),
-			"expected only 1 error event in %s", testServerStream.String())
-		waitForRangefeedCount(t, smMetrics, 0)
-	})
-}
-
 func TestBufferedSenderReturnsErrorAfterManagerStop(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)