code: def test_failure_with_storage(workflow_start_regular):
      with tempfile.TemporaryDirectory() as temp_dir:
          debug_store = DebugStorage(temp_dir)
          _alter_storage(debug_store)
  
          wf = construct_workflow(length=3)
          result = wf.run(workflow_id="complex_workflow")
          index = _locate_initial_commit(debug_store) + 1
          debug_store.log_off()
  
          def resume(num_records_replayed):
              key = debug_store.wrapped_storage.make_key(
                  "complex_workflow"
              )
              asyncio_run(
                  debug_store.wrapped_storage.delete_prefix(key)
              )
  
              async def replay():
                  # We need to replay one by one to avoid conflict
                  for i in range(num_records_replayed):
                      await debug_store.replay(i)
  
              asyncio_run(replay())
              return workflow.resume(workflow_id="complex_workflow")
  
          with pytest.raises(ValueError):
              # in cases, the replayed records are too few to resume the
              # workflow.
              resume(index - 1)
  
          if isinstance(
              debug_store.wrapped_storage, FilesystemStorageImpl
          ):
              # filesystem is faster, so we can cover all cases
              task_len = 1
          else:
              task_len = max((len(debug_store) - index) // 5, 1)
  
          for j in range(index, len(debug_store), task_len):
              assert resume(j) == result
  
file_path: ray/python/ray/workflow/tests/test_storage_failure.py
imports: from hashlib import sha1
  import tempfile
  import pytest
  import ray
  from ray import workflow
  from ray.workflow.storage.debug import DebugStorage
  from ray.workflow.workflow_storage import STEP_OUTPUTS_METADATA
  from ray.workflow.common import asyncio_run
  from ray.workflow.storage.filesystem import FilesystemStorageImpl
  from ray.workflow.tests.utils import _alter_storage
label_desc: Use image node_id 8 for calling a global function with example usage: test_failure_with_storage(workflow_start_regular) and returns: workflow