feat: make organic job execution flow sync

Impacts: validator

Author: emnoor-reef

validator/app/src/compute_horde_validator/settings.py

@@ -134,6 +134,7 @@ def wrapped(*args, **kwargs):
 CONSTANCE_BACKEND = "constance.backends.database.DatabaseBackend"
 CONSTANCE_DATABASE_CACHE_BACKEND = "default"
 CONSTANCE_CONFIG = {
+    "SYNC_ORGANIC_JOBS": (False, "SYNC_ORGANIC_JOBS", bool),
     "SERVING": (
         not env.bool("MIGRATING", default=False),
         "Whether this validator is serving jobs and setting weights",
@@ -257,6 +258,11 @@ def wrapped(*args, **kwargs):
         "Maximum number of organic jobs each miner can get scores for. Negative value means unlimited.",
         int,
     ),
+    "DYNAMIC_MAX_OVERALL_ORGANIC_JOB_TIME_LIMIT": (
+        300,
+        "Overall time to complete an organic job",
+        int,
+    ),
     "DYNAMIC_EXECUTOR_RESERVATION_TIME_LIMIT": (
         7,
         "Time for miner to accept or decline an organic job in seconds",

validator/app/src/compute_horde_validator/validator/job_excuses.py

@@ -114,3 +114,30 @@ async def get_expected_miner_executor_count(
         return 0
 
     return latest_manifest.online_executor_count
+
+
+def get_expected_miner_executor_count_sync(
+    check_time: datetime,
+    miner_hotkey: str,
+    executor_class: ExecutorClass,
+) -> int:
+    latest_manifest = (
+        MinerManifest.objects.filter(
+            miner__hotkey=miner_hotkey,
+            executor_class=executor_class,
+            created_at__lte=check_time,
+        )
+        .order_by("created_at")
+        .only("online_executor_count")
+        .first()
+    )
+
+    if latest_manifest is None:
+        logger.warning(
+            f"Cannot check expected miner executor count: "
+            f"manifest not found "
+            f"({miner_hotkey} {executor_class} {check_time})"
+        )
+        return 0
+
+    return latest_manifest.online_executor_count

validator/app/src/compute_horde_validator/validator/management/commands/debug_run_organic_job.py

@@ -1,5 +1,6 @@
 import shlex
 import sys
+import time
 import uuid
 
 from asgiref.sync import async_to_sync
@@ -125,6 +126,11 @@ def handle(self, *args, **options):
             streaming_start_time_limit=options["streaming_start_time_limit"],
         )
 
+        if settings.DEBUG_USE_MOCK_BLOCK_NUMBER:
+            block = 5136476 + int((time.time() - 1742076533) / 12)
+        else:
+            block = allowance().get_current_block()
+
         job = OrganicJob.objects.create(
             job_uuid=str(job_request.uuid),
             miner=miner,
@@ -134,7 +140,7 @@ def handle(self, *args, **options):
             namespace=job_request.job_namespace or job_request.docker_image or None,
             executor_class=job_request.executor_class,
             job_description="User job from facilitator",
-            block=allowance().get_current_block(),
+            block=block,
         )
 
         async def _run_job():