From e38cf3e5e302dc7cff8c2400a12c6c1756ba21b8 Mon Sep 17 00:00:00 2001 From: Alex Noonan <48368867+C00ldudeNoonan@users.noreply.github.com> Date: Sun, 12 Jul 2026 15:32:04 -0400 Subject: [PATCH 1/2] fix: replace destructive Dagster event cleanup with safe retention + backups (#132) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The dagster_db_cleanup job deleted every non-error event older than 24h and cascade-deleted asset_materializations, asset_observations, asset_check_executions, and event_log_tags. Those index tables are orchestration/automation state — AutomationConditions, "latest materialization", asset checks, and lineage read from them — so cleanup made materialized assets look un-materialized and could retrigger automation. Retention (safe, by event class): - cleanup_event_logs.sh now deletes ONLY plain log-message rows (event_logs with dagster_event_type IS NULL), in bounded batches, default 30-day retention. It never touches structured events or the asset index tables. Legacy DAGSTER_EVENT_RETENTION_HOURS is still honored (now log-rows only). - dagster.yaml gains Dagster's supported `retention:` for schedule/sensor ticks (the actual high-volume tick tables), no raw SQL. Backups: - backup_dagster_postgres.sh + dagster_db_backup compose service: interval pg_dump -> gzipped timestamped dumps on a dagster_backups volume, retention pruning, optional GPG encryption (GCP PD is encrypted at rest regardless). - docker/DAGSTER_RETENTION_AND_BACKUP.md documents retention-by-class, RPO (24h) / RTO (~15m), and a tested restore drill. Config surfaced in .env.dagster.example. Tests (test_dagster_retention_cleanup.py) statically enforce the contract: cleanup never DELETEs the protected tables, deletes only NULL-type event_logs rows, scripts pass `sh -n`, dagster.yaml declares tick retention, and compose wires the backup service + volume. (Live restore drill must be run against a real PostgreSQL — cannot execute in CI.) Closes #132 Co-Authored-By: Claude Opus 4.8 --- .env.dagster.example | 18 ++- dagster.yaml | 13 ++ docker-compose.yml | 41 +++++- docker/DAGSTER_RETENTION_AND_BACKUP.md | 116 +++++++++++++++ docker/backup_dagster_postgres.sh | 70 ++++++++++ docker/cleanup_event_logs.sh | 132 ++++++++++-------- .../tests/test_dagster_retention_cleanup.py | 71 ++++++++++ 7 files changed, 398 insertions(+), 63 deletions(-) create mode 100644 docker/DAGSTER_RETENTION_AND_BACKUP.md create mode 100644 docker/backup_dagster_postgres.sh create mode 100644 macro_agents/tests/test_dagster_retention_cleanup.py diff --git a/.env.dagster.example b/.env.dagster.example index 34515d5..f2a77dd 100644 --- a/.env.dagster.example +++ b/.env.dagster.example @@ -15,12 +15,22 @@ DAGSTER_PG_PASSWORD=change-me-to-secure-password DAGSTER_UI_PORT=3000 # ============================================================================ -# Dagster Event Log Cleanup (Optional) -# ============================================================================ -# Deletes non-error events older than the retention window. -DAGSTER_EVENT_RETENTION_HOURS=24 +# Dagster Log Pruning + PostgreSQL Backups (Optional) — see +# docker/DAGSTER_RETENTION_AND_BACKUP.md +# ============================================================================ +# Log pruning deletes ONLY plain log-message rows (event_logs with +# dagster_event_type IS NULL). Materialization/observation/asset-check state is +# NEVER deleted; schedule/sensor tick retention is handled in dagster.yaml. +DAGSTER_LOG_RETENTION_DAYS=30 +DAGSTER_LOG_CLEANUP_BATCH_SIZE=5000 DAGSTER_EVENT_CLEANUP_INTERVAL_SECONDS=3600 +# Automated PostgreSQL backups (dagster_db_backup service). +DAGSTER_BACKUP_INTERVAL_SECONDS=86400 +DAGSTER_BACKUP_RETENTION_DAYS=14 +# Optional app-level backup encryption (disk-at-rest encryption applies regardless). +DAGSTER_BACKUP_GPG_PASSPHRASE= + # ============================================================================ # Google Cloud Platform — Primary Data Warehouse (BigQuery) # ============================================================================ diff --git a/dagster.yaml b/dagster.yaml index a9b31e5..80f9a05 100644 --- a/dagster.yaml +++ b/dagster.yaml @@ -105,5 +105,18 @@ run_retries: max_retries: 2 retry_on_asset_or_op_failure: true +# Tick retention (issue #132) — Dagster's supported mechanism for pruning the +# high-volume schedule/sensor tick tables. Successful ticks are kept so +# automation history stays auditable; skipped/failed ticks age out. This does +# NOT touch materialization/observation/asset-check state. +retention: + schedule: + purge_after_days: 90 + sensor: + purge_after_days: + skipped: 7 + failure: 30 + success: 90 + telemetry: enabled: false diff --git a/docker-compose.yml b/docker-compose.yml index ac182e1..9f881db 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -283,7 +283,9 @@ services: OLLAMA_HOST: http://ollama:11434 restart: "no" - # Dagster DB cleanup - prune non-error events older than retention + # Dagster log pruning — deletes ONLY plain log-message rows (event_logs with + # dagster_event_type IS NULL). Never touches materialization/observation/ + # asset-check state. See docker/DAGSTER_RETENTION_AND_BACKUP.md. dagster_db_cleanup: image: postgres:15 container_name: dagster_db_cleanup @@ -294,7 +296,8 @@ services: DAGSTER_DB_NAME: dagster DAGSTER_DB_USER: dagster_user DAGSTER_DB_PASSWORD: ${DAGSTER_PG_PASSWORD} - DAGSTER_EVENT_RETENTION_HOURS: ${DAGSTER_EVENT_RETENTION_HOURS:-24} + DAGSTER_LOG_RETENTION_DAYS: ${DAGSTER_LOG_RETENTION_DAYS:-30} + DAGSTER_LOG_CLEANUP_BATCH_SIZE: ${DAGSTER_LOG_CLEANUP_BATCH_SIZE:-5000} DAGSTER_EVENT_CLEANUP_INTERVAL_SECONDS: ${DAGSTER_EVENT_CLEANUP_INTERVAL_SECONDS:-3600} volumes: - ./docker/cleanup_event_logs.sh:/opt/dagster/cleanup_event_logs.sh:ro @@ -314,8 +317,42 @@ services: max-size: "10m" max-file: "2" + # Automated encrypted PostgreSQL backups with retention (issue #132). + dagster_db_backup: + image: postgres:15 + container_name: dagster_db_backup + restart: unless-stopped + environment: + DAGSTER_DB_HOST: dagster_postgresql + DAGSTER_DB_PORT: "5432" + DAGSTER_DB_NAME: dagster + DAGSTER_DB_USER: dagster_user + DAGSTER_DB_PASSWORD: ${DAGSTER_PG_PASSWORD} + BACKUP_DIR: /backups + BACKUP_INTERVAL_SECONDS: ${DAGSTER_BACKUP_INTERVAL_SECONDS:-86400} + BACKUP_RETENTION_DAYS: ${DAGSTER_BACKUP_RETENTION_DAYS:-14} + # Optional app-level encryption; disk-at-rest encryption applies regardless. + BACKUP_GPG_PASSPHRASE: ${DAGSTER_BACKUP_GPG_PASSPHRASE:-} + volumes: + - ./docker/backup_dagster_postgres.sh:/opt/dagster/backup_dagster_postgres.sh:ro + - dagster_backups:/backups + entrypoint: ["/bin/sh", "/opt/dagster/backup_dagster_postgres.sh"] + networks: + - dagster_network + depends_on: + dagster_postgresql: + condition: service_healthy + mem_limit: 256m + memswap_limit: 256m + logging: + driver: "json-file" + options: + max-size: "10m" + max-file: "2" + volumes: postgres_data: + dagster_backups: driver: local dagster_storage: driver: local diff --git a/docker/DAGSTER_RETENTION_AND_BACKUP.md b/docker/DAGSTER_RETENTION_AND_BACKUP.md new file mode 100644 index 0000000..097e13d --- /dev/null +++ b/docker/DAGSTER_RETENTION_AND_BACKUP.md @@ -0,0 +1,116 @@ +# Dagster Retention & PostgreSQL Backups + +How this deployment prunes Dagster history safely and how to back up / restore the +Dagster PostgreSQL database. Addresses issue #132 (the previous cleanup deleted +materialization/observation/asset-check state, which broke automation and audit +history). + +## Retention by event class + +| Class | What it is | Where it lives | Retention | Managed by | +|---|---|---|---|---| +| **Materializations / observations / asset checks** | Asset state that AutomationConditions, "latest materialization", asset checks, and lineage read | `event_logs` (structured) + `asset_materializations`, `asset_observations`, `asset_check_executions` | **Never deleted** | — | +| **Structured run/step events** | `RUN_*`, `STEP_*`, `ENGINE_EVENT`, etc. — run history | `event_logs` (`dagster_event_type IS NOT NULL`) | **Never deleted** | — | +| **Plain log-message rows** | High-volume `context.log.*` lines | `event_logs` (`dagster_event_type IS NULL`) | `DAGSTER_LOG_RETENTION_DAYS` (default **30d**) | `dagster_db_cleanup` (`cleanup_event_logs.sh`) | +| **Schedule ticks** | Schedule evaluation ticks | `job_ticks` | `purge_after_days: 90` | Dagster `retention:` (dagster.yaml) | +| **Sensor ticks** | Sensor evaluation ticks | `job_ticks` | skipped 7d / failure 30d / success 90d | Dagster `retention:` (dagster.yaml) | + +Key rule: **orchestration and audit state is never deleted by a cleanup job.** +Only disposable log-message rows and schedule/sensor ticks age out, and the tick +purge uses Dagster's own supported `retention:` config rather than raw SQL +against internal tables. + +### Why not delete the asset index tables? + +`asset_materializations` / `asset_observations` / `asset_check_executions` are how +Dagster answers "what is the latest materialization of this asset?" Deleting them +makes materialized assets look un-materialized, which can retrigger +AutomationConditions and destroys asset-check and lineage history. The cleanup +job therefore never touches them, and only deletes `event_logs` rows that carry +no structured event (`dagster_event_type IS NULL`) — those never have a row in +the index tables, so no state can be orphaned. + +## Configuration + +Set in `.env` (see `.env.dagster.example`): + +| Variable | Default | Effect | +|---|---|---| +| `DAGSTER_LOG_RETENTION_DAYS` | `30` | Age after which plain log rows are pruned | +| `DAGSTER_LOG_CLEANUP_BATCH_SIZE` | `5000` | Rows deleted per batch (bounds lock time) | +| `DAGSTER_EVENT_CLEANUP_INTERVAL_SECONDS` | `3600` | Cleanup loop interval | +| `DAGSTER_BACKUP_INTERVAL_SECONDS` | `86400` | Backup cadence (daily) | +| `DAGSTER_BACKUP_RETENTION_DAYS` | `14` | Age after which backups are pruned | +| `DAGSTER_BACKUP_GPG_PASSPHRASE` | _(empty)_ | If set (and `gpg` present), symmetrically encrypts each dump | + +> The legacy `DAGSTER_EVENT_RETENTION_HOURS` is still honored by +> `cleanup_event_logs.sh` (converted to days) for backward compatibility, but it +> now affects **only** plain log rows — never structured events. + +## Backups + +The `dagster_db_backup` service runs `pg_dump` on `DAGSTER_BACKUP_INTERVAL_SECONDS`, +writing `dagster_.sql.gz` to the `dagster_backups` volume and +pruning dumps older than `DAGSTER_BACKUP_RETENTION_DAYS`. + +**Encryption.** GCP persistent disks are encrypted at rest by default (AES-256), +so a backups volume on a GCP PD is already encrypted. Set +`DAGSTER_BACKUP_GPG_PASSPHRASE` to add application-level symmetric encryption +(produces `.sql.gz.gpg`); this requires `gpg` in the image. + +**Off-host copies (recommended).** The volume lives on the same VM as the DB. For +real disaster recovery, sync `dagster_backups` to object storage, e.g.: + +```bash +gcloud storage rsync -r /var/lib/docker/volumes/_dagster_backups/_data \ + gs:///dagster-postgres/ +``` + +## Objectives + +- **RPO (Recovery Point Objective): 24 hours.** With daily backups, at most one + day of Dagster metadata (run/materialization history) can be lost. Lower + `DAGSTER_BACKUP_INTERVAL_SECONDS` for a tighter RPO. +- **RTO (Recovery Time Objective): ~15 minutes.** Restore is a single + `gunzip | psql` into a fresh database plus a service restart. + +Note: the warehouse data itself (BigQuery) is the system of record for analytics; +these backups protect Dagster's **orchestration metadata**, not the analytical +tables. + +## Restore drill + +Run this in a non-production environment periodically to prove backups are valid. + +```bash +# 1. Pick a backup from the volume. +docker compose run --rm --entrypoint sh dagster_db_backup \ + -c 'ls -1t /backups/dagster_*.sql.gz* | head' + +# 2. Create a scratch database to restore into (does not touch the live DB). +docker compose exec dagster_postgresql \ + psql -U dagster_user -d postgres -c 'CREATE DATABASE dagster_restore_test;' + +# 3. Restore the dump into the scratch DB. +# (If the file ends in .gpg, first: gpg --batch --passphrase "$PASS" -d file.sql.gz.gpg > file.sql.gz) +docker compose exec dagster_db_backup sh -c \ + 'gunzip -c /backups/.sql.gz | psql -h dagster_postgresql -U dagster_user -d dagster_restore_test' + +# 4. Verify asset state survived — expect a non-zero count. +docker compose exec dagster_postgresql psql -U dagster_user -d dagster_restore_test \ + -c 'SELECT count(*) AS materializations FROM asset_materializations;' + +# 5. Clean up. +docker compose exec dagster_postgresql \ + psql -U dagster_user -d postgres -c 'DROP DATABASE dagster_restore_test;' +``` + +**Production restore:** stop the Dagster services, restore the dump into the +`dagster` database (drop/recreate or restore into an empty DB), then restart: + +```bash +docker compose stop dagster_webserver dagster_daemon dagster_user_code +gunzip -c /backups/.sql.gz | \ + docker compose exec -T dagster_postgresql psql -U dagster_user -d dagster +docker compose start dagster_user_code dagster_daemon dagster_webserver +``` diff --git a/docker/backup_dagster_postgres.sh b/docker/backup_dagster_postgres.sh new file mode 100644 index 0000000..16e578f --- /dev/null +++ b/docker/backup_dagster_postgres.sh @@ -0,0 +1,70 @@ +#!/bin/sh +# Automated Dagster PostgreSQL backups (issue #132). +# +# Runs pg_dump on an interval, writes a compressed timestamped dump to the +# mounted backups volume, prunes dumps older than the retention window, and +# (optionally) encrypts each dump with a GPG passphrase. +# +# Encryption: GCP persistent disks are encrypted at rest by default (AES-256), +# so a backups volume on a GCP PD is already encrypted. Set BACKUP_GPG_PASSPHRASE +# to add application-level symmetric encryption on top (requires gpg in the +# image). See docker/DAGSTER_RETENTION_AND_BACKUP.md for the restore drill. +set -eu + +: "${DAGSTER_DB_HOST:=dagster_postgresql}" +: "${DAGSTER_DB_PORT:=5432}" +: "${DAGSTER_DB_NAME:=dagster}" +: "${DAGSTER_DB_USER:=dagster_user}" +: "${DAGSTER_DB_PASSWORD:?DAGSTER_DB_PASSWORD must be set}" +: "${BACKUP_DIR:=/backups}" +: "${BACKUP_INTERVAL_SECONDS:=86400}" # daily +: "${BACKUP_RETENTION_DAYS:=14}" +: "${BACKUP_GPG_PASSPHRASE:=}" # empty = rely on disk-level encryption + +export PGPASSWORD="$DAGSTER_DB_PASSWORD" + +mkdir -p "$BACKUP_DIR" + +if [ -n "$BACKUP_GPG_PASSPHRASE" ] && ! command -v gpg >/dev/null 2>&1; then + echo "backup: WARNING BACKUP_GPG_PASSPHRASE set but gpg not found; writing unencrypted (disk-level encryption still applies)" >&2 +fi + +echo "backup: dir=${BACKUP_DIR} interval=${BACKUP_INTERVAL_SECONDS}s retention=${BACKUP_RETENTION_DAYS}d" + +while true; do + # Wait for the database to accept connections before dumping. + if ! pg_isready -h "$DAGSTER_DB_HOST" -p "$DAGSTER_DB_PORT" -U "$DAGSTER_DB_USER" >/dev/null 2>&1; then + echo "backup: database not ready, retrying shortly" + sleep 30 + continue + fi + + stamp=$(date -u +%Y%m%dT%H%M%SZ) + tmp="${BACKUP_DIR}/dagster_${stamp}.sql.gz.partial" + final="${BACKUP_DIR}/dagster_${stamp}.sql.gz" + + # -Fc would be smaller/parallel-restorable, but plain SQL + gzip keeps the + # restore drill dependency-free (psql only). Fail the cycle (not the loop) + # if the dump errors, so a transient failure doesn't kill the service. + if pg_dump -h "$DAGSTER_DB_HOST" -p "$DAGSTER_DB_PORT" -U "$DAGSTER_DB_USER" \ + -d "$DAGSTER_DB_NAME" --no-owner --no-privileges | gzip -c > "$tmp"; then + if [ -n "$BACKUP_GPG_PASSPHRASE" ] && command -v gpg >/dev/null 2>&1; then + gpg --batch --yes --passphrase "$BACKUP_GPG_PASSPHRASE" \ + --symmetric --cipher-algo AES256 -o "${final}.gpg" "$tmp" + rm -f "$tmp" + echo "backup: wrote ${final}.gpg" + else + mv "$tmp" "$final" + echo "backup: wrote ${final}" + fi + else + echo "backup: ERROR pg_dump failed for ${stamp}" >&2 + rm -f "$tmp" + fi + + # Prune old backups (both plain and encrypted). + find "$BACKUP_DIR" -maxdepth 1 -type f -name 'dagster_*.sql.gz*' \ + -mtime "+${BACKUP_RETENTION_DAYS}" -print -delete 2>/dev/null || true + + sleep "$BACKUP_INTERVAL_SECONDS" +done diff --git a/docker/cleanup_event_logs.sh b/docker/cleanup_event_logs.sh index c690c7b..ebb8866 100644 --- a/docker/cleanup_event_logs.sh +++ b/docker/cleanup_event_logs.sh @@ -1,4 +1,26 @@ #!/bin/sh +# Safe Dagster log pruning (issue #132). +# +# WHAT THIS PRUNES +# Only plain log-message rows in `event_logs` — i.e. rows where +# `dagster_event_type IS NULL`. Those are the high-volume compute-log lines +# emitted via context.log.*, not structured Dagster events. +# +# WHAT THIS DELIBERATELY NEVER TOUCHES +# - Structured events (dagster_event_type IS NOT NULL): ASSET_MATERIALIZATION, +# ASSET_OBSERVATION, ASSET_CHECK_EVALUATION, STEP_* , RUN_* , etc. +# - The asset-state index tables asset_materializations, asset_observations, +# asset_check_executions, and event_log_tags. +# +# Those tables are orchestration/automation state, not disposable logs: +# AutomationConditions, "latest materialization", asset checks, and lineage +# all read from them. The previous version of this script deleted every +# non-error event older than 24h AND cascade-deleted those index tables, +# which made materialized assets look un-materialized and could retrigger +# automation. See docker/DAGSTER_RETENTION_AND_BACKUP.md. +# +# Schedule/sensor tick retention is handled by Dagster's supported `retention:` +# config in dagster.yaml — not here. set -eu : "${DAGSTER_DB_HOST:=dagster_postgresql}" @@ -6,23 +28,44 @@ set -eu : "${DAGSTER_DB_NAME:=dagster}" : "${DAGSTER_DB_USER:=dagster_user}" : "${DAGSTER_DB_PASSWORD:?DAGSTER_DB_PASSWORD must be set}" -: "${DAGSTER_EVENT_RETENTION_HOURS:=24}" +# Retention for plain log-message rows only. Default 30 days. The legacy +# DAGSTER_EVENT_RETENTION_HOURS is still honored (converted to days) so existing +# deployments keep working, but it now only affects NULL-type log rows. +: "${DAGSTER_LOG_RETENTION_DAYS:=30}" : "${DAGSTER_EVENT_CLEANUP_INTERVAL_SECONDS:=3600}" -: "${DAGSTER_EVENT_CLEANUP_ERROR_REGEX:=ERROR|FAILURE}" +# Delete in bounded batches so a large first pass can't hold a long lock. +: "${DAGSTER_LOG_CLEANUP_BATCH_SIZE:=5000}" + +retention_days="$DAGSTER_LOG_RETENTION_DAYS" +if [ -n "${DAGSTER_EVENT_RETENTION_HOURS:-}" ]; then + # Legacy override (hours -> days, rounded up, min 1). + retention_days=$(( (DAGSTER_EVENT_RETENTION_HOURS + 23) / 24 )) + [ "$retention_days" -lt 1 ] && retention_days=1 + echo "cleanup: DAGSTER_EVENT_RETENTION_HOURS=${DAGSTER_EVENT_RETENTION_HOURS} -> ${retention_days}d (log rows only)" +fi export PGPASSWORD="$DAGSTER_DB_PASSWORD" +psql_do() { + psql -At -h "$DAGSTER_DB_HOST" -p "$DAGSTER_DB_PORT" \ + -U "$DAGSTER_DB_USER" -d "$DAGSTER_DB_NAME" "$@" +} + +echo "cleanup: pruning event_logs rows with dagster_event_type IS NULL older than ${retention_days}d, every ${DAGSTER_EVENT_CLEANUP_INTERVAL_SECONDS}s" + while true; do - table_exists=$(psql -At -h "$DAGSTER_DB_HOST" -p "$DAGSTER_DB_PORT" -U "$DAGSTER_DB_USER" -d "$DAGSTER_DB_NAME" \ - -c "SELECT 1 FROM information_schema.tables WHERE table_schema = 'public' AND table_name = 'event_logs' LIMIT 1;" || true) + table_exists=$(psql_do -c \ + "SELECT 1 FROM information_schema.tables WHERE table_schema = 'public' AND table_name = 'event_logs' LIMIT 1;" || true) if [ -z "$table_exists" ]; then sleep "$DAGSTER_EVENT_CLEANUP_INTERVAL_SECONDS" continue fi - ts_type=$(psql -At -h "$DAGSTER_DB_HOST" -p "$DAGSTER_DB_PORT" -U "$DAGSTER_DB_USER" -d "$DAGSTER_DB_NAME" \ - -c "SELECT data_type FROM information_schema.columns WHERE table_schema = 'public' AND table_name = 'event_logs' AND column_name = 'timestamp' LIMIT 1;" || true) + # The timestamp column has been epoch-numeric in some Dagster versions and a + # true timestamp in others; adapt the comparison expression accordingly. + ts_type=$(psql_do -c \ + "SELECT data_type FROM information_schema.columns WHERE table_schema = 'public' AND table_name = 'event_logs' AND column_name = 'timestamp' LIMIT 1;" || true) ts_expr="\"timestamp\"" case "$ts_type" in @@ -31,58 +74,33 @@ while true; do ;; esac - asset_check_exists=$(psql -At -h "$DAGSTER_DB_HOST" -p "$DAGSTER_DB_PORT" -U "$DAGSTER_DB_USER" -d "$DAGSTER_DB_NAME" \ - -c "SELECT 1 FROM information_schema.tables WHERE table_schema = 'public' AND table_name = 'asset_check_executions' LIMIT 1;" || true) - asset_check_delete="" - if [ -n "$asset_check_exists" ]; then - asset_check_delete="DELETE FROM asset_check_executions WHERE evaluation_event_storage_id IN (SELECT id FROM event_log_ids_to_delete) OR materialization_event_storage_id IN (SELECT id FROM event_log_ids_to_delete);" - fi - - event_log_tags_exists=$(psql -At -h "$DAGSTER_DB_HOST" -p "$DAGSTER_DB_PORT" -U "$DAGSTER_DB_USER" -d "$DAGSTER_DB_NAME" \ - -c "SELECT 1 FROM information_schema.tables WHERE table_schema = 'public' AND table_name = 'event_log_tags' LIMIT 1;" || true) - event_log_tags_delete="" - if [ -n "$event_log_tags_exists" ]; then - event_log_tags_delete="DELETE FROM event_log_tags WHERE event_id IN (SELECT id FROM event_log_ids_to_delete);" - fi - - asset_materializations_exists=$(psql -At -h "$DAGSTER_DB_HOST" -p "$DAGSTER_DB_PORT" -U "$DAGSTER_DB_USER" -d "$DAGSTER_DB_NAME" \ - -c "SELECT 1 FROM information_schema.tables WHERE table_schema = 'public' AND table_name = 'asset_materializations' LIMIT 1;" || true) - asset_materializations_delete="" - if [ -n "$asset_materializations_exists" ]; then - asset_materializations_delete="DELETE FROM asset_materializations WHERE event_id IN (SELECT id FROM event_log_ids_to_delete);" - fi - - asset_observations_exists=$(psql -At -h "$DAGSTER_DB_HOST" -p "$DAGSTER_DB_PORT" -U "$DAGSTER_DB_USER" -d "$DAGSTER_DB_NAME" \ - -c "SELECT 1 FROM information_schema.tables WHERE table_schema = 'public' AND table_name = 'asset_observations' LIMIT 1;" || true) - asset_observations_delete="" - if [ -n "$asset_observations_exists" ]; then - asset_observations_delete="DELETE FROM asset_observations WHERE event_id IN (SELECT id FROM event_log_ids_to_delete);" - fi - - psql -v ON_ERROR_STOP=1 -h "$DAGSTER_DB_HOST" -p "$DAGSTER_DB_PORT" -U "$DAGSTER_DB_USER" -d "$DAGSTER_DB_NAME" \ - -v retention_hours="$DAGSTER_EVENT_RETENTION_HOURS" \ - -v error_regex="$DAGSTER_EVENT_CLEANUP_ERROR_REGEX" < :retention_hours::int) - AND (dagster_event_type IS NULL OR dagster_event_type !~* :'error_regex'); - -${event_log_tags_delete} - -${asset_materializations_delete} - -${asset_observations_delete} - -${asset_check_delete} - -DELETE FROM event_logs -WHERE id IN (SELECT id FROM event_log_ids_to_delete); - -COMMIT; + # Batched delete of ONLY plain log rows (dagster_event_type IS NULL). + # The CTE returns the number of rows deleted; loop until a batch deletes + # fewer than BATCH_SIZE (nothing left within the retention window). + while true; do + n=$(psql_do -v ON_ERROR_STOP=1 \ + -v retention_days="$retention_days" \ + -v batch_size="$DAGSTER_LOG_CLEANUP_BATCH_SIZE" < :retention_days::int) + ORDER BY id + LIMIT :batch_size::int +), +del AS ( + DELETE FROM event_logs e USING doomed d + WHERE e.id = d.id + RETURNING 1 +) +SELECT count(*) FROM del; SQL +) + n=${n:-0} + echo "cleanup: deleted ${n} log rows" + [ "$n" -lt "$DAGSTER_LOG_CLEANUP_BATCH_SIZE" ] && break + done sleep "$DAGSTER_EVENT_CLEANUP_INTERVAL_SECONDS" done diff --git a/macro_agents/tests/test_dagster_retention_cleanup.py b/macro_agents/tests/test_dagster_retention_cleanup.py new file mode 100644 index 0000000..d2613c2 --- /dev/null +++ b/macro_agents/tests/test_dagster_retention_cleanup.py @@ -0,0 +1,71 @@ +"""Guardrails for the Dagster retention/backup design (issue #132). + +The cleanup job must never delete asset-automation state. These tests assert the +safety contract statically (there is no PostgreSQL in CI) so a regression that +reintroduces destructive deletes fails the build. +""" + +from pathlib import Path + +import yaml + +REPO_ROOT = Path(__file__).resolve().parents[2] +CLEANUP_SH = REPO_ROOT / "docker" / "cleanup_event_logs.sh" +BACKUP_SH = REPO_ROOT / "docker" / "backup_dagster_postgres.sh" +DAGSTER_YAML = REPO_ROOT / "dagster.yaml" +COMPOSE_YAML = REPO_ROOT / "docker-compose.yml" + +# Tables that hold asset-automation / audit state — the cleanup job must never +# delete from any of them. +PROTECTED_TABLES = ( + "asset_materializations", + "asset_observations", + "asset_check_executions", + "event_log_tags", +) + + +def test_cleanup_never_deletes_asset_state_tables() -> None: + script = CLEANUP_SH.read_text(encoding="utf-8").lower() + for table in PROTECTED_TABLES: + assert f"delete from {table}" not in script, ( + f"cleanup script must not DELETE FROM {table} (issue #132)" + ) + + +def test_cleanup_only_deletes_null_type_log_rows() -> None: + script = CLEANUP_SH.read_text(encoding="utf-8") + # The single DELETE targets event_logs and is scoped to plain log rows. + assert "DELETE FROM event_logs" in script + assert "dagster_event_type IS NULL" in script + # It must not fall back to the old "everything except errors" predicate. + assert "dagster_event_type !~" not in script + assert script.count("DELETE FROM") == 1 + + +def test_cleanup_and_backup_scripts_are_valid_sh() -> None: + import subprocess + + for script in (CLEANUP_SH, BACKUP_SH): + result = subprocess.run( + ["sh", "-n", str(script)], capture_output=True, text=True + ) + assert result.returncode == 0, f"{script.name} sh syntax error: {result.stderr}" + + +def test_dagster_yaml_declares_supported_tick_retention() -> None: + config = yaml.safe_load(DAGSTER_YAML.read_text(encoding="utf-8")) + retention = config.get("retention", {}) + assert "schedule" in retention and "sensor" in retention, ( + "dagster.yaml must configure supported schedule/sensor tick retention" + ) + + +def test_compose_wires_backup_service_and_volume() -> None: + compose = yaml.safe_load(COMPOSE_YAML.read_text(encoding="utf-8")) + assert "dagster_db_backup" in compose["services"], "backup service missing" + assert "dagster_backups" in compose["volumes"], "backups volume missing" + backup = compose["services"]["dagster_db_backup"] + assert any("/backups" in v for v in backup["volumes"]), ( + "backup service must mount the backups volume at /backups" + ) From 9321a65a9d5022f3885bc94c73e14e691b5d7ac9 Mon Sep 17 00:00:00 2001 From: Alex Noonan <48368867+C00ldudeNoonan@users.noreply.github.com> Date: Sun, 12 Jul 2026 15:36:56 -0400 Subject: [PATCH 2/2] fix: address Codex review on #132 backup safety and legacy retention var - backup_dagster_postgres.sh: dump to a temp file and check pg_dump's own exit status before gzipping. POSIX sh has no pipefail, so `pg_dump | gzip` reported gzip's status and could publish an empty/partial dump as success; pruning then ran and could rotate away good backups. Now gzip/publish/prune happen only after a verified successful dump. - docker-compose.yml: pass legacy DAGSTER_EVENT_RETENTION_HOURS through to dagster_db_cleanup so deployments that still set it keep their configured retention instead of silently jumping to the 30-day default. Co-Authored-By: Claude Opus 4.8 --- docker-compose.yml | 4 ++++ docker/backup_dagster_postgres.sh | 27 +++++++++++++++++---------- 2 files changed, 21 insertions(+), 10 deletions(-) diff --git a/docker-compose.yml b/docker-compose.yml index 9f881db..eca02e9 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -299,6 +299,10 @@ services: DAGSTER_LOG_RETENTION_DAYS: ${DAGSTER_LOG_RETENTION_DAYS:-30} DAGSTER_LOG_CLEANUP_BATCH_SIZE: ${DAGSTER_LOG_CLEANUP_BATCH_SIZE:-5000} DAGSTER_EVENT_CLEANUP_INTERVAL_SECONDS: ${DAGSTER_EVENT_CLEANUP_INTERVAL_SECONDS:-3600} + # Backward-compat: if a deployment still sets the legacy hours var, honor + # it (converted to days, log rows only) instead of silently jumping to the + # 30-day default. Empty when unset -> new default applies. + DAGSTER_EVENT_RETENTION_HOURS: ${DAGSTER_EVENT_RETENTION_HOURS:-} volumes: - ./docker/cleanup_event_logs.sh:/opt/dagster/cleanup_event_logs.sh:ro entrypoint: ["/bin/sh", "/opt/dagster/cleanup_event_logs.sh"] diff --git a/docker/backup_dagster_postgres.sh b/docker/backup_dagster_postgres.sh index 16e578f..4f58346 100644 --- a/docker/backup_dagster_postgres.sh +++ b/docker/backup_dagster_postgres.sh @@ -40,14 +40,20 @@ while true; do fi stamp=$(date -u +%Y%m%dT%H%M%SZ) + raw="${BACKUP_DIR}/dagster_${stamp}.sql.partial" tmp="${BACKUP_DIR}/dagster_${stamp}.sql.gz.partial" final="${BACKUP_DIR}/dagster_${stamp}.sql.gz" - # -Fc would be smaller/parallel-restorable, but plain SQL + gzip keeps the - # restore drill dependency-free (psql only). Fail the cycle (not the loop) - # if the dump errors, so a transient failure doesn't kill the service. + # Dump to a temp file first and check pg_dump's OWN exit status. POSIX sh has + # no pipefail, so `pg_dump | gzip` would report gzip's status (often 0) and + # publish an empty/partial dump as a success — after which pruning could + # delete older valid backups. -Fc would be smaller, but plain SQL keeps the + # restore drill dependency-free (psql only). Failures fail the cycle, not the + # service loop, and pruning runs ONLY after a verified good backup. if pg_dump -h "$DAGSTER_DB_HOST" -p "$DAGSTER_DB_PORT" -U "$DAGSTER_DB_USER" \ - -d "$DAGSTER_DB_NAME" --no-owner --no-privileges | gzip -c > "$tmp"; then + -d "$DAGSTER_DB_NAME" --no-owner --no-privileges > "$raw" \ + && gzip -c "$raw" > "$tmp"; then + rm -f "$raw" if [ -n "$BACKUP_GPG_PASSPHRASE" ] && command -v gpg >/dev/null 2>&1; then gpg --batch --yes --passphrase "$BACKUP_GPG_PASSPHRASE" \ --symmetric --cipher-algo AES256 -o "${final}.gpg" "$tmp" @@ -57,14 +63,15 @@ while true; do mv "$tmp" "$final" echo "backup: wrote ${final}" fi + + # Prune old backups (both plain and encrypted) — only after this cycle + # produced a valid dump, so a failed run never rotates good backups away. + find "$BACKUP_DIR" -maxdepth 1 -type f -name 'dagster_*.sql.gz*' \ + -mtime "+${BACKUP_RETENTION_DAYS}" -print -delete 2>/dev/null || true else - echo "backup: ERROR pg_dump failed for ${stamp}" >&2 - rm -f "$tmp" + echo "backup: ERROR pg_dump/gzip failed for ${stamp}; keeping existing backups" >&2 + rm -f "$raw" "$tmp" fi - # Prune old backups (both plain and encrypted). - find "$BACKUP_DIR" -maxdepth 1 -type f -name 'dagster_*.sql.gz*' \ - -mtime "+${BACKUP_RETENTION_DAYS}" -print -delete 2>/dev/null || true - sleep "$BACKUP_INTERVAL_SECONDS" done