Don't require probes to pass in router worker sync

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(

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 []
 

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,
         )

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,30 +489,54 @@ 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)
         sync_row.lock_owner = ServiceRouterWorkerSyncPipeline.__name__
         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,