code: def test_controller_recover_and_delete(shutdown_ray_and_serve):
      """Ensure that in-progress deletion can finish even after controller dies."""
  
      ray_context = ray.init()
      serve.start()
      client = _get_global_client()
  
      num_replicas = 10
  
      @serve.deployment(
          num_replicas=num_replicas,
          ray_actor_options={"num_cpus": 0.001},
      )
      def f():
          pass
  
      serve.run(f.bind())
  
      actors = list_actors(
          address=ray_context.address_info["address"],
          filters=[("state", "=", "ALIVE")],
      )
  
      # Try to delete the application and kill the controller right after
      serve.delete(SERVE_DEFAULT_APP_NAME, _blocking=False)
      ray.kill(client._controller, no_restart=False)
  
      # All replicas should be removed already or after the controller revives
      wait_for_condition(
          lambda: len(
              list_actors(
                  address=ray_context.address_info["address"],
                  filters=[("state", "=", "ALIVE")],
              )
          )
          < len(actors)
      )
  
      wait_for_condition(
          lambda: len(
              list_actors(
                  address=ray_context.address_info["address"],
                  filters=[("state", "=", "ALIVE")],
              )
          )
          == len(actors) - num_replicas
      )
  
      # The application should be deleted.
      wait_for_condition(
          lambda: SERVE_DEFAULT_APP_NAME
          not in serve.status().applications
      )
  
imports: import os
  import subprocess
  import sys
  from contextlib import contextmanager
  from tempfile import NamedTemporaryFile
  import pytest
  import requests
  import ray
  import ray._private.state
  import ray.actor
  from ray import serve
  from ray._private.test_utils import wait_for_condition
  from ray.exceptions import RayActorError
  from ray.serve._private.constants import SERVE_DEFAULT_APP_NAME, SERVE_NAMESPACE
  from ray.serve.context import _get_global_client
  from ray.tests.conftest import call_ray_stop_only
  from ray.util.state import list_actors
label_desc: Use image node_id 13 for calling a global function with example usage: test_controller_recover_and_delete(shutdown_ray_and_serve) without return types