Merge #117016 #117096

117016: kv/closedts: remove use and propagation of synthetic timestamp bit  r=nvanbenschoten a=nvanbenschoten

Informs #101938.

This PR removes the handling of synthetic timestamps from the closed timestamp tracker data structure.

It then stops setting the synthetic timestamp bit on the closed timestamps selected for ranges with a `LEAD_FOR_GLOBAL_READS` closed timestamp policy.

This flag has been deprecated since v22.2 and is no longer consulted in uncertainty interval checks or by transaction commit-wait. It does not need to be propagated from evaluating requests to the closed timestamp.

Release note: None

117096: workload/tpcc: optionally disable txn retry loops r=michae2 a=nvanbenschoten

Informs #115191.

This commit adds a `--txn-retries` flag to `tpcc`, allowing users of the workload to disable transaction retry loops for 40001 errors.

For Serializable transactions (controllable with `--isolation-level`), this quickly leads to errors being thrown by the workload. For Read Committed transaction, the workload eventually hits an error due to a lease transfer. Combined with a prototype fix for #61986, Read Committed transactions survive for at least a few minutes without error.

Release note: None

Co-authored-by: Nathan VanBenschoten <nvanbenschoten@gmail.com>

pkg/kv/kvserver/client_split_test.go

@@ -3602,28 +3602,32 @@ func TestStoreRangeSplitAndMergeWithGlobalReads(t *testing.T) {
 	defer log.Scope(t).Close(t)
 
 	// Detect splits and merges over the global read ranges. Assert that the split
-	// and merge transactions commit with synthetic timestamps, and that the
+	// and merge transactions commit with pushed write timestamps, and that the
 	// commit-wait sleep for these transactions is performed before running their
 	// commit triggers instead of run on the kv client. For details on why this is
 	// necessary, see maybeCommitWaitBeforeCommitTrigger.
-	var clock atomic.Value
-	var splitsWithSyntheticTS, mergesWithSyntheticTS int64
+	var clockPtr atomic.Pointer[hlc.Clock]
+	var splits, merges int64
 	respFilter := func(ctx context.Context, ba *kvpb.BatchRequest, br *kvpb.BatchResponse) *kvpb.Error {
+		clock := clockPtr.Load()
+		if clock == nil {
+			return nil
+		}
 		if req, ok := ba.GetArg(kvpb.EndTxn); ok {
 			endTxn := req.(*kvpb.EndTxnRequest)
-			if br.Txn.Status == roachpb.COMMITTED && br.Txn.WriteTimestamp.Synthetic {
+			if br.Txn.Status == roachpb.COMMITTED && br.Txn.MinTimestamp.Less(br.Txn.WriteTimestamp) {
 				if ct := endTxn.InternalCommitTrigger; ct != nil {
 					// The server-side commit-wait sleep should ensure that the commit
 					// triggers are only run after the commit timestamp is below present
 					// time.
-					now := clock.Load().(*hlc.Clock).Now()
+					now := clock.Now()
 					require.True(t, br.Txn.WriteTimestamp.Less(now))
 
 					switch {
 					case ct.SplitTrigger != nil:
-						atomic.AddInt64(&splitsWithSyntheticTS, 1)
+						atomic.AddInt64(&splits, 1)
 					case ct.MergeTrigger != nil:
-						atomic.AddInt64(&mergesWithSyntheticTS, 1)
+						atomic.AddInt64(&merges, 1)
 					}
 				}
 			}
@@ -3655,7 +3659,6 @@ func TestStoreRangeSplitAndMergeWithGlobalReads(t *testing.T) {
 	tdb.Exec(t, `SET CLUSTER SETTING kv.closed_timestamp.target_duration = '20ms'`)
 	tdb.Exec(t, `SET CLUSTER SETTING kv.closed_timestamp.side_transport_interval = '20ms'`)
 	tdb.Exec(t, `SET CLUSTER SETTING kv.rangefeed.closed_timestamp_refresh_interval = '20ms'`)
-	clock.Store(s.Clock())
 	store, err := s.GetStores().(*kvserver.Stores).GetStore(s.GetFirstStoreID())
 	require.NoError(t, err)
 	config.TestingSetupZoneConfigHook(s.Stopper())
@@ -3667,6 +3670,10 @@ func TestStoreRangeSplitAndMergeWithGlobalReads(t *testing.T) {
 	_, pErr := kv.SendWrapped(ctx, store.TestSender(), splitArgs)
 	require.Nil(t, pErr)
 
+	// Set the clock to the store's clock, which also serves to engage the
+	// response filter.
+	clockPtr.Store(s.Clock())
+
 	// Perform a write to the system config span being watched by
 	// the SystemConfigProvider.
 	tdb.Exec(t, "CREATE TABLE foo ()")
@@ -3685,8 +3692,8 @@ func TestStoreRangeSplitAndMergeWithGlobalReads(t *testing.T) {
 		if splitCount != store.Metrics().CommitWaitsBeforeCommitTrigger.Count() {
 			return errors.Errorf("commit wait count is %d", store.Metrics().CommitWaitsBeforeCommitTrigger.Count())
 		}
-		if splitCount != atomic.LoadInt64(&splitsWithSyntheticTS) {
-			return errors.Errorf("num splits is %d", atomic.LoadInt64(&splitsWithSyntheticTS))
+		if splitCount != atomic.LoadInt64(&splits) {
+			return errors.Errorf("num splits is %d", atomic.LoadInt64(&splits))
 		}
 		return nil
 	})
@@ -3703,7 +3710,7 @@ func TestStoreRangeSplitAndMergeWithGlobalReads(t *testing.T) {
 	require.Nil(t, pErr)
 	splitCount++
 	require.Equal(t, splitCount, store.Metrics().CommitWaitsBeforeCommitTrigger.Count())
-	require.Equal(t, splitCount, atomic.LoadInt64(&splitsWithSyntheticTS))
+	require.Equal(t, splitCount, atomic.LoadInt64(&splits))
 
 	repl := store.LookupReplica(roachpb.RKey(splitKey))
 	require.Equal(t, splitKey, repl.Desc().StartKey.AsRawKey())
@@ -3713,7 +3720,7 @@ func TestStoreRangeSplitAndMergeWithGlobalReads(t *testing.T) {
 	_, pErr = kv.SendWrapped(ctx, store.TestSender(), mergeArgs)
 	require.Nil(t, pErr)
 	require.Equal(t, splitCount+1, store.Metrics().CommitWaitsBeforeCommitTrigger.Count())
-	require.Equal(t, int64(1), atomic.LoadInt64(&mergesWithSyntheticTS))
+	require.Equal(t, int64(1), atomic.LoadInt64(&merges))
 
 	repl = store.LookupReplica(roachpb.RKey(splitKey))
 	require.Equal(t, descKey, repl.Desc().StartKey.AsRawKey())

pkg/kv/kvserver/closedts/policy.go

@@ -121,8 +121,7 @@ func TargetForPolicy(
 			leadTimeAtSender = leadTargetOverride
 		}
 
-		// Mark as synthetic, because this time is in the future.
-		res = now.ToTimestamp().Add(leadTimeAtSender.Nanoseconds(), 0).WithSynthetic(true)
+		res = now.ToTimestamp().Add(leadTimeAtSender.Nanoseconds(), 0)
 	default:
 		panic("unexpected RangeClosedTimestampPolicy")
 	}

pkg/kv/kvserver/closedts/policy_test.go

@@ -55,23 +55,21 @@ func TestTargetForPolicy(t *testing.T) {
 			expClosedTSTarget: now.
 				Add((maxClockOffset +
 					millis(275) /* sideTransportPropTime */ +
-					millis(25) /* bufferTime */).Nanoseconds(), 0).
-				WithSynthetic(true),
+					millis(25) /* bufferTime */).Nanoseconds(), 0),
 		},
 		{
 			sideTransportCloseInterval: millis(50),
 			rangePolicy:                roachpb.LEAD_FOR_GLOBAL_READS,
 			expClosedTSTarget: now.
 				Add((maxClockOffset +
 					millis(245) /* raftTransportPropTime */ +
-					millis(25) /* bufferTime */).Nanoseconds(), 0).
-				WithSynthetic(true),
+					millis(25) /* bufferTime */).Nanoseconds(), 0),
 		},
 		{
 			leadTargetOverride:         millis(1234),
 			sideTransportCloseInterval: millis(200),
 			rangePolicy:                roachpb.LEAD_FOR_GLOBAL_READS,
-			expClosedTSTarget:          now.Add(millis(1234).Nanoseconds(), 0).WithSynthetic(true),
+			expClosedTSTarget:          now.Add(millis(1234).Nanoseconds(), 0),
 		},
 	} {
 		t.Run("", func(t *testing.T) {

pkg/kv/kvserver/closedts/sidetransport/receiver_test.go

@@ -62,9 +62,9 @@ func (m *mockStores) getAndClearRecording() []rangeUpdate {
 var ts10 = hlc.Timestamp{WallTime: 10}
 var ts11 = hlc.Timestamp{WallTime: 11}
 var ts12 = hlc.Timestamp{WallTime: 12}
-var ts20 = hlc.Timestamp{WallTime: 20, Synthetic: true}
-var ts21 = hlc.Timestamp{WallTime: 21, Synthetic: true}
-var ts22 = hlc.Timestamp{WallTime: 22, Synthetic: true}
+var ts20 = hlc.Timestamp{WallTime: 20}
+var ts21 = hlc.Timestamp{WallTime: 21}
+var ts22 = hlc.Timestamp{WallTime: 22}
 var laiZero = kvpb.LeaseAppliedIndex(0)
 
 const lai100 = kvpb.LeaseAppliedIndex(100)

pkg/kv/kvserver/closedts/tracker/lockfree_tracker.go

@@ -156,13 +156,13 @@ func (t *lockfreeTracker) Track(ctx context.Context, ts hlc.Timestamp) RemovalTo
 	// b1 and the rest (the "high" ones) go on to create and join b2). But that's
 	// harder to implement.
 	if !initialized || wts <= t1 {
-		return b1.extendAndJoin(ctx, wts, ts.Synthetic)
+		return b1.extendAndJoin(ctx, wts)
 	}
 
 	// We know that b1 < wts. We can technically join either bucket, but we always
 	// prefer b2 in order to let b1 drain as soon as possible (at which point
 	// we'll be able to create a new bucket).
-	return b2.extendAndJoin(ctx, wts, ts.Synthetic)
+	return b2.extendAndJoin(ctx, wts)
 }
 
 // Untrack is part of the Tracker interface.
@@ -176,7 +176,6 @@ func (t *lockfreeTracker) Untrack(ctx context.Context, tok RemovalToken) {
 	if b.refcnt == 0 {
 		// Reset the bucket, so that future Track() calls can create a new one.
 		b.ts = 0
-		b.synthetic = 0
 		// If we reset b1, swap the pointers, so that, if b2 is currently
 		// initialized, it becomes b1. If a single bucket is initialized, we want it
 		// to be b1.
@@ -196,9 +195,8 @@ func (t *lockfreeTracker) LowerBound(ctx context.Context) hlc.Timestamp {
 		return hlc.Timestamp{}
 	}
 	return hlc.Timestamp{
-		WallTime:  ts,
-		Logical:   0,
-		Synthetic: t.b1.isSynthetic(),
+		WallTime: ts,
+		Logical:  0,
 	}
 }
 
@@ -213,9 +211,8 @@ func (t *lockfreeTracker) Count() int {
 // A bucket can be initialized or uninitialized. It's initialized when the ts is
 // set.
 type bucket struct {
-	ts        int64 // atomic, nanos
-	refcnt    int32 // atomic
-	synthetic int32 // atomic
+	ts     int64 // atomic, nanos
+	refcnt int32 // atomic
 }
 
 func (b *bucket) String() string {
@@ -234,18 +231,12 @@ func (b *bucket) timestamp() (int64, bool) {
 	return ts, ts != 0
 }
 
-// isSynthetic returns true if the bucket's timestamp (i.e. the bucket's lower
-// bound) should be considered a synthetic timestamp.
-func (b *bucket) isSynthetic() bool {
-	return atomic.LoadInt32(&b.synthetic) != 0
-}
-
 // extendAndJoin extends the bucket downwards (if necessary) so that its
 // timestamp is <= ts, and then adds a timestamp to the bucket. It returns a
 // token to be used for removing the timestamp from the bucket.
 //
 // If the bucket it not initialized, it will be initialized to ts.
-func (b *bucket) extendAndJoin(ctx context.Context, ts int64, synthetic bool) lockfreeToken {
+func (b *bucket) extendAndJoin(ctx context.Context, ts int64) lockfreeToken {
 	// Loop until either we set the bucket's timestamp, or someone else sets it to
 	// an even lower value.
 	var t int64
@@ -258,16 +249,6 @@ func (b *bucket) extendAndJoin(ctx context.Context, ts int64, synthetic bool) lo
 			break
 		}
 	}
-	// If we created the bucket, then we dictate if its lower bound will be
-	// considered a synthetic timestamp or not. It's possible that we're now
-	// inserting a synthetic timestamp into the bucket but, over time, a higher
-	// non-synthetic timestamp joins. Or, that a lower non-synthetic timestamp
-	// joins. In either case, the bucket will remain "synthetic" although it'd be
-	// correct to make it non-synthetic. We don't make an effort to keep the
-	// synthetic bit up to date within a bucket.
-	if t == 0 && synthetic {
-		atomic.StoreInt32(&b.synthetic, 1)
-	}
 	atomic.AddInt32(&b.refcnt, 1)
 	return lockfreeToken{b: b}
 }

pkg/kv/kvserver/closedts/tracker/tracker.go

@@ -81,11 +81,6 @@ type Tracker interface {
 	// The returned timestamp might be smaller than the lowest timestamp ever
 	// inserted into the set. Implementations are allowed to round timestamps
 	// down.
-	//
-	// Synthetic timestamps: The Tracker doesn't necessarily track synthetic /
-	// physical timestamps precisely; the only guarantee implementations need to
-	// make is that, if no synthethic timestamp is inserted into the tracked set
-	// for a while, eventually the LowerBound value will not be synthetic.
 	LowerBound(context.Context) hlc.Timestamp
 
 	// Count returns the current size of the tracked set.

pkg/kv/kvserver/closedts/tracker/tracker_test.go

@@ -79,19 +79,6 @@ func testTracker(ctx context.Context, t *testing.T, tr Tracker) {
 	}
 	tr.Untrack(ctx, tok30)
 	require.True(t, tr.LowerBound(ctx).IsEmpty())
-
-	// Check that synthetic timestamps are tracked as such.
-	synthTS := hlc.Timestamp{
-		WallTime:  10,
-		Synthetic: true,
-	}
-	tok := tr.Track(ctx, synthTS)
-	require.Equal(t, synthTS, tr.LowerBound(ctx))
-	// Check that after the Tracker is emptied, lowerbounds are not synthetic any
-	// more.
-	tr.Untrack(ctx, tok)
-	tr.Track(ctx, ts(10))
-	require.Equal(t, ts(10), tr.LowerBound(ctx))
 }
 
 // Test the tracker by throwing random requests at it. We verify that, at all

pkg/kv/kvserver/kvserverpb/lease_status.go

@@ -60,10 +60,5 @@ func (st LeaseStatus) Expiration() hlc.Timestamp {
 // Until a new lease is acquired, all writes will be pushed into this last
 // nanosecond of the lease.
 func (st LeaseStatus) ClosedTimestampUpperBound() hlc.Timestamp {
-	// HACK(andrei): We declare the lease expiration to be synthetic by fiat,
-	// because it frequently is synthetic even though currently it's not marked
-	// as such. See the TODO in Timestamp.Add() about the work remaining to
-	// properly mark these timestamps as synthetic. We need to make sure it's
-	// synthetic here so that the results of Backwards() can be synthetic.
-	return st.Expiration().WithSynthetic(true).WallPrev()
+	return st.Expiration().WallPrev()
 }

pkg/kv/kvserver/replica_proposal_buf_test.go

@@ -924,8 +924,7 @@ func TestProposalBufferClosedTimestamp(t *testing.T) {
 	nowMinusTwiceClosedLag := nowTS.Add(-2*closedts.TargetDuration.Get(&st.SV).Nanoseconds(), 0)
 	nowPlusGlobalReadLead := nowTS.Add((maxOffset +
 		275*time.Millisecond /* sideTransportPropTime */ +
-		25*time.Millisecond /* bufferTime */).Nanoseconds(), 0).
-		WithSynthetic(true)
+		25*time.Millisecond /* bufferTime */).Nanoseconds(), 0)
 	expiredLeaseTimestamp := nowTS.Add(-1000, 0)
 	someClosedTS := nowTS.Add(-2000, 0)
 

pkg/workload/tpcc/delivery.go

@@ -16,7 +16,6 @@ import (
 	"fmt"
 	"strings"
 
-	crdbpgx "github.com/cockroachdb/cockroach-go/v2/crdb/crdbpgxv5"
 	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 	"github.com/cockroachdb/cockroach/pkg/workload"
 	"github.com/cockroachdb/errors"
@@ -87,8 +86,8 @@ func (del *delivery) run(ctx context.Context, wID int) (interface{}, error) {
 	oCarrierID := rng.Intn(10) + 1
 	olDeliveryD := timeutil.Now()
 
-	err := crdbpgx.ExecuteTx(
-		ctx, del.mcp.Get(), pgx.TxOptions{},
+	err := del.config.executeTx(
+		ctx, del.mcp.Get(),
 		func(tx pgx.Tx) error {
 			// 2.7.4.2. For each district:
 			dIDoIDPairs := make(map[int]int)

pkg/workload/tpcc/new_order.go

@@ -17,7 +17,6 @@ import (
 	"strings"
 	"time"
 
-	crdbpgx "github.com/cockroachdb/cockroach-go/v2/crdb/crdbpgxv5"
 	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 	"github.com/cockroachdb/cockroach/pkg/workload"
 	"github.com/cockroachdb/errors"
@@ -211,8 +210,8 @@ func (n *newOrder) run(ctx context.Context, wID int) (interface{}, error) {
 
 	d.oEntryD = timeutil.Now()
 
-	err := crdbpgx.ExecuteTx(
-		ctx, n.mcp.Get(), pgx.TxOptions{},
+	err := n.config.executeTx(
+		ctx, n.mcp.Get(),
 		func(tx pgx.Tx) error {
 			// Select the district tax rate and next available order number, bumping it.
 			var dNextOID int

pkg/workload/tpcc/order_status.go

@@ -14,7 +14,6 @@ import (
 	"context"
 	"time"
 
-	crdbpgx "github.com/cockroachdb/cockroach-go/v2/crdb/crdbpgxv5"
 	"github.com/cockroachdb/cockroach/pkg/util/bufalloc"
 	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 	"github.com/cockroachdb/cockroach/pkg/workload"
@@ -133,8 +132,8 @@ func (o *orderStatus) run(ctx context.Context, wID int) (interface{}, error) {
 		d.cID = o.config.randCustomerID(rng)
 	}
 
-	if err := crdbpgx.ExecuteTx(
-		ctx, o.mcp.Get(), pgx.TxOptions{},
+	if err := o.config.executeTx(
+		ctx, o.mcp.Get(),
 		func(tx pgx.Tx) error {
 			// 2.6.2.2 explains this entire transaction.
 

pkg/workload/tpcc/payment.go

@@ -15,7 +15,6 @@ import (
 	"fmt"
 	"time"
 
-	crdbpgx "github.com/cockroachdb/cockroach-go/v2/crdb/crdbpgxv5"
 	"github.com/cockroachdb/cockroach/pkg/util/bufalloc"
 	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 	"github.com/cockroachdb/cockroach/pkg/workload"
@@ -189,8 +188,8 @@ func (p *payment) run(ctx context.Context, wID int) (interface{}, error) {
 		d.cID = p.config.randCustomerID(rng)
 	}
 
-	if err := crdbpgx.ExecuteTx(
-		ctx, p.mcp.Get(), pgx.TxOptions{},
+	if err := p.config.executeTx(
+		ctx, p.mcp.Get(),
 		func(tx pgx.Tx) error {
 			var wName, dName string
 			// Update warehouse with payment

pkg/workload/tpcc/stock_level.go

@@ -13,7 +13,6 @@ package tpcc
 import (
 	"context"
 
-	crdbpgx "github.com/cockroachdb/cockroach-go/v2/crdb/crdbpgxv5"
 	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 	"github.com/cockroachdb/cockroach/pkg/workload"
 	"github.com/jackc/pgx/v5"
@@ -96,8 +95,8 @@ func (s *stockLevel) run(ctx context.Context, wID int) (interface{}, error) {
 		dID:       rng.Intn(10) + 1,
 	}
 
-	if err := crdbpgx.ExecuteTx(
-		ctx, s.mcp.Get(), pgx.TxOptions{},
+	if err := s.config.executeTx(
+		ctx, s.mcp.Get(),
 		func(tx pgx.Tx) error {
 			var dNextOID int
 			if err := s.selectDNextOID.QueryRowTx(