Just reraise exception when failed to send

proxy/common_neon/utils/pickable_data_server.py

@@ -42,7 +42,6 @@ async def handle_client(self, reader: StreamReader, writer: StreamWriter):
                 writer.write(result_data)
                 await writer.drain()
             except ConnectionResetError:
-                self.error(f"Client connection has been closed")
                 break
             except asyncio.exceptions.IncompleteReadError as err:
                 self.error(f"Incomplete read error: {err}")

proxy/mempool/executor_mng.py

@@ -64,6 +64,7 @@ def _get_executor(self) -> Tuple[int, MpExecutorClient]:
         return executor_id, executor_info.client
 
     def on_no_liquidity(self, resource_id: int):
+        self.debug(f"No liquidity, executor: {resource_id} - will be unblocked in: {ExecutorMng.BRING_BACK_EXECUTOR_TIMEOUT_SEC} sec")
         asyncio.get_event_loop().create_task(self._release_executor_later(resource_id))
 
     async def _release_executor_later(self, executor_id: int):

proxy/mempool/mem_pool.py

@@ -25,14 +25,15 @@ def __init__(self, executor: IMemPoolExecutor):
 
     async def enqueue_mp_request(self, mp_request: MemPoolRequest):
         tx_hash = mp_request.neon_tx.hash_signed().hex()
+        log_ctx = {"context": {"req_id": mp_request.req_id}}
         try:
-            self.debug(f"Got mp_tx_request: 0x{tx_hash} to be scheduled on the mempool")
+            self.debug(f"Got mp_tx_request: 0x{tx_hash} to be scheduled on the mempool", extra=log_ctx)
             if len(self._tx_req_queue) > MemPool.TX_QUEUE_MAX_SIZE:
                 self._tx_req_queue = self._tx_req_queue[-MemPool.TX_QUEUE_SIZE:]
             bisect.insort_left(self._tx_req_queue, mp_request)
             await self._kick_tx_queue()
         except Exception as err:
-            self.error(f"Failed enqueue tx: {tx_hash} into queue: {err}")
+            self.error(f"Failed enqueue tx: {tx_hash} into queue: {err}", extra=log_ctx)
 
     async def process_tx_queue(self):
         while True:
@@ -76,15 +77,17 @@ async def check_processing_tasks(self):
 
     async def _process_mp_result(self, resource_id: int, mp_result: MemPoolResult, mp_request: MemPoolRequest):
         hash = "0x" + mp_request.neon_tx.hash_signed().hex()
+        log_ctx = {"context": {"req_id": mp_request.req_id}}
         if mp_result.code == MemPoolResultCode.Done:
-            self.debug(f"Neon tx: {hash} - processed on executor: {resource_id} - done")
+            self.debug(f"Neon tx: {hash} - processed on executor: {resource_id} - done", extra=log_ctx)
             self._on_request_done(mp_request)
             self._executor.release_resource(resource_id)
             await self._kick_tx_queue()
             return
-        self.warning(f"Failed to process tx: {hash} - on executor: {resource_id}, status: {mp_result} - reschedule")
-        if mp_result.code == MemPoolResultCode.ToBeRepeat:
+        self.warning(f"Failed to process tx: {hash} - on executor: {resource_id}, status: {mp_result} - reschedule", extra=log_ctx)
+        if mp_result.code == MemPoolResultCode.BlockedAccount:
             self._executor.release_resource(resource_id)
+            await self.enqueue_mp_request(mp_request)
             await self._kick_tx_queue()
         elif mp_result.code == MemPoolResultCode.NoLiquidity:
             self._executor.on_no_liquidity(resource_id)

proxy/mempool/mempool_api.py

@@ -31,6 +31,7 @@ def release_resource(self, resource_id: int):
 
 @dataclass(order=True)
 class MemPoolRequest:
+    req_id: int
     signature: str
     neon_tx: NeonTx = field(compare=False)
     neon_tx_exec_cfg: NeonTxExecCfg = field(compare=False)
@@ -44,8 +45,10 @@ def __post_init__(self):
 
 class MemPoolResultCode(IntEnum):
     Done = 0
-    ToBeRepeat = 1,
-    NoLiquidity = 2,
+    BlockedAccount = 1,
+    SolanaUnavailable = 2,
+    NoLiquidity = 3,
+    Unspecified = 4,
     Dummy = -1
 
 

