fix: replace destructive Dagster event cleanup with safe retention + backups (#132)

.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)
 # ============================================================================

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

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,8 +296,13 @@ 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}
+      # 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"]
@@ -314,8 +321,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

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_<UTC-timestamp>.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/<project>_dagster_backups/_data \
+  gs://<your-backup-bucket>/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/<chosen>.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/<chosen>.sql.gz | \
+  docker compose exec -T dagster_postgresql psql -U dagster_user -d dagster
+docker compose start dagster_user_code dagster_daemon dagster_webserver
+```

docker/backup_dagster_postgres.sh

@@ -0,0 +1,77 @@
+#!/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)
+  raw="${BACKUP_DIR}/dagster_${stamp}.sql.partial"
+  tmp="${BACKUP_DIR}/dagster_${stamp}.sql.gz.partial"
+  final="${BACKUP_DIR}/dagster_${stamp}.sql.gz"
+
+  # 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 > "$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"
+      rm -f "$tmp"
+      echo "backup: wrote ${final}.gpg"
+    else
+      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/gzip failed for ${stamp}; keeping existing backups" >&2
+    rm -f "$raw" "$tmp"
+  fi
+
+  sleep "$BACKUP_INTERVAL_SECONDS"
+done