added artifacts_dir and artifacts fields to job messages and results

Author: adal-chiriliuc-reef

compute_horde/compute_horde/em_protocol/executor_requests.py

@@ -55,6 +55,7 @@ class V0FinishedRequest(BaseExecutorRequest, JobMixin):
     message_type: RequestType = RequestType.V0FinishedRequest
     docker_process_stdout: str  # TODO: add max_length
     docker_process_stderr: str  # TODO: add max_length
+    artifacts: dict[str, str] | None = None
 
 
 class GenericError(BaseExecutorRequest):

compute_horde/compute_horde/em_protocol/miner_requests.py

@@ -48,6 +48,7 @@ class V0JobRequest(BaseMinerRequest, JobMixin):
     docker_run_cmd: list[str]
     volume: Volume | None = None
     output_upload: OutputUpload | None = None
+    artifacts_dir: str | None = None
 
     @model_validator(mode="after")
     def validate_at_least_docker_image_or_raw_script(self) -> Self:

compute_horde/compute_horde/fv_protocol/facilitator_requests.py

@@ -122,6 +122,7 @@ class SignedFields(BaseModel):
     args: str
     env: dict[str, str]
     use_gpu: bool
+    artifacts_dir: str
 
     volumes: list[JsonValue]
     uploads: list[JsonValue]
@@ -137,6 +138,7 @@ def from_facilitator_sdk_json(data: JsonValue):
             args=str(data.get("args", "")),
             env=typing.cast(dict[str, str], data.get("env", None)),
             use_gpu=typing.cast(bool, data.get("use_gpu")),
+            artifacts_dir=str(data.get("artifacts_dir", "")),
             volumes=typing.cast(list[JsonValue], data.get("volumes", [])),
             uploads=typing.cast(list[JsonValue], data.get("uploads", [])),
         )
@@ -166,6 +168,7 @@ class V2JobRequest(BaseModel, extra="forbid"):
     use_gpu: bool
     volume: Volume | None = None
     output_upload: OutputUpload | None = None
+    artifacts_dir: str | None = None
     # !!! all fields above are included in the signed json payload
 
     def get_args(self):
@@ -198,6 +201,7 @@ def get_signed_fields(self) -> SignedFields:
             args=" ".join(self.args),
             env=self.env,
             use_gpu=self.use_gpu,
+            artifacts_dir=self.artifacts_dir or "",
             volumes=volumes,
             uploads=uploads,
         )

compute_horde/compute_horde/miner_client/organic.py

@@ -381,6 +381,7 @@ class OrganicJobDetails:
     total_job_timeout: int = 300
     volume: Volume | None = None
     output: OutputUpload | None = None
+    artifacts_dir: str | None = None
 
     def __post_init__(self):
         if (self.docker_image, self.raw_script) == (None, None):
@@ -470,6 +471,7 @@ async def run_organic_job(
                     docker_run_cmd=job_details.docker_run_cmd,
                     volume=job_details.volume,
                     output_upload=job_details.output,
+                    artifacts_dir=job_details.artifacts_dir,
                 )
             )
 
@@ -487,7 +489,11 @@ async def run_organic_job(
                     score=0,  # no score for organic jobs (at least right now)
                 )
 
-                return final_response.docker_process_stdout, final_response.docker_process_stderr
+                return (
+                    final_response.docker_process_stdout,
+                    final_response.docker_process_stderr,
+                    final_response.artifacts,
+                )
             except TimeoutError as exc:
                 raise OrganicJobError(FailureReason.FINAL_RESPONSE_TIMED_OUT) from exc
         except Exception:

compute_horde/compute_horde/mv_protocol/miner_requests.py

@@ -90,6 +90,7 @@ class V0JobFinishedRequest(BaseMinerRequest, JobMixin):
     message_type: RequestType = RequestType.V0JobFinishedRequest
     docker_process_stdout: str  # TODO: add max_length
     docker_process_stderr: str  # TODO: add max_length
+    artifacts: dict[str, str] | None = None
 
 
 class V0MachineSpecsRequest(BaseMinerRequest, JobMixin):

compute_horde/compute_horde/mv_protocol/validator_requests.py

@@ -86,6 +86,7 @@ class V0JobRequest(BaseValidatorRequest, JobMixin):
     docker_run_cmd: list[str]
     volume: Volume | None = None
     output_upload: OutputUpload | None = None
+    artifacts_dir: str | None = None
 
     @model_validator(mode="after")
     def validate_at_least_docker_image_or_raw_script(self) -> Self:

compute_horde/tests/test_run_organic_job.py

@@ -56,7 +56,7 @@ async def test_run_organic_job__success(keypair):
         transport=mock_transport,
     )
     job_details = OrganicJobDetails(job_uuid=JOB_UUID, docker_image="mock")
-    stdout, stderr = await run_organic_job(client, job_details, wait_timeout=2)
+    stdout, stderr, artifacts = await run_organic_job(client, job_details, wait_timeout=2)
 
     assert stdout == "stdout"
     assert stderr == "stderr"

executor/app/src/compute_horde_executor/executor/management/commands/run_executor.py

@@ -3,6 +3,7 @@
 import csv
 import io
 import logging
+import os
 import pathlib
 import random
 import re
@@ -69,6 +70,7 @@
 CVE_2022_0492_TIMEOUT_SECONDS = 120
 CVE_2024_0132_TIMEOUT_SECONDS = 120
 MAX_RESULT_SIZE_IN_RESPONSE = 2000
+MAX_ARTIFACT_SIZE = 16000
 TRUNCATED_RESPONSE_PREFIX_LEN = 1000
 TRUNCATED_RESPONSE_SUFFIX_LEN = 1000
 INPUT_VOLUME_UNPACK_TIMEOUT_SECONDS = 60 * 15
@@ -234,6 +236,7 @@ async def send_finished(self, job_result: "JobResult"):
                 job_uuid=self.job_uuid,
                 docker_process_stdout=job_result.stdout,
                 docker_process_stderr=job_result.stderr,
+                artifacts=job_result.artifacts,
             )
         )
 
@@ -276,6 +279,7 @@ class JobResult(pydantic.BaseModel):
     timeout: bool
     stdout: str
     stderr: str
+    artifacts: dict[str, str]
     specs: MachineSpecs | None = None
 
 
@@ -474,6 +478,7 @@ def __init__(self, initial_job_request: V0InitialJobRequest | V1InitialJobReques
         self.volume_mount_dir = self.temp_dir / "volume"
         self.output_volume_mount_dir = self.temp_dir / "output"
         self.specs_volume_mount_dir = self.temp_dir / "specs"
+        self.artifacts_mount_dir = self.temp_dir / "artifacts"
         self.download_manager = DownloadManager()
 
         self.job_container_name = f"ch-{settings.EXECUTOR_TOKEN}-job"
@@ -507,6 +512,7 @@ async def cleanup_potential_old_jobs(self):
     async def prepare(self):
         self.volume_mount_dir.mkdir(exist_ok=True)
         self.output_volume_mount_dir.mkdir(exist_ok=True)
+        self.artifacts_mount_dir.mkdir(exist_ok=True)
 
         logger.info("preparing in progress")
 
@@ -554,6 +560,7 @@ async def start_job(self, job_request: V0JobRequest) -> JobResult | None:
                 timeout=False,
                 stdout=ex.description,
                 stderr="",
+                artifacts={},
             )
 
         docker_image = job_request.docker_image_name
@@ -582,6 +589,7 @@ async def start_job(self, job_request: V0JobRequest) -> JobResult | None:
                 timeout=False,
                 stdout="",
                 stderr="",
+                artifacts={},
                 exit_status=None,
             )
 
@@ -595,6 +603,12 @@ async def start_job(self, job_request: V0JobRequest) -> JobResult | None:
             )
             await process.wait()
 
