[Bug]: Multi-node task gets stuck in provisioning if job provisioning fails

[un-def]

Steps to reproduce

To reproduce, we need a run to be assigned to a fleet but fail to provision one of its jobs due to no capacity (that is, no job_provisioning_data, status SUBMITTED -> TERMINATING).

The most reliable way to reproduce is to use blocks (due to loose checks when blocks are enabled, see _run_can_fit_into_fleet). The other possible case is when there is no enough cloud provider capacity for the second instance while the first one is already provisioned.

1. Prepare a fleet with 2 instances divided into blocks and make sure that one instance is busy (at least one block or the whole instance -- does not matter) while the other is idle:

    NAME           NODES  GPU  SPOT       BACKEND             PRICE    STATUS    CREATED
    test-fleet     2           on-demand  *                            active    8 mins ago
       instance=0         -    on-demand  nebius (eu-north1)  $0.0992  4/4 busy  8 mins ago
       instance=1         -    on-demand  nebius (eu-north1)  $0.0992  0/4 busy  8 mins ago
   

2. Submit a run with nodes: 2. Normally, such a run should fail fast (it does not fit into the fleet; no offers in the run plan), but with blocks this check is effectively bypassed; one offer is displayed in the run plan).

3. Wait.

Actual behaviour

Job statuses change as follows:

job=0	job=1	note
submitted	submitted
provisioning	submitted
provisioning	terminating	exception in the server logs
provisioning	no capacity	gets stuck here infinitely

The run gets stuck in "provisioning". dstack stop <run-name> won't help. The exception in the logs is repeated over and over again.

Expected behaviour

No response

dstack version

0.21.0

Server logs

dstack._internal.server.background.pipeline_tasks.base:357 Processing jobs item f49daabe-6429-4414-87e7-e7ad337d2edb
dstack._internal.server.background.pipeline_tasks.base:361 Unexpected exception when processing item
Traceback (most recent call last):
  File "/home/def/dev/dstack/src/dstack/_internal/server/background/pipeline_tasks/base.py", line 359, in start
    await self.process(item)
  File "/home/def/dev/dstack/src/dstack/_internal/server/utils/tracing.py", line 26, in wrapper
    result = await f(*args, **kwargs)
  File "/home/def/dev/dstack/src/dstack/_internal/server/background/pipeline_tasks/jobs_running.py", line 324, in process
    result = await _process_running_job(context=context)
  File "/home/def/dev/dstack/src/dstack/_internal/server/background/pipeline_tasks/jobs_running.py", line 466, in
_process_running_job
    startup_context = await _prepare_startup_context(context=context, result=result)
  File "/home/def/dev/dstack/src/dstack/_internal/server/background/pipeline_tasks/jobs_running.py", line 573, in
_prepare_startup_context
    cluster_info = _get_cluster_info(
  File "/home/def/dev/dstack/src/dstack/_internal/server/background/pipeline_tasks/jobs_running.py", line 1740, in
_get_cluster_info
    get_or_error(job.job_submissions[-1].job_provisioning_data).internal_ip or ""
  File "/home/def/dev/dstack/src/dstack/_internal/utils/common.py", line 315, in get_or_error
    raise ValueError("Optional value is None")
ValueError: Optional value is None

Additional information

No response