code: def test_actor_restart_on_node_failure(ray_start_cluster):
      config = {
          "health_check_failure_threshold": 10,
          "health_check_period_ms": 100,
          "health_check_initial_delay_ms": 0,
          "object_timeout_milliseconds": 1000,
          "task_retry_delay_ms": 100,
      }
      cluster = ray_start_cluster
      # Head node with no resources.
      cluster.add_node(num_cpus=0, _system_config=config)
      cluster.wait_for_nodes()
      ray.init(address=cluster.address)
  
      # Node to place the actor.
      actor_node = cluster.add_node(num_cpus=1)
      cluster.wait_for_nodes()
  
      @ray.remote(num_cpus=1, max_restarts=1, max_task_retries=-1)
      class RestartableActor:
          """An actor that will be reconstructed at most once."""
  
          def __init__(self):
              self.value = 0
  
          def increase(self):
              self.value += 1
              return self.value
  
          def ready(self):
              return
  
      actor = RestartableActor.options(lifetime="detached").remote()
      ray.get(actor.ready.remote())
      results = [actor.increase.remote() for _ in range(100)]
      # Kill actor node, while the above task is still being executed.
      cluster.remove_node(actor_node)
      cluster.add_node(num_cpus=1)
      cluster.wait_for_nodes()
      # Check that none of the tasks failed and the actor is restarted.
      seq = list(range(1, 101))
      results = ray.get(results)
      failed_task_index = None
      # Make sure that all tasks were executed in order before and after the
      # actor's death.
      for i, res in enumerate(results):
          elm = seq.pop(0)
          if res != elm:
              if failed_task_index is None:
                  failed_task_index = i
              assert res + failed_task_index == elm
      # Check that we can still call the actor.
      result = ray.get(actor.increase.remote())
      assert result == 1 or result == results[-1] + 1
  
imports: import atexit
  import asyncio
  import collections
  import numpy
  import os
  import pytest
  import signal
  import sys
  import time
  import ray
  from ray.actor import exit_actor
  import ray.cluster_utils
  from ray._private.test_utils import wait_for_condition, wait_for_pid_to_exit, generate_system_config_map, SignalActor
label_desc: Use image node_id 6 for calling a global function with example usage: test_actor_restart_on_node_failure(ray_start_cluster) and returns: self