+        if job_request.artifacts_dir:
+            extra_volume_flags += [
+                "-v",
+                f"{self.artifacts_mount_dir.as_posix()}/:{job_request.artifacts_dir}",
+            ]
+
         self.cmd = [
             "docker",
             "run",
@@ -648,6 +662,7 @@ async def start_job(self, job_request: V0JobRequest) -> JobResult | None:
                     stdout="",
                     stderr=msg,
                     exit_status=None,
+                    artifacts={},
                 )
 
         return None
@@ -708,6 +723,18 @@ async def wait_for_job(self, job_request: V0JobRequest) -> JobResult:
 
         success = exit_status == 0
 
+        artifacts = {}
+        for artifact_filename in os.listdir(self.artifacts_mount_dir):
+            artifact_path = self.artifacts_mount_dir / artifact_filename
+            if os.path.isfile(artifact_path):
+                with open(artifact_path, "rb") as f:
+                    content = f.read()
+                artifact_size = len(content)
+                if artifact_size < MAX_ARTIFACT_SIZE:
+                    artifacts[artifact_filename] = base64.b64encode(content).decode()
+                else:
+                    logger.error(f"Artefact {artifact_filename} too large: {artifact_size:,} bytes")
+
         if success:
             # upload the output if requested and job succeeded
             if job_request.output_upload:
@@ -740,6 +767,7 @@ async def wait_for_job(self, job_request: V0JobRequest) -> JobResult:
             timeout=timeout,
             stdout=stdout,
             stderr=stderr,
+            artifacts=artifacts,
         )
 
     async def clean(self):

miner/app/src/compute_horde_miner/miner/liveness_check.py

@@ -182,6 +182,7 @@ async def drive_executor() -> float:
             contents=get_dummy_inline_zip_volume(),
         ),
         output_upload=None,
+        artifacts_dir=None,
     )
 
     validator, _ = await Validator.objects.aget_or_create(
@@ -227,6 +228,7 @@ async def drive_executor() -> float:
                 docker_run_cmd=job_request.docker_run_cmd,
                 volume=job_request.volume,
                 output_upload=job_request.output_upload,
+                artifacts_dir=job_request.artifacts_dir,
             ).model_dump(),
         },
     )

miner/app/src/compute_horde_miner/miner/migrations/0012_acceptedjob_artifacts.py

@@ -0,0 +1,19 @@
+# Generated by Django 4.2.16 on 2025-02-07 13:52
+
+from django.db import migrations, models
+
+import compute_horde_miner.miner.models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("miner", "0011_acceptedjob_executor_address"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="acceptedjob",
+            name="artifacts",
+            field=models.JSONField(encoder=compute_horde_miner.miner.models.EnumEncoder, null=True),
+        ),
+    ]

miner/app/src/compute_horde_miner/miner/miner_consumer/executor_interface.py

@@ -131,13 +131,15 @@ async def handle(self, msg: BaseExecutorRequest):
             self.job.status = AcceptedJob.Status.FINISHED
             self.job.stderr = msg.docker_process_stderr
             self.job.stdout = msg.docker_process_stdout
+            self.job.artifacts = msg.artifacts or {}
 
             await self.job.asave()
             await self.send_executor_finished(
                 job_uuid=msg.job_uuid,
                 executor_token=self.executor_token,
                 stdout=msg.docker_process_stdout,
                 stderr=msg.docker_process_stderr,
+                artifacts=msg.artifacts,
             )
         if isinstance(msg, executor_requests.V0MachineSpecsRequest):
             await self.send_executor_specs(
@@ -168,6 +170,7 @@ async def _miner_job_request(self, msg: JobRequest):
                 docker_run_cmd=msg.docker_run_cmd,
                 volume=msg.volume,
                 output_upload=msg.output_upload,
+                artifacts_dir=msg.artifacts_dir,
             ).model_dump_json()
         )
 

miner/app/src/compute_horde_miner/miner/miner_consumer/layer_utils.py

@@ -45,6 +45,7 @@ class JobRequest(pydantic.BaseModel):
     docker_run_cmd: list[str]
     volume: Volume | None = None
     output_upload: OutputUpload | None = None
