crosscluster/logical: zero catchup/initial scan metrics on flow shutdown

Informs #147620

Release note: none

Author: msbutler

pkg/crosscluster/logical/logical_replication_job.go

@@ -321,6 +321,15 @@ func (r *logicalReplicationResumer) ingest(
 		return err
 	}
 
+	defer func() {
+		if l := payload.MetricsLabel; l != "" {
+			metrics.LabeledScanningRanges.Update(map[string]string{"label": l}, 0)
+			metrics.LabeledCatchupRanges.Update(map[string]string{"label": l}, 0)
+		}
+		metrics.ScanningRanges.Update(0)
+		metrics.CatchupRanges.Update(0)
+	}()
+
 	err = ctxgroup.GoAndWait(ctx, execPlan, replanner, startHeartbeat)
 	if errors.Is(err, sql.ErrPlanChanged) {
 		metrics.ReplanCount.Inc(1)