pause_flow_run(wait_for_input=...)

410 Gone

pause_flow_run(wait_for_input=str)

Concurrency lease renewal failed - slots are no longer reserved.
...
Lease not found - it may have expired or been revoked
HTTP 410 Gone

pause_flow_run(wait_for_input=...)

pause_flow_run(...)

pause_flow_run(...)

pause_flow_run(wait_for_input=...)

from datetime import date

from prefect import flow, pause_flow_run

from core.logger import get_logger


@flow(flow_run_name="breakdown-production-{customer}-{breakdown_type}")
def breakdown_production(
    customer: str,
    breakdown_type: str,
    reference_date: date | None = None,
    dispatch: bool = False,
):
    logger = get_logger()

    <http://logger.info|logger.info>("Executing data quality")

    try:
        raise Exception("Data quality failed")
    except Exception:
        logger.exception("Data quality check failed.")

        action = pause_flow_run(
            wait_for_input=str,
            timeout=300,
        )

        if action == "go":
            <http://logger.info|logger.info>("User chose to continue")
        else:
            raise ValueError("User chose not to continue")

410 Gone

410 Gone

pause_flow_run(...)

pause_flow_run

pause_flow_run(
    wait_for_input=None,
    timeout=3600,
    poll_interval=10,
    key=None,
)

action = pause_flow_run(wait_for_input=str)

from datetime import date
from uuid import uuid4

from prefect import flow
from core.logger import get_logger


def create_approval_request(customer: str, breakdown_type: str, reference_date: date | None) -> str:
    approval_id = str(uuid4())

    # Store this in your DB / S3 / app backend / ticketing system
    # status = "pending"
    # include enough context for the approver and for the continuation step

    return approval_id


@flow(flow_run_name="breakdown-production-{customer}-{breakdown_type}")
def breakdown_production(
    customer: str,
    breakdown_type: str,
    reference_date: date | None = None,
    dispatch: bool = False,
):
    logger = get_logger()

    <http://logger.info|logger.info>("Executing data quality")

    try:
        # simulate data quality failure
        raise Exception("Data quality failed")

    except Exception:
        logger.exception("Data quality check failed.")

        approval_id = create_approval_request(
            customer=customer,
            breakdown_type=breakdown_type,
            reference_date=reference_date,
        )

        <http://logger.info|logger.info>(f"Approval required. approval_id={approval_id}")

        # End this run cleanly or fail intentionally depending on your business logic
        return {
            "status": "waiting_for_approval",
            "approval_id": approval_id,
        }

from datetime import date

from prefect import flow
from core.logger import get_logger


def load_approval_decision(approval_id: str) -> str:
    # Read from your DB/backend
    # return "go" or "stop"
    return "go"


@flow(flow_run_name="resume-breakdown-{customer}-{breakdown_type}")
def resume_breakdown_production(
    approval_id: str,
    customer: str,
    breakdown_type: str,
    reference_date: date | None = None,
    dispatch: bool = False,
):
    logger = get_logger()

    action = load_approval_decision(approval_id)

    if action == "go":
        <http://logger.info|logger.info>("User chose to continue")
        # continue the rest of the pipeline here
    else:
        raise ValueError("User chose not to continue")

from datetime import date

from prefect import flow, pause_flow_run
from core.logger import get_logger


@flow(flow_run_name="breakdown-production-{customer}-{breakdown_type}")
def breakdown_production(
    customer: str,
    breakdown_type: str,
    reference_date: date | None = None,
    dispatch: bool = False,
):
    logger = get_logger()

    <http://logger.info|logger.info>("Executing data quality")

    try:
        raise Exception("Data quality failed")

    except Exception:
        logger.exception("Data quality check failed.")

        action = pause_flow_run(
            wait_for_input=str,
            timeout=180,
            poll_interval=5,
        )

        if action == "go":
            <http://logger.info|logger.info>("User chose to continue")
        else:
            raise ValueError("User chose not to continue")

timeout=180

poll_interval=5

except Exception:

except:

suspend_flow_run

pause_flow_run

suspend_flow_run

asuspend_flow_run

suspend_flow_run

suspend_flow_run(...)

Then trigger a second deployment when the person approves.

---

*If you insist on keeping the pause in one flow*

Then I would at least use this safer version:

wait_for_input=str

pause_flow_run()

suspend_flow_run()

pause_flow_run()

suspend_flow_run

pause_flow_run

pause_flow_run

pause_flow_run(...)

suspend_flow_run(...)

persist_result=True

from datetime import date

from prefect import flow
from prefect.flow_runs import suspend_flow_run

