https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> Pattern for shutting down infra durin...
# ask-marvin
g
@Marvin Pattern for shutting down infra during long external-API polls — what's the idiomatic approach? Prefect 3.6.22, KubernetesWorker on GKE. Parent flow auto_label_flow fans out chunks via a task wrapper that invokes core_batch_flow as a subflow. Each core_batch_flow submits an OpenAI batch, then polls every ~60 min until it completes (jobs run hours to a day). Today the polling is implemented as task retries: @task(retries=24, retry_condition_fn=_retry_on_not_ready) def poll_status_task(client, batch_id): ... poll_status_task.with_options(retry_delay_seconds=3600)(client, batch_id) Confirmed in task_engine.py891 897 retry_delay_seconds is implemented as in-process time.sleep(). The pod stays alive holding 2 CPU + 4G memory reservation while doing nothing for the full hour. GKE Cluster Autoscaler picks up the idle pod for node consolidation → SIGTERM → reschedule. Variable interval (14–48 min observed). Confirmed by pod event deleting pod for node scale down. I want the pod to actually shut down during the wait and a fresh pod to come up when it's time to poll again. I tried to reach for suspend_flow_run(timeout=...) but hit two blockers in source: 1. core_policy.py:1352-1357 — pause_reschedule cannot be applied to subflows (if context.run.parent_task_run_id: abort_transition). My core_batch_flow is a subflow of auto_label_flow. 2. server/services/pause_expirations.py:44 — pause_timeout fires Failed, not auto-resume. There's no built-in "wake up after N seconds" for Suspended. Questions: 1. Is there an idiomatic Prefect 3 pattern for "long external poll, no infra during the wait, auto-resume when due"? Specifically: is there a server-side mechanism I missed for auto-resuming Suspended flows on a schedule? 2. If not, is the recommended path: (a) run_deployment per chunk to make each chunk a top-level run, plus suspend_flow_run plus a Prefect Automation that resumes on a delay, OR (b) decouple entirely — submit-flow writes batch_ids to durable storage, separate cron-scheduled harvester flow polls and downloads? 3. Are there other primitives (events, webhooks, automations of type "schedule resume") that fit this shape better than what I'm reaching for?
m
✅ thought for 272.7 seconds
Great question — and thanks for the precise context. You’ve identified the key constraints correctly. Short answers 1) There’s no built-in “resume this paused run at time T” in Prefect 3.x. Subflows cannot use suspend-with-reschedule, and pause timeouts fail the run (don’t auto-resume). Your reading of the source is spot on. 2) Recommended patterns for “long external poll; no infra during wait”: - Best when you don’t have webhooks: self-rescheduling top-level runs (each run does a single check, then schedules the next run at T and exits). - Best when you do have an external signal: suspend a top-level run and resume it via an event-driven Automation. 3) Other primitives that fit well: events + automations (ResumeFlowRun) if you can get a webhook/timer to emit an event; or a tiny “ticker” flow (cron) that resumes paused runs when due. Tasks/retries aren’t a fit here because their retry delay sleeps in-process. Details and patterns Why subflows won’t work for suspend - Subflows are explicitly blocked from pause_reschedule; orchestration aborts the transition when 
context.run.parent_task_run_id
is set. That’s why 
suspend_flow_run(timeout=...)
won’t do what you want from a subflow. - If you want suspend/resume, the unit must be a top-level flow run with a deployment. Pattern A (recommended here): self-rescheduling “one-check” flow runs - Make 
core_batch_flow
its own deployment. - Each run performs one status check. If not ready, it schedules a new run of the same deployment with 
scheduled_time=now+poll_interval
and exits. No sleeping pod; new pod starts exactly when due. Example:
Copy code
from datetime import datetime, timedelta, timezone
from prefect import flow, get_run_logger
from prefect.deployments import run_deployment

def check_batch_status(batch_id: str) -> str:
    # your OpenAI API call here
    ...

def download_results(batch_id: str):
    ...

