Great question. Assuming you’re on Prefect 3.x, the simplest way to “hand off” work to a new flow run (so your current run can end before Cloud Run’s 24h limit) is to trigger a deployment from within your flow, then just return. Two common approaches: 1) Use run_deployment (easy, works in both sync/async flows) - Set as_subflow=False so the new run is fully decoupled from the current one. - Optionally add an idempotency_key to prevent duplicate runs if your flow retries.

from prefect import flow
from prefect.deployments import run_deployment

@flow
def long_job(offset: int = 0):
    # ... do a chunk of work ...

    if more_to_do(offset):
        _ = run_deployment(
            name="long_job/cloud-run",                  # "flow_name/deployment_name"
            parameters={"offset": next_offset(offset)},
            as_subflow=False,                           # decouple from this run
            tags=["continuation"],
            idempotency_key=f"continue-{next_offset(offset)}"
        )
    # end current flow run (no special call needed)
    return "done"

2) Use the Prefect client (more control, async) - Create the next run directly via the API. This is useful if you need to look up the deployment by name or manage runs at a lower level.

from prefect import flow, get_client

@flow
async def long_job(offset: int = 0):
    # ... do a chunk of work ...

    if more_to_do(offset):
        async with get_client() as client:
            dep = await client.read_deployment_by_name("long_job/cloud-run")
            fr = await client.create_flow_run_from_deployment(
                dep.id,
                parameters={"offset": next_offset(offset)},
                tags=["continuation"],
                idempotency_key=f"continue-{next_offset(offset)}"
            )
    # end current flow run
    return "done"

Why this helps with Cloud Run’s 24h limit - You split the work into chunks that each complete well under 24h. - Before hitting the limit, the current run triggers a new run that picks up where it left off (e.g., via an offset/cursor parameter), then exits. Repeat as needed. - For horizontal scaling, you can trigger multiple runs at once (shards/partitions) with different parameters and let your work pool/worker fleet pick them up in parallel. Best practices for this pattern - Persist progress/cursors externally (DB, object storage, etc.) or pass them forward as parameters so the next run knows where to resume. - Use idempotency_key to avoid duplicate triggers during retries. - Ensure your work pool has enough workers/replicas to pick up the new runs for actual parallelism. Docs - Run deployments from code: Run deployments (Prefect 3.x) - Deployments overview: Deployments (concepts) If you can share: - The deployment name you’re triggering - Whether you’re retriggering the same deployment or a different one - Your worker/work-pool setup on Cloud Run …I can tailor the code and scaling guidance to your exact setup.