code: def test_new_replica_while_all_workers_in_flight(ws):
      """A task is stuck in 'fetch' state because all workers that hold a replica are in
      flight. While in this state, a new replica appears on a different worker and the
      scheduler informs the waiting worker through a new acquire-replicas or
      compute-task op.
  
      In real life, this will typically happen when the Active Memory Manager replicates a
      key to multiple workers and some workers are much faster than others to acquire it,
      due to unrelated tasks being in flight, so 2 seconds later the AMM reiterates the
      request, passing a larger who_has.
  
      Test that, when this happens, the task is immediately acquired from the new worker,
      without waiting for the original replica holders to get out of flight.
      """
      ws2 = "127.0.0.1:2"
      ws3 = "127.0.0.1:3"
      instructions = ws.handle_stimulus(
          AcquireReplicasEvent(
              who_has={"x": [ws2]},
              nbytes={"x": 1},
              stimulus_id="s1",
          ),
          AcquireReplicasEvent(
              who_has={"y": [ws2]},
              nbytes={"y": 1},
              stimulus_id="s2",
          ),
          AcquireReplicasEvent(
              who_has={"y": [ws2, ws3]},
              nbytes={"y": 1},
              stimulus_id="s3",
          ),
      )
      assert instructions == [
          GatherDep(
              worker=ws2,
              to_gather={"x"},
              total_nbytes=1,
              stimulus_id="s1",
          ),
          GatherDep(
              worker=ws3,
              to_gather={"y"},
              total_nbytes=1,
              stimulus_id="s3",
          ),
      ]
      assert ws.tasks["x"].state == "flight"
      assert ws.tasks["y"].state == "flight"
  
imports: from __future__ import annotations
  import asyncio
  import gc
  import pickle
  import sys
  from collections import defaultdict
  from collections.abc import Iterator
  from time import sleep
  import pytest
  from tlz import first
  from dask.sizeof import sizeof
  import distributed.profile
  from distributed import Nanny, Worker, wait
  from distributed.compatibility import MACOS, WINDOWS
  from distributed.protocol.serialize import Serialize
  from distributed.scheduler import TaskState
  from distributed.utils import recursive_to_dict
  from distributed.utils_test import NO_AMM, _LockedCommPool, assert_story, async_poll_for, freeze_data_fetching, gen_cluster, inc, wait_for_state
  from distributed.worker_state_machine import AcquireReplicasEvent, AddKeysMsg, ComputeTaskEvent, Execute, ExecuteFailureEvent, ExecuteSuccessEvent, FreeKeysEvent, GatherDep, GatherDepBusyEvent, GatherDepFailureEvent, GatherDepNetworkFailureEvent, GatherDepSuccessEvent, Instruction, InvalidTaskState, InvalidTransition, PauseEvent, RecommendationsConflict, RefreshWhoHasEvent, ReleaseWorkerDataMsg, RemoveReplicasEvent, RemoveWorkerEvent, RequestRefreshWhoHasMsg, RescheduleEvent, RescheduleMsg, RetryBusyWorkerEvent, RetryBusyWorkerLater, SecedeEvent, StateMachineEvent, TaskErredMsg, TaskState, TransitionCounterMaxExceeded, UnpauseEvent, UpdateDataEvent, merge_recs_instructions
label_desc: Use image node_id 23 for calling a global function with example usage: test_new_replica_while_all_workers_in_flight(ws) without return types