code: def fail(
      self, shuffle_id: ShuffleId, run_id: int, message: str
  ) -> None:
      stale_run_id = self._stale_run_ids.setdefault(shuffle_id, run_id)
      if stale_run_id < run_id:
          self._stale_run_ids[shuffle_id] = run_id
  
      shuffle_run = self._active_runs.get(shuffle_id, None)
      if shuffle_run is None or shuffle_run.run_id != run_id:
          return
      self._active_runs.pop(shuffle_id)
      exception = RuntimeError(message)
      shuffle_run.fail(exception)
  
      self._plugin.worker._ongoing_background_tasks.call_soon(
          self.close, shuffle_run
      )
  
imports: from __future__ import annotations
  import asyncio
  import logging
  from collections.abc import Sequence
  from concurrent.futures import ThreadPoolExecutor
  from typing import TYPE_CHECKING, Any, overload
  from dask.context import thread_state
  from dask.typing import Key
  from dask.utils import parse_bytes
  from distributed.diagnostics.plugin import WorkerPlugin
  from distributed.protocol.serialize import ToPickle
  from distributed.shuffle._core import NDIndex, ShuffleId, ShuffleRun, ShuffleRunSpec, ShuffleSpec
  from distributed.shuffle._exceptions import ShuffleClosedError
  from distributed.shuffle._limiter import ResourceLimiter
  from distributed.utils import log_errors, sync
label_desc: Use image node_id 3 for calling the _ShuffleRunManager obj's underlying member method code with example usage: obj.fail(shuffle_id, run_id, message) without return types