Merge #159861

159861: logical: fix tombstone lease expiration check r=jeffswenson a=jeffswenson

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

Co-authored-by: Jeff Swenson <[email protected]>

Author: craig[bot]

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)