Bound transaction trace polling in test harness

tests/TestHarness/queries.py

@@ -240,22 +240,34 @@ def isBlockFinalized(self, blockNum):
         return self.isBlockPresent(blockNum, blockType=BlockType.lib)
 
     def getTransaction(self, transId, silentErrors=False, exitOnError=False, delayedRetry=True):
+        """Fetch a transaction trace, bounding each poll so missing traces cannot stall bootstrap."""
         assert(isinstance(transId, str))
         exitOnErrorForDelayed=not delayedRetry and exitOnError
-        timeout=3
+        pollTimeout=3
+        pollBudget=60
+        pollDeadline=time.perf_counter() + pollBudget
         cmdDesc="get transaction_trace"
         cmd="%s %s" % (cmdDesc, transId)
-        msg="(transaction id=%s)" % (transId);
-        for i in range(0,(int(60/timeout) - 1)):
-            trans=self.processClioCmd(cmd, cmdDesc, silentErrors=True, exitOnError=exitOnErrorForDelayed, exitMsg=msg)
+        msg="(transaction id=%s)" % (transId)
+        while True:
+            remainingBudget=pollDeadline - time.perf_counter()
+            finalAttempt=remainingBudget <= pollTimeout
+            attemptStart=time.perf_counter()
+            trans=self.processClioCmd(cmd, cmdDesc,
+                                      silentErrors=silentErrors if finalAttempt else True,
+                                      exitOnError=exitOnError if finalAttempt else exitOnErrorForDelayed,
+                                      exitMsg=msg, timeout=pollTimeout)
             if trans is not None or not delayedRetry:
                 return trans
+            if finalAttempt:
+                break
             if Utils.Debug: Utils.Print("Could not find transaction with id %s, delay and retry" % (transId))
-            time.sleep(timeout)
+            attemptElapsed=time.perf_counter() - attemptStart
+            time.sleep(max(0, pollTimeout - attemptElapsed))
 
         self.missingTransaction=True
         # either it is there or the transaction has timed out
-        return self.processClioCmd(cmd, cmdDesc, silentErrors=silentErrors, exitOnError=exitOnError, exitMsg=msg)
+        return None
 
     def isTransInBlock(self, transId, blockId, exitOnError=False):
         """Check if transId is within block identified by blockId"""
@@ -561,7 +573,8 @@ def getTableColumns(self, contract, scope, table):
         keys=list(row["value"].keys())
         return keys
 
-    def processClioCmd(self, cmd, cmdDesc, silentErrors=True, exitOnError=False, exitMsg=None, returnType=ReturnType.json):
+    def processClioCmd(self, cmd, cmdDesc, silentErrors=True, exitOnError=False, exitMsg=None, returnType=ReturnType.json, timeout=None):
+        """Run clio and decode the requested return type, optionally bounding the subprocess runtime."""
         assert(isinstance(returnType, ReturnType))
         cmd="%s %s %s" % (Utils.SysClientPath, self.sysClientArgs(), cmd)
         if Utils.Debug: Utils.Print("cmd: %s" % (cmd))