from core.logger import get_logger


@flow(
    flow_run_name="breakdown-production-{customer}-{breakdown_type}",
    persist_result=True,
)
def breakdown_production(
    customer: str,
    breakdown_type: str,
    reference_date: date | None = None,
    dispatch: bool = False,
):
    logger = get_logger()

    <http://logger.info|logger.info>("Executing data quality")

    try:
        # simulate data quality failure
        raise Exception("Data quality failed")

    except Exception:
        logger.exception("Data quality check failed.")

        action = suspend_flow_run(
            wait_for_input=str,
            timeout=3600,
            key="dq-approval",
        )

        if action == "go":
            <http://logger.info|logger.info>("User chose to continue")
        else:
            raise ValueError("User chose not to continue")

persist_result=True

except Exception:

except:

key="dq-approval"

suspend_flow_run(...)

action = suspend_flow_run(
    wait_for_input=str,
    timeout=3600,
    key="dq-approval",
)

pause_flow_run

suspend_flow_run

pause_flow_run

suspend_flow_run(...)

---

*How to test it*

I’d test in this order:

*Test 1 — minimal suspend*
- deploy the simplified flow
- trigger it
- verify that once it reaches `suspend_flow_run`, the ECS task stops cleanly
- verify the flow run remains resumable in UI

*Test 2 — manual resume with input*
- resume it with `go`
- verify continuation works

*Test 3 — rejection path*
- resume it with any value other than `go`
- verify it fails with your intended business error

*Test 4 — long wait*
- leave it suspended longer than the 4-minute window that was breaking `pause_flow_run`
- confirm there is no lease-renewal crash while waiting

That last test is the important proof.

---

*What I expect relative to your current issue*

With `pause_flow_run`, you saw:

- lease renewal attempts while waiting
- HTTP 410
- runtime cancellation/crash

With `suspend_flow_run`, I would expect that class of failure to go away *while the run is waiting*, because the ECS task should not be sitting there renewing the lease the same way.

---

*One warning about user input type*

You are using:

That is fine for a quick test.

For production, I’d recommend a structured input model instead, so the resume payload is explicit and less error-prone.

Example:

and then:

That tends to be cleaner in real HITL flows.

---

*Recommended next test version*

If you want the smallest diff from your current code, use this:

suspend_flow_run

suspend_flow_run(...)

Suspended(type=PAUSED)

Finished in state Cached(type=COMPLETED)

suspend_flow_run(...)

@flow(persist_result=True)
def my_flow():
    run_big_sql()
    upload_files()
    action = suspend_flow_run(wait_for_input=str)
    publish_results()

---

*Why this structure helps*

`run_data_preparation(...)`
- encapsulates all expensive pre-approval work
- persisted/cached, so resume can reuse it

`evaluate_data_quality(...)`
- clearly separates decision logic

`suspend_flow_run(...)`
- stops infra while waiting for human input

`dispatch_manifest(...)`
- only runs after approval
- avoids re-triggering irreversible side effects before approval

---

*Important operational rule*

Anything before suspend should be one of these:

- *idempotent*, or
- *cached/persisted*, or
- *both*

Anything after suspend should be:
- safe to run exactly once after approval
- ideally isolated into its own task(s)

---

*About caching specifically*

You said:

> the theme now is that i have to cache what i've done till the dataquality checks

Yes, exactly.

But more precisely, you want a mix of:

- *task result persistence*
- *cache policies / cache keys*
- *idempotent external writes*

These are related, but not identical.

---

*Practical guidance for your pre-approval steps*

## If a step reads data / computes data
Use a task with persisted results and caching.

Example:

## If a step writes to S3
Prefer one of these:
- deterministic output path, so rerun overwrites same object safely
- check-if-exists logic
- persist the produced path as the task result

Example:

## If a step writes to a database
Make it idempotent:
- upsert instead of insert
- use a run/business key
- avoid “append duplicate row every retry/resume”

---

*One subtle but important point*

Caching helps when Prefect can recognize the task as already done for the same inputs.

But for *external side effects*, do not rely on caching alone.

For example:
- if a task uploaded a file but its state/result was not available for some reason
- Prefect may rerun the task
- so the operation itself should still be safe

That’s why *idempotency* matters in addition to caching.

---

*Suggested minimal adaptation of your current code*

If you don’t want a big rewrite yet, do this first:

1. move all pre-DQ work into one or more tasks
2. add `persist_result=True` to flow and tasks
3. suspend only after those tasks
4. move post-approval side effects into separate tasks

Minimal pattern:

pause_flow_run

suspend_flow_run