feat(retention): chain-aware retention and pruning

- Add a pure retention engine in internal/dumps (GroupChains + Plan, no
  I/O or clock): the catalog is grouped into restorable chains (a base
  plus its incrementals) and kept or pruned as a unit, so an incremental
  is never orphaned from its base. A chain's age is its newest member,
  so an actively-appended chain is never pruned mid-life.
- Support three composable rules with union semantics (a chain is kept
  if it satisfies any active rule, so a rule can only ever protect):
  keep_last N, max_age duration, and GFS daily/weekly/monthly. An
  all-zero policy keeps everything — prune is a no-op until configured.
- Add app.Prune: scopes to a profile, plans over chains, and on --apply
  deletes leaf-inward (incrementals before base) so an interrupted prune
  leaves at worst a complete shorter chain. Per-dump failures are
  collected, not fatal; the run reports them and exits non-zero.
- Rewire siphon prune to the engine: --profile, --keep-last, --max-age,
  --gfs-daily/weekly/monthly, dry-run by default, --apply to delete.
- Add a retention config block (defaults + per-profile override that
  replaces wholesale) with fail-fast validation; precedence is CLI flags
  > profile > defaults > keep-everything.
- Document in docs/RETENTION.md; update README and CHANGELOG. Unit-test
  the engine (GFS bucketing, union, chain grouping, keep-everything),
  the executor (leaf-inward order, scoping, collected failures), and the
  config (validation, override precedence).

Author: nixrajput

CHANGELOG.md

@@ -9,6 +9,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Phase G (ops) — **retention & lifecycle** (second G cycle):
+  - Chain-aware retention engine in `internal/dumps` (`GroupChains` + `Plan`, pure and DB/clock-free): the catalog is grouped into restorable **chains** (a base plus its incrementals), and retention keeps or prunes each chain as a unit, so an incremental is never orphaned from its base. A chain's age is its newest member, so an actively-appended chain is never pruned mid-life.
+  - Three composable policy rules with **union** semantics (a chain is kept if it satisfies any active rule, so adding a rule only ever protects more): `keep_last: N`, `max_age: <duration>`, and grandfather-father-son `gfs: {daily, weekly, monthly}`. An all-zero/omitted policy keeps everything — prune is a no-op unless a rule is explicitly configured.
+  - `siphon prune` rewired to the engine: `--profile` scoping, `--keep-last` / `--max-age` / `--gfs-daily|weekly|monthly` flags, dry-run by default with `--apply` to delete. Deletion is leaf-inward (incrementals before base), so an interrupted prune leaves at worst a complete shorter chain. Per-dump failures are collected and reported; the run exits non-zero on any failure but still prunes the rest. Works over any storage backend via the existing `Store.Delete`.
+  - New `retention:` config block (`defaults` + per-profile override that replaces the default wholesale; `keep_last`, `max_age`, `gfs`) with fail-fast validation. Resolution precedence: CLI flags > profile > defaults > keep-everything. See [docs/RETENTION.md](docs/RETENTION.md).
+
 - Phase G (ops) — **cloud storage backends** (first G cycle):
   - New `internal/storage` leaf defining a backend-neutral `Store` interface (`Put`/`Get`/`Delete`/`List`/`Stat`, all context-aware) with two implementations: **local** (the previous filesystem layout, now behind the interface — temp-write+rename for atomic publish, keys map verbatim to file names so existing catalogs need no migration) and **s3** (S3 and S3-compatible services like MinIO/R2 via AWS SDK v2; streaming transfer-manager upload that is atomic on completion, path-style for custom endpoints).
   - `internal/dumps.Catalog` refactored to hold a `Store` and address dumps by key (`<id>.dump`, `<id>.meta.json`) instead of filesystem paths; `Path`/`MetaPath`/`Root` removed. Catalog methods (`PutDump`/`OpenDump`/`ReadMeta`/`WriteMeta`/`List`/`Delete`/`ResolveChain`/`PruneDryRun`) are now context-aware. `backup`, `restore`, `verify`, and `dumps` stream through the store; SHA-256 integrity is computed app-side over `envelope ++ body`, so a dump verifies identically whether it lives locally or in S3.

