chore: sync dynamic config (#798)

Impacts: validator, facilitator

Author: adalc-reef

facilitator/app/src/project/core/utils.py

@@ -1,83 +1,5 @@
-import asyncio
-from concurrent.futures import ThreadPoolExecutor
-
 import wandb
-from asgiref.sync import AsyncToSync, SyncToAsync
-from constance import config
 from django.conf import settings
-from structlog import get_logger
-
-log = get_logger(__name__)
-
-
-def get_thread_sensitive_executor(loop):
-    """Returns executor to run sync code in as if sync code were wrapped with sync_to_async
-
-    This is exactly copied implementation from SyncToAsync but instead of running code in thread
-    and awaiting in the loop we return executor in which SyncToAsync would run sync code
-    and we allow to pass function directly allowing for blocking waiting for future result.
-
-    TODO: handling deadlock is not implemented so it can be unsafe in some conditions
-    """
-    current_thread_executor = getattr(AsyncToSync.executors, "current", None)
-    if current_thread_executor:
-        # If we have a parent sync thread above somewhere, use that
-        executor = current_thread_executor
-    elif SyncToAsync.thread_sensitive_context.get(None):
-        # If we have a way of retrieving the current context, attempt
-        # to use a per-context thread pool executor
-        thread_sensitive_context = SyncToAsync.thread_sensitive_context.get()
-
-        if thread_sensitive_context in SyncToAsync.context_to_thread_executor:
-            # Re-use thread executor in current context
-            executor = SyncToAsync.context_to_thread_executor[thread_sensitive_context]
-        else:
-            # Create new thread executor in current context
-            executor = ThreadPoolExecutor(max_workers=1)
-    elif loop in AsyncToSync.loop_thread_executors:
-        # Re-use thread executor for running loop
-        executor = AsyncToSync.loop_thread_executors[loop]
-    elif SyncToAsync.deadlock_context.get(False):
-        raise RuntimeError("Single thread executor already being used, would deadlock")
-    else:
-        # Otherwise, we run it in a fixed single thread
-        executor = SyncToAsync.single_thread_executor
-        # TODO: we don't set deadlock here - maybe we should clean it all up
-        # SyncToAsync.deadlock_context.set(True)
-    return executor
-
-
-def safe_sync(fun, *args, **kwargs):
-    """Runs sensitive sync function in sync or async context
-
-    When running in sync context it is the same as if it was called directly.
-    In async context function is run as if it would be run called using sync_to_async
-    but it blocks async context when waiting for future result. This allows writing
-    sync sensitive functions that can be run in async context without need to write
-    async version of a function that allows await, eg. istead of writing two versions
-    one using obj.save() and second await obj.asave() we can write one function
-    with safe_sync(obj.save)
-    """
-    try:
-        # Try to get the event loop. If we're in an async context, this will succeed.
-        loop = asyncio.get_running_loop()
-    except RuntimeError:
-        loop = None
-    if loop is not None:
-        executor = get_thread_sensitive_executor(loop)
-        future = executor.submit(fun, *args, **kwargs)
-        return future.result()
-    else:
-        # We're in a synchronous context without an event loop
-        return fun(*args, **kwargs)
-
-
-class SafeConfig:
-    def __getattr__(self, name):
-        return safe_sync(getattr, config, name)
-
-
-safe_config = SafeConfig()
 
 
 def fetch_compute_subnet_hardware() -> dict:

miner/app/src/compute_horde_miner/miner/dynamic_config.py

@@ -4,5 +4,9 @@
 from constance import config
 
 
+def get_config(key: str) -> Any:
+    return getattr(config, key)
+
+
 async def aget_config(key: str) -> Any:
     return await sync_to_async(lambda: getattr(config, key))()

validator/app/src/compute_horde_validator/validator/allowance/utils/blocks.py

@@ -5,10 +5,10 @@
 import turbobt
 from celery.utils.log import get_task_logger
 from compute_horde_core.executor_class import ExecutorClass
-from constance import config
 from django.db import IntegrityError, transaction
 from django.db.models import Case, Exists, Max, Min, OuterRef, Q, Sum, When
 
+from compute_horde_validator.validator.dynamic_config import get_config
 from compute_horde_validator.validator.locks import Lock, LockType
 
 from ...models import SystemEvent
@@ -142,7 +142,7 @@ def process_block_allowance(
             creation_timestamp=supertensor_.get_block_timestamp(block_number),
         )
 
-        dynamic_multiplier = cast(float, config.DYNAMIC_BLOCK_ALLOWANCE_MULTIPLIER)
+        dynamic_multiplier = cast(float, get_config("DYNAMIC_BLOCK_ALLOWANCE_MULTIPLIER"))
         neurons = supertensor_.list_neurons(block_number)
         save_neurons(neurons, block_number)
 

validator/app/src/compute_horde_validator/validator/allowance/utils/spending.py

@@ -5,7 +5,8 @@
 from typing import NamedTuple, cast
 
 from compute_horde_core.executor_class import ExecutorClass
-from constance import config
+
+from compute_horde_validator.validator.dynamic_config import get_config
 
 from ...models import Block, BlockAllowance
 from .. import settings
@@ -44,10 +45,10 @@ class SpendingBookkeeperBase(ABC):
     def __init__(self) -> None:
         self._block_expiry = settings.BLOCK_EXPIRY
         self._validation_leeway_lower = cast(
-            int, config.DYNAMIC_SPENDING_VALIDATION_BLOCK_LEEWAY_LOWER
+            int, get_config("DYNAMIC_SPENDING_VALIDATION_BLOCK_LEEWAY_LOWER")
         )
         self._validation_leeway_upper = cast(
-            int, config.DYNAMIC_SPENDING_VALIDATION_BLOCK_LEEWAY_UPPER
+            int, get_config("DYNAMIC_SPENDING_VALIDATION_BLOCK_LEEWAY_UPPER")
         )
 
     def spend(

validator/app/src/compute_horde_validator/validator/collateral/default.py

@@ -11,7 +11,6 @@
 import turbobt
 from asgiref.sync import async_to_sync
 from compute_horde.smart_contracts.utils import get_web3_connection
-from constance import config
 from django.conf import settings
 from eth_account import Account
 from eth_account.signers.local import LocalAccount
@@ -21,6 +20,7 @@
 from web3.exceptions import Web3RPCError
 from web3.types import Wei
 
+from compute_horde_validator.validator.dynamic_config import get_config
 from compute_horde_validator.validator.models import Miner
 
 from .base import CollateralBase
@@ -109,7 +109,7 @@ def slash_collateral(
 
         w3 = get_web3_connection(network=settings.BITTENSOR_NETWORK)
 
-        amount_wei = config.DYNAMIC_COLLATERAL_SLASH_AMOUNT_WEI
+        amount_wei = get_config("DYNAMIC_COLLATERAL_SLASH_AMOUNT_WEI")
         if amount_wei <= 0:
             raise SlashCollateralError("Slash amount must be greater than 0")
 

validator/app/src/compute_horde_validator/validator/collateral/tests/helpers/env.py

@@ -8,6 +8,7 @@
 
 from compute_horde_validator.validator.collateral import default as collateral_default
 from compute_horde_validator.validator.collateral.default import Collateral
+from compute_horde_validator.validator.tests.helpers import patch_constance
 
 from .stubs import (
     HttpStub,
@@ -46,6 +47,7 @@ def __init__(
         self.fetch_log: list[dict[str, Any]] = []
         self.system_events = SystemEventRecorder()
         self.contract_log: list[tuple[str, Any]] = []
+        self._constance_overlay: dict[str, Any] | None = None
 
     def __enter__(self) -> CollateralTestEnvironment:
         self._stack = ExitStack()
@@ -68,7 +70,12 @@ def __enter__(self) -> CollateralTestEnvironment:
         )
         self._config = Mock()
         self._config.DYNAMIC_COLLATERAL_SLASH_AMOUNT_WEI = self.slash_amount
-        self._stack.enter_context(patch.object(collateral_default, "config", self._config))
+        self._constance_overlay = {
+            "DYNAMIC_COLLATERAL_SLASH_AMOUNT_WEI": self.slash_amount,
+        }
+        self._stack.enter_context(
+            patch_constance(self._constance_overlay),
+        )
         self._stack.enter_context(
             patch(
                 "compute_horde_validator.validator.collateral.default.settings.BITTENSOR_WALLET",
@@ -139,6 +146,8 @@ def set_receipt(self, receipt: dict[str, Any]) -> None:
     def set_slash_amount(self, amount: int) -> None:
         self.slash_amount = amount
         self._config.DYNAMIC_COLLATERAL_SLASH_AMOUNT_WEI = amount
+        if self._constance_overlay is not None:
+            self._constance_overlay["DYNAMIC_COLLATERAL_SLASH_AMOUNT_WEI"] = amount
 
     def _record_transaction(
         self,

validator/app/src/compute_horde_validator/validator/dynamic_config.py

@@ -7,6 +7,10 @@
 from constance import config
 
 
+def get_config(key: str) -> Any:
+    return getattr(config, key)
+
+
 async def aget_config(key: str) -> Any:
     return await sync_to_async(lambda: getattr(config, key))()
 
@@ -30,7 +34,7 @@ def executor_class_value_map_parser(
 
 
 def get_miner_max_executors_per_class_sync() -> dict[ExecutorClass, int]:
-    miner_max_executors_per_class: str = config.DYNAMIC_MINER_MAX_EXECUTORS_PER_CLASS
+    miner_max_executors_per_class: str = get_config("DYNAMIC_MINER_MAX_EXECUTORS_PER_CLASS")
     result = {
         executor_class: count
         for executor_class, count in executor_class_value_map_parser(
@@ -43,5 +47,5 @@ def get_miner_max_executors_per_class_sync() -> dict[ExecutorClass, int]:
 
 def get_executor_class_weights() -> dict[ExecutorClass, float]:
     return executor_class_value_map_parser(
-        config.DYNAMIC_EXECUTOR_CLASS_WEIGHTS, value_parser=float
+        get_config("DYNAMIC_EXECUTOR_CLASS_WEIGHTS"), value_parser=float
     )

validator/app/src/compute_horde_validator/validator/routing/default.py

@@ -8,7 +8,6 @@
     V2JobRequest,
 )
 from compute_horde_core.executor_class import ExecutorClass
-from constance import config
 from django.conf import settings
 from django.db.models import Count
 from django.utils import timezone
@@ -22,6 +21,7 @@
     Miner as AllowanceMiner,
 )
 from compute_horde_validator.validator.collateral.default import collateral
+from compute_horde_validator.validator.dynamic_config import get_config
 from compute_horde_validator.validator.models import Miner, MinerIncident, SystemEvent
 from compute_horde_validator.validator.receipts.default import receipts
 from compute_horde_validator.validator.routing.base import RoutingBase
@@ -138,7 +138,7 @@ def _pick_miner_for_job_v2(request: V2JobRequest) -> JobRoute:
         raise
 
     # Collateral filtering: build a set of miners with sufficient collateral if threshold > 0
-    collateral_threshold = config.DYNAMIC_MINIMUM_COLLATERAL_AMOUNT_WEI
+    collateral_threshold = get_config("DYNAMIC_MINIMUM_COLLATERAL_AMOUNT_WEI")
     if collateral_threshold > 0:
         try:
             eligible_collateral_hotkeys = {
@@ -169,7 +169,9 @@ def _pick_miner_for_job_v2(request: V2JobRequest) -> JobRoute:
     manifests = allowance().get_manifests()
 
     reliability_score_per_hotkey = _get_miners_reliability_score(
-        reliability_window=timedelta(hours=float(config.DYNAMIC_ROUTING_RELIABILITY_WINDOW_HOURS)),
+        reliability_window=timedelta(
+            hours=float(get_config("DYNAMIC_ROUTING_RELIABILITY_WINDOW_HOURS"))
+        ),
         executor_class=executor_class,
         manifests=manifests,
     )
@@ -185,8 +187,8 @@ def _pick_miner_for_job_v2(request: V2JobRequest) -> JobRoute:
         items=suitable_hotkeys,
         weights=hotkey_weights,
         # With steepness >5 there is a strong cutoff at ~center*2, hence center~=cutoff/2
-        center=float(config.DYNAMIC_ROUTING_RELIABILITY_SOFT_CUTOFF) / 2,
-        steepness=float(config.DYNAMIC_ROUTING_RELIABILITY_SEPARATION),
+        center=float(get_config("DYNAMIC_ROUTING_RELIABILITY_SOFT_CUTOFF")) / 2,
+        steepness=float(get_config("DYNAMIC_ROUTING_RELIABILITY_SEPARATION")),
     )
 
     # Get receipts instance once to avoid bound method issues with async_to_sync

validator/app/src/compute_horde_validator/validator/scoring/calculations.py

@@ -5,7 +5,6 @@
 
 import numpy as np
 from compute_horde_core.executor_class import ExecutorClass
-from constance import config
 
 from compute_horde_validator.validator.allowance.default import allowance
 from compute_horde_validator.validator.allowance.types import (
@@ -14,6 +13,7 @@
     SpendingDetails,
 )
 from compute_horde_validator.validator.allowance.utils.spending import Triplet
+from compute_horde_validator.validator.dynamic_config import get_config
 from compute_horde_validator.validator.models import Miner, OrganicJob
 from compute_horde_validator.validator.receipts import receipts
 from compute_horde_validator.validator.scoring.metrics import (
@@ -167,8 +167,8 @@ def horde_score(
 def score_organic_jobs(jobs: Sequence[OrganicJob]) -> dict[str, float]:
     """Score organic jobs."""
     batch_scores: defaultdict[str, float] = defaultdict(float)
-    score = config.DYNAMIC_ORGANIC_JOB_SCORE
-    limit = config.DYNAMIC_SCORE_ORGANIC_JOBS_LIMIT
+    score = get_config("DYNAMIC_ORGANIC_JOB_SCORE")
+    limit = get_config("DYNAMIC_SCORE_ORGANIC_JOBS_LIMIT")
 
     for job in jobs:
         batch_scores[job.miner.hotkey] += score
@@ -202,7 +202,7 @@ def calculate_organic_scores(organic_jobs: list[OrganicJob]) -> dict[str, dict[s
             executor_class_organic_jobs[job.executor_class].append(job)
 
     organic_scores_by_executor: dict[str, dict[str, float]] = {}
-    organic_job_score = config.DYNAMIC_ORGANIC_JOB_SCORE
+    organic_job_score = get_config("DYNAMIC_ORGANIC_JOB_SCORE")
 
     for executor_class, jobs in executor_class_organic_jobs.items():
         executor_class_scores: dict[str, float] = {}

validator/app/src/compute_horde_validator/validator/scoring/tasks.py

@@ -20,6 +20,7 @@
 
 from compute_horde_validator.celery import app
 from compute_horde_validator.validator.clean_me_up import bittensor_client
+from compute_horde_validator.validator.dynamic_config import get_config
 from compute_horde_validator.validator.locks import Locked, LockType, get_advisory_lock
 from compute_horde_validator.validator.models import SystemEvent
 from compute_horde_validator.validator.models.scoring.internal import (
@@ -60,7 +61,7 @@ def set_scores(bittensor: turbobt.Bittensor):
         logger.warning("Not setting scores, SERVING is disabled in constance config")
         return
 
-    commit_reveal_weights_enabled = config.DYNAMIC_COMMIT_REVEAL_WEIGHTS_ENABLED
+    commit_reveal_weights_enabled = get_config("DYNAMIC_COMMIT_REVEAL_WEIGHTS_ENABLED")
 
     current_block = async_to_sync(bittensor.blocks.head)()
 
@@ -101,11 +102,21 @@ def set_scores(bittensor: turbobt.Bittensor):
                     SystemEvent.EventSubType.SUCCESS,
                     "",
                     {
-                        "DYNAMIC_COMMIT_REVEAL_WEIGHTS_INTERVAL": config.DYNAMIC_COMMIT_REVEAL_WEIGHTS_INTERVAL,
-                        "DYNAMIC_COMMIT_REVEAL_COMMIT_START_OFFSET": config.DYNAMIC_COMMIT_REVEAL_COMMIT_START_OFFSET,
-                        "DYNAMIC_COMMIT_REVEAL_COMMIT_END_BUFFER": config.DYNAMIC_COMMIT_REVEAL_COMMIT_END_BUFFER,
-                        "DYNAMIC_COMMIT_REVEAL_REVEAL_END_BUFFER": config.DYNAMIC_COMMIT_REVEAL_REVEAL_END_BUFFER,
-                        "DYNAMIC_COMMIT_REVEAL_WEIGHTS_ENABLED": config.DYNAMIC_COMMIT_REVEAL_WEIGHTS_ENABLED,
+                        "DYNAMIC_COMMIT_REVEAL_WEIGHTS_INTERVAL": get_config(
+                            "DYNAMIC_COMMIT_REVEAL_WEIGHTS_INTERVAL"
+                        ),
+                        "DYNAMIC_COMMIT_REVEAL_COMMIT_START_OFFSET": get_config(
+                            "DYNAMIC_COMMIT_REVEAL_COMMIT_START_OFFSET"
+                        ),
+                        "DYNAMIC_COMMIT_REVEAL_COMMIT_END_BUFFER": get_config(
+                            "DYNAMIC_COMMIT_REVEAL_COMMIT_END_BUFFER"
+                        ),
+                        "DYNAMIC_COMMIT_REVEAL_REVEAL_END_BUFFER": get_config(
+                            "DYNAMIC_COMMIT_REVEAL_REVEAL_END_BUFFER"
+                        ),
+                        "DYNAMIC_COMMIT_REVEAL_WEIGHTS_ENABLED": get_config(
+                            "DYNAMIC_COMMIT_REVEAL_WEIGHTS_ENABLED"
+                        ),
                         "current_block_number": current_block.number,
                     },
                 )
@@ -212,12 +223,16 @@ def __init__(
         reveal_end_buffer: int | None = None,
     ):
         self.current_block = current_block
-        self.length = length or config.DYNAMIC_COMMIT_REVEAL_WEIGHTS_INTERVAL
-        self.commit_start_offset = (
-            commit_start_offset or config.DYNAMIC_COMMIT_REVEAL_COMMIT_START_OFFSET
+        self.length = length or get_config("DYNAMIC_COMMIT_REVEAL_WEIGHTS_INTERVAL")
+        self.commit_start_offset = commit_start_offset or get_config(
+            "DYNAMIC_COMMIT_REVEAL_COMMIT_START_OFFSET"
+        )
+        self.commit_end_buffer = commit_end_buffer or get_config(
+            "DYNAMIC_COMMIT_REVEAL_COMMIT_END_BUFFER"
+        )
+        self.reveal_end_buffer = reveal_end_buffer or get_config(
+            "DYNAMIC_COMMIT_REVEAL_REVEAL_END_BUFFER"
         )
-        self.commit_end_buffer = commit_end_buffer or config.DYNAMIC_COMMIT_REVEAL_COMMIT_END_BUFFER
-        self.reveal_end_buffer = reveal_end_buffer or config.DYNAMIC_COMMIT_REVEAL_REVEAL_END_BUFFER
 
     @cached_property
     def start(self):
@@ -386,9 +401,9 @@ def apply_dancing_burners(
     min_allowed_weights: int,
     max_weight_limit: int,
 ) -> tuple["torch.Tensor", "torch.FloatTensor"] | tuple[NDArray[np.int64], NDArray[np.float32]]:
-    burner_hotkeys = config.DYNAMIC_BURN_TARGET_SS58ADDRESSES.split(",")
-    burn_rate = config.DYNAMIC_BURN_RATE
-    burn_partition = config.DYNAMIC_BURN_PARTITION
+    burner_hotkeys = get_config("DYNAMIC_BURN_TARGET_SS58ADDRESSES").split(",")
+    burn_rate = get_config("DYNAMIC_BURN_RATE")
+    burn_partition = get_config("DYNAMIC_BURN_PARTITION")
 
     hotkey_to_uid = {neuron.hotkey: neuron.uid for neuron in neurons}
     registered_burner_hotkeys = sorted(