A few likely culprits, based on how dbt is being invoked from Prefect: - If you’re running dbt via a subprocess (e.g.,

DbtCoreOperation

from older prefect-dbt versions or `prefect-shell`’s

ShellOperation

), Prefect’s timeout/cancellation is cooperative. Prefect will cancel the task and send SIGTERM to the subprocess, but if dbt (or a child process it spawned) doesn’t exit, Prefect waits during cleanup and the run can sit “Running” with no new logs. This exactly fits “logs stop, never times out, stays running until manual cancel.” - If the timeout was applied at the flow level instead of the task that actually invokes dbt, you can see similar behavior because the cleanup step is shielded from further cancellation; a stuck subprocess can keep the run alive. - If you’re on an older

prefect-dbt

(pre-0.7),

DbtCoreOperation

shells out to dbt under the hood, which has the same subprocess caveats above. Quick questions to confirm: - How are you invoking dbt:

PrefectDbtRunner

,

DbtCoreOperation

,

ShellOperation

, or a custom

run_process

? - Where is the timeout applied (flow or task), and what Prefect/prefect-dbt versions are you on? - What worker/run environment are you using (process, Docker, K8s)? Recommended fixes 1) Best fix: run dbt in-process with PrefectDbtRunner (prefect-dbt ≥ 0.7) - This avoids subprocesses entirely, so timeouts/cancellation are reliable and logs keep flowing. - Apply the timeout at the task that runs dbt. Example:

from prefect import flow, task
from prefect_dbt import PrefectDbtRunner, PrefectDbtSettings

@task(timeout_seconds=600)  # 10-minute timeout
def run_dbt(command: str = "build"):
    runner = PrefectDbtRunner(
        settings=PrefectDbtSettings(
            project_dir="./my_dbt_project",
            profiles_dir="~/.dbt"
        ),
        raise_on_failure=False,
    )
    runner.invoke(command.split())

@flow
def dbt_flow():
    run_dbt("build")

- Integration: prefect-dbt 2) If you must use the CLI (subprocess) - Wrap the dbt call in a task with

timeout_seconds=600

. - Be aware Prefect will send SIGTERM to the process; if dbt or a child ignores it, the task may still hang during cleanup. - To make it more robust: - Start dbt in its own process group (Unix) so you can signal the whole group. - On cancellation, send SIGTERM, then escalate to SIGKILL on the group after a short grace period (e.g., 15–30s). - Ensure you’re streaming stdout/stderr so logs aren’t buffered and “go silent.” For ShellOperation, a minimal pattern would be:

from prefect import flow, task
from prefect_shell import ShellOperation

@task(timeout_seconds=600)
def run_dbt_cli():
    ShellOperation(
        commands=["dbt build --threads 4"],
        env={"DBT_PROFILES_DIR": "~/.dbt"}
    ).run()

@flow
def dbt_flow():
    run_dbt_cli()

- Shell plugin: prefect-shell Note: ShellOperation and general subprocess patterns do not escalate to SIGKILL by default, and child processes can outlive the parent if not in the same process group. That’s why the in-process

PrefectDbtRunner

is strongly recommended. If you can share how you’re running dbt today (code snippet and versions), I can point to the exact change to make this stop hanging and honor your 10-minute timeout.

Thanks for the snippet — this explains the behavior. You’re calling the synchronous

PrefectDbtRunner.invoke(...)

from inside an async task (

async def

). That blocks the event loop, which prevents Prefect’s cooperative cancellation/timeout from being delivered. The result is exactly what you’re seeing: logs go quiet and the run never actually times out or cancels until you intervene. Fix: make the task synchronous (regular def) so Prefect can interrupt it reliably

from pathlib import Path
from typing import Optional
from prefect import task, get_run_logger
from prefect_dbt import PrefectDbtRunner, PrefectDbtSettings

@task(log_prints=True, timeout_seconds=300)  # 5-minute timeout
def run_single_dbt_command(
    command: str,
    project_dir: Path,
    profiles_dir: Path,
    additional_args: Optional[list[str]] = None,
) -> None:
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"Running dbt command: {command}")
    if additional_args:
        <http://logger.info|logger.info>(f"Additional arguments: {' '.join(additional_args)}")

    settings = PrefectDbtSettings(
        project_dir=str(project_dir),
        profiles_dir=str(profiles_dir),
    )
    runner = PrefectDbtRunner(settings=settings, raise_on_failure=False)

    cmd_parts = command.split()
    if additional_args:
        cmd_parts.extend(additional_args)

    full_command = ' '.join(cmd_parts)
    <http://logger.info|logger.info>(f"Executing: dbt {full_command}")
    runner.invoke(cmd_parts)  # synchronous call in a sync task
    <http://logger.info|logger.info>(f"Completed: dbt {full_command}")

Notes - You can keep your flow async or sync — the important part is that the task running dbt is synchronous. If your flow is async, just call the task normally (or submit it) — Prefect will handle it. - With a sync task, Prefect 3’s timeout/cancellation can inject a cancellation into the task’s thread, which will interrupt the in-process dbt run. - Your

raise_on_failure=False

is fine; it doesn’t interfere with timeouts/cancellation. If you truly need the task to remain async, you’d need an async

invoke

(not available in current prefect-dbt) or to restructure so the blocking work doesn’t pin the event loop — but that still won’t give you reliable cancellation for the underlying work. The sync task is the robust path. References - prefect-dbt (PrefectDbtRunner): prefect-dbt If this change doesn’t resolve the hang, share your Prefect version and worker type (process/Docker/K8s), and we can dig deeper.

Thanks for the added details — that pinpoints it. What’s happening - Your task is now sync, but it’s still running inside a worker thread (Docker worker + default task runner for an async flow). In Prefect 3.x, timeouts for sync tasks running in threads are enforced by injecting an exception into the thread. That cannot interrupt blocking C/IO calls (e.g., database drivers used by dbt adapters). The exception is only raised once control returns to Python. - dbt frequently spends time inside adapter/database calls, so if a query hangs, the task won’t see the timeout until the driver returns. That’s why logs stop and the task never actually times out or cancels. How to fix it (pick one or combine) 1) Enforce timeouts at the db/dbt level (most reliable) - Add a per-statement timeout so the database stops the query and returns control to Python before your Prefect timeout. Examples: - Postgres/Redshift (milliseconds): - Profile: include

connect_timeout

- Project/model config pre-hook:

models:
        +pre-hook: "set statement_timeout = 300000"

- Snowflake:

models:
        +pre-hook: "alter session set STATEMENT_TIMEOUT_IN_SECONDS=300"

- BigQuery: - Prefer a model config that sets a per-job timeout if your dbt-bigquery version supports it (e.g.,

job_execution_timeout_seconds

), or implement a pre-hook equivalent where possible. - This is the best way to make sure the underlying blocking call returns and Prefect can enforce its own timeout cleanly. 2) Switch the task runner to a process-based runner (harder stops) - Running the task in its own process lets Prefect terminate the process on timeout, which is more reliable than thread-level exception injection for blocking C/IO. - Try:

from prefect import flow
  from prefect.task_runners import ProcessPoolTaskRunner

  @flow(task_runner=ProcessPoolTaskRunner())
  def my_flow():
      run_single_dbt_command(...)

- Trade-offs: less efficient for many tiny tasks, but far more robust for “must stop” workloads. 3) Last resort: run dbt in a subprocess you control and enforce kill escalation - If you absolutely must hard-enforce time, wrap dbt in a subprocess and kill the process group on timeout (SIGTERM → short wait → SIGKILL). This is stricter than PrefectDbtRunner but guarantees termination even if the driver is stuck. - Simple pattern:

@task(timeout_seconds=300)
  def run_dbt_cli_with_kill(cmd_parts):
      import os, signal, subprocess, time

      # New process group so children get signaled too
      proc = subprocess.Popen(
          ["dbt", *cmd_parts],
          preexec_fn=os.setsid,
          stdout=subprocess.PIPE,
          stderr=subprocess.STDOUT,
          text=True,
          bufsize=1,
      )
      try:
          for line in proc.stdout:
              print(line.rstrip())  # stream logs
          rc = proc.wait(timeout=300)
          if rc != 0:
              raise RuntimeError(f"dbt exited {rc}")
      except subprocess.TimeoutExpired:
          os.killpg(proc.pid, signal.SIGTERM)
          try:
              proc.wait(timeout=15)
          except subprocess.TimeoutExpired:
              os.killpg(proc.pid, signal.SIGKILL)
          raise

- Note: this uses the dbt CLI, not PrefectDbtRunner. 4) Optional: run tasks on the main thread for signal-based timeouts - If you don’t need concurrency, running the flow with a single-threaded task runner (PrefectTaskRunner) can allow signal-based timeouts in the main thread, which are better at interrupting some blocking cases. Still, DB drivers may handle EINTR and continue, so DB-level timeouts are still recommended. Given you’re on prefect-dbt 0.7.9 and Docker workers, I’d recommend: - Add a database-level statement timeout (pre-hook) matching or slightly below your Prefect timeout. - If you still need a hard stop guarantee, switch the flow to

