Fix services ending when replicas exit with 0 (#4482)

A service whose replicas exited with 0 either had them resubmitted by
scaling or ended with ALL_JOBS_DONE, depending on whether all of them
finished in the same tick. With `stop_criteria: master-done`, any
replica reaching DONE could end the whole service, but only if the run
pipeline saw the job after it left TERMINATING.

- Services never end with ALL_JOBS_DONE. Replicas that exit with 0 are
  brought back by scaling, and an all-DONE tick moves the run to
  PENDING, so it is resubmitted with a backoff.
- `master-done` applies within a replica: once the master job is DONE
  or finishing with DONE_BY_RUNNER, the replica counts as DONE. It has
  no effect on single-job service replicas. A worker failing in the
  same tick the master finishes no longer fails the run.

Fixes: #4479
Fixes: #4481
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

Author: un-def

contributing/RUNS-AND-JOBS.md

@@ -21,12 +21,12 @@ A run can spawn one or multiple jobs, depending on the configuration. A task tha
 - 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`.
+    2. Otherwise, if any job is `RUNNING`, the run becomes `RUNNING`.
+    3. Otherwise, if any job is `PROVISIONING` or `PULLING`, the run becomes `PROVISIONING`.
+    4. Otherwise, if jobs are still waiting for placement or provisioning, the run stays `SUBMITTED`.
+    5. Otherwise, if all contributing jobs are `DONE`, the run becomes `TERMINATING` with `ALL_JOBS_DONE`.
+    6. Otherwise, if no active replicas remain and the run should be retried, the run becomes `PENDING`.
+  - `stop_criteria` applies within a replica. With `MASTER_DONE`, once the replica's master job (`job_num == 0`) is `DONE` or `TERMINATING` with `DONE_BY_RUNNER`, the whole replica counts as `DONE`: the other jobs are neither waited for nor retried, and their failures are ignored.
   - 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`.
@@ -47,6 +47,7 @@ Services' run lifecycle has some modifications:
 - 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 2, a service never ends with `ALL_JOBS_DONE`. A replica that exits with 0 is inactive, so scaling resubmits it regardless of `retry`. If all replicas are `DONE` at once, the run becomes `PENDING` instead and is resubmitted as a whole after the backoff.
 - 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.
 

mkdocs/docs/concepts/services.md

@@ -1388,6 +1388,8 @@ on a dedicated domain name by setting up a [gateway](gateways.md).
 
 By default, if `dstack` can't find capacity, or the service exits with an error, or the instance is interrupted, the run will fail.
 
+A service runs until it's stopped. If a replica exits with `0`, `dstack` restarts it regardless of the retry policy.
+
 If you'd like `dstack` to automatically retry, configure the
 [retry](../reference/dstack.yml/service.md#retry) property accordingly:
 <!-- TODO: Add a relevant example -->

src/dstack/_internal/server/background/pipeline_tasks/runs/active.py

@@ -127,7 +127,7 @@ async def process_active_run(context: ActiveContext) -> ActiveResult:
     run_spec = context.run_spec
 
     fleet_id = _detect_fleet_id_from_jobs(run_model)
-    analysis = await _analyze_active_run(run_model)
+    analysis = await _analyze_active_run(run_model, run_spec.merged_profile.stop_criteria)
     transition = _get_active_run_transition(run_spec, run_model, analysis)
 
     run_update_map = _build_run_update_map(run_model, run_spec, transition, fleet_id)
@@ -183,11 +183,14 @@ def _detect_fleet_id_from_jobs(run_model: RunModel) -> Optional[uuid.UUID]:
     return None
 
 
-async def _analyze_active_run(run_model: RunModel) -> _RunAnalysis:
+async def _analyze_active_run(
+    run_model: RunModel, stop_criteria: Optional[StopCriteria]
+) -> _RunAnalysis:
     run_analysis = _RunAnalysis()
     for replica_num, job_models in group_jobs_by_replica_latest(run_model.jobs):
         replica_analysis = await _analyze_active_run_replica(
             run_model=run_model,
+            stop_criteria=stop_criteria,
             replica_num=replica_num,
             job_models=job_models,
         )
@@ -197,9 +200,19 @@ async def _analyze_active_run(run_model: RunModel) -> _RunAnalysis:
 
 async def _analyze_active_run_replica(
     run_model: RunModel,
+    stop_criteria: Optional[StopCriteria],
     replica_num: int,
     job_models: List[JobModel],
 ) -> _ReplicaAnalysis:
+    if stop_criteria == StopCriteria.MASTER_DONE and _master_job_is_done(job_models):
+        # The master decides the replica's outcome. The other jobs are neither waited for
+        # nor retried, and their failures are ignored.
+        return _ReplicaAnalysis(
+            replica_num=replica_num,
+            job_models=job_models,
+            contributed_statuses={RunStatus.DONE},
+        )
+
     contributed_statuses: Set[RunStatus] = set()
     termination_reasons: Set[RunTerminationReason] = set()
     needs_retry = False
@@ -258,6 +271,10 @@ def _apply_replica_analysis(
         analysis.contributed_statuses.update(replica_analysis.contributed_statuses)
 
 
+def _master_job_is_done(job_models: List[JobModel]) -> bool:
+    return any(j.job_num == 0 and _job_is_done_or_finishing_done(j) for j in job_models)
+
+
 def _job_is_done_or_finishing_done(job_model: JobModel) -> bool:
     return job_model.status == JobStatus.DONE or (
         job_model.status == JobStatus.TERMINATING
@@ -361,15 +378,6 @@ def _is_retry_duration_exceeded(job_model: JobModel, current_duration: timedelta
     return current_duration > timedelta(seconds=job_spec.retry.duration)
 
 
-def _should_stop_on_master_done(run_spec: RunSpec, run_model: RunModel) -> bool:
-    if run_spec.merged_profile.stop_criteria != StopCriteria.MASTER_DONE:
-        return False
-    for job_model in run_model.jobs:
-        if job_model.job_num == 0 and job_model.status == JobStatus.DONE:
-            return True
-    return False
-
-
 def _get_active_run_transition(
     run_spec: RunSpec,
     run_model: RunModel,
@@ -394,19 +402,18 @@ def _get_active_run_transition(
             termination_reason=RunTerminationReason.GATEWAY_ERROR,
         )
 
-    if _should_stop_on_master_done(run_spec, run_model):
-        return _ActiveRunTransition(
-            new_status=RunStatus.TERMINATING,
-            termination_reason=RunTerminationReason.ALL_JOBS_DONE,
-        )
-
     if RunStatus.RUNNING in analysis.contributed_statuses:
         return _ActiveRunTransition(new_status=RunStatus.RUNNING)
     if RunStatus.PROVISIONING in analysis.contributed_statuses:
         return _ActiveRunTransition(new_status=RunStatus.PROVISIONING)
     if RunStatus.SUBMITTED in analysis.contributed_statuses:
         return _ActiveRunTransition(new_status=RunStatus.SUBMITTED)
     if RunStatus.DONE in analysis.contributed_statuses and not analysis.replicas_to_retry:
+        if run_spec.configuration.type == "service":
+            # A service runs until stopped. While other replicas keep the run active,
+            # scaling brings back replicas that exit with 0. When none are left,
+            # the run is resubmitted as a whole, with a backoff.
+            return _ActiveRunTransition(new_status=RunStatus.PENDING)
         return _ActiveRunTransition(
             new_status=RunStatus.TERMINATING,
             termination_reason=RunTerminationReason.ALL_JOBS_DONE,

src/tests/_internal/server/background/pipeline_tasks/runs/test_active.py

@@ -1,6 +1,7 @@
 import json
 import uuid
 from datetime import timedelta
+from typing import Optional
 from unittest.mock import AsyncMock, patch
 
 import gpuhunt
@@ -842,15 +843,35 @@ async def test_transitions_to_pending_when_retry_duration_exceeded(
         assert run.termination_reason == RunTerminationReason.RETRY_LIMIT_EXCEEDED
         assert run.lock_token is None
 
+    @pytest.mark.parametrize(
+        ("master_status", "worker_status", "worker_termination_reason"),
+        [
+            pytest.param(JobStatus.DONE, JobStatus.RUNNING, None, id="master-done"),
+            pytest.param(JobStatus.TERMINATING, JobStatus.RUNNING, None, id="master-finishing"),
+            pytest.param(
+                JobStatus.DONE,
+                JobStatus.FAILED,
+                JobTerminationReason.CONTAINER_EXITED_WITH_ERROR,
+                id="worker-failed",
+            ),
+        ],
+    )
     async def test_stops_on_master_done(
-        self, test_db, session: AsyncSession, worker: RunWorker
+        self,
+        test_db,
+        session: AsyncSession,
+        worker: RunWorker,
+        master_status: JobStatus,
+        worker_status: JobStatus,
+        worker_termination_reason: Optional[JobTerminationReason],
     ) -> None:
         project = await create_project(session=session)
         user = await create_user(session=session)
         repo = await create_repo(session=session, project_id=project.id)
         run_spec = get_run_spec(
             repo_id=repo.name,
             profile=Profile(name="default", stop_criteria=StopCriteria.MASTER_DONE),
+            configuration=TaskConfiguration(commands=["echo hello"], nodes=2),
         )
         run = await create_run(
             session=session,
@@ -860,14 +881,20 @@ async def test_stops_on_master_done(
             run_spec=run_spec,
             status=RunStatus.RUNNING,
         )
-        # Master job (job_num=0) is done
         await create_job(
             session=session,
             run=run,
-            status=JobStatus.DONE,
+            status=master_status,
             termination_reason=JobTerminationReason.DONE_BY_RUNNER,
             job_num=0,
         )
+        await create_job(
+            session=session,
+            run=run,
+            status=worker_status,
+            termination_reason=worker_termination_reason,
+            job_num=1,
+        )
         lock_run(run)
         await session.commit()
 
@@ -1111,6 +1138,126 @@ async def test_service_zero_scale_noop(
         assert run.status == RunStatus.PENDING
         assert run.lock_token is None
 
+    # `stop_criteria` is about jobs within a replica, and service replicas have a single job
+    @pytest.mark.parametrize("stop_criteria", [StopCriteria.ALL_DONE, StopCriteria.MASTER_DONE])
+    @pytest.mark.parametrize("done_job_status", [JobStatus.DONE, JobStatus.TERMINATING])
+    async def test_service_resubmits_done_replica(
+        self,
+        test_db,
+        session: AsyncSession,
+        worker: RunWorker,
+        stop_criteria: StopCriteria,
+        done_job_status: JobStatus,
+    ) -> None:
+        project = await create_project(session=session)
+        user = await create_user(session=session)
+        repo = await create_repo(session=session, project_id=project.id)
+        run_spec = get_run_spec(
+            repo_id=repo.name,
+            run_name="service-run",
+            profile=Profile(name="default", stop_criteria=stop_criteria),
+            configuration=ServiceConfiguration(
+                port=8080,
+                commands=["echo Hi!"],
+                replicas=Range[int](min=2, max=2),
+            ),
+        )
+        run = await create_run(
+            session=session,
+            project=project,
+            repo=repo,
+            user=user,
+            run_name="service-run",
+            run_spec=run_spec,
+            status=RunStatus.RUNNING,
+        )
+        await create_job(
+            session=session,
+            run=run,
+            status=done_job_status,
+            termination_reason=JobTerminationReason.DONE_BY_RUNNER,
+            replica_num=0,
+        )
+        await create_job(
+            session=session,
+            run=run,
+            status=JobStatus.RUNNING,
+            replica_num=1,
+        )
+        lock_run(run)
+        await session.commit()
+
+        await worker.process(run_to_pipeline_item(run))
+
+        await session.refresh(run)
+        assert run.status == RunStatus.RUNNING
+        res = await session.execute(
+            select(JobModel)
+            .where(JobModel.run_id == run.id)
+            .order_by(JobModel.replica_num, JobModel.submission_num)
+        )
+        jobs = list(res.scalars().all())
+        assert [(j.replica_num, j.submission_num, j.status) for j in jobs] == [
+            (0, 0, done_job_status),
+            (0, 1, JobStatus.SUBMITTED),
+            (1, 0, JobStatus.RUNNING),
+        ]
+
+    @pytest.mark.parametrize("stop_criteria", [StopCriteria.ALL_DONE, StopCriteria.MASTER_DONE])
+    async def test_service_transitions_to_pending_when_all_replicas_done(
+        self,
+        test_db,
+        session: AsyncSession,
+        worker: RunWorker,
+        stop_criteria: StopCriteria,
+    ) -> None:
+        project = await create_project(session=session)
+        user = await create_user(session=session)
+        repo = await create_repo(session=session, project_id=project.id)
+        run_spec = get_run_spec(
+            repo_id=repo.name,
+            run_name="service-run",
+            profile=Profile(name="default", stop_criteria=stop_criteria),
+            configuration=ServiceConfiguration(
+                port=8080,
+                commands=["echo Hi!"],
+                replicas=Range[int](min=2, max=2),
+            ),
+        )
+        run = await create_run(
+            session=session,
+            project=project,
+            repo=repo,
+            user=user,
+            run_name="service-run",
+            run_spec=run_spec,
+            status=RunStatus.RUNNING,
+        )
+        await create_job(
+            session=session,
+            run=run,
+            status=JobStatus.DONE,
+            termination_reason=JobTerminationReason.DONE_BY_RUNNER,
+            replica_num=0,
+        )
+        await create_job(
+            session=session,
+            run=run,
+            status=JobStatus.TERMINATING,
+            termination_reason=JobTerminationReason.DONE_BY_RUNNER,
+            replica_num=1,
+        )
+        lock_run(run)
+        await session.commit()
+
+        await worker.process(run_to_pipeline_item(run))
+
+        await session.refresh(run)
+        assert run.status == RunStatus.PENDING
+        assert run.termination_reason is None
+        assert run.resubmission_attempt == 1
+        assert run.lock_token is None
+
     async def test_noops_when_run_lock_changes_after_processing(
         self, test_db, session: AsyncSession, worker: RunWorker
     ) -> None: