code: def test_recover_deleting_application(serve_instance):
      """Test that replicas that are stuck on __del__ when the controller crashes,
      is properly recovered when the controller is recovered.
  
      This is similar to the test test_replica_deletion_after_controller_recover,
      except what's blocking the deployment is __del__ instead of ongoing requests
      """
  
      signal = SignalActor.remote()
  
      @serve.deployment
      class A:
          async def __del__(self):
              await signal.wait.remote()
  
      id = DeploymentID("A", SERVE_DEFAULT_APP_NAME)
      serve.run(A.bind())
  
      @ray.remote
      def delete_task():
          serve.delete(SERVE_DEFAULT_APP_NAME)
  
      # Delete application and make sure it is stuck on deleting
      delete_ref = delete_task.remote()
      print("Started task to delete application `default`")
  
      def application_deleting():
          # Confirm application is in deleting state
          app_status = serve.status().applications[
              SERVE_DEFAULT_APP_NAME
          ]
          assert app_status.status == "DELETING"
  
          # Confirm deployment is in updating state
          status = serve_instance.get_all_deployment_statuses()[0]
          assert status.name == "A" and status.status == "UPDATING"
  
          # Confirm replica is stopping
          replicas = ray.get(
              serve_instance._controller._dump_replica_states_for_testing.remote(
                  id
              )
          )
          assert replicas.count(states=[ReplicaState.STOPPING]) == 1
  
          # Confirm delete task is still blocked
          finished, pending = ray.wait([delete_ref], timeout=0)
          assert pending and not finished
          return True
  
      def check_deleted():
          deployment_statuses = (
              serve_instance.get_all_deployment_statuses()
          )
          if len(deployment_statuses) != 0:
              return False
  
          finished, pending = ray.wait([delete_ref], timeout=0)
          return finished and not pending
  
      wait_for_condition(application_deleting)
      for _ in range(10):
          time.sleep(0.1)
          assert application_deleting()
  
      print(
          "Confirmed that application `default` is stuck on deleting."
      )
  
      # Kill controller while the application is stuck on deleting
      ray.kill(
          serve.context._global_client._controller, no_restart=False
      )
      print("Finished killing the controller (with restart).")
  
      def check_controller_alive():
          all_current_actors = list_actors(
              filters=[("state", "=", "ALIVE")]
          )
          for actor in all_current_actors:
              if actor["class_name"] == "ServeController":
                  return True
          return False
  
      wait_for_condition(check_controller_alive)
      print("Controller is back alive.")
  
      wait_for_condition(application_deleting)
      # Before we send the signal, the application should still be deleting
      for _ in range(10):
          time.sleep(0.1)
          assert application_deleting()
  
      print("Confirmed that application is still stuck on deleting.")
  
      # Since we've confirmed the replica is in a stopping state, we can grab
      # the reference to the in-progress graceful shutdown task
      replicas = ray.get(
          serve_instance._controller._dump_replica_states_for_testing.remote(
              id
          )
      )
      graceful_shutdown_ref = replicas.get()[
          0
      ]._actor._graceful_shutdown_ref
  
      signal.send.remote()
      print("Sent signal to unblock deletion of application")
      wait_for_condition(check_deleted)
      print(
          "Confirmed that application finished deleting and delete task has returned."
      )
  
      # Make sure graceful shutdown ran successfully
      ray.get(graceful_shutdown_ref)
      print("Confirmed that graceful shutdown ran successfully.")
  
imports: import asyncio
  import logging
  import os
  import re
  import sys
  import time
  from collections import defaultdict
  import pytest
  import requests
  import ray
  from ray import serve
  from ray._private.test_utils import SignalActor, wait_for_condition
  from ray.exceptions import RayTaskError
  from ray.serve._private.common import DeploymentID, ReplicaState
  from ray.serve._private.constants import SERVE_CONTROLLER_NAME, SERVE_DEFAULT_APP_NAME, SERVE_NAMESPACE, SERVE_PROXY_NAME
  from ray.serve._private.utils import get_random_string
  from ray.serve.schema import LoggingConfig
  from ray.serve.tests.test_failure import request_with_retries
  from ray.util.state import list_actors
label_desc: Use image node_id 5 for calling a global function with example usage: test_recover_deleting_application(serve_instance) and returns: True, unknown, False, False, True