span: make MultiFrontier.Frontiers return read-only frontiers

Previously, `MultiFrontier`'s `Frontiers` method returned an iterator
over the actual backing frontiers. This was unsafe because it meant
callers could modify the underlying frontiers and cause the internal
heap invariant to be violated. This change restricts the frontiers
returned by the `Frontiers` iterator to be read-only.

Release note: None

Author: andyyang890

pkg/ccl/changefeedccl/resolvedspan/frontier.go

@@ -437,8 +437,8 @@ type maybeTablePartitionedFrontier interface {
 	// Frontiers returns an iterator over the table ID and sub-frontiers
 	// being tracked by the frontier. If the frontier is not tracking
 	// on a per-table basis, the iterator will return a single frontier
-	// with an ID of 0.
-	Frontiers() iter.Seq2[descpb.ID, span.Frontier]
+	// with descpb.InvalidID.
+	Frontiers() iter.Seq2[descpb.ID, span.ReadOnlyFrontier]
 }
 
 var _ maybeTablePartitionedFrontier = (*span.MultiFrontier[descpb.ID])(nil)
@@ -456,9 +456,9 @@ type notTablePartitionedFrontier struct {
 var _ maybeTablePartitionedFrontier = notTablePartitionedFrontier{}
 
 // Frontiers implements maybeTablePartitionedFrontier.
-func (f notTablePartitionedFrontier) Frontiers() iter.Seq2[descpb.ID, span.Frontier] {
-	return func(yield func(descpb.ID, span.Frontier) bool) {
-		yield(0, f.spanFrontier)
+func (f notTablePartitionedFrontier) Frontiers() iter.Seq2[descpb.ID, span.ReadOnlyFrontier] {
+	return func(yield func(descpb.ID, span.ReadOnlyFrontier) bool) {
+		yield(descpb.InvalidID, f.spanFrontier)
 	}
 }
 

pkg/ccl/changefeedccl/resolvedspan/frontier_test.go

@@ -176,7 +176,7 @@ type frontier interface {
 	InBackfill(jobspb.ResolvedSpan) bool
 	AtBoundary() (bool, jobspb.ResolvedSpan_BoundaryType, hlc.Timestamp)
 	All() iter.Seq[jobspb.ResolvedSpan]
-	Frontiers() iter.Seq2[descpb.ID, span.Frontier]
+	Frontiers() iter.Seq2[descpb.ID, span.ReadOnlyFrontier]
 }
 
 func testBackfillSpan(

pkg/util/span/multi_frontier.go

@@ -241,9 +241,22 @@ func (f *MultiFrontier[P]) String() string {
 	return buf.String()
 }
 
-// Frontiers returns an iterator over the sub-frontiers.
-func (f *MultiFrontier[P]) Frontiers() iter.Seq2[P, Frontier] {
-	return func(yield func(P, Frontier) bool) {
+// ReadOnlyFrontier is a subset of Frontier with only the methods
+// that are read-only.
+type ReadOnlyFrontier interface {
+	Frontier() hlc.Timestamp
+	PeekFrontierSpan() roachpb.Span
+	Entries() iter.Seq2[roachpb.Span, hlc.Timestamp]
+	SpanEntries(span roachpb.Span) iter.Seq2[roachpb.Span, hlc.Timestamp]
+	Len() int
+	String() string
+}
+
+var _ ReadOnlyFrontier = Frontier(nil)
+
+// Frontiers returns an iterator over the sub-frontiers (with read-only access).
+func (f *MultiFrontier[P]) Frontiers() iter.Seq2[P, ReadOnlyFrontier] {
+	return func(yield func(P, ReadOnlyFrontier) bool) {
 		f.mu.Lock()
 		defer f.mu.Unlock()