feat: consensus thread improvements

backend/database_handler/chain_snapshot.py

@@ -2,6 +2,7 @@
 
 from typing import List
 from sqlalchemy.orm import Session
+from collections import defaultdict
 
 from backend.database_handler.transactions_processor import (
     TransactionStatus,
@@ -18,14 +19,20 @@ def __init__(self, session: Session):
         self.all_validators = self.validators_registry.get_all_validators()
         self.pending_transactions = self._load_pending_transactions()
         self.num_validators = len(self.all_validators)
-        self.accepted_transactions = self._load_accepted_transactions()
+        self.accepted_undetermined_transactions = (
+            self._load_accepted_undetermined_transactions()
+        )
 
     def _load_pending_transactions(self) -> List[dict]:
         """Load and return the list of pending transactions from the database."""
 
         pending_transactions = (
             self.session.query(Transactions)
-            .filter(Transactions.status == TransactionStatus.PENDING)
+            .filter(
+                Transactions.status == TransactionStatus.PENDING,
+                Transactions.queued != True,
+            )
+            .order_by(Transactions.created_at)
             .all()
         )
         return [
@@ -41,19 +48,30 @@ def get_all_validators(self):
         """Return the list of all validators."""
         return self.all_validators
 
-    def _load_accepted_transactions(self) -> List[dict]:
-        """Load and return the list of accepted transactions from the database."""
+    def _load_accepted_undetermined_transactions(self) -> dict[str, List[dict]]:
+        """Load and return the list of accepted and undetermined transactions from the database,
+        grouped by address."""
 
-        accepted_transactions = (
+        accepted_undetermined_transactions = (
             self.session.query(Transactions)
-            .filter(Transactions.status == TransactionStatus.ACCEPTED)
+            .filter(
+                (Transactions.status == TransactionStatus.ACCEPTED)
+                | (Transactions.status == TransactionStatus.UNDETERMINED),
+                Transactions.queued != True,
+            )
+            .order_by(Transactions.created_at)
             .all()
         )
-        return [
-            TransactionsProcessor._parse_transaction_data(transaction)
-            for transaction in accepted_transactions
-        ]
 
-    def get_accepted_transactions(self):
-        """Return the list of accepted transactions."""
-        return self.accepted_transactions
+        # Group transactions by address
+        transactions_by_address = defaultdict(list)
+        for transaction in accepted_undetermined_transactions:
+            address = transaction.to_address or transaction.from_address
+            transactions_by_address[address].append(
+                TransactionsProcessor._parse_transaction_data(transaction)
+            )
+        return transactions_by_address
+
+    def get_accepted_undetermined_transactions(self):
+        """Return the list of accepted and undetermined transactions."""
+        return self.accepted_undetermined_transactions

backend/database_handler/migration/versions/a4a32d27dde2_appeal_undetermined.py

@@ -0,0 +1,55 @@
+"""appeal_undetermined
+
+Revision ID: a4a32d27dde2
+Revises: 2a4ac5eb9455
+Create Date: 2024-11-25 14:49:46.916279
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = "a4a32d27dde2"
+down_revision: Union[str, None] = "2a4ac5eb9455"
+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("appeal_undetermined", sa.Boolean(), nullable=True)
+    )
+    op.execute(
+        "UPDATE transactions SET appeal_undetermined = FALSE WHERE appeal_undetermined IS NULL"
+    )
+    op.alter_column("transactions", "appeal_undetermined", nullable=False)
+    op.add_column(
+        "transactions",
+        sa.Column("timestamp_awaiting_finalization", sa.BigInteger(), nullable=True),
+    )
+    op.execute(
+        "UPDATE transactions SET timestamp_awaiting_finalization = 0 WHERE timestamp_awaiting_finalization IS NULL"
+    )
+    op.drop_column("transactions", "timestamp_accepted")
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column(
+        "transactions",
+        sa.Column(
+            "timestamp_accepted", sa.BIGINT(), autoincrement=False, nullable=True
+        ),
+    )
+    op.execute(
+        "UPDATE transactions SET timestamp_accepted = 0 WHERE timestamp_accepted IS NULL"
+    )
+    op.drop_column("transactions", "timestamp_awaiting_finalization")
+    op.drop_column("transactions", "appeal_undetermined")
+    # ### end Alembic commands ###

backend/database_handler/migration/versions/ba7e1d24cc0e_queued_transactions.py

@@ -0,0 +1,31 @@
+"""queued_transactions
+
+Revision ID: ba7e1d24cc0e
+Revises: a4a32d27dde2
+Create Date: 2025-01-09 13:48:43.456221
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = "ba7e1d24cc0e"
+down_revision: Union[str, None] = "a4a32d27dde2"
+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("queued", sa.Boolean(), nullable=True))
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column("transactions", "queued")
+    # ### end Alembic commands ###

backend/database_handler/models.py

@@ -96,6 +96,7 @@ class Transactions(Base):
     v: Mapped[Optional[int]] = mapped_column(Integer)
     ghost_contract_address: Mapped[Optional[str]] = mapped_column(String(255))
     appeal_failed: Mapped[Optional[int]] = mapped_column(Integer)
+    queued: Mapped[Optional[bool]] = mapped_column(Boolean)
 
     # Relationship for triggered transactions
     triggered_by_hash: Mapped[Optional[str]] = mapped_column(
@@ -116,7 +117,10 @@ class Transactions(Base):
         init=False,
     )
     appealed: Mapped[bool] = mapped_column(Boolean, default=False)
-    timestamp_accepted: Mapped[Optional[int]] = mapped_column(BigInteger, default=None)
+    appeal_undetermined: Mapped[bool] = mapped_column(Boolean, default=False)
+    timestamp_awaiting_finalization: Mapped[Optional[int]] = mapped_column(
+        BigInteger, default=None
+    )
 
 
 class Validators(Base):

backend/database_handler/transactions_processor.py

@@ -61,8 +61,10 @@ def _parse_transaction_data(transaction_data: Transactions) -> dict:
             ],
             "ghost_contract_address": transaction_data.ghost_contract_address,
             "appealed": transaction_data.appealed,
-            "timestamp_accepted": transaction_data.timestamp_accepted,
+            "timestamp_awaiting_finalization": transaction_data.timestamp_awaiting_finalization,
             "appeal_failed": transaction_data.appeal_failed,
+            "appeal_undetermined": transaction_data.appeal_undetermined,
+            "queued": transaction_data.queued,
         }
 
     @staticmethod
