catalog/lease: disallow leases until range feed starts

Previusly, the lease manager allowed descriptors to be leased before the
range feed was fully initialized. This was problematic because if an
descriptor was updated before the range feed is started we would have no
way of knowing. For example an early query during startup could fetch
the system database descriptor before the range feed starts, and this
could kept forever. If an upgrade bumped the descriptor version between
the initial fetch and start of range feed then we could end up with a
node stuck with the sale version forever. To address this, this patch
forces any early lease acqusitions to wait for the range feed to start
up.

Fixes: cockroachdb#139837
Fixes: cockroachdb#139405
Fixes: cockroachdb#139356
Fixes: cockroachdb#139100

Release note: None

pkg/sql/catalog/lease/lease.go

@@ -1125,6 +1125,10 @@ type Manager struct {
 
 		// rangeFeed current range feed on system.descriptors.
 		rangeFeed *rangefeed.RangeFeed
+
+		// Indicates that initialization is complete and leases
+		// can be acquired now.
+		initComplete bool
 	}
 
 	// closeTimeStamp for the range feed, which is the timestamp
@@ -1154,6 +1158,10 @@ type Manager struct {
 	// detected by the lease manager. Once this count is incremented new data
 	// is available.
 	leaseGeneration atomic.Int64
+
+	// waitForInit used when the lease manager is starting up prevent leases from
+	// being acquired before the range feed.
+	waitForInit *sync.Cond
 }
 
 const leaseConcurrencyLimit = 5
@@ -1248,6 +1256,7 @@ func NewLeaseManager(
 	lm.storage.writer = newKVWriter(codec, db.KV(), keys.LeaseTableID, settingsWatcher, lm)
 	lm.stopper.AddCloser(lm.sem.Closer("stopper"))
 	lm.mu.descriptors = make(map[descpb.ID]*descriptorState)
+	lm.waitForInit = sync.NewCond(&lm.mu)
 	// We are going to start the range feed later when RefreshLeases
 	// is invoked inside pre-start. So, that guarantees all range feed events
 	// that will be generated will be after the current time. So, historical
@@ -1603,8 +1612,19 @@ func (m *Manager) isDescriptorStateEmpty(id descpb.ID) bool {
 	return len(st.mu.active.data) == 0
 }
 
+// maybeWaitForInitLocked waits for the lease manager to startup.
+func (m *Manager) maybeWaitForInitLocked() {
+	// If initialization is complete we have nothing
+	// else to do.
+	if m.mu.initComplete {
+		return
+	}
+	m.waitForInit.Wait()
+}
+
 // If create is set, cache and stopper need to be set as well.
 func (m *Manager) findDescriptorState(id descpb.ID, create bool) *descriptorState {
+	m.maybeWaitForInitLocked()
 	m.mu.Lock()
 	defer m.mu.Unlock()
 	t := m.mu.descriptors[id]
@@ -1622,6 +1642,7 @@ func (m *Manager) findDescriptorState(id descpb.ID, create bool) *descriptorStat
 func (m *Manager) RefreshLeases(ctx context.Context, s *stop.Stopper, db *kv.DB) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
+	defer m.waitForInit.Broadcast()
 	m.watchForUpdates(ctx)
 	_ = s.RunAsyncTask(ctx, "refresh-leases", func(ctx context.Context) {
 		for {
@@ -1733,6 +1754,7 @@ func (m *Manager) RefreshLeases(ctx context.Context, s *stop.Stopper, db *kv.DB)
 			}
 		}
 	})
+	m.mu.initComplete = true
 }
 
 // GetLeaseGeneration provides an integer which will change whenever new