@task(persist_result=True, result_storage=s3_block)
async def execute_node(
    upstream_states:List[Union[Coroutine,State,FlowRun]] = [],
    name: str = None,
    timeout: int = None,
    handle_upstream_strategy: Any = None,
) ->   FlowRun:
    # Return the actual FLOW so it can be chained into dependent flows
    #random string for flow_run_name
    flow_run_name = ''.join(random.choices(string.ascii_uppercase +
                             string.digits, k=N))

    if upstream_states:
        #Cascade await triggers upstream states to run
        
        #if the values in upstream_states are coroutines, await them else if they are FlowRun, use resolve_flow_run else use them as is
        awaited_states = [(await x).result().get() if isinstance(x, Coroutine) else resolve_flow_run(x) if isinstance(x, FlowRun) else x for x in upstream_states]
        awaited_states = [await x.result().get() for x in upstream_states]
        flow_run_name = str(awaited_states)

    run = run_deployment(
        name=name,
        flow_run_name=flow_run_name,
        tags=["test"],
        parameters=dict(input_str=name),
        timeout=timeout,
    )
    run_result = await run
    return run_result


@flow(name="test-orchestrator", persist_result=True, result_storage=s3_block,log_prints=True)
def test_flow(input_str: str):
    res = ''.join(random.choices(string.ascii_uppercase +
                             string.digits, k=N))
    if input_str:
        res = f"*{input_str}{res}*"

    #randomly raise an exception with 50% chance
    if random.random() > 0.5:
        raise Exception("Random Exception")

    #randomly sleep for 30-120 seconds
    random_sleep = random.randint(10,120)
    print("randomly sleeping for",random_sleep)
    time.sleep(random_sleep)
    return res


@flow(name="debug-orchestrator")
async def debug_orchestrator():
    res1 =  execute_node.submit(upstream_states=[], name='test-orchestrator/local_orch_test',return_state=True)
    #join the two results
    res2 =  execute_node.submit(upstream_states=[], name='test-orchestrator/local_orch_test',return_state=True)

    res3 = execute_node.submit(upstream_states=[res1], name='test-orchestrator/local_orch_test',return_state=True)
    res4 = execute_node.submit(upstream_states=[res2,res3], name='test-orchestrator/local_orch_test',return_state=True)

    final_res =  execute_node.submit(upstream_states=[res3,res4], name='test-orchestrator/local_orch_test',return_state=True)

    await final_res