@@ -225,8 +227,10 @@ def insert_transaction(
             ),
             ghost_contract_address=ghost_contract_address,
             appealed=False,
-            timestamp_accepted=None,
+            timestamp_awaiting_finalization=None,
             appeal_failed=0,
+            appeal_undetermined=False,
+            queued=False,
         )
 
         self.session.add(new_transaction)
@@ -347,17 +351,20 @@ def set_transaction_appeal(self, transaction_hash: str, appeal: bool):
             self.session.query(Transactions).filter_by(hash=transaction_hash).one()
         )
         transaction.appealed = appeal
+        self.session.commit()
 
-    def set_transaction_timestamp_accepted(
-        self, transaction_hash: str, timestamp_accepted: int = None
+    def set_transaction_timestamp_awaiting_finalization(
+        self, transaction_hash: str, timestamp_awaiting_finalization: int = None
     ):
         transaction = (
             self.session.query(Transactions).filter_by(hash=transaction_hash).one()
         )
-        if timestamp_accepted:
-            transaction.timestamp_accepted = timestamp_accepted
+        if timestamp_awaiting_finalization:
+            transaction.timestamp_awaiting_finalization = (
+                timestamp_awaiting_finalization
+            )
         else:
-            transaction.timestamp_accepted = int(time.time())
+            transaction.timestamp_awaiting_finalization = int(time.time())
 
     def set_transaction_appeal_failed(self, transaction_hash: str, appeal_failed: int):
         if appeal_failed < 0:
@@ -366,3 +373,18 @@ def set_transaction_appeal_failed(self, transaction_hash: str, appeal_failed: in
             self.session.query(Transactions).filter_by(hash=transaction_hash).one()
         )
         transaction.appeal_failed = appeal_failed
+
+    def set_transaction_appeal_undetermined(
+        self, transaction_hash: str, appeal_undetermined: bool
+    ):
+        transaction = (
+            self.session.query(Transactions).filter_by(hash=transaction_hash).one()
+        )
+        transaction.appeal_undetermined = appeal_undetermined
+
+    def set_transaction_queued(self, transaction_hash: str, queued: bool):
+        transaction = (
+            self.session.query(Transactions).filter_by(hash=transaction_hash).one()
+        )
+        transaction.queued = queued
+        self.session.commit()

backend/domain/types.py

@@ -83,8 +83,10 @@ class Transaction:
     created_at: str | None = None
     ghost_contract_address: str | None = None
     appealed: bool = False
-    timestamp_accepted: int | None = None
+    timestamp_awaiting_finalization: int | None = None
     appeal_failed: int = 0
+    appeal_undetermined: bool = False
+    queued: bool | None = False
 
     def to_dict(self):
         return {
@@ -106,8 +108,10 @@ def to_dict(self):
             "created_at": self.created_at,
             "ghost_contract_address": self.ghost_contract_address,
             "appealed": self.appealed,
-            "timestamp_accepted": self.timestamp_accepted,
+            "timestamp_awaiting_finalization": self.timestamp_awaiting_finalization,
             "appeal_failed": self.appeal_failed,
+            "appeal_undetermined": self.appeal_undetermined,
+            "queued": self.queued,
         }
 
     @classmethod
@@ -131,6 +135,10 @@ def from_dict(cls, input: dict) -> "Transaction":
             created_at=input.get("created_at"),
             ghost_contract_address=input.get("ghost_contract_address"),
             appealed=input.get("appealed"),
-            timestamp_accepted=input.get("timestamp_accepted"),
+            timestamp_awaiting_finalization=input.get(
+                "timestamp_awaiting_finalization"
+            ),
             appeal_failed=input.get("appeal_failed", 0),
+            appeal_undetermined=input.get("appeal_undetermined", False),
+            queued=input.get("queued", False),
         )

