[3.2] Backport: Create integration test for sending copies of the same transaction into the network

tests/CMakeLists.txt

@@ -223,6 +223,10 @@ set_property(TEST plugin_http_api_test PROPERTY LABELS nonparallelizable_tests)
 add_test(NAME resource_monitor_plugin_test COMMAND tests/resource_monitor_plugin_test.py WORKING_DIRECTORY ${CMAKE_BINARY_DIR})
 set_property(TEST resource_monitor_plugin_test PROPERTY LABELS long_running_tests)
 
+add_test(NAME nodeos_repeat_transaction_lr_test COMMAND tests/nodeos_high_transaction_test.py -v --clean-run --dump-error-detail -p 4 -n 8 --num-transactions 1000 --max-transactions-per-second 500 --send-duplicates WORKING_DIRECTORY ${CMAKE_BINARY_DIR})
+set_property(TEST nodeos_repeat_transaction_lr_test PROPERTY LABELS long_running_tests)
+
+
 if(ENABLE_COVERAGE_TESTING)
 
   set(Coverage_NAME ${PROJECT_NAME}_coverage)

tests/Node.py

@@ -526,15 +526,27 @@ def __call__(self):
     def waitForIrreversibleBlock(self, blockNum, timeout=None, reportInterval=None):
         return self.waitForBlock(blockNum, timeout=timeout, blockType=BlockType.lib, reportInterval=reportInterval)
 
-    def __transferFundsCmdArr(self, source, destination, amountStr, memo, force, retry, sign):
+    def __transferFundsCmdArr(self, source, destination, amountStr, memo, force, retry, sign, dontSend, expiration):
         assert isinstance(amountStr, str)
         assert(source)
         assert(isinstance(source, Account))
         assert(destination)
         assert(isinstance(destination, Account))
+        assert(expiration is None or isinstance(expiration, int))
 
-        cmd="%s %s -v transfer --expiration 90 %s -j" % (
-            Utils.EosClientPath, self.eosClientArgs(), self.getRetryCmdArg(retry))
+        dontSendStr = ""
+        if dontSend:
+            dontSendStr = "--dont-broadcast "
+            if expiration is None:
+                # default transaction expiration to be 4 minutes in the future
+                expiration = 240
+
+        expirationStr = ""
+        if expiration is not None:
+            expirationStr = "--expiration %d " % (expiration)
+
+        cmd="%s %s -v transfer %s -j %s %s" % (
+            Utils.EosClientPath, self.eosClientArgs(), self.getRetryCmdArg(retry), dontSendStr, expirationStr)
         cmdArr=cmd.split()
         # not using __sign_str, since cmdArr messes up the string
         if sign:
@@ -552,16 +564,17 @@ def __transferFundsCmdArr(self, source, destination, amountStr, memo, force, ret
         return cmdArr
 
     # Trasfer funds. Returns "transfer" json return object
-    def transferFunds(self, source, destination, amountStr, memo="memo", force=False, waitForTransBlock=False, exitOnError=True, reportStatus=True, retry=None, sign=False):
-        cmdArr = self.__transferFundsCmdArr(source, destination, amountStr, memo, force, retry, sign)
+    def transferFunds(self, source, destination, amountStr, memo="memo", force=False, waitForTransBlock=False, exitOnError=True, reportStatus=True, retry=None, sign=False, dontSend=False, expiration=90):
+        cmdArr = self.__transferFundsCmdArr(source, destination, amountStr, memo, force, retry, sign, dontSend, expiration)
         trans=None
         start=time.perf_counter()
         try:
             trans=Utils.runCmdArrReturnJson(cmdArr)
             if Utils.Debug:
                 end=time.perf_counter()
                 Utils.Print("cmd Duration: %.3f sec" % (end-start))
-            self.trackCmdTransaction(trans, reportStatus=reportStatus)
+            if not dontSend:
+                self.trackCmdTransaction(trans, reportStatus=reportStatus)
         except subprocess.CalledProcessError as ex:
             end=time.perf_counter()
             msg=ex.output.decode("utf-8")
@@ -578,8 +591,8 @@ def transferFunds(self, source, destination, amountStr, memo="memo", force=False
         return self.waitForTransBlockIfNeeded(trans, waitForTransBlock, exitOnError=exitOnError)
 
     # Trasfer funds. Returns (popen, cmdArr) for checkDelayedOutput
-    def transferFundsAsync(self, source, destination, amountStr, memo="memo", force=False, exitOnError=True, retry=None, sign=False):
-        cmdArr = self.__transferFundsCmdArr(source, destination, amountStr, memo, force, retry, sign)
+    def transferFundsAsync(self, source, destination, amountStr, memo="memo", force=False, exitOnError=True, retry=None, sign=False, dontSend=False, expiration=90):
+        cmdArr = self.__transferFundsCmdArr(source, destination, amountStr, memo, force, retry, sign, dontSend, expiration)
         start=time.perf_counter()
         try:
             popen=Utils.delayedCheckOutput(cmdArr)
@@ -820,11 +833,11 @@ def getTableColumns(self, contract, scope, table):
         keys=list(row.keys())
         return keys
 
+    # returns tuple with indication if transaction was successfully sent and either the transaction or else the exception output
     def pushTransaction(self, trans, opts="", silentErrors=False, permissions=None):
         assert(isinstance(trans, dict))
         if isinstance(permissions, str):
             permissions=[permissions]
-        reportStatus = True
 
         cmd="%s %s push transaction -j" % (Utils.EosClientPath, self.eosClientArgs())
         cmdArr=cmd.split()
@@ -867,7 +880,6 @@ def pushMessage(self, account, action, data, opts, silentErrors=False, signature
             cmdArr.append(data)
         if opts is not None:
             cmdArr += opts.split()
-        s=" ".join(cmdArr)
         if Utils.Debug: Utils.Print("cmd: %s" % (cmdArr))
         start=time.perf_counter()
         try:
@@ -897,7 +909,6 @@ def setPermission(self, account, code, pType, requirement, waitForTransBlock=Fal
         assert(isinstance(account, Account))
         assert(isinstance(code, Account))
         signStr = Node.__sign_str(sign, [ account.activePublicKey ])
-        Utils.Print("REMOVE signStr: <%s>" % (signStr))
         cmdDesc="set action permission"
         cmd="%s -j %s %s %s %s %s" % (cmdDesc, signStr, account.name, code.name, pType, requirement)
         trans=self.processCleosCmd(cmd, cmdDesc, silentErrors=False, exitOnError=exitOnError)

tests/TestHelper.py

@@ -11,20 +11,21 @@ def __init__(self):
         self.args=[]
 
     class AppArg:
-        def __init__(self, flag, type, help, default, choices=None):
+        def __init__(self, flag, help, type=None, default=None, choices=None, action=None):
             self.flag=flag
             self.type=type
             self.help=help
             self.default=default
             self.choices=choices
+            self.action=action
 
     def add(self, flag, type, help, default, choices=None):
-        arg=self.AppArg(flag, type, help, default, choices)
+        arg=self.AppArg(flag, help, type=type, default=default, choices=choices)
         self.args.append(arg)
 
 
     def add_bool(self, flag, help, action='store_true'):
-        arg=self.AppArg(flag=flag, help=help, action=action)
+        arg=self.AppArg(flag, help, action=action)
         self.args.append(arg)
 
 # pylint: disable=too-many-instance-attributes
@@ -112,7 +113,10 @@ def parse_args(includeArgs, applicationSpecificArgs=AppArgs()):
             parser.add_argument("--alternate-version-labels-file", type=str, help="Provide a file to define the labels that can be used in the test and the path to the version installation associated with that.")
 
         for arg in applicationSpecificArgs.args:
-            parser.add_argument(arg.flag, type=arg.type, help=arg.help, choices=arg.choices, default=arg.default)
+            if arg.type is not None:
+                parser.add_argument(arg.flag, type=arg.type, help=arg.help, choices=arg.choices, default=arg.default)
+            else:
+                parser.add_argument(arg.flag, help=arg.help, action=arg.action)
 
         args = parser.parse_args()
         return args

tests/nodeos_high_transaction_test.py

@@ -1,6 +1,7 @@
 #!/usr/bin/env python3
 
 from testUtils import Utils
+import signal
 import time
 from Cluster import Cluster
 from Cluster import NamedAccounts
@@ -30,6 +31,7 @@
 extraArgs = appArgs.add(flag="--num-transactions", type=int, help="How many total transactions should be sent", default=10000)
 extraArgs = appArgs.add(flag="--max-transactions-per-second", type=int, help="How many transactions per second should be sent", default=500)
 extraArgs = appArgs.add(flag="--total-accounts", type=int, help="How many accounts should be involved in sending transfers.  Must be greater than %d" % (minTotalAccounts), default=100)
+extraArgs = appArgs.add_bool(flag="--send-duplicates", help="If identical transactions should be sent to all nodes")
 args = TestHelper.parse_args({"-p", "-n","--dump-error-details","--keep-logs","-v","--leave-running","--clean-run"}, applicationSpecificArgs=appArgs)
 
 Utils.Debug=args.v
@@ -50,7 +52,7 @@
 transBlocksBehind=args.transaction_time_delta * blocksPerSec
 numTransactions = args.num_transactions
 maxTransactionsPerSecond = args.max_transactions_per_second
-assert args.total_accounts >= minTotalAccounts, Print("ERROR: Only %d was selected for --total-accounts, must have at least %d" % (args.total_accounts, minTotalAccounts))
+assert args.total_accounts >= minTotalAccounts, "ERROR: Only %d was selected for --total-accounts, must have at least %d" % (args.total_accounts, minTotalAccounts)
 if numTransactions % args.total_accounts > 0:
     oldNumTransactions = numTransactions
     numTransactions = int((oldNumTransactions + args.total_accounts - 1)/args.total_accounts) * args.total_accounts
@@ -67,6 +69,9 @@
 WalletdName=Utils.EosWalletName
 ClientName="cleos"
 
+maxTransactionAttempts = 2            # max number of attempts to try to send a transaction
+maxTransactionAttemptsNoSend = 1      # max number of attempts to try to create a transaction to be sent as a duplicate
+
 try:
     TestHelper.printSystemInfo("BEGIN")
 
@@ -79,7 +84,7 @@
     if cluster.launch(pnodes=totalProducerNodes,
                       totalNodes=totalNodes, totalProducers=totalProducers,
                       useBiosBootFile=False,
-                      extraNodeosArgs=traceNodeosArgs) is False:
+                      extraNodeosArgs=traceNodeosArgs, topo="ring") is False:
         Utils.cmdError("launcher")
         Utils.errorExit("Failed to stand up eos cluster.")
 
@@ -110,8 +115,9 @@
 
     nonProdNodes=[]
     prodNodes=[]
+    allNodes=cluster.getNodes()
     for i in range(0, totalNodes):
-        node=cluster.getNode(i)
+        node=allNodes[i]
         nodeProducers=Cluster.parseProducers(i)
         numProducers=len(nodeProducers)
         Print("node has producers=%s" % (nodeProducers))
@@ -176,15 +182,16 @@ def cacheTransIdInBlock(transId, transToBlock, node):
 
             lastIrreversibleBlockNum = node.getIrreversibleBlockNum()
             blockNum = Node.getTransBlockNum(trans)
-            assert blockNum is not None, Print("ERROR: could not retrieve block num from transId: %s, trans: %s" % (transId, json.dumps(trans, indent=2)))
+            assert blockNum is not None, "ERROR: could not retrieve block num from transId: %s, trans: %s" % (transId, json.dumps(trans, indent=2))
             block = node.getBlock(blockNum)
             if block is not None:
                 transactions = block["transactions"]
                 for trans_receipt in transactions:
                     btrans = trans_receipt["trx"]
-                    assert btrans is not None, Print("ERROR: could not retrieve \"trx\" from transaction_receipt: %s, from transId: %s that led to block: %s" % (json.dumps(trans_receipt, indent=2), transId, json.dumps(block, indent=2)))
+                    assert btrans is not None, "ERROR: could not retrieve \"trx\" from transaction_receipt: %s, from transId: %s that led to block: %s" % (json.dumps(trans_receipt, indent=2), transId, json.dumps(block, indent=2))
                     btransId = btrans["id"]
-                    assert btransId is not None, Print("ERROR: could not retrieve \"id\" from \"trx\": %s, from transId: %s that led to block: %s" % (json.dumps(btrans, indent=2), transId, json.dumps(block, indent=2)))
+                    assert btransId is not None, "ERROR: could not retrieve \"id\" from \"trx\": %s, from transId: %s that led to block: %s" % (json.dumps(btrans, indent=2), transId, json.dumps(block, indent=2))
+                    assert btransId not in transToBlock, "ERROR: transaction_id: %s found in block: %d, but originally seen in block number: %d" % (btransId, blockNum, transToBlock[btransId]["block_num"])
                     transToBlock[btransId] = block
 
                 break
@@ -205,8 +212,8 @@ def findTransInBlock(transId, transToBlock, node):
         if transId in transToBlock:
             return
         (block, trans) = cacheTransIdInBlock(transId, transToBlock, node)
-        assert trans is not None, Print("ERROR: could not find transaction for transId: %s" % (transId))
-        assert block is not None, Print("ERROR: could not retrieve block with block num: %d, from transId: %s, trans: %s" % (blockNum, transId, json.dumps(trans, indent=2)))
+        assert trans is not None, "ERROR: could not find transaction for transId: %s" % (transId)
+        assert block is not None, "ERROR: could not retrieve block with block num: %d, from transId: %s, trans: %s" % (blockNum, transId, json.dumps(trans, indent=2))
 
     transToBlock = {}
     for transId in checkTransIds:
@@ -219,6 +226,20 @@ def findTransInBlock(transId, transToBlock, node):
     #verify nodes are in sync and advancing
     cluster.waitOnClusterSync(blockAdvancing=5)
 
+    nodeOrder = []
+    if args.send_duplicates:
+        # kill bios, since it prevents the ring topography from really being a ring
+        cluster.biosNode.kill(signal.SIGTERM)
+        nodeOrder.append(0)
+        # jump to node furthest in ring from node 0
+        next = int((totalNodes + 1) / 2)
+        nodeOrder.append(next)
+        # then just fill in the rest of the nodes
+        for i in range(1, next):
+            nodeOrder.append(i)
+        for i in range(next + 1, totalNodes):
+            nodeOrder.append(i)
+
     Print("Sending %d transfers" % (numTransactions))
     delayAfterRounds = int(maxTransactionsPerSecond / args.total_accounts)
     history = []
@@ -238,22 +259,46 @@ def findTransInBlock(transId, transToBlock, node):
             time.sleep(delayTime)
 
         transferAmount = Node.currencyIntToStr(round + 1, CORE_SYMBOL)
+
         Print("Sending round %d, transfer: %s" % (round, transferAmount))
         for accountIndex in range(0, args.total_accounts):
             fromAccount = accounts[accountIndex]
             toAccountIndex = accountIndex + 1 if accountIndex + 1 < args.total_accounts else 0
             toAccount = accounts[toAccountIndex]
             node = nonProdNodes[accountIndex % nonProdNodeCount]
-            trans=node.transferFunds(fromAccount, toAccount, transferAmount, "transfer round %d" % (round), exitOnError=False, reportStatus=False)
-            if trans is None:
-                # delay and see if transfer is accepted now
-                Utils.Print("Transfer rejected, delay 1 second and see if it is then accepted")
-                time.sleep(1)
-                trans=node.transferFunds(fromAccount, toAccount, transferAmount, "transfer round %d" % (round), exitOnError=False, reportStatus=False)
-
-            assert trans is not None, Print("ERROR: failed round: %d, fromAccount: %s, toAccount: %s" % (round, accountIndex, toAccountIndex))
-            # store off the transaction id, which we can use with the node.transCache
-            history.append(Node.getTransId(trans))
+            trans = None
+            attempts = 0
+            maxAttempts = maxTransactionAttempts if not args.send_duplicates else maxTransactionAttemptsNoSend  # for send_duplicates we are just constructing a transaction, so should never require a second attempt
+            # can try up to maxAttempts times to send the transfer
+            while trans is None and attempts < maxAttempts:
+                if attempts > 0:
+                    # delay and see if transfer is accepted now
+                    Utils.Print("Transfer rejected, delay 1 second and see if it is then accepted")
+                    time.sleep(1)
+                expiration=None if args.send_duplicates else 90
+                trans=node.transferFunds(fromAccount, toAccount, transferAmount, "transfer round %d" % (round), exitOnError=False, reportStatus=False, sign=True, dontSend=args.send_duplicates, expiration=expiration)
+                attempts += 1
+
+            if args.send_duplicates:
+                sendTrans = trans
+                trans = None
+                numAccepted = 0
+                attempts = 0
+                while trans is None and attempts < maxTransactionAttempts:
+                    for node in map(lambda ordinal: allNodes[ordinal], nodeOrder):
+                        repeatTrans = node.pushTransaction(sendTrans, opts="--skip-sign", silentErrors=True)
+                        if repeatTrans is not None:
+                            if trans is None and repeatTrans[0]:
+                                trans = repeatTrans[1]
+                                transId = Node.getTransId(trans)
+
+                            numAccepted += 1
+
+                    attempts += 1
+
+            assert trans is not None, "ERROR: failed round: %d, fromAccount: %s, toAccount: %s" % (round, accountIndex, toAccountIndex)
+            transId = Node.getTransId(trans)
+            history.append(transId)
 
     nextTime = time.perf_counter()
     Print("Sending transfers took %s sec" % (nextTime - startTransferTime))
@@ -288,11 +333,11 @@ def findTransInBlock(transId, transToBlock, node):
                 if newestBlockNum > lastBlockNum:
                     missingTransactions[-1]["highest_block_seen"] = newestBlockNum
             blockNum = Node.getTransBlockNum(trans)
-            assert blockNum is not None, Print("ERROR: could not retrieve block num from transId: %s, trans: %s" % (transId, json.dumps(trans, indent=2)))
+            assert blockNum is not None, "ERROR: could not retrieve block num from transId: %s, trans: %s" % (transId, json.dumps(trans, indent=2))
         else:
             block = transToBlock[transId]
             blockNum = block["block_num"]
-            assert blockNum is not None, Print("ERROR: could not retrieve block num for block retrieved for transId: %s, block: %s" % (transId, json.dumps(block, indent=2)))
+            assert blockNum is not None, "ERROR: could not retrieve block num for block retrieved for transId: %s, block: %s" % (transId, json.dumps(block, indent=2))
 
         if lastBlockNum is not None:
             if blockNum > lastBlockNum + transBlocksBehind or blockNum + transBlocksBehind < lastBlockNum: