Feature: Parallel message processing #796

[1yam]

The goal of this PR is to process messages in parallel based on address

Related Clickup or Jira tickets : ALEPH-XXX

Self proofreading checklist

• Is my code clear enough and well documented
• Are my files well typed
• New translations have been added or updated if new strings have been introduced in the frontend
• Database migrations file are included
• Are there enough tests
• Documentation has been included (for new feature)

Changes

This pull request introduces significant changes to migrate the database session handling from synchronous to asynchronous operations across multiple modules. Additionally, a new column and indexes are added to the pending_messages table to optimize querying. Below are the most important changes grouped by theme:

Database Migration

• Added a new content_address column to the pending_messages table, which is computed from the content JSON field, and created two new indexes (ix_pending_messages_content_address and ix_pending_messages_content_address_attempt) to improve query performance.

Transition to Asynchronous Database Sessions

• Updated DbSessionFactory and DbSession to their asynchronous counterparts (AsyncDbSessionFactory and AsyncDbSession) across all modules, including src/aleph/api_entrypoint.py, src/aleph/chains/bsc.py, src/aleph/chains/ethereum.py, src/aleph/chains/nuls2.py, src/aleph/chains/connector.py, and src/aleph/chains/indexer_reader.py. [1] [2] [3] [4] [5] [6]

Refactoring to Support Asynchronous Operations

• Replaced synchronous session management (with self.session_factory() as session) with asynchronous context managers (async with self.session_factory() as session) and updated database operations to use await for methods like commit, upsert, and query execution. [1] [2] [3] [4]

Updates to Chain Connectors and Services

• Modified constructors and methods in chain connectors (e.g., BscConnector, EthereumConnector, Nuls2Connector) and services (e.g., ChainDataService, AlephIndexerReader) to use AsyncDbSessionFactory and updated their methods to handle asynchronous database interactions. [1] [2] [3] [4]

Asynchronous Pending Transaction Handling

• Refactored methods for handling pending transactions (e.g., add_pending_tx, add_and_publish_pending_tx) to use asynchronous database operations and ensure consistency between the database and message queue. [1] [2]

These changes collectively improve the scalability and responsiveness of the system by enabling non-blocking database operations and optimizing query performance.

How to test

Explain how to test your PR.
Full node syncing to ensure no issue

Print screen / video

Notes

If a node have lot's of pending messages the migrations might take some times since it's need to create new index and and computed the value of the address (based on the content address) to simplify access during the processing