# PiTR Purge Compaction Design

**Date:** 2026-05-18  
**Status:** Approved

## Problem

Compaction-accelerated PiTR restores create files under two subdirectories of the log-backup storage prefix:

```
{logbackup.Storage}/v1/compactions/
{logbackup.Storage}/v1/migrations/
```

These files are never cleaned up after restore, causing unbounded storage growth and future billing issues.

## Goal

After a compaction-accelerated PiTR restore completes, automatically delete the `v1/compactions` and `v1/migrations` directories. Non-compaction-accelerated PiTR restores must be skipped.

## Design

### Two-Layer Architecture

The purge work is split across two DAGs to keep concerns separate:

1. **PiTR restore DAG** — adds a thin `ActionDeleteCompactionForPiTR` that calls `compactBackupModel.AsyncDelete()` as a fire-and-forget trigger.
2. **`DagDeleteCompactBackup`** — extended with a purge action and a final-step action to perform the actual storage deletion and mark the entity as `deleted`.

This means a PiTR restore is never blocked by storage deletion failures. If deletion fails, the `CompactBackup` record stays in `deleting` state, which is surfaced via metrics and triggers an alert.

---

### Part 1: PiTR DAG Changes

#### New constant

```go
// app/workflow/constants.go
ActionDeleteCompactionForPiTR = "delete_compaction_for_pitr"
```

#### New action: `delete_compaction.go`

File: `app/workflow/action/restore/pitr/create/delete_compaction.go`

**Dependencies injected:**
- `restoreRepo` — load restore entity
- `compactBackupModel` — call `AsyncDelete`

**Flow:**

1. Parse `restoreID` from params.
2. Load `restore` from DB.
3. **Skip gate:** if `restore.ExtraInfo == nil || restore.ExtraInfo.CompactBackupID == 0`, log "no compaction, skipping" and return `nil`.
4. Call `compactBackupModel.AsyncDelete(ctx, &domain.AsyncDeleteCompactBackupReq{CompactBackupID: restore.ExtraInfo.CompactBackupID})`.
5. If `AsyncDelete` returns an error, log a warning and return `nil` — PiTR restore must not be blocked.

**Idempotency:** `AsyncDelete` creates a DAG instance. If the DAG is already running (duplicate call), `workflowCli.RunDagWithTags` returns a conflict that `AsyncDelete` handles. The action is safe to retry.

#### Updated DAG task graph (`restore_pitr.go`)

```
ActionCheckAndCreateClusterForRestoring
        ↓
ActionInitRestoreConfig   ActionAddBucketPolicyForRestoring
                               ↓
                    ActionRunCompactBackupForPiTR
                               ↓
                    ActionRunPitRestoring
                         ↙           ↘
  ActionRestoreSqlBindings    ActionDeleteCompactionForPiTR  ← NEW (parallel)
                         ↘           ↙
             ActionDeleteClusterJobForRestoring
                               ↓
                    ActionRestorePitBackupFinalStep
```

`ActionDeleteCompactionForPiTR` depends on `ActionRunPitRestoring`. `ActionDeleteClusterJobForRestoring` depends on both `ActionRestoreSqlBindings` and `ActionDeleteCompactionForPiTR`.

Timeout for `ActionDeleteCompactionForPiTR`: 1 minute (it only triggers a DAG, does no heavy work).

---

### Part 2: `DagDeleteCompactBackup` Extended

Current pipeline:
```
ActionCleanupCompactBackup
```

New pipeline:
```
ActionCleanupCompactBackup  (existing — cancels TiKV job, cleans downstream CR)
        ↓
ActionPurgeCompactionStorage  (new — deletes S3/GCS/Azure files)
        ↓
ActionDeleteCompactBackupFinalStep  (new — marks entity deleted)
```

#### New constant

```go
// app/workflow/constants.go
ActionPurgeCompactionStorage         = "purge_compaction_storage"
ActionDeleteCompactBackupFinalStep   = "delete_compact_backup_final_step"
```

#### New action: `ActionPurgeCompactionStorage`

File: `app/workflow/action/backup/compact/purge_compaction_storage.go`

