code: def test_fault_tolerance_chained_task_fail(
      shutdown_only, exit_type, actor_or_normal_tasks
  ):
      ray.init(_system_config=_SYSTEM_CONFIG)
  
      def sleep_or_fail(pid_actor=None, exit_type=None):
          if exit_type is None:
              time.sleep(999)
          # Wait until the children run
          if pid_actor:
              wait_for_condition(
                  lambda: len(ray.get(pid_actor.get_pids.remote()))
                  == 3,
              )
  
          if exit_type == "exit_kill":
              os._exit(1)
          else:
              raise ValueError("Expected to fail")
  
      # Test a chain of tasks
      @ray.remote(max_retries=0)
      def A(exit_type, pid_actor):
          x = B.remote(pid_actor)
          ray.get(pid_actor.report_pid.remote("A", os.getpid()))
          sleep_or_fail(pid_actor, exit_type)
          ray.get(x)
  
      @ray.remote(max_retries=0)
      def B(pid_actor):
          x = C.remote(pid_actor)
          ray.get(pid_actor.report_pid.remote("B", os.getpid()))
          sleep_or_fail()
          ray.get(x)
  
      @ray.remote(max_retries=0)
      def C(pid_actor):
          ray.get(pid_actor.report_pid.remote("C", os.getpid()))
          sleep_or_fail()
  
      @ray.remote(max_restarts=0, max_task_retries=0)
      class Actor:
          def run(self, pid_actor):
              with pytest.raises(
                  (
                      ray.exceptions.RayTaskError,
                      ray.exceptions.WorkerCrashedError,
                  )
              ):
                  ray.get(
                      A.remote(exit_type=exit_type, pid_actor=pid_actor)
                  )
  
      pid_actor = PidActor.remote()
  
      if actor_or_normal_tasks == "normal_task":
          with pytest.raises(
              (
                  ray.exceptions.RayTaskError,
                  ray.exceptions.WorkerCrashedError,
              )
          ):
              ray.get(
                  A.remote(exit_type=exit_type, pid_actor=pid_actor)
              )
      else:
          a = Actor.remote()
          ray.get(a.run.remote(pid_actor=pid_actor))
  
      wait_for_condition(
          verify_tasks_running_or_terminated,
          task_pids=ray.get(pid_actor.get_pids.remote()),
          expect_num_tasks=3,
      )
  
imports: import asyncio
  from collections import defaultdict
  import os
  from typing import Dict
  import pytest
  import sys
  import time
  from ray._private import ray_constants
  from functools import reduce
  import ray
  from ray._private.state_api_test_utils import PidActor, get_state_api_manager, verify_tasks_running_or_terminated, verify_failed_task
  from ray.util.state.common import ListApiOptions, StateResource
  from ray._private.test_utils import async_wait_for_condition_async_predicate, run_string_as_driver, run_string_as_driver_nonblocking, wait_for_condition
  from ray.util.state import StateApiClient, list_actors, list_jobs, list_tasks
label_desc: Use image node_id 13 for calling a global function with example usage: test_fault_tolerance_chained_task_fail(shutdown_only, exit_type, actor_or_normal_tasks) without return types