backend/protocol_rpc/server.py

@@ -161,10 +161,12 @@ def handle_unsubscribe(topics):
 thread_crawl_snapshot = threading.Thread(target=consensus.run_crawl_snapshot_loop)
 thread_crawl_snapshot.start()
 
-# Thread for the run_consensus method
-thread_consensus = threading.Thread(target=consensus.run_consensus_loop)
-thread_consensus.start()
+# Thread for the process_pending_transactions method
+thread_process_pending_transactions = threading.Thread(
+    target=consensus.run_process_pending_transactions_loop
+)
+thread_process_pending_transactions.start()
 
 # Thread for the appeal_window method
-thread_consensus = threading.Thread(target=consensus.run_appeal_window_loop)
-thread_consensus.start()
+thread_appeal_window = threading.Thread(target=consensus.run_appeal_window_loop)
+thread_appeal_window.start()

frontend/src/components/Simulator/TransactionItem.vue

@@ -18,6 +18,7 @@
 
 const props = defineProps<{
   transaction: TransactionItem;
+  finalityWindow: number;
 }>();
 
 const isDetailsModalOpen = ref(false);
@@ -44,20 +45,17 @@
   return props.transaction.hash?.slice(0, 6);
 });
 
-const isAppealed = ref(false);
+const appealed = ref(props.transaction.data.appealed);
 
 const handleSetTransactionAppeal = () => {
   transactionsStore.setTransactionAppeal(props.transaction.hash);
-
-  isAppealed.value = true;
+  appealed.value = true;
 };
 
 watch(
-  () => props.transaction.status,
-  (newStatus) => {
-    if (newStatus !== 'ACCEPTED') {
-      isAppealed.value = false;
-    }
+  () => props.transaction.data.appealed,
+  (newVal) => {
+    appealed.value = newVal;
   },
 );
 
@@ -155,8 +153,15 @@
       <!-- <TransactionStatusBadge
         as="button"
         @click.stop="handleSetTransactionAppeal"
-        :class="{ '!bg-green-500': isAppealed }"
-        v-if="transaction.status == 'ACCEPTED'"
+        :class="{ '!bg-green-500': appealed }"
+        v-if="
+          transaction.data.leader_only == false &&
+          (transaction.status == 'ACCEPTED' ||
+            transaction.status == 'UNDETERMINED') &&
+          Date.now() / 1000 -
+            transaction.data.timestamp_awaiting_finalization <=
+            finalityWindow
+        "
         v-tooltip="'Appeal transaction'"
       >
         <div class="flex items-center gap-1">
@@ -226,6 +231,25 @@
             >
               {{ transaction.status }}
             </TransactionStatusBadge>
+            <!-- <TransactionStatusBadge
+              as="button"
+              @click.stop="handleSetTransactionAppeal"
+              :class="{ '!bg-green-500': appealed }"
+              v-if="
+                transaction.data.leader_only == false &&
+                (transaction.status == 'ACCEPTED' ||
+                  transaction.status == 'UNDETERMINED') &&
+                Date.now() / 1000 -
+                  transaction.data.timestamp_awaiting_finalization <=
+                  finalityWindow
+              "
+              v-tooltip="'Appeal transaction'"
+            >
+              <div class="flex items-center gap-1">
+                APPEAL
+                <GavelIcon class="h-3 w-3" />
+              </div>
+            </TransactionStatusBadge> -->
           </p>
         </div>