Check message status before checking the payment

src/aleph/vm/orchestrator/messages.py

@@ -1,10 +1,12 @@
 import asyncio
 import copy
 
-from aiohttp import ClientConnectorError, ClientResponseError
+from aiohttp import ClientConnectorError, ClientResponseError, ClientSession
 from aiohttp.web_exceptions import HTTPNotFound, HTTPServiceUnavailable
 from aleph_message.models import ExecutableMessage, ItemHash, MessageType
+from aleph_message.status import MessageStatus
 
+from aleph.vm.conf import settings
 from aleph.vm.storage import get_latest_amend, get_message
 
 
@@ -69,3 +71,19 @@
  message = copy.deepcopy(original_message)
  await update_message(message)
  return message, original_message
+
+
+async def get_message_status(item_hash: ItemHash) -> MessageStatus:
+ """
+ Fetch the status of an execution from the reference API server.
+ We use a normal API call to the CCN instead to use the connector because we want to get the updated status of the
+ message and bypass the messages cache.
+ """
+ async with ClientSession() as session:
+ url = f"{settings.API_SERVER}/api/v0/messages/{item_hash}"
+ resp = await session.get(url)
+ # Raise an error if the request failed
+ resp.raise_for_status()
+
+ resp_data = await resp.json()
+ return resp_data["status"]

src/aleph/vm/orchestrator/tasks.py

@@ -16,13 +16,14 @@
  ProgramMessage,
  parse_message,
 )
+from aleph_message.status import MessageStatus
 from yarl import URL
 
 from aleph.vm.conf import settings
 from aleph.vm.pool import VmPool
 from aleph.vm.utils import create_task_log_exceptions
 
-from .messages import load_updated_message
+from .messages import get_message_status, load_updated_message
 from .payment import (
  compute_required_balance,
  compute_required_flow,
@@ -148,6 +149,14 @@
  while True:
  await asyncio.sleep(settings.PAYMENT_MONITOR_INTERVAL)
 
+ # Check if the executions continues existing or are forgotten before checking the payment
+ for vm_hash in pool.executions.keys():
+ message_status = await get_message_status(vm_hash)
+ if message_status != MessageStatus.PROCESSED:
+ logger.debug(f"Stopping {vm_hash} execution due to {message_status} message status")
+ await pool.stop_vm(vm_hash)
+ pool.forget_vm(vm_hash)
+
  # Check if the balance held in the wallet is sufficient holder tier resources (Not do it yet)
  for sender, chains in pool.get_executions_by_sender(payment_type=PaymentType.hold).items():
  for chain, executions in chains.items():
@@ -171,6 +180,7 @@
  logger.debug(
  f"Get stream flow from Sender {sender} to Receiver {settings.PAYMENT_RECEIVER_ADDRESS} of {stream}"
  )
+
  required_stream = await compute_required_flow(executions)
  logger.debug(f"Required stream for Sender {sender} executions: {required_stream}")
  # Stop executions until the required stream is reached

tests/supervisor/test_status.py

@@ -6,21 +6,6 @@
 from aleph.vm.orchestrator.status import check_internet
 
 
-@pytest.mark.asyncio
-async def test_check_internet_no_server_header():
- vm_id = ItemHash("cafecafecafecafecafecafecafecafecafecafecafecafecafecafecafecafe")
-
- mock_session = Mock()
- mock_session.get = MagicMock()
- mock_session.get.__aenter__.return_value.json = AsyncMock(return_value={"result": 200})
-
- # A "Server" header is always expected in the result from the VM.
- # If it is not present, the diagnostic VM is not working correctly
- # and an error must be raised.
- with pytest.raises(ValueError):
- await check_internet(mock_session, vm_id)
-
-
 @pytest.mark.asyncio
 async def test_check_internet_wrong_result_code():
  vm_id = ItemHash("cafecafecafecafecafecafecafecafecafecafecafecafecafecafecafecafe")