kvnemesis: add FlushLockTableOperation

This adds FlushLockTable to the set of KVNemsis operations. It doesn't
add any particularly interesting validations.

Epic: none
Release note: None

Author: stevendanna

pkg/kv/batch.go

@@ -300,6 +300,7 @@ func (b *Batch) fillResults(ctx context.Context) {
 			case *kvpb.MigrateRequest:
 			case *kvpb.QueryResolvedTimestampRequest:
 			case *kvpb.BarrierRequest:
+			case *kvpb.FlushLockTableRequest:
 			case *kvpb.LinkExternalSSTableRequest:
 			case *kvpb.ExciseRequest:
 			default:
@@ -1147,6 +1148,27 @@ func (b *Batch) barrier(s, e interface{}, withLAI bool) {
 	b.initResult(1, 0, notRaw, nil)
 }
 
+func (b *Batch) flushLockTable(s, e interface{}) {
+	begin, err := marshalKey(s)
+	if err != nil {
+		b.initResult(0, 0, notRaw, err)
+		return
+	}
+	end, err := marshalKey(e)
+	if err != nil {
+		b.initResult(0, 0, notRaw, err)
+		return
+	}
+	req := &kvpb.FlushLockTableRequest{
+		RequestHeader: kvpb.RequestHeader{
+			Key:    begin,
+			EndKey: end,
+		},
+	}
+	b.appendReqs(req)
+	b.initResult(1, 0, notRaw, nil)
+}
+
 func (b *Batch) bulkRequest(
 	numKeys int, requestFactory func() (req kvpb.RequestUnion, kvSize int),
 ) {

pkg/kv/db.go

@@ -899,6 +899,23 @@ func (db *DB) Barrier(ctx context.Context, begin, end interface{}) (hlc.Timestam
 	return resp.Timestamp, nil
 }
 
+func (db *DB) FlushLockTable(ctx context.Context, begin, end interface{}) error {
+	b := &Batch{}
+	b.flushLockTable(begin, end)
+	if err := getOneErr(db.Run(ctx, b), b); err != nil {
+		return err
+	}
+	if l := len(b.response.Responses); l != 1 {
+		return errors.Errorf("got %d responses for FlushLockTable", l)
+	}
+	resp := b.response.Responses[0].GetFlushLockTable()
+	if resp == nil {
+		return errors.Errorf("unexpected response %T for FlushLockTable",
+			b.response.Responses[0].GetInner())
+	}
+	return nil
+}
+
 // BarrierWithLAI is like Barrier, but also returns the lease applied index and
 // range descriptor at which the barrier was applied. In this case, the barrier
 // can't span multiple ranges, otherwise a RangeKeyMismatchError is returned.

pkg/kv/kvnemesis/applier.go

@@ -143,6 +143,8 @@ func applyOp(ctx context.Context, env *Env, db *kv.DB, op *Operation) {
 			_, err = db.Barrier(ctx, o.Key, o.EndKey)
 		}
 		o.Result = resultInit(ctx, err)
+	case *FlushLockTableOperation:
+		o.Result = resultInit(ctx, db.FlushLockTable(ctx, o.Key, o.EndKey))
 	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.
@@ -449,6 +451,16 @@ func applyClientOp(
 			})
 		})
 		o.Result = resultInit(ctx, err)
+	case *FlushLockTableOperation:
+		_, _, err := dbRunWithResultAndTimestamp(ctx, db, func(b *kv.Batch) {
+			b.AddRawRequest(&kvpb.FlushLockTableRequest{
+				RequestHeader: kvpb.RequestHeader{
+					Key:    o.Key,
+					EndKey: o.EndKey,
+				},
+			})
+		})
+		o.Result = resultInit(ctx, err)
 	case *BatchOperation:
 		b := &kv.Batch{}
 		applyBatchOp(ctx, b, db.Run, o)