@@ -574,26 +587,40 @@ def processClioCmd(self, cmd, cmdDesc, silentErrors=True, exitOnError=False, exi
         start=time.perf_counter()
         while retries > 0:
             retries -= 1
+            attemptStart=time.perf_counter()
             try:
                 if returnType==ReturnType.json:
-                    trans=Utils.runCmdReturnJson(cmd, silentErrors=silentErrors)
+                    trans=Utils.runCmdReturnJson(cmd, silentErrors=silentErrors, timeout=timeout)
                 elif returnType==ReturnType.raw:
-                    trans=Utils.runCmdReturnStr(cmd)
+                    trans=Utils.runCmdReturnStr(cmd, timeout=timeout)
                 else:
                     unhandledEnumType(returnType)
 
                 if Utils.Debug:
                     end=time.perf_counter()
-                    Utils.Print("cmd Duration: %.3f sec" % (end-start))
+                    Utils.Print("cmd Duration: %.3f sec" % (end-attemptStart))
             except subprocess.CalledProcessError as ex:
                 if not silentErrors:
                     end=time.perf_counter()
-                    out=ex.output.decode("utf-8")
-                    msg=ex.stderr.decode("utf-8")
+                    out=Utils.decodeProcessOutput(ex.output)
+                    msg=Utils.decodeProcessOutput(ex.stderr)
                     if retries > 0 and "tx_cpu_usage_exceeded" in out:
                         Utils.Print(f"Retrying {cmdDesc} due to: tx_cpu_usage_exceeded")
                         continue # try again
-                    errorMsg="Exception during \"%s\". Exception message: %s.  stdout: %s.  cmd Duration=%.3f sec. %s" % (cmdDesc, msg, out, end-start, exitMsg)
+                    errorMsg="Exception during \"%s\". Exception message: %s.  stdout: %s.  cmd Duration=%.3f sec. %s" % (cmdDesc, msg, out, end-attemptStart, exitMsg)
+                    if exitOnError:
+                        Utils.cmdError(errorMsg)
+                        Utils.errorExit(errorMsg)
+                    else:
+                        Utils.Print("ERROR: %s" % (errorMsg))
+                return None
+            except subprocess.TimeoutExpired as ex:
+                if not silentErrors:
+                    end=time.perf_counter()
+                    out=Utils.decodeProcessOutput(ex.output)
+                    msg=Utils.decodeProcessOutput(ex.stderr)
+                    errorMsg=("Timeout during \"%s\" after %.3f sec. cmd timeout=%s. stderr: %s. stdout: %s. %s" %
+                              (cmdDesc, end-attemptStart, ex.timeout, msg, out, exitMsg))
                     if exitOnError:
                         Utils.cmdError(errorMsg)
                         Utils.errorExit(errorMsg)

tests/TestHarness/testUtils.py

@@ -197,9 +197,12 @@ def getChainStrategies():
         return chainSyncStrategies
 
     @staticmethod
-    def checkOutput(cmd, ignoreError=False):
+    def checkOutput(cmd, ignoreError=False, timeout=None):
+        """Run a command and return stdout, optionally bounding the subprocess runtime."""
+        assert timeout is None or isinstance(cmd, list), (
+            "timeout with shell command strings can leave orphaned children")
         popen = Utils.delayedCheckOutput(cmd)
-        return Utils.checkDelayedOutput(popen, cmd, ignoreError=ignoreError)
+        return Utils.checkDelayedOutput(popen, cmd, ignoreError=ignoreError, timeout=timeout)
 
     @staticmethod
     def delayedCheckOutput(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE):
@@ -210,15 +213,27 @@ def delayedCheckOutput(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE):
         return popen
 
     @staticmethod
-    def checkDelayedOutput(popen, cmd, ignoreError=False):
+    def checkDelayedOutput(popen, cmd, ignoreError=False, timeout=None):
+        """Collect a subprocess result and terminate it if it exceeds the requested timeout."""
         assert isinstance(popen, subprocess.Popen)
         assert isinstance(cmd, (str,list))
         start=Utils.timestamp()
-        (output,error)=popen.communicate()
+        try:
+            (output,error)=popen.communicate(timeout=timeout)
+        except subprocess.TimeoutExpired as ex:
+            popen.kill()
+            (output,error)=popen.communicate()
+            Utils.checkOutputFileWrite(start, cmd, output, error)
+            raise subprocess.TimeoutExpired(cmd=cmd, timeout=ex.timeout, output=output, stderr=error) from ex
         Utils.checkOutputFileWrite(start, cmd, output, error)
         if popen.returncode != 0 and not ignoreError:
             raise subprocess.CalledProcessError(returncode=popen.returncode, cmd=cmd, output=output, stderr=error)
-        return output.decode("utf-8") if popen.returncode == 0 else error.decode("utf-8")
+        return Utils.decodeProcessOutput(output) if popen.returncode == 0 else Utils.decodeProcessOutput(error)
+
+    @staticmethod
+    def decodeProcessOutput(output):
+        """Decode subprocess bytes without hiding diagnostics if a killed process wrote partial UTF-8."""
+        return output.decode("utf-8", errors="replace") if output is not None else ""
 
     @staticmethod
     def errorExit(msg="", raw=False, errorCode=1):
@@ -315,28 +330,33 @@ def toJson(retStr, trace=False, silentErrors=True):
             raise
 
     @staticmethod
-    def runCmdArrReturnJson(cmdArr, trace=False, silentErrors=True):
-        retStr=Utils.checkOutput(cmdArr)
+    def runCmdArrReturnJson(cmdArr, trace=False, silentErrors=True, timeout=None):
+        """Run a command array and parse its JSON output, optionally with a subprocess timeout."""
+        retStr=Utils.checkOutput(cmdArr, timeout=timeout)
         return Utils.toJson(retStr, trace, silentErrors)
 
     @staticmethod
-    def runCmdReturnStr(cmd, trace=False, ignoreError=False):
+    def runCmdReturnStr(cmd, trace=False, ignoreError=False, timeout=None):
+        """Run a shell command string and return stdout, optionally with a subprocess timeout."""
         cmdArr=shlex.split(cmd)
-        return Utils.runCmdArrReturnStr(cmdArr, ignoreError=ignoreError)
+        return Utils.runCmdArrReturnStr(cmdArr, ignoreError=ignoreError, timeout=timeout)
 
     @staticmethod
-    def runCmdArrReturnStr(cmdArr, trace=False, ignoreError=False):
-        retStr=Utils.checkOutput(cmdArr, ignoreError=ignoreError)
+    def runCmdArrReturnStr(cmdArr, trace=False, ignoreError=False, timeout=None):
+        """Run a command array and return stdout, optionally with a subprocess timeout."""
+        retStr=Utils.checkOutput(cmdArr, ignoreError=ignoreError, timeout=timeout)
         if trace: Utils.Print ("RAW > %s" % (retStr))
         return retStr
 
     @staticmethod
-    def runCmdReturnJson(cmd, trace=False, silentErrors=False):
+    def runCmdReturnJson(cmd, trace=False, silentErrors=False, timeout=None):
+        """Run a shell command string and parse its JSON output, optionally with a subprocess timeout."""
         cmdArr=shlex.split(cmd)
-        return Utils.runCmdArrReturnJson(cmdArr, trace=trace, silentErrors=silentErrors)
+        return Utils.runCmdArrReturnJson(cmdArr, trace=trace, silentErrors=silentErrors, timeout=timeout)
 
     @staticmethod
-    def processSysioUtilCmd(cmd, cmdDesc, silentErrors=True, exitOnError=False, exitMsg=None):
+    def processSysioUtilCmd(cmd, cmdDesc, silentErrors=True, exitOnError=False, exitMsg=None, timeout=None):
+        """Run sys-util and return stdout, optionally bounding the subprocess runtime."""
         cmd="%s %s" % (Utils.SysioClientPath, cmd)
         if Utils.Debug: Utils.Print("cmd: %s" % (cmd))
         if exitMsg is not None:
@@ -346,22 +366,35 @@ def processSysioUtilCmd(cmd, cmdDesc, silentErrors=True, exitOnError=False, exit
         output=None
         start=time.perf_counter()
         try:
-            output=Utils.runCmdReturnStr(cmd)
+            output=Utils.runCmdReturnStr(cmd, timeout=timeout)
 
             if Utils.Debug:
                 end=time.perf_counter()
                 Utils.Print("cmd Duration: %.3f sec" % (end-start))
         except subprocess.CalledProcessError as ex:
             if not silentErrors:
                 end=time.perf_counter()
-                msg=ex.stderr.decode("utf-8")
+                msg=Utils.decodeProcessOutput(ex.stderr)
                 errorMsg="Exception during \"%s\". Exception message: %s.  cmd Duration=%.3f sec. %s" % (cmdDesc, msg, end-start, exitMsg)
                 if exitOnError:
                     Utils.cmdError(errorMsg)
                     Utils.errorExit(errorMsg)
                 else:
                     Utils.Print("ERROR: %s" % (errorMsg))
             return None
+        except subprocess.TimeoutExpired as ex:
+            if not silentErrors:
+                end=time.perf_counter()
+                out=Utils.decodeProcessOutput(ex.output)
+                msg=Utils.decodeProcessOutput(ex.stderr)
+                errorMsg=("Timeout during \"%s\" after %.3f sec. cmd timeout=%s. stderr: %s. stdout: %s. %s" %
+                          (cmdDesc, end-start, ex.timeout, msg, out, exitMsg))
+                if exitOnError:
+                    Utils.cmdError(errorMsg)
+                    Utils.errorExit(errorMsg)
+                else:
+                    Utils.Print("ERROR: %s" % (errorMsg))
+            return None
 
         if exitOnError and output is None:
             Utils.cmdError("could not \"%s\". %s" % (cmdDesc,exitMsg))

tests/TestHarness/transactions.py

@@ -12,6 +12,8 @@
 
 class Transactions(NodeopQueries):
     retry_num_blocks_default = 1
+    # Default clio action-push subprocess timeout, in seconds, used to bound a stuck push.
+    push_message_timeout_default = 45
 
     def __init__(self, host, port, walletMgr=None):
         super().__init__(host, port, walletMgr)
@@ -121,7 +123,7 @@ def transferFunds(self, source, destination, amountStr, memo="memo", waitForTran
                 self.trackCmdTransaction(trans, reportStatus=reportStatus)
         except subprocess.CalledProcessError as ex:
             end=time.perf_counter()
-            msg=ex.stderr.decode("utf-8")
+            msg=Utils.decodeProcessOutput(ex.stderr)
             Utils.Print("ERROR: Exception during funds transfer.  cmd Duration: %.3f sec.  %s" % (end-start, msg))
             if exitOnError:
                 Utils.cmdError("could not transfer \"%s\" from %s to %s" % (amountStr, source, destination))
@@ -261,7 +263,11 @@ def pushTransaction(self, trans, opts="", silentErrors=False, permissions=None):
             return (False, msg)
 
     # returns tuple with transaction execution status and transaction
-    def pushMessage(self, account, action, data, opts, silentErrors=False, signatures=None, expectTrxTrace=True):
+    def pushMessage(self, account, action, data, opts, silentErrors=False, signatures=None, expectTrxTrace=True,
+                    timeout=None):
+        """Push an action with clio, bounding the clio subprocess runtime by default."""
+        if timeout is None:
+            timeout = self.push_message_timeout_default
         cmd="%s %s push action -j %s %s" % (Utils.SysClientPath, self.sysClientArgs(), account, action)
         cmdArr=cmd.split()
         # not using sign_str, since cmdArr messes up the string
@@ -278,22 +284,33 @@ def pushMessage(self, account, action, data, opts, silentErrors=False, signature
         while retries > 0:
             retries -= 1
             try:
-                trans=Utils.runCmdArrReturnJson(cmdArr)
+                trans=Utils.runCmdArrReturnJson(cmdArr, timeout=timeout)
                 self.trackCmdTransaction(trans, ignoreNonTrans=True)
                 if Utils.Debug:
                     end=time.perf_counter()
                     Utils.Print("cmd Duration: %.3f sec" % (end-start))
                 return (NodeopQueries.getTransStatus(trans) == 'executed' if expectTrxTrace else True, trans)
             except subprocess.CalledProcessError as ex:
-                msg=ex.stderr.decode("utf-8")
-                output=ex.output.decode("utf-8")
+                msg=Utils.decodeProcessOutput(ex.stderr)
+                output=Utils.decodeProcessOutput(ex.output)
                 if not silentErrors:
                     end=time.perf_counter()
                     Utils.Print("ERROR: Exception during push message. retry %s. stderr: %s. stdout: %s.  cmd Duration=%.3f sec." % (retries, msg, output, end - start))
                 if retries > 0 and "tx_cpu_usage_exceeded" in output:
                     Utils.Print(f"Retrying {cmd} due to: tx_cpu_usage_exceeded")
                     continue # try again
                 return (False, msg)
+            except subprocess.TimeoutExpired as ex:
+                msg=str(ex)
+                output=Utils.decodeProcessOutput(ex.output)
+                error=Utils.decodeProcessOutput(ex.stderr)
+                if not silentErrors:
+                    end=time.perf_counter()
+                    Utils.Print(
+                        "ERROR: Timeout during push message. retry %s. cmd timeout=%s. stderr: %s. stdout: %s. cmd Duration=%.3f sec." %
+                        (retries, ex.timeout, error, output, end - start))
+                # The killed clio process may have already broadcast the transaction before timing out; do not retry.
+                return (False, msg)
 
     def setPermission(self, account, code, pType, requirement, waitForTransBlock=False, exitOnError=False, sign=False):
         assert(isinstance(account, Account))