Merge pull request #577 from backend-developers-ltd/job-error-propagation

Job error propagation

Author: kkowalski-reef

compute_horde/compute_horde/fv_protocol/validator_requests.py

@@ -1,9 +1,18 @@
-from enum import StrEnum
 from typing import Any, Literal, Self
 
 import bittensor
 from pydantic import BaseModel
 
+from compute_horde.protocol_consts import (
+    HordeFailureReason,
+    JobFailureReason,
+    JobParticipantType,
+    JobRejectionReason,
+    JobStage,
+    JobStatus,
+)
+from compute_horde.protocol_messages import FailureContext
+
 
 class V0Heartbeat(BaseModel, extra="forbid"):
     """Message sent from validator to facilitator to keep connection alive"""
@@ -39,46 +48,71 @@ def ss58_address(self) -> str:
         return address
 
 
-class MinerResponse(BaseModel, extra="allow"):
-    job_uuid: str
-    message_type: str | None
+class JobResultDetails(BaseModel, extra="allow"):
     docker_process_stderr: str
     docker_process_stdout: str
     artifacts: dict[str, str] | None = None
     upload_results: dict[str, str] | None = None
 
 
+class JobRejectionDetails(BaseModel):
+    rejected_by: JobParticipantType
+    reason: JobRejectionReason
+    message: str
+    context: FailureContext | None = None
+
+
+class JobFailureDetails(BaseModel):
+    reason: JobFailureReason
+    stage: JobStage
+    message: str
+    context: FailureContext | None = None
+    docker_process_exit_status: int | None = None
+    docker_process_stdout: str | None = None
+    docker_process_stderr: str | None = None
+
+
+class HordeFailureDetails(BaseModel):
+    reported_by: JobParticipantType
+    reason: HordeFailureReason
+    message: str
+    context: FailureContext | None = None
+
+
 class StreamingServerDetails(BaseModel, extra="forbid"):
     streaming_server_cert: str | None = None
     streaming_server_address: str | None = None
     streaming_server_port: int | None = None
 
 
 class JobStatusMetadata(BaseModel, extra="allow"):
-    comment: str
-    miner_response: MinerResponse | None = None
+    miner_response: JobResultDetails | None = None
+    job_rejection_details: JobRejectionDetails | None = None
+    job_failure_details: JobFailureDetails | None = None
+    horde_failure_details: HordeFailureDetails | None = None
     streaming_details: StreamingServerDetails | None = None
 
+    @classmethod
+    def from_uncaught_exception(cls, reported_by: JobParticipantType, exception: Exception) -> Self:
+        return cls(
+            horde_failure_details=HordeFailureDetails(
+                reported_by=reported_by,
+                reason=HordeFailureReason.UNHANDLED_EXCEPTION,
+                message="Uncaught exception",
+                context={"exception_type": type(exception).__qualname__},
+            ),
+        )
+
 
 class JobStatusUpdate(BaseModel, extra="forbid"):
+    # TODO(post error propagation): remove "extra"
     """
-    Message sent from validator to facilitator in response to NewJobRequest.
+    Message sent from validator to facilitator when the job's state changes.
     """
 
-    class Status(StrEnum):
-        RECEIVED = "received"
-        ACCEPTED = "accepted"
-        EXECUTOR_READY = "executor_ready"
-        VOLUMES_READY = "volumes_ready"
-        EXECUTION_DONE = "execution_done"
-        COMPLETED = "completed"
-        REJECTED = "rejected"
-        FAILED = "failed"
-        STREAMING_READY = "streaming_ready"
-
     message_type: Literal["V0JobStatusUpdate"] = "V0JobStatusUpdate"
     uuid: str
-    status: Status
+    status: JobStatus
     metadata: JobStatusMetadata | None = None
 
 

compute_horde/compute_horde/job_errors.py

@@ -0,0 +1,86 @@
+import logging
+from typing import Generic, TypeVar
+
+from compute_horde.protocol_consts import HordeFailureReason, JobFailureReason, JobRejectionReason
+from compute_horde.protocol_messages import FailureContext
+
+ReasonType = TypeVar("ReasonType")
+
+logger = logging.getLogger(__name__)
+
+
+class BaseJobError(Exception, Generic[ReasonType]):
+    """
+    Common structure for job-related errors.
+    """
+
+    def __init__(self, message: str, reason: ReasonType, context: FailureContext | None = None):
+        """
+        Parameters:
+            message: str
+                Short message for a human seeing the error somewhere.
+            reason: ReasonType
+                Structured error reason.
+            context: FailureContext | None
+                Optional arbitrary key-value context to be sent along the error.
+                Avoid
+        """
+        super().__init__(message)
+        self.message = message
+        self.reason = reason
+        self.context = context
+
+    def add_context(self, context: FailureContext):
+        if self.context is None:
+            self.context = {**context}
+        else:
+            self.context.update(context)
+
+    def __str__(self):
+        return f"{self.message}, {self.reason=}"
+
+    def __repr__(self):
+        return f"{type(self).__name__}: {str(self)}"
+
+
+class JobRejection(BaseJobError[JobRejectionReason]):
+    def __str__(self):
+        return f"Job rejected ({self.reason}): {self.message} "
+
+
+class JobError(BaseJobError[JobFailureReason]):
+    def __str__(self):
+        return f"Job failed ({self.reason}): {self.message} "
+
+
+class HordeError(BaseJobError[HordeFailureReason]):
+    def __init__(
+        self,
+        message: str,
+        reason: HordeFailureReason = HordeFailureReason.GENERIC_ERROR,
+        context: FailureContext | None = None,
+    ):
+        super().__init__(message, reason, context)
+
+    def __str__(self):
+        return f"Horde failed ({self.reason}): {self.message}"
+
+    @classmethod
+    def wrap_unhandled(cls, e: Exception, context: FailureContext | None = None) -> "HordeError":
+        """
+        Intended for "catch-all" except blocks - consistently wraps the exception in a HordeError.
+        For catching errors somewhere in the job code, do instead:
+            `raise HordeError("Spline reticulation failed") from e`
+        """
+        if isinstance(e, HordeError):
+            # Don't wrap another horde failure
+            return e
+
+        if isinstance(e, BaseJobError):
+            # If another job error is to be wrapped by accident - don't blow up, but scream for help
+            logger.error(f"Wrapping a {type(e).__qualname__} in a HordeError", exc_info=False)
+
+        failure = cls("Unhandled exception", HordeFailureReason.UNHANDLED_EXCEPTION, context)
+        failure.__cause__ = e
+        failure.add_context({"exception_type": type(e).__qualname__})
+        return failure