From 16dbce5af5d5822c82d6ea9c987ab845ac74a750 Mon Sep 17 00:00:00 2001 From: Dmitry Meyer Date: Fri, 25 Sep 2026 10:38:48 +0000 Subject: [PATCH] Don't require probes to pass in router worker sync The sync only picked up router and worker jobs with `JobModel.ready`, i.e. with all probes passed. A router probe that depends on registered workers, such as `/health_generate`, never passed, since workers are only registered once the router is picked up, so the service deadlocked. The sync now takes all running jobs, filtered by `job_num == 0` instead. Workers are registered once the sync's own check (`/server_info` or `GetServerInfo`) reports them ready, so `probes` can no longer delay worker registration. As probes are configured per service, not per replica group, such gating only worked when the router and workers share the transport and API. The `job_num == 0` condition is a no-op for now, as services don't support multi-node replicas. Previously, it was implied by `ready`, which is only set for the first node. Fixes: https://github.com/dstackai/dstack/issues/4310 Co-Authored-By: Claude Opus 5.5 (1M context) --- .../service_router_worker_sync.py | 11 ++++- .../services/jobs/configurators/base.py | 8 ++-- .../services/runs/router_worker_sync.py | 2 +- .../test_service_router_worker_sync.py | 47 ++++++++++++++----- 4 files changed, 49 insertions(+), 19 deletions(-) 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,