From ce22097b6ffbb285b308ecf27277c39001fb374e Mon Sep 17 00:00:00 2001 From: Nikhil Rajput Date: Thu, 25 Jun 2026 11:09:06 +0530 Subject: [PATCH 1/2] feat(retention): chain-aware retention and pruning MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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). --- CHANGELOG.md | 6 + README.md | 14 +- docs/RETENTION.md | 94 +++++++++++++ internal/app/prune.go | 104 ++++++++++++++ internal/app/prune_test.go | 221 ++++++++++++++++++++++++++++++ internal/cli/dumps.go | 127 +++++++++++++++-- internal/config/config.go | 86 ++++++++++-- internal/config/retention_test.go | 53 +++++++ internal/dumps/prune.go | 28 ++-- internal/dumps/retention.go | 183 +++++++++++++++++++++++++ internal/dumps/retention_test.go | 182 ++++++++++++++++++++++++ 11 files changed, 1054 insertions(+), 44 deletions(-) create mode 100644 docs/RETENTION.md create mode 100644 internal/app/prune.go create mode 100644 internal/app/prune_test.go create mode 100644 internal/config/retention_test.go create mode 100644 internal/dumps/retention.go create mode 100644 internal/dumps/retention_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index c232c4d..b4c757b 100644 --- a/CHANGELOG.md +++ b/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: `, 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 (`.dump`, `.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. diff --git a/README.md b/README.md index 8c12ede..5852587 100644 --- a/README.md +++ b/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 ` 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): diff --git a/docs/RETENTION.md b/docs/RETENTION.md new file mode 100644 index 0000000..96166a0 --- /dev/null +++ b/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: ` | 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. diff --git a/internal/app/prune.go b/internal/app/prune.go new file mode 100644 index 0000000..36d1287 --- /dev/null +++ b/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 +} diff --git a/internal/app/prune_test.go b/internal/app/prune_test.go new file mode 100644 index 0000000..9fef882 --- /dev/null +++ b/internal/app/prune_test.go @@ -0,0 +1,221 @@ +package app + +import ( + "bytes" + "context" + "errors" + "io" + "strings" + "testing" + "time" + + "github.com/nixrajput/siphon/internal/dumps" + "github.com/nixrajput/siphon/internal/jobs" +) + +// recordingStore is an in-memory storage.Store that records the order of Delete +// calls, so prune's leaf-inward deletion order can be asserted. A key in +// failKeys returns an error from Delete (to test collected-failure handling). +type recordingStore struct { + objects map[string][]byte + deletes []string // keys passed to Delete, in call order + failKeys map[string]bool +} + +func newRecordingStore() *recordingStore { + return &recordingStore{objects: map[string][]byte{}, failKeys: map[string]bool{}} +} + +func (s *recordingStore) Put(_ context.Context, key string, r io.Reader) error { + b, err := io.ReadAll(r) + if err != nil { + return err + } + s.objects[key] = b + return nil +} + +func (s *recordingStore) Get(_ context.Context, key string) (io.ReadCloser, error) { + b, ok := s.objects[key] + if !ok { + return nil, errors.New("not found") + } + return io.NopCloser(bytes.NewReader(b)), nil +} + +func (s *recordingStore) Delete(_ context.Context, key string) error { + s.deletes = append(s.deletes, key) + if s.failKeys[key] { + return errors.New("simulated delete failure") + } + delete(s.objects, key) + return nil +} + +func (s *recordingStore) List(_ context.Context) ([]string, error) { + keys := make([]string, 0, len(s.objects)) + for k := range s.objects { + keys = append(keys, k) + } + return keys, nil +} + +func (s *recordingStore) Stat(_ context.Context, key string) (int64, bool, error) { + b, ok := s.objects[key] + return int64(len(b)), ok, nil +} + +// seedDump writes a dump body + meta directly into the catalog over the store. +func seedDump(t *testing.T, cat *dumps.Catalog, id, profile string, created time.Time, baseID, parentID string) { + t.Helper() + ctx := context.Background() + if err := cat.PutDump(ctx, id, strings.NewReader("body-"+id)); err != nil { + t.Fatalf("PutDump %s: %v", id, err) + } + if err := cat.WriteMeta(ctx, &dumps.Meta{ + ID: id, Profile: profile, Driver: "fake", Created: created, + BaseID: baseID, ParentID: parentID, SizeBytes: 10, + }); err != nil { + t.Fatalf("WriteMeta %s: %v", id, err) + } +} + +func pruneDeps(store *recordingStore) Deps { + return Deps{Dumps: dumps.New(store), Runner: jobs.NewRunner()} +} + +func TestPrune_DryRunDeletesNothing(t *testing.T) { + store := newRecordingStore() + cat := dumps.New(store) + now := time.Now() + seedDump(t, cat, "new", "p", now, "", "") + seedDump(t, cat, "old", "p", now.AddDate(0, 0, -40), "", "") + + res, err := Prune(context.Background(), pruneDeps(store), PruneOpts{ + Policy: dumps.RetentionPolicy{KeepLast: 1}, Apply: false, + }) + if err != nil { + t.Fatalf("Prune: %v", err) + } + if len(store.deletes) != 0 { + t.Errorf("dry-run performed %d deletes, want 0: %v", len(store.deletes), store.deletes) + } + // The plan still reports one chain as pruned. + var prunedChains int + for _, oc := range res.Outcomes { + if oc.Pruned { + prunedChains++ + } + } + if prunedChains != 1 { + t.Errorf("plan pruned %d chains, want 1", prunedChains) + } +} + +func TestPrune_ApplyDeletesLeafInward(t *testing.T) { + store := newRecordingStore() + cat := dumps.New(store) + now := time.Now() + // Keep this fresh single chain. + seedDump(t, cat, "keep", "p", now, "", "") + // Prune this old base + 2 incrementals; deletion must be inc-before-base. + seedDump(t, cat, "base", "p", now.AddDate(0, 0, -40), "base", "") + seedDump(t, cat, "inc1", "p", now.AddDate(0, 0, -39), "base", "base") + seedDump(t, cat, "inc2", "p", now.AddDate(0, 0, -38), "base", "inc1") + + res, err := Prune(context.Background(), pruneDeps(store), PruneOpts{ + Policy: dumps.RetentionPolicy{KeepLast: 1}, Apply: true, + }) + if err != nil { + t.Fatalf("Prune: %v", err) + } + if res.Failed != 0 { + t.Fatalf("Failed = %d, want 0", res.Failed) + } + // Order: the chain's members delete newest-first, i.e. inc2.dump, inc2.meta, + // inc1.dump, inc1.meta, base.dump, base.meta. The base's keys must come AFTER + // both incrementals' keys. + lastBase := lastIndexWithPrefix(store.deletes, "base") + firstInc := firstIndexWithPrefix(store.deletes, "inc") + if lastBase < 0 || firstInc < 0 { + t.Fatalf("missing deletes: %v", store.deletes) + } + if !lastBeforeFirst(store.deletes, "inc2", "inc1") || !lastBeforeFirst(store.deletes, "inc1", "base") { + t.Errorf("deletion order not leaf-inward: %v", store.deletes) + } + // "keep" must be untouched. + for _, k := range store.deletes { + if strings.HasPrefix(k, "keep") { + t.Errorf("kept chain's dump was deleted: %s", k) + } + } +} + +func TestPrune_ProfileScope(t *testing.T) { + store := newRecordingStore() + cat := dumps.New(store) + old := time.Now().AddDate(0, 0, -40) + seedDump(t, cat, "p1old", "p1", old, "", "") + seedDump(t, cat, "p2old", "p2", old, "", "") + + // Prune p1 only, keep-last 0 + max-age tiny so the old chain is pruned. + _, err := Prune(context.Background(), pruneDeps(store), PruneOpts{ + Profile: "p1", Policy: dumps.RetentionPolicy{MaxAge: time.Hour}, Apply: true, + }) + if err != nil { + t.Fatalf("Prune: %v", err) + } + for _, k := range store.deletes { + if strings.HasPrefix(k, "p2") { + t.Errorf("prune of profile p1 deleted a p2 dump: %s", k) + } + } + if firstIndexWithPrefix(store.deletes, "p1old") < 0 { + t.Errorf("p1's old chain was not pruned: %v", store.deletes) + } +} + +func TestPrune_CollectsDeletionFailures(t *testing.T) { + store := newRecordingStore() + cat := dumps.New(store) + old := time.Now().AddDate(0, 0, -40) + seedDump(t, cat, "old", "p", old, "", "") + store.failKeys["old.dump"] = true // the body delete will fail + + res, err := Prune(context.Background(), pruneDeps(store), PruneOpts{ + Policy: dumps.RetentionPolicy{MaxAge: time.Hour}, Apply: true, + }) + if err != nil { + t.Fatalf("Prune returned error, want collected-failure result: %v", err) + } + if res.Failed != 1 { + t.Errorf("Failed = %d, want 1", res.Failed) + } +} + +// helpers + +func firstIndexWithPrefix(s []string, p string) int { + for i, v := range s { + if strings.HasPrefix(v, p) { + return i + } + } + return -1 +} + +func lastIndexWithPrefix(s []string, p string) int { + idx := -1 + for i, v := range s { + if strings.HasPrefix(v, p) { + idx = i + } + } + return idx +} + +// lastBeforeFirst reports whether the last delete of prefix a precedes the first +// delete of prefix b — i.e. all of a's keys come before any of b's. +func lastBeforeFirst(s []string, a, b string) bool { + return lastIndexWithPrefix(s, a) < firstIndexWithPrefix(s, b) +} diff --git a/internal/cli/dumps.go b/internal/cli/dumps.go index 6d7d565..002f662 100644 --- a/internal/cli/dumps.go +++ b/internal/cli/dumps.go @@ -2,11 +2,15 @@ package cli import ( "fmt" + "io" "time" "github.com/spf13/cobra" + "github.com/nixrajput/siphon/internal/app" + "github.com/nixrajput/siphon/internal/config" "github.com/nixrajput/siphon/internal/dumps" + "github.com/nixrajput/siphon/internal/errs" ) func newDumpsCmd() *cobra.Command { @@ -56,34 +60,127 @@ func dumpsInspectCmd() *cobra.Command { } func dumpsPruneCmd() *cobra.Command { - var maxAge time.Duration - var apply bool + var ( + profileName string + keepLast int + maxAge time.Duration + gfsDaily, gfsWeekly int + gfsMonthly int + apply bool + keepLastSet, maxAgeSet bool + gfsDSet, gfsWSet, gfsMSet bool + ) cmd := &cobra.Command{ - Use: "prune", Short: "Show or apply retention policy", + Use: "prune", Short: "Apply a retention policy, pruning whole dump chains", + Long: "prune groups the catalog into restorable chains (a base plus its " + + "incrementals) and keeps or deletes each chain as a unit, so an " + + "incremental is never orphaned from its base. The policy comes from the " + + "config retention block (per-profile override, else defaults); the flags " + + "below override it per run. Dry-run by default — pass --apply to delete.", RunE: func(c *cobra.Command, _ []string) error { + cfg, err := config.Load() + if err != nil { + return err + } + policy := resolveRetentionPolicy(cfg, profileName) + // CLI flags override the config-derived policy, but only when set, so + // an unset flag does not zero out a configured rule. + if keepLastSet { + policy.KeepLast = keepLast + } + if maxAgeSet { + policy.MaxAge = maxAge + } + if gfsDSet { + policy.GFS.Daily = gfsDaily + } + if gfsWSet { + policy.GFS.Weekly = gfsWeekly + } + if gfsMSet { + policy.GFS.Monthly = gfsMonthly + } + deps, err := buildDeps() if err != nil { return err } - rep, err := deps.Dumps.PruneDryRun(c.Context(), dumps.RetentionPolicy{MaxAge: maxAge}) + res, err := app.Prune(c.Context(), deps, app.PruneOpts{ + Profile: profileName, Policy: policy, Apply: apply, + }) if err != nil { return err } - for _, m := range rep.Would { - if apply { - if delErr := deps.Dumps.Delete(c.Context(), m.ID); delErr != nil { - _, _ = fmt.Fprintf(c.OutOrStdout(), " ! failed to delete %s: %v\n", m.ID, delErr) - continue - } - _, _ = fmt.Fprintf(c.OutOrStdout(), " ✗ deleted %s\n", m.ID) - } else { - _, _ = fmt.Fprintf(c.OutOrStdout(), " would delete %s\n", m.ID) - } + renderPruneResult(c.OutOrStdout(), res) + if res.Failed > 0 { + return &errs.Error{Op: "prune", Code: errs.CodeSystem, Hint: "some dumps could not be deleted"} } return nil }, } - cmd.Flags().DurationVar(&maxAge, "max-age", 0, "Delete dumps older than this duration") + cmd.Flags().StringVar(&profileName, "profile", "", "Limit pruning to one profile's dumps") + cmd.Flags().IntVar(&keepLast, "keep-last", 0, "Keep the N newest chains") + cmd.Flags().DurationVar(&maxAge, "max-age", 0, "Keep chains younger than this duration") + cmd.Flags().IntVar(&gfsDaily, "gfs-daily", 0, "GFS: keep newest chain in each of N recent days") + cmd.Flags().IntVar(&gfsWeekly, "gfs-weekly", 0, "GFS: keep newest chain in each of N recent ISO weeks") + cmd.Flags().IntVar(&gfsMonthly, "gfs-monthly", 0, "GFS: keep newest chain in each of N recent months") cmd.Flags().BoolVar(&apply, "apply", false, "Actually delete (default is dry-run)") + // Record which override flags were explicitly set so unset flags don't clobber config. + cmd.PreRun = func(c *cobra.Command, _ []string) { + keepLastSet = c.Flags().Changed("keep-last") + maxAgeSet = c.Flags().Changed("max-age") + gfsDSet = c.Flags().Changed("gfs-daily") + gfsWSet = c.Flags().Changed("gfs-weekly") + gfsMSet = c.Flags().Changed("gfs-monthly") + } return cmd } + +// resolveRetentionPolicy maps a profile's effective config retention block into +// a dumps.RetentionPolicy. A nil block (none configured) yields the zero policy +// — keep everything. The map duration is pre-validated at config load, so the +// parse here cannot fail in practice; a parse error degrades to "no max-age". +func resolveRetentionPolicy(cfg *config.Config, profile string) dumps.RetentionPolicy { + rc := cfg.EffectiveRetention(profile) + if rc == nil { + return dumps.RetentionPolicy{} + } + var maxAge time.Duration + if rc.MaxAge != "" { + maxAge, _ = time.ParseDuration(rc.MaxAge) + } + return dumps.RetentionPolicy{ + KeepLast: rc.KeepLast, + MaxAge: maxAge, + GFS: dumps.GFSPolicy{Daily: rc.GFS.Daily, Weekly: rc.GFS.Weekly, Monthly: rc.GFS.Monthly}, + } +} + +// renderPruneResult prints the plan: kept chains, then pruned chains with their +// member dumps and (on apply) deletion outcome. +func renderPruneResult(w io.Writer, res *app.PruneResult) { + verb := "would prune" + if res.Apply { + verb = "pruned" + } + var prunedChains int + for _, oc := range res.Outcomes { + if !oc.Pruned { + continue + } + prunedChains++ + _, _ = fmt.Fprintf(w, "%s chain %s (%d dump(s), %d bytes)\n", verb, oc.Root, len(oc.DumpIDs), oc.SizeBytes) + for _, e := range oc.Errors { + _, _ = fmt.Fprintf(w, " ! %s\n", e) + } + } + if prunedChains == 0 { + _, _ = fmt.Fprintln(w, "nothing to prune under the current policy") + return + } + if res.Apply { + _, _ = fmt.Fprintf(w, "reclaimed %d bytes; %d deletion failure(s)\n", res.Reclaimed, res.Failed) + } else { + _, _ = fmt.Fprintf(w, "%d chain(s) would be pruned (dry-run; pass --apply to delete)\n", prunedChains) + } +} diff --git a/internal/config/config.go b/internal/config/config.go index 1f62397..198cf6a 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -6,6 +6,7 @@ import ( "fmt" "os" "strings" + "time" "gopkg.in/yaml.v3" ) @@ -19,10 +20,50 @@ type Config struct { } type Defaults struct { - DumpDir string `yaml:"dump_dir"` - Jobs int `yaml:"jobs"` - Compression int `yaml:"compression"` - SecretBackend string `yaml:"secret_backend"` + DumpDir string `yaml:"dump_dir"` + Jobs int `yaml:"jobs"` + Compression int `yaml:"compression"` + SecretBackend string `yaml:"secret_backend"` + Retention *RetentionConfig `yaml:"retention,omitempty"` +} + +// RetentionConfig is the YAML shape of a retention policy. It lives in config +// (not dumps) so config stays a leaf; the app layer maps it to a +// dumps.RetentionPolicy. A nil pointer (block omitted) means "keep everything". +// A profile's retention block REPLACES the defaults block wholesale — it is not +// field-merged — so a profile's policy is read as a single coherent rule set. +type RetentionConfig struct { + KeepLast int `yaml:"keep_last,omitempty"` // keep the N newest chains + MaxAge string `yaml:"max_age,omitempty"` // Go duration string, e.g. "720h" + GFS GFSConfig `yaml:"gfs,omitempty"` // grandfather-father-son tiers +} + +// GFSConfig is the YAML shape of a grandfather-father-son rule. +type GFSConfig struct { + Daily int `yaml:"daily,omitempty"` + Weekly int `yaml:"weekly,omitempty"` + Monthly int `yaml:"monthly,omitempty"` +} + +// Validate rejects nonsensical retention settings (negative counts, an +// unparseable duration). An all-zero policy is valid and means "keep +// everything", so an empty or misconfigured block can never wipe the catalog. +func (r *RetentionConfig) Validate() error { + if r == nil { + return nil + } + if r.KeepLast < 0 { + return fmt.Errorf("retention.keep_last must be >= 0, got %d", r.KeepLast) + } + if r.GFS.Daily < 0 || r.GFS.Weekly < 0 || r.GFS.Monthly < 0 { + return errors.New("retention.gfs tiers must be >= 0") + } + if r.MaxAge != "" { + if _, err := time.ParseDuration(r.MaxAge); err != nil { + return fmt.Errorf("retention.max_age %q is not a valid duration: %w", r.MaxAge, err) + } + } + return nil } // StorageConfig selects where the dump catalog physically lives. Type "local" @@ -60,15 +101,16 @@ func (s StorageConfig) Validate() error { // Name is NOT read from YAML — it is populated by Load() from the map key // in Config.Profiles so callers don't need to thread the name separately. type ProfileConfig struct { - Name string `yaml:"-"` - Driver string `yaml:"driver"` - Host string `yaml:"host"` - Port int `yaml:"port"` - User string `yaml:"user"` - Password string `yaml:"password"` // may be a SecretRef like ${VAR} or keychain://... - Database string `yaml:"database"` - SSLMode string `yaml:"sslmode"` - Group string `yaml:"group"` + Name string `yaml:"-"` + Driver string `yaml:"driver"` + Host string `yaml:"host"` + Port int `yaml:"port"` + User string `yaml:"user"` + Password string `yaml:"password"` // may be a SecretRef like ${VAR} or keychain://... + Database string `yaml:"database"` + SSLMode string `yaml:"sslmode"` + Group string `yaml:"group"` + Retention *RetentionConfig `yaml:"retention,omitempty"` // overrides Defaults.Retention wholesale } type GroupConfig struct { @@ -116,10 +158,28 @@ func Load() (*Config, error) { if err := cfg.Storage.Validate(); err != nil { return nil, fmt.Errorf("config storage: %w", err) } + if err := cfg.Defaults.Retention.Validate(); err != nil { + return nil, fmt.Errorf("config defaults.retention: %w", err) + } + for name, p := range cfg.Profiles { + if err := p.Retention.Validate(); err != nil { + return nil, fmt.Errorf("config profile %q retention: %w", name, err) + } + } return cfg, nil } +// EffectiveRetention returns the RetentionConfig for a profile: its own block if +// present, else the defaults block, else nil (keep everything). The profile +// block replaces the defaults wholesale — it is not merged. +func (c *Config) EffectiveRetention(profile string) *RetentionConfig { + if p, ok := c.Profiles[profile]; ok && p.Retention != nil { + return p.Retention + } + return c.Defaults.Retention +} + // Save writes cfg to the configured Path, creating directories as needed. // Note: Name is yaml:"-" so it is not serialized — the map key is the // canonical source of a profile's name on disk. diff --git a/internal/config/retention_test.go b/internal/config/retention_test.go new file mode 100644 index 0000000..54c834e --- /dev/null +++ b/internal/config/retention_test.go @@ -0,0 +1,53 @@ +package config + +import "testing" + +func TestRetentionConfig_Validate(t *testing.T) { + cases := []struct { + name string + cfg *RetentionConfig + wantErr bool + }{ + {"nil keeps everything", nil, false}, + {"all-zero is valid", &RetentionConfig{}, false}, + {"valid full policy", &RetentionConfig{KeepLast: 7, MaxAge: "720h", GFS: GFSConfig{Daily: 7, Weekly: 4, Monthly: 6}}, false}, + {"negative keep_last", &RetentionConfig{KeepLast: -1}, true}, + {"negative gfs tier", &RetentionConfig{GFS: GFSConfig{Weekly: -2}}, true}, + {"bad duration", &RetentionConfig{MaxAge: "not-a-duration"}, true}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if err := tc.cfg.Validate(); (err != nil) != tc.wantErr { + t.Errorf("Validate() error = %v, wantErr %v", err, tc.wantErr) + } + }) + } +} + +func TestEffectiveRetention_Precedence(t *testing.T) { + def := &RetentionConfig{KeepLast: 7} + prof := &RetentionConfig{KeepLast: 30} + cfg := &Config{ + Defaults: Defaults{Retention: def}, + Profiles: map[string]ProfileConfig{ + "prod": {Name: "prod", Retention: prof}, + "staging": {Name: "staging"}, // no override + }, + } + + if got := cfg.EffectiveRetention("prod"); got == nil || got.KeepLast != 30 { + t.Errorf("prod effective = %+v, want profile override keep_last=30", got) + } + if got := cfg.EffectiveRetention("staging"); got == nil || got.KeepLast != 7 { + t.Errorf("staging effective = %+v, want defaults keep_last=7", got) + } + if got := cfg.EffectiveRetention("unknown"); got == nil || got.KeepLast != 7 { + t.Errorf("unknown-profile effective = %+v, want defaults", got) + } + + // No defaults block at all → nil (keep everything). + empty := &Config{Profiles: map[string]ProfileConfig{}} + if got := empty.EffectiveRetention("x"); got != nil { + t.Errorf("no config effective = %+v, want nil (keep everything)", got) + } +} diff --git a/internal/dumps/prune.go b/internal/dumps/prune.go index 5260f74..3c36a1e 100644 --- a/internal/dumps/prune.go +++ b/internal/dumps/prune.go @@ -5,33 +5,31 @@ import ( "time" ) -// RetentionPolicy is filled in by Phase G. Phase B ships only a dry-run prune -// that lists what *would* be deleted given a max-age policy. -type RetentionPolicy struct { - MaxAge time.Duration -} - -// PruneReport contains the outcome of a dry-run prune operation. +// PruneReport contains the outcome of a dry-run prune operation, flattened to +// individual dumps (chain members of pruned chains are listed in Would, members +// of kept chains in Kept). type PruneReport struct { Would []Meta Kept []Meta } // PruneDryRun returns which dumps would be deleted under policy p without -// actually deleting anything. +// deleting anything. It groups the catalog into chains, runs the chain-aware +// retention engine, and flattens the plan back to dumps. Chain-aware: a base is +// only in Would when its whole chain is pruned, so an incremental is never +// orphaned. func (c *Catalog) PruneDryRun(ctx context.Context, p RetentionPolicy) (PruneReport, error) { all, err := c.List(ctx) if err != nil { return PruneReport{}, err } - now := time.Now() + plan := Plan(GroupChains(all), p, time.Now()) var report PruneReport - for _, m := range all { - if p.MaxAge > 0 && now.Sub(m.Created) > p.MaxAge { - report.Would = append(report.Would, m) - } else { - report.Kept = append(report.Kept, m) - } + for _, ch := range plan.Prune { + report.Would = append(report.Would, ch.Members...) + } + for _, ch := range plan.Keep { + report.Kept = append(report.Kept, ch.Members...) } return report, nil } diff --git a/internal/dumps/retention.go b/internal/dumps/retention.go new file mode 100644 index 0000000..061f2a2 --- /dev/null +++ b/internal/dumps/retention.go @@ -0,0 +1,183 @@ +package dumps + +import ( + "fmt" + "sort" + "time" +) + +// GFSPolicy is a grandfather-father-son retention rule: keep the newest chain +// in each of the most-recent Daily calendar days, Weekly ISO weeks, and Monthly +// calendar months. A zero field disables that tier; an all-zero GFSPolicy is off. +type GFSPolicy struct { + Daily int + Weekly int + Monthly int +} + +func (g GFSPolicy) active() bool { return g.Daily > 0 || g.Weekly > 0 || g.Monthly > 0 } + +// RetentionPolicy decides which dump chains to keep. A chain is kept if it +// satisfies ANY active rule (union semantics): adding a rule can only ever +// protect more chains, never fewer, so a misconfiguration cannot silently +// delete data a rule meant to keep. An all-zero policy keeps everything (prune +// is a no-op) — the dangerous "delete everything" direction requires explicit +// configuration, never silence. +type RetentionPolicy struct { + KeepLast int // keep the N newest chains (0 = rule off) + MaxAge time.Duration // keep chains younger than this (0 = rule off) + GFS GFSPolicy // keep-by-calendar-bucket (all-zero = off) +} + +// IsEmpty reports whether no rule is active, i.e. the policy keeps everything. +func (p RetentionPolicy) IsEmpty() bool { + return p.KeepLast <= 0 && p.MaxAge <= 0 && !p.GFS.active() +} + +// Chain is a restorable unit: a base dump plus its ordered incrementals. The +// chain is the unit of retention — it is kept or pruned as a whole, so an +// incremental can never be orphaned from its base. +type Chain struct { + Root string // root dump ID (the base), the chain's stable key + Members []Meta // base first, then incrementals in apply order +} + +// Age timestamp: a chain is "as young as" its NEWEST member, not its base. An +// actively-appended chain (old base, fresh incremental) is therefore treated as +// young and is never pruned mid-life by a max-age or GFS rule. +func (c Chain) newest() time.Time { + t := c.Members[0].Created + for _, m := range c.Members[1:] { + if m.Created.After(t) { + t = m.Created + } + } + return t +} + +// RetentionPlan is the engine's decision: which chains to keep and which to +// prune, with no side effects. +type RetentionPlan struct { + Keep []Chain + Prune []Chain +} + +// GroupChains folds a flat dump list into chains keyed by root BaseID. A full +// backup (BaseID == ID, or legacy empty BaseID) is a singleton chain; its +// incrementals attach to it via their BaseID. A dump whose root is missing from +// the set anchors its own chain rather than being dropped, so a partially +// present catalog never loses entries silently. Within each chain, members are +// ordered by Created (base first). +func GroupChains(dumps []Meta) []Chain { + byRoot := map[string][]Meta{} + for _, m := range dumps { + root := m.BaseID + if root == "" { + root = m.ID // legacy full backup: its own root + } + byRoot[root] = append(byRoot[root], m) + } + chains := make([]Chain, 0, len(byRoot)) + for root, members := range byRoot { + sort.SliceStable(members, func(i, j int) bool { + return members[i].Created.Before(members[j].Created) + }) + chains = append(chains, Chain{Root: root, Members: members}) + } + // Deterministic order: newest chain (by newest member) first. + sort.SliceStable(chains, func(i, j int) bool { + return chains[i].newest().After(chains[j].newest()) + }) + return chains +} + +// Plan decides which chains to keep vs prune under p, as of now. It is pure: no +// I/O, no clock — now is injected — so every rule and edge case is unit-testable +// with synthetic fixtures. An empty policy keeps everything. +func Plan(chains []Chain, p RetentionPolicy, now time.Time) RetentionPlan { + if p.IsEmpty() { + return RetentionPlan{Keep: append([]Chain(nil), chains...)} + } + + // chains is already sorted newest-first by GroupChains; copy defensively. + ordered := append([]Chain(nil), chains...) + sort.SliceStable(ordered, func(i, j int) bool { + return ordered[i].newest().After(ordered[j].newest()) + }) + + keep := map[string]bool{} // union of every active rule's keep set + + if p.KeepLast > 0 { + for i := 0; i < len(ordered) && i < p.KeepLast; i++ { + keep[ordered[i].Root] = true + } + } + if p.MaxAge > 0 { + for _, c := range ordered { + if now.Sub(c.newest()) < p.MaxAge { + keep[c.Root] = true + } + } + } + if p.GFS.active() { + for _, root := range gfsKeep(ordered, p.GFS) { + keep[root] = true + } + } + + var plan RetentionPlan + for _, c := range ordered { + if keep[c.Root] { + plan.Keep = append(plan.Keep, c) + } else { + plan.Prune = append(plan.Prune, c) + } + } + return plan +} + +// gfsKeep returns the roots of chains retained by the GFS rule: the newest chain +// in each of the most-recent Daily days, Weekly ISO weeks, and Monthly months. +// `ordered` must be newest-first. A chain that is the newest in more than one +// tier (e.g. today's daily AND this week's weekly) is simply listed once by the +// caller's set. +func gfsKeep(ordered []Chain, g GFSPolicy) []string { + var roots []string + // For each tier, walk newest→oldest assigning chains to calendar buckets; + // the FIRST chain seen for a bucket (newest, since input is newest-first) is + // that bucket's representative. Keep representatives of the most-recent N + // distinct buckets. + pick := func(limit int, bucketKey func(time.Time) string) { + if limit <= 0 { + return + } + seen := map[string]bool{} + kept := 0 + for _, c := range ordered { + k := bucketKey(c.newest()) + if seen[k] { + continue // older chain in an already-represented bucket + } + seen[k] = true + roots = append(roots, c.Root) + kept++ + if kept >= limit { + return + } + } + } + + pick(g.Daily, func(t time.Time) string { + y, m, d := t.Date() + return fmt.Sprintf("%04d-%02d-%02d", y, int(m), d) + }) + pick(g.Weekly, func(t time.Time) string { + y, w := t.ISOWeek() + return fmt.Sprintf("%04d-W%02d", y, w) + }) + pick(g.Monthly, func(t time.Time) string { + y, m, _ := t.Date() + return fmt.Sprintf("%04d-%02d", y, int(m)) + }) + return roots +} diff --git a/internal/dumps/retention_test.go b/internal/dumps/retention_test.go new file mode 100644 index 0000000..923674c --- /dev/null +++ b/internal/dumps/retention_test.go @@ -0,0 +1,182 @@ +package dumps + +import ( + "sort" + "testing" + "time" +) + +// fixed reference "now" so age/GFS math is deterministic. +var refNow = time.Date(2026, 3, 15, 12, 0, 0, 0, time.UTC) + +// dump builds a Meta with an ID, profile, creation time, and optional chain links. +func dump(id, profile string, created time.Time, baseID, parentID string) Meta { + return Meta{ID: id, Profile: profile, Created: created, BaseID: baseID, ParentID: parentID, SizeBytes: 100} +} + +func daysAgo(n int) time.Time { return refNow.AddDate(0, 0, -n) } +func roots(chains []Chain) []string { + out := make([]string, len(chains)) + for i, c := range chains { + out[i] = c.Root + } + sort.Strings(out) + return out +} + +func TestGroupChains(t *testing.T) { + t.Run("singletons and an incremental chain", func(t *testing.T) { + ds := []Meta{ + dump("full1", "p", daysAgo(10), "", ""), // legacy full (empty BaseID) + dump("base2", "p", daysAgo(5), "base2", ""), // self-rooted base + dump("inc2a", "p", daysAgo(4), "base2", "base2"), // incremental of base2 + dump("inc2b", "p", daysAgo(3), "base2", "inc2a"), // incremental of inc2a + } + chains := GroupChains(ds) + if len(chains) != 2 { + t.Fatalf("got %d chains, want 2: %+v", len(chains), chains) + } + got := roots(chains) + if got[0] != "base2" || got[1] != "full1" { + t.Errorf("roots = %v, want [base2 full1]", got) + } + // base2's chain has 3 members, base first. + for _, c := range chains { + if c.Root == "base2" { + if len(c.Members) != 3 || c.Members[0].ID != "base2" { + t.Errorf("base2 chain = %+v, want 3 members base-first", c.Members) + } + } + } + }) + + t.Run("orphaned incremental anchors its own chain (no crash, no drop)", func(t *testing.T) { + // inc points at a base that isn't in the set. + ds := []Meta{dump("inc", "p", daysAgo(1), "missingbase", "missingbase")} + chains := GroupChains(ds) + if len(chains) != 1 || chains[0].Root != "missingbase" { + t.Fatalf("orphan grouping = %+v, want one chain rooted at missingbase", chains) + } + }) +} + +func TestPlan_EmptyPolicyKeepsEverything(t *testing.T) { + ds := []Meta{ + dump("a", "p", daysAgo(100), "", ""), + dump("b", "p", daysAgo(200), "", ""), + } + plan := Plan(GroupChains(ds), RetentionPolicy{}, refNow) + if len(plan.Prune) != 0 { + t.Errorf("empty policy pruned %d chains, want 0 (keep everything)", len(plan.Prune)) + } + if len(plan.Keep) != 2 { + t.Errorf("empty policy kept %d chains, want 2", len(plan.Keep)) + } +} + +func TestPlan_KeepLast(t *testing.T) { + var ds []Meta + for i := 0; i < 5; i++ { + ds = append(ds, dump("c"+string(rune('0'+i)), "p", daysAgo(i), "", "")) + } + plan := Plan(GroupChains(ds), RetentionPolicy{KeepLast: 2}, refNow) + if len(plan.Keep) != 2 || len(plan.Prune) != 3 { + t.Fatalf("keep-last-2 = keep %d / prune %d, want 2/3", len(plan.Keep), len(plan.Prune)) + } + // The two kept are the newest (c0 = today, c1 = yesterday). + got := roots(plan.Keep) + if got[0] != "c0" || got[1] != "c1" { + t.Errorf("kept roots = %v, want [c0 c1]", got) + } +} + +func TestPlan_MaxAge_UsesNewestMember(t *testing.T) { + // An OLD base with a FRESH incremental: the chain must be treated as young + // (newest member governs age) and kept under a 7-day max-age. + ds := []Meta{ + dump("oldbase", "p", daysAgo(30), "oldbase", ""), + dump("freshinc", "p", daysAgo(1), "oldbase", "oldbase"), + } + plan := Plan(GroupChains(ds), RetentionPolicy{MaxAge: 7 * 24 * time.Hour}, refNow) + if len(plan.Prune) != 0 { + t.Errorf("chain with fresh incremental was pruned by max-age: %+v", plan.Prune) + } +} + +func TestPlan_Union_RuleOnlyProtects(t *testing.T) { + // keep-last-1 alone would prune the 5-day-old chain; adding max-age 7d must + // protect it (union, not intersection). + ds := []Meta{ + dump("today", "p", daysAgo(0), "", ""), + dump("fivedays", "p", daysAgo(5), "", ""), + dump("tendays", "p", daysAgo(10), "", ""), + } + keepLastOnly := Plan(GroupChains(ds), RetentionPolicy{KeepLast: 1}, refNow) + if len(keepLastOnly.Keep) != 1 { + t.Fatalf("keep-last-1 kept %d, want 1", len(keepLastOnly.Keep)) + } + union := Plan(GroupChains(ds), RetentionPolicy{KeepLast: 1, MaxAge: 7 * 24 * time.Hour}, refNow) + got := roots(union.Keep) + // today (both rules) + fivedays (max-age) kept; tendays pruned. + if len(got) != 2 || got[0] != "fivedays" || got[1] != "today" { + t.Errorf("union kept = %v, want [fivedays today]", got) + } +} + +func TestPlan_GFS_Daily(t *testing.T) { + // Two chains on the same day + one on a prior day. gfs-daily:2 keeps the + // newest chain of each of the 2 most-recent days = 2 chains; the older + // same-day chain is pruned. + ds := []Meta{ + dump("day0_late", "p", refNow.Add(-1*time.Hour), "", ""), + dump("day0_early", "p", refNow.Add(-5*time.Hour), "", ""), + dump("day1", "p", daysAgo(1), "", ""), + dump("day2", "p", daysAgo(2), "", ""), + } + plan := Plan(GroupChains(ds), RetentionPolicy{GFS: GFSPolicy{Daily: 2}}, refNow) + got := roots(plan.Keep) + // kept: day0_late (today's representative) + day1 (yesterday). day0_early and + // day2 pruned. + if len(got) != 2 || got[0] != "day0_late" || got[1] != "day1" { + t.Errorf("gfs-daily:2 kept = %v, want [day0_late day1]", got) + } +} + +func TestPlan_GFS_MultiTierKeptOnce(t *testing.T) { + // A single recent chain is the representative of today's day AND this week's + // week AND this month's month. It must be kept exactly once, not duplicated. + ds := []Meta{dump("recent", "p", refNow.Add(-1*time.Hour), "", "")} + plan := Plan(GroupChains(ds), RetentionPolicy{GFS: GFSPolicy{Daily: 1, Weekly: 1, Monthly: 1}}, refNow) + if len(plan.Keep) != 1 || len(plan.Prune) != 0 { + t.Errorf("multi-tier single chain = keep %d / prune %d, want 1/0", len(plan.Keep), len(plan.Prune)) + } +} + +func TestPlan_GFS_FewerChainsThanBuckets(t *testing.T) { + // gfs-daily:30 with only 2 chains: keep both, prune none (no crash, no + // phantom buckets). + ds := []Meta{ + dump("a", "p", daysAgo(0), "", ""), + dump("b", "p", daysAgo(1), "", ""), + } + plan := Plan(GroupChains(ds), RetentionPolicy{GFS: GFSPolicy{Daily: 30}}, refNow) + if len(plan.Keep) != 2 || len(plan.Prune) != 0 { + t.Errorf("gfs over-provisioned = keep %d / prune %d, want 2/0", len(plan.Keep), len(plan.Prune)) + } +} + +func TestPlan_PrunedChainKeepsMembersTogether(t *testing.T) { + // A pruned multi-dump chain must surface ALL its members for deletion. + ds := []Meta{ + dump("keep", "p", daysAgo(0), "", ""), + dump("oldbase", "p", daysAgo(40), "oldbase", ""), + dump("oldinc", "p", daysAgo(39), "oldbase", "oldbase"), + } + plan := Plan(GroupChains(ds), RetentionPolicy{KeepLast: 1}, refNow) + if len(plan.Prune) != 1 { + t.Fatalf("pruned %d chains, want 1", len(plan.Prune)) + } + if len(plan.Prune[0].Members) != 2 { + t.Errorf("pruned chain has %d members, want 2 (base+inc together)", len(plan.Prune[0].Members)) + } +} From 1a4a69d320872a265297ab39a5ec9d12ddf25035 Mon Sep 17 00:00:00 2001 From: Nikhil Rajput Date: Fri, 26 Jun 2026 01:33:45 +0530 Subject: [PATCH 2/2] fix(retention): apply CodeRabbit review fixes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Apply post-review fixes to the retention cycle. - Order chain members topologically (base first, each child after its parent via ParentID), with Created only as a tie-breaker. Ordering by Created alone could place a child ahead of its base on tied or skewed timestamps, and the leaf-inward delete then removes the base before a surviving incremental — orphaning it. Stragglers from broken/cyclic links are appended, never dropped. - Stop deleting a chain's members at the first failure (return, not continue): once a leaf delete fails, deleting its ancestors would orphan it, so abort the chain and leave a complete shorter prefix. Other chains still proceed. - Make chain ordering deterministic: break equal newest() timestamps by Root so keep_last / GFS selection is reproducible (map iteration is randomized; a planner must not be). Same comparator used in Plan. - Reject negative retention.max_age in config validation: ParseDuration accepts "-1h", which would push the cutoff into the future and make the whole catalog deletable. - Reject negative CLI flag overrides before planning, mirroring the config-load validation that flags otherwise bypass. - Label a partially deleted chain "partially pruned", not "pruned". - Fix the command name in README and docs/RETENTION.md: it is "siphon dumps prune", not "siphon prune". - Add tests: topological member order on tied/skewed timestamps, deterministic chain order on tied timestamps, and that a chain's deletion stops (base untouched) when the leaf delete fails. --- README.md | 4 +- docs/RETENTION.md | 8 +-- internal/app/prune.go | 11 ++-- internal/app/prune_test.go | 30 +++++++++ internal/cli/dumps.go | 15 ++++- internal/config/config.go | 9 ++- internal/dumps/retention.go | 102 ++++++++++++++++++++++++++++--- internal/dumps/retention_test.go | 65 ++++++++++++++++++++ 8 files changed, 222 insertions(+), 22 deletions(-) diff --git a/README.md b/README.md index 5852587..29027dc 100644 --- a/README.md +++ b/README.md @@ -181,7 +181,7 @@ 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: +A `retention:` block (default + optional per-profile override) drives `siphon dumps prune`, which deletes old backups as whole chains so an incremental is never orphaned: ```yaml defaults: @@ -191,7 +191,7 @@ defaults: 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. +`siphon dumps 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 diff --git a/docs/RETENTION.md b/docs/RETENTION.md index 96166a0..0e13015 100644 --- a/docs/RETENTION.md +++ b/docs/RETENTION.md @@ -1,6 +1,6 @@ # Retention & pruning -`siphon prune` applies a retention policy to the dump catalog, deleting old +`siphon dumps 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. @@ -70,13 +70,13 @@ duration) fails fast at config load. ```bash # Dry-run (default): show which chains the policy would prune, for one profile. -siphon prune --profile prod +siphon dumps prune --profile prod # Override the configured policy for this run, then actually delete. -siphon prune --profile prod --keep-last 14 --apply +siphon dumps 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 +siphon dumps prune --gfs-daily 7 --gfs-weekly 4 --gfs-monthly 6 --apply ``` `--apply` performs deletions; without it, prune only prints the plan. Flags diff --git a/internal/app/prune.go b/internal/app/prune.go index 36d1287..c9f8ab4 100644 --- a/internal/app/prune.go +++ b/internal/app/prune.go @@ -78,16 +78,19 @@ func Prune(ctx context.Context, d Deps, opt PruneOpts) (*PruneResult, error) { } // 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. +// before the base) and STOPS at the first failure within the chain. Stopping is +// the invariant: if a leaf delete fails (its meta may be gone but the dump still +// catalogued), continuing to delete its ancestors would orphan that leaf — +// exactly what the chain-aware design prevents. Aborting the chain leaves a +// complete, still-restorable prefix. Other chains are unaffected; the failure is +// recorded and the run continues with them. 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 + return // do not delete this member's ancestors — would orphan it } oc.Deleted = append(oc.Deleted, m.ID) result.Reclaimed += m.SizeBytes diff --git a/internal/app/prune_test.go b/internal/app/prune_test.go index 9fef882..4638b46 100644 --- a/internal/app/prune_test.go +++ b/internal/app/prune_test.go @@ -193,6 +193,36 @@ func TestPrune_CollectsDeletionFailures(t *testing.T) { } } +// TestPrune_StopsChainOnFirstFailure proves the orphan-prevention invariant: +// when deleting the leaf incremental of a pruned chain fails, the prune must NOT +// proceed to delete the base — otherwise the surviving leaf is orphaned. +func TestPrune_StopsChainOnFirstFailure(t *testing.T) { + store := newRecordingStore() + cat := dumps.New(store) + old := time.Now().AddDate(0, 0, -40) + seedDump(t, cat, "base", "p", old, "base", "") + seedDump(t, cat, "inc", "p", old.Add(time.Hour), "base", "base") + // The leaf (inc) is deleted first (leaf-inward). Make its meta delete fail — + // Catalog.Delete removes meta first, so this is the first Delete call. + store.failKeys["inc.meta.json"] = true + + res, err := Prune(context.Background(), pruneDeps(store), PruneOpts{ + Policy: dumps.RetentionPolicy{MaxAge: time.Hour}, Apply: true, + }) + if err != nil { + t.Fatalf("Prune: %v", err) + } + if res.Failed != 1 { + t.Errorf("Failed = %d, want 1", res.Failed) + } + // The base must NOT have been deleted — neither its meta nor its body. + for _, k := range store.deletes { + if strings.HasPrefix(k, "base") { + t.Errorf("base was deleted after leaf failure (orphans the leaf): deletes=%v", store.deletes) + } + } +} + // helpers func firstIndexWithPrefix(s []string, p string) int { diff --git a/internal/cli/dumps.go b/internal/cli/dumps.go index 002f662..7377e39 100644 --- a/internal/cli/dumps.go +++ b/internal/cli/dumps.go @@ -100,6 +100,13 @@ func dumpsPruneCmd() *cobra.Command { if gfsMSet { policy.GFS.Monthly = gfsMonthly } + // Config-backed policy is validated at load; flag overrides bypass that, + // so reject negatives here too before a destructive run sees an + // undefined policy. + if policy.KeepLast < 0 || policy.MaxAge < 0 || + policy.GFS.Daily < 0 || policy.GFS.Weekly < 0 || policy.GFS.Monthly < 0 { + return &errs.Error{Op: "prune", Code: errs.CodeUser, Hint: "retention values must be non-negative"} + } deps, err := buildDeps() if err != nil { @@ -169,7 +176,13 @@ func renderPruneResult(w io.Writer, res *app.PruneResult) { continue } prunedChains++ - _, _ = fmt.Fprintf(w, "%s chain %s (%d dump(s), %d bytes)\n", verb, oc.Root, len(oc.DumpIDs), oc.SizeBytes) + // On apply, a chain with errors was only partially deleted (the delete + // aborted leaf-inward at the failure), so don't label it fully "pruned". + label := verb + if res.Apply && len(oc.Errors) > 0 { + label = "partially pruned" + } + _, _ = fmt.Fprintf(w, "%s chain %s (%d dump(s), %d bytes)\n", label, oc.Root, len(oc.DumpIDs), oc.SizeBytes) for _, e := range oc.Errors { _, _ = fmt.Fprintf(w, " ! %s\n", e) } diff --git a/internal/config/config.go b/internal/config/config.go index 198cf6a..8fa9a81 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -59,9 +59,16 @@ func (r *RetentionConfig) Validate() error { return errors.New("retention.gfs tiers must be >= 0") } if r.MaxAge != "" { - if _, err := time.ParseDuration(r.MaxAge); err != nil { + d, err := time.ParseDuration(r.MaxAge) + if err != nil { return fmt.Errorf("retention.max_age %q is not a valid duration: %w", r.MaxAge, err) } + // time.ParseDuration accepts signed inputs ("-1h"); a negative max-age + // would push the cutoff into the future and make the whole catalog + // eligible for deletion, so reject it. + if d < 0 { + return fmt.Errorf("retention.max_age must be >= 0, got %q", r.MaxAge) + } } return nil } diff --git a/internal/dumps/retention.go b/internal/dumps/retention.go index 061f2a2..18c3919 100644 --- a/internal/dumps/retention.go +++ b/internal/dumps/retention.go @@ -66,8 +66,14 @@ type RetentionPlan struct { // backup (BaseID == ID, or legacy empty BaseID) is a singleton chain; its // incrementals attach to it via their BaseID. A dump whose root is missing from // the set anchors its own chain rather than being dropped, so a partially -// present catalog never loses entries silently. Within each chain, members are -// ordered by Created (base first). +// present catalog never loses entries silently. +// +// Within each chain, members are ordered by ParentID/BaseID TOPOLOGY (base +// first, then each child after its parent), with Created only as a tie-breaker. +// This is the contract the leaf-inward delete path relies on: deleting members +// last-to-first must remove every descendant before its ancestor. Ordering by +// Created alone would break that on tied or skewed timestamps — a child could +// sort ahead of its parent and the base could be deleted while a leaf survives. func GroupChains(dumps []Meta) []Chain { byRoot := map[string][]Meta{} for _, m := range dumps { @@ -79,18 +85,88 @@ func GroupChains(dumps []Meta) []Chain { } chains := make([]Chain, 0, len(byRoot)) for root, members := range byRoot { - sort.SliceStable(members, func(i, j int) bool { - return members[i].Created.Before(members[j].Created) - }) - chains = append(chains, Chain{Root: root, Members: members}) + chains = append(chains, Chain{Root: root, Members: orderMembers(root, members)}) } - // Deterministic order: newest chain (by newest member) first. + // Deterministic order: newest chain first, Root breaking ties so the plan is + // reproducible across runs even when timestamps collide (map iteration is + // randomized; a destructive planner must not be). sort.SliceStable(chains, func(i, j int) bool { - return chains[i].newest().After(chains[j].newest()) + ti, tj := chains[i].newest(), chains[j].newest() + if ti.Equal(tj) { + return chains[i].Root < chains[j].Root + } + return ti.After(tj) }) return chains } +// orderMembers returns a chain's members in topological apply order: the base +// (root) first, then each incremental placed immediately it can be (its parent +// already emitted). Created breaks ties among siblings and gives a stable order. +// Any members left unplaceable by broken/cyclic ParentID links are appended in +// Created order so nothing is dropped — a corrupt chain still surfaces all its +// dumps for the caller to handle. +func orderMembers(root string, members []Meta) []Meta { + // Stable Created order first, so ties resolve deterministically. + sort.SliceStable(members, func(i, j int) bool { + if members[i].Created.Equal(members[j].Created) { + return members[i].ID < members[j].ID + } + return members[i].Created.Before(members[j].Created) + }) + + byID := make(map[string]Meta, len(members)) + for _, m := range members { + byID[m.ID] = m + } + + ordered := make([]Meta, 0, len(members)) + emitted := make(map[string]bool, len(members)) + + // Emit the base first if present (its ID == root, or it self-roots / is a + // legacy empty-BaseID full backup). + if base, ok := byID[root]; ok { + ordered = append(ordered, base) + emitted[base.ID] = true + } + + // attachable reports whether m can be emitted now: its parent is already + // emitted, or its parent is not part of this chain (absent/self/empty), in + // which case it attaches at the root level. The base itself is handled above. + attachable := func(m Meta) bool { + if m.ID == root { + return false // already emitted (or will be appended as a straggler) + } + p := m.ParentID + return p == "" || p == m.ID || emitted[p] || byID[p].ID == "" + } + + // Iterate to a fixed point; members are Created-sorted, so each pass emits the + // earliest now-attachable child. + for { + progressed := false + for _, m := range members { + if emitted[m.ID] || !attachable(m) { + continue + } + ordered = append(ordered, m) + emitted[m.ID] = true + progressed = true + } + if !progressed { + break + } + } + + // Append any stragglers (cycles) in Created order — never drop a member. + for _, m := range members { + if !emitted[m.ID] { + ordered = append(ordered, m) + } + } + return ordered +} + // Plan decides which chains to keep vs prune under p, as of now. It is pure: no // I/O, no clock — now is injected — so every rule and edge case is unit-testable // with synthetic fixtures. An empty policy keeps everything. @@ -99,10 +175,16 @@ func Plan(chains []Chain, p RetentionPolicy, now time.Time) RetentionPlan { return RetentionPlan{Keep: append([]Chain(nil), chains...)} } - // chains is already sorted newest-first by GroupChains; copy defensively. + // chains is already sorted newest-first by GroupChains; copy and re-sort with + // the SAME deterministic comparator (Root breaks ties) so callers that build + // a Chain slice by hand still get reproducible keep_last / GFS selection. ordered := append([]Chain(nil), chains...) sort.SliceStable(ordered, func(i, j int) bool { - return ordered[i].newest().After(ordered[j].newest()) + ti, tj := ordered[i].newest(), ordered[j].newest() + if ti.Equal(tj) { + return ordered[i].Root < ordered[j].Root + } + return ti.After(tj) }) keep := map[string]bool{} // union of every active rule's keep set diff --git a/internal/dumps/retention_test.go b/internal/dumps/retention_test.go index 923674c..e8ad14a 100644 --- a/internal/dumps/retention_test.go +++ b/internal/dumps/retention_test.go @@ -58,6 +58,71 @@ func TestGroupChains(t *testing.T) { t.Fatalf("orphan grouping = %+v, want one chain rooted at missingbase", chains) } }) + + t.Run("members ordered by topology, not Created, on tied timestamps", func(t *testing.T) { + // Base and incremental share the SAME Created time (clock skew / coarse + // granularity). Topological order MUST still put the base first, because + // the leaf-inward delete path deletes last-to-first and would otherwise + // remove the base before its surviving child. + same := daysAgo(2) + ds := []Meta{ + dump("inc", "p", same, "base", "base"), // listed first, same timestamp + dump("base", "p", same, "base", ""), + } + chains := GroupChains(ds) + if len(chains) != 1 { + t.Fatalf("got %d chains, want 1", len(chains)) + } + got := chains[0].Members + if len(got) != 2 || got[0].ID != "base" || got[1].ID != "inc" { + t.Errorf("member order = [%s %s], want [base inc] (base first despite tie)", got[0].ID, got[1].ID) + } + }) + + t.Run("members topologically ordered when a child predates its parent", func(t *testing.T) { + // Pathological: the incremental's Created is BEFORE the base's (skew). + // Topology must still win — base first. + ds := []Meta{ + dump("base", "p", daysAgo(1), "base", ""), + dump("inc", "p", daysAgo(5), "base", "base"), // older than its base + } + chains := GroupChains(ds) + got := chains[0].Members + if got[0].ID != "base" { + t.Errorf("member order = [%s ...], want base first despite older child", got[0].ID) + } + }) +} + +func TestGroupChains_DeterministicOnTiedChainTimestamps(t *testing.T) { + // Two independent chains with identical newest() timestamps must sort in a + // stable order (by Root) across runs — a destructive planner cannot depend on + // randomized map iteration. Run the grouping repeatedly and assert identical + // output. + same := daysAgo(3) + ds := []Meta{ + dump("zzz", "p", same, "", ""), + dump("aaa", "p", same, "", ""), + dump("mmm", "p", same, "", ""), + } + first := roots(GroupChains(ds)) + for i := 0; i < 20; i++ { + if got := roots(GroupChains(ds)); !equalStrings(got, first) { + t.Fatalf("non-deterministic chain order: run %d = %v, first = %v", i, got, first) + } + } +} + +func equalStrings(a, b []string) bool { + if len(a) != len(b) { + return false + } + for i := range a { + if a[i] != b[i] { + return false + } + } + return true } func TestPlan_EmptyPolicyKeepsEverything(t *testing.T) {