a

pkg/kv/kvserver/rangefeed/buffered_sender.go

@@ -127,9 +127,10 @@ func NewBufferedSender(
 // SendBuffered buffers the event before sending them to the underlying
 // ServerStreamSender.
 //
-// alloc.Release is nil-safe. SendBuffered will take the ownership of the alloc
-// and release it if the return error is non-nil. Note that it is safe to send
-// error events without being blocked for too long.
+// alloc.Release is nil-safe. SendBuffered will take the ownership of the alloc.
+// It is responsible for using if the return error is nil and releasing the
+// token in the end. Note that it is safe to send error events here without
+// being worried about getting blocked for too long.
 func (bs *BufferedSender) SendBuffered(
 	ev *kvpb.MuxRangeFeedEvent, alloc *SharedBudgetAllocation,
 ) (err error) {

pkg/kv/kvserver/rangefeed/unbuffered_registration.go

@@ -394,9 +394,6 @@ func (ubr *unbufferedRegistration) publishCatchUpBuffer(ctx context.Context) err
 		ubr.metrics.RangefeedCatachUpBufDrainingNanos.Inc(timeutil.Since(start).Nanoseconds())
 	}()
 
-	ubr.mu.Lock()
-	defer ubr.mu.Unlock()
-
 	// TODO(wenyihu6): check if we can just drain without holding the lock first
 	// during reviews We shouldn't be reading from the buffer at the same time
 	publish := func() error {