kv: use origin timestamp to validate lww semantics

The PR updates the implementation of put, cput, delete, and range delete
to return an error if a replicated write with an origin timestamp would
overwrite a value with a more recent origin or MVCC timestamp.

No explicit version checks are included. In a mixed-version environment,
a SQL writer running 25.1 and writing to a 25.2 KV server will encounter
"condition failed" errors if an insert would overwrite a tombstone and
retry the write until it is upgraded or the LDR PTS ages out. This is
acceptable, as LDR remains in preview and the SQL writer is neither the
default nor the recommended writer.

Release note: none
Epic: CRDB-48647

Author: jeffswenson

pkg/kv/BUILD.bazel

@@ -51,6 +51,7 @@ go_test(
         "admission_test.go",
         "client_test.go",
         "db_test.go",
+        "kv_client_lww_test.go",
         "main_test.go",
         "range_lookup_test.go",
         "txn_external_test.go",

pkg/kv/kv_client_lww_test.go

@@ -0,0 +1,355 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package kv_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/cockroachdb/cockroach/pkg/base"
+	"github.com/cockroachdb/cockroach/pkg/kv"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
+	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
+	"github.com/cockroachdb/cockroach/pkg/util/hlc"
+	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
+	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/cockroachdb/errors"
+	"github.com/stretchr/testify/require"
+)
+
+// unwrapConditionFailedError unwraps an error into a kvpb.ConditionFailedError.
+// It fails the test if there is no error or if the error cannot be unwrapped
+// into the expected type.
+func unwrapConditionFailedError(t *testing.T, err error) *kvpb.ConditionFailedError {
+	t.Helper()
+	require.Error(t, err, "expected an error")
+
+	var conditionFailedErr *kvpb.ConditionFailedError
+	require.True(t, errors.As(err, &conditionFailedErr),
+		"expected error to unwrap to *kvpb.ConditionFailedError, got: %v", err)
+
+	return conditionFailedErr
+}
+
+func TestDeleteWithOriginTimestamp(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	s, _, db := serverutils.StartServer(t, base.TestServerArgs{})
+	defer s.Stopper().Stop(context.Background())
+
+	ctx := context.Background()
+	key := roachpb.Key("lww-test-key")
+	value := []byte("value")
+
+	// Put a value with the current timestamp.
+	require.NoError(t, db.Put(ctx, key, value))
+
+	// Get the current timestamp to use as a reference.
+	gr, err := db.Get(ctx, key)
+	require.NoError(t, err)
+	currentTS := gr.Value.Timestamp
+
+	// Helper function to attempt a delete with a specific origin timestamp
+	attemptDelete := func(ts hlc.Timestamp) error {
+		b := &kv.Batch{}
+		b.Del(key)
+		b.Header.WriteOptions = &kvpb.WriteOptions{
+			OriginTimestamp: ts,
+		}
+		return db.Run(ctx, b)
+	}
+
+	t.Run("OlderTimestamp", func(t *testing.T) {
+		// Delete with an origin timestamp older than the mvcc timestamp should fail
+		err := attemptDelete(currentTS.Prev())
+		condErr := unwrapConditionFailedError(t, err)
+		require.False(t, condErr.OriginTimestampOlderThan.IsEmpty(), "expected OriginTimestampOlderThan to be set")
+		require.Equal(t, currentTS, condErr.OriginTimestampOlderThan, "expected OriginTimestampOlderThan to match current timestamp")
+
+		// Verify the original value still exists
+		gr, err := db.Get(ctx, key)
+		require.NoError(t, err)
+		require.NotNil(t, gr.Value, "expected value to exist")
+		require.Equal(t, value, gr.ValueBytes(), "value should match what was put")
+	})
+
+	t.Run("NewerTimestamp", func(t *testing.T) {
+		// Delete with an origin timestamp newer than the mvcc timestamp should succeed
+		err := attemptDelete(currentTS.Next())
+		require.NoError(t, err)
+
+		// Verify the value was deleted
+		gr, err := db.Get(ctx, key)
+		require.NoError(t, err)
+		require.Nil(t, gr.Value, "expected value to be deleted")
+	})
+}
+
+func TestDelRangeWithOriginTimestamp(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	s, _, db := serverutils.StartServer(t, base.TestServerArgs{})
+	defer s.Stopper().Stop(context.Background())
+
+	ctx := context.Background()
+
+	// Setup test data once for all subtests
+	keys := []roachpb.Key{
+		roachpb.Key("lww-delrange-a"),
+		roachpb.Key("lww-delrange-b"),
+		roachpb.Key("lww-delrange-c"),
+	}
+	startKey := keys[0]
+	endKey := keys[len(keys)-1].Next()
+	value := []byte("value")
+
+	// Put values for all keys
+	for _, key := range keys {
+		require.NoError(t, db.Put(ctx, key, value))
+	}
+
+	// Get the current timestamp to use as a reference
+	gr, err := db.Get(ctx, keys[2])
+	require.NoError(t, err)
+	currentTS := gr.Value.Timestamp
+
+	t.Run("OlderTimestamp", func(t *testing.T) {
+		// DelRange with an origin timestamp older than the mvcc timestamp should fail
+		b := &kv.Batch{}
+		b.DelRange(startKey, endKey, false)
+		b.Header.WriteOptions = &kvpb.WriteOptions{
+			OriginTimestamp: currentTS.Prev(),
+		}
+		err := db.Run(ctx, b)
+		condErr := unwrapConditionFailedError(t, err)
+		require.False(t, condErr.OriginTimestampOlderThan.IsEmpty(), "expected OriginTimestampOlderThan to be set")
+		require.Equal(t, currentTS, condErr.OriginTimestampOlderThan, "expected OriginTimestampOlderThan to match current timestamp")
+
+		// Verify the original values still exist
+		for _, key := range keys {
+			gr, err := db.Get(ctx, key)
+			require.NoError(t, err)
+			require.NotNil(t, gr.Value, "expected value to exist at %s", key)
+			require.Equal(t, value, gr.ValueBytes(), "value should match what was put")
+		}
+	})
+
+	t.Run("NewerTimestamp", func(t *testing.T) {
+		// Delete the range with the newer timestamp
+		b := &kv.Batch{}
+		b.DelRange(startKey, endKey, false)
+		b.Header.WriteOptions = &kvpb.WriteOptions{
+			OriginTimestamp: currentTS.Next(),
+		}
+		err := db.Run(ctx, b)
+		require.NoError(t, err, "unexpected error when deleting range with newer origin timestamp")
+
+		// Verify all values were deleted
+		for _, key := range keys {
+			gr, err := db.Get(ctx, key)
+			require.NoError(t, err)
+			require.Nil(t, gr.Value, "expected value to be deleted at %s", key)
+		}
+	})
+}
+
+func TestPutWithOriginTimestamp(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	s, _, db := serverutils.StartServer(t, base.TestServerArgs{})
+	defer s.Stopper().Stop(context.Background())
+
+	ctx := context.Background()
+	key := roachpb.Key("lww-put-test-key")
+	value1 := []byte("value1")
+	value2 := []byte("value2")
+
+	// Put a value with the current timestamp.
+	require.NoError(t, db.Put(ctx, key, value1))
+
+	// Get the current timestamp to use as a reference.
+	gr, err := db.Get(ctx, key)
+	require.NoError(t, err)
+	currentTS := gr.Value.Timestamp
+
+	// Helper function to attempt a put with a specific origin timestamp
+	attemptPut := func(ts hlc.Timestamp, value []byte) error {
+		b := &kv.Batch{}
+		b.Put(key, value)
+		b.Header.WriteOptions = &kvpb.WriteOptions{
+			OriginTimestamp: ts,
+		}
+		return db.Run(ctx, b)
+	}
+
+	t.Run("OlderTimestamp", func(t *testing.T) {
+		// Put with an origin timestamp older than the mvcc timestamp should fail
+		olderTS := currentTS.Prev()
+		err := attemptPut(olderTS, value2)
+		condErr := unwrapConditionFailedError(t, err)
+		require.False(t, condErr.OriginTimestampOlderThan.IsEmpty(), "expected OriginTimestampOlderThan to be set")
+		require.Equal(t, currentTS, condErr.OriginTimestampOlderThan, "expected OriginTimestampOlderThan to match current timestamp")
+
+		// Verify the original value still exists
+		gr, err := db.Get(ctx, key)
+		require.NoError(t, err)
+		require.NotNil(t, gr.Value, "expected value to exist")
+		require.Equal(t, value1, gr.ValueBytes(), "value should match expected")
+	})
+
+	t.Run("NewerTimestamp", func(t *testing.T) {
+		// Put with an origin timestamp newer than the mvcc timestamp should succeed
+		newerTS := currentTS.Next()
+		err := attemptPut(newerTS, value2)
+		require.NoError(t, err, "unexpected error when putting with newer origin timestamp")
+
+		// Verify the value was updated
+		gr, err := db.Get(ctx, key)
+		require.NoError(t, err)
+		require.NotNil(t, gr.Value, "expected value to exist")
+		require.Equal(t, value2, gr.ValueBytes(), "value should match expected")
+
+		// Get the new timestamp
+		gr, err = db.Get(ctx, key)
+		require.NoError(t, err)
+		updatedTS := gr.Value.Timestamp
+
+		// Try to put with an even newer timestamp
+		newestValue := []byte("newest-value")
+		newestTS := updatedTS.Next()
+		err = attemptPut(newestTS, newestValue)
+		require.NoError(t, err, "unexpected error when putting with newest origin timestamp")
+
+		// Verify the newest value was written
+		gr, err = db.Get(ctx, key)
+		require.NoError(t, err)
+		require.NotNil(t, gr.Value, "expected value to exist")
+		require.Equal(t, newestValue, gr.ValueBytes(), "value should match new value")
+	})
+}
+
+func TestCPutWithOriginTimestamp(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	s, _, db := serverutils.StartServer(t, base.TestServerArgs{})
+	defer s.Stopper().Stop(context.Background())
+
+	ctx := context.Background()
+	key := roachpb.Key("lww-cput-test-key")
+	initialValue := []byte("initial")
+	newValue := []byte("new-value")
+	wrongExpectedValue := []byte("wrong-expected")
+
+	// Put a value with the current timestamp.
+	require.NoError(t, db.Put(ctx, key, initialValue))
+
+	// Get the current timestamp to use as a reference.
+	gr, err := db.Get(ctx, key)
+	require.NoError(t, err)
+	currentTS := gr.Value.Timestamp
+
+	// Helper function to attempt a conditional put with a specific origin timestamp
+	attemptCPut := func(ts hlc.Timestamp, expectedValue, newValue []byte) error {
+		b := &kv.Batch{}
+		b.CPut(key, newValue, expectedValue)
+		b.Header.WriteOptions = &kvpb.WriteOptions{
+			OriginTimestamp: ts,
+		}
+		return db.Run(ctx, b)
+	}
+
+	t.Run("ConditionFailsAndLWWFails", func(t *testing.T) {
+		// Both the condition fails (wrong expected value) and the origin timestamp is older
+		olderTS := currentTS.Prev()
+		err := attemptCPut(olderTS, wrongExpectedValue, newValue)
+		condErr := unwrapConditionFailedError(t, err)
+
+		// The error should indicate the origin timestamp is older
+		require.False(t, condErr.OriginTimestampOlderThan.IsEmpty(), "expected OriginTimestampOlderThan to be set")
+		require.Equal(t, currentTS, condErr.OriginTimestampOlderThan, "expected OriginTimestampOlderThan to match current timestamp")
+
+		// Verify the original value still exists
+		gr, err := db.Get(ctx, key)
+		require.NoError(t, err)
+		require.NotNil(t, gr.Value, "expected value to exist")
+		require.Equal(t, initialValue, gr.ValueBytes(), "value should match initial value")
+	})
+
+	t.Run("ConditionPassesButLWWFails", func(t *testing.T) {
+		// The condition passes (correct expected value) but the origin timestamp is older
+		olderTS := currentTS.Prev()
+		err := attemptCPut(olderTS, initialValue, newValue)
+		condErr := unwrapConditionFailedError(t, err)
+
+		// The error should indicate the origin timestamp is older
+		require.False(t, condErr.OriginTimestampOlderThan.IsEmpty(), "expected OriginTimestampOlderThan to be set")
+		require.Equal(t, currentTS, condErr.OriginTimestampOlderThan, "expected OriginTimestampOlderThan to match current timestamp")
+
+		// Verify the original value still exists
+		gr, err := db.Get(ctx, key)
+		require.NoError(t, err)
+		require.NotNil(t, gr.Value, "expected value to exist")
+		require.Equal(t, initialValue, gr.ValueBytes(), "value should match initial value")
+	})
+
+	t.Run("ConditionFailsButLWWPasses", func(t *testing.T) {
+		// The condition fails (wrong expected value) but the origin timestamp is newer
+		newerTS := currentTS.Next()
+		err := attemptCPut(newerTS, wrongExpectedValue, newValue)
+		condErr := unwrapConditionFailedError(t, err)
+
+		// The error should indicate the condition failed but not that the timestamp is older
+		require.True(t, condErr.OriginTimestampOlderThan.IsEmpty(), "expected OriginTimestampOlderThan to be empty")
+		require.NotNil(t, condErr.ActualValue, "expected ActualValue to be set")
+
+		// Verify the original value still exists
+		gr, err := db.Get(ctx, key)
+		require.NoError(t, err)
+		require.NotNil(t, gr.Value, "expected value to exist")
+		require.Equal(t, initialValue, gr.ValueBytes(), "value should match initial value")
+	})
+
+	t.Run("BothConditionAndLWWPass", func(t *testing.T) {
+		// Create a new key for this test to avoid interference from previous test cases
+		successKey := roachpb.Key("lww-cput-success-key")
+
+		// Put an initial value
+		require.NoError(t, db.Put(ctx, successKey, initialValue))
+
+		// Get the current value and timestamp
+		gr, err := db.Get(ctx, successKey)
+		require.NoError(t, err)
+		require.NotNil(t, gr.Value)
+		currentValue := gr.Value.TagAndDataBytes()
+		currentTS := gr.Value.Timestamp
+
+		// Both the condition passes and the origin timestamp is newer
+		newerTS := currentTS.Next()
+
+		// Use a helper function to perform the CPut with the current value
+		cputErr := func() error {
+			b := &kv.Batch{}
+			b.CPut(successKey, newValue, currentValue)
+			b.Header.WriteOptions = &kvpb.WriteOptions{
+				OriginTimestamp: newerTS,
+			}
+			return db.Run(ctx, b)
+		}()
+
+		require.NoError(t, cputErr, "unexpected error when both condition and timestamp are valid")
+
+		// Verify the value was updated
+		gr, err = db.Get(ctx, successKey)
+		require.NoError(t, err)
+		require.NotNil(t, gr.Value, "expected value to exist")
+		require.Equal(t, newValue, gr.ValueBytes(), "value should match new value")
+	})
+}