Thanks for the detailed examples — this is a great question. High-level behavior in Prefect 3.x - Flow-level

timeout_seconds

is enforced while the flow run is in a Running state. If the flow is paused or rescheduled, the time spent paused does not count, and the timeout is reset on resume (per attempt). - RunInputs (

RunInput.receive()

/

receive_one()

) do not pause the flow by themselves — they poll for input while the flow stays Running and use their own

anyio.fail_after(timeout)

internally. - A strict wall-clock cap only holds if the run remains a single, uninterrupted Running attempt. If your run is paused, rescheduled, or crashes and is restarted by the worker, the flow timeout will restart on resume. Why your callback-style flow can exceed

timeout_seconds

- If the flow is ever Paused or Rescheduled (explicitly or due to infra issues), the flow’s

timeout_seconds

“clock” resets on resume. This is the most common reason for seeing overall wall-clock exceed the configured timeout. - If your code spawns background tasks or threads that outlive the flow’s cancellation scope, they can keep work going, but in your snippet you’re awaiting inside the flow so this is less likely. - If cancellation is accidentally ignored (e.g., catching

BaseException

or

CancelledError

), the flow can overrun. Your

receive_one

only catches

TimeoutError

, which is good — it should not swallow the flow timeout. What I recommend for callback flows with a strict 10‑minute cap You have two solid patterns, depending on whether you want to pause the flow or keep it Running. 1) Preferred for “wait for callback” UX (no polling): use an explicit pause with an input form and its own timeout - This makes the flow actually Paused (clear in the UI), avoids polling, and gives you a hard deadline for the callback. - Important: time spent paused does not count toward flow

timeout_seconds

. If you use pause-based waiting, lean on the pause timeout instead of the flow timeout for the cap. Example:

from prefect import flow
from prefect.flow_runs import pause_flow_run, FlowPauseTimeout
from prefect.input import RunInput

class CompanyAggkSearchFlowCallback(RunInput):
    # define fields expected from your callback
    company_id: str
    # ...

@flow
async def n8n_flow_aggk_search(search: ...):
    # Call your webhook that will eventually POST the input back
    await call_n8n_webhook(...)

    try:
        # Pause and wait up to 10 minutes for input; no polling while paused
        result = await pause_flow_run(
            wait_for_input=CompanyAggkSearchFlowCallback,
            timeout=600,
        )
    except FlowPauseTimeout:
        # Hard 10 minute cap for the callback
        raise RuntimeError("Callback not received within 10 minutes")

    # Process result here
    ...

2) If you want to keep the run in Running and poll (your current approach): rely on the flow timeout, and ensure there are no pauses/reschedules - Keep using

RunInput.receive_one(timeout=...)

(pure polling). - Keep

@flow(timeout_seconds=600)

. - Verify the run does not get Paused, Suspended, Crashed, or retried — any of those reset the flow timeout on resume. Concrete things to check to explain overruns - In the Prefect UI, open the flow run and check the timeline/state history: - Do you see Paused/Suspended or multiple Running attempts? If so, the timeout reset explains the overrun. - Do you see a Crash followed by the worker re-submitting the run? That’s another reset. - Look at worker logs for restarts or lost heartbeats around the time of overrun. - Approximately how far past 600s are you seeing? A small overrun (e.g., 10–30s) can be explained by polling intervals and cooperative cancellation. Large overruns point to reschedules or retries. - Confirm you aren’t catching

CancelledError

or

BaseException

anywhere around the wait; that would inadvertently ignore the flow’s timeout cancellation.

If you want a strict, end-to-end 10‑minute wall‑clock budget even across pauses/reschedules - Store a deadline at the start (e.g., in parameters or a persisted key/value) and check it on resume. - Fail fast if

time.monotonic() > deadline

whenever the flow continues after a pause/reschedule. Example:

import time
from prefect import flow

@flow
async def n8n_flow_aggk_search(..., deadline: float | None = None):
    deadline = deadline or (time.monotonic() + 600)

    # ... issue webhook

    # If you might pause/resume, always check
    if time.monotonic() > deadline:
        raise RuntimeError("Overall deadline exceeded")

    # Wait for input (pause or poll)
    # On resume/continue, check the deadline again
    if time.monotonic() > deadline:
        raise RuntimeError("Overall deadline exceeded")

Next steps / questions to help pinpoint your case - Which execution environment/worker type are you using (process, Docker, K8s, ECS, etc.)? - Do the overrunning flow runs show any Paused/Suspended/Crash/Retry events in their state history? - Roughly how long past 600s are they running? - If you can share a run ID that overran, I can point to the exact state transitions that caused a reset. If your goal is “kill after 10 minutes waiting on external API,” I’d use the pause/input pattern with

