logical: fix tombstone lease expiration check

The tombstone updater has the wrong condition when checking for expired
leases. It refreshes the descriptor when the hlc comes before the
expiration, when it should referesh the descriptor if the hlc is equal
to or after expiration.

In theory this could lead to the tombstone updater using a stale
descriptor. It's unlikely this can cause a problem in practice. LDR
periodically releases leases whenever it processes a checkpoint and the
tombstone updater refreshes the lease every time it is called.

Fixes: #159860
Release note: none

Author: jeffswenson

pkg/crosscluster/logical/tombstone_updater.go

@@ -153,9 +153,13 @@ func (tu *tombstoneUpdater) addToBatch(
 	)
 }
 
+func (tu *tombstoneUpdater) hasValidLease(ctx context.Context, now hlc.Timestamp) bool {
+	return tu.leased.descriptor != nil && tu.leased.descriptor.Expiration(ctx).After(now)
+}
+
 func (tu *tombstoneUpdater) getDeleter(ctx context.Context, txn *kv.Txn) (row.Deleter, error) {
 	timestamp := txn.ProvisionalCommitTimestamp()
-	if tu.leased.descriptor == nil || !timestamp.After(tu.leased.descriptor.Expiration(ctx)) {
+	if !tu.hasValidLease(ctx, timestamp) {
 		tu.ReleaseLeases(ctx)
 
 		var err error

pkg/crosscluster/logical/tombstone_updater_test.go

@@ -16,11 +16,44 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/lease"
 	"github.com/cockroachdb/cockroach/pkg/sql/isql"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
+	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/stretchr/testify/require"
 )
 
+type fakeDescriptor struct {
+	lease.LeasedDescriptor
+	expiration hlc.Timestamp
+}
+
+func (e *fakeDescriptor) Expiration(ctx context.Context) hlc.Timestamp {
+	return e.expiration
+}
+
+func TestTombstoneDescriptorLease(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	ctx := context.Background()
+	clock := hlc.NewClockForTesting(nil)
+
+	before := clock.Now()
+	expiration := before.Add(1, 0)
+	after := expiration.Add(1, 0)
+
+	tu := &tombstoneUpdater{}
+	require.False(t, tu.hasValidLease(ctx, before))
+
+	tu.leased.descriptor = &fakeDescriptor{
+		expiration: expiration,
+	}
+
+	require.True(t, tu.hasValidLease(ctx, before))
+	require.False(t, tu.hasValidLease(ctx, expiration))
+	require.False(t, tu.hasValidLease(ctx, after))
+}
+
 func TestTombstoneUpdaterSetsOriginID(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)