catalog/lease: handle range feed recovery for dropped descriptors

Previously, when range feed recovery was executed by the lease manager,
the operation would handle missing descriptors by removing them.
However, it also needs to gracefully handle dropped descriptors in the
same way, since dropped descriptors cannot be leased. This patch updates
the recovery process and adds a test to confirm that dropped descriptors
are correctly released by the lease manager.

Fixes: #148598
Fixes: #148273
Fixes: #146536

Release note: None

Author: fqazi

pkg/sql/catalog/lease/lease.go

@@ -1817,7 +1817,7 @@ func (m *Manager) refreshSomeLeases(ctx context.Context, refreshAndPurgeAllDescr
 				if _, err := acquireNodeLease(ctx, m, id, AcquireBackground); err != nil {
 					log.Errorf(ctx, "refreshing descriptor: %d lease failed: %s", id, err)
 
-					if errors.Is(err, catalog.ErrDescriptorNotFound) {
+					if errors.Is(err, catalog.ErrDescriptorNotFound) || errors.Is(err, catalog.ErrDescriptorDropped) {
 						// Lease renewal failed due to removed descriptor; Remove this descriptor from cache.
 						if err := purgeOldVersions(
 							ctx, m.storage.db.KV(), id, true /* dropped */, 0 /* minVersion */, m,

pkg/sql/catalog/lease/lease_test.go

@@ -3765,9 +3765,17 @@ func TestLeaseDescriptorRangeFeedFailure(t *testing.T) {
 							p.Params.ExecutionPhase != scop.PostCommitPhase {
 							return nil
 						}
+						foundDescriptorDrop := false
+						// Ensure we have a descriptor drop that will be executed before
+						// we mess with the range feed.
+						for _, op := range p.Stages[stageIdx].EdgeOps {
+							if _, ok := op.(*scop.MarkDescriptorAsDropped); ok {
+								foundDescriptorDrop = true
+							}
+						}
 						// Once this stage completes, we can "resume" the range feed,
 						// so the update is detected.
-						if stageIdx == 0 {
+						if foundDescriptorDrop {
 							rangeFeedResetChan = srv.ApplicationLayer(1).LeaseManager().(*lease.Manager).TestingSetDisableRangeFeedCheckpointFn(true)
 							enableAfterStageKnob.Swap(false)
 							grp.Go(func() error {
@@ -3791,15 +3799,20 @@ func TestLeaseDescriptorRangeFeedFailure(t *testing.T) {
 	// detects a problem.
 	defer srv.ApplicationLayer(1).LeaseManager().(*lease.Manager).TestingSetDisableRangeFeedCheckpointFn(false)
 	firstConn.Exec(t, "CREATE TABLE t1(n int)")
+	firstConn.Exec(t, "CREATE TABLE t2(n int)")
 	require.NoError(t, srv.WaitForFullReplication())
 	tx := secondConn.Begin(t)
 	_, err := tx.Exec("SELECT * FROM t1;")
 	require.NoError(t, err)
+	_, err = tx.Exec("SELECT * FROM t2;")
+	require.NoError(t, err)
 	// This schema change will wait for the connection on
 	// node 1 to release the lease. Because the rangefeed is
 	// disabled it will never know about the new version.
 	enableAfterStageKnob.Store(true)
-	firstConn.Exec(t, "ALTER TABLE t1 ADD COLUMN j INT DEFAULT 64")
+	firstConn.Exec(t, "SET autocommit_before_ddl=false")
+	firstConn.Exec(t, "SET use_declarative_schema_changer='unsafe_always'")
+	firstConn.Exec(t, "ALTER TABLE t1 ADD COLUMN j INT DEFAULT 64; DROP TABLE t2;")
 	_, err = tx.Exec("INSERT INTO t1 VALUES (32)")
 	if err != nil {
 		t.Fatal(err)