pause_flow_run(wait_for_input=..., timeout=600)

and treat that as the canonical deadline. It’s simpler, visible in the UI, and not subject to reschedule-induced resets.

Thanks for the extra context — that helps narrow it down. Given: - You’re keeping the run in Running (no pauses), and - The run can exceed

timeout_seconds

by many hours, and - Clicking “Cancel” leaves it stuck in Cancelling, the most likely cause is that the flow is spending its time inside a non‑cancellable section of code (e.g., a blocking sync I/O call or a cancellation “shield”), so neither the flow’s timeout nor the UI cancellation can be delivered. Prefect’s flow timeout and cancellation are cooperative: they’re delivered at await/cancellation points. If your code is sitting inside a blocking call (requests, time.sleep, a synchronous DB driver call, a long-running CPU loop, etc.), the cancellation won’t be observed until that call returns. Where to look in your code - The behavior difference you’re seeing is between: - simple “sleep longer than timeout” (works), vs. - waiting for a callback via your

receive_one()

path (overruns). - You shared

receive_one()

, but the crucial bit is the implementation of

cls.receive(...)

that you iterate with

async for

. If that generator uses any blocking operations (e.g., requests, time.sleep, synchronous client methods), the flow won’t see cancellation/timeout until those calls finish. - Also scan for: - any use of

anyio.CancelScope(shield=True)

or a similar “shield” around the wait loop, - broad exception handling that might swallow cancellation (in some stacks, cancellation can surface as

asyncio.CancelledError

, which subclasses Exception; if you catch

Exception

and keep going, you’ll ignore cancellation). Why this matches your symptoms - Flow timeout not firing: if your run is inside a blocking call, the flow’s cancellation scope can’t interrupt it; the timeout only takes effect at the next await/interruptible point. - UI stuck in “Cancelling”: the cancel signal was sent, but the user code did not reach a cancellation point, so it never transitioned to Cancelled. Concrete steps to diagnose - Please share: - Prefect version (exact 3.x patch), Python version - Worker/infrastructure type (process, Docker, Kubernetes, ECS, etc.) - The implementation of

ResumableInput.receive()

(the async generator) - What HTTP/DB libraries are used in that path (requests vs httpx.AsyncClient, time.sleep vs anyio.sleep, etc.) - Check the flow run’s logs around the 10-minute mark for “flow exceeded timeout” messages; and the worker logs for “received cancel” messages when you clicked Cancel. - In the Prefect UI for run 8ed34ad2-669d-4699-9f19-f6e10bcf8bd7, review the state history to confirm it remained a single Running attempt (no Paused/Suspended/Crash); that aligns with a blocking section. Immediate mitigations - Make the wait path fully async and cancellable: - Use

httpx.AsyncClient

(not

requests

) - Use

anyio.sleep

(not

time.sleep

) - Use async DB drivers if applicable - Do not wrap the polling loop in a shield; avoid catching

Exception

around it; if you must, re-raise

CancelledError

explicitly - Add a belt-and-suspenders outer guard at the call site, so even if an inner helper misbehaves, you still enforce a hard cap:

from anyio import fail_after, CancelledError
  from prefect import flow

  @flow(timeout_seconds=600)  # optional; rely on fail_after too
  async def n8n_flow_aggk_search(...):
      try:
          with fail_after(600):
              await n8n_generic_callback_flow_task(...)
      except TimeoutError:
          # hard 10-minute cap regardless of inner helpers
          raise RuntimeError("Callback not received within 10 minutes")
      except CancelledError:
          # never swallow cancellation
          raise

- If you discover you must perform blocking I/O in the wait loop, move it into a separate thread/subprocess and make the outer layer cancellable:

- For threads:

await anyio.to_thread.run_sync(blocking_call, cancellable=True)

and ensure you periodically check for cancellation (threads can’t be force-killed). - For truly uninterruptible calls, run them in a subprocess you can terminate on timeout. A note on “keeping it Running so other flows do not start” - If this is to throttle load on the external API, consider Prefect’s concurrency limits to gate the API calls explicitly. That lets you pause/resume safely without accidentally releasing your “reservation” via a Paused state. If you share more about how you’re gating today, I can suggest a concrete setup. Next actions - If you can post the

receive()

generator and confirm the libraries used in that path, I can point to the exact uninterruptible spot and propose a drop-in async replacement. - Also, please share the Prefect/worker versions and environment. The fact that Cancel didn’t transition suggests the process never hit a cancellation point; worker type can help explain the mechanics.

Thanks, that detail helps. Given: - You’re using `RunInput`’s stock implementation (so pure async polling with

anyio.sleep

+

anyio.fail_after

) - You keep the run in Running (no pauses) - Affected runs can exceed

timeout_seconds

