Unconfirmed blocks

src/monitoring_service/events.py

@@ -4,22 +4,22 @@
 from raiden_libs.events import Event
 
 
-@dataclass
+@dataclass(frozen=True)
 class ScheduledEvent(Event):
     """ An event to be triggered a t a certain block number. """
 
     trigger_block_number: BlockNumber
     event: Event
 
 
-@dataclass
+@dataclass(frozen=True)
 class ActionMonitoringTriggeredEvent(Event):
     token_network_address: TokenNetworkAddress
     channel_identifier: ChannelID
     non_closing_participant: Address
 
 
-@dataclass
+@dataclass(frozen=True)
 class ActionClaimRewardTriggeredEvent(Event):
     token_network_address: TokenNetworkAddress
     channel_identifier: ChannelID

src/monitoring_service/handlers.py

@@ -21,6 +21,7 @@
 )
 from raiden.utils.typing import BlockNumber, TokenNetworkAddress, TransactionHash
 from raiden_contracts.constants import CONTRACT_TOKEN_NETWORK, ChannelState
+from raiden_libs.blockchain import get_pessimistic_udc_balance
 from raiden_libs.contract_info import CONTRACT_MANAGER
 from raiden_libs.events import (
     Event,
@@ -45,6 +46,7 @@ class Context:
     monitoring_service_contract: Contract
     user_deposit_contract: Contract
     min_reward: int
+    required_confirmations: int
 
 
 def channel_opened_event_handler(event: Event, context: Context) -> None:
@@ -406,67 +408,84 @@ def action_monitoring_triggered_event_handler(event: Event, context: Context) ->
     if channel.update_status:
         last_onchain_nonce = channel.update_status.nonce
 
-    user_address = monitor_request.non_closing_signer
-    user_deposit = context.user_deposit_contract.functions.effectiveBalance(user_address).call()
-
     if monitor_request.nonce <= last_onchain_nonce:
         log.info(
             "Another MS submitted the last known channel state", monitor_request=monitor_request
         )
+        return
 
+    latest_block = context.w3.eth.blockNumber
+    last_confirmed_block = latest_block - context.required_confirmations
+    user_address = monitor_request.non_closing_signer
+    user_deposit = get_pessimistic_udc_balance(
+        udc=context.user_deposit_contract,
+        address=user_address,
+        from_block=last_confirmed_block,
+        to_block=latest_block,
+    )
+    context.user_deposit_contract.functions.effectiveBalance(user_address).call()
     if monitor_request.reward_amount < context.min_reward:
         log.info(
             "Monitor request not executed due to insufficient reward amount",
             monitor_request=monitor_request,
             min_reward=context.min_reward,
         )
+        return
+
+    if user_deposit < monitor_request.reward_amount * DEFAULT_PAYMENT_RISK_FAKTOR:
+        log.debug(
+            "User deposit is insufficient -> try monitoring again later",
+            monitor_request=monitor_request,
+            min_reward=context.min_reward,
+        )
+        context.db.upsert_scheduled_event(
+            ScheduledEvent(trigger_block_number=BlockNumber(last_confirmed_block + 1), event=event)
+        )
+        return
 
-    call_monitor = (
+    assert (
         channel.closing_tx_hash is None
-        and monitor_request.nonce > last_onchain_nonce
-        and user_deposit >= monitor_request.reward_amount * DEFAULT_PAYMENT_RISK_FAKTOR
-        and monitor_request.reward_amount >= context.min_reward
-    )
-    if call_monitor:
-        try:
-            # Attackers might be able to construct MRs that make this fail.
-            # Since we execute a gas estimation before doing the `transact`,
-            # the gas estimation will fail before any gas is used.
-            # If we stop doing a gas estimation, a `call` has to be done before
-            # the `transact` to prevent attackers from wasting the MS's gas.
-            tx_hash = TransactionHash(
-                bytes(
-                    context.monitoring_service_contract.functions.monitor(
-                        monitor_request.signer,
-                        monitor_request.non_closing_signer,
-                        monitor_request.balance_hash,
-                        monitor_request.nonce,
-                        monitor_request.additional_hash,
-                        monitor_request.closing_signature,
-                        monitor_request.non_closing_signature,
-                        monitor_request.reward_amount,
-                        monitor_request.token_network_address,
-                        monitor_request.reward_proof_signature,
-                    ).transact({"from": context.ms_state.address})
-                )
+    ), "This MS already monitored this channel. Should be impossible."
+
+    try:
+        # Attackers might be able to construct MRs that make this fail.
+        # Since we execute a gas estimation before doing the `transact`,
+        # the gas estimation will fail before any gas is used.
+        # If we stop doing a gas estimation, a `call` has to be done before
+        # the `transact` to prevent attackers from wasting the MS's gas.
+        tx_hash = TransactionHash(
+            bytes(
+                context.monitoring_service_contract.functions.monitor(
+                    monitor_request.signer,
+                    monitor_request.non_closing_signer,
+                    monitor_request.balance_hash,
+                    monitor_request.nonce,
+                    monitor_request.additional_hash,
+                    monitor_request.closing_signature,
+                    monitor_request.non_closing_signature,
+                    monitor_request.reward_amount,
+                    monitor_request.token_network_address,
+                    monitor_request.reward_proof_signature,
+                ).transact({"from": context.ms_state.address})
             )
+        )
 
-            log.info(
-                "Sent transaction calling `monitor` for channel",
-                token_network_address=channel.token_network_address,
-                channel_identifier=channel.identifier,
-                transaction_hash=encode_hex(tx_hash),
-            )
-            assert tx_hash is not None
+        log.info(
+            "Sent transaction calling `monitor` for channel",
+            token_network_address=channel.token_network_address,
+            channel_identifier=channel.identifier,
+            transaction_hash=encode_hex(tx_hash),
+        )
+        assert tx_hash is not None
 
-            with context.db.conn:
-                # Add tx hash to list of waiting transactions
-                context.db.add_waiting_transaction(tx_hash)
+        with context.db.conn:
+            # Add tx hash to list of waiting transactions
+            context.db.add_waiting_transaction(tx_hash)
 
-                channel.closing_tx_hash = tx_hash
-                context.db.upsert_channel(channel)
-        except Exception as exc:  # pylint: disable=broad-except
-            log.error("Sending tx failed", exc_info=True, err=exc)
+            channel.closing_tx_hash = tx_hash
+            context.db.upsert_channel(channel)
+    except Exception as exc:  # pylint: disable=broad-except
+        log.error("Sending tx failed", exc_info=True, err=exc)
 
 
 def action_claim_reward_triggered_event_handler(event: Event, context: Context) -> None:

src/monitoring_service/service.py

@@ -99,6 +99,7 @@ def __init__(  # pylint: disable=too-many-arguments
             monitoring_service_contract=monitoring_contract,
             user_deposit_contract=user_deposit_contract,
             min_reward=min_reward,
+            required_confirmations=required_confirmations,
         )
 
     def start(

src/pathfinding_service/api.py

@@ -35,6 +35,7 @@
 from raiden.exceptions import InvalidSignature
 from raiden.utils.signer import recover
 from raiden.utils.typing import Address, PaymentAmount, Signature, TokenAmount, TokenNetworkAddress
+from raiden_libs.blockchain import get_pessimistic_udc_balance
 from raiden_libs.marshmallow import ChecksumAddress, HexedBytes
 
 log = structlog.get_logger(__name__)
@@ -230,7 +231,13 @@ def process_payment(  # pylint: disable=too-many-branches
 
     # Check client's deposit in UserDeposit contract
     udc = pathfinding_service.user_deposit_contract
-    udc_balance = udc.functions.effectiveBalance(iou.sender).call()
+    latest_block = pathfinding_service.web3.eth.blockNumber
+    udc_balance = get_pessimistic_udc_balance(
+        udc=udc,
+        address=iou.sender,
+        from_block=latest_block - pathfinding_service.required_confirmations,
+        to_block=latest_block,
+    )
     required_deposit = round(expected_amount * UDC_SECURITY_MARGIN_FACTOR)
     if udc_balance < required_deposit:
         raise exceptions.DepositTooLow(required_deposit=required_deposit)

src/pathfinding_service/service.py

@@ -69,7 +69,7 @@ def __init__(  # pylint: disable=too-many-arguments
         self.user_deposit_contract = contracts[CONTRACT_USER_DEPOSIT]
         self.chain_id = ChainID(int(web3.net.version))
         self.address = private_key_to_address(private_key)
-        self._required_confirmations = required_confirmations
+        self.required_confirmations = required_confirmations
         self._poll_interval = poll_interval
         self._is_running = gevent.event.Event()
 
@@ -128,7 +128,7 @@ def _run(self) -> None:  # pylint: disable=method-hidden
             start_block=self.database.get_latest_known_block(),
         )
         while not self._is_running.is_set():
-            last_confirmed_block = self.web3.eth.blockNumber - self._required_confirmations
+            last_confirmed_block = self.web3.eth.blockNumber - self.required_confirmations
 
             max_query_interval_end_block = (
                 self.database.get_latest_known_block() + MAX_FILTER_INTERVAL

src/raiden_libs/blockchain.py

@@ -5,10 +5,10 @@
 from eth_utils import decode_hex, encode_hex, to_checksum_address
 from eth_utils.abi import event_abi_to_log_topic
 from web3 import Web3
-from web3.contract import get_event_data
+from web3.contract import Contract, get_event_data
 from web3.utils.abi import filter_by_type
 
-from raiden.utils.typing import Address, BlockNumber
+from raiden.utils.typing import Address, BlockNumber, TokenAmount
 from raiden_contracts.constants import (
     CONTRACT_MONITORING_SERVICE,
     CONTRACT_TOKEN_NETWORK,
@@ -238,3 +238,16 @@ def get_monitoring_blockchain_events(
             )
 
     return events
+
+
+def get_pessimistic_udc_balance(
+    udc: Contract, address: Address, from_block: BlockNumber, to_block: BlockNumber
+) -> TokenAmount:
+    """ Get the effective UDC balance using the block with the lowest result.
+
+    Blocks between the latest confirmed block and the latest block are considered.
+    """
+    return min(
+        udc.functions.effectiveBalance(address).call(block_identifier=block)
+        for block in range(from_block, to_block + 1)
+    )

tests/libs/mocks/web3.py

@@ -14,6 +14,7 @@ class Web3Mock(Mock):
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
         self.net.version = 1  # chain_id
+        self.eth.blockNumber = 100
 
     def _get_child_mock(self, **kwargs):
         return Mock(**kwargs)

tests/libs/test_blockchain.py

@@ -1,10 +1,11 @@
+import eth_tester
 import pytest
 from web3 import Web3
 
 from raiden.utils.typing import Address, BlockNumber
 from raiden_contracts.constants import CONTRACT_TOKEN_NETWORK_REGISTRY, EVENT_TOKEN_NETWORK_CREATED
 from raiden_contracts.contract_manager import ContractManager
-from raiden_libs.blockchain import query_blockchain_events
+from raiden_libs.blockchain import get_pessimistic_udc_balance, query_blockchain_events
 
 
 def create_tnr_contract_events_query(
@@ -101,3 +102,35 @@ def test_limit_inclusivity_in_query_blockchain_events(
         to_block=registry_event_block,
     )
     assert len(events) == 1
+
+
+def test_get_pessimistic_udc_balance(user_deposit_contract, web3, deposit_to_udc, get_accounts):
+    address, = get_accounts(1)
+    deposit_to_udc(address, 10)
+    deposit_block = web3.eth.blockNumber
+    web3.testing.mine(5)
+
+    def deposit(from_offset, to_offset):
+        return get_pessimistic_udc_balance(
+            udc=user_deposit_contract,
+            address=address,
+            from_block=deposit_block + from_offset,
+            to_block=deposit_block + to_offset,
+        )
+
+    assert deposit(0, 0) == 10
+    assert deposit(0, 1) == 10
+    assert deposit(0, 5) == 10
+    assert deposit(-1, -1) == 0
+    assert deposit(-1, 0) == 0
+    assert deposit(-1, 1) == 0
+    assert deposit(-5, 5) == 0
+
+    # Hmm, I would expect this to fail, but maybe one block into the future is
+    # allowed and will call into the block that is currently being created!?
+    # with pytest.raises(eth_tester.exceptions.BlockNotFound):
+    #     deposit(0, 6)
+
+    # That's two blocks that do not exist, yet!
+    with pytest.raises(eth_tester.exceptions.BlockNotFound):
+        deposit(0, 7)

tests/monitoring/monitoring_service/test_handlers.py

@@ -1,8 +1,9 @@
 # pylint: disable=redefined-outer-name
-from unittest.mock import Mock
+from unittest.mock import Mock, patch
 
 import pytest
 from eth_utils import to_checksum_address
+from tests.libs.mocks.web3 import Web3Mock
 from tests.monitoring.monitoring_service.factories import (
     DEFAULT_CHANNEL_IDENTIFIER,
     DEFAULT_PARTICIPANT1,
@@ -12,6 +13,7 @@
     create_signed_monitor_request,
 )
 
+from monitoring_service.constants import DEFAULT_PAYMENT_RISK_FAKTOR
 from monitoring_service.database import Database
 from monitoring_service.events import (
     ActionClaimRewardTriggeredEvent,
@@ -96,11 +98,12 @@ def context(ms_database: Database):
     return Context(
         ms_state=ms_database.load_state(),
         db=ms_database,
-        w3=Mock(),
+        w3=Web3Mock(),
         last_known_block=0,
         monitoring_service_contract=Mock(),
         user_deposit_contract=Mock(),
         min_reward=1,
+        required_confirmations=1,
     )
 
 
@@ -482,6 +485,41 @@ def test_action_monitoring_triggered_event_handler_does_not_trigger_monitor_call
     assert channel.closing_tx_hash is None
 
 
+def test_action_monitoring_rescheduling_when_user_lacks_funds(context: Context):
+    reward_amount = TokenAmount(10)
+    context = setup_state_with_closed_channel(context)
+    context.db.upsert_monitor_request(
+        create_signed_monitor_request(nonce=Nonce(6), reward_amount=reward_amount)
+    )
+
+    event = ActionMonitoringTriggeredEvent(
+        token_network_address=DEFAULT_TOKEN_NETWORK_ADDRESS,
+        channel_identifier=DEFAULT_CHANNEL_IDENTIFIER,
+        non_closing_participant=DEFAULT_PARTICIPANT2,
+    )
+    scheduled_events_before = context.db.get_scheduled_events(max_trigger_block=BlockNumber(10000))
+
+    # Try to call monitor when the user has insufficient funds
+    with patch("monitoring_service.handlers.get_pessimistic_udc_balance", Mock(return_value=0)):
+        action_monitoring_triggered_event_handler(event, context)
+    assert not context.monitoring_service_contract.functions.monitor.called
+
+    # Now the event must have been rescheduled
+    # TODO: check that the event is rescheduled to trigger at the right block
+    scheduled_events_after = context.db.get_scheduled_events(max_trigger_block=BlockNumber(10000))
+    new_events = set(scheduled_events_after) - set(scheduled_events_before)
+    assert len(new_events) == 1
+    assert new_events.pop().event == event
+
+    # With sufficient funds it must succeed
+    with patch(
+        "monitoring_service.handlers.get_pessimistic_udc_balance",
+        Mock(return_value=reward_amount * DEFAULT_PAYMENT_RISK_FAKTOR),
+    ):
+        action_monitoring_triggered_event_handler(event, context)
+    assert context.monitoring_service_contract.functions.monitor.called
+
+
 def test_action_monitoring_triggered_event_handler_with_sufficient_balance_does_trigger_monitor_call(  # noqa
     context: Context,
 ):

tests/pathfinding/fixtures/network_service.py

@@ -323,6 +323,7 @@ def pathfinding_service_web3_mock(
             },
             private_key="3a1076bf45ab87712ad64ccb3b10217737f7faacbf2872e88fdd9a537d8fe266",
             db_filename=":memory:",
+            required_confirmations=1,
         )
 
         yield pathfinding_service