Implement submitted jobs pipeline (#3670)

* Add submitted jobs pipeline fetcher scaffold

* Implement submitted jobs pipeline assignment

* Refine submitted jobs pipeline assignment

* Implement submitted jobs pipeline provisioning

* Clarify submitted jobs pipeline result names

* Wire submitted jobs pipeline

* Tighten submitted jobs pipeline cleanup

* Add FIXME

* Fixes after review

* Fix missing volumes lock

* Lock cluster master fleet in submitted jobs pipeline

* Prefer fleet current master in provisioning lookup

* Add docstrings

* Restore todos

* Drop autocreated fleets broken lock

* Fix submitted jobs placement group cleanup

* Do not lock non-empty fleets

* Fix submitted jobs cluster fleet locking

* Fix tests

* Adjust pipelines params

* Fix tasks selects

* Fix tests

Author: r4victor

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

@@ -6,6 +6,9 @@
 from dstack._internal.server.background.pipeline_tasks.gateways import GatewayPipeline
 from dstack._internal.server.background.pipeline_tasks.instances import InstancePipeline
 from dstack._internal.server.background.pipeline_tasks.jobs_running import JobRunningPipeline
+from dstack._internal.server.background.pipeline_tasks.jobs_submitted import (
+    JobSubmittedPipeline,
+)
 from dstack._internal.server.background.pipeline_tasks.jobs_terminating import (
     JobTerminatingPipeline,
 )
@@ -24,6 +27,7 @@ def __init__(self) -> None:
             ComputeGroupPipeline(),
             FleetPipeline(),
             GatewayPipeline(),
+            JobSubmittedPipeline(),
             JobRunningPipeline(),
             JobTerminatingPipeline(),
             InstancePipeline(),
@@ -60,14 +64,17 @@ def hinter(self):
 class PipelineHinter:
     def __init__(self, pipelines: list[Pipeline]) -> None:
         self._pipelines = pipelines
-        self._hint_fetch_map = {p.hint_fetch_model_name: p for p in self._pipelines}
+        self._hint_fetch_map: dict[str, list[Pipeline]] = {}
+        for pipeline in self._pipelines:
+            self._hint_fetch_map.setdefault(pipeline.hint_fetch_model_name, []).append(pipeline)
 
     def hint_fetch(self, model_name: str):
-        pipeline = self._hint_fetch_map.get(model_name)
-        if pipeline is None:
+        pipelines = self._hint_fetch_map.get(model_name)
+        if pipelines is None:
             logger.warning("Model %s not registered for fetch hints", model_name)
             return
-        pipeline.hint_fetch()
+        for pipeline in pipelines:
+            pipeline.hint_fetch()
 
 
 def start_pipeline_tasks() -> PipelineManager:

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

@@ -114,7 +114,7 @@ class JobRunningPipelineItem(PipelineItem):
 class JobRunningPipeline(Pipeline[JobRunningPipelineItem]):
     def __init__(
         self,
-        workers_num: int = 10,
+        workers_num: int = 20,
         queue_lower_limit_factor: float = 0.5,
         queue_upper_limit_factor: float = 2.0,
         min_processing_interval: timedelta = timedelta(seconds=10),
@@ -462,6 +462,9 @@ async def _refetch_locked_job_model(
 
 
 async def _fetch_run_model(session: AsyncSession, run_id: uuid.UUID) -> RunModel:
+    # FIXME: Selecting all run's jobs on every processing iteration is highly inefficient:
+    # it's quadratic w.r.t. the number jobs within a run.
+    # Avoid selecting other jobs as much as possible.
     latest_submissions_sq = (
         select(
             JobModel.run_id.label("run_id"),