This message was deleted.

Your message was too long, here's your message at the allowed limit: I am having a problem where some types of flows do not time out when they are supposed to for example this test flow times out as expected where the tasks just sleep for longer than the timeout

Minimal test flow to verify tasks appear in Prefect UI import asyncio from prefect import flow task from prefect settings import PREFECT_API_URL from prefect logging import get_run_logger @task async def task_one(value str) -> str First task - waits 3 seconds logger = get_run_logger() <http //logger info|logger info>(f Task one starting with value {value} ) await asyncio sleep(10) <http //logger info|logger info>( Task one complete ) return f one-{value} @task async def task_two(value str) -> str Second task - waits 5 seconds logger = get_run_logger() <http //logger info|logger info>(f Task two starting with value {value} ) await asyncio sleep(20) <http //logger info|logger info>( Task two complete ) return f two-{value} @task async def task_three(val1 str val2 str) -> str Third task - combines results waits 2 seconds logger = get_run_logger() <http //logger info|logger info>(f Task three combining {val1} + {val2} ) await asyncio sleep(30) result = f {val1}|{val2} <http //logger info|logger info>(f Task three complete {result} ) return result # @flow(timeout_seconds=60) # async def test_task_visibility_flow(input_value str = test ) -> str # logger = get_run_logger() # <http //logger info|logger info>(f PREFECT_API_URL {PREFECT_API_URL} ) # <http //logger info|logger info>(f Flow starting with input {input_value} ) # # Call tasks with await - these should show in UI # result_one = await task_one(input_value) # result_two = await task_two(input_value) # result_three = await task_three(result_one result_two) # <http //logger info|logger info>(f Flow complete with result {result_three} ) # return result_three @flow(timeout_seconds=15) async def test_task_visibility_flow(input_value str = test ) -> str logger = get_run_logger() <http //logger info|logger info>(f PREFECT_API_URL {PREFECT_API_URL} ) <http //logger info|logger info>(f Flow starting with input {input_value} ) <http //logger info|logger info>( Running with submit instead of await ) r1 = task_one submit(input_value) result() r2 = task_two submit(input_value) result() r3 = task_three submit(r1 r2) result() <http //logger info|logger info>(f Flow complete with result {r3} ) return r3 if __name__ == __main__ asyncio run(test_task_visibility_flow())

however flows like this that wait for a callback exceed the timeout regularly (we expect the callback to be slow and want to kill the flow after 10 minutes if the external API is experiencing lots of latency ```@flow(timeout_seconds=600) async def n8n_flow_aggk_search(search N8NFlowCompanySearchSchema) -> None await n8n_generic_callback_flow_task( get_run_context() N8N_AGGK_COMPANY_SEARCH_WEBHOOK_ID /search/company/aggk CompanyAggkSearchFlowCallback { search search model_dump(mode= json )} ) async def n8n_generic_callback_flow_task( context FlowRunContext webhook_id str callback_prefix str wait_for_input Type[T] payload dict[str Any] | None = None timeout float = 600 0 testing bool = False ) -> T Trigger an n8n webhook and wait for callback with input Args context The current flow run context webhook_id The n8n webhook ID to call callback_prefix Prefix for the callback URL wait_for_input Type of input to wait for (ResumableInput subclass) payload JSON payload to send to the webhook timeout Timeout in seconds to wait for input testing Whether to use test mode when calling

@Marvin I am having a problem where some types of flows do not time out when they are supposed to. for example, this test flow times out as expected where the tasks just sleep for longer than the timeout:

"""Minimal test flow to verify tasks appear in Prefect UI."""

import asyncio

from prefect import flow, task
from prefect.settings import PREFECT_API_URL
from prefect.logging import get_run_logger


@task
async def task_one(value: str) -> str:
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"Task one starting with value: {value}")
    await asyncio.sleep(30)
    <http://logger.info|logger.info>("Task one complete")
    return f"one-{value}"

@flow(timeout_seconds=15)
async def test_task_visibility_flow(input_value: str = "test") -> str:
    logger = get_run_logger()
    r1 = task_one.submit(input_value).result()
    <http://logger.info|logger.info>(f"Flow complete with result: {r3}")
    return r3

however, flows like this that wait for a callback exceed the timeout regularly (we expect the callback to be slow and want to kill the flow after 10 minutes if the external API is experiencing lots of latency

@flow(timeout_seconds=600)
async def n8n_flow_aggk_search(search: N8NFlowCompanySearchSchema) -> None:
    await n8n_generic_callback_flow_task(
        get_run_context(),
        N8N_AGGK_COMPANY_SEARCH_WEBHOOK_ID,
        "/search/company/aggk",
        CompanyAggkSearchFlowCallback,
        {"search": search.model_dump(mode="json")},
    )

async def n8n_generic_callback_flow_task(
    context: FlowRunContext,
    webhook_id: str,
    callback_prefix: str,
    wait_for_input: Type[T],
    payload: dict[str, Any] | None = None,
    timeout: float = 600.0,
    testing: bool = False,
) -> T:
    logger = get_run_logger(context)
    flow_run_id = context.flow_run.id
    base_url = config.sourcing.api_url.encoded_string().removesuffix("/")
    callback_url = f"{base_url}{callback_prefix}/{flow_run_id}/callback"
    await call_n8n_webhook(
        webhook_id=webhook_id,
        payload=(payload or {}) | {"callback": callback_url},
        testing=testing,
    )
    <http://logger.info|logger.info>(f"Waiting for callback on POST {callback_url} (timeout: {timeout})")
    value = await wait_for_input.receive_one(
        timeout,
        flow_run_id=flow_run_id,
    )
    <http://logger.info|logger.info>(f"Received callback input: {value}")
    return value

class ResumableInput(RunInput):
    @classmethod
    async def receive_one(
        cls,
        timeout: float = 3600,
        poll_interval: float = 10,
        exclude_keys: set[str] | None = None,
        key_prefix: str | None = None,
        flow_run_id: UUID | None = None,
    ) -> "ResumableInput":
        logger = get_run_logger()

        try:
            with anyio.fail_after(timeout + HARD_TIMEOUT_DELAY_SECS):
                gen = cls.receive(
                    timeout=timeout,
                    poll_interval=poll_interval,
                    raise_timeout_error=True,
                    exclude_keys=exclude_keys,
                    key_prefix=key_prefix,
                    flow_run_id=flow_run_id,
                )
                async for item in gen:
                    return item
        except TimeoutError:
            logger.warning(
                f"No input received by {cls}.receive_one - "
                "possible key_prefix mismatch or timeout."
            )
            raise

🔄 Researching your answer... this may take a while 🔧 Using tool:

research_prefect_topic

📊 Tools used: •

display_callable_signature

(1x) •

explore_module_offerings

(2x) •

read_github_issues

(1x) •

research_prefect_topic

(3x)