sql/inspect: serialize inspect progress metadata pushes

Concurrent inspect processors were racing when calling
DistSQLReceiver.Push, which is not concurrency-safe. This commit
introduces a per-processor mutex to serialize all Push calls.

Resolves: #155309, #155311, #155312, #155313, #155314, #155315
Epic: CRDB-30356
Release note: none

Author: spilchen

pkg/sql/inspect/inspect_processor.go

@@ -23,6 +23,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/ctxgroup"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/cockroachdb/cockroach/pkg/util/syncutil"
 	"github.com/cockroachdb/cockroach/pkg/util/tracing"
 	"github.com/cockroachdb/errors"
 	pbtypes "github.com/gogo/protobuf/types"
@@ -71,6 +72,12 @@ type inspectProcessor struct {
 	logger         inspectLogger
 	concurrency    int
 	clock          *hlc.Clock
+	mu             struct {
+		// Guards calls to output.Push because DistSQLReceiver.Push is not
+		// concurrency safe and progress metadata can be emitted from multiple
+		// worker goroutines.
+		syncutil.Mutex
+	}
 }
 
 var _ execinfra.Processor = (*inspectProcessor)(nil)
@@ -310,7 +317,7 @@ func (p *inspectProcessor) sendInspectProgress(
 		},
 	}
 
-	output.Push(nil, meta)
+	p.pushProgressMeta(output, meta)
 	return nil
 }
 
@@ -350,10 +357,21 @@ func (p *inspectProcessor) sendSpanCompletionProgress(
 		},
 	}
 
-	output.Push(nil, meta)
+	p.pushProgressMeta(output, meta)
 	return nil
 }
 
+// pushProgressMeta serializes metadata pushes so only one goroutine interacts
+// with the DistSQL receiver at a time (DistSQLReceiver.Push is not concurrency
+// safe).
+func (p *inspectProcessor) pushProgressMeta(
+	output execinfra.RowReceiver, meta *execinfrapb.ProducerMetadata,
+) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	output.Push(nil, meta)
+}
+
 // newInspectProcessor constructs a new inspectProcessor from the given InspectSpec.
 // It parses the spec to generate a set of inspectCheck factories, sets up the span source,
 // and wires in logging and concurrency controls.