Drop deprecated scheduled tasks (#3749)

* Drop deprecated scheduled tasks

* Clean up dead code

* Update AUTOSCALING.md and RUNS-AND-JOBS.md for pipelines

* Make JobSubmittedPipeline to wait for master election

* Fix tests

Author: r4victor

contributing/AUTOSCALING.md

@@ -4,13 +4,16 @@
 
 - STEP 1: `dstack-gateway` parses nginx `access.log` to collect per-second statistics about requests to the service and request times.
 - STEP 2: `dstack-gateway` aggregates statistics over a 1-minute window.
-- STEP 3: The dstack server pulls all service statistics in the `process_gateways` background task.
-- STEP 4: The `process_runs` background task passes statistics and current replicas to the autoscaler.
-- STEP 5: The autoscaler (configured via the `dstack.yml` file) returns the replica change as an int.
-- STEP 6: `process_runs` calls `scale_run_replicas` to add or remove replicas.
-- STEP 7: `scale_run_replicas` terminates or starts replicas.
-    - `SUBMITTED` and `PROVISIONING` replicas get terminated before `RUNNING`.
-    - Replicas are terminated by descending `replica_num` and launched by ascending `replica_num`.
+- STEP 3: The server keeps gateway connections alive in the scheduled `process_gateways_connections` task and continuously collects stats from active gateways. This is separate from `GatewayPipeline`, which handles gateway provisioning and deletion.
+- STEP 4: When `RunPipeline` processes a service run, it loads the latest collected gateway stats for that service.
+- STEP 5: The autoscaler (configured via `dstack.yml`) computes the desired replica count for each replica group.
+- STEP 6: `RunPipeline` applies that desired state.
+    - For scale-up, it creates new `SUBMITTED` jobs. `JobSubmittedPipeline` then assigns existing capacity or provisions new capacity for them.
+    - For scale-down, it marks the least-important active replicas as `TERMINATING` with `SCALED_DOWN`. `JobTerminatingPipeline` unregisters and cleans them up.
+- STEP 7: If the service is in rolling deployment, `RunPipeline` handles that in the same active-run processing path.
+    - It allows only a limited surge of replacement replicas.
+    - It delays teardown of old replicas until replacement capacity is available.
+    - It also cleans up replicas that belong to replica groups removed from the configuration.
 
 ## RPSAutoscaler
 

contributing/RUNS-AND-JOBS.md

@@ -17,31 +17,38 @@ A run can spawn one or multiple jobs, depending on the configuration. A task tha
 
 ## Run's Lifecycle
 
-- STEP 1: The user submits the run. `services.runs.submit_run` creates jobs with status `SUBMITTED`. Now the run has status `SUBMITTED`.
-- STEP 2: `background.tasks.process_runs` periodically pulls unfinished runs and processes them:
-	- If any job is `RUNNING`, the run becomes `RUNNING`.
-	- If any job is `PROVISIONING` or `PULLING`, the run becomes `PROVISIONING`.
-	- If any job fails and cannot be retried, the run becomes `TERMINATING`, and after processing, `FAILED`.
-	- If all jobs are `DONE`, the run becomes `TERMINATING`, and after processing, `DONE`.
-	- If any job fails, can be retried, and there is any other active job, the failed job will be resubmitted in-place.
-	- If any jobs in a replica fail and can be retried and there is other active replicas, the jobs of the failed replica are resubmitted in-place (without stopping other replicas). But if some jobs in a replica fail, then all the jobs in a replica are terminated and resubmitted. This include multi-node tasks that represent one replica with multiple jobs.
-	- If all jobs fail and can be resubmitted, the run becomes `PENDING`.
-- STEP 3: If the run is `TERMINATING`, the server makes all jobs `TERMINATING`. `background.tasks.process_runs` sets their status to `TERMINATING`, assigns `JobTerminationReason`, and sends a graceful stop command to `dstack-runner`. `process_terminating_jobs` then ensures that jobs are terminated assigns a finished status.
-- STEP 4: Once all jobs are finished, the run becomes `TERMINATED`, `DONE`, or `FAILED` based on `RunTerminationReason`.
-- STEP 0: If the run is `PENDING`, `background.tasks.process_runs` will resubmit jobs. The run becomes `SUBMITTED` again.
-
-> Use `switch_run_status()` for all status transitions. Do not set `RunModel.status` directly.
-
-> No one must assign the finished status to the run, except `services.runs.process_terminating_run`. To terminate the run, assign `TERMINATING` status and `RunTerminationReason`.
+- STEP 1: The user submits the run. `services.runs.submit_run` creates jobs with status `SUBMITTED`. The run starts in `SUBMITTED`.
+- STEP 2: `RunPipeline` continuously processes unfinished runs.
+  - For active runs, it derives the run status from the latest job states in priority order:
+    1. If any non-retryable failure is present, the run becomes `TERMINATING` with the relevant `RunTerminationReason`.
+    2. If `stop_criteria == MASTER_DONE` and the master job is done, the run becomes `TERMINATING` with `ALL_JOBS_DONE`.
+    3. Otherwise, if any job is `RUNNING`, the run becomes `RUNNING`.
+    4. Otherwise, if any job is `PROVISIONING` or `PULLING`, the run becomes `PROVISIONING`.
+    5. Otherwise, if jobs are still waiting for placement or provisioning, the run stays `SUBMITTED`.
+    6. Otherwise, if all contributing jobs are `DONE`, the run becomes `TERMINATING` with `ALL_JOBS_DONE`.
+    7. Otherwise, if no active replicas remain and the run should be retried, the run becomes `PENDING`.
+  - Retryable replica failures are handled before the final transition is applied:
+    - If a replica fails with a retryable reason while other replicas are still active, `RunPipeline` creates a new `SUBMITTED` submission for that replica and terminates the old jobs in that replica.
+    - If all remaining work is retryable, the run ends up in `PENDING`.
+- STEP 3: If the run is `PENDING`, `RunPipeline` processes it in the pending phase.
+  - For retrying runs, it waits for an exponential backoff before resubmitting.
+  - For scheduled runs, it waits until `next_triggered_at`.
+  - For scaled-to-zero services, it can keep the run in `PENDING` until autoscaling wants replicas again.
+  - Once the run is ready to continue, `RunPipeline` creates new `SUBMITTED` jobs and moves the run back to `SUBMITTED`.
+- STEP 4: If the run is `TERMINATING`, `RunPipeline` marks active jobs as `TERMINATING` and assigns the corresponding `JobTerminationReason`.
+- STEP 5: Once all jobs are finished, the terminating phase of `RunPipeline` either:
+  - assigns the final run status (`TERMINATED`, `DONE`, or `FAILED`), or
+  - for scheduled runs that were not stopped or aborted by the user, returns the run to `PENDING` and computes a new `next_triggered_at`.
 
 ### Services
 
-Services' lifecycle has some modifications:
+Services' run lifecycle has some modifications:
 
-- During STEP 1, the service is registered on the gateway. If the gateway is not accessible or the domain name is taken, the run submission fails.
-- During STEP 2, downscaled jobs are ignored.
-- During STEP 4, the service is unregistered on the gateway.
-- During STEP 0, the service can stay in `PENDING` status if it was downscaled to zero (WIP).
+- During STEP 1, the service itself is registered on the gateway or the in-server proxy. If the gateway is not accessible or the domain name is taken, submission fails.
+- During STEP 2, active run processing also computes desired replica counts from gateway stats and handles scale-up, scale-down, rolling deployment, and cleanup of removed replica groups.
+- During STEP 2, jobs already marked `SCALED_DOWN` do not contribute to the run status.
+- During STEP 3, a service can stay in `PENDING` when autoscaling currently wants zero replicas.
+- During STEP 5, the terminating phase of `RunPipeline` unregisters the service from the gateway.
 
 ### When can the job be retried?
 
@@ -54,29 +61,25 @@ Services' lifecycle has some modifications:
 ## Job's Lifecycle
 
 - STEP 1: A newly submitted job has status `SUBMITTED`. It is not assigned to any instance yet.
-- STEP 2: `background.tasks.process_submitted_jobs` tries to assign an existing instance or provision a new one.
-	- On success, the job becomes `PROVISIONING`.
-	- On failure, the job becomes `TERMINATING`, and after processing, `FAILED` because of `FAILED_TO_START_DUE_TO_NO_CAPACITY`.
-- STEP 3: `background.tasks.process_running_jobs` periodically pulls unfinished jobs and processes them.
-	- While `dstack-shim`/`dstack-runner` is not responding, the job stays `PROVISIONING`.
-	- Once `dstack-shim` (for VM-featured backends) becomes available, it submits the docker image name, and the job becomes `PULLING`.
-	- Once `dstack-runner` inside a docker container becomes available, it submits the code and the job spec, and the job becomes `RUNNING`.
-	- If `dstack-shim` or `dstack-runner` don't respond for a long time or fail to respond after successful connection and multiple retries, the job becomes `TERMINATING`, and after processing, `FAILED`.
-- STEP 4: `background.tasks.process_running_jobs` processes `RUNNING` jobs, pulling job logs, runner logs, and job status.
-	- If the pulled status is `DONE`, the job becomes `TERMINATING`, and after processing, `DONE`.
-	- Otherwise, the job becomes `TERMINATING`, and after processing, `FAILED`.
-- STEP 5: `background.tasks.process_terminating_jobs` processes `TERMINATING` jobs.
-	- If the job has `remove_at` in the future, nothing happens. This is to give the job some time for a graceful stop.
-	- Once `remove_at` is in the past, it stops the container via `dstack-shim`, detaches instance volumes, and releases the instance. The job becomes `TERMINATED`, `DONE`, `FAILED`, or `ABORTED` based on `JobTerminationReason`.
-	- If some volumes fail to detach, it keeps the job `TERMINATING` and checks volumes attachment status.
-
-> Use `switch_job_status()` for all status transitions. Do not set `JobModel.status` directly.
-
-> No one must assign the finished status to the job, except `services.jobs.process_terminating_job`. To terminate the job, assign `TERMINATING` status and `JobTerminationReason`.
+- STEP 2: `JobSubmittedPipeline` tries to assign an existing instance or provision new capacity.
+  - On success, the job becomes `PROVISIONING`.
+  - On failure, the job becomes `TERMINATING`. `JobTerminatingPipeline` later assigns the final failed status.
+- STEP 3: `JobRunningPipeline` processes `PROVISIONING`, `PULLING`, and `RUNNING` jobs.
+  - While `dstack-shim` / `dstack-runner` is not responding, the job stays `PROVISIONING`.
+  - Once `dstack-shim` (for VM-featured backends) becomes available, the pipeline submits the image and the job becomes `PULLING`.
+  - Once `dstack-runner` inside the container becomes available, the pipeline uploads the code and job spec, and the job becomes `RUNNING`.
+  - While the job is `RUNNING`, the pipeline keeps collecting logs and runner status.
+  - If startup, runner communication, or replica registration fails, the job becomes `TERMINATING`.
+- STEP 4: Once the job is actually ready, `JobRunningPipeline` initializes probes.
+- STEP 5: `JobTerminatingPipeline` processes `TERMINATING` jobs.
+  - If the job has `remove_at` in the future, it waits. This gives the job time for a graceful stop.
+  - Once `remove_at` is in the past, it stops the container, detaches volumes, unregisters service replicas if needed, and releases the instance assignment.
+  - If some volumes are not detached yet, the job stays `TERMINATING` and is retried.
+  - When cleanup is complete, the job becomes `TERMINATED`, `DONE`, `FAILED`, or `ABORTED` based on `JobTerminationReason`.
 
 ### Services' Jobs
 
 Services' jobs lifecycle has some modifications:
 
-- During STEP 3, once the job becomes `RUNNING`, it is registered on the gateway as a replica. If the gateway is not accessible, the job fails.
-- During STEP 5, the job is unregistered on the gateway (WIP).
+- During STEP 3, once the primary job of a replica is `RUNNING` and ready to receive traffic, `JobRunningPipeline` registers that replica on the gateway. If the gateway is not accessible, the job fails with a gateway-related termination reason.
+- During STEP 5, `JobTerminatingPipeline` unregisters the replica from receiving requests before the job is fully cleaned up.

docs/docs/reference/environment-variables.md

@@ -130,7 +130,6 @@ For more details on the options below, refer to the [server deployment](../guide
 - `DSTACK_SERVER_GCS_BUCKET`{ #DSTACK_SERVER_GCS_BUCKET } - The bucket that repo diffs will be uploaded to if set. If unset, diffs are uploaded to the database.
 - `DSTACK_DB_POOL_SIZE`{ #DSTACK_DB_POOL_SIZE } - The client DB connections pool size. Defaults to `20`,
 - `DSTACK_DB_MAX_OVERFLOW`{ #DSTACK_DB_MAX_OVERFLOW } - The client DB connections pool allowed overflow. Defaults to `20`.
-- `DSTACK_SERVER_BACKGROUND_PROCESSING_FACTOR`{ #DSTACK_SERVER_BACKGROUND_PROCESSING_FACTOR } - The number of background jobs for processing server resources. Increase if you need to process more resources per server replica quickly. Defaults to `1`.
 - `DSTACK_SERVER_BACKGROUND_PROCESSING_DISABLED`{ #DSTACK_SERVER_BACKGROUND_PROCESSING_DISABLED } - Disables background processing if set to any value. Useful to run only web frontend and API server.
 - `DSTACK_SERVER_MAX_PROBES_PER_JOB`{ #DSTACK_SERVER_MAX_PROBES_PER_JOB } - Maximum number of probes allowed in a run configuration. Validated at apply time.
 - `DSTACK_SERVER_MAX_PROBE_TIMEOUT`{ #DSTACK_SERVER_MAX_PROBE_TIMEOUT } - Maximum allowed timeout for a probe. Validated at apply time.

src/dstack/_internal/server/app.py

@@ -171,9 +171,8 @@ async def lifespan(app: FastAPI):
     pipeline_manager = None
     if settings.SERVER_BACKGROUND_PROCESSING_ENABLED:
         scheduler = start_scheduled_tasks()
-        if core_settings.FeatureFlags.PIPELINE_PROCESSING_ENABLED:
-            pipeline_manager = start_pipeline_tasks()
-            app.state.pipeline_manager = pipeline_manager
+        pipeline_manager = start_pipeline_tasks()
+        app.state.pipeline_manager = pipeline_manager
     else:
         logger.info("Background processing is disabled")
     PROBES_SCHEDULER.start()

…ver/background/scheduled_tasks/common.py …rver/background/pipeline_tasks/common.pysrc/dstack/_internal/server/background/scheduled_tasks/common.py renamed to src/dstack/_internal/server/background/pipeline_tasks/common.py src/dstack/_internal/server/background/scheduled_tasks/common.py renamed to src/dstack/_internal/server/background/pipeline_tasks/common.py

@@ -5,8 +5,7 @@
 
 def get_provisioning_timeout(backend_type: BackendType, instance_type_name: str) -> timedelta:
     """
-    This timeout is used in a few places, but roughly refers to the max time between
-    requesting instance creation and the instance becoming ready to accept jobs.
+    This timeout refers to the max time between requesting instance creation and the instance becoming ready to accept jobs.
     For container-based backends, this also includes the image pulling time.
     """
     if backend_type == BackendType.LAMBDA:

src/dstack/_internal/server/background/pipeline_tasks/instances/common.py

@@ -20,7 +20,7 @@
     ItemUpdateMap,
     UpdateMapDateTime,
 )
-from dstack._internal.server.background.scheduled_tasks.common import get_provisioning_timeout
+from dstack._internal.server.background.pipeline_tasks.common import get_provisioning_timeout
 from dstack._internal.server.models import FleetModel, InstanceModel, PlacementGroupModel
 from dstack._internal.server.services.fleets import get_fleet_spec
 from dstack._internal.utils.common import UNSET, Unset, get_current_datetime

src/dstack/_internal/server/background/pipeline_tasks/jobs_running.py

@@ -48,7 +48,7 @@
     set_processed_update_map_fields,
     set_unlock_update_map_fields,
 )
-from dstack._internal.server.background.scheduled_tasks.common import get_provisioning_timeout
+from dstack._internal.server.background.pipeline_tasks.common import get_provisioning_timeout
 from dstack._internal.server.db import get_db, get_session_ctx
 from dstack._internal.server.models import (
     FleetModel,

src/dstack/_internal/server/background/pipeline_tasks/jobs_submitted.py

@@ -319,6 +319,10 @@ async def process(self, item: JobSubmittedPipelineItem):
         if context.job_model.instance_assigned:
             logger.debug("%s: provisioning has started", fmt(context.job_model))
             provisioning = await _process_provisioning(item=item, context=context)
+            _hint_pipelines_fetch(
+                pipeline_hinter=self._pipeline_hinter,
+                result=provisioning,
+            )
             await _apply_provisioning_result(
                 item=item,
                 provisioning=provisioning,
@@ -327,6 +331,10 @@ async def process(self, item: JobSubmittedPipelineItem):
 
         logger.debug("%s: assignment has started", fmt(context.job_model))
         assignment = await _process_assignment(context=context)
+        _hint_pipelines_fetch(
+            pipeline_hinter=self._pipeline_hinter,
+            result=assignment,
+        )
         await _apply_assignment_result(
             item=item,
             context=context,
@@ -365,6 +373,7 @@ class _DeferSubmittedJobResult:
     """The job is not ready yet, so apply should just mark it processed and unlock it."""
 
     log_message: str
+    hint_fleet_pipeline: bool = False
 
 
 @dataclass
@@ -1179,6 +1188,17 @@ async def _process_new_capacity_provisioning(
             job=context.job,
         )
     )
+    if (
+        is_master_job(context.job)
+        and fleet_model is not None
+        and _get_cluster_fleet_spec(fleet_model) is not None
+        and any(not instance.deleted for instance in fleet_model.instances)
+        and master_provisioning_data is None
+    ):
+        return _DeferSubmittedJobResult(
+            log_message="waiting for fleet master instance election",
+            hint_fleet_pipeline=True,
+        )
     provision_new_capacity_result = await _provision_new_capacity(
         project=context.project,
         fleet_model=fleet_model,
@@ -1834,6 +1854,17 @@ def _get_fleet_master_provisioning_data(
     )
 
 
+def _hint_pipelines_fetch(
+    pipeline_hinter: PipelineHinterProtocol,
+    result: Union[_AssignmentResult, _ProvisioningResult],
+) -> None:
+    if not isinstance(result, _DeferSubmittedJobResult):
+        return
+
+    if result.hint_fleet_pipeline:
+        pipeline_hinter.hint_fetch(FleetModel.__name__)
+
+
 def _select_jobs_to_provision(job: Job, replica_jobs: list[Job], job_model: JobModel) -> list[Job]:
     jobs_to_provision = [job]
     if is_multinode_job(job) and is_master_job(job) and job_model.waiting_master_job is not None: