Merge pull request #123676 from cockroachdb/blathers/backport-release…

…-24.1.0-rc-123519 release-24.1.0-rc: rangefeed: fix sendMetadata deadlock
cockroachdb · May 7, 2024 · 6205244 · 6205244
2 parents 418905b + bb9a627
commit 6205244
Show file tree

Hide file tree

Showing 3 changed files with 23 additions and 5 deletions.
diff --git a/pkg/ccl/streamingccl/streamproducer/event_stream.go b/pkg/ccl/streamingccl/streamproducer/event_stream.go
@@ -306,7 +306,13 @@ func (s *eventStream) onMetadata(ctx context.Context, metadata *kvpb.RangeFeedMe
 	if metadata.FromManualSplit && !metadata.Span.Key.Equal(metadata.ParentStartKey) {
 		// Only send new manual split keys (i.e. a child rangefeed start key that
 		// differs from the parent start key)
+		if s.addMu != nil {
+			// Split points can be sent concurrently during the initial scan.
+			s.addMu.Lock()
+			defer s.addMu.Unlock()
+		}
 		s.seb.addSplitPoint(metadata.Span.Key)
+		s.setErr(s.maybeFlushBatch(ctx))
 	}
 }
 

diff --git a/pkg/kv/kvclient/kvcoord/dist_sender_mux_rangefeed.go b/pkg/kv/kvclient/kvcoord/dist_sender_mux_rangefeed.go
@@ -237,7 +237,9 @@ func (m *rangefeedMuxer) startSingleRangeFeed(
 	if m.cfg.withMetadata {
 		// Send metadata after the stream successfully registers to avoid sending
 		// metadata about a rangefeed that never starts.
-		sendMetadata(m.eventCh, span, parentRangefeedMetadata)
+		if err := sendMetadata(ctx, m.eventCh, span, parentRangefeedMetadata); err != nil {
+			return err
+		}
 	}
 
 	return nil

diff --git a/pkg/kv/kvclient/kvcoord/dist_sender_rangefeed.go b/pkg/kv/kvclient/kvcoord/dist_sender_rangefeed.go
@@ -285,7 +285,9 @@ func (ds *DistSender) RangeFeedSpans(
 				})
 
 				if cfg.withMetadata {
-					sendMetadata(eventCh, span, sri.parentRangeFeedMetadata)
+					if err := sendMetadata(ctx, eventCh, span, sri.parentRangeFeedMetadata); err != nil {
+						return err
+					}
 				}
 
 			case <-ctx.Done():
@@ -520,9 +522,13 @@ type parentRangeFeedMetadata struct {
 }
 
 func sendMetadata(
-	eventCh chan<- RangeFeedMessage, span roachpb.Span, parentMetadata parentRangeFeedMetadata,
-) {
-	eventCh <- RangeFeedMessage{
+	ctx context.Context,
+	eventCh chan<- RangeFeedMessage,
+	span roachpb.Span,
+	parentMetadata parentRangeFeedMetadata,
+) error {
+	select {
+	case eventCh <- RangeFeedMessage{
 		RangeFeedEvent: &kvpb.RangeFeedEvent{
 			Metadata: &kvpb.RangeFeedMetadata{
 				Span:            span,
@@ -531,6 +537,10 @@ func sendMetadata(
 			},
 		},
 		RegisteredSpan: span,
+	}:
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
 	}
 }