code: def test_workerstate_fail_to_pickle_flight(ws):
      """Same as test_workerstate_fail_to_pickle_execute_1, but the task was
      computed on another host and for whatever reason it did not fail to pickle when it
      was sent over the network.
  
      See also
      --------
      test_fail_to_pickle_execute_1
      test_workerstate_fail_to_pickle_execute_1
      test_fail_to_pickle_execute_2
      test_fail_to_pickle_spill
  
      See also test_worker_state_machine.py::test_gather_dep_failure, where the task
      instead fails to unpickle when leaving the network stack.
      """
      assert not ws.data
      ws.data = FailStoreDict()
      ws.total_resources = {"R": 1}
      ws.available_resources = {"R": 1}
      ws2 = "127.0.0.1:2"
  
      instructions = ws.handle_stimulus(
          ComputeTaskEvent.dummy(
              "y",
              who_has={"x": [ws2]},
              resource_restrictions={"R": 1},
              stimulus_id="s1",
          ),
          GatherDepSuccessEvent(
              worker=ws2,
              total_nbytes=1,
              data={"x": 123},
              stimulus_id="s2",
          ),
      )
      assert instructions == [
          GatherDep(
              worker=ws2,
              to_gather={"x"},
              total_nbytes=1,
              stimulus_id="s1",
          ),
          TaskErredMsg.match(key="x", stimulus_id="s2"),
      ]
      assert ws.tasks["x"].state == "error"
      assert ws.tasks["y"].state == "waiting"  # Not constrained
  
      # FIXME https://github.com/dask/distributed/issues/6705
      ws.validate = False
  
imports: from __future__ import annotations
  import asyncio
  import glob
  import logging
  import os
  import signal
  from collections import Counter, UserDict
  from time import sleep
  import psutil
  import pytest
  from tlz import merge
  import dask.config
  from dask.utils import format_bytes, parse_bytes
  import distributed.system
  from distributed import Client, Event, KilledWorker, Nanny, Scheduler, Worker, wait
  from distributed.compatibility import MACOS, WINDOWS
  from distributed.core import Status
  from distributed.metrics import monotonic
  from distributed.utils import RateLimiterFilter
  from distributed.utils_test import NO_AMM, async_poll_for, captured_logger, gen_cluster, inc, wait_for_state
  from distributed.worker_memory import parse_memory_limit
  from distributed.worker_state_machine import ComputeTaskEvent, ExecuteSuccessEvent, GatherDep, GatherDepSuccessEvent, TaskErredMsg
label_desc: Use image node_id 5 for calling a global function with example usage: test_workerstate_fail_to_pickle_flight(ws) without return types