code: def test_reconstruction(monkeypatch, ray_start_cluster, delay):
      with monkeypatch.context() as m:
          if delay:
              m.setenv(
                  "RAY_testing_asio_delay_us",
                  "CoreWorkerService.grpc_server."
                  "ReportGeneratorItemReturns=10000:1000000",
              )
          cluster = ray_start_cluster
          # Head node with no resources.
          cluster.add_node(
              num_cpus=0,
              _system_config=RECONSTRUCTION_CONFIG,
              enable_object_reconstruction=True,
          )
          ray.init(address=cluster.address)
          # Node to place the initial object.
          node_to_kill = cluster.add_node(
              num_cpus=1, object_store_memory=10**8
          )
          cluster.wait_for_nodes()
  
      @ray.remote(max_retries=2)
      def dynamic_generator(num_returns):
          for i in range(num_returns):
              yield np.ones(1_000_000, dtype=np.int8) * i
  
      @ray.remote
      def fetch(x):
          return x[0]
  
      # Test recovery of all dynamic objects through re-execution.
      gen = dynamic_generator.remote(10)
      refs = []
  
      for i in range(5):
          refs.append(next(gen))
  
      cluster.remove_node(node_to_kill, allow_graceful=False)
      node_to_kill = cluster.add_node(
          num_cpus=1, object_store_memory=10**8
      )
  
      for i, ref in enumerate(refs):
          print("first trial.")
          print("fetching ", i)
          assert ray.get(fetch.remote(ref)) == i
  
      # Try second retry.
      cluster.remove_node(node_to_kill, allow_graceful=False)
      node_to_kill = cluster.add_node(
          num_cpus=1, object_store_memory=10**8
      )
  
      for i in range(4):
          refs.append(next(gen))
  
      for i, ref in enumerate(refs):
          print("second trial")
          print("fetching ", i)
          assert ray.get(fetch.remote(ref)) == i
  
      # third retry should fail.
      cluster.remove_node(node_to_kill, allow_graceful=False)
      node_to_kill = cluster.add_node(
          num_cpus=1, object_store_memory=10**8
      )
  
      for i in range(1):
          refs.append(next(gen))
  
      for i, ref in enumerate(refs):
          print("third trial")
          print("fetching ", i)
          with pytest.raises(ray.exceptions.RayTaskError) as e:
              ray.get(fetch.remote(ref))
          assert (
              "the maximum number of task retries has been exceeded"
              in str(e.value)
          )
  
file_path: ray/python/ray/tests/test_streaming_generator_2.py
imports: import asyncio
  import pytest
  import numpy
  import sys
  import time
  import gc
  import ray
  from ray.experimental.state.api import list_actors
label_desc: Use image node_id 2 for calling a global function with example usage: test_reconstruction(monkeypatch, ray_start_cluster, delay) and returns: x