by many hours and do not respond to Cancel (stuck in Cancelling) the core symptom points to the flow code being stuck in a non-cancellable section (or otherwise never hitting an await/cancellation point), so neither the flow’s timeout nor the cancel signal can be delivered. Why that’s surprising here - Prefect’s

RunInput.receive()

uses

anyio.sleep()

in a loop plus per-call HTTP requests to the API. Both should be cancellation-friendly. If the flow-level timeout or a UI cancel occurs, it should interrupt on the next await in the polling loop. - Since you’re inheriting

RunInput.receive()

unchanged and not catching

CancelledError

in your

receive_one

, cancellation should propagate. What could still cause 19–24h overruns and “stuck Cancelling” - A blocking operation around or inside the polling path that never yields back to the event loop (e.g., a sync HTTP client,

time.sleep

, CPU-bound loop, or a library call that blocks the event loop). Even one such call in the loop will prevent timeouts/cancel from being observed. - An outer cancellation shield or broad

except Exception:

that swallows

CancelledError

. - The worker process is alive but the actual run task crashed into an internal state that never returns control to Prefect’s engine (rare, but logs would show). Targeted checks to pinpoint it 1) Minimal repro in your environment - If this simple flow respects the 5s timeout and is cancelable, the issue is in your callback path; if it doesn’t, we likely have an environment/version-specific bug.

from prefect import flow
from prefect.input import RunInput
from anyio import fail_after

class TestInput(RunInput):
    msg: str

@flow(timeout_seconds=5)
async def timeout_probe():
    # Keep it Running, no pause, pure polling
    with fail_after(5):
        await TestInput.receive_one(timeout=60, poll_interval=1)

2) Add debug logs in your wait loop - Temporarily add a log each iteration right before/after the

anyio.sleep(poll_interval)

inside your

receive_one

wrapper (or set

poll_interval=1

just for the test). - If logs stop for long stretches, something is blocking the event loop. - If logs continue past 10 minutes and you don’t see your own

TimeoutError

log from the

except TimeoutError:

block, there’s a mismatch between expected timeout values. 3) Confirm there’s no shielding or exception swallowing - Search your code path for

CancelScope(shield=True)

and for

except Exception:

around the callback wait. If you must catch broad exceptions, explicitly re-raise

CancelledError

. 4) Worker + run logs at timeout/cancel moments - Run the worker with DEBUG logging and capture logs: - Right after the flow reaches 600s, do you see any “timeout fired” or cancellation logs? - When you click Cancel in the UI, do you see the worker receive and act on the cancel? 5) Version/environment - Please share your Prefect version (server and worker) and Python version used in the worker. If the minimal flow above fails to respect timeouts/cancel, I’ll escalate as a bug with those details. Immediate mitigations you can deploy now - Belt-and-suspenders timeout at the call site (this enforces a hard cap even if a helper gets stuck):

from anyio import fail_after, CancelledError

@flow(timeout_seconds=600)  # keep or remove; the fail_after is the hard cap
async def n8n_flow_aggk_search(...):
    try:
        with fail_after(600):
            await n8n_generic_callback_flow_task(...)
    except TimeoutError:
        # This will trip even if inner logic blocks, ensuring the flow ends
        raise RuntimeError("Callback not received within 10 minutes")
    except CancelledError:
        # Never swallow cancellations
        raise

- Reduce

poll_interval

(e.g., to 1–2s) during experiments so cancellation is observed more quickly, and to make the “is the loop alive?” logging conclusive. About keeping the run Running to avoid starting more flows - A safer long-term control is Prefect’s concurrency limits around the external API call(s). That lets you control how many flows can initiate the webhook, regardless of whether a particular run later pauses or keeps running. If you want, I can help map your “max 2 inflight requests to service X” requirement to a

prefect.concurrency

setup. It won’t hold a slot while Paused, but you can acquire the slot before calling your webhook and release it only after the callback completes, ensuring only N flows can be in the “waiting on API X” phase. Next step - Could you share: - Prefect version(s) and Python version for the worker - Whether the minimal

timeout_probe

flow respects timeout/cancel in your environment - Any logs around run 8ed34ad2-669d-4699-9f19-f6e10bcf8bd7 at the 10-minute mark and when you clicked Cancel If the minimal probe also overruns or ignores cancel, I’ll treat this as a bug and help get it into our tracker with your run ID and environment info. In the meantime, wrapping the wait with an outer

fail_after(600)

in your flow is the most reliable way to enforce the SLA today. You can also try cancelling via CLI to confirm behavior:

prefect flow-run cancel 8ed34ad2-669d-4699-9f19-f6e10bcf8bd7
prefect flow-run inspect 8ed34ad2-669d-4699-9f19-f6e10bcf8bd7