code: def test_recovery_cluster_failure_resume_all(tmp_path, shutdown_only):
      ray.shutdown()
  
      tmp_path = tmp_path
      workflow_dir = tmp_path / "workflow"
      subprocess.check_call(
          ["ray", "start", "--head", f"--storage={workflow_dir}"]
      )
      time.sleep(1)
      lock_file = tmp_path / "lock_file"
      lock = FileLock(lock_file)
      lock.acquire()
  
      proc = run_string_as_driver_nonblocking(
          f"""
  import time
  import ray
  from ray import workflow
  from filelock import FileLock
  
  @ray.remote
  def foo(x):
      with FileLock("{str(lock_file)}"):
          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)
      lock.release()
      ray.init(storage=str(workflow_dir))
      workflow.init()
      resumed = workflow.resume_all()
      assert len(resumed) == 1
      (wid, obj_ref) = resumed[0]
      assert wid == "cluster_failure"
      assert ray.get(obj_ref) == 20
  
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 13 for calling a global function with example usage: test_recovery_cluster_failure_resume_all(tmp_path, shutdown_only) without return types