Rolling deployments for files (#2862)

Support rolling deployments for services when the
`files` property is updated or the contents of
file archives change.

Currently, `dstack` always detects phantom changes
to file archive contents, since file archives are
omitted when getting the run plan. Addressing this
is out of scope for this PR.

This PR does not address compatibility with
previously submitted jobs that are using files,
because the files feature has not been released
yet, so no users have such jobs.

Author: jvstme

runner/internal/executor/files.go

@@ -51,7 +51,7 @@ func (ex *RunExecutor) setupFiles(ctx context.Context) error {
 		}
 	}
 
-	for _, fa := range ex.run.RunSpec.FileArchives {
+	for _, fa := range ex.jobSpec.FileArchives {
 		log.Trace(ctx, "Extracting file archive", "id", fa.Id, "path", fa.Path)
 
 		p := path.Clean(fa.Path)

runner/internal/schemas/schemas.go

@@ -47,7 +47,6 @@ type RunSpec struct {
 	RunName           string        `json:"run_name"`
 	RepoId            string        `json:"repo_id"`
 	RepoData          RepoData      `json:"repo_data"`
-	FileArchives      []FileArchive `json:"file_archives"`
 	Configuration     Configuration `json:"configuration"`
 	ConfigurationPath string        `json:"configuration_path"`
 }
@@ -71,7 +70,8 @@ type JobSpec struct {
 	// `RepoData` is optional for compatibility with jobs submitted before 0.19.17.
 	// Use `RunExecutor.getRepoData()` to get non-nil `RepoData`.
 	// TODO: make required when supporting jobs submitted before 0.19.17 is no longer relevant.
-	RepoData *RepoData `json:"repo_data"`
+	RepoData     *RepoData     `json:"repo_data"`
+	FileArchives []FileArchive `json:"file_archives"`
 }
 
 type ClusterInfo struct {

src/dstack/_internal/core/compatibility/runs.py

@@ -138,6 +138,8 @@ def get_job_spec_excludes(job_specs: list[JobSpec]) -> Optional[dict]:
         spec_excludes["repo_code_hash"] = True
     if all(s.repo_data is None for s in job_specs):
         spec_excludes["repo_data"] = True
+    if all(not s.file_archives for s in job_specs):
+        spec_excludes["file_archives"] = True
 
     if spec_excludes:
         return spec_excludes

src/dstack/_internal/core/models/runs.py

@@ -226,6 +226,7 @@ class JobSpec(CoreModel):
     # submitted before 0.19.17. See `_get_repo_code_hash` on how to get the correct `repo_code_hash`
     # TODO: drop this comment when supporting jobs submitted before 0.19.17 is no longer relevant.
     repo_code_hash: Optional[str] = None
+    file_archives: list[FileArchiveMapping] = []
 
 
 class JobProvisioningData(CoreModel):

src/dstack/_internal/server/background/tasks/process_running_jobs.py

@@ -244,7 +244,7 @@ async def _process_running_job(session: AsyncSession, job_model: JobModel):
                 # the runner is not ready yet
                 file_archives = await _get_job_file_archives(
                     session=session,
-                    archive_mappings=run.run_spec.file_archives,
+                    archive_mappings=job.job_spec.file_archives,
                     user=run_model.user,
                 )
                 code = await _get_job_code(
@@ -296,7 +296,7 @@ async def _process_running_job(session: AsyncSession, job_model: JobModel):
             # the runner is not ready yet
             file_archives = await _get_job_file_archives(
                 session=session,
-                archive_mappings=run.run_spec.file_archives,
+                archive_mappings=job.job_spec.file_archives,
                 user=run_model.user,
             )
             code = await _get_job_code(

src/dstack/_internal/server/schemas/runner.py

@@ -57,7 +57,6 @@ class SubmitBody(CoreModel):
                     "repo_data",
                     "configuration",
                     "configuration_path",
-                    "file_archives",
                 },
             }
         ),
@@ -79,6 +78,7 @@ class SubmitBody(CoreModel):
                 "ssh_key",
                 "working_dir",
                 "repo_data",
+                "file_archives",
             }
         ),
     ]

src/dstack/_internal/server/services/jobs/configurators/base.py

@@ -151,6 +151,7 @@ async def _get_job_spec(
             ssh_key=self._ssh_key(jobs_per_replica),
             repo_data=self.run_spec.repo_data,
             repo_code_hash=self.run_spec.repo_code_hash,
+            file_archives=self.run_spec.file_archives,
         )
         return job_spec
 

src/dstack/_internal/server/services/runs.py

@@ -918,6 +918,7 @@ def _validate_run_spec_and_set_defaults(run_spec: RunSpec):
         # rolling deployment
         "repo_data",
         "repo_code_hash",
+        "file_archives",
     ],
 }
 _CONF_UPDATABLE_FIELDS = ["priority"]
@@ -930,6 +931,7 @@ def _validate_run_spec_and_set_defaults(run_spec: RunSpec):
         # rolling deployment
         "resources",
         "volumes",
+        "files",
         "image",
         "user",
         "privileged",

src/tests/_internal/server/routers/test_runs.py

@@ -246,6 +246,7 @@ def get_dev_env_run_plan_dict(
                     "working_dir": ".",
                     "repo_code_hash": None,
                     "repo_data": {"repo_dir": "/repo", "repo_type": "local"},
+                    "file_archives": [],
                 },
                 "offers": [json.loads(o.json()) for o in offers],
                 "total_offers": total_offers,
@@ -440,6 +441,7 @@ def get_dev_env_run_dict(
                     "working_dir": ".",
                     "repo_code_hash": None,
                     "repo_data": {"repo_dir": "/repo", "repo_type": "local"},
+                    "file_archives": [],
                 },
                 "job_submissions": [
                     {