diff --git a/src/dstack/_internal/server/background/pipeline_tasks/service_router_worker_sync.py b/src/dstack/_internal/server/background/pipeline_tasks/service_router_worker_sync.py index ca4201757..f946688a1 100644 --- a/src/dstack/_internal/server/background/pipeline_tasks/service_router_worker_sync.py +++ b/src/dstack/_internal/server/background/pipeline_tasks/service_router_worker_sync.py @@ -244,7 +244,16 @@ async def process(self, item: ServiceRouterWorkerSyncPipelineItem) -> None: selectinload( RunModel.jobs.and_( JobModel.status == JobStatus.RUNNING, - JobModel.ready == True, + # `JobModel.ready` is deliberately not checked. A router only + # passes probes that depend on workers (e.g. `/health_generate`) + # after workers are registered, so waiting for it would deadlock. + # Workers are registered as soon as the sync's own check + # (`/server_info` or `GetServerInfo`) reports them ready, so + # `probes` cannot delay registration. + # Only the first node of a replica is synced. Services don't + # support multi-node replicas yet, so this is a no-op kept for + # forward compatibility. + JobModel.job_num == 0, ) ) .load_only( diff --git a/src/dstack/_internal/server/services/jobs/configurators/base.py b/src/dstack/_internal/server/services/jobs/configurators/base.py index c2da60797..c4c5bfd60 100644 --- a/src/dstack/_internal/server/services/jobs/configurators/base.py +++ b/src/dstack/_internal/server/services/jobs/configurators/base.py @@ -519,11 +519,9 @@ def _probes(self) -> list[ProbeSpec]: return [_openai_model_probe_spec(model.name, model.prefix)] group = self._replica_group() if group is not None and group.router is not None: - # A router only answers chat completions once dstack has registered workers - # with it, and registration skips routers that are not ready yet. Probing - # chat completions here would deadlock: readiness would wait on registration - # while registration waits on readiness. Probe the router's own liveness - # endpoint instead, which does not depend on any worker. + # Probe the router's own liveness endpoint, which does not depend on any + # worker. Workers get no default probe: they may not serve HTTP at all + # (gRPC workers), and they don't receive traffic directly. return [_router_health_probe_spec()] return [] diff --git a/src/dstack/_internal/server/services/runs/router_worker_sync.py b/src/dstack/_internal/server/services/runs/router_worker_sync.py index 918711be1..100597dc0 100644 --- a/src/dstack/_internal/server/services/runs/router_worker_sync.py +++ b/src/dstack/_internal/server/services/runs/router_worker_sync.py @@ -566,7 +566,7 @@ async def sync_router_workers_for_run_model(run_model: RunModel) -> None: router_job = _get_router_job(run_model, router_group) if router_job is None: logger.debug( - "%s: no ready router job in group %s, skipping worker sync", + "%s: no running router job in group %s, skipping worker sync", fmt(run_model), router_group.name, ) diff --git a/src/tests/_internal/server/background/pipeline_tasks/test_service_router_worker_sync.py b/src/tests/_internal/server/background/pipeline_tasks/test_service_router_worker_sync.py index 26633a8ba..98a4c3998 100644 --- a/src/tests/_internal/server/background/pipeline_tasks/test_service_router_worker_sync.py +++ b/src/tests/_internal/server/background/pipeline_tasks/test_service_router_worker_sync.py @@ -460,14 +460,12 @@ async def test_process_calls_sync_and_unlocks_on_success( assert sync_row.lock_owner is None assert sync_row.last_processed_at is not None - async def test_process_skips_sync_when_router_replica_not_ready( + async def test_process_syncs_running_first_node_jobs_regardless_of_probes( self, test_db, session: AsyncSession, worker: ServiceRouterWorkerSyncWorker, - caplog: pytest.LogCaptureFixture, ): - caplog.set_level(level=logging.DEBUG, logger=router_worker_sync.__name__) project = await create_project(session=session) user = await create_user(session=session) repo = await create_repo(session=session, project_id=project.id) @@ -480,8 +478,9 @@ async def test_process_skips_sync_when_router_replica_not_ready( run_spec=_router_service_run_spec(repo.name), ) instance = await create_instance(session=session, project=project) - # The router replica is still starting up, the worker replica is already serving. - await create_job( + # The router's probes may depend on registered workers, so it must be synced before + # they pass. Otherwise, neither the probes nor the sync could make progress. + router_job = await create_job( session=session, run=run, instance=instance, @@ -490,16 +489,36 @@ async def test_process_skips_sync_when_router_replica_not_ready( replica_group_name="router", job_provisioning_data=make_job_provisioning_data(), ) + worker_job = await create_job( + session=session, + run=run, + instance=instance, + status=JobStatus.RUNNING, + ready=False, + replica_num=1, + replica_group_name="worker", + job_provisioning_data=make_job_provisioning_data(), + ) + # Not the first node of the replica await create_job( session=session, run=run, instance=instance, status=JobStatus.RUNNING, - ready=True, + job_num=1, replica_num=1, replica_group_name="worker", job_provisioning_data=make_job_provisioning_data(), ) + # Not running + await create_job( + session=session, + run=run, + instance=instance, + status=JobStatus.PROVISIONING, + replica_num=2, + replica_group_name="worker", + ) sync_row = await _add_service_router_worker_sync_row(session, run.id) sync_row.lock_token = uuid.uuid4() sync_row.lock_expires_at = get_current_datetime() + timedelta(seconds=30) @@ -507,13 +526,17 @@ async def test_process_skips_sync_when_router_replica_not_ready( await session.commit() item = _sync_row_to_pipeline_item(sync_row) - await worker.process(item) + with patch( + "dstack._internal.server.background.pipeline_tasks.service_router_worker_sync" + ".sync_router_workers_for_run_model", + new_callable=AsyncMock, + ) as sync_mock: + await worker.process(item) - assert "no ready router job in group router, skipping worker sync" in caplog.text - # The run stays eligible for the next sync attempt. - await session.refresh(sync_row) - assert sync_row.deleted is False - assert sync_row.lock_token is None + sync_mock.assert_awaited_once() + assert sync_mock.await_args is not None + called_run = sync_mock.await_args.args[0] + assert {j.id for j in called_run.jobs} == {router_job.id, worker_job.id} async def test_process_logs_router_job_when_router_connection_fails( self,