code: def test_task_failure(workflow_start_regular_shared, tmp_path):
      @ray.remote(max_retries=10, retry_exceptions=True)
      def unstable_task_exception(n):
          v = int((tmp_path / "test").read_text())
          (tmp_path / "test").write_text(f"{v + 1}")
          if v < n:
              raise ValueError("Invalid")
          return v
  
      @ray.remote(max_retries=10)
      def unstable_task_crash(n):
          v = int((tmp_path / "test").read_text())
          (tmp_path / "test").write_text(f"{v + 1}")
          if v < n:
              import os
  
              os.kill(os.getpid(), 9)
          return v
  
      @ray.remote(max_retries=10, retry_exceptions=True)
      def unstable_task_crash_then_exception(n):
          v = int((tmp_path / "test").read_text())
          (tmp_path / "test").write_text(f"{v + 1}")
          if v < n / 2:
              import os
  
              os.kill(os.getpid(), 9)
          elif v < n:
              raise ValueError("Invalid")
          return v
  
      with pytest.raises(Exception):
          unstable_task_exception.options(max_retries=-2)
  
      for task in (
          unstable_task_exception,
          unstable_task_crash,
          unstable_task_crash_then_exception,
      ):
          (tmp_path / "test").write_text("0")
          assert workflow.run(task.bind(10)) == 10
  
          (tmp_path / "test").write_text("0")
          with pytest.raises(workflow.WorkflowExecutionError):
              workflow.run(task.bind(11))
  
      # TODO(suquark): catch crash as an exception
      for task in (
          unstable_task_exception,
          unstable_task_crash_then_exception,
      ):
          (tmp_path / "test").write_text("0")
          (ret, err) = workflow.run(
              task.options(
                  **workflow.options(catch_exceptions=True)
              ).bind(10)
          )
          assert ret == 10
          assert err is None
  
          (tmp_path / "test").write_text("0")
          (ret, err) = workflow.run(
              task.options(
                  **workflow.options(catch_exceptions=True)
              ).bind(11)
          )
          assert ret is None
          assert err is not None
  
      (tmp_path / "test").write_text("0")
      with pytest.raises(workflow.WorkflowExecutionError):
          workflow.run(
              unstable_task_exception.options(
                  retry_exceptions=False
              ).bind(10)
          )
  
      (tmp_path / "test").write_text("0")
      workflow.run(
          unstable_task_crash.options(retry_exceptions=False).bind(10)
      )
  
      (tmp_path / "test").write_text("0")
      with pytest.raises(workflow.WorkflowExecutionError):
          workflow.run(
              unstable_task_crash_then_exception.options(
                  retry_exceptions=False
              ).bind(10)
          )
  
file_path: ray/python/ray/workflow/tests/test_error_handling.py
imports: import pytest
  import ray
  from ray import workflow
  from ray.tests.conftest import *
label_desc: Use image node_id 1 for calling a global function with example usage: test_task_failure(workflow_start_regular_shared, tmp_path) and returns: v, v, v