code: def test_dedupe_cluster_failure(shutdown_only, tmp_path):
      """
      ======== driver 1 ===========
      1. Checkpoing the input args
          * Uploads
      2. Begin to run task
          * Crash
  
      ====== driver 2 ============
      1. Recover inputs
          * Creates a new object ref
      2. Finish running task
      3. Checkpoint task output
          * Should not trigger upload
      """
      lock_file = tmp_path / "lock"
      workflow_dir = tmp_path / "workflow"
  
      driver_script = f"""
  import time
  import ray
  from ray import workflow
  from filelock import FileLock
  
  @ray.remote
  def foo(objrefs):
      with FileLock("{str(lock_file)}"):
          return objrefs
  
  if __name__ == "__main__":
      ray.init(storage="{str(workflow_dir)}")
      workflow.init()
      arg = ray.put("hello world")
  
      workflow.run(foo.bind([arg, arg]))
      assert False
      """
  
      with FileLock(lock_file):
          run_string_as_driver_nonblocking(driver_script)
          time.sleep(10)
          subprocess.check_call(["ray", "stop", "--force"])
  
      ray.init(storage=str(workflow_dir))
      workflow.init()
      resumed = workflow.resume_all()
      assert len(resumed) == 1
      objref = resumed.pop()[1]
      ray.get(objref)
  
      # The object ref will be different before and after recovery, so it will
      # get uploaded twice.
      assert get_num_uploads() == 1
      ray.shutdown()
  
file_path: ray/python/ray/workflow/tests/test_serialization2.py
imports: import subprocess
  import time
  import pytest
  from filelock import FileLock
  import ray
  from ray import workflow
  from ray._private.test_utils import run_string_as_driver_nonblocking
  from ray.tests.conftest import *
  from ray.workflow import serialization, workflow_storage
label_desc: Use image node_id 2 for calling a global function with example usage: test_dedupe_cluster_failure(shutdown_only, tmp_path) without return types