Implement a stability test for mempool limits (#1647)

* Copy some boilerplate from the other stability tests.

* Fix a copy+paste error in run.sh.

* Copy over some stability test boilerplate go code.

* Run kaspad in the background.

* Catch panics and initialize the RPC client.

* Mine enough blocks to fund filling up the mempool.

* Extract coinbase transactions out of the generated blocks.

* Tidy up a bit.

* Implement submitting transactions.

* Lower the amount of outputs in each transaction.

* Verify that the mempool size has the expected amount of transactions.

* Pregenerate enough funds before submitting the first transaction so that block creation doesn't interfere with the test.

* Empty mempool out by continuously adding blocks to the DAG.

* Handle orphan transactions when overfilling the mempool.

* Increase mempoolSizeLimit to 1m.

* Fix a comment.

* Fix a comment.

* Add mempool-limits to run-slow.sh.

* Rename generateTransactionsWithLotsOfOutputs to generateTransactionsWithMultipleOutputs.

* Rename generateCoinbaseTransaction to mineBlockAndGetCoinbaseTransaction.

* Make generateFundingCoinbaseTransactions return an object instead of store a global variable.

* Convert mempool-limits into a Go test.

* Convert panics to t.Fatalfs.

* Fix a comment.

* Increase mempoolSizeLimit to 1m.

* Run TestMempoolLimits only if RUN_STABILITY_TESTS is set.

* Move the run of mempool-limits in run-slow.sh.

* Add a comment above fundingCoinbaseTransactions.

* Make a couple of stylistic changes.

* Use transactionhelper.CoinbaseTransactionIndex instead of hardcoding 0.

* Make uninteresting errors print %+v instead of %s.

Co-authored-by: Svarog <feanorr@gmail.com>

app/appmessage/rpc_get_info.go

@@ -11,8 +11,8 @@ func (msg *GetInfoRequestMessage) Command() MessageCommand {
 	return CmdGetInfoRequestMessage
 }
 
-// NewGeInfoRequestMessage returns a instance of the message
+// NewGetInfoRequestMessage returns a instance of the message
-func NewGeInfoRequestMessage() *GetInfoRequestMessage {
+func NewGetInfoRequestMessage() *GetInfoRequestMessage {
 	return &GetInfoRequestMessage{}
 }
 

infrastructure/network/rpcclient/rpc_get_info.go

@@ -0,0 +1,20 @@
+package rpcclient
+
+import "github.com/kaspanet/kaspad/app/appmessage"
+
+// GetInfo sends an RPC request respective to the function's name and returns the RPC server's response
+func (c *RPCClient) GetInfo() (*appmessage.GetInfoResponseMessage, error) {
+	err := c.rpcRouter.outgoingRoute().Enqueue(appmessage.NewGetInfoRequestMessage())
+	if err != nil {
+		return nil, err
+	}
+	response, err := c.route(appmessage.CmdGetInfoResponseMessage).DequeueWithTimeout(c.timeout)
+	if err != nil {
+		return nil, err
+	}
+	getInfoResponse := response.(*appmessage.GetInfoResponseMessage)
+	if getInfoResponse.Error != nil {
+		return nil, c.convertRPCError(getInfoResponse.Error)
+	}
+	return getInfoResponse, nil
+}

stability-tests/mempool-limits/README.md

@@ -0,0 +1,14 @@
+# Mempool Limits tool
+
+This tool:
+
+1. Fills up the mempool beyond its transaction limit to make sure eviction works correctly
+2. Mines blocks until the mempool is expected to become empty
+
+## Running
+
+1. `go install` kaspad and mempool-limits.
+2. `cd run`
+3. `./run.sh`
+
+

stability-tests/mempool-limits/config.go

@@ -0,0 +1,44 @@
+package main
+
+import (
+	"path/filepath"
+
+	"github.com/jessevdk/go-flags"
+	"github.com/kaspanet/kaspad/stability-tests/common"
+)
+
+const (
+	defaultLogFilename    = "mempool-limits.log"
+	defaultErrLogFilename = "mempool-limits_err.log"
+)
+
+var (
+	// Default configuration options
+	defaultLogFile    = filepath.Join(common.DefaultAppDir, defaultLogFilename)
+	defaultErrLogFile = filepath.Join(common.DefaultAppDir, defaultErrLogFilename)
+)
+
+type configFlags struct {
+	LogLevel         string `long:"loglevel" description:"Set log level {trace, debug, info, warn, error, critical}"`
+	Profile          string `long:"profile" description:"Enable HTTP profiling on given port -- NOTE port must be between 1024 and 65536"`
+	KaspadRPCAddress string `long:"rpc-address" description:"RPC address of the kaspad node"`
+}
+
+var cfg *configFlags
+
+func activeConfig() *configFlags {
+	return cfg
+}
+
+func parseConfig() error {
+	cfg = &configFlags{}
+	parser := flags.NewParser(cfg, flags.PrintErrors|flags.HelpFlag|flags.IgnoreUnknown)
+	_, err := parser.Parse()
+	if err != nil {
+		return err
+	}
+
+	initLog(defaultLogFile, defaultErrLogFile)
+
+	return nil
+}

stability-tests/mempool-limits/log.go

@@ -0,0 +1,30 @@
+package main
+
+import (
+	"fmt"
+	"os"
+
+	"github.com/kaspanet/kaspad/infrastructure/logger"
+	"github.com/kaspanet/kaspad/stability-tests/common"
+	"github.com/kaspanet/kaspad/util/panics"
+)
+
+var (
+	backendLog = logger.NewBackend()
+	log        = backendLog.Logger("MPLM")
+	spawn      = panics.GoroutineWrapperFunc(log)
+)
+
+func initLog(logFile, errLogFile string) {
+	level := logger.LevelInfo
+	if activeConfig().LogLevel != "" {
+		var ok bool
+		level, ok = logger.LevelFromString(activeConfig().LogLevel)
+		if !ok {
+			fmt.Fprintf(os.Stderr, "Log level %s doesn't exists", activeConfig().LogLevel)
+			os.Exit(1)
+		}
+	}
+	log.SetLevel(level)
+	common.InitBackend(backendLog, logFile, errLogFile)
+}

stability-tests/mempool-limits/main_test.go

@@ -0,0 +1,111 @@
+package main
+
+import (
+	"github.com/kaspanet/kaspad/infrastructure/network/rpcclient"
+	"github.com/kaspanet/kaspad/stability-tests/common"
+	"github.com/kaspanet/kaspad/util/panics"
+	"github.com/kaspanet/kaspad/util/profiling"
+	"os"
+	"testing"
+)
+
+const (
+	mempoolSizeLimit        = 1_000_000
+	overfillMempoolByAmount = 1_000
+)
+
+func TestMempoolLimits(t *testing.T) {
+	if os.Getenv("RUN_STABILITY_TESTS") == "" {
+		t.Skip()
+	}
+
+	defer panics.HandlePanic(log, "mempool-limits-main", nil)
+	err := parseConfig()
+	if err != nil {
+		t.Fatalf("error in parseConfig: %s", err)
+	}
+	defer backendLog.Close()
+	common.UseLogger(backendLog, log.Level())
+
+	cfg := activeConfig()
+	if cfg.Profile != "" {
+		profiling.Start(cfg.Profile, log)
+	}
+
+	payAddressKeyPair := decodePayAddressKeyPair(t)
+	payToPayAddressScript := buildPayToPayAddressScript(t)
+	rpcClient := buildRPCClient(t)
+
+	// Create enough funds for the test
+	fundingTransactions := generateFundingCoinbaseTransactions(t, rpcClient)
+
+	// Fill up the mempool to the brim
+	submitAnAmountOfTransactionsToTheMempool(t, rpcClient, payAddressKeyPair,
+		payToPayAddressScript, fundingTransactions, mempoolSizeLimit, false)
+
+	// Make sure that the mempool size is exactly the limit
+	mempoolSize := getMempoolSize(t, rpcClient)
+	if mempoolSize != mempoolSizeLimit {
+		t.Fatalf("Unexpected mempool size. Want: %d, got: %d",
+			mempoolSizeLimit, mempoolSize)
+	}
+
+	// Add some more transactions to the mempool. We expect the
+	// mempool to either not grow or even to shrink, since an eviction
+	// may also remove any dependant (chained) transactions.
+	// Note that we pass ignoreOrphanRejects: true because we
+	// expect some of the submitted transactions to depend on
+	// transactions that had been evicted from the mempool
+	submitAnAmountOfTransactionsToTheMempool(t, rpcClient, payAddressKeyPair,
+		payToPayAddressScript, fundingTransactions, overfillMempoolByAmount, true)
+
+	// Make sure that the mempool size is the limit or smaller
+	mempoolSize = getMempoolSize(t, rpcClient)
+	if mempoolSize > mempoolSizeLimit {
+		t.Fatalf("Unexpected mempool size. Want at most: %d, got: %d",
+			mempoolSizeLimit, mempoolSize)
+	}
+
+	// Empty mempool out by continuously adding blocks to the DAG
+	emptyOutMempool(t, rpcClient)
+
+	log.Infof("mempool-limits passed")
+}
+
+func buildRPCClient(t *testing.T) *rpcclient.RPCClient {
+	client, err := rpcclient.NewRPCClient(activeConfig().KaspadRPCAddress)
+	if err != nil {
+		t.Fatalf("error connecting to %s: %s", activeConfig().KaspadRPCAddress, err)
+	}
+	return client
+}
+
+func getMempoolSize(t *testing.T, rpcClient *rpcclient.RPCClient) uint64 {
+	getInfoResponse, err := rpcClient.GetInfo()
+	if err != nil {
+		t.Fatalf("GetInfo: %+v", err)
+	}
+	return getInfoResponse.MempoolSize
+}
+
+func emptyOutMempool(t *testing.T, rpcClient *rpcclient.RPCClient) {
+	log.Infof("Adding blocks until mempool shrinks to 0 transactions")
+	getInfoResponse, err := rpcClient.GetInfo()
+	if err != nil {
+		t.Fatalf("GetInfo: %+v", err)
+	}
+	currentMempoolSize := getInfoResponse.MempoolSize
+	for currentMempoolSize > 0 {
+		mineBlockAndGetCoinbaseTransaction(t, rpcClient)
+		getInfoResponse, err := rpcClient.GetInfo()
+		if err != nil {
+			t.Fatalf("GetInfo: %+v", err)
+		}
+		if getInfoResponse.MempoolSize == currentMempoolSize {
+			t.Fatalf("Mempool did not shrink after a block was added to the DAG")
+		}
+		log.Infof("Mempool shrank from %d transactions to %d transactions",
+			currentMempoolSize, getInfoResponse.MempoolSize)
+		currentMempoolSize = getInfoResponse.MempoolSize
+	}
+}

stability-tests/mempool-limits/run/run.sh

@@ -0,0 +1,29 @@
+#!/bin/bash
+
+APPDIR=/tmp/kaspad-temp
+KASPAD_RPC_PORT=29587
+
+rm -rf "${APPDIR}"
+
+kaspad --simnet --appdir="${APPDIR}" --rpclisten=0.0.0.0:"${KASPAD_RPC_PORT}" --profile=6061 &
+KASPAD_PID=$!
+
+sleep 1
+
+RUN_STABILITY_TESTS=true go test ../ -- --rpc-address=127.0.0.1:"${KASPAD_RPC_PORT}" --profile=7000
+TEST_EXIT_CODE=$?
+
+kill $KASPAD_PID
+
+wait $KASPAD_PID
+KASPAD_EXIT_CODE=$?
+
+echo "Exit code: $TEST_EXIT_CODE"
+echo "Kaspad exit code: $KASPAD_EXIT_CODE"
+
+if [ $TEST_EXIT_CODE -eq 0 ] && [ $KASPAD_EXIT_CODE -eq 0 ]; then
+  echo "mempool-limits test: PASSED"
+  exit 0
+fi
+echo "mempool-limits test: FAILED"
+exit 1

stability-tests/mempool-limits/transactions.go

@@ -0,0 +1,200 @@
+package main
+
+import (
+	"encoding/hex"
+	"github.com/kaspanet/go-secp256k1"
+	"github.com/kaspanet/kaspad/app/appmessage"
+	"github.com/kaspanet/kaspad/domain/consensus/model/externalapi"
+	"github.com/kaspanet/kaspad/domain/consensus/utils/consensushashing"
+	"github.com/kaspanet/kaspad/domain/consensus/utils/constants"
+	"github.com/kaspanet/kaspad/domain/consensus/utils/subnetworks"
+	"github.com/kaspanet/kaspad/domain/consensus/utils/transactionhelper"
+	"github.com/kaspanet/kaspad/domain/consensus/utils/txscript"
+	utxopkg "github.com/kaspanet/kaspad/domain/consensus/utils/utxo"
+	"github.com/kaspanet/kaspad/domain/dagconfig"
+	"github.com/kaspanet/kaspad/infrastructure/network/rpcclient"
+	"github.com/kaspanet/kaspad/stability-tests/common/mine"
+	"github.com/kaspanet/kaspad/util"
+	"strings"
+	"testing"
+)
+
+const (
+	payAddress                       = "kaspasim:qr79e37hxdgkn4xjjmfxvqvayc5gsmsql2660d08u9ej9vnc8lzcywr265u64"
+	payAddressPrivateKey             = "0ec5d7308f65717f3f0c3e4d962d73056c1c255a16593b3989589281b51ad5bc"
+	fundingCoinbaseTransactionAmount = 1000
+	outputsPerTransaction            = 3
+	transactionFee                   = 1000
+	coinbaseMaturity                 = 100
+)
+
+// fundingCoinbaseTransactions contains a collection of transactions
+// to be utilized when generating further transactions to fill up
+// the mempool.
+// It's a separate type because we modify the transactions in place
+// whenever we pass an instance of this type into
+// submitAnAmountOfTransactionsToTheMempool.
+type fundingCoinbaseTransactions struct {
+	transactions []*externalapi.DomainTransaction
+}
+
+func generateFundingCoinbaseTransactions(t *testing.T, rpcClient *rpcclient.RPCClient) *fundingCoinbaseTransactions {
+	// Mine a block, since we need at least one block above the genesis
+	// to create a spendable UTXO
+	mineBlockAndGetCoinbaseTransaction(t, rpcClient)
+
+	log.Infof("Generating funding coinbase transactions")
+	fundingCoinbaseTransactions := &fundingCoinbaseTransactions{
+		transactions: make([]*externalapi.DomainTransaction, fundingCoinbaseTransactionAmount),
+	}
+	for i := 0; i < fundingCoinbaseTransactionAmount; i++ {
+		fundingCoinbaseTransactions.transactions[i] = mineBlockAndGetCoinbaseTransaction(t, rpcClient)
+	}
+
+	log.Infof("Maturing funding coinbase transactions")
+	for i := 0; i < coinbaseMaturity; i++ {
+		mineBlockAndGetCoinbaseTransaction(t, rpcClient)
+	}
+
+	return fundingCoinbaseTransactions
+}
+
+func submitAnAmountOfTransactionsToTheMempool(t *testing.T, rpcClient *rpcclient.RPCClient,
+	payAddressKeyPair *secp256k1.SchnorrKeyPair, payToPayAddressScript *externalapi.ScriptPublicKey,
+	fundingTransactions *fundingCoinbaseTransactions, amountToSubmit int, ignoreOrphanRejects bool) {
+
+	log.Infof("Generating %d transactions", amountToSubmit)
+	transactions := make([]*externalapi.DomainTransaction, 0)
+	for len(transactions) < amountToSubmit {
+		var coinbaseTransaction *externalapi.DomainTransaction
+		coinbaseTransaction, fundingTransactions.transactions = fundingTransactions.transactions[0], fundingTransactions.transactions[1:]
+
+		unspentTransactions := []*externalapi.DomainTransaction{coinbaseTransaction}
+		for len(transactions) < amountToSubmit && len(unspentTransactions) > 0 {
+			var transactionToSpend *externalapi.DomainTransaction
+			transactionToSpend, unspentTransactions = unspentTransactions[0], unspentTransactions[1:]
+			spendingTransactions := generateTransactionsWithMultipleOutputs(t, payAddressKeyPair, payToPayAddressScript, transactionToSpend)
+			transactions = append(transactions, spendingTransactions...)
+			unspentTransactions = append(unspentTransactions, spendingTransactions...)
+		}
+		log.Infof("Generated %d transactions", len(transactions))
+	}
+
+	transactions = transactions[:amountToSubmit]
+	log.Infof("Submitting %d transactions", len(transactions))
+
+	for i, transaction := range transactions {
+		rpcTransaction := appmessage.DomainTransactionToRPCTransaction(transaction)
+		_, err := rpcClient.SubmitTransaction(rpcTransaction)
+		if err != nil {
+			if ignoreOrphanRejects && strings.Contains(err.Error(), "orphan") {
+				continue
+			}
+			t.Fatalf("SubmitTransaction: %+v", err)
+		}
+		log.Infof("Submitted %d transactions", i+1)
+	}
+}
+
+func mineBlockAndGetCoinbaseTransaction(t *testing.T, rpcClient *rpcclient.RPCClient) *externalapi.DomainTransaction {
+	getBlockTemplateResponse, err := rpcClient.GetBlockTemplate(payAddress)
+	if err != nil {
+		t.Fatalf("GetBlockTemplate: %+v", err)
+	}
+	templateBlock, err := appmessage.RPCBlockToDomainBlock(getBlockTemplateResponse.Block)
+	if err != nil {
+		t.Fatalf("RPCBlockToDomainBlock: %+v", err)
+	}
+	mine.SolveBlock(templateBlock)
+	_, err = rpcClient.SubmitBlock(templateBlock)
+	if err != nil {
+		t.Fatalf("SubmitBlock: %+v", err)
+	}
+	return templateBlock.Transactions[transactionhelper.CoinbaseTransactionIndex]
+}
+
+func generateTransactionsWithMultipleOutputs(t *testing.T,
+	payAddressKeyPair *secp256k1.SchnorrKeyPair, payToPayAddressScript *externalapi.ScriptPublicKey,
+	fundingTransaction *externalapi.DomainTransaction) []*externalapi.DomainTransaction {
+
+	var transactions []*externalapi.DomainTransaction
+	for fundingTransactionOutputIndex, fundingTransactionOutput := range fundingTransaction.Outputs {
+		if fundingTransactionOutput.Value < transactionFee {
+			continue
+		}
+		outputValue := (fundingTransactionOutput.Value - transactionFee) / outputsPerTransaction
+
+		fundingTransactionID := consensushashing.TransactionID(fundingTransaction)
+		spendingTransactionInputs := []*externalapi.DomainTransactionInput{
+			{
+				PreviousOutpoint: externalapi.DomainOutpoint{
+					TransactionID: *fundingTransactionID,
+					Index:         uint32(fundingTransactionOutputIndex),
+				},
+				UTXOEntry: utxopkg.NewUTXOEntry(
+					fundingTransactionOutput.Value,
+					payToPayAddressScript,
+					false,
+					0),
+			},
+		}
+
+		spendingTransactionOutputs := make([]*externalapi.DomainTransactionOutput, outputsPerTransaction)
+		for i := 0; i < outputsPerTransaction; i++ {
+			spendingTransactionOutputs[i] = &externalapi.DomainTransactionOutput{
+				Value:           outputValue,
+				ScriptPublicKey: payToPayAddressScript,
+			}
+		}
+
+		spendingTransaction := &externalapi.DomainTransaction{
+			Version:      constants.MaxTransactionVersion,
+			Inputs:       spendingTransactionInputs,
+			Outputs:      spendingTransactionOutputs,
+			LockTime:     0,
+			SubnetworkID: subnetworks.SubnetworkIDNative,
+			Gas:          0,
+			Payload:      nil,
+		}
+
+		for spendingTransactionInputIndex, spendingTransactionInput := range spendingTransactionInputs {
+			signatureScript, err := txscript.SignatureScript(
+				spendingTransaction,
+				spendingTransactionInputIndex,
+				consensushashing.SigHashAll,
+				payAddressKeyPair,
+				&consensushashing.SighashReusedValues{})
+			if err != nil {
+				t.Fatalf("SignatureScript: %+v", err)
+			}
+			spendingTransactionInput.SignatureScript = signatureScript
+		}
+
+		transactions = append(transactions, spendingTransaction)
+	}
+	return transactions
+}
+
+func decodePayAddressKeyPair(t *testing.T) *secp256k1.SchnorrKeyPair {
+	privateKeyBytes, err := hex.DecodeString(payAddressPrivateKey)
+	if err != nil {
+		t.Fatalf("DecodeString: %+v", err)
+	}
+	keyPair, err := secp256k1.DeserializeSchnorrPrivateKeyFromSlice(privateKeyBytes)
+	if err != nil {
+		t.Fatalf("DeserializeSchnorrPrivateKeyFromSlice: %+v", err)
+	}
+	return keyPair
+}
+
+func buildPayToPayAddressScript(t *testing.T) *externalapi.ScriptPublicKey {
+	address, err := util.DecodeAddress(payAddress, dagconfig.SimnetParams.Prefix)
+	if err != nil {
+		t.Fatalf("DecodeAddress: %+v", err)
+	}
+	script, err := txscript.PayToAddrScript(address)
+	if err != nil {
+		t.Fatalf("PayToAddrScript: %+v", err)
+	}
+	return script
+}

stability-tests/run/run-slow.sh

@@ -38,6 +38,10 @@ echo "Running reorg"
 cd "${PROJECT_ROOT}/reorg/run" && ./run.sh || failedTests+=("reorg")
 echo "Done running reorg"
 
+echo "Running mempool-limits"
+cd "${PROJECT_ROOT}/mempool-limits/run" && ./run.sh || failedTests+=("mempool-limits")
+echo "Done running mempool-limits"
+
 echo "Running netsync - slow"
 cd ${PROJECT_ROOT}/netsync/run"" && ./run.sh || failedTests+=("netsync")
 echo "Done running netsync - slow"