feat: improve logging system in terminal and UI

backend/consensus/base.py

@@ -29,6 +29,11 @@
 from backend.node.base import Node
 from backend.node.genvm.types import ExecutionMode, Receipt, Vote
 from backend.protocol_rpc.message_handler.base import MessageHandler
+from backend.protocol_rpc.message_handler.types import (
+    LogEvent,
+    EventType,
+    EventScope,
+)
 
 
 def node_factory(
@@ -146,6 +151,9 @@ async def exec_transaction(
             Node,
         ] = node_factory,
     ):
+        msg_handler = self.msg_handler.with_client_session(
+            transaction.client_session_id
+        )
         if (
             transactions_processor.get_transaction_by_hash(transaction.hash)["status"]
             != TransactionStatus.PENDING.value
@@ -161,8 +169,8 @@ async def exec_transaction(
         # If transaction is a transfer, execute it
         # TODO: consider when the transfer involves a contract account, bridging, etc.
         if transaction.type == TransactionType.SEND:
-            return self.execute_transfer(
-                transaction, transactions_processor, accounts_manager
+            return ConsensusAlgorithm.execute_transfer(
+                transaction, transactions_processor, accounts_manager, msg_handler
             )
 
         # Select Leader and validators
@@ -180,8 +188,11 @@ async def exec_transaction(
 
         for validators in rotate(involved_validators):
             # Update transaction status
-            transactions_processor.update_transaction_status(
-                transaction.hash, TransactionStatus.PROPOSING
+            ConsensusAlgorithm.dispatch_transaction_status_update(
+                transactions_processor,
+                transaction.hash,
+                TransactionStatus.PROPOSING,
+                msg_handler,
             )
 
             [leader, *remaining_validators] = validators
@@ -199,16 +210,19 @@ async def exec_transaction(
                 ExecutionMode.LEADER,
                 contract_snapshot,
                 None,
-                self.msg_handler,
+                msg_handler,
                 contract_snapshot_factory,
             )
 
             # Leader executes transaction
             leader_receipt = await leader_node.exec_transaction(transaction)
             votes = {leader["address"]: leader_receipt.vote.value}
             # Update transaction status
-            transactions_processor.update_transaction_status(
-                transaction.hash, TransactionStatus.COMMITTING
+            ConsensusAlgorithm.dispatch_transaction_status_update(
+                transactions_processor,
+                transaction.hash,
+                TransactionStatus.COMMITTING,
+                msg_handler,
             )
 
             # Create Validators
@@ -218,7 +232,7 @@ async def exec_transaction(
                     ExecutionMode.VALIDATOR,
                     contract_snapshot,
                     leader_receipt,
-                    self.msg_handler,
+                    msg_handler,
                     contract_snapshot_factory,
                 )
                 for validator in remaining_validators
@@ -235,9 +249,11 @@ async def exec_transaction(
 
             for i, validation_result in enumerate(validation_results):
                 votes[validator_nodes[i].address] = validation_result.vote.value
-            transactions_processor.update_transaction_status(
+            ConsensusAlgorithm.dispatch_transaction_status_update(
+                transactions_processor,
                 transaction.hash,
                 TransactionStatus.REVEALING,
+                msg_handler,
             )
 
             if (
@@ -252,13 +268,27 @@ async def exec_transaction(
 
         else:  # this block is executed if the loop above is not broken
             print("Consensus not reached for transaction: ", transaction)
-            transactions_processor.update_transaction_status(
-                transaction.hash, TransactionStatus.UNDETERMINED
+            msg_handler.send_message(
+                LogEvent(
+                    "consensus_failed",
+                    EventType.ERROR,
+                    EventScope.CONSENSUS,
+                    "Failed to reach consensus",
+                )
+            )
+            ConsensusAlgorithm.dispatch_transaction_status_update(
+                transactions_processor,
+                transaction.hash,
+                TransactionStatus.UNDETERMINED,
+                msg_handler,
             )
             return
 
-        transactions_processor.update_transaction_status(
-            transaction.hash, TransactionStatus.ACCEPTED
+        ConsensusAlgorithm.dispatch_transaction_status_update(
+            transactions_processor,
+            transaction.hash,
+            TransactionStatus.ACCEPTED,
+            msg_handler,
         )
 
         final = False
@@ -269,6 +299,16 @@ async def exec_transaction(
             validators=validation_results,
         ).to_dict()
 
+        msg_handler.send_message(
+            LogEvent(
+                "consensus_reached",
+                EventType.SUCCESS,
+                EventScope.CONSENSUS,
+                "Reached consensus",
+                consensus_data,
+            )
+        )
+
         # Register contract if it is a new contract
         if transaction.type == TransactionType.DEPLOY_CONTRACT:
             new_contract = {
@@ -280,10 +320,27 @@ async def exec_transaction(
             }
             contract_snapshot.register_contract(new_contract)
 
+            msg_handler.send_message(
+                LogEvent(
+                    "deployed_contract",
+                    EventType.SUCCESS,
+                    EventScope.GENVM,
+                    "Contract deployed",
+                    new_contract,
+                )
+            )
+
         # Update contract state if it is an existing contract
         else:
             contract_snapshot.update_contract_state(leader_receipt.contract_state)
 
+        ConsensusAlgorithm.dispatch_transaction_status_update(
+            transactions_processor,
+            transaction.hash,
+            TransactionStatus.FINALIZED,
+            msg_handler,
+        )
+
         # Finalize transaction
         transactions_processor.set_transaction_result(
             transaction.hash,
@@ -303,13 +360,15 @@ async def exec_transaction(
                 value=0,  # No value gets transferred?
                 type=TransactionType.RUN_CONTRACT.value,
                 leader_only=transaction.leader_only,  # Cascade
+                client_session_id=transaction.client_session_id,
             )
 
+    @staticmethod
     def execute_transfer(
-        self,
         transaction: Transaction,
         transactions_processor: TransactionsProcessor,
         accounts_manager: AccountsManager,
+        msg_handler: MessageHandler,
     ):
         """
         Executes a native token transfer between Externally Owned Accounts (EOAs).
@@ -333,8 +392,10 @@ def execute_transfer(
 
             # If the sender does not have enough balance, set the transaction status to UNDETERMINED
             if from_balance < transaction.value:
-                transactions_processor.update_transaction_status(
-                    transaction.hash, TransactionStatus.UNDETERMINED
+                MessageHandler.dispatch_transaction_status_update(
+                    transactions_processor,
+                    transaction.hash,
+                    TransactionStatus.UNDETERMINED,
                 )
                 return
 
@@ -352,8 +413,33 @@ def execute_transfer(
                 transaction.to_address, to_balance + transaction.value
             )
 
-        transactions_processor.update_transaction_status(
-            transaction.hash, TransactionStatus.FINALIZED
+        ConsensusAlgorithm.dispatch_transaction_status_update(
+            transactions_processor,
+            transaction.hash,
+            TransactionStatus.FINALIZED,
+            msg_handler,
+        )
+
+    @staticmethod
+    def dispatch_transaction_status_update(
+        transactions_processor: TransactionsProcessor,
+        transaction_hash: str,
+        new_status: TransactionStatus,
+        msg_handler: MessageHandler,
+    ):
+        transactions_processor.update_transaction_status(transaction_hash, new_status)
+
+        msg_handler.send_message(
+            LogEvent(
+                "transaction_status_updated",
+                EventType.INFO,
+                EventScope.CONSENSUS,
+                f"{str(new_status.value)} {str(transaction_hash)}",
+                {
+                    "hash": str(transaction_hash),
+                    "new_status": str(new_status.value),
+                },
+            )
         )
 
 

backend/database_handler/migration/versions/a32f85df2806_add_client_session_id_to_transactions.py

@@ -0,0 +1,34 @@
+"""add client session id to transactions
+
+Revision ID: a32f85df2806
+Revises: 3566595124f6
+Create Date: 2024-10-01 12:09:19.995482
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = "a32f85df2806"
+down_revision: Union[str, None] = "3566595124f6"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column(
+        "transactions",
+        sa.Column("client_session_id", sa.String(length=255), nullable=True),
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column("transactions", "client_session_id")
+    # ### end Alembic commands ###

backend/database_handler/models.py

@@ -82,6 +82,9 @@ class Transactions(Base):
     created_at: Mapped[Optional[datetime.datetime]] = mapped_column(
         DateTime(True), server_default=func.current_timestamp(), init=False
     )
+    client_session_id: Mapped[Optional[str]] = mapped_column(
+        String(255)
+    )  # Used to identify the client session that is subscribed to this transaction's events
     leader_only: Mapped[bool] = mapped_column(Boolean)
     r: Mapped[Optional[int]] = mapped_column(Integer)
     s: Mapped[Optional[int]] = mapped_column(Integer)

backend/database_handler/transactions_processor.py

@@ -10,7 +10,10 @@
 
 
 class TransactionsProcessor:
-    def __init__(self, session: Session):
+    def __init__(
+        self,
+        session: Session,
+    ):
         self.session = session
 
     @staticmethod
@@ -31,6 +34,7 @@ def _parse_transaction_data(transaction_data: Transactions) -> dict:
             "v": transaction_data.v,
             "created_at": transaction_data.created_at.isoformat(),
             "leader_only": transaction_data.leader_only,
+            "client_session_id": transaction_data.client_session_id,
         }
 
     @staticmethod
@@ -75,6 +79,7 @@ def insert_transaction(
         value: float,
         type: int,
         leader_only: bool,
+        client_session_id: str | None,
     ) -> int:
         nonce = (
             self.session.query(Transactions)
@@ -103,6 +108,7 @@ def insert_transaction(
             s=None,
             v=None,
             leader_only=leader_only,
+            client_session_id=client_session_id,
         )
 
         self.session.add(new_transaction)
@@ -149,9 +155,4 @@ def set_transaction_result(self, transaction_hash: str, consensus_data: dict):
         transaction.status = TransactionStatus.FINALIZED
         transaction.consensus_data = consensus_data
 
-        print(
-            "Updating transaction status",
-            transaction_hash,
-            TransactionStatus.FINALIZED.value,
-        )
         self.session.commit()

backend/domain/types.py

@@ -79,6 +79,7 @@ class Transaction:
     leader_only: bool = (
         False  # Flag to indicate if this transaction should be processed only by the leader. Used for fast and cheap execution of transactions.
     )
+    client_session_id: str | None = None
 
     def to_dict(self):
         return {
@@ -97,6 +98,7 @@ def to_dict(self):
             "s": self.s,
             "v": self.v,
             "leader_only": self.leader_only,
+            "client_session_id": self.client_session_id,
         }
 
 
@@ -117,4 +119,5 @@ def transaction_from_dict(input: dict) -> Transaction:
         s=input.get("s"),
         v=input.get("v"),
         leader_only=input.get("leader_only", False),
+        client_session_id=input["client_session_id"],
     )

backend/node/base.py

@@ -89,27 +89,14 @@ async def run_contract(
     def get_contract_data(
         self, code: str, state: str, method_name: str, method_args: list
     ):
-        output_buffer = io.StringIO()
-
-        result = GenVM.get_contract_data(
+        result = self.genvm.get_contract_data(
             code,
             state,
             method_name,
             method_args,
             self.contract_snapshot_factory,
-            output_buffer,
         )
 
-        if self.genvm.contract_runner.mode == ExecutionMode.LEADER:
-            # Retrieve the captured stdout and stderr
-            captured_out = output_buffer.getvalue()
-            if captured_out:
-                socket_message = {
-                    "function": "intelligent_contract_execution",
-                    "response": {"status": "info", "message": captured_out},
-                }
-                self.msg_handler.socket_emit(socket_message)
-
         return result
 
     def get_contract_schema(self, code: str):