Task._update_task crashes with "not enough values to unpack" when describe_tasks returns empty tasks

[fspoljar]

Describe the issue:

Task._update_task in dask_cloudprovider/aws/ecs.py unpacks the ECS describe_tasks response into exactly one element with no guard:

[self.task] = (
    await ecs.describe_tasks(cluster=self.cluster_arn, tasks=[self.task_arn])
)["tasks"]

ECS can return an empty tasks list (with the task under failures as MISSING) during the RunTask -> DescribeTasks read-after-write consistency window — the ~1s right after a task is created. When that happens the unpack raises ValueError: not enough values to unpack (expected 1, got 0), which distributed rewraps as RuntimeError: Cluster failed to start, killing FargateCluster startup.

This is the same defect as #33 / #36 ("add task error handling"), which added a guard to the sibling run_task unpack in start() but left _update_task unguarded. Worse, _update_task is called from start() after its 60s retry loop, so the error is terminal rather than retried.

Minimal Complete Verifiable Example:

# The real trigger is a timing race: a FargateCluster whose scheduler task is
# polled in the brief window where ECS DescribeTasks still reports it MISSING.
#   from dask_cloudprovider.aws import FargateCluster
#   cluster = FargateCluster(...)   # intermittently fails on startup
#
# Deterministic reduction — describe_tasks returns this documented shape
# during the consistency window, and _update_task does the unguarded unpack:
response = {"tasks": [], "failures": [{"reason": "MISSING"}]}
[task] = response["tasks"]   # ValueError: not enough values to unpack (expected 1, got 0)

Anything else we need to know?:

Observed against a real run: ECS RunTask for the scheduler succeeded (one task returned, failures: [], status PROVISIONING); the very next DescribeTasks ~1s later returned tasks: [] / failures: [{reason: MISSING}].

Suggested fix — mirror #36 inside _update_task's loop so an empty tasks (transient MISSING) retries instead of crashing:

resp = await ecs.describe_tasks(cluster=self.cluster_arn, tasks=[self.task_arn])
if not resp["tasks"]:
    await asyncio.sleep(wait_duration)
    continue
[self.task] = resp["tasks"]

The bug is present on the latest release and on main.

Environment:

• Dask version: dask-cloudprovider 2025.9.0; dask / distributed 2025.5.1
• Python version: 3.11
• Operating System: Linux (AWS Fargate container)
• Install method (conda, pip, source): pip