fix(jobs): prevent Docker job scheduling races under queue pressure

services/core/jobs/src/nmp/core/jobs/controllers/backends/exceptions.py

@@ -18,5 +18,13 @@ def __init__(self, message: str = "Failed to allocate resource"):
         self.message = message
 
 
+class SchedulingDeferred(Exception):
+    """Exception raised when a backend has no immediate capacity to accept a step."""
+
+    def __init__(self, message: str = "Scheduling deferred"):
+        super().__init__(message)
+        self.message = message
+
+
 class JobStorageError(Exception):
     """Exception raised when there's an issue with job storage."""

services/core/jobs/src/nmp/core/jobs/controllers/backends/subprocess.py

@@ -161,15 +161,6 @@ def shutdown(self) -> None:
             self._finish_logs(metadata)
 
     def schedule(self, executor_config: SubprocessExecutionProvider, step: PlatformJobStepWithContext) -> JobUpdate:
-        ttl_seconds = self._execution_profile_config.ttl_seconds_before_active
-        if self.should_enforce_before_active_ttl(step) and self.check_step_ttl_before_active(step, ttl_seconds):
-            message = f"Job timed out after reaching max TTL of {ttl_seconds} seconds"
-            return JobUpdate(
-                status=PlatformJobStatus.ERROR.value,
-                status_details={"message": message},
-                error_details={"message": message},
-            )
-
         if not executor_config.command:
             return JobUpdate(
                 status=PlatformJobStatus.ERROR.value,

services/core/jobs/src/nmp/core/jobs/controllers/reconciler.py

@@ -3,6 +3,7 @@
 
 import logging
 import threading
+import time
 
 from nemo_platform import APIError, APIStatusError, NeMoPlatform
 from nemo_platform.types.jobs import PlatformJobStepWithContext
@@ -54,6 +55,7 @@ def step(self):
             logger.debug("Stop signal received, skipping reconciliation step")
             return
 
+        fetch_started_at = time.monotonic()
         with tracer.start_as_current_span("jobs_reconciler/fetch_steps_for_reconciliation"):
             try:
                 statuses: list[SDKPlatformJobStatus] = [
@@ -73,6 +75,10 @@ def step(self):
             logger.info(f"Got {len(steps_to_reconcile)} job steps to reconcile")
         else:
             logger.debug("No job steps to reconcile")
+        logger.debug(
+            "Reconciler fetched job steps",
+            extra={"count": len(steps_to_reconcile), "duration_seconds": time.monotonic() - fetch_started_at},
+        )
         for step in steps_to_reconcile:
             with start_span_with_ctx(
                 tracer,
@@ -86,7 +92,21 @@ def step(self):
                     with scoped_app_ctx(
                         JobBackendContext(provider=provider, profile=profile, name=str(backend)),
                     ):
+                        sync_started_at = time.monotonic()
                         job_update = backend.sync(step)
+                        logger.debug(
+                            "Reconciler backend sync completed",
+                            extra={
+                                "job": step.job,
+                                "step": step.name,
+                                "workspace": step.workspace,
+                                "provider": provider,
+                                "profile": profile,
+                                "from_status": step.status,
+                                "to_status": job_update.status,
+                                "duration_seconds": time.monotonic() - sync_started_at,
+                            },
+                        )
                         logger.info(f"Updating job step status from '{step.status}' to '{job_update.status}'")
                         if (
                             job_update.status == PlatformJobStatus.ERROR.value
@@ -98,13 +118,13 @@ def step(self):
                                 logger=self._logger,
                                 context="step transitioned to error during reconciliation",
                             )
-                        self._nmp_sdk.jobs.steps.update_status(
-                            step.name,
-                            workspace=step.workspace,
-                            job=step.job,
+                        self._update_step_status_with_timing(
+                            step=step,
+                            provider=provider,
+                            profile=profile,
                             status=job_update.status,
-                            status_details=job_update.status_details,  # type: ignore
-                            error_details=job_update.error_details,  # type: ignore
+                            status_details=job_update.status_details,
+                            error_details=job_update.error_details,
                         )
                 except APIStatusError as e:
                     # In cases when attempting to update job step status results in a conflict (409),
@@ -149,6 +169,56 @@ def step(self):
                 except Exception:
                     logger.exception("Could not complete cleanup steps for backend", exc_info=True)
 
+    def _update_step_status_with_timing(
+        self,
+        *,
+        step: PlatformJobStepWithContext,
+        provider: str,
+        profile: str,
+        status: str,
+        status_details: dict | None = None,
+        error_details: dict | None = None,
+    ):
+        started_at = time.monotonic()
+        try:
+            response = self._nmp_sdk.jobs.steps.update_status(
+                step.name,
+                workspace=step.workspace,
+                job=step.job,
+                status=status,
+                status_details=status_details,  # type: ignore
+                error_details=error_details,  # type: ignore
+            )
+        except Exception:
+            logger.warning(
+                "Reconciler step status update failed",
+                extra={
+                    "job": step.job,
+                    "step": step.name,
+                    "workspace": step.workspace,
+                    "provider": provider,
+                    "profile": profile,
+                    "from_status": step.status,
+                    "to_status": status,
+                    "duration_seconds": time.monotonic() - started_at,
+                },
+            )
+            raise
+        logger.debug(
+            "Reconciler step status update succeeded",
+            extra={
+                "job": step.job,
+                "step": step.name,
+                "workspace": step.workspace,
+                "provider": provider,
+                "profile": profile,
+                "from_status": step.status,
+                "to_status": status,
+                "duration_seconds": time.monotonic() - started_at,
+            },
+        )
+        return response
+
     def get_steps_for_reconciliation(self, statuses: list[SDKPlatformJobStatus]) -> list[PlatformJobStepWithContext]:
         """
         Return the list of steps to reconcile.

services/core/jobs/src/nmp/core/jobs/controllers/scheduler.py

@@ -3,18 +3,21 @@
 
 import logging
 import threading
+import time
 import traceback
+from typing import TypedDict, cast, get_args
 
 import nemo_platform
 from nemo_platform import APIStatusError, NeMoPlatform
+from nemo_platform.types import PlatformJobStatus as SDKPlatformJobStatus
 from nemo_platform.types.jobs import PlatformJobStepWithContext
 from nemo_platform.types.jobs.platform_job_steps_list_filter_param import PlatformJobStepsListFilterParam
 from nmp.common.controller import Controller
 from nmp.common.jobs.schemas import PlatformJobStatus
 from nmp.common.observability import start_span_with_ctx
 from nmp.core.jobs.app.ctx import JobBackendContext, JobContext
 from nmp.core.jobs.controllers.backends import JobUpdate, extract_provider_profile
-from nmp.core.jobs.controllers.backends.exceptions import ResourceAllocationError
+from nmp.core.jobs.controllers.backends.exceptions import ResourceAllocationError, SchedulingDeferred
 from nmp.core.jobs.controllers.backends.registry import BackendRegistry
 from nmp.core.jobs.controllers.diagnostics import log_job_diagnostics_if_debug
 from opentelemetry import metrics, trace
@@ -25,6 +28,18 @@
 
 DEFAULT_PROFILE = "default"
 DEFAULT_PROVIDER = "cpu"
+SDK_PLATFORM_JOB_STATUSES = frozenset(get_args(SDKPlatformJobStatus))
+
+
+class StepStatusDetailParams(TypedDict, total=False):
+    status_details: dict[str, object]
+    error_details: dict[str, object]
+
+
+def as_sdk_platform_job_status(status: str) -> SDKPlatformJobStatus:
+    if status not in SDK_PLATFORM_JOB_STATUSES:
+        raise ValueError(f"Unsupported platform job status: {status}")
+    return cast(SDKPlatformJobStatus, status)
 
 
 class JobScheduler(Controller):
@@ -60,6 +75,7 @@ def step(self):
             return
 
         steps = []
+        fetch_started_at = time.monotonic()
         with tracer.start_as_current_span("jobs_scheduler/fetch_steps_for_scheduling"):
             try:
                 steps = self.get_steps_for_scheduling()
@@ -73,21 +89,34 @@ def step(self):
             logger.info(f"Got {len(steps)} job steps to schedule")
         else:
             logger.debug("No job steps to schedule")
+        logger.debug(
+            "Scheduler fetched job steps",
+            extra={"count": len(steps), "duration_seconds": time.monotonic() - fetch_started_at},
+        )
         for step in steps:
             with start_span_with_ctx(
                 tracer, "jobs_scheduler/schedule_step", JobContext(id=step.job, step_name=step.name)
             ):
                 try:
+                    schedule_started_at = time.monotonic()
                     update = self.schedule_step(step)
-                    logger.info("Scheduled job step")
+                    logger.debug(
+                        "Scheduled job step",
+                        extra={
+                            "job": step.job,
+                            "step": step.name,
+                            "workspace": step.workspace,
+                            "duration_seconds": time.monotonic() - schedule_started_at,
+                            "status": update.status,
+                        },
+                    )
                     try:
-                        self._nmp_sdk.jobs.steps.update_status(
-                            step.name,
-                            workspace=step.workspace,
-                            job=step.job,
+                        self._update_step_status_with_timing(
+                            step=step,
+                            phase="schedule",
                             status=update.status,
-                            status_details=update.status_details,  # type: ignore
-                            error_details=update.error_details,  # type: ignore
+                            status_details=update.status_details,
+                            error_details=update.error_details,
                         )
                     except APIStatusError as e:
                         # Stopgap for a scheduler/reconciler race: by the time the scheduler persists
@@ -119,14 +148,23 @@ def step(self):
                         context="resource allocation error during scheduling",
                     )
                     self._step_scheduling_errors.add(1, attributes={"error_type": "resource_allocation"})
-                    self._nmp_sdk.jobs.steps.update_status(
-                        step.name,
-                        workspace=step.workspace,
-                        job=step.job,
+                    self._update_step_status_with_timing(
+                        step=step,
+                        phase="resource_allocation_error",
                         status=PlatformJobStatus.ERROR.value,
                         status_details={"message": e.message},
                         error_details={"message": e.message},
                     )
+                except SchedulingDeferred as e:
+                    logger.debug(
+                        "Scheduling deferred for job step",
+                        extra={
+                            "job": step.job,
+                            "step": step.name,
+                            "workspace": step.workspace,
+                            "reason": e.message,
+                        },
+                    )
                 except Exception as e:
                     logger.exception("Could not schedule job step", exc_info=True)
                     log_job_diagnostics_if_debug(
@@ -136,15 +174,65 @@ def step(self):
                         context="unexpected scheduling error",
                     )
                     self._step_scheduling_errors.add(1, attributes={"error_type": "unknown"})
-                    self._nmp_sdk.jobs.steps.update_status(
-                        step.name,
-                        workspace=step.workspace,
-                        job=step.job,
+                    self._update_step_status_with_timing(
+                        step=step,
+                        phase="unexpected_error",
                         status=PlatformJobStatus.ERROR.value,
                         status_details={"message": str(e)},
                         error_details={"message": str(e), "error": traceback.format_exc()},
                     )
 
+    def _update_step_status_with_timing(
+        self,
+        *,
+        step: PlatformJobStepWithContext,
+        phase: str,
+        status: str,
+        status_details: dict[str, object] | None = None,
+        error_details: dict[str, object] | None = None,
+    ):
+        started_at = time.monotonic()
+        detail_params: StepStatusDetailParams = {}
+        if status_details is not None:
+            detail_params["status_details"] = status_details
+        if error_details is not None:
+            detail_params["error_details"] = error_details
+        try:
+            response = self._nmp_sdk.jobs.steps.update_status(
+                step.name,
+                workspace=step.workspace,
+                job=step.job,
+                status=as_sdk_platform_job_status(status),
+                **detail_params,
+            )
+        except Exception:
+            logger.warning(
+                "Scheduler step status update failed",
+                extra={
+                    "job": step.job,
+                    "step": step.name,
+                    "workspace": step.workspace,
+                    "phase": phase,
+                    "from_status": step.status,
+                    "to_status": status,
+                    "duration_seconds": time.monotonic() - started_at,
+                },
+            )
+            raise
+        logger.debug(
+            "Scheduler step status update succeeded",
+            extra={
+                "job": step.job,
+                "step": step.name,
+                "workspace": step.workspace,
+                "phase": phase,
+                "from_status": step.status,
+                "to_status": status,
+                "duration_seconds": time.monotonic() - started_at,
+            },
+        )
+        return response
+
     def get_steps_for_scheduling(self) -> list[PlatformJobStepWithContext]:
         """
         Return the oldest set of steps to schedule. We using the
@@ -159,7 +247,7 @@ def get_steps_for_scheduling(self) -> list[PlatformJobStepWithContext]:
             name="-",  # Use "-" to indicate all jobs
             workspace="-",  # Cross-workspace query
             filter=filter_params,
-            sort="-created_at",
+            sort="created_at",
         ):
             steps.append(step)
         return steps