code: def test_crash_after_commit(workflow_start_regular):
      """Ensure that we don't re-call poll_for_event after `event_checkpointed`
      returns, even after a crash. Here we must call `event_checkpointed`
      twice, because there's no way to know if we called it after
      checkpointing.
      """
  
      from ray._private import storage
      from ray.workflow.tests.utils import skip_client_mode_test
  
      # This test restarts the cluster, so we cannot test under client mode.
      skip_client_mode_test()
  
      storage_uri = storage._storage_uri
  
      class MyEventListener(workflow.EventListener):
          async def poll_for_event(self):
              assert not utils.check_global_mark("committed")
  
          async def event_checkpointed(self, event):
              utils.set_global_mark("committed")
              if utils.check_global_mark("first"):
                  utils.set_global_mark("second")
              else:
                  utils.set_global_mark("first")
                  await asyncio.sleep(1000000)
  
      event_promise = workflow.wait_for_event(MyEventListener)
      workflow.run_async(event_promise, workflow_id="workflow")
  
      while not utils.check_global_mark("first"):
          time.sleep(0.1)
  
      ray.shutdown()
      subprocess.check_output(["ray", "stop", "--force"])
  
      ray.init(num_cpus=4, storage=storage_uri)
      workflow.init()
      workflow.resume_async("workflow")
  
      workflow.get_output("workflow")
      assert utils.check_global_mark("second")
  
file_path: ray/python/ray/workflow/tests/test_events_with_crash.py
imports: import asyncio
  import subprocess
  import time
  import pytest
  import ray
  from ray import workflow
  from ray.tests.conftest import *
  from ray.workflow.tests import utils
label_desc: Use image node_id 2 for calling a global function with example usage: test_crash_after_commit(workflow_start_regular) without return types