Delete dull job and instance models created on no capacity retry

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

@@ -332,6 +332,7 @@ class _UnregisterReplicaResult:
 class _ProcessResult:
     job_update_map: _JobUpdateMap = field(default_factory=_JobUpdateMap)
     instance_update_map: Optional[_InstanceUpdateMap] = None
+    delete_instance: bool = False
     volume_update_rows: list[_VolumeUpdateRow] = field(default_factory=list)
     detached_volume_ids: set[uuid.UUID] = field(default_factory=set)
     unassign_event_message: Optional[str] = None
@@ -476,7 +477,11 @@ async def _apply_process_result(
 ) -> None:
     set_processed_update_map_fields(result.job_update_map)
     set_unlock_update_map_fields(result.job_update_map)
-    if instance_model is not None and result.instance_update_map is None:
+    if (
+        instance_model is not None
+        and result.instance_update_map is None
+        and not result.delete_instance
+    ):
         result.instance_update_map = _InstanceUpdateMap()
     if result.instance_update_map is not None:
         set_processed_update_map_fields(result.instance_update_map)
@@ -514,7 +519,24 @@ async def _apply_process_result(
                 )
             return
 
-        if instance_model is not None and instance_update_map is not None:
+        if instance_model is not None and result.delete_instance:
+            res = await session.execute(
+                delete(InstanceModel)
+                .where(
+                    InstanceModel.id == instance_model.id,
+                    InstanceModel.lock_token == item.lock_token,
+                    InstanceModel.lock_owner == related_instance_lock_owner,
+                )
+                .returning(InstanceModel.id)
+            )
+            deleted_ids = list(res.scalars().all())
+            if len(deleted_ids) == 0:
+                logger.error(
+                    "Failed to delete placeholder instance %s for terminating job %s.",
+                    instance_model.id,
+                    item.id,
+                )
+        elif instance_model is not None and instance_update_map is not None:
             res = await session.execute(
                 update(InstanceModel)
                 .where(
@@ -647,13 +669,11 @@ async def _process_terminating_job(
         return result
 
     if is_placeholder_instance(instance_model):
-        # Placeholder has no VM and no provisioning data. Skip graceful stop,
-        # container stop, and volume detach.
-        instance_update_map = get_or_error(result.instance_update_map)
-        if instance_model.status != InstanceStatus.TERMINATING:
-            instance_update_map["status"] = InstanceStatus.TERMINATING
-            instance_update_map["skip_min_processing_interval"] = True
-        instance_update_map["termination_reason"] = InstanceTerminationReason.JOB_FINISHED
+        # Placeholder has no VM and no provisioning data, so there is nothing to terminate.
+        # Skip graceful stop, container stop, and volume detach. Delete the instance right away
+        # to avoid placeholders accumulating for runs retrying on no capacity.
+        result.instance_update_map = None
+        result.delete_instance = True
         result.job_update_map["instance_id"] = None
         await _unregister_replica_and_update_result(result=result, job_model=job_model)
         result.job_update_map["status"] = _get_job_termination_status(job_model)

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

@@ -25,6 +25,9 @@
     set_processed_update_map_fields,
     set_unlock_update_map_fields,
 )
+from dstack._internal.server.background.pipeline_tasks.runs.common import (
+    delete_superseded_no_capacity_job_submissions,
+)
 from dstack._internal.server.db import get_db, get_session_ctx
 from dstack._internal.server.models import InstanceModel, JobModel, ProjectModel, RunModel
 from dstack._internal.server.services import events
@@ -415,14 +418,17 @@ async def _apply_pending_result(
                 actor=events.SystemActor(),
                 targets=[events.Target.from_model(job_model)],
             )
-
+        await delete_superseded_no_capacity_job_submissions(
+            session=session,
+            run_id=item.id,
+            new_job_models=result.new_job_models,
+        )
         emit_run_status_change_event(
             session=session,
             run_model=context.run_model,
             old_status=context.run_model.status,
             new_status=result.run_update_map.get("status", context.run_model.status),
         )
-
         await _unlock_related_jobs(
             session=session,
             item=item,
@@ -596,6 +602,11 @@ async def _apply_active_result(
                 actor=events.SystemActor(),
                 targets=[events.Target.from_model(job_model)],
             )
+        await delete_superseded_no_capacity_job_submissions(
+            session=session,
+            run_id=item.id,
+            new_job_models=result.new_job_models,
+        )
 
         old_status = run_model.status
         new_status = result.run_update_map.get("status", old_status)

src/dstack/_internal/server/background/pipeline_tasks/runs/active.py

@@ -475,8 +475,8 @@ async def _build_retry_job_models(
                 run_model=context.run_model,
                 job=new_job,
                 status=JobStatus.SUBMITTED,
+                submission_num=old_job_model.submission_num + 1,
             )
-            job_model.submission_num = old_job_model.submission_num + 1
             new_job_models.append(job_model)
     return new_job_models
 

src/dstack/_internal/server/background/pipeline_tasks/runs/common.py

@@ -1,12 +1,16 @@
 import json
+import uuid
 from datetime import datetime
 from typing import Optional
 
+from sqlalchemy import delete
+from sqlalchemy.ext.asyncio import AsyncSession
+
 from dstack._internal.core.models.configurations import (
     DEFAULT_REPLICA_GROUP_NAME,
     ServiceConfiguration,
 )
-from dstack._internal.core.models.runs import JobStatus, RunSpec
+from dstack._internal.core.models.runs import JobStatus, JobTerminationReason, RunSpec
 from dstack._internal.proxy.gateway.schemas.stats import PerWindowStats
 from dstack._internal.server.models import JobModel, RunModel
 from dstack._internal.server.services.jobs import get_job_spec, get_jobs_from_run_spec
@@ -86,8 +90,8 @@ async def build_scale_up_job_models(
                 run_model=run_model,
                 job=new_job,
                 status=JobStatus.SUBMITTED,
+                submission_num=old_job_model.submission_num + 1,
             )
-            job_model.submission_num = old_job_model.submission_num + 1
             new_job_models.append(job_model)
         scheduled_replicas += 1
 
@@ -115,3 +119,32 @@ async def build_scale_up_job_models(
                 new_job_models.append(job_model)
 
     return new_job_models
+
+
+async def delete_superseded_no_capacity_job_submissions(
+    session: AsyncSession,
+    run_id: uuid.UUID,
+    new_job_models: list[JobModel],
+) -> None:
+    """
+    Delete previous job submissions that failed to start due to no capacity without
+    ever provisioning. Such submissions are created each time a run is retrying
+    on no capacity and carry no information beyond the events.
+    The direct predecessor of each new submission is kept so that the run's
+    `status_message` can still show `retrying` while the new submission is in flight.
+    """
+    for job_model in new_job_models:
+        if job_model.submission_num < 2:
+            continue
+        await session.execute(
+            delete(JobModel).where(
+                JobModel.run_id == run_id,
+                JobModel.replica_num == job_model.replica_num,
+                JobModel.job_num == job_model.job_num,
+                JobModel.submission_num < job_model.submission_num - 1,
+                JobModel.status.in_(JobStatus.finished_statuses()),
+                JobModel.termination_reason
+                == JobTerminationReason.FAILED_TO_START_DUE_TO_NO_CAPACITY,
+                JobModel.job_provisioning_data.is_(None),
+            )
+        )

src/dstack/_internal/server/migrations/versions/2026/07_22_1256_ad348ea93493_add_eventtargetmodel_entity_run_id.py

@@ -0,0 +1,51 @@
+"""Add EventTargetModel.entity_run_id
+
+Revision ID: ad348ea93493
+Revises: e9c5e7e26c78
+Create Date: 2026-07-22 12:56:04.527041+00:00
+
+"""
+
+import sqlalchemy as sa
+import sqlalchemy_utils
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "ad348ea93493"
+down_revision = "e9c5e7e26c78"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table("event_targets", schema=None) as batch_op:
+        batch_op.add_column(
+            sa.Column(
+                "entity_run_id", sqlalchemy_utils.types.uuid.UUIDType(binary=False), nullable=True
+            )
+        )
+        batch_op.create_index(
+            batch_op.f("ix_event_targets_entity_run_id"), ["entity_run_id"], unique=False
+        )
+        batch_op.create_foreign_key(
+            batch_op.f("fk_event_targets_entity_run_id_runs"),
+            "runs",
+            ["entity_run_id"],
+            ["id"],
+            ondelete="CASCADE",
+        )
+
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table("event_targets", schema=None) as batch_op:
+        batch_op.drop_constraint(
+            batch_op.f("fk_event_targets_entity_run_id_runs"), type_="foreignkey"
+        )
+        batch_op.drop_index(batch_op.f("ix_event_targets_entity_run_id"))
+        batch_op.drop_column("entity_run_id")
+
+    # ### end Alembic commands ###

src/dstack/_internal/server/migrations/versions/2026/07_22_1331_87d4312605e5_backfill_eventtargetmodel_entity_run_id.py

@@ -0,0 +1,45 @@
+"""Backfill EventTargetModel.entity_run_id
+
+Revision ID: 87d4312605e5
+Revises: ad348ea93493
+Create Date: 2026-07-22 13:31:47.070886+00:00
+
+"""
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "87d4312605e5"
+down_revision = "ad348ea93493"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # Events recorded before entity_run_id was introduced have it unset.
+    # The within_runs events filter relies on entity_run_id, so backfill it.
+    # Job targets are backfilled via the jobs table, so this migration must run
+    # before the job models the events reference are deleted
+    # (e.g. superseded no capacity submissions deleted on resubmission).
+    # Old replicas can still record events without entity_run_id while
+    # this migration is being deployed. Such events won't be backfilled and
+    # won't be returned by the within_runs events filter.
+    op.execute(
+        """
+        UPDATE event_targets SET entity_run_id = entity_id
+        WHERE entity_type = 'run' AND entity_run_id IS NULL
+        """
+    )
+    op.execute(
+        """
+        UPDATE event_targets SET entity_run_id = jobs.run_id
+        FROM jobs
+        WHERE jobs.id = event_targets.entity_id
+            AND event_targets.entity_type = 'job'
+            AND event_targets.entity_run_id IS NULL
+        """
+    )
+
+
+def downgrade() -> None:
+    pass

src/dstack/_internal/server/models.py

@@ -1183,6 +1183,11 @@ class EventTargetModel(BaseModel):
     )
     entity_project: Mapped[Optional["ProjectModel"]] = relationship()
 
+    entity_run_id: Mapped[Optional[uuid.UUID]] = mapped_column(
+        ForeignKey("runs.id", ondelete="CASCADE"), nullable=True, index=True
+    )
+    entity_run: Mapped[Optional["RunModel"]] = relationship()
+
     entity_type: Mapped[EventTargetType] = mapped_column(
         EnumAsString(EventTargetType, 100), index=True
     )

src/dstack/_internal/server/services/events.py

@@ -76,6 +76,7 @@ class Target:
     project_id: Optional[uuid.UUID]
     id: uuid.UUID
     name: str
+    run_id: Optional[uuid.UUID] = None
 
     def __post_init__(self):
         if self.type == EventTargetType.USER and self.project_id is not None:
@@ -84,6 +85,10 @@ def __post_init__(self):
             raise ValueError(f"{self.type} target must have project_id")
         if self.type == EventTargetType.PROJECT and self.id != self.project_id:
             raise ValueError("Project target id must be equal to project_id")
+        if self.type in [EventTargetType.RUN, EventTargetType.JOB] and self.run_id is None:
+            raise ValueError(f"{self.type} target must have run_id")
+        if self.type == EventTargetType.RUN and self.id != self.run_id:
+            raise ValueError("Run target id must be equal to run_id")
 
     @staticmethod
     def from_model(
@@ -126,6 +131,7 @@ def from_model(
                 project_id=model.project_id or model.project.id,
                 id=model.id,
                 name=model.job_name,
+                run_id=model.run_id,
             )
         if isinstance(model, ProjectModel):
             return Target(
@@ -140,6 +146,7 @@ def from_model(
                 project_id=model.project_id or model.project.id,
                 id=model.id,
                 name=model.run_name,
+                run_id=model.id,
             )
         if isinstance(model, SecretModel):
             return Target(
@@ -225,6 +232,7 @@ def emit(session: AsyncSession, message: str, actor: AnyActor, targets: list[Tar
                 entity_project_id=target.project_id,
                 entity_id=target.id,
                 entity_name=target.name,
+                entity_run_id=target.run_id,
             )
         )
     session.add(event)
@@ -354,23 +362,7 @@ async def list_events(
             )
         )
     if within_runs is not None:
-        query = select(JobModel.id).where(JobModel.run_id.in_(within_runs))
-        res = await session.execute(query)
-        # In Postgres, fetching job IDs separately is orders of magnitude faster
-        # than using a subquery.
-        job_ids = list(res.unique().scalars().all())
-        target_filters.append(
-            or_(
-                and_(
-                    EventTargetModel.entity_type == EventTargetType.RUN,
-                    EventTargetModel.entity_id.in_(within_runs),
-                ),
-                and_(
-                    EventTargetModel.entity_type == EventTargetType.JOB,
-                    EventTargetModel.entity_id.in_(job_ids),
-                ),
-            )
-        )
+        target_filters.append(EventTargetModel.entity_run_id.in_(within_runs))
     if include_target_types is not None:
         target_filters.append(EventTargetModel.entity_type.in_(include_target_types))
 

src/dstack/_internal/server/services/runs/__init__.py

@@ -833,6 +833,7 @@ def create_job_model_for_new_submission(
     run_model: RunModel,
     job: Job,
     status: JobStatus,
+    submission_num: int = 0,
 ) -> JobModel:
     """
     Create a new job.
@@ -849,7 +850,7 @@ def create_job_model_for_new_submission(
         job_name=f"{job.job_spec.job_name}",
         replica_num=job.job_spec.replica_num,
         deployment_num=run_model.deployment_num,
-        submission_num=len(job.job_submissions),
+        submission_num=submission_num,
         submitted_at=now,
         last_processed_at=now,
         status=status,