Merge #34066

34066: rangefeed: Fix iterator leak when stopping cluster r=danhhz a=mrtracy

When converting rangefeeds to use a per-registration output loop in
issue #33557, I accidentally removed two places where the passed
"catchupIter" for a registration was being closed if the processor is
terminated before the registration is actually completed. This happened
due to some work-in-progress confusion where the catchupIter was
removed, but eventually restored. This was caught promptly by our
regular stress tests.

Fixes #34051

Release note: None

Co-authored-by: Matt Tracy <[email protected]>

pkg/storage/rangefeed/processor.go

@@ -219,6 +219,9 @@ func (p *Processor) Start(stopper *stop.Stopper, rtsIter engine.SimpleIterator)
 					}
 				}
 				if err := stopper.RunAsyncTask(ctx, "rangefeed: output loop", runOutputLoop); err != nil {
+					if r.catchupIter != nil {
+						r.catchupIter.Close() // clean up
+					}
 					r.disconnect(roachpb.NewError(err))
 					p.reg.Unregister(&r)
 				}
@@ -354,6 +357,9 @@ func (p *Processor) Register(
 	select {
 	case p.regC <- r:
 	case <-p.stoppedC:
+		if catchupIter != nil {
+			catchupIter.Close() // clean up
+		}
 		// errC has a capacity of 1. If it is already full, we don't need to send
 		// another error.
 		select {