Implement instance pipeline (#3636)

* Move delete_instance_health_checks to a separate module

* Move utils/provisioning.py to ssh_fleets/provisioning.py

* Use SSHProvisioningError for ssh instances errors

* Fix _add_remote() nested try-excepts

* Refactor _resolve_ssh_instance_network

* Refactor _process_instance() into thin dispatcher

* Refactor instance check code

* Add fetchers tests

* Add TestInstanceWorker

* Run pyright for pipeline tests

* WIP: InstanceWorker

* Fix volumes pipeline processing active

* Refactor log_lock_token

* Build instance events from update map

* Rename

* Refactor instance pipeline into modules

* Inline _get_effective_ helpers

* Process new instance immediately

* Do not refetch status

* Fix sibling_update_rows

* Drop redundant synchronize_session=False

* Add ProcessContext

* Simplify placement groups code

* Drop _PlacementGroupState

* Restore comments

* Fix result.sibling_update_rows append

* Fix unset typing

* Add migration

* Lock instances in fleet pipeline

* Optimize instance lock in fleet pipeline

* Respect instance lock in delete_fleets

* Skip locked instances in process_next_terminating_job

* Respect instance lock in submitted_jobs

* Add ix_instances_pipeline_fetch_q_index

* Wire instance pipeline

* Set current_master_instance

* Refactor current_master_instance

* Terminate instances with MASTER_FAILED if the master dies with NO_OFFERS

* Fix instance unlock in fleet pipeline

* Remove extra fleet_model_to_fleet

* Wire pipeline_hinter

* Remove extra fleet_model_to_fleet

* Fix index name

* Rebase migrations

* Fix redundant fleet.instances loads in instance pipeline

* Do not lock all instances in delete_fleets

* Add FIXME

* Fix tests

* Retry instance lock in delete_fleets

* Retry lock in all delete endpoints

* Fix created_at and last_processed_at init values

Author: r4victor

AGENTS.md

@@ -18,6 +18,7 @@
 - Python targets 3.9+ with 4-space indentation and max line length of 99 (see `ruff.toml`; `E501` is ignored but keep lines readable).
 - Imports are sorted via Ruff’s isort settings (`dstack` treated as first-party).
 - Keep primary/public functions before local helper functions in a module section.
+- Keep private classes, exceptions, and similar implementation-specific types close to the private functions that use them unless they are shared more broadly in the module.
 - Prefer pydantic-style models in `core/models`.
 - Tests use `test_*.py` modules and `test_*` functions; fixtures live near usage.
 

pyproject.toml

@@ -106,6 +106,7 @@ include = [
     "src/dstack/_internal/core/backends/runpod",
     "src/dstack/_internal/cli/services/configurators",
     "src/dstack/_internal/cli/commands",
+    "src/tests/_internal/server/background/pipeline_tasks",
 ]
 ignore = [
     "src/dstack/_internal/server/migrations/versions",

src/dstack/_internal/core/errors.py

@@ -136,6 +136,10 @@ class ConfigurationError(DstackError):
     pass
 
 
+class SSHProvisioningError(DstackError):
+    pass
+
+
 class SSHError(DstackError):
     pass
 

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

@@ -4,6 +4,7 @@
 from dstack._internal.server.background.pipeline_tasks.compute_groups import ComputeGroupPipeline
 from dstack._internal.server.background.pipeline_tasks.fleets import FleetPipeline
 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.placement_groups import (
     PlacementGroupPipeline,
 )
@@ -19,6 +20,7 @@ def __init__(self) -> None:
             ComputeGroupPipeline(),
             FleetPipeline(),
             GatewayPipeline(),
+            InstancePipeline(),
             PlacementGroupPipeline(),
             VolumePipeline(),
         ]

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

@@ -1,6 +1,8 @@
 import asyncio
+import logging
 import math
 import random
+import time
 import uuid
 from abc import ABC, abstractmethod
 from collections.abc import Iterable, Sequence
@@ -331,14 +333,20 @@ async def start(self):
         self._running = True
         while self._running:
             item = await self._queue.get()
+            start_time = time.time()
             logger.debug("Processing %s item %s", item.__tablename__, item.id)
             try:
                 await self.process(item)
             except Exception:
                 logger.exception("Unexpected exception when processing item")
             finally:
                 await self._heartbeater.untrack(item)
-            logger.debug("Processed %s item %s", item.__tablename__, item.id)
+            logger.debug(
+                "Processed %s item %s in %.3f",
+                item.__tablename__,
+                item.id,
+                time.time() - start_time,
+            )
 
     def stop(self):
         self._running = False
@@ -416,3 +424,40 @@ def resolve_now_placeholders(update_values: _ResolveNowInput, now: datetime):
     for key, value in update_values.items():
         if value is NOW_PLACEHOLDER:
             update_values[key] = now
+
+
+def log_lock_token_mismatch(
+    logger: logging.Logger,
+    item: PipelineItem,
+    action: str = "process",
+) -> None:
+    logger.warning(
+        "Failed to %s %s item %s: lock_token mismatch."
+        " The item is expected to be processed and updated on another fetch iteration.",
+        action,
+        item.__tablename__,
+        item.id,
+    )
+
+
+def log_lock_token_changed_after_processing(
+    logger: logging.Logger,
+    item: PipelineItem,
+    action: str = "update",
+    expected_outcome: str = "updated",
+) -> None:
+    logger.warning(
+        "Failed to %s %s item %s after processing: lock_token changed."
+        " The item is expected to be processed and %s on another fetch iteration.",
+        action,
+        item.__tablename__,
+        item.id,
+        expected_outcome,
+    )
+
+
+def log_lock_token_changed_on_reset(logger: logging.Logger) -> None:
+    logger.warning(
+        "Failed to reset lock: lock_token changed."
+        " The item is expected to be processed and updated on another fetch iteration."
+    )

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

@@ -20,6 +20,8 @@
     PipelineItem,
     UpdateMapDateTime,
     Worker,
+    log_lock_token_changed_after_processing,
+    log_lock_token_mismatch,
     resolve_now_placeholders,
     set_processed_update_map_fields,
     set_unlock_update_map_fields,
@@ -194,12 +196,7 @@ async def process(self, item: PipelineItem):
             )
             compute_group_model = res.unique().scalar_one_or_none()
             if compute_group_model is None:
-                logger.warning(
-                    "Failed to process %s item %s: lock_token mismatch."
-                    " The item is expected to be processed and updated on another fetch iteration.",
-                    item.__tablename__,
-                    item.id,
-                )
+                log_lock_token_mismatch(logger, item)
                 return
 
         result = _TerminateResult()
@@ -228,12 +225,7 @@ async def process(self, item: PipelineItem):
             )
             updated_ids = list(res.scalars().all())
             if len(updated_ids) == 0:
-                logger.warning(
-                    "Failed to update %s item %s after processing: lock_token changed."
-                    " The item is expected to be processed and updated on another fetch iteration.",
-                    item.__tablename__,
-                    item.id,
-                )
+                log_lock_token_changed_after_processing(logger, item)
                 return
             if not result.instances_update_map:
                 return
@@ -249,6 +241,8 @@ async def process(self, item: PipelineItem):
                     instance_model=instance_model,
                     old_status=instance_model.status,
                     new_status=InstanceStatus.TERMINATED,
+                    termination_reason=instance_model.termination_reason,
+                    termination_reason_message=instance_model.termination_reason_message,
                 )