fix: Reorganize Unit Tests Following Clean Architecture

.github/workflows/backend_integration_tests_pr.yml

@@ -50,7 +50,11 @@ jobs:
           OPENAIKEY: ${{ secrets.OPENAIKEY }}
         run: |
           sed -i "s/<add_your_openai_api_key_here>/${OPENAIKEY}/g" .env
-          sed -i "s/FINALITY_WINDOW =.*/FINALITY_WINDOW = 10/" .env
+          sed -i "s/VITE_FINALITY_WINDOW=\".*\"/VITE_FINALITY_WINDOW=\"10\"/" .env
+          sed -i "s/COMPOSE_CPU_LIMIT=\".*\"/COMPOSE_CPU_LIMIT=\"4\"/" .env
+          sed -i "s/COMPOSE_CPU_RESERVATION=\".*\"/COMPOSE_CPU_RESERVATION=\"1\"/" .env
+          sed -i "s/COMPOSE_MEM_LIMIT=\".*\"/COMPOSE_MEM_LIMIT=\"6gb\"/" .env
+          sed -i "s/COMPOSE_MEM_RESERVATION=\".*\"/COMPOSE_MEM_RESERVATION=\"2gb\"/" .env
           echo >> .env
           if [[ "${{ needs.triggers.outputs.is_pull_request_review_approved }}" == "true" ]]; then
             echo "TEST_WITH_MOCK_LLMS=false" >> .env
@@ -67,18 +71,18 @@ jobs:
         with:
           files: |
             ./docker-compose.yml
-            ./docker-compose.production.yml
           targets: |
             database-migration
             jsonrpc
+            consensus-worker
           set: |
             *.cache-from=type=gha
             *.cache-to=type=gha,mode=max
           load: true
 
       # Start services without rebuilding images
       - name: Run Docker Compose
-        run: docker compose up -d --no-build
+        run: docker compose up -d --no-build jsonrpc consensus-worker
 
 
 

backend/consensus/worker_service.py

@@ -101,6 +101,15 @@ def get_session():
     validators_manager = validators.Manager(SessionLocal())
     await validators_manager.restart()
 
+    # Subscribe to validator change events
+    async def handle_validator_change(event_data):
+        """Reload validators when notified of changes."""
+        logger.info(f"Received validator change event: {event_data}")
+        await validators_manager.restart()
+
+    # Subscribe to validator events channel
+    await msg_handler.subscribe_to_validator_events(handle_validator_change)
+
     # Create and start the worker
     worker = ConsensusWorker(
         get_session=get_session,

backend/database_handler/llm_providers.py

@@ -2,6 +2,7 @@
 from backend.node.create_nodes.providers import get_default_providers
 from .models import LLMProviderDBModel
 from sqlalchemy.orm import Session
+from sqlalchemy.exc import IntegrityError
 
 
 class LLMProviderRegistry:
@@ -18,7 +19,11 @@ def reset_defaults(self):
         for provider in providers:
             self.session.add(_to_db_model(provider, is_default=True))
 
-        self.session.commit()
+        try:
+            self.session.commit()
+        except IntegrityError:
+            # Another worker already seeded the providers, rollback and continue
+            self.session.rollback()
 
     def update_defaults(self):
         """Update default providers while preserving custom ones."""
@@ -62,7 +67,11 @@ def update_defaults(self):
                     }
                 )
 
