code: def test_recovery_cluster_failure(tmp_path, shutdown_only):
      ray.shutdown()
      subprocess.check_call(
          ["ray", "start", "--head", f"--storage={tmp_path}"]
      )
      time.sleep(1)
      proc = run_string_as_driver_nonblocking(
          """
  import time
  import ray
  from ray import workflow
  
  @ray.remote
  def foo(x):
      print("Executing", x)
      time.sleep(1)
      if x < 20:
          return workflow.continuation(foo.bind(x + 1))
      else:
          return 20
  
  if __name__ == "__main__":
      ray.init()
      assert workflow.run(foo.bind(0), workflow_id="cluster_failure") == 20
  """
      )
      time.sleep(10)
      subprocess.check_call(["ray", "stop"])
      proc.kill()
      time.sleep(1)
      ray.init(storage=str(tmp_path))
      workflow.init()
      assert workflow.resume("cluster_failure") == 20
      ray.shutdown()
  
imports: import subprocess
  import tempfile
  import time
  from ray.tests.conftest import *
  import pytest
  from filelock import FileLock
  import ray
  from ray._private.test_utils import run_string_as_driver_nonblocking
  from ray import workflow
  from ray.workflow import workflow_storage
  from ray.workflow.storage.debug import DebugStorage
  from ray.workflow.tests import utils
  from ray.workflow.exceptions import WorkflowNotResumableError
label_desc: Use image node_id 12 for calling a global function with example usage: test_recovery_cluster_failure(tmp_path, shutdown_only) without return types