catalog/lease: fix error handling for purgeOldVersions

Previously, when purging old versions we would acquire a lease on the
latest version, which was introduced when we added logic for acquiring
leases on the previous version. The logic to acquire the previous
version would clear the error, preventing errors from correctly
surfacing and causing issues with dropped / offline descriptors. To
address this, ensure the acquisition logic for the previous version is
only executed if a clean up will occur.

This patch also enhances logging for error scenarios where we could
fail to acquire new leases.

Informs: #156176
Release note: None

Author: fqazi

pkg/sql/catalog/lease/descriptor_state.go

@@ -138,7 +138,7 @@ func (t *descriptorState) findForTimestampImpl(
 
 	// If we have the initial version of the descriptor, and it satisfies the read
 	// timestamp, then the object was just created. We can confirm it satisfies
-	// the request, by executing findForTimestampImpl with the readTimestamp instead.
+	// the request by executing findForTimestampImpl with the readTimestamp instead.
 	if oldest := t.mu.active.findOldest(); hasDifferentReadTimeStamp &&
 		oldest != nil &&
 		oldest.GetVersion() == 1 &&

pkg/sql/catalog/lease/lease.go

@@ -1274,7 +1274,7 @@ func (m *Manager) purgeOldVersions(
 		retry.Options{
 			MaxDuration: time.Second * 30}); r.Next(); {
 		// Acquire a refcount on the descriptor on the latest version to maintain an
-		// active lease, so that it doesn't get released when removeInactives()
+		// active lease so that it doesn't get released when removeInactives()
 		// is called below. Release this lease after calling removeInactives().
 		desc, _, err = t.findForTimestamp(ctx, TimestampToReadTimestamp(m.storage.clock.Now()))
 		if err == nil || !errors.Is(err, errRenewLease) {
@@ -1290,7 +1290,7 @@ func (m *Manager) purgeOldVersions(
 		// Assert this should never happen due to a fixed expiration, since the range
 		// feed is responsible for purging old versions and acquiring new versions.
 		if newest.hasFixedExpiration() {
-			return errors.AssertionFailedf("the latest version of the descriptor has" +
+			return errors.AssertionFailedf("the latest version of the descriptor has " +
 				"a fixed expiration, this should never happen")
 		}
 		// Otherwise, we ran into some type of transient issue, where the sqllivness
@@ -1307,9 +1307,10 @@ func (m *Manager) purgeOldVersions(
 		err = nil
 	}
 
-	// Optionally, acquire the refcount on the previous version.
+	// Optionally, acquire the refcount on the previous version for the locked
+	// leasing mode.
 	acquireLeaseOnPrevious := func() error {
-		if dropped || !LockedLeaseTimestamp.Get(&m.storage.settings.SV) {
+		if !LockedLeaseTimestamp.Get(&m.storage.settings.SV) {
 			return nil
 		}
 		var handles []*closeTimeStampHandle
@@ -1376,11 +1377,12 @@ func (m *Manager) purgeOldVersions(
 		return gatheredErrors
 	}
 
-	if err = acquireLeaseOnPrevious(); err != nil {
-		log.Dev.Errorf(ctx, "unable to acquire lease on previous version of descriptor: %s", err)
-	}
-
 	if isInactive := catalog.HasInactiveDescriptorError(err); err == nil || isInactive {
+		// If previous versions are released, then acquire a lease on the previous
+		// version for the locked leasing mode.
+		if acquirePreviousErr := acquireLeaseOnPrevious(); acquirePreviousErr != nil {
+			log.Dev.Errorf(ctx, "unable to acquire lease on previous version of descriptor: %s", acquirePreviousErr)
+		}
 		removeInactives(isInactive)
 		if desc != nil {
 			t.release(ctx, desc)
@@ -2325,12 +2327,25 @@ func (m *Manager) StartRefreshLeasesTask(ctx context.Context, s *stop.Stopper, d
 	defer m.initComplete.Swap(true)
 	m.watchForUpdates(ctx)
 	_ = s.RunAsyncTask(ctx, "refresh-leases", func(ctx context.Context) {
+		defer func() {
+			if err := recover(); err != nil {
+				log.Dev.Warningf(ctx, "panic in refresh-leases: %v", err)
+				panic(err)
+			}
+		}()
+
 		for {
 			select {
 			case id := <-m.descDelCh:
 				// Descriptor is marked as deleted, so mark it for deletion or
 				// remove it if it's no longer in use.
 				_ = s.RunAsyncTask(ctx, "purgeOldVersionsOrAcquireInitialVersion deleted descriptor", func(ctx context.Context) {
+					defer func() {
+						if err := recover(); err != nil {
+							log.Dev.Warningf(ctx, "panic in purgeOldVersionsOrAcquireInitialVersion deleted descriptor: %v", err)
+							panic(err)
+						}
+					}()
 					// Once the descriptor is purged notify that some change has occurred.
 					defer m.leaseGeneration.Add(1)
 					state := m.findNewest(id)
@@ -2412,6 +2427,12 @@ func (m *Manager) StartRefreshLeasesTask(ctx context.Context, s *stop.Stopper, d
 				// of increased latency right as the descriptor has been committed.
 				if now := db.Clock().Now(); now.Less(desc.GetModificationTime()) {
 					_ = s.RunAsyncTask(ctx, "wait to purgeOldVersionsOrAcquireInitialVersion", func(ctx context.Context) {
+						defer func() {
+							if err := recover(); err != nil {
+								log.Dev.Warningf(ctx, "panic in wait to purgeOldVersionsOrAcquireInitialVersion: %v", err)
+								panic(err)
+							}
+						}()
 						toWait := time.Duration(desc.GetModificationTime().WallTime - now.WallTime)
 						select {
 						case <-time.After(toWait):