Merge #150141 #154662 #154863

150141: kvnemesis: add network partitions r=stevendanna a=miraradeva

**rpc: extend the test partitioner to dynamic partitions**

The `Partitioner` is a testing tool that uses the unary and stream client interceptor knobs to induce network partitions between nodes at the gRPC level.

This commit extends the `Partitioner` to be able to add and remove partitions dynamically during a test, and to be able to create asymmetric partitions.

Release note: None

----

**kvnemesis: add network partitions**

This commit adds support for symmetric, asymmetric, partial and full partitions to kvnemesis, using the `rpc.Partitioner`.

Release note: None

----

**kvnemesis: introduce safety and liveness modes**

Currently, all kvnemesis test variants test both safety and liveness properties: safety in the sense of validating serializability, and liveness in the sense of failing if unavailability is encountered.

With the introduction of various faults (e.g. network partitions), kvnemesis tests can continue to operate the same way only if we carefully craft the fault patterns in order to avoid unavailability. While this is valuable, it is also important to test for safety testing in the presence of unavailability.

This commit introduces two new modes of running kvnemesis tests, both of which still validate serializability:
- Safety mode: all fault patterns are allowed; unavailability errors are   ignored.
- Liveness mode: faults are introduced carefully to ensure a  well-connected quorum is preserved; unavailability errors fail the  test.

Fixes: #114814

Release note: None

154662: sql: fix transaction diagnostics for implicit txns r=kyle-a-wong a=kyle-a-wong

Fixes transaction diagnostic collection to work for implicit transactions

