code: def test_recovery_simple_3(workflow_start_regular):
      @ray.remote
      def append1(x):
          return x + "[append1]"
  
      @ray.remote
      def append2(x):
          return x + "[append2]"
  
      @ray.remote
      def simple(x):
          x = append1.bind(x)
          y = the_failed_task.bind(x)
          z = append2.bind(y)
          return workflow.continuation(z)
  
      utils.unset_global_mark()
      workflow_id = "test_recovery_simple_3"
      with pytest.raises(workflow.WorkflowExecutionError):
          # internally we get WorkerCrashedError
          workflow.run(simple.bind("x"), workflow_id=workflow_id)
  
      assert (
          workflow.get_status(workflow_id)
          == workflow.WorkflowStatus.FAILED
      )
  
      utils.set_global_mark()
      assert workflow.resume(workflow_id) == "foo(x[append1])[append2]"
      utils.unset_global_mark()
      # resume from workflow output checkpoint
      assert workflow.resume(workflow_id) == "foo(x[append1])[append2]"
  
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 9 for calling a global function with example usage: test_recovery_simple_3(workflow_start_regular) and returns: unknown, unknown, workflow