You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: txn_recovery_integration_test.go
    path of file: ./repos/cockroach/pkg/kv/kvserver
    
    the code of the file until where you have to start completion: // Copyright 2019 The Cockroach Authors.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt.
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0, included in the file
// licenses/APL.txt.

package kvserver

import (
	"bytes"
	"context"
	"fmt"
	"testing"

	"github.com/cockroachdb/cockroach/pkg/kv"
	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/concurrency/isolation"
	"github.com/cockroachdb/cockroach/pkg/roachpb"
	"github.com/cockroachdb/cockroach/pkg/testutils"
	"github.com/cockroachdb/cockroach/pkg/util/hlc"
	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
	"github.com/cockroachdb/cockroach/pkg/util/log"
	"github.com/cockroachdb/cockroach/pkg/util/stop"
	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
	"github.com/cockroachdb/errors"
	"github.com/stretchr/testify/require"
)

// TestTxnRecoveryFromStaging tests the recovery process for a transaction that
// stages its transaction reco
    
func TestTxnRecoveryFromStagingWithoutHighPriority(t *testing.T) {
	defer leaktest.AfterTest(t)()
	defer log.Scope(t).Close(t)
	ctx := context.Background()

	run := func(t *testing.T, pusheeIsoLevel isolation.Level, pusheeCommits, pusherWriting bool) {
		stopper := stop.NewStopper()
		defer stopper.Stop(ctx)
		manual := timeutil.NewManualTime(timeutil.Unix(0, 123))
		cfg := TestStoreConfig(hlc.NewClockForTesting(manual))
		store := createTestStoreWithConfig(ctx, t, stopper, testStoreOpts{createSystemRanges: true}, &cfg)

		// Create a transaction that will get stuck performing a parallel
		// commit.
		keyA, keyB := roachpb.Key("a"), roachpb.Key("b")
		txn := newTransaction("txn", keyA, 1, store.Clock())
		txn.IsoLevel = pusheeIsoLevel

		// Issue two writes, which will be considered in-flight at the time of
		// the transaction's EndTxn request.
		keyAVal := []byte("value")
		pArgs := putArgs(keyA, keyAVal)
		pArgs.Sequence = 1
		h := kvpb.Header{Txn: txn}
		_, pErr := kv.SendWrappedWith(ctx, store.TestSender(), h, &pArgs)
		require.Nil(t, pErr, "error: %s", pErr)

		pArgs = putArgs(keyB, []byte("value2"))
		pArgs.Sequence = 2
		h2 := kvpb.Header{Txn: txn.Clone()}
		if !pusheeCommits {
			// If we're not going to have the pushee commit, make sure it never enters
			// the implicit commit state by bumping the timestamp of one of its writes.
			manual.Advance(100)
			h2.Txn.WriteTimestamp = store.Clock().Now()
		}
		_, pErr = kv.SendWrappedWith(ctx, store.TestSender(), h2, &pArgs)
		require.Nil(t, pErr, "error: %s", pErr)

		// Issue a parallel commit, which will put the transaction into a
		// STAGING state. Include both writes as the EndTxn's in-flight writes.
		et, etH := endTxnArgs(txn, true)
		et.InFlightWrites = []roachpb.SequencedWrite{
			{Key: keyA, Sequence: 1},
			{Key: keyB, Sequence: 2},
		}
		etReply, pErr := kv.SendWrappedWith(ctx, store.TestSender(), etH, &et)
		require.Nil(t, pErr, "error: %s", pErr)
		require.Equal(t, roachpb.STAGING, etReply.Header().Txn.Status)

		// Issue a conflicting, normal-priority operation.
		var conflictArgs kvpb.Request
		if pusherWriting {
			pArgs = putArgs(keyB, []byte("value3"))
			conflictArgs = &pArgs
		} else {
			gArgs := getArgs(keyB)
			conflictArgs = &gArgs
		}
		manual.Advance(100)
		pErrC := make(chan *kvpb.Error, 1)
		require.NoError(t, stopper.RunAsyncTask(ctx, "conflict", func(ctx context.Context) {
			_, pErr := kv.SendWrapped(ctx, store.TestSender(), conflictArgs)
			pErrC <- pErr
		}))

		// Wait for the conflict to push and be queued in the txn wait queue.
		testutils.SucceedsSoon(t, func() error {
			select {
			case pErr := <-pErrC:
				t.Fatalf("conflicting operation unexpectedly completed: pErr=%s", pErr)
			default:
			}
			if v := store.txnWaitMetrics.PusherWaiting.Value(); v != 1 {
				return errors.Errorf("expected 1 pusher waiting, found %d", v)
			}
			return nil
		})

		// Finalize the STAGING txn, either by committing it or by aborting it.
		et2, et2H := endTxnArgs(txn, pusheeCommits)
		etReply, pErr = kv.SendWrappedWith(ctx, store.TestSender(), et2H, &et2)
		require.Nil(t, pErr, "error: %s", pErr)
		expStatus := roachpb.COMMITTED
		if !pusheeCommits {
			expStatus = roachpb.ABORTED
		}
		require.Equal(t, expStatus, etReply.Header().Txn.Status)

		// This will unblock the conflicting operation, which should succeed.
		pErr = <-pErrC
		require.Nil(t, pErr, "error: %s", pErr)
	}

	for _, pusheeIsoLevel := range isolation.Levels() {
		t.Run("pushee_iso_level="+pusheeIsoLevel.String(), func(t *testing.T) {
			testutils.RunTrueAndFalse(t, "pushee_commits", func(t *testing.T, pusheeCommits bool) {
				testutils.RunTrueAndFalse(t, "pusher_writing", func(t *testing.T, pusherWriting bool) {
					run(t, pusheeIsoLevel, pusheeCommits, pusherWriting)
				})
			})
		})
	}
}