Description
What happened + What you expected to happen
I got interesting behaviour when trying to know my workflow status:
Option 1 : ray.workflow.get_status(workflow_id) --> RESUMABLE
Option 2 : ray.workflow.get_metadata(workflow_id)["status"] --> RUNNING
I call those two options with same workflow_id at the same time, and those two options does return the same status.
Note that bug does not appear for every workflow status, but at least for "RESUMABLE" state.
I think option 2 is wrong because even if I wait for a long time, the workflow does not change its state. So it does not seem RUNNING anymore.
Versions / Dependencies
Ray 2.1.0
Reproduction script
import time
import ray
@ray.remote
def task():
time.sleep(10)
workflow_id = '12345'
ray.workflow.run_async(task.bind(), workflow_id=workflow_id)
time.sleep(1)
print(f'\n{workflow_id}: status from metadata: {ray.workflow.get_metadata(workflow_id)["status"]}')
print(f'{workflow_id}: status from status: {ray.workflow.get_status(workflow_id)}')
'''Do something that makes the workflow resumable but I don't know what... '''
time.sleep(1)
print(f'\n{workflow_id}: status from metadata: {ray.workflow.get_metadata(workflow_id)["status"]}')
print(f'{workflow_id}: status from status: {ray.workflow.get_status(workflow_id)}')
Issue Severity
Medium: It is a significant difficulty but I can work around it.