README.md

@@ -55,7 +55,7 @@ A single binary that turns the painful, error-prone sprawl of `pg_dump` → `pg_
 | **D** — Driver hardening          | Shared cross-driver test harness (`RunDriverSuite`), capability-gating helper (`RequireCapability`), Postgres connection-probe retry, and a `docs/DRIVERS.md` contributor guide                                                                                                                                                                                                                                                         | ✅ Complete |
 | **E** — MySQL + MariaDB           | Both drivers via a shared `_mysqlcommon` package (shared `Conn`, `mysqldump`/`mariadb-dump` backup, client-pipe restore), exercised by the Phase D `RunDriverSuite` harness                                                                                                                                                                                                                                                             | ✅ Complete |
 | **F** — Advanced transfer         | All four advanced-transfer modes work end-to-end: bounded-buffer streaming sync; **incremental** backup/restore (`backup --incremental --base <id>` captures a bounded change set via Postgres logical decoding / MySQL-MariaDB binlog, `restore` replays the base→incremental chain, Postgres orphan-slot sweep); **cross-engine** sync (`sync --cross-engine` — typed `SchemaInspector` introspection → canonical type-mapping, e.g. Postgres → MySQL); and **CDC** (`siphon cdc` / `sync --continuous` — unbounded change streaming with snapshot→stream handoff, resumable, same- and cross-engine). Live DB paths are integration-tested in CI — see [docs/INCREMENTAL.md](docs/INCREMENTAL.md), [docs/CROSS_ENGINE.md](docs/CROSS_ENGINE.md), [docs/CDC.md](docs/CDC.md) | ✅ Complete |
-| **G** — Ops features              | **Cloud storage** ships first: the dump catalog can live in an S3 / S3-compatible bucket via a pluggable `storage.Store` backend (local + S3 today; GCS/Azure are a fast-follow), selected by a `storage:` config block, with SHA-256 integrity preserved end-to-end — see [docs/STORAGE.md](docs/STORAGE.md). Secret backends, profile groups + 2FA, team mode, audit log, retention, and telemetry follow in later G cycles.                                                                                                                                                                                                                                                                                                              | 🟡 In progress |
+| **G** — Ops features              | **Cloud storage** (✅): the dump catalog can live in an S3 / S3-compatible bucket via a pluggable `storage.Store` backend (local + S3; GCS/Azure are a fast-follow), `storage:` config block, SHA-256 integrity end-to-end — see [docs/STORAGE.md](docs/STORAGE.md). **Retention** (✅): chain-aware pruning (`siphon prune`) with keep-last-N / max-age / GFS rules, per-profile `retention:` config, dry-run by default — never orphans an incremental's base; see [docs/RETENTION.md](docs/RETENTION.md). Secret backends, profile groups + 2FA, team mode, audit log, and telemetry follow in later G cycles.                                                                                                                                                                                                                                                              | 🟡 In progress |
 | **H** — Distribution              | GoReleaser, Homebrew tap, Scoop bucket, install script, docs site                                                                                                                                                                                                                                                                                                                                                                       | ⏳ Planned  |
 
 ## Requirements
@@ -181,6 +181,18 @@ storage:
 
 Credentials are resolved from the standard AWS chain (env vars, `~/.aws`, instance role) — never stored in the config file. See [docs/STORAGE.md](docs/STORAGE.md) for details.
 
+A `retention:` block (default + optional per-profile override) drives `siphon prune`, which deletes old backups as whole chains so an incremental is never orphaned:
+
+```yaml
+defaults:
+  retention:
+    keep_last: 7
+    max_age: 720h            # 30 days
+    gfs: { daily: 7, weekly: 4, monthly: 6 }
+```
+
+`siphon prune` is dry-run by default; pass `--apply` to delete. Flags override the configured policy per run. See [docs/RETENTION.md](docs/RETENTION.md) for details.
+
 ## Architecture
 
 siphon is a strictly layered Go application; imports flow **downward only**, enforced at lint time by `golangci-lint`'s `depguard` (an upward import fails CI):

docs/RETENTION.md

@@ -0,0 +1,94 @@
+# Retention & pruning
+
+`siphon prune` applies a retention policy to the dump catalog, deleting old
+backups while guaranteeing it never orphans an incremental from its base. It
+works against any storage backend (local or S3), since it prunes through the
+same `Store.Delete` the catalog already uses.
+
+## Table of contents
+
+- [The chain is the unit](#the-chain-is-the-unit)
+- [Policy rules](#policy-rules)
+- [Configuration](#configuration)
+- [The CLI](#the-cli)
+- [Safety properties](#safety-properties)
+
+## The chain is the unit
+
+An incremental dump depends on its base (and any intermediate incrementals) to
+restore. Retention therefore operates on **chains**, not individual dumps: a base
+plus its incrementals is kept or deleted as a whole. A full backup is a
+single-member chain. Counting and bucketing are done over chains, so "keep the
+last 7" means seven restorable backups, not seven files.
+
+A chain's age is the timestamp of its **newest** member, so a chain that is
+still being appended to (an old base with a fresh incremental) is treated as
+young and is never pruned mid-life.
+
+## Policy rules
+
+Three rules, each independently enableable. A chain is kept if it satisfies
+**any** active rule (union) — adding a rule can only protect more chains, never
+fewer, so a policy change can't silently delete data a rule meant to keep.
+
+| Rule | Meaning |
+| --- | --- |
+| `keep_last: N` | keep the N newest chains |
+| `max_age: <duration>` | keep chains younger than the duration (Go duration string, e.g. `720h`) |
+| `gfs: {daily, weekly, monthly}` | grandfather-father-son: keep the newest chain in each of the most-recent N days / ISO weeks / months |
+
+An **all-zero / omitted policy keeps everything** — prune becomes a no-op. The
+destructive direction always requires explicit configuration.
+
+## Configuration
+
+Retention lives in a `retention:` block. Set a default for all profiles, and
+optionally override per profile (the profile block **replaces** the default
+block wholesale — it is not field-merged).
+
+```yaml
+defaults:
+  retention:
+    keep_last: 7
+    max_age: 720h            # 30 days
+    gfs: { daily: 7, weekly: 4, monthly: 6 }
+
+profiles:
+  prod:
+    driver: postgres
+    # ...
+    retention:               # prod keeps more, independent of the default
+      keep_last: 30
+      gfs: { daily: 14, weekly: 8, monthly: 12 }
+```
+
+Precedence, highest first: **CLI flags → profile block → defaults block →
+built-in (keep everything)**. An invalid block (negative count, unparseable
+duration) fails fast at config load.
+
+## The CLI
+
+```bash
+# Dry-run (default): show which chains the policy would prune, for one profile.
+siphon prune --profile prod
+
+# Override the configured policy for this run, then actually delete.
+siphon prune --profile prod --keep-last 14 --apply
+
+# Pure flag-driven policy (no config), GFS only.
+siphon prune --gfs-daily 7 --gfs-weekly 4 --gfs-monthly 6 --apply
+```
+
+`--apply` performs deletions; without it, prune only prints the plan. Flags
+(`--keep-last`, `--max-age`, `--gfs-daily/-weekly/-monthly`) override the
+config-derived policy, but only when explicitly set — an unset flag never zeroes
+out a configured rule.
+
+## Safety properties
+
+- **Dry-run by default** — deletion requires an explicit `--apply`.
+- **Keep-everything on empty** — a missing or all-zero policy prunes nothing.
+- **Union semantics** — every rule can only protect; none can cause a surprise deletion.
+- **Chain-atomic** — a base is only deleted when its whole chain is pruned, so an incremental is never left unrestorable.
+- **Leaf-inward deletion** — within a pruned chain, incrementals are deleted before the base, so an interrupted prune (Ctrl-C, network blip) leaves at worst a complete shorter chain, never a base missing under a surviving incremental.
+- **Collected failures** — a single dump that fails to delete is reported and the run exits non-zero, but the remaining chains are still pruned.

internal/app/prune.go

@@ -0,0 +1,104 @@
+package app
+
+import (
+	"context"
+	"time"
+
+	"github.com/nixrajput/siphon/internal/dumps"
+)
+
+// PruneOpts configures a prune run. Policy is resolved by the caller (the CLI
+// maps config + flags into it), so this layer stays config-agnostic. Profile
+// scopes the catalog to one profile's dumps ("" = all profiles). Apply performs
+// deletions; otherwise the run is a dry-run that only computes the plan.
+type PruneOpts struct {
+	Profile string
+	Policy  dumps.RetentionPolicy
+	Apply   bool
+}
+
+// ChainOutcome is one chain's place in the plan, flattened for reporting.
+type ChainOutcome struct {
+	Root      string
+	DumpIDs   []string
+	SizeBytes int64
+	Pruned    bool     // true = scheduled for (or performed) deletion
+	Deleted   []string // dump IDs actually deleted (Apply only)
+	Errors    []string // per-dump deletion failures (Apply only)
+}
+
+// PruneResult is the structured outcome the CLI renders.
+type PruneResult struct {
+	Profile   string
+	Apply     bool
+	Outcomes  []ChainOutcome
+	Reclaimed int64 // bytes freed (sum of successfully deleted dumps; Apply only)
+	Failed    int   // count of dumps that failed to delete
+}
+
+// Prune groups the catalog into chains, plans retention over them, and (when
+// Apply) deletes the pruned chains. It is synchronous (like Verify): prune is a
+// list + a few deletes, not a long stream, so it returns a structured result
+// directly rather than over the job channel.
+//
+// Deletion is chain-aware and leaf-inward: within a pruned chain, incrementals
+// are removed before the base, so an interrupted prune leaves at worst a
+// complete shorter chain — never a base missing under a surviving incremental.
+func Prune(ctx context.Context, d Deps, opt PruneOpts) (*PruneResult, error) {
+	all, err := d.Dumps.List(ctx)
+	if err != nil {
+		return nil, err
+	}
+	// Scope to the requested profile before grouping, so chains and the plan
+	// reflect only this profile's backups.
+	if opt.Profile != "" {
+		filtered := make([]dumps.Meta, 0, len(all))
+		for _, m := range all {
+			if m.Profile == opt.Profile {
+				filtered = append(filtered, m)
+			}
+		}
+		all = filtered
+	}
+
+	plan := dumps.Plan(dumps.GroupChains(all), opt.Policy, time.Now())
+
+	result := &PruneResult{Profile: opt.Profile, Apply: opt.Apply}
+	for _, ch := range plan.Keep {
+		result.Outcomes = append(result.Outcomes, chainOutcome(ch, false))
+	}
+	for _, ch := range plan.Prune {
+		oc := chainOutcome(ch, true)
+		if opt.Apply {
+			applyChainDeletion(ctx, d, ch, &oc, result)
+		}
+		result.Outcomes = append(result.Outcomes, oc)
+	}
+	return result, nil
+}
+
+// applyChainDeletion deletes a pruned chain's members leaf-inward (incrementals
+// before the base). A per-dump failure is collected and does not abort the rest
+// of the chain or the run — a single stuck object should not block reclaiming
+// the others.
+func applyChainDeletion(ctx context.Context, d Deps, ch dumps.Chain, oc *ChainOutcome, result *PruneResult) {
+	for i := len(ch.Members) - 1; i >= 0; i-- {
+		m := ch.Members[i]
+		if err := d.Dumps.Delete(ctx, m.ID); err != nil {
+			oc.Errors = append(oc.Errors, m.ID+": "+err.Error())
+			result.Failed++
+			continue
+		}
+		oc.Deleted = append(oc.Deleted, m.ID)
+		result.Reclaimed += m.SizeBytes
+	}
+}
+
+func chainOutcome(ch dumps.Chain, pruned bool) ChainOutcome {
+	oc := ChainOutcome{Root: ch.Root, Pruned: pruned}
+	for _, m := range ch.Members {
+		oc.DumpIDs = append(oc.DumpIDs, m.ID)
+		oc.SizeBytes += m.SizeBytes
+	}
+	return oc
+}