**Dependencies injected:**
- `compactBackupRepo` — load `CompactBackup`
- `logBackupRepo` — load `LogBackup` (for `Storage`, `BucketID`, `TenantID`, `ProjectID`)
- `bucketRepo` — load `Bucket` for credentials
- `encryptor` — decrypt bucket credentials (Azure SAS token)

**Flow:**

1. Read `compactBackupID` from share data.
2. Load `compactBackup`; load `logBackup` via `compactBackup.LogBackupID`.
3. Resolve storage URI using the same logic as `CreateCompactBackupTaskAction.getStorageURI`:
   - If `compactBackup.BucketID == 0`: `storageURI = logBackup.Storage`, credentials from `logBackup.BucketID`.
   - If `compactBackup.BucketID != 0`: load bucket via `compactBackup.BucketID`; `storageURI = bucket.URIInBucket(logBackup.Storage)`.
4. For Azure infra type: convert URL using the same logic as `CleanLogBackupStorageAction.getAzureBlobURI` (parse with `azblob.ParseAzblobURL`, set SAS token, convert to inner URL).
5. Create `blob.Storage` via `blob.NewStorage(ctx, urlStr, opt)`.
6. Call `storage.DeleteSubDir(ctx, "v1/compactions", nil)`.
7. Call `storage.DeleteSubDir(ctx, "v1/migrations", nil)`.
8. **On any error:** return the error — DAG marks action as failed. `CompactBackup` remains in `deleting` state, which triggers an alert via metrics.
9. **On success:** return `nil`.

Timeout: 10 minutes.

#### New action: `ActionDeleteCompactBackupFinalStep`

File: `app/workflow/action/backup/compact/delete_compact_backup_final_step.go`

Mirrors `RestorePitBackupFinalStep` pattern.

**Flow:**
1. Read `compactBackupID` from share data.
2. Load `compactBackup`.
3. If already `deleted`, log and return `nil` (idempotent).
4. Set `Status = CompactBackupStatusDeleted`, set `FinishedAt = now()`, save.

Timeout: 2 minutes.

#### Model change: `AsyncDelete`

Before launching the DAG, `AsyncDelete` transitions `compactBackup.Status` from `success`/`failed` → `deleting` in a transaction. If status is already `deleting` or `deleted`, skip the status update and proceed directly to ensuring the DAG is running (idempotent path):

```go
// in AsyncDelete, inside transaction:
compactBackup.Status = entity.CompactBackupStatusDeleting
if err := c.compactBackupRepo.Save(ctx, compactBackup); err != nil { return err }
// then launch DAG
```

This ensures the record is in `deleting` state from the moment deletion starts.

---

### Part 3: Entity Changes

#### `app/repo/entity/constant.go`

```go
CompactBackupStatusDeleting CompactBackupStatus = "deleting"
CompactBackupStatusDeleted  CompactBackupStatus = "deleted"
```

Update `IsFinishedStatus()`:
```go
func (s CompactBackupStatus) IsFinishedStatus() bool {
    return s == CompactBackupStatusSuccess ||
        s == CompactBackupStatusFailed ||
        s == CompactBackupStatusDeleted
}
```

#### `app/repo/entity/entity.go`

Update GORM ENUM tag:
```go
Status CompactBackupStatus `gorm:"type:ENUM('pending','running','success','failed','deleting','deleted');not null"`
```

#### DB Migration

File: `migrate/record_202605181000_compact_backup_status_deleting.go`

```go
var _ = AddDataflowDBMigrateRecord(
    "202605181000",
    func(db *gorm.DB) error {
        return db.Exec(`ALTER TABLE compact_backups MODIFY COLUMN status ENUM('pending','running','success','failed','deleting','deleted') NOT NULL`).Error
    },
)
```

After migration, run `make generate-entity` to regenerate query-set methods.

---

### Part 4: Metrics

New package: `app/metrics/compact_backup/`

Structure mirrors `app/metrics/backup/`:

```
app/metrics/compact_backup/
├── metric/metric.go         — GaugeVec definition + status int map
├── collector/collector.go   — Prometheus collector (lists non-terminal compactions)
└── hooks/status.go          — AfterSave hook updates metric
```