@@ -569,6 +581,8 @@ func applyBatchOp(
 			panic(errors.AssertionFailedf(`AddSSTable cannot be used in batches`))
 		case *BarrierOperation:
 			panic(errors.AssertionFailedf(`Barrier cannot be used in batches`))
+		case *FlushLockTableOperation:
+			panic(errors.AssertionFailedf(`FlushLockOperation cannot be used in batches`))
 		default:
 			panic(errors.AssertionFailedf(`unknown batch operation type: %T %v`, subO, subO))
 		}

pkg/kv/kvnemesis/generator.go

@@ -261,6 +261,10 @@ type ClientOperationConfig struct {
 	AddSSTable int
 	// Barrier is an operation that waits for in-flight writes to complete.
 	Barrier int
+
+	// FlushLockTable is an operation that moves unreplicated locks in the
+	// in-memory lock table into the
+	FlushLockTable int
 }
 
 // BatchOperationConfig configures the relative probability of generating a
@@ -401,6 +405,7 @@ func newAllOperationsConfig() GeneratorConfig {
 		DeleteRangeUsingTombstone:                          1,
 		AddSSTable:                                         1,
 		Barrier:                                            1,
+		FlushLockTable:                                     1,
 	}
 	batchOpConfig := BatchOperationConfig{
 		Batch: 4,
@@ -536,6 +541,11 @@ func NewDefaultConfig() GeneratorConfig {
 	config.Ops.ClosureTxn.CommitBatchOps.Barrier = 0
 	config.Ops.ClosureTxn.TxnClientOps.Barrier = 0
 	config.Ops.ClosureTxn.TxnBatchOps.Ops.Barrier = 0
+
+	config.Ops.Batch.Ops.FlushLockTable = 0
+	config.Ops.ClosureTxn.CommitBatchOps.FlushLockTable = 0
+	config.Ops.ClosureTxn.TxnClientOps.FlushLockTable = 0
+	config.Ops.ClosureTxn.TxnBatchOps.Ops.FlushLockTable = 0
 	return config
 }
 
@@ -833,6 +843,7 @@ func (g *generator) registerClientOps(allowed *[]opGen, c *ClientOperationConfig
 	addOpGen(allowed, randDelRangeUsingTombstone, c.DeleteRangeUsingTombstone)
 	addOpGen(allowed, randAddSSTable, c.AddSSTable)
 	addOpGen(allowed, randBarrier, c.Barrier)
+	addOpGen(allowed, randFlushLockTable, c.FlushLockTable)
 }
 
 func (g *generator) registerBatchOps(allowed *[]opGen, c *BatchOperationConfig) {
@@ -1138,6 +1149,19 @@ func randBarrier(g *generator, rng *rand.Rand) Operation {
 	return barrier(key, endKey, withLAI)
 }
 
+func randFlushLockTable(g *generator, rng *rand.Rand) Operation {
+	// FlushLockTable can't span multiple ranges. We want to test a combination of
+	// requests that span the entire range and those that span part of a range.
+	key, endKey := randRangeSpan(rng, g.currentSplits)
+
+	wholeRange := rng.Float64() < 0.5
+	if !wholeRange {
+		key = randKeyBetween(rng, key, endKey)
+	}
+
+	return flushLockTable(key, endKey)
+}
+
 func randScan(g *generator, rng *rand.Rand) Operation {
 	key, endKey := randSpan(rng)
 	return scan(key, endKey)
@@ -1976,6 +2000,13 @@ func barrier(key, endKey string, withLAI bool) Operation {
 	}}
 }
 
+func flushLockTable(key, endKey string) Operation {
+	return Operation{FlushLockTable: &FlushLockTableOperation{
+		Key:    []byte(key),
+		EndKey: []byte(endKey),
+	}}
+}
+
 func createSavepoint(id int) Operation {
 	return Operation{SavepointCreate: &SavepointCreateOperation{ID: int32(id)}}
 }

pkg/kv/kvnemesis/generator_test.go

@@ -248,6 +248,8 @@ func TestRandStep(t *testing.T) {
 				client.AddSSTable++
 			case *BarrierOperation:
 				client.Barrier++
+			case *FlushLockTableOperation:
+				client.FlushLockTable++
 			case *BatchOperation:
 				batch.Batch++
 				countClientOps(&batch.Ops, nil, o.Ops...)
@@ -284,7 +286,8 @@ func TestRandStep(t *testing.T) {
 			*DeleteRangeOperation,
 			*DeleteRangeUsingTombstoneOperation,
 			*AddSSTableOperation,
-			*BarrierOperation:
+			*BarrierOperation,
+			*FlushLockTableOperation:
 			countClientOps(&counts.DB, &counts.Batch, step.Op)
 		case *ClosureTxnOperation:
 			countClientOps(&counts.ClosureTxn.TxnClientOps, &counts.ClosureTxn.TxnBatchOps, o.Ops...)

pkg/kv/kvnemesis/operations.go

@@ -38,6 +38,8 @@ func (op Operation) Result() *Result {
 		return &o.Result
 	case *BarrierOperation:
 		return &o.Result
+	case *FlushLockTableOperation:
+		return &o.Result
 	case *SplitOperation:
 		return &o.Result
 	case *MergeOperation:
@@ -138,6 +140,8 @@ func (op Operation) format(w *strings.Builder, fctx formatCtx) {
 		o.format(w, fctx)
 	case *BarrierOperation:
 		o.format(w, fctx)
+	case *FlushLockTableOperation:
+		o.format(w, fctx)
 	case *SplitOperation:
 		o.format(w, fctx)
 	case *MergeOperation:
@@ -364,6 +368,11 @@ func (op BarrierOperation) format(w *strings.Builder, fctx formatCtx) {
 	op.Result.format(w)
 }
 
+func (op FlushLockTableOperation) format(w *strings.Builder, fctx formatCtx) {
+	fmt.Fprintf(w, `%s.FlushLockTable(ctx, %s, %s)`, fctx.receiver, fmtKey(op.Key), fmtKey(op.EndKey))
+	op.Result.format(w)
+}
+
 func (op SplitOperation) format(w *strings.Builder, fctx formatCtx) {
 	fmt.Fprintf(w, `%s.AdminSplit(ctx, %s, hlc.MaxTimestamp)`, fctx.receiver, fmtKey(op.Key))
 	op.Result.format(w)

pkg/kv/kvnemesis/operations.proto

@@ -97,6 +97,12 @@ message BarrierOperation {
   Result result = 4 [(gogoproto.nullable) = false];
 }
 
+message FlushLockTableOperation {
+  bytes key = 1;
+  bytes end_key = 2;
+  Result result = 3 [(gogoproto.nullable) = false];
+}
+
 message SplitOperation {
   bytes key = 1;
   Result result = 2 [(gogoproto.nullable) = false];
@@ -182,6 +188,7 @@ message Operation {
   SavepointReleaseOperation savepoint_release = 21;
   SavepointRollbackOperation savepoint_rollback = 22;
   BarrierOperation barrier = 23;
+  FlushLockTableOperation flush_lock_table = 24;
 }
 
 enum ResultType {

pkg/kv/kvnemesis/validator.go

@@ -733,8 +733,19 @@ func (v *validator) processOp(op Operation) {
 			// Fail or retry on other errors, depending on type.
 			v.checkNonAmbError(op, t.Result, exceptUnhandledRetry)
 		}
-		// We don't yet actually check the barrier guarantees here, i.e. that all
-		// concurrent writes are applied by the time it completes. Maybe later.
+	// We don't yet actually check the barrier guarantees here, i.e. that all
+	// concurrent writes are applied by the time it completes. Maybe later.
+	case *FlushLockTableOperation:
+		// TODO(ssd): Should this be true? Right now it needs
+		// to be or the test fails; but shoujld we be
+		// returning a timestamp?
+		execTimestampStrictlyOptional = true
+		if resultHasErrorType(t.Result, &kvpb.RangeKeyMismatchError{}) {
+			// FlushLockTableOperation may race with a split.
+		} else {
+			// Fail or retry on other errors, depending on type.
+			v.checkNonAmbError(op, t.Result, exceptUnhandledRetry)
+		}
 	case *ScanOperation:
 		if _, isErr := v.checkError(op, t.Result); isErr {
 			break