proxy/mempool/mempool_executor.py

@@ -3,7 +3,7 @@
 import socket
 import time
 
-from logged_groups import logged_group
+from logged_groups import logged_group, logging_context
 
 from ..common_neon.solana_interactor import SolanaInteractor
 from ..common_neon.config import IConfig
@@ -39,13 +39,14 @@ def _init_in_proc(self):
         self._solana = SolanaInteractor(self._config.get_solana_url())
         self._db = MemDB(self._solana)
 
-    def execute_neon_tx(self, mempool_tx_cfg: MemPoolRequest):
-        try:
-            self.execute_neon_tx_impl(mempool_tx_cfg)
-        except Exception as err:
-            self.error(f"Failed to execute neon_tx: {err}")
-            return MemPoolResult(MemPoolResultCode.ToBeRepeat, None)
-        return MemPoolResult(MemPoolResultCode.Done, None)
+    def execute_neon_tx(self, mempool_request: MemPoolRequest):
+        with logging_context(req_id=mempool_request.req_id, exectr=self._id):
+            try:
+                self.execute_neon_tx_impl(mempool_request)
+            except Exception as err:
+                self.error(f"Failed to execute neon_tx: {err}")
+                return MemPoolResult(MemPoolResultCode.BlockedAccount, None)
+            return MemPoolResult(MemPoolResultCode.Done, None)
 
     def execute_neon_tx_impl(self, mempool_tx_cfg: MemPoolRequest):
         neon_tx = mempool_tx_cfg.neon_tx

proxy/mempool/neon_tx_stages.py

@@ -29,7 +29,7 @@ def build(self):
         pass
 
 
-@logged_group("neon.Proxy")
+@logged_group("neon.MemPool")
 class NeonCancelTxStage(NeonTxStage, abc.ABC):
     NAME = 'cancelWithNonce'
 
@@ -78,7 +78,7 @@ def _create_account_with_seed(self):
         return self.s.builder.create_account_with_seed_instruction(self.sol_account, self._seed, self.balance, self.size)
 
 
-@logged_group("neon.Proxy")
+@logged_group("neon.MemPool")
 class NeonCreateAccountTxStage(NeonTxStage):
     NAME = 'createNeonAccount'
 
@@ -98,7 +98,7 @@ def build(self):
         self.tx.add(self._create_account())
 
 
-@logged_group("neon.Proxy")
+@logged_group("neon.MemPool")
 class NeonCreateERC20TxStage(NeonTxStage, abc.ABC):
     NAME = 'createERC20Account'
 
@@ -124,7 +124,7 @@ def build(self):
         self.tx.add(self._create_erc20_account())
 
 
-@logged_group("neon.Proxy")
+@logged_group("neon.MemPool")
 class NeonCreateContractTxStage(NeonCreateAccountWithSeedStage, abc.ABC):
     NAME = 'createNeonContract'
 
@@ -150,7 +150,7 @@ def build(self):
         self.tx.add(self._create_account())
 
 
-@logged_group("neon.Proxy")
+@logged_group("neon.MemPool")
 class NeonResizeContractTxStage(NeonCreateAccountWithSeedStage, abc.ABC):
     NAME = 'resizeNeonContract'
 

proxy/mempool/operator_resource_list.py

@@ -42,7 +42,7 @@ def secret_key(self) -> bytes:
         return self.signer.secret_key()
 
 
-@logged_group("neon.Proxy")
+@logged_group("neon.MemPool")
 class OperatorResourceList:
     # These variables are global for class, they will be initialized one time
     _manager = mp.Manager()
@@ -282,7 +282,7 @@ def free_resource_info(self):
         self._free_resource_list.append(resource.idx)
 
 
-@logged_group("neon.Proxy")
+@logged_group("neon.MemPool")
 class NeonCreatePermAccount(NeonCreateAccountWithSeedStage, abc.ABC):
     NAME = 'createPermAccount'
 

proxy/mempool/transaction_sender.py

@@ -30,7 +30,7 @@
 from .operator_resource_list import OperatorResourceInfo
 
 
-@logged_group("neon.Proxy")
+@logged_group("neon.MemPool")
 class NeonTxSender:
     def __init__(self, db: MemDB, solana: SolanaInteractor, eth_tx: EthTx, steps: int):
         self._db = db
