Merge pull request #21 from derekpierre/timeout-strategy

Modifications/Testing of `TimeoutStrategy`

atxm/machine.py

@@ -14,7 +14,7 @@
     AsyncTxStrategy,
     FixedRateSpeedUp,
     InsufficientFundsPause,
-    TimeoutPause,
+    TimeoutStrategy,
 )
 from atxm.tracker import _TxTracker
 from atxm.tx import (
@@ -91,7 +91,7 @@ class _Machine(StateMachine):
 
     STRATEGIES: List[Type[AsyncTxStrategy]] = [
         InsufficientFundsPause,
-        TimeoutPause,
+        TimeoutStrategy,
         FixedRateSpeedUp,
     ]
 

atxm/strategies.py

@@ -1,5 +1,5 @@
-import time
 from abc import ABC
+from datetime import datetime, timedelta
 from typing import Tuple, Optional
 
 from web3 import Web3
@@ -84,46 +84,53 @@ def execute(self, pending: PendingTx) -> TxParams:
         return pending.params
 
 
-class TimeoutPause(AsyncTxStrategy):
-    """Pause strategy for pending transactions."""
+class TimeoutStrategy(AsyncTxStrategy):
+    """Timeout strategy for pending transactions."""
 
     _NAME = "timeout"
+
     _TIMEOUT = 60 * 60  # 1 hour in seconds
 
+    _WARN_FACTOR = 0.05  # 10% of timeout remaining
+
     def __init__(self, w3: Web3, timeout: Optional[int] = None):
         super().__init__(w3)
         self.timeout = timeout or self._TIMEOUT
 
     def __active_timed_out(self, pending: PendingTx) -> bool:
         """Returns True if the active transaction has timed out."""
-        if not pending:
-            return False
-        timeout = (time.time() - pending.created) > self.timeout
-        if timeout:
+        # seconds specificity (ignore microseconds)
+        now = datetime.now().replace(microsecond=0)
+        creation_time = datetime.fromtimestamp(pending.created).replace(microsecond=0)
+
+        elapsed_time = now - creation_time
+        if elapsed_time.seconds > self.timeout:
             return True
 
-        time_remaining = round(self.timeout - (time.time() - pending.created))
-        minutes = round(time_remaining / 60)
-        remainder_seconds = time_remaining % 60
-        end_time = time.time() + time_remaining
-        human_end_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(end_time))
-        if time_remaining < (60 * 2):
+        end_time = creation_time + timedelta(seconds=self.timeout)
+        time_remaining = end_time - now
+        human_end_time = end_time.strftime("%Y-%m-%d %H:%M:%S")
+        if time_remaining.seconds < (self.timeout * self._WARN_FACTOR):
             self.log.warn(
-                f"[timeout] Transaction {pending.txhash.hex()} will timeout in "
-                f"{minutes}m{remainder_seconds}s at {human_end_time}"
+                f"[pending_timeout] Transaction {pending.txhash.hex()} will timeout in "
+                f"{time_remaining} at {human_end_time}"
             )
         else:
             self.log.info(
-                f"[pending] {pending.txhash.hex()} \n"
-                f"[pending] {round(time.time() - pending.created)}s Elapsed | "
-                f"{minutes}m{remainder_seconds}s Remaining | "
+                f"[pending] {pending.txhash.hex()} "
+                f"{elapsed_time.seconds}s Elapsed | "
+                f"{time_remaining} Remaining | "
                 f"Timeout at {human_end_time}"
             )
         return False
 
     def execute(self, pending: PendingTx) -> TxParams:
-        timeout = self.__active_timed_out(pending)
-        if timeout:
+        if not pending:
+            # should never get here
+            raise RuntimeError("pending tx should not be None")
+
+        timedout = self.__active_timed_out(pending)
+        if timedout:
             raise TransactionFaulted(
                 tx=pending,
                 fault=Fault.TIMEOUT,

tests/test_strategy.py

@@ -0,0 +1,77 @@
+from datetime import datetime, timedelta
+
+import pytest
+from hexbytes import HexBytes
+from twisted.logger import LogLevel, globalLogPublisher
+from web3.types import TxParams
+
+from atxm.exceptions import Fault, TransactionFaulted
+from atxm.strategies import TimeoutStrategy
+from atxm.tx import PendingTx
+
+
+def test_timeout_strategy(w3, mocker):
+    TIMEOUT = 600  # 10 mins
+
+    # default timeout
+    timeout_strategy = TimeoutStrategy(w3)
+    assert timeout_strategy.timeout == TimeoutStrategy._TIMEOUT
+
+    # specific timeout
+    timeout_strategy = TimeoutStrategy(w3, timeout=TIMEOUT)
+    assert timeout_strategy.timeout == TIMEOUT
+
+    assert timeout_strategy.name == timeout_strategy._NAME
+
+    # None tx does not time out
+    with pytest.raises(RuntimeError):
+        assert timeout_strategy.execute(None)
+
+    # mock pending tx
+    pending_tx = mocker.Mock(spec=PendingTx)
+    pending_tx.txhash = HexBytes("0xdeadbeef")
+    params = mocker.Mock(spec=TxParams)
+    pending_tx.params = params
+    now = datetime.now()
+    pending_tx.created = now.timestamp()
+
+    # can't mock datetime, so instead use creation time to manipulate scenarios
+
+    # 1) tx just created a does not time out
+    for i in range(3):
+        assert timeout_strategy.execute(pending_tx) == params
+
+    # 2) remaining time is < warn factor; still doesn't time out but we warn about it
+    pending_tx.created = (
+        now - timedelta(seconds=(TIMEOUT * (1 - timeout_strategy._WARN_FACTOR) + 1))
+    ).timestamp()
+
+    warnings = []
+
+    def warning_trapper(event):
+        if event["log_level"] == LogLevel.warn:
+            warnings.append(event)
+
+    globalLogPublisher.addObserver(warning_trapper)
+    assert timeout_strategy.execute(pending_tx) == params
+    globalLogPublisher.removeObserver(warning_trapper)
+
+    assert len(warnings) == 1
+    warning = warnings[0]["log_format"]
+    assert (
+        f"[pending_timeout] Transaction {pending_tx.txhash.hex()} will timeout in"
+        in warning
+    )
+
+    # 3) close to timeout but not quite (5s short)
+    pending_tx.created = (now - timedelta(seconds=(TIMEOUT - 5))).timestamp()
+    assert timeout_strategy.execute(pending_tx) == params
+
+    # 4) timeout
+    pending_tx.created = (now - timedelta(seconds=(TIMEOUT + 1))).timestamp()
+    with pytest.raises(TransactionFaulted) as exc_info:
+        timeout_strategy.execute(pending_tx)
+    e = exc_info.value
+    e.tx = pending_tx
+    e.fault = Fault.TIMEOUT
+    e.message = "Transaction has timed out"