code: def test_long_poll_restarts(serve_instance):
      @ray.remote(
          max_restarts=-1,
          max_task_retries=-1,
      )
      class RestartableLongPollHost:
          def __init__(self) -> None:
              print("actor started")
              self.host = LongPollHost()
              self.host.notify_changed("timer", time.time())
              self.should_exit = False
  
          async def listen_for_change(self, key_to_ids):
              print("listening for change ", key_to_ids)
              return await self.host.listen_for_change(key_to_ids)
  
          async def set_exit(self):
              self.should_exit = True
  
          async def exit_if_set(self):
              if self.should_exit:
                  print("actor exit")
                  os._exit(1)
  
      host = RestartableLongPollHost.remote()
      updated_values = ray.get(
          host.listen_for_change.remote({"timer": -1})
      )
      timer: UpdatedObject = updated_values["timer"]
  
      on_going_ref = host.listen_for_change.remote(
          {"timer": timer.snapshot_id}
      )
      ray.get(host.set_exit.remote())
      # This task should trigger the actor to exit.
      # But the retried task will not because self.should_exit is false.
      host.exit_if_set.remote()
  
      # on_going_ref should return succesfully with a differnt value.
      new_timer: UpdatedObject = ray.get(on_going_ref)["timer"]
      assert new_timer.snapshot_id != timer.snapshot_id + 1
      assert new_timer.object_snapshot != timer.object_snapshot
  
file_path: ray/python/ray/serve/tests/test_long_poll.py
imports: import asyncio
  import os
  import sys
  import time
  from typing import Dict
  import pytest
  import ray
  from ray._private.test_utils import async_wait_for_condition
  from ray._private.utils import get_or_create_event_loop
  from ray.serve._private.common import EndpointInfo, EndpointTag, RunningReplicaInfo
  from ray.serve._private.long_poll import LongPollClient, LongPollHost, LongPollNamespace, LongPollState, UpdatedObject
  from ray.serve.generated.serve_pb2 import ActorNameList, EndpointSet, LongPollRequest, LongPollResult
label_desc: Use image node_id 4 for calling a global function with example usage: test_long_poll_restarts(serve_instance) and returns: unknown