Great question. Assuming you’re on Prefect 3.x, here are practical ways to keep your database from growing unbounded. If you’re on Prefect Cloud, retention is managed by your plan; if you’re self-hosting, you’ll want to put a few controls in place. Key docs: - Database maintenance guide - Settings reference (event retention) - EventPersister service - Using the Prefect API client - REST: Delete flow run and Delete flow run logs Best practices (self-hosted) 1) Configure automatic event retention (built-in) - Events can be pruned automatically by the server. Set the retention period via env var: -

PREFECT_SERVER_EVENTS_RETENTION_PERIOD=P30D

(ISO-8601 duration; default is

P7D

) - TOML path:

server.events.retention_period

- Confirm your current value:

prefect config view | grep EVENTS_RETENTION_PERIOD

- EventPersister tunables (for throughput): batch size, flush interval, delete batch size. See docs for details. - EventPersister 2) Control how much you store - Be intentional with logging: - Prefer INFO/WARN over DEBUG for chatty flows. - Consider not sending all logs to the API for extremely verbose workloads; ship detailed logs to an external system (e.g., CloudWatch, ELK, Loki) and keep summaries in Prefect. - Avoid storing overly large artifacts or results in the DB; use external result storage where appropriate. 3) Periodically prune flow runs, task runs, and logs - There isn’t a built-in TTL for flow runs/logs; plan a recurring cleanup job. You can do this with a Prefect flow using the client API. Example: nightly pruning flow runs older than 90 days (terminal states) and their logs

from datetime import datetime, timedelta, timezone
from prefect import flow
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import FlowRunFilterStartTime, FlowRunFilterStateType, FlowRunFilterState
from prefect.client.schemas import SortDirection
from prefect.client.schemas.sorting import FlowRunSort

@flow
async def prune_old_runs(days=90, limit_per_batch=1000):
    cutoff = datetime.now(timezone.utc) - timedelta(days=days)
    async with get_client() as client:
        # Read runs in terminal states older than cutoff, in batches
        while True:
            runs = await client.read_flow_runs(
                flows=None,
                flow_runs={
                    "start_time": FlowRunFilterStartTime(before_=cutoff),
                    "state": FlowRunFilterState(type=FlowRunFilterStateType(any_=["COMPLETED","FAILED","CANCELLED","CRASHED"])),
                },
                sort=FlowRunSort.BY_START_TIME,  # oldest first
                sort_direction=SortDirection.ASC,
                limit=limit_per_batch,
            )
            if not runs:
                break

            for fr in runs:
                # Optional: delete logs for this run
                try:
                    await client.delete_flow_run_logs(flow_run_id=fr.id)
                except Exception:
                    pass  # logs may already be gone or endpoint not available in your version

                # Delete the flow run (cascades to task runs and states)
                await client.delete_flow_run(flow_run_id=fr.id)

if __name__ == "__main__":
    prune_old_runs()

Notes: - Deletion in batches avoids long locks. Run this during a quiet window. - If you prefer to keep the run metadata but drop logs, just call the logs deletion part.

- You can schedule this pruning flow to run nightly. 4) SQL-based pruning (DBA-friendly, do in batches) - For very large cleanups, SQL can be faster. Test with SELECT first; then DELETE in chunks to avoid long locks. - Example (adapt to your schema):

-- Preview
SELECT count(*) FROM log WHERE timestamp < now() - interval '30 days';

-- Batch delete logs (repeat until 0 rows)
WITH old AS (
  SELECT id FROM log
  WHERE timestamp < now() - interval '30 days'
  LIMIT 50000
)
DELETE FROM log WHERE id IN (SELECT id FROM old);

-- Flow run cleanup (terminal states, older than 90 days from start_time)
WITH old_runs AS (
  SELECT id
  FROM flow_run
  WHERE start_time < now() - interval '90 days'
    AND state_type IN ('COMPLETED','FAILED','CANCELLED','CRASHED')
  LIMIT 10000
)
DELETE FROM flow_run WHERE id IN (SELECT id FROM old_runs);

- Follow up with routine PostgreSQL maintenance (autovacuum tuned appropriately). For heavy fragmentation, consider

pg_repack

instead of

VACUUM FULL

to avoid long locks. 5) Monitor size and growth - Keep an eye on your largest tables (often

event

,

log

,

flow_run

,

task_run

) and index bloat. Track DB size growth and adjust retention accordingly. For Prefect Cloud - Cloud enforces plan-based data retention for runs, logs, and events. You typically don’t need to prune manually; instead, pick a plan that matches your retention needs. If you have specific retention targets, let me know your plan and we can map expectations. Questions to tailor this for you - Are you on Prefect Cloud or self-hosted? If self-hosted, which Postgres version and approximate DB size? - What retention goals do you have for events, logs, and run history (e.g., 7/30/90/365 days)? - How frequently do you want cleanup to run, and do you have a maintenance window? I can provide a ready-to-run pruning flow configured to your retention targets and environment once I have the above details.