#### `metric/metric.go`

Labels: `cluster_id`, `compact_backup_id`, `log_backup_id`, `bucket_id`.

Status int map (for Prometheus `_info`-style gauge):
```
pending   → 1
running   → 2
success   → 3
failed    → 4
deleting  → 5
deleted   → 6
```

GaugeVec name: `dataflow_compact_backup_status`  
Help: `"compact backup status: 1-pending, 2-running, 3-success, 4-failed, 5-deleting, 6-deleted"`

`tenant_id` and `project_id` are not included as labels to keep cardinality low; `cluster_id` is sufficient for per-cluster alerting and triage. `log_backup_id` and `bucket_id` are included for cross-referencing compaction records with their parent resources.

#### `collector/collector.go`

Lists `CompactBackup` records where `Status IN (pending, running, deleting)` (all non-terminal, excluding `success`/`failed`/`deleted`). Follows `base.NewMetricsCollector` pattern.

#### `hooks/status.go`

`AfterSave` hook: calls `UpdateCompactBackupStatusMetric`. Schedules metric deletion after `WaitMetricScrapeInterval` (2 minutes) when status reaches `deleted`, `failed`, or `success`.

#### Registration

`app/metrics/register/register.go`:
```go
prometheus.MustRegister(compact_backup_metric.CompactBackupStatus)
prometheus.MustRegister(compact_backup_collector.NewCompactBackupMetricsCollector())
```

The `CompactBackupMetricsHooker` is registered alongside other entity hooks.

---

### Part 5: Alert Rules (Documentation Only)

Alert conditions (to be configured in the monitoring system):

| Alert | Condition | Severity |
|-------|-----------|----------|
| `CompactBackupStuckDeleting` | `dataflow_compact_backup_status == 5` for > 30 minutes | Warning |
| `CompactBackupStuckRunning` | `dataflow_compact_backup_status == 2` for > 2 hours | Warning |

---

## Files Changed / Created

| File | Change |
|------|--------|
| `app/workflow/constants.go` | Add `ActionDeleteCompactionForPiTR`, `ActionPurgeCompactionStorage`, `ActionDeleteCompactBackupFinalStep` |
| `app/workflow/dag/restore/restore_pitr.go` | Add `ActionDeleteCompactionForPiTR` to dag tasks and actions slice |
| `app/workflow/dag/backup/compact/create.go` | Extend `GetDeleteDag` with 2 new tasks + actions |
| `app/workflow/action/restore/pitr/create/delete_compaction.go` | New — thin trigger action |
| `app/workflow/action/backup/compact/purge_compaction_storage.go` | New — storage deletion action |
| `app/workflow/action/backup/compact/delete_compact_backup_final_step.go` | New — mark deleted final step |
| `app/models/backup/impl/compact_backup_model.go` | `AsyncDelete`: set status=deleting before DAG launch |
| `app/repo/entity/constant.go` | Add `deleting`/`deleted` status constants, update `IsFinishedStatus` |
| `app/repo/entity/entity.go` | Extend GORM ENUM |
| `app/repo/entity/autogenerated_entity.go` | Regenerated (via `make generate-entity`) |
| `migrate/record_202605181000_compact_backup_status_deleting.go` | New migration |
| `app/metrics/compact_backup/metric/metric.go` | New |
| `app/metrics/compact_backup/collector/collector.go` | New |
| `app/metrics/compact_backup/hooks/status.go` | New |
| `app/metrics/register/register.go` | Register new collector + gauge |

## Risks

- **ENUM migration**: MySQL `MODIFY COLUMN` on ENUM acquires a metadata lock but does not rewrite rows (additive change). Safe for online deployment.
- **Partial purge**: If `DeleteSubDir("v1/compactions")` succeeds but `DeleteSubDir("v1/migrations")` fails, the action is retried by fastflow. `DeleteSubDir` is idempotent (deletes existing objects, ignores missing). Safe to retry.
- **Concurrent deletes**: `AsyncDelete` sets status=deleting in a transaction before DAG launch. Duplicate PiTR calls on the same compaction will hit the same `CompactBackupID`; the second DAG run will find the entity already `deleted` or `deleting` and be idempotent.