-        self.session.commit()
+        try:
+            self.session.commit()
+        except IntegrityError:
+            # Another worker already seeded the providers, rollback and continue
+            self.session.rollback()
 
     def get_all(self) -> list[LLMProvider]:
         return [

backend/database_handler/migration/versions/96840ab9133a_add_blocked_at_and_worker_id_to_.py

@@ -5,28 +5,34 @@
 Create Date: 2025-09-11 13:18:28.225878
 
 """
+
 from typing import Sequence, Union
 
 from alembic import op
 import sqlalchemy as sa
 
 
 # revision identifiers, used by Alembic.
-revision: str = '96840ab9133a'
-down_revision: Union[str, None] = '035ef00c2779'
+revision: str = "96840ab9133a"
+down_revision: Union[str, None] = "035ef00c2779"
 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('blocked_at', sa.DateTime(timezone=True), nullable=True))
-    op.add_column('transactions', sa.Column('worker_id', sa.String(length=255), nullable=True))
+    op.add_column(
+        "transactions",
+        sa.Column("blocked_at", sa.DateTime(timezone=True), nullable=True),
+    )
+    op.add_column(
+        "transactions", sa.Column("worker_id", sa.String(length=255), nullable=True)
+    )
     # ### end Alembic commands ###
 
 
 def downgrade() -> None:
     # ### commands auto generated by Alembic - please adjust! ###
-    op.drop_column('transactions', 'worker_id')
-    op.drop_column('transactions', 'blocked_at')
+    op.drop_column("transactions", "worker_id")
+    op.drop_column("transactions", "blocked_at")
     # ### end Alembic commands ###

backend/database_handler/validators_registry.py

@@ -44,9 +44,13 @@ def _get_validator_or_fail(self, validator_address: str) -> Validators:
         return validator_data
 
     def count_validators(self) -> int:
+        # Expire all objects to ensure we get fresh data from the database
+        self.session.expire_all()
         return self.session.query(Validators).count()
 
     def get_all_validators(self, include_private_key: bool = True) -> List[dict]:
+        # Expire all objects to ensure we get fresh data from the database
+        self.session.expire_all()
         validators_data = self.session.query(Validators).all()
         return [
             to_dict(validator, include_private_key) for validator in validators_data
@@ -55,6 +59,8 @@ def get_all_validators(self, include_private_key: bool = True) -> List[dict]:
     def get_validator(
         self, validator_address: str, include_private_key: bool = True
     ) -> dict:
+        # Expire all objects to ensure we get fresh data from the database
+        self.session.expire_all()
         return to_dict(
             self._get_validator_or_fail(validator_address), include_private_key
         )

backend/protocol_rpc/app_lifespan.py

@@ -126,13 +126,16 @@ def _verify_database_ready(db_manager: DatabaseSessionManager) -> None:
 async def _initialise_validators(
     validators_config_json: Optional[str],
     db_manager: DatabaseSessionManager,
+    validators_manager: validators.Manager,
 ) -> None:
     if not validators_config_json:
         return
 
     init_session = db_manager.open_session()
     try:
-        await initialize_validators(validators_config_json, init_session)
+        await initialize_validators(
+            validators_config_json, init_session, validators_manager
+        )
         init_session.commit()
     finally:
         init_session.close()
@@ -201,7 +204,9 @@ async def rpc_app_lifespan(app, settings: RPCAppSettings) -> AsyncIterator[RPCAp
     # Delete all validators from the database and create new ones based on env.VAlIDATORS_CONFIG_JSON
     if settings.validators_config_json:
         logger.info("[STARTUP] Initializing validators from config")
-        await _initialise_validators(settings.validators_config_json, db_manager)
+        await _initialise_validators(
+            settings.validators_config_json, db_manager, validators_manager
+        )
 
     # Restart web and llm modules, created the validators Snapshot, and registers providers and models to the LLM module
     logger.info("[STARTUP] Restarting validators and creating snapshot")
@@ -311,6 +316,20 @@ def get_session() -> Session:
         )
         await redis_subscriber.connect()
         await redis_subscriber.start()
+
+        # Register handler for validator change events
+        async def handle_validator_change(event_data):
+            """Reload validators when they change."""
+            logger.info(f"RPC worker reloading validators due to change event")
+            await validators_manager.restart()
+
+        redis_subscriber.register_handler("validator_created", handle_validator_change)
+        redis_subscriber.register_handler("validator_updated", handle_validator_change)
+        redis_subscriber.register_handler("validator_deleted", handle_validator_change)
+        redis_subscriber.register_handler(
+            "all_validators_deleted", handle_validator_change
+        )
+
         logger.info(
             f"[STARTUP] Redis subscriber connected at {redis_url} for worker event broadcasting"
         )

backend/protocol_rpc/endpoints.py

@@ -192,6 +192,7 @@ def delete_provider(session: Session, id: int) -> None:
 
 async def create_validator(
     session: Session,
+    validators_manager: validators.Manager,
     stake: int,
     provider: str,
     model: str,
@@ -215,11 +216,10 @@ async def create_validator(
         validate_provider(llm_provider)
 
     accounts_manager = AccountsManager(session)
-    validators_registry = ModifiableValidatorsRegistry(session)
 
     account = accounts_manager.create_new_account()
 
-    return await validators_registry.create_validator(
+    return await validators_manager.registry.create_validator(
         Validator(
             address=account.address,
             private_key=account.key,
@@ -257,7 +257,6 @@ async def create_random_validators(
     limit_models: list[str] = None,
 ) -> list[dict]:
     accounts_manager = AccountsManager(session)
-    validators_registry = ModifiableValidatorsRegistry(session)
     llm_provider_registry = LLMProviderRegistry(session)
 
     limit_providers = limit_providers or []
@@ -276,7 +275,7 @@ async def create_random_validators(
         stake = random.randint(min_stake, max_stake)
         validator_account = accounts_manager.create_new_account()
 
-        validator = await validators_registry.create_validator(
+        validator = await validators_manager.registry.create_validator(
             Validator(
                 address=validator_account.address,
                 private_key=validator_account.key,
@@ -292,6 +291,7 @@ async def create_random_validators(
 @check_forbidden_method_in_hosted_studio
 async def update_validator(
     session: Session,
+    validators_manager: validators.Manager,
     validator_address: str,
     stake: int,
     provider: str,
@@ -321,38 +321,33 @@ async def update_validator(
         )
         validate_provider(llm_provider)
 
-    validators_registry = ModifiableValidatorsRegistry(session)
-
     validator = Validator(
         address=validator_address,
         stake=stake,
         llmprovider=llm_provider,
     )
-    return await validators_registry.update_validator(validator)
+    return await validators_manager.registry.update_validator(validator)
 
 
 @check_forbidden_method_in_hosted_studio
 async def delete_validator(
-    session: Session,
+    validators_manager: validators.Manager,
     validator_address: str,
 ) -> str:
     # Remove validation while adding migration to update the db address
     # if not accounts_manager.is_valid_address(validator_address):
     #     raise InvalidAddressError(validator_address)
 
-    validators_registry = ModifiableValidatorsRegistry(session)
-
-    await validators_registry.delete_validator(validator_address)
+    await validators_manager.registry.delete_validator(validator_address)
     return validator_address
 
 
 @check_forbidden_method_in_hosted_studio
 async def delete_all_validators(
-    session: Session,
+    validators_manager: validators.Manager,
 ) -> list:
-    validators_registry = ModifiableValidatorsRegistry(session)
-    await validators_registry.delete_all_validators()
-    return validators_registry.get_all_validators()
+    await validators_manager.registry.delete_all_validators()
+    return validators_manager.registry.get_all_validators()
 
 
 def get_all_validators(validators_registry: ValidatorsRegistry) -> list:

backend/protocol_rpc/message_handler/redis_worker_handler.py

@@ -25,6 +25,7 @@ class RedisWorkerMessageHandler(MessageHandler):
     CONSENSUS_CHANNEL = "consensus:events"
     TRANSACTION_CHANNEL = "transaction:events"
     GENERAL_CHANNEL = "general:events"
+    VALIDATOR_CHANNEL = "validator:events"
 
     def __init__(
         self,
@@ -70,6 +71,51 @@ async def initialize(self):
                 logger.error(f"Worker {self.worker_id} failed to connect to Redis: {e}")
                 raise
 
+    async def subscribe_to_validator_events(self, callback):
+        """
+        Subscribe to validator change events.
+        Used by consensus-worker to reload validators when they change.
+
+        Args:
+            callback: Async function to call when validator event received
+        """
+        if not self.redis_client:
+            await self.initialize()
+
+        if not self.redis_client:
+            logger.error(
+                "Cannot subscribe to validator events: Redis client not initialized"
+            )
+            return
+
+        redis_client = self.redis_client
+
+        async def listener():
+            """Listen for validator events and call callback."""
+            redis_pubsub = redis_client.pubsub()
+            await redis_pubsub.subscribe(self.VALIDATOR_CHANNEL)
+            logger.info(
+                f"Worker {self.worker_id} subscribed to {self.VALIDATOR_CHANNEL}"
+            )
+
+            try:
+                async for message in redis_pubsub.listen():
+                    if message["type"] == "message":
+                        try:
+                            event_data = json.loads(message["data"])
+                            await callback(event_data)
+                        except Exception as e:
+                            logger.error(f"Error processing validator event: {e}")
+            except asyncio.CancelledError:
+                logger.info(
+                    f"Worker {self.worker_id} validator event listener cancelled"
+                )
+                await redis_pubsub.unsubscribe(self.VALIDATOR_CHANNEL)
+                await redis_pubsub.close()
+
+        # Start listener in background
+        asyncio.create_task(listener())
+
     def _get_channel_for_event(self, log_event: LogEvent) -> str:
         """
         Determine the appropriate Redis channel for an event.