ProcessPoolTaskRunner()

.

If you can share which dbt adapter you’re using (Postgres, Snowflake, BigQuery, etc.), I’ll give you the exact model/profile config to set the timeout. References - prefect-dbt repo: prefect-dbt - Prefect source (timeouts/cancellation internals): Prefect

Got it — thanks for the details. What you’re seeing is expected with PrefectDbtRunner + Snowflake: - dbt emits a NodeStart event when it submits a query, then nothing until the query completes (NodeFinished). Snowflake adapter does not emit mid-query events. So the Prefect UI will go quiet during long-running statements even though the model is still executing in Snowflake. - Your task timeout won’t reliably interrupt a blocking Snowflake call running in a worker thread. The timeout exception can’t be injected until control returns to Python, which doesn’t happen until the database call finishes. This explains both “logs stop” and “timeout is never hit while the model runs.” What to do 1) Enforce a statement timeout in Snowflake (best fix) - Set a Snowflake session parameter so long queries end and return control to Python. Then Prefect can transition the task at or before your 5-minute cap. dbt_project.yml (global)

on-run-start:
  - "alter session set STATEMENT_TIMEOUT_IN_SECONDS=300"
  - "alter session set QUERY_TAG='prefect:{{ env_var('PREFECT_FLOW_RUN_ID') }}'"

Or per-model:

models:
  +pre-hook:
    - "alter session set STATEMENT_TIMEOUT_IN_SECONDS=300"

Also set these in your Snowflake profile for better tracking/robustness: profiles.yml

your_profile:
  target: dev
  outputs:
    dev:
      type: snowflake
      account: ...
      user: ...
      password: ...
      role: ...
      database: ...
      warehouse: ...
      schema: ...
      threads: 4
      client_session_keep_alive: true
      query_tag: "prefect:{{ env_var('PREFECT_FLOW_RUN_ID') }}"

- STATEMENT_TIMEOUT_IN_SECONDS caps per-statement execution time. - QUERY_TAG helps you confirm progress directly in Snowflake while Prefect logs are quiet. - Keep your Prefect task timeout slightly above the Snowflake statement timeout (e.g., db timeout 300s, Prefect task timeout 330–360s) so cleanup/logging can complete. References: - Snowflake statement timeout: STATEMENT_TIMEOUT_IN_SECONDS - dbt Snowflake profile options: dbt-snowflake profile docs - dbt hooks: dbt hooks 2) If you absolutely need a hard-stop at the Prefect layer - Run the flow with a process-based task runner, or run dbt via a subprocess that you can SIGTERM/SIGKILL on timeout. This provides a “kill switch” even if Snowflake is still busy, at the cost of using the CLI instead of the in-process runner. Process-based runner:

from prefect import flow
from prefect.task_runners import ProcessPoolTaskRunner

@flow(task_runner=ProcessPoolTaskRunner())
def my_flow():
    run_single_dbt_command(...)

CLI with kill escalation (for strict enforcement):

@task(timeout_seconds=300)
def run_dbt_cli_with_kill(cmd_parts):
    import os, signal, subprocess

    proc = subprocess.Popen(
        ["dbt", *cmd_parts],
        preexec_fn=os.setsid,  # new process group (Unix)
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
        bufsize=1,
    )
    try:
        for line in proc.stdout:
            print(line.rstrip())
        rc = proc.wait(timeout=300)
        if rc != 0:
            raise RuntimeError(f"dbt exited {rc}")
    except subprocess.TimeoutExpired:
        os.killpg(proc.pid, signal.SIGTERM)
        try:
            proc.wait(timeout=15)
        except subprocess.TimeoutExpired:
            os.killpg(proc.pid, signal.SIGKILL)
        raise

3) Visibility while queries run - Because dbt + Snowflake doesn’t emit mid-query events, it’s normal for Prefect logs to be quiet. Rely on QUERY_TAG in Snowflake to confirm progress during those silent periods.

- If you want periodic “still running” messages in the UI, you can log heartbeats from a separate monitoring task (but note: that won’t affect cancellation of the dbt execution itself; the database/session timeout is what actually returns control). Recap - Silence in Prefect logs mid-query is normal with Snowflake; it doesn’t mean Prefect hung. - Prefect task timeouts won’t interrupt a blocking Snowflake call in a thread; add a Snowflake statement timeout. - If you need a hard cap from Prefect, use a process-based runner or a subprocess with signal escalation. If you share your dbt_project.yml and profiles.yml snippets, I can help you place the hooks and tags precisely for your setup.