@flow
def core_batch_flow(batch_id: str, poll_interval_seconds: int = 3600):
    logger = get_run_logger()
    status = check_batch_status(batch_id)

    if status == "completed":
        download_results(batch_id)
        <http://logger.info|logger.info>("Batch %s completed; results downloaded.", batch_id)
        return

    # Not ready — schedule a new run at the desired time, then exit
    wake_at = datetime.now(timezone.utc) + timedelta(seconds=poll_interval_seconds)
    <http://logger.info|logger.info>("Not ready; scheduling next check at %s", wake_at.isoformat())

    run_deployment(
        name="core-batch-flow/poller",                 # your deployment name
        parameters={"batch_id": batch_id, "poll_interval_seconds": poll_interval_seconds},
        scheduled_time=wake_at,
        as_subflow=False,                              # ensure a top-level run
        flow_run_name=f"batch-{batch_id}-check-{int(wake_at.timestamp())}",
        tags=["batch-poller"],
    )
Then from your parent flow, fan out top-level runs instead of subflows:
Copy code
from prefect import flow
from prefect.deployments import run_deployment

@flow
def auto_label_flow(batch_ids: list[str]):
    for b in batch_ids:
        run_deployment(
            name="core-batch-flow/poller",
            parameters={"batch_id": b},
            as_subflow=False,    # make each chunk its own top-level run
            flow_run_name=f"batch-{b}-start",
            tags=["batch-poller"],
        )
Notes: - This gives you predictable wake-ups driven by Prefect’s scheduler. No infra idling and no in-process sleeps. - Persist anything you need across runs (e.g., attempt counters) in Prefect Variables or external storage. - API reference for 
run_deployment
(notice the 
scheduled_time
and 
as_subflow
args): run_deployment docs Pattern B: suspend + event-driven resume (if you can get a webhook/signal) - Make 
core_batch_flow
a top-level flow (deployment).
- When not ready, call 
suspend_flow_run(timeout=...)
. This immediately releases the pod. - Resume the run when your external system emits a signal — e.g., via a small webhook that sends a Prefect event or calls the ResumeFlowRun action. Skeleton:
Copy code
from prefect import flow
from prefect.flow_runs import suspend_flow_run

@flow
def core_batch_flow(batch_id: str):
    status = check_batch_status(batch_id)
    if status != "completed":
        suspend_flow_run(timeout=86400, pause_key=f"batch:{batch_id}")  # releases infra
        # Execution will resume from here after resume action
    download_results(batch_id)
Then create an Automation that resumes runs when your event arrives (e.g., “openai.batch.ready”): - Actions: 
ResumeFlowRun
- Triggers: 
EventTrigger(expect={"openai.batch.ready"})
matching the appropriate flow-run resource Automation/action docs: - Automations & actions overview: Automations overview - Actions API reference (including ResumeFlowRun): events actions API Note: There isn’t a built-in “delay N minutes then resume” action. If you truly need time-based resume (not event-based), add a tiny “ticker”: - Either a cron-scheduled flow that queries the Prefect API for paused runs with a given 
pause_key
and resumes those that are due, or an external timer (Cloud Scheduler) that calls Prefect’s REST API to resume specific runs. Why not task retries for long waits? - In Prefect 3.x, 
retry_delay_seconds
on tasks is an in-process sleep (as you saw in 
task_engine.py
), which holds pods and confuses autoscaling. Prefer flow-level scheduling/suspension patterns above. - States/docs: States concept Which path should you choose? - If OpenAI can’t notify you when a batch is ready, self-rescheduling top-level runs (Pattern A) is the simplest and most robust for “wake up exactly at T, do a quick check, and go back to sleep”. - If you can wire an external event when a batch is ready, suspend + event-driven resume (Pattern B) minimizes checks and wakes up immediately. References - Interactive flows (pause/suspend) and best practices: Interactive flows - Flow run suspension/resume API: flow_runs API - Automations and actions (ResumeFlowRun, SuspendFlowRun): events actions API One question to help me tailor a concrete recommendation: - Do you have any way to receive a webhook/event from OpenAI when a batch completes? If not, I’ll sketch a complete “self-rescheduling” version for your 
core_batch_flow
including a small state store for idempotency and a parent fan-out pattern using 
run_deployment(as_subflow=False)
.
Open in Slack
PreviousNext
Powered by