kvpb: fix NPE in test-only assertion

This assertion was recently added but didn't account for the case when a
batch was being evaluated as 1PC, in which case the transaction
has been stripped off.

Fixes #148788
Release note: None

Author: stevendanna

pkg/kv/kvpb/api.go

@@ -2562,6 +2562,12 @@ func validateExclusionTimestampForBatch(ts hlc.Timestamp, h Header) error {
 		return nil
 	}
 
+	// If we don't have a transaction, we can't know whether
+	// CanForwardReadTimestamp is valid or not.
+	if h.Txn == nil {
+		return nil
+	}
+
 	// Unless the IsoLevel allows per-statement read snapshots,
 	// CanForwardReadTimestamp implies we haven't served a read, so it makes no
 	// sense that ExpectExclusionSince would be set since it is the result of a

pkg/kv/kvpb/api_test.go

@@ -11,6 +11,7 @@ import (
 	"testing"
 
 	"github.com/cockroachdb/cockroach/pkg/kv/kvnemesis/kvnemesisutil"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/concurrency/isolation"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/concurrency/lock"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/storage/enginepb"
@@ -384,6 +385,35 @@ func TestGetValidate(t *testing.T) {
 	})
 }
 
+func TestDeleteValidate(t *testing.T) {
+	t.Run("ExpectExclusionSinceWithCanForwardReadTimestamp", func(t *testing.T) {
+		delReq := &DeleteRequest{
+			ExpectExclusionSince: hlc.Timestamp{WallTime: 1},
+		}
+		require.Error(t, delReq.Validate(Header{
+			CanForwardReadTimestamp: true,
+			Txn:                     &roachpb.Transaction{},
+		}))
+	})
+	t.Run("ExpectExclusionSinceWithCanForwardReadTimestampAtWeakerIsolation", func(t *testing.T) {
+		delReq := &DeleteRequest{
+			ExpectExclusionSince: hlc.Timestamp{WallTime: 1},
+		}
+		txn := &roachpb.Transaction{}
+		txn.IsoLevel = isolation.ReadCommitted
+		require.NoError(t, delReq.Validate(Header{
+			CanForwardReadTimestamp: true,
+			Txn:                     txn,
+		}))
+	})
+	t.Run("ExpectExclusionSinceWithCanForwardReadTimestampButNoTxn", func(t *testing.T) {
+		delReq := &DeleteRequest{
+			ExpectExclusionSince: hlc.Timestamp{WallTime: 1},
+		}
+		require.NoError(t, delReq.Validate(Header{CanForwardReadTimestamp: true}))
+	})
+}
+
 func TestRequestHeaderRoundTrip(t *testing.T) {
 	var seq kvnemesisutil.Container
 	seq.Set(123)