Merge pull request #375 from macrocosm-os/staging

Staging

Author: schampoux

.env.example

@@ -1,5 +1,6 @@
 S3_REGION=nyc3
 S3_ENDPOINT=https://nyc3.digitaloceanspaces.com
+S3_BUCKET=sn25-folding-mainnet
 S3_KEY=s3_key
 S3_SECRET=secret_key
 RQLITE_HTTP_ADDR=0.0.0.0:4001

documentation/get_started.md

@@ -51,6 +51,7 @@ Bash install will use poetry to build the environment correctly.
 SN25 uses DigitalOcean S3 data buckets for data transfer. Therefore, the following environment variables need to be set in your system or application environment (`.env` file):
 - `S3_REGION = "nyc3"`: The AWS region or S3-compatible region where the bucket is located.
 - `S3_ENDPOINT = "https://nyc3.digitaloceanspaces.com"`: The endpoint URL for your S3-compatible service.
+- `S3_BUCKET = sn25-folding-mainnet`: The name of the s3 bucket. 
 - `S3_KEY`: Your S3 access key ID.
 - `S3_SECRET`: Your S3 secret access key.
 

folding/__init__.py

@@ -1,4 +1,4 @@
-__version__ = "2.1.2"
+__version__ = "2.1.3"
 version_split = __version__.split(".")
 __spec_version__ = (
     (10000 * int(version_split[0]))

folding/registries/miner_registry.py

@@ -56,12 +56,29 @@ def add_credibilities(self, miner_uid: int, task: str, credibilities: List[float
 
     def update_credibility(self, miner_uid: int, task: str):
         """
-        Updates the credibility of a miner based:
-        1. The credibility of the miner's previous results. Intially set as STARTING_CREDIBILITY
+        Updates the credibility of a miner based on:
+        1. The credibility of the miner's previous results. Initially set as STARTING_CREDIBILITY
         2. The credibility of the miner's current results.
         3. The number of previous and current entries to act as a weighting factor
         4. The EMA with credibility_alpha as the smoothing factor
+
+        If the miner_uid doesn't exist in the registry, it will be instantiated first.
+
+        Args:
+            miner_uid (int): The unique identifier of the miner
+            task (str): The task name to update credibility for
         """
+        # Check if miner_uid exists, if not instantiate it
+        if miner_uid not in self.registry:
+            self.registry[miner_uid] = {}
+            self.registry[miner_uid]["overall_credibility"] = c.STARTING_CREDIBILITY
+            for task_name in self.tasks:
+                self.registry[miner_uid][task_name] = {
+                    "credibility": c.STARTING_CREDIBILITY,
+                    "credibilities": [],
+                    "score": 0.0,
+                    "results": [],
+                }
 
         task_credibilities = list(
             chain.from_iterable(self.registry[miner_uid][task]["credibilities"])

folding/utils/config.py

@@ -254,7 +254,7 @@ def add_args(cls, parser):
         "--s3.bucket_name",
         type=str,
         help="The name of the S3 bucket to log to.",
-        default="vali-s3-demo-do",
+        default="sn25-folding-mainnet",
     )
 
 

folding/utils/s3_utils.py

@@ -86,7 +86,8 @@ async def upload_to_s3(
     try:
         s3_links = {}
         input_time = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
-
+        s3_endpoint = os.getenv("S3_ENDPOINT")
+        s3_bucket = os.getenv("S3_BUCKET")
         for file_type in ["pdb", "cpt"]:
             if file_type == "cpt":
                 file_path = os.path.join(validator_directory, simulation_cpt)
@@ -104,9 +105,7 @@ async def upload_to_s3(
                 location=location,
                 public=True,
             )
-            s3_links[file_type] = os.path.join(
-                "https://nyc3.digitaloceanspaces.com/vali-s3-demo-do/", key
-            )
+            s3_links[file_type] = os.path.join(f"{s3_endpoint}/{s3_bucket}/", key)
             await asyncio.sleep(0.10)
 
         return s3_links
@@ -138,6 +137,9 @@ async def upload_output_to_s3(
     Raises:
         Exception: If any error occurs during file upload.
     """
+    s3_endpoint = os.getenv("S3_ENDPOINT")
+    s3_bucket = os.getenv("S3_BUCKET")
+
     try:
         output_time = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
         location = os.path.join(
@@ -149,7 +151,7 @@ async def upload_output_to_s3(
             location=location,
             public=True,
         )
-        return os.path.join("https://nyc3.digitaloceanspaces.com/vali-s3-demo-do/", key)
+        return os.path.join(f"{s3_endpoint}/{s3_bucket}/", key)
     except Exception as e:
         logger.error(f"Exception during output file upload: {str(e)}")
         raise

folding/validators/forward.py

@@ -297,7 +297,7 @@ async def try_prepare_md_challenge(self, config, pdb_id: str) -> Dict:
             if "validator_search_status" not in event:
                 if not config.s3.off:
                     try:
-                        logger.info(f"Uploading to {self.handler}")
+                        logger.info(f"Uploading to {self.handler.bucket_name}")
                         s3_links = await upload_to_s3(
                             handler=self.handler,
                             pdb_location=protein.pdb_location,

folding/validators/reward.py

@@ -50,11 +50,10 @@ def evaluate(
     responses: List[JobSubmissionSynapse],
     uids: List[int],
     job_type: str,
-    event: dict,
 ):
     reported_energies = np.zeros(len(uids))
     evaluators = [None] * len(uids)
-    seed = []
+    seed = [-1] * len(uids)
     best_cpt = [""] * len(uids)
     process_md_output_time = [0.0] * len(uids)
 
@@ -67,6 +66,7 @@ def evaluate(
                 continue
 
             start_time = time.time()
+            seed[i] = resp.miner_seed
             evaluator = EVALUATION_REGISTRY[job_type](
                 pdb_id=protein.pdb_id,
                 pdb_location=protein.pdb_location,
@@ -82,7 +82,6 @@ def evaluate(
             can_process = evaluator.evaluate()
             if not can_process:
                 continue
-            seed.append(resp.miner_seed)
             best_cpt[i] = (
                 evaluator.checkpoint_path
                 if hasattr(evaluator, "checkpoint_path")
@@ -135,7 +134,7 @@ def get_energies(
 
     # Get initial evaluations
     reported_energies, evaluators, seed, best_cpt, process_md_output_time = evaluate(
-        protein, responses, uids, job_type, event
+        protein, responses, uids, job_type
     )
 
     # Sort all lists by reported energy
@@ -157,7 +156,7 @@ def get_energies(
     unique_energies = set()  # Track unique energy values
 
     # Process responses until we get TOP_K valid non-duplicate ones or run out of responses
-    for i, (reported_energy, response, uid, evaluator, s, bc, pmt) in enumerate(
+    for i, (reported_energy, response, uid, evaluator, seed, best_cpt, process_md_output_time) in enumerate(
         sorted_data
     ):
         try:

neurons/validator.py

@@ -26,7 +26,7 @@
 from folding.store import Job, SQLiteJobStore
 from folding.utils.logger import logger
 from folding.utils.logging import log_event
-from folding.utils.uids import get_random_uids
+from folding.utils.uids import get_all_miner_uids
 from folding.utils.s3_utils import upload_output_to_s3
 from folding.utils.s3_utils import DigitalOceanS3Handler
 from folding.validators.forward import create_new_challenge, run_ping_step, run_step
@@ -46,9 +46,7 @@ def __init__(self, config=None):
 
         # Sample all the uids on the network, and return only the uids that are non-valis.
         logger.info("Determining all miner uids...⏳")
-        self.all_miner_uids: List = get_random_uids(
-            self, k=int(self.metagraph.n), exclude=None
-        ).tolist()
+        self.all_miner_uids: List = get_all_miner_uids(self)
 
         # If we do not have any miner registry saved to the machine, create.
         if not hasattr(self, "miner_registry"):
@@ -316,14 +314,15 @@ async def update_job(self, job: Job):
             # If there is an exploit on the cpt file detected via the state-checkpoint, reduce score.
             if reason == "state-checkpoint":
                 logger.warning(
-                    f"Setting uid {uid} score to zero, State-checkpoint check failed."
+                    f"Reducing uid {uid} score, State-checkpoint check failed."
                 )
                 self.scores[uid] = 0.5 * self.scores[uid]
 
-            credibility = [0.0] if reason != "" else [1.0]
+            credibility = [0.0] if reason != "valid" else [1.0]
             self.miner_registry.add_credibilities(
                 miner_uid=uid, task=job.job_type, credibilities=credibility
             )
+            self.miner_registry.update_credibility(miner_uid=uid, task=job.job_type)
 
         best_index = np.argmin(energies)
         best_loss = energies[best_index].item()  # item because it's a torch.tensor
@@ -487,7 +486,6 @@ async def update_jobs(self):
                 await asyncio.sleep(self.config.neuron.update_interval)
 
                 logger.info("Updating jobs.")
-                logger.info(f"step({self.step}) block({self.block})")
 
                 for job in self.store.get_queue(
                     ready=True, validator_hotkey=self.wallet.hotkey.ss58_address
@@ -513,6 +511,8 @@ async def update_jobs(self):
                     # Determine the status of the job based on the current energy and the previous values (early stopping)
                     # Update the DB with the current status
                     await self.update_job(job=job)
+                logger.info(f"step({self.step}) block({self.block})")
+
             except Exception as e:
                 logger.error(f"Error in update_jobs: {traceback.format_exc()}")
 
@@ -571,7 +571,12 @@ async def monitor_db(self):
         while True:
             try:
                 await asyncio.sleep(60)
-                outdated = await self.store.monitor_db()
+                try:
+                    outdated = await self.store.monitor_db()
+                except Exception as e:
+                    logger.error(f"Error in monitor_db: {traceback.format_exc()}")
+                    await self.start_rqlite()
+
                 if outdated:
                     logger.error("Database is outdated. Restarting rqlite.")
                     await self.start_rqlite()