code: def test_cluster_crash_before_checkpoint(
      workflow_start_regular_shared_serve,
  ):
      """If the cluster crashed before the event was checkpointed, after the cluster restarted
      and the workflow resumed, the new event message is processed by the workflow.
      """
  
      class CustomHTTPListener(HTTPListener):
          async def poll_for_event(self, event_key):
              workflow_id = workflow_context.get_current_workflow_id()
              if event_key is None:
                  raise WorkflowEventHandleError(
                      workflow_id, "poll_for_event() needs event_key"
                  )
              payload = await self.handle.get_event_payload.remote(
                  workflow_id, event_key
              )
  
              if utils.check_global_mark("after_cluster_restarted"):
                  return payload
              else:
                  utils.set_global_mark("simulate_cluster_crash")
                  await asyncio.sleep(10000)
  
      from ray._private import storage
      from ray.workflow.tests.utils import skip_client_mode_test
  
      storage_uri = storage._storage_uri
  
      # This test restarts the cluster, so we cannot test under client mode.
      skip_client_mode_test()
  
      def send_event(msg):
          try:
              resp = requests.post(
                  "http://127.0.0.1:8000/event/send_event/"
                  + "workflow_test_cluster_crash_before_checkpoint",
                  json={"event_key": "event_key", "event_payload": msg},
                  timeout=5,
              )
              return resp
          except requests.Timeout:
              return 500
  
      event_promise = workflow.wait_for_event(
          CustomHTTPListener, event_key="event_key"
      )
      workflow.run_async(
          event_promise,
          workflow_id="workflow_test_cluster_crash_before_checkpoint",
      )
  
      # wait until HTTPEventProvider is ready
      def check_app_running():
          status = serve.status().applications[
              common.HTTP_EVENT_PROVIDER_NAME
          ]
          assert status.status == "RUNNING"
          return True
  
      wait_for_condition(check_app_running)
  
      test_msg = "first_try"
  
      while True:
          res = send_event(test_msg)
          if not isinstance(res, int):
              if res.status_code == 404:
                  sleep(0.5)
              else:
                  break
          else:
              break
  
      while not utils.check_global_mark("simulate_cluster_crash"):
          sleep(0.1)
  
      if utils.check_global_mark("simulate_cluster_crash"):
          serve.delete(common.HTTP_EVENT_PROVIDER_NAME)
          serve.shutdown()
          ray.shutdown()
          subprocess.check_output(["ray", "stop", "--force"])
  
          ray.init(num_cpus=4, storage=storage_uri)
          serve.start(detached=True)
          utils.set_global_mark("after_cluster_restarted")
  
          workflow.resume_async(
              workflow_id="workflow_test_cluster_crash_before_checkpoint"
          )
          status_after_resume = workflow.get_status(
              workflow_id="workflow_test_cluster_crash_before_checkpoint"
          )
  
          wait_for_condition(check_app_running)
  
          assert status_after_resume == WorkflowStatus.RUNNING
  
          test_msg = "second_try"
  
          while True:
              res = send_event(test_msg)
              if not isinstance(res, int):
                  if res.status_code == 404:
                      sleep(0.5)
                  else:
                      break
              else:
                  break
  
          key, event_message = workflow.get_output(
              workflow_id="workflow_test_cluster_crash_before_checkpoint"
          )
          assert event_message == "second_try"
  
      else:
          assert False
  
file_path: ray/python/ray/workflow/tests/test_event_resume_after_crash.py
imports: import asyncio
  import subprocess
  from time import sleep
  import pytest
  import ray
  from ray import workflow, serve
  from ray.workflow.http_event_provider import HTTPListener, WorkflowEventHandleError
  from ray.tests.conftest import *
  from ray.workflow.tests import utils
  from ray.workflow.common import WorkflowStatus
  from ray.workflow import common, workflow_context
  from ray._private.test_utils import wait_for_condition
  import requests
label_desc: Use image node_id 1 for calling a global function with example usage: test_cluster_crash_before_checkpoint(workflow_start_regular_shared_serve) and returns: True, resp, payload, int