Epic: [CRDB-53541](https://cockroachlabs.atlassian.net/browse/CRDB-53541)
Release note: None

----

This is a stacked PR, only the last commit needs to be reviewed

154863: changefeedccl: prevent panic in per-table PTS mixed-version r=asg0451 a=aerfrei

Fixes a panic that occurred when evaluating per-table protected timestamp settings. The fix adds a nil check on the change aggregator spec ProgressConfig before accessing the per-table PTS flag.

Fixes: #154830
Epic: CRDB-1421
Release note: None

Co-authored-by: Mira Radeva <[email protected]>
Co-authored-by: Kyle Wong <[email protected]>
Co-authored-by: Aerin Freilich <[email protected]>

Author: 4 people

pkg/ccl/changefeedccl/changefeed_processors.go

@@ -1975,7 +1975,7 @@ func (cf *changeFrontier) manageProtectedTimestamps(
 		return cf.frontier.Frontier()
 	}()
 
-	if cf.spec.ProgressConfig.PerTableProtectedTimestamps {
+	if cf.spec.ProgressConfig != nil && cf.spec.ProgressConfig.PerTableProtectedTimestamps {
 		return cf.managePerTableProtectedTimestamps(ctx, txn, &ptsEntries, highwater)
 	}
 

pkg/kv/kvclient/kvcoord/dist_sender_server_test.go

@@ -4229,10 +4229,7 @@ func TestProxyTracing(t *testing.T) {
 	ctx := context.Background()
 
 	testutils.RunValues(t, "lease-type", roachpb.TestingAllLeaseTypes(), func(t *testing.T, leaseType roachpb.LeaseType) {
-		if leaseType == roachpb.LeaseExpiration {
-			skip.UnderRace(t, "too slow")
-			skip.UnderDeadlock(t, "too slow")
-		} else if leaseType == roachpb.LeaseEpoch {
+		if leaseType == roachpb.LeaseEpoch {
 			// With epoch leases this test doesn't work reliably. It passes
 			// in cases where it should fail and fails in cases where it
 			// should pass.
@@ -4241,6 +4238,9 @@ func TestProxyTracing(t *testing.T) {
 			skip.IgnoreLint(t, "flaky with epoch leases")
 		}
 
+		skip.UnderRace(t, "too slow")
+		skip.UnderDeadlock(t, "too slow")
+
 		const numServers = 3
 		const numRanges = 3
 		st := cluster.MakeTestingClusterSettings()
@@ -4254,6 +4254,9 @@ func TestProxyTracing(t *testing.T) {
 		closedts.SideTransportCloseInterval.Override(ctx, &st.SV, 10*time.Millisecond)
 
 		var p rpc.Partitioner
+		// Partition between n1 and n3.
+		require.NoError(t, p.AddPartition(roachpb.NodeID(1), roachpb.NodeID(3)))
+		require.NoError(t, p.AddPartition(roachpb.NodeID(3), roachpb.NodeID(1)))
 		tc := testcluster.StartTestCluster(t, numServers, base.TestClusterArgs{
 			ServerArgs: base.TestServerArgs{
 				DefaultDRPCOption: base.TestDRPCDisabled,
@@ -4262,8 +4265,7 @@ func TestProxyTracing(t *testing.T) {
 				perNode := make(map[int]base.TestServerArgs)
 				for i := 0; i < numServers; i++ {
 					ctk := rpc.ContextTestingKnobs{}
-					// Partition between n1 and n3.
-					p.RegisterTestingKnobs(roachpb.NodeID(i+1), [][2]roachpb.NodeID{{1, 3}}, &ctk)
+					p.RegisterTestingKnobs(roachpb.NodeID(i+1), &ctk)
 					perNode[i] = base.TestServerArgs{
 						Settings: st,
 						Knobs: base.TestingKnobs{
@@ -4343,7 +4345,7 @@ func TestProxyTracing(t *testing.T) {
 			return checkLeaseCount(3, numRanges)
 		})
 
-		p.EnablePartition(true)
+		p.EnablePartitions(true)
 
 		_, err = conn.Exec("SET TRACING = on; SELECT FROM t where i = 987654321; SET TRACING = off")
 		require.NoError(t, err)

pkg/kv/kvclient/kvcoord/partial_partition_test.go

@@ -58,7 +58,6 @@ func TestPartialPartitionDirectFiveFail(t *testing.T) {
 // Additionally validate that a rangefeed sees the update.
 func testPartialPartition(t *testing.T, useProxy bool, numServers int) {
 	skip.UnderDuress(t, "test does heavy lifting")
-	partition := [][2]roachpb.NodeID{{1, 2}}
 	ctx := context.Background()
 
 	t.Run(fmt.Sprintf("%t-%d", useProxy, numServers), func(t *testing.T) {
@@ -86,6 +85,9 @@ func testPartialPartition(t *testing.T, useProxy bool, numServers int) {
 			zoneConfig.NumVoters = &numNodes
 
 			var p rpc.Partitioner
+			// Partition between n1 and n2.
+			require.NoError(t, p.AddPartition(roachpb.NodeID(1), roachpb.NodeID(2)))
+			require.NoError(t, p.AddPartition(roachpb.NodeID(2), roachpb.NodeID(1)))
 			tc := testcluster.StartTestCluster(t, numServers, base.TestClusterArgs{
 				ServerArgs: base.TestServerArgs{
 					DefaultDRPCOption: base.TestDRPCDisabled,
@@ -94,7 +96,7 @@ func testPartialPartition(t *testing.T, useProxy bool, numServers int) {
 					perNode := make(map[int]base.TestServerArgs)
 					for i := 0; i < numServers; i++ {
 						ctk := rpc.ContextTestingKnobs{}
-						p.RegisterTestingKnobs(roachpb.NodeID(i+1), partition, &ctk)
+						p.RegisterTestingKnobs(roachpb.NodeID(i+1), &ctk)
 						perNode[i] = base.TestServerArgs{
 							Settings:         st,
 							DisableSQLServer: true,
@@ -148,7 +150,7 @@ func testPartialPartition(t *testing.T, useProxy bool, numServers int) {
 				return nil
 			})
 
-			p.EnablePartition(true)
+			p.EnablePartitions(true)
 
 			txn := tc.ApplicationLayer(0).DB().NewTxn(ctx, "test")
 			// DistSender will retry forever. For the failure cases we want

pkg/kv/kvnemesis/BUILD.bazel

@@ -32,6 +32,7 @@ go_library(
         "//pkg/kv/kvserver/liveness",
         "//pkg/kv/kvtestutils",
         "//pkg/roachpb",
+        "//pkg/rpc",
         "//pkg/settings/cluster",
         "//pkg/sql/catalog/bootstrap",
         "//pkg/storage",
@@ -77,6 +78,7 @@ go_test(
     deps = [
         "//pkg/base",
         "//pkg/config/zonepb",
+        "//pkg/gossip",
         "//pkg/kv",
         "//pkg/kv/kvclient/kvcoord",
         "//pkg/kv/kvnemesis/kvnemesisutil",
@@ -89,12 +91,15 @@ go_test(
         "//pkg/kv/kvserver/kvserverpb",
         "//pkg/kv/kvtestutils",
         "//pkg/roachpb",
+        "//pkg/rpc",
         "//pkg/security/securityassets",
         "//pkg/security/securitytest",
         "//pkg/server",
         "//pkg/settings/cluster",
+        "//pkg/spanconfig",
         "//pkg/storage",
         "//pkg/storage/enginepb",
+        "//pkg/testutils",
         "//pkg/testutils/datapathutils",
         "//pkg/testutils/echotest",
         "//pkg/testutils/serverutils",

pkg/kv/kvnemesis/applier.go

@@ -7,6 +7,7 @@ package kvnemesis
 
 import (
 	"context"
+	"strings"
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/config/zonepb"
@@ -98,7 +99,8 @@ func exceptUnhandledRetry(err error) bool {
 }
 
 func exceptAmbiguous(err error) bool { // true if ambiguous result
-	return errors.HasInterface(err, (*kvpb.ClientVisibleAmbiguousError)(nil))
+	return errors.HasInterface(err, (*kvpb.ClientVisibleAmbiguousError)(nil)) ||
+		strings.Contains(err.Error(), "result is ambiguous")
 }
 
 func exceptDelRangeUsingTombstoneStraddlesRangeBoundary(err error) bool {
@@ -109,6 +111,15 @@ func exceptConditionFailed(err error) bool {
 	return errors.HasType(err, (*kvpb.ConditionFailedError)(nil))
 }
 
+func exceptReplicaUnavailable(err error) bool {
+	return errors.HasType(err, (*kvpb.ReplicaUnavailableError)(nil))
+}
+
+func exceptContextCanceled(err error) bool {
+	return errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) ||
+		strings.Contains(err.Error(), "query execution canceled")
+}
+
 func applyOp(ctx context.Context, env *Env, db *kv.DB, op *Operation) {
 	switch o := op.GetValue().(type) {
 	case *GetOperation,
@@ -150,6 +161,12 @@ func applyOp(ctx context.Context, env *Env, db *kv.DB, op *Operation) {
 		o.Result = resultInit(ctx, err)
 	case *FlushLockTableOperation:
 		o.Result = resultInit(ctx, db.FlushLockTable(ctx, o.Key, o.EndKey))
+	case *AddNetworkPartitionOperation:
+		err := env.Partitioner.AddPartition(roachpb.NodeID(o.FromNode), roachpb.NodeID(o.ToNode))
+		o.Result = resultInit(ctx, err)
+	case *RemoveNetworkPartitionOperation:
+		err := env.Partitioner.RemovePartition(roachpb.NodeID(o.FromNode), roachpb.NodeID(o.ToNode))
+		o.Result = resultInit(ctx, err)
 	case *ClosureTxnOperation:
 		// Use a backoff loop to avoid thrashing on txn aborts. Don't wait between
 		// epochs of the same transaction to avoid waiting while holding locks.
@@ -745,7 +762,12 @@ func getRangeDesc(ctx context.Context, key roachpb.Key, dbs ...*kv.DB) roachpb.R
 	var opts = retry.Options{}
 	for r := retry.StartWithCtx(ctx, opts); r.Next(); dbIdx = (dbIdx + 1) % len(dbs) {
 		sender := dbs[dbIdx].NonTransactionalSender()
-		descs, _, err := kv.RangeLookup(ctx, sender, key, kvpb.CONSISTENT, 0, false)
+		// Use kvpb.INCONSISTENT because kv.CONSISTENT requires a transactional
+		// sender. In the generator, range lookups are usually used for finding
+		// replica/lease change targets, so it's ok if these are not consistent.
+		// Using kv.CONSISTENT with a non-transactional sender and in the presence
+		// of network partitions can lead to infinitely stuck lookups.
+		descs, _, err := kv.RangeLookup(ctx, sender, key, kvpb.INCONSISTENT, 0, false)
 		if err != nil {
 			log.Dev.Infof(ctx, "looking up descriptor for %s: %+v", key, err)
 			continue
@@ -756,12 +778,11 @@ func getRangeDesc(ctx context.Context, key roachpb.Key, dbs ...*kv.DB) roachpb.R
 		}
 		return descs[0]
 	}
-	panic(`unreachable`)
+	return roachpb.RangeDescriptor{}
 }
 
 func newGetReplicasFn(dbs ...*kv.DB) GetReplicasFn {
-	ctx := context.Background()
-	return func(key roachpb.Key) ([]roachpb.ReplicationTarget, []roachpb.ReplicationTarget) {
+	return func(ctx context.Context, key roachpb.Key) ([]roachpb.ReplicationTarget, []roachpb.ReplicationTarget) {
 		desc := getRangeDesc(ctx, key, dbs...)
 		replicas := desc.Replicas().Descriptors()
 		var voters []roachpb.ReplicationTarget

pkg/kv/kvnemesis/env.go

@@ -15,6 +15,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/config/zonepb"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvtestutils"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/rpc"
 	"github.com/cockroachdb/cockroach/pkg/util/protoutil"
 	"github.com/cockroachdb/errors"
 )
@@ -29,9 +30,10 @@ type Logger interface {
 // Env manipulates the environment (cluster settings, zone configurations) that
 // the Applier operates in.
 type Env struct {
-	SQLDBs  []*gosql.DB
-	Tracker *SeqTracker
-	L       Logger
+	SQLDBs      []*gosql.DB
+	Tracker     *SeqTracker
+	L           Logger
+	Partitioner *rpc.Partitioner
 }
 
 func (e *Env) anyNode() *gosql.DB {