+    artifacts_dir: str | None = None
 
     @model_validator(mode="after")
     def validate_at_least_docker_image_or_raw_script(self) -> Self:
@@ -62,6 +63,7 @@ class ExecutorFinished(pydantic.BaseModel):
     job_uuid: str
     docker_process_stdout: str
     docker_process_stderr: str
+    artifacts: dict[str, str] | None
 
 
 class ExecutorFailed(pydantic.BaseModel):
@@ -180,6 +182,7 @@ async def send_job_request(self, executor_token, job_request: validator_requests
                     docker_run_cmd=job_request.docker_run_cmd,
                     volume=job_request.volume,
                     output_upload=job_request.output_upload,
+                    artifacts_dir=job_request.artifacts_dir,
                 ).model_dump(),
             },
         )
@@ -252,7 +255,12 @@ async def send_executor_specs(self, job_uuid: str, executor_token: str, specs: M
         )
 
     async def send_executor_finished(
-        self, job_uuid: str, executor_token: str, stdout: str, stderr: str
+        self,
+        job_uuid: str,
+        executor_token: str,
+        stdout: str,
+        stderr: str,
+        artifacts: dict[str, str] | None,
     ):
         group_name = ValidatorInterfaceMixin.group_name(executor_token)
         await self.channel_layer.group_send(
@@ -263,6 +271,7 @@ async def send_executor_finished(
                     job_uuid=job_uuid,
                     docker_process_stdout=stdout,
                     docker_process_stderr=stderr,
+                    artifacts=artifacts,
                 ).model_dump(),
             },
         )

miner/app/src/compute_horde_miner/miner/miner_consumer/validator_interface.py

@@ -228,6 +228,7 @@ async def handle_authentication(self, msg: validator_requests.V0AuthenticateRequ
                         job_uuid=str(job.job_uuid),
                         docker_process_stdout=job.stdout,
                         docker_process_stderr=job.stderr,
+                        artifacts=job.artifacts,
                     ).model_dump_json()
                 )
                 logger.debug(
@@ -568,6 +569,7 @@ async def _executor_finished(self, msg: ExecutorFinished):
                 job_uuid=msg.job_uuid,
                 docker_process_stdout=msg.docker_process_stdout,
                 docker_process_stderr=msg.docker_process_stderr,
+                artifacts=msg.artifacts,
             ).model_dump_json()
         )
         logger.debug(f"Finished job {msg.job_uuid} reported to validator {self.validator_key}")

miner/app/src/compute_horde_miner/miner/models.py

@@ -56,6 +56,7 @@ class Status(models.TextChoices):
     created_at = models.DateTimeField(auto_now_add=True)
     updated_at = models.DateTimeField(auto_now=True)
     executor_address = models.TextField(null=True)
+    artifacts = models.JSONField(encoder=EnumEncoder, null=True)
 
     def __str__(self):
         return (

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

@@ -127,6 +127,7 @@ async def run_streaming_job(options, wait_timeout: int = 300):
                     docker_run_cmd=job_details.docker_run_cmd,
                     volume=job_details.volume,
                     output_upload=job_details.output,
+                    artifacts_dir=job_details.artifacts_dir,
                 )
             )
 

validator/app/src/compute_horde_validator/validator/migrations/0050_organicjob_artifacts.py

@@ -0,0 +1,17 @@
+# Generated by Django 4.2.16 on 2025-02-07 13:53
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("validator", "0049_alter_syntheticjobbatch_block_and_more"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="organicjob",
+            name="artifacts",
+            field=models.JSONField(blank=True, default=dict),
+        ),
+    ]

validator/app/src/compute_horde_validator/validator/models.py

@@ -233,6 +233,7 @@ class SyntheticJob(JobBase):
 class OrganicJob(JobBase):
     stdout = models.TextField(blank=True, default="")
     stderr = models.TextField(blank=True, default="")
+    artifacts = models.JSONField(blank=True, default=dict)
     block = models.BigIntegerField(
         null=True, help_text="Block number on which this job is scheduled"
     )