Merge pull request #168 from valory-xyz/fix/redeeming

Fix/redeeming

packages/packages.json

@@ -1,17 +1,17 @@
 {
     "dev": {
         "skill/valory/market_manager_abci/0.1.0": "bafybeietothjpt3472icsfoelihq437dclsckldsstz4sui46lblxfugsq",
-        "skill/valory/decision_maker_abci/0.1.0": "bafybeic474g73pukavqdpqu5pocsyhfa4eididi6v36jbzo53c3lo4tmfy",
-        "skill/valory/trader_abci/0.1.0": "bafybeieu5ufh3hvksdhzzur7hwfzz3dohqlehvuavbet3xpbw4dmzbf2q4",
+        "skill/valory/decision_maker_abci/0.1.0": "bafybeiepex3nukg5w6x75xngdiimjettpakpti64mnpfmjoapomt2hxqam",
+        "skill/valory/trader_abci/0.1.0": "bafybeibuaonngf33m63co5ezcerovfvjdfdit4gz3l44a3c5dl52nskfxq",
         "contract/valory/market_maker/0.1.0": "bafybeiehszwvmurz5b5vpqbxbvcm5ty4z553jqpsnernqwdq6d62r3z5wa",
-        "agent/valory/trader/0.1.0": "bafybeifn45eermdfe66i3tx3aevvtngsr3punhu55kvrpmlk5ut3cgzif4",
-        "service/valory/trader/0.1.0": "bafybeiccwofazsng2vgws22ior4dxouwk7f75teqx6y6xdvof75xpehera",
+        "agent/valory/trader/0.1.0": "bafybeic3a2dxfw4xvg7ptunpnjxnji73qdvl7l2fa3pgto4k4a2k7hr4by",
+        "service/valory/trader/0.1.0": "bafybeig47lvkke2vg22obi3vkhbyr6kjxiqvxpxlsku4kzlidi6unhqmlq",
         "contract/valory/erc20/0.1.0": "bafybeihisiakfcftxs2hwx7yxzovao4qjuk3kqbzntiekzv5v75vpax3ui",
-        "skill/valory/tx_settlement_multiplexer_abci/0.1.0": "bafybeihkjceuiw6cwaul5a5ngydma7gik46cgp2cinhibhsysv7qvc3mba",
-        "contract/valory/mech/0.1.0": "bafybeieo65rtidzgrx22la7z7azoqbisd35hnoqfw7l3xxctotvctjlkju",
-        "contract/valory/realitio/0.1.0": "bafybeieoily22pasgkzzcyawhqyrsko7b52lghnvwuskhomavb7tlb6pxa",
+        "skill/valory/tx_settlement_multiplexer_abci/0.1.0": "bafybeihvwp3rlxaepfggkzi6ptcojbpayaul2hrruawz4pcjpuqfutcjqi",
+        "contract/valory/mech/0.1.0": "bafybeigvha4ya64ccjl7tbroa3zjqlxm64wxsqx5k2cz2xs3a7igu5qgxa",
+        "contract/valory/realitio/0.1.0": "bafybeicllaewp72lf5mlohsgu2h2gqemaltn7lkbzotwlq3wyg3wk2nunm",
         "contract/valory/realitio_proxy/0.1.0": "bafybeidx37xzjjmapwacedgzhum6grfzhp5vhouz4zu3pvpgdy5pgb2fr4",
-        "contract/valory/conditional_tokens/0.1.0": "bafybeifu5axib5ifzq6bomfscs7nnx5qknkzymlz6gfn7ohjsb2shghrei",
+        "contract/valory/conditional_tokens/0.1.0": "bafybeigucumqbsk74nj4rpm4p2cpiky4dj6uws7nfmgpimuviaxcamwqnu",
         "contract/valory/agent_registry/0.1.0": "bafybeicznp6b7t7n6gju35xa2mr2vo3vshtj72l5odwhtdvz3uard27csq",
         "contract/valory/service_staking_token/0.1.0": "bafybeifrbsyszkg6hhqnx32ha2wwyidi2nzaece74niopsnzmb4szbpahi",
         "skill/valory/staking_abci/0.1.0": "bafybeiebu4qeblyjbacjvkiiiorxb4p3yy4nuhvsssgbailmqqtgrhwcca"

packages/valory/agents/trader/aea-config.yaml

@@ -21,9 +21,9 @@ contracts:
 - valory/market_maker:0.1.0:bafybeiehszwvmurz5b5vpqbxbvcm5ty4z553jqpsnernqwdq6d62r3z5wa
 - valory/erc20:0.1.0:bafybeihisiakfcftxs2hwx7yxzovao4qjuk3kqbzntiekzv5v75vpax3ui
 - valory/multisend:0.1.0:bafybeig5byt5urg2d2bsecufxe5ql7f4mezg3mekfleeh32nmuusx66p4y
-- valory/mech:0.1.0:bafybeieo65rtidzgrx22la7z7azoqbisd35hnoqfw7l3xxctotvctjlkju
-- valory/conditional_tokens:0.1.0:bafybeifu5axib5ifzq6bomfscs7nnx5qknkzymlz6gfn7ohjsb2shghrei
-- valory/realitio:0.1.0:bafybeieoily22pasgkzzcyawhqyrsko7b52lghnvwuskhomavb7tlb6pxa
+- valory/mech:0.1.0:bafybeigvha4ya64ccjl7tbroa3zjqlxm64wxsqx5k2cz2xs3a7igu5qgxa
+- valory/conditional_tokens:0.1.0:bafybeigucumqbsk74nj4rpm4p2cpiky4dj6uws7nfmgpimuviaxcamwqnu
+- valory/realitio:0.1.0:bafybeicllaewp72lf5mlohsgu2h2gqemaltn7lkbzotwlq3wyg3wk2nunm
 - valory/realitio_proxy:0.1.0:bafybeidx37xzjjmapwacedgzhum6grfzhp5vhouz4zu3pvpgdy5pgb2fr4
 - valory/agent_registry:0.1.0:bafybeicznp6b7t7n6gju35xa2mr2vo3vshtj72l5odwhtdvz3uard27csq
 - valory/service_staking_token:0.1.0:bafybeifrbsyszkg6hhqnx32ha2wwyidi2nzaece74niopsnzmb4szbpahi
@@ -43,10 +43,10 @@ skills:
 - valory/reset_pause_abci:0.1.0:bafybeidzajbe3erygeh2xbd6lrjv7nsptznjuzrt24ykgvhgotdeyhfnba
 - valory/termination_abci:0.1.0:bafybeie4zvjfxvdu7qrulmur3chpjz3kpj5m4bjsxvpk4gvj5zbyyayfaa
 - valory/transaction_settlement_abci:0.1.0:bafybeiaefgqbs7zsn5xe5kdwrujj7ivygkn3ujpw6crnvi3knvxw75qmja
-- valory/tx_settlement_multiplexer_abci:0.1.0:bafybeihkjceuiw6cwaul5a5ngydma7gik46cgp2cinhibhsysv7qvc3mba
+- valory/tx_settlement_multiplexer_abci:0.1.0:bafybeihvwp3rlxaepfggkzi6ptcojbpayaul2hrruawz4pcjpuqfutcjqi
 - valory/market_manager_abci:0.1.0:bafybeietothjpt3472icsfoelihq437dclsckldsstz4sui46lblxfugsq
-- valory/decision_maker_abci:0.1.0:bafybeic474g73pukavqdpqu5pocsyhfa4eididi6v36jbzo53c3lo4tmfy
-- valory/trader_abci:0.1.0:bafybeieu5ufh3hvksdhzzur7hwfzz3dohqlehvuavbet3xpbw4dmzbf2q4
+- valory/decision_maker_abci:0.1.0:bafybeiepex3nukg5w6x75xngdiimjettpakpti64mnpfmjoapomt2hxqam
+- valory/trader_abci:0.1.0:bafybeibuaonngf33m63co5ezcerovfvjdfdit4gz3l44a3c5dl52nskfxq
 - valory/staking_abci:0.1.0:bafybeiebu4qeblyjbacjvkiiiorxb4p3yy4nuhvsssgbailmqqtgrhwcca
 default_ledger: ethereum
 required_ledgers:
@@ -207,6 +207,7 @@ models:
       agent_balance_threshold: ${int:10000000000000000}
       refill_check_interval: ${int:10}
       tool_punishment_multiplier: ${int:1}
+      contract_timeout: ${float:300.0}
 ---
 public_id: valory/p2p_libp2p_client:0.1.0
 type: connection

packages/valory/contracts/conditional_tokens/contract.py

@@ -18,8 +18,8 @@
 # ------------------------------------------------------------------------------
 
 """This module contains the conditional tokens contract definition."""
-
-from typing import List
+import concurrent.futures
+from typing import List, Any, Dict, Union, Callable
 
 from requests.exceptions import ReadTimeout as RequestsReadTimeoutError
 from urllib3.exceptions import ReadTimeoutError as Urllib3ReadTimeoutError
@@ -30,11 +30,39 @@
 from hexbytes import HexBytes
 
 
+FIVE_MINUTES = 300.0
+
 class ConditionalTokensContract(Contract):
     """The ConditionalTokens smart contract."""
 
     contract_id = PublicId.from_str("valory/conditional_tokens:0.1.0")
 
+    @staticmethod
+    def execute_with_timeout(func: Callable, timeout: float) -> Any:
+        """Execute a function with a timeout."""
+
+        # Create a ProcessPoolExecutor with a maximum of 1 worker (process)
+        with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
+            # Submit the function to the executor
+            future = executor.submit(
+                func,
+            )
+
+            try:
+                # Wait for the result with a 5-minute timeout
+                data = future.result(timeout=timeout)
+            except TimeoutError:
+                # Handle the case where the execution times out
+                err = f"The RPC didn't respond in {timeout}."
+                return None, err
+
+            # Check if an error occurred
+            if isinstance(data, str):
+                # Handle the case where the execution failed
+                return None, data
+
+            return data, None
+
     @classmethod
     def check_redeemed(
         cls,
@@ -47,31 +75,41 @@ def check_redeemed(
         parent_collection_ids: List[bytes],
         condition_ids: List[HexBytes],
         index_sets: List[List[int]],
+        timeout: float = FIVE_MINUTES,
     ) -> JSONLike:
         """Filter to find out whether a position has already been redeemed."""
-        contract_instance = cls.get_instance(ledger_api, contract_address)
-        to_checksum = ledger_api.api.to_checksum_address
-        redeemer_checksummed = to_checksum(redeemer)
-        collateral_tokens_checksummed = [
-            to_checksum(token) for token in collateral_tokens
-        ]
-        try:
-            payout_filter = contract_instance.events.PayoutRedemption.build_filter()
-            payout_filter.args.redeemer.match_single(redeemer_checksummed)
-            payout_filter.args.collateralToken.match_any(*collateral_tokens_checksummed)
-            payout_filter.args.parentCollectionId.match_any(*parent_collection_ids)
-            payout_filter.args.conditionId.match_any(*condition_ids)
-            payout_filter.args.indexSets.match_any(*index_sets)
-            payout_filter.fromBlock = from_block
-            payout_filter.toBlock = to_block
-            redeemed = list(payout_filter.deploy(ledger_api.api).get_all_entries())
-        except (Urllib3ReadTimeoutError, RequestsReadTimeoutError):
-            msg = (
-                "The RPC timed out! This usually happens if the filtering is too wide. "
-                f"The service tried to filter from block {from_block} to {to_block}. "
-                f"If this issue persists, please try lowering the `EVENT_FILTERING_BATCH_SIZE`!"
-            )
-            return dict(error=msg)
+
+        def get_redeem_events() -> Union[List[Dict[str, Any]], str]:
+            """Get the redeem events."""
+            contract_instance = cls.get_instance(ledger_api, contract_address)
+            to_checksum = ledger_api.api.to_checksum_address
+            redeemer_checksummed = to_checksum(redeemer)
+            collateral_tokens_checksummed = [
+                to_checksum(token) for token in collateral_tokens
+            ]
+            try:
+                payout_filter = contract_instance.events.PayoutRedemption.build_filter()
+                payout_filter.args.redeemer.match_single(redeemer_checksummed)
+                payout_filter.args.collateralToken.match_any(*collateral_tokens_checksummed)
+                payout_filter.args.parentCollectionId.match_any(*parent_collection_ids)
+                payout_filter.args.conditionId.match_any(*condition_ids)
+                payout_filter.args.indexSets.match_any(*index_sets)
+                payout_filter.fromBlock = from_block
+                payout_filter.toBlock = to_block
+                redeemed = list(payout_filter.deploy(ledger_api.api).get_all_entries())
+                return redeemed
+
+            except (Urllib3ReadTimeoutError, RequestsReadTimeoutError):
+                msg = (
+                    "The RPC timed out! This usually happens if the filtering is too wide. "
+                    f"The service tried to filter from block {from_block} to {to_block}. "
+                    f"If this issue persists, please try lowering the `EVENT_FILTERING_BATCH_SIZE`!"
+                )
+                return msg
+
+        redeemed, err = cls.execute_with_timeout(get_redeem_events, timeout)
+        if err is not None:
+            return dict(error=err)
 
         payouts = {}
         for redeeming in redeemed:

packages/valory/contracts/conditional_tokens/contract.yaml

@@ -8,7 +8,7 @@ aea_version: '>=1.0.0, <2.0.0'
 fingerprint:
   __init__.py: bafybeidhdxio3oq5gqdnxmngumvt3fcd6zyiyrpk5f2k4dwhflbg4e5iky
   build/ConditionalTokens.json: bafybeia2ahis7zx2yhhf23kpkcxu56hto6fwg6ptjg5ld46lp4dgz7cz3e
-  contract.py: bafybeic34mnydr5m7xncg42tgg5qrjxkltsxxudnvog4jjcs6elg64z5ty
+  contract.py: bafybeiesym65thjsazplbgjhqleci6sdriytubqljpc3smtjitrpthhaam
 fingerprint_ignore_patterns: []
 class_name: ConditionalTokensContract
 contract_interface_paths:

packages/valory/contracts/mech/contract.py

@@ -18,8 +18,8 @@
 # ------------------------------------------------------------------------------
 
 """This module contains the class to connect to a Mech contract."""
-
-from typing import Dict, Optional, cast, List, Any
+import concurrent.futures
+from typing import Dict, Optional, cast, List, Any, Callable
 
 from aea.common import JSONLike
 from aea.configurations.base import PublicId
@@ -30,13 +30,40 @@
 from web3.types import TxReceipt, EventData, BlockIdentifier, BlockData
 
 PUBLIC_ID = PublicId.from_str("valory/mech:0.1.0")
+FIVE_MINUTES = 300.0
 
 
 class Mech(Contract):
     """The Mech contract."""
 
     contract_id = PUBLIC_ID
 
+    @staticmethod
+    def execute_with_timeout(func: Callable, timeout: float) -> Any:
+        """Execute a function with a timeout."""
+
+        # Create a ProcessPoolExecutor with a maximum of 1 worker (process)
+        with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
+            # Submit the function to the executor
+            future = executor.submit(
+                func,
+            )
+
+            try:
+                # Wait for the result with a 5-minute timeout
+                data = future.result(timeout=timeout)
+            except TimeoutError:
+                # Handle the case where the execution times out
+                err = f"The RPC didn't respond in {timeout}."
+                return None, err
+
+            # Check if an error occurred
+            if isinstance(data, str):
+                # Handle the case where the execution failed
+                return None, data
+
+            return data, None
+
     @classmethod
     def get_price(
         cls,
@@ -154,36 +181,43 @@ def get_response(
         request_id: int,
         from_block: BlockIdentifier = "earliest",
         to_block: BlockIdentifier = "latest",
+        timeout: float = FIVE_MINUTES,
     ) -> JSONLike:
         """Filter the `Deliver` events emitted by the contract and get the data of the given `request_id`."""
-        ledger_api = cast(EthereumApi, ledger_api)
-        contract_instance = cls.get_instance(ledger_api, contract_address)
-
-        deliver_filter = contract_instance.events.Deliver.build_filter()
-        deliver_filter.fromBlock = from_block
-        deliver_filter.toBlock = to_block
-        deliver_filter.args.requestId.match_single(request_id)
-        delivered = list(deliver_filter.deploy(ledger_api.api).get_all_entries())
-        n_delivered = len(delivered)
-
-        if n_delivered == 0:
-            info = f"The mech ({contract_address}) has not delivered a response yet for request with id {request_id}."
-            return {"info": info}
-
-        if n_delivered != 1:
-            error = (
-                f"A single response was expected by the mech ({contract_address}) for request with id {request_id}. "
-                f"Received {n_delivered} responses: {delivered}."
-            )
-            return {"error": error}
-
-        delivered_event = delivered.pop()
-        deliver_args = delivered_event.get("args", None)
-        if deliver_args is None or "data" not in deliver_args:
-            error = f"The mech's response does not match the expected format: {delivered_event}"
-            return {"error": error}
-
-        return dict(data=deliver_args["data"])
+        def get_responses() -> Any:
+            """Get the responses from the contract."""
+            contract_instance = cls.get_instance(ledger_api, contract_address)
+            deliver_filter = contract_instance.events.Deliver.build_filter()
+            deliver_filter.fromBlock = from_block
+            deliver_filter.toBlock = to_block
+            deliver_filter.args.requestId.match_single(request_id)
+            delivered = list(deliver_filter.deploy(ledger_api.api).get_all_entries())
+            n_delivered = len(delivered)
+
+            if n_delivered == 0:
+                info = f"The mech ({contract_address}) has not delivered a response yet for request with id {request_id}."
+                return {"info": info}
+
+            if n_delivered != 1:
+                error = (
+                    f"A single response was expected by the mech ({contract_address}) for request with id {request_id}. "
+                    f"Received {n_delivered} responses: {delivered}."
+                )
+                return error
+
+            delivered_event = delivered.pop()
+            deliver_args = delivered_event.get("args", None)
+            if deliver_args is None or "data" not in deliver_args:
+                error = f"The mech's response does not match the expected format: {delivered_event}"
+                return error
+
+            return {"data": deliver_args["data"]}
+
+        data, err = cls.execute_with_timeout(get_responses, timeout=timeout)
+        if err is not None:
+            return {"error": err}
+
+        return data
 
     @classmethod
     def get_mech_id(

packages/valory/contracts/mech/contract.yaml

@@ -9,7 +9,7 @@ fingerprint:
   README.md: bafybeibl4uw7rs6mwh7zuvdnqmj2o2xyr7nx5qk3w7torwx3jg6farn6ca
   __init__.py: bafybeicx5pxh3cxnml2biuuoebvafvu5tvy6mgkzyjzuubuoeebb5yzjsm
   build/mech.json: bafybeia3zf47nhljgahe4ijvw6mfcswhn6r5ntw572y63hbse7bvgsydaq
-  contract.py: bafybeifbfa6p3jcwn6j7s5aiiqxb3ne4vbmvoggr5zpptmd727gpsjqjpe
+  contract.py: bafybeifnpv6d7vwxayii4rsoauc6ymzujbxc5b4sqj56e544tapcjr32ji
 fingerprint_ignore_patterns: []
 contracts: []
 class_name: Mech