@@ -205,7 +205,7 @@ def done_account_tx_list(self, skip_create_accounts=False):
             self.create_account_tx.instructions.clear()
 
 
-@logged_group("neon.Proxy")
+@logged_group("neon.MemPool")
 class BaseNeonTxStrategy(metaclass=abc.ABCMeta):
     NAME = 'UNKNOWN STRATEGY'
 
@@ -259,7 +259,7 @@ def _validate_gas_limit(self):
         return False
 
 
-@logged_group("neon.Proxy")
+@logged_group("neon.MemPool")
 class SimpleNeonTxSender(SolTxListSender):
     def __init__(self, strategy: BaseNeonTxStrategy, *args, **kwargs):
         SolTxListSender.__init__(self, *args, **kwargs)
@@ -282,7 +282,7 @@ def _on_post_send(self):
                 raise RuntimeError('Run out of attempts to execute transaction')
 
 
-@logged_group("neon.Proxy")
+@logged_group("neon.MemPool")
 class SimpleNeonTxStrategy(BaseNeonTxStrategy, abc.ABC):
     NAME = 'CallFromRawEthereumTX'
     IS_SIMPLE = True
@@ -329,7 +329,7 @@ def execute(self) -> (NeonTxResultInfo, [str]):
         return tx_sender.neon_res, tx_sender.success_sign_list
 
 
-@logged_group("neon.Proxy")
+@logged_group("neon.MemPool")
 class IterativeNeonTxSender(SimpleNeonTxSender):
     def __init__(self, *args, **kwargs):
         SimpleNeonTxSender.__init__(self, *args, **kwargs)
@@ -443,7 +443,7 @@ def _on_post_send(self):
             self._tx_list.append(self._strategy.build_tx())
 
 
-@logged_group("neon.Proxy")
+@logged_group("neon.MemPool")
 class IterativeNeonTxStrategy(BaseNeonTxStrategy, abc.ABC):
     NAME = 'PartialCallOrContinueFromRawEthereumTX'
     IS_SIMPLE = False
@@ -492,7 +492,7 @@ def execute(self) -> (NeonTxResultInfo, [str]):
         return tx_sender.neon_res, tx_sender.success_sign_list
 
 
-@logged_group("neon.Proxy")
+@logged_group("neon.MemPool")
 class HolderNeonTxStrategy(IterativeNeonTxStrategy, abc.ABC):
     NAME = 'ExecuteTrxFromAccountDataIterativeOrContinue'
 
@@ -531,7 +531,7 @@ def _build_preparation_tx_list(self) -> [TransactionWithComputeBudget]:
         return tx_list
 
 
-@logged_group("neon.Proxy")
+@logged_group("neon.MemPool")
 class NoChainIdNeonTxStrategy(HolderNeonTxStrategy, abc.ABC):
     NAME = 'ExecuteTrxFromAccountDataIterativeOrContinueNoChainId'
 

proxy/neon_rpc_api_model/neon_rpc_api_model.py

@@ -1,11 +1,10 @@
 import json
 import multiprocessing
-import socket
 import traceback
 from typing import Optional, Union, Tuple
 
 import sha3
-from logged_groups import logged_group
+from logged_groups import logged_group, LogMng
 from web3.auto import w3
 
 from ..common_neon.address import EthereumAddress
@@ -450,13 +449,13 @@ def eth_sendRawTransaction(self, rawTrx: str) -> str:
             neon_tx_cfg, emulating_result = self.precheck(trx)
 
             self._stat_tx_success()
-
-            mempool_tx_request = MemPoolRequest(signature=eth_signature,
+            req_id = LogMng.get_logging_context().get("req_id")
+            mempool_tx_request = MemPoolRequest(req_id=req_id,
+                                                signature=eth_signature,
                                                 neon_tx=trx,
                                                 neon_tx_exec_cfg=neon_tx_cfg,
                                                 emulating_result=emulating_result)
             self._mempool_client.send_raw_transaction(mempool_tx_request)
-
             return eth_signature
 
         except PendingTxError as err:

requirements.txt

@@ -10,4 +10,4 @@ ethereum
 py-solc-x==1.1.0
 flask
 prometheus_client==0.13.1
-git+https://github.com/neonlabsorg/[email protected].4
+git+https://github.com/neonlabsorg/[email protected].5