Best answer depends on where your Prefect metadata lives: If you're on Prefect Cloud - retention is mostly plan-managed, not something you fully control yourself - historical run/artifact retention depends on your Cloud plan - if you need longer/custom retention behavior, see pricing or contact If you're self-hosting Prefect Server / API - the best way is to use Prefect's built-in database vacuum service for metadata retention - then separately clean up persisted result storage (S3/GCS/local disk), because deleting DB records does not automatically delete stored result files For your goal of keeping only 90 days, I'd recommend: 1. Enable Prefect's DB vacuum for flow runs This is the safest built-in cleanup path for: - flow runs - associated logs - associated artifacts - orphaned records created by deletions Set:

bash
export PREFECT_SERVER_SERVICES_DB_VACUUM_ENABLED="events,flow_runs"
export PREFECT_SERVER_SERVICES_DB_VACUUM_RETENTION_PERIOD=7776000
export PREFECT_SERVER_SERVICES_DB_VACUUM_LOOP_SECONDS=3600

7776000

seconds = 90 days. This is the main built-in retention mechanism for self-hosted Prefect 3.x. 2. Handle result persistence separately If you persist results to: - local filesystem - S3 - GCS - Azure Blob - etc. those files/objects are not automatically removed when old flow runs are deleted from Prefect's database. Best practice: - for S3/GCS/Azure: use bucket/container lifecycle policies to delete objects older than 90 days - for local disk: use a cron job / OS cleanup job Example for local storage:

bash
find ~/.prefect/storage -type f -mtime +90 -delete

3. If you're on Cloud or need custom cleanup behavior You can also run a scheduled cleanup flow that: - queries old terminal flow runs - deletes them via the orchestration API one by one But for self-hosted, the vacuum service is usually better than writing your own deletion loop. Important caveats - deleting flow runs should target terminal runs older than your cutoff - DB cleanup and persisted object cleanup are two separate jobs - direct SQL deletion is possible but not recommended, since it can bypass Prefect's cleanup logic Useful links - Retention docs: Database maintenance - Settings reference: Settings reference - Prefect Cloud app: app.prefect.cloud If you want, I can give you either: 1. a docker-compose/Kubernetes example for setting the 90-day vacuum config, or 2. a Python cleanup flow for manual deletion of old runs.

Yes — a scheduled cleanup flow is a good pattern if you want different retention policies for different users / groups / deployments. A good design is: - query old terminal flow runs - apply your own retention rules - delete matching runs in batches - optionally delete related artifacts - separately clean persisted result files/objects if needed A few important details for Prefect 3.x: -

client.delete_flow_run(flow_run_id)

deletes a single run -

client.bulk_delete_flow_runs(...)

exists, but has a max 50 runs per request - logs related to deleted flow runs are cleaned up asynchronously - artifacts are not automatically guaranteed to be removed the way logs are, so if artifacts matter, explicitly clean them too - persisted results in storage are still separate and need their own cleanup strategy Here’s a solid starting point for a scheduled cleanup flow. ```python from datetime import datetime, timedelta, timezone from typing import Optional from prefect import flow, task, get_run_logger from prefect.client.orchestration import get_client from prefect.client.schemas.filters import ( FlowRunFilter, FlowRunFilterEndTime, FlowRunFilterState, FlowRunFilterStateType, FlowRunFilterTags, ArtifactFilter, ArtifactFilterFlowRunId, ) from prefect.client.schemas.objects import StateType RETENTION_BY_TAG = { "team-a": 30, "team-b": 90, "sandbox": 7, } DEFAULT_RETENTION_DAYS = 90 BATCH_SIZE = 50 # bulk delete limit @task async def delete_artifacts_for_flow_run(flow_run_id): logger = get_run_logger() async with get_client() as client: artifacts = await client.read_artifacts( artifact_filter=ArtifactFilter( flow_run_id=ArtifactFilterFlowRunId(any_=[flow_run_id]) ), limit=1000, ) deleted = 0 for artifact in artifacts: try: await client.delete_artifact(artifact.id) deleted += 1 except Exception as exc: logger.warning( f"Failed to delete artifact {artifact.id} for flow run {flow_run_id}: {exc}" ) return deleted @task async def cleanup_old_runs_for_tag(tag: Optional[str], retention_days: int): logger = get_run_logger() cutoff = datetime.now(timezone.utc) - timedelta(days=retention_days) async with get_client() as client: flow_run_filter = FlowRunFilter( end_time=FlowRunFilterEndTime(before_=cutoff), state=FlowRunFilterState( type=FlowRunFilterStateType( any_=[ StateType.COMPLETED, StateType.FAILED, StateType.CANCELLED, StateType.CRASHED, ] ) ), tags=FlowRunFilterTags(all_=[tag]) if tag else None, ) total_deleted = 0 while True: flow_runs = await client.read_flow_runs( flow_run_filter=flow_run_filter, limit=BATCH_SIZE, ) if not flow_runs: break flow_run_ids = [flow_run.id for flow_run in flow_runs] for flow_run_id in flow_run_ids: await delete_artifacts_for_flow_run.fn(flow_run_id) result = await client.bulk_delete_flow_runs( flow_runs=flow_run_filter, limit=BATCH_SIZE, ) deleted_now = len(result.deleted) total_deleted += deleted_now logger.info( f"Deleted {deleted_now} flow runs for tag={tag!r} older than {retention_days} days" ) if deleted_now == 0: break logger.info( f"Finished cleanup for tag={tag!r}, retention_days={retention_days}, total_deleted={total_deleted}" ) return total_deleted

@flow(name="historical-data-retention") async def historical_data_retention(): logger = get_run_logger() grand_total = 0 for tag, retention_days in RETENTION_BY_TAG.items(): deleted = await cleanup_old_runs_for_tag(tag, retention_days) grand_total += deleted deleted_default = await cleanup_old_runs_for_tag(None, DEFAULT_RETENTION_DAYS) grand_total += deleted_default logger.info(f"Retention flow complete. Total deleted flow runs: {grand_total}")

A couple of notes on this example:

*How to apply “per-user” retention*
Prefect flow runs don’t really have a simple “username retention” primitive you’d usually manage against in the UI. In practice, people usually model this by:
- *tags*
- *deployment*
- *flow name*
- *creator type*
- *workspace separation*

So instead of “user A gets 30 days, user B gets 90 days”, the cleanest implementation is often:
- `team-a` tag => 30 days
- `team-b` tag => 90 days

That tends to be much easier to maintain.

*Be careful with the “default” pass*
In the example above, `tag=None` means “all runs older than default retention”. If you also deleted some tagged runs earlier in the same flow, that’s fine, but you may want to *exclude* specially managed categories from the default pass if you want strict non-overlap.

For example, you might instead:
- explicitly process each policy bucket
- then process only runs with no retention tag

If you want that, I can rewrite the filter logic.

*Scheduling it*
You can deploy this flow on a daily schedule and let it enforce retention continuously.

If you’re deploying from Python in Prefect 3.x, use `flow.serve(...)` or `flow.from_source(...).deploy(...)` depending on your setup.

Example local serving pattern:

```python
from prefect import flow

@flow
async def historical_data_retention():
    ...

if __name__ == "__main__":
    historical_data_retention.serve(
        name="historical-data-retention",
        cron="0 2 * * *",
    )

If you want CLI-based deployment, the supported command is

prefect deploy

. What this does not clean This only cleans Prefect metadata: - flow runs - associated logs - optionally artifacts It does not automatically remove: - result files in S3/GCS/Azure/local storage - any custom files your flows wrote For those, you should add a second cleanup task or use storage lifecycle rules. My recommendation Best practical setup: 1. use a scheduled Prefect cleanup flow for policy-driven run/artifact deletion 2. use storage lifecycle rules for persisted results / blobs 3. keep the built-in DB vacuum enabled too if self-hosted, because it helps with general DB hygiene Useful references: - Self-hosted DB maintenance: database maintenance - Settings reference: settings reference - Deployments: deployments If you want, I can turn this into either: - a tag-based retention flow - a deployment-based retention flow - or a creator/user-mapped retention flow with a config dictionary.

Yes — Prefect's built-in DB vacuum is an application-level cleanup service for self-hosted Prefect. It is not PostgreSQL

VACUUM

. Difference in one sentence: - Prefect DB vacuum = deletes old Prefect data like old flow runs, orphaned logs, orphaned artifacts, old events - Postgres VACUUM = reclaims storage / dead rows after deletes happened So if your question is "how do old flow runs actually get deleted?", Prefect DB vacuum is the thing that does that automatically. How it helps with flow-run deletion If you enable the

flow_runs

vacuum component, Prefect will periodically: 1. delete old terminal top-level flow runs - completed - failed - cancelled - crashed - older than your configured retention period 2. delete orphaned logs - logs whose flow run no longer exists 3. delete orphaned artifacts - artifacts tied to deleted flow runs 4. repair/delete stale artifact collections - if their latest artifact was deleted So for self-hosted retention, this is usually the simplest built-in solution. Important behavior It does not mean "delete everything instantly at exactly 90 days". Instead, it runs in a loop and cleans in batches. That means: - a run becomes eligible once it is older than retention - then the vacuum loop eventually deletes it - associated orphan cleanup happens as part of subsequent cleanup cycles Main settings For a 90-day retention target:

bash
PREFECT_SERVER_SERVICES_DB_VACUUM_ENABLED="events,flow_runs"
PREFECT_SERVER_SERVICES_DB_VACUUM_RETENTION_PERIOD="7776000"
PREFECT_SERVER_SERVICES_DB_VACUUM_LOOP_SECONDS="3600"
PREFECT_SERVER_SERVICES_DB_VACUUM_BATCH_SIZE="200"

Meaning: -

events,flow_runs

enables both event cleanup and flow-run cleanup -

7776000

seconds = 90 days - loop every hour - delete in batches of 200 rows On Kubernetes, where does it run? This is the most important operational point: You need Prefect background services running. In production-ish k8s setups, people often run: - API pods - separate background-services pod(s) The vacuum service runs in the background services process, typically started with:

bash
prefect server services start

If you only run the API with no services, vacuum will not run. So on k8s, check that you have a deployment/pod for Prefect background services and that those env vars are set there. Typical k8s pattern -

prefect-api

deployment: - serves API -

prefect-background

deployment: - runs

prefect server services start

- includes DB vacuum settings Example env section for the background pod:

yaml
env:
  - name: PREFECT_SERVER_SERVICES_DB_VACUUM_ENABLED
    value: "events,flow_runs"
  - name: PREFECT_SERVER_SERVICES_DB_VACUUM_RETENTION_PERIOD
    value: "7776000"
  - name: PREFECT_SERVER_SERVICES_DB_VACUUM_LOOP_SECONDS
    value: "3600"
  - name: PREFECT_SERVER_SERVICES_DB_VACUUM_BATCH_SIZE
    value: "200"

Why this is better than writing your own delete script Built-in vacuum is nice because it already knows how to: - safely identify old flow runs - delete in batches - avoid giant transactions - clean orphaned logs/artifacts - continuously enforce retention So if your retention rule is simply: - "keep 90 days for everybody" then built-in DB vacuum is probably the best choice. If your rule is: - "different retention depending on team/user/tag/deployment" then a scheduled Prefect cleanup flow gives you more control. A common hybrid approach is: - use built-in DB vacuum as the baseline 90-day cleanup - use a custom cleanup flow for exceptions / shorter retention for certain tags or deployments What it does not clean DB vacuum only cleans Prefect database records. It does not automatically delete: - S3 result files - GCS result files - local persisted result files - any custom files written by your flows Those need separate cleanup, usually via: - bucket lifecycle policies - cronjob - or another Prefect flow

How to confirm it's working Check logs from your background services pod for messages like: -

Database vacuum: deleted X old flow runs

-

Database vacuum: deleted Y orphaned logs

-

Database vacuum: deleted Z orphaned artifacts

Useful links - Self-hosted database maintenance: database maintenance - Settings reference: settings reference - Prefect Helm repo: prefect-helm If you want, I can help you with either: 1. a Helm values.yaml example to enable 90-day vacuum on k8s, or 2. a custom retention flow for per-user/per-tag retention on top of that.