From c67ecf86a9c28f131a9dcc9e4acdb0af6aa0a121 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Mon, 8 Jun 2026 19:26:48 +0000 Subject: [PATCH] Add prune retention planning --- docs/config.md | 20 ++++ docs/internal/app.md | 10 +- docs/internal/config.md | 4 +- docs/internal/state.md | 8 +- internal/app/prune.go | 106 ++++++++++++++++++++ internal/app/prune_test.go | 162 +++++++++++++++++++++++++++++++ internal/config/config.go | 11 +++ internal/config/load_test.go | 34 +++++++ internal/config/validate.go | 18 ++++ internal/config/validate_test.go | 45 +++++++++ internal/state/prune.go | 121 +++++++++++++++++++++++ internal/state/prune_test.go | 96 ++++++++++++++++++ 12 files changed, 628 insertions(+), 7 deletions(-) create mode 100644 internal/app/prune.go create mode 100644 internal/app/prune_test.go create mode 100644 internal/state/prune.go create mode 100644 internal/state/prune_test.go diff --git a/docs/config.md b/docs/config.md index 42e1aa3..57c1e5f 100644 --- a/docs/config.md +++ b/docs/config.md @@ -286,6 +286,7 @@ destinations: - `links`: optional public URL metadata policy. - `state`: optional destination state ownership policy. - `reconciliation`: optional managed-output reconciliation policy. +- `retention`: optional managed-output retention policy. - `transfer`: optional destination comparison action policy. Destination ids must be unique within a pipeline. @@ -448,6 +449,24 @@ destinations: `links.primary_url` is selected from the newly planned outputs for the current run. Retained outputs keep their prior output metadata and timestamps. +## Retention Policy + +```yaml +retention: + prune: + enabled: false + older_than: 168h + keep_latest: 3 +``` + +- `retention.prune.enabled`: optional boolean. Default is `false`. +- `retention.prune.older_than`: optional duration. When pruning is enabled, outputs older than this duration are eligible in prune planning. +- `retention.prune.keep_latest`: optional non-negative integer. When pruning is enabled, this many newest managed outputs are preserved before age-based pruning is considered. + +When `retention.prune.enabled` is `true`, at least one of `older_than` or `keep_latest` is required. `older_than` must be greater than zero, and `keep_latest` must be zero or greater. + +Prune planning uses managed output `updated_at` timestamps from destination state. It is owner-scoped for shared-root state. The current implementation parses, validates, and plans pruning from this policy; it does not delete destination files. + ## Transfer Policy ```yaml @@ -504,6 +523,7 @@ Defaults are applied after YAML decoding and before validation: - `links.primary: auto` when a `links` block is present and `primary` is omitted - `state.mode: single_owner` - `reconciliation.mode: replace` +- `retention.prune.enabled: false` - `transfer.on_destination_same: skip` - `transfer.on_destination_older: replace` - `transfer.on_destination_newer: skip` diff --git a/docs/internal/app.md b/docs/internal/app.md index b1cb81f..36466ee 100644 --- a/docs/internal/app.md +++ b/docs/internal/app.md @@ -4,13 +4,13 @@ Audience: developers and LLM coding agents changing `internal/app`. ## Purpose -`internal/app` owns top-level application use cases: run, single-pipeline run, staged-source run, validate, inspect, manifest creation, reconcile-state planning/repair, and HTTP upload serving. It coordinates config loading, secret resolution, backend construction, source discovery, destination selection, publish planning/execution, state repair reporting, notification handoff, output projection, and upload coordination. +`internal/app` owns top-level application use cases: run, single-pipeline run, staged-source run, validate, inspect, manifest creation, reconcile-state planning/repair, prune planning, and HTTP upload serving. It coordinates config loading, secret resolution, backend construction, source discovery, destination selection, publish planning/execution, state repair reporting, retention plan reporting, notification handoff, output projection, and upload coordination. ## Inputs And Outputs Inputs include app option structs, contexts, config paths, pipeline ids, local source roots, dry-run/force flags, output format, stdout writers, HTTP requests, and optional notifier implementations. -Outputs include `RunReport`, `ReconcileStateReport`, validate/inspect/manifest results, CLI text/JSON projections, HTTP upload responses, upload status records, and errors. Destination-scoped failures can return a partial run report plus an aggregated error; fatal setup failures return before a complete report exists. +Outputs include `RunReport`, `ReconcileStateReport`, `PrunePlanReport`, validate/inspect/manifest results, CLI text/JSON projections, HTTP upload responses, upload status records, and errors. Destination-scoped failures can return a partial run report plus an aggregated error; fatal setup failures return before a complete report exists. ## Boundaries @@ -20,7 +20,7 @@ User-facing command parsing stays in `internal/cli`, including `reconcile-state` ## Config Fields Used -The package consumes the loaded `config.Config`: `server.http`, `secrets.directory`, pipeline ids, source and destination backend fields, validation policy, publish policy, transform policy, path mapping, links, reconciliation policy, and transfer policy. +The package consumes the loaded `config.Config`: `server.http`, `secrets.directory`, pipeline ids, source and destination backend fields, validation policy, publish policy, transform policy, path mapping, links, state policy, reconciliation policy, retention policy, and transfer policy. Config fields are validated and defaulted by `internal/config` before app workflows use them. @@ -36,6 +36,8 @@ Run workflows discover and validate source bundles through `internal/bundle`. De Reconcile-state workflows load one configured pipeline/destination selector, open that destination root, parse the root `.distributor.json`, and report missing managed output records plus unmanaged storage entries. Managed output existence checks use storage `Stat`; unmanaged reporting uses bounded storage `Walk` and excludes `.distributor.json` plus all paths already recorded as managed. Apply mode removes missing managed output records from state and rewrites valid state only; dry-run reports the same repair without writing. Text output reports `changed`, `would_change`, or `unchanged`; JSON output uses the shared app envelope. It does not validate output digests, delete destination files, adopt unmanaged files, or rewrite invalid or mismatched state. +Prune planning consumes a parsed destination state document and a validated retention prune policy, then returns owner-scoped managed output records that would be pruned or preserved. Planning uses output `updated_at` timestamps, applies `keep_latest` before `older_than` when both are configured, and does not open storage, delete files, or rewrite state. + HTTP uploads stage and validate archives before enqueueing a pipeline run with a local staged source root. Go producers can use the public `pkg/upload` package to create client-side gzip tar uploads for this server contract; `internal/app` remains the server-side orchestration boundary and does not import that producer package. Upload idempotency is owned by the upload coordinator. Optional `Idempotency-Key` values are scoped to token id, pipeline id, and key. The coordinator reserves a key while staging is in progress, records the accepted run id with the validated source manifest identity after staging succeeds, returns the original accepted record for the same scoped key and same manifest, and rejects the same scoped key with a different manifest as a conflict. @@ -57,6 +59,7 @@ HTTP upload startup fails if upload tokens are missing, empty, or duplicated. Up ## Tests To Inspect - `internal/app/*_test.go` +- `internal/app/prune_test.go` - `internal/cli/reconcile_state_test.go` - `internal/cli/root_test.go` - `internal/config/*_test.go` @@ -73,3 +76,4 @@ HTTP upload startup fails if upload tokens are missing, empty, or duplicated. Up - Idempotent upload retries compare normalized source manifest identity, not archive bytes. - Runtime backend registration remains app-owned. - Reconcile-state repairs state records only; it never deletes or adopts destination files. +- Prune planning is report-only until execution code applies a plan. diff --git a/docs/internal/config.md b/docs/internal/config.md index 084343c..26f5ee9 100644 --- a/docs/internal/config.md +++ b/docs/internal/config.md @@ -18,7 +18,7 @@ The canonical user-facing config reference is `docs/config.md`. ## Config Fields Used -The package defines all user-visible config fields: `server.http`, `secrets`, `pipelines`, source and destination backend fields, validation policy, publish policy, transform policy, path mapping, links, and transfer policy. +The package defines all user-visible config fields: `server.http`, `secrets`, `pipelines`, source and destination backend fields, validation policy, publish policy, transform policy, path mapping, links, state policy, reconciliation policy, retention policy, and transfer policy. ## Adapters Used @@ -26,7 +26,7 @@ No external storage adapters are used directly. The package exposes normalized c ## State And Manifest Behavior -The package does not parse source manifests or destination state. It validates config values that later affect manifest validation and destination state, such as publish/transform combinations, links, transfer policy, backend roots, S3 prefix shape, and HTTP upload source settings. +The package does not parse source manifests or destination state. It validates config values that later affect manifest validation and destination state, such as publish/transform combinations, links, state policy, reconciliation policy, retention policy, transfer policy, backend roots, S3 prefix shape, and HTTP upload source settings. ## Skip And Resume Behavior diff --git a/docs/internal/state.md b/docs/internal/state.md index d0b79fa..8befb76 100644 --- a/docs/internal/state.md +++ b/docs/internal/state.md @@ -36,9 +36,9 @@ Shared-root publish conversion is explicit. Compatible single-owner state for th Embedded source manifests are parsed and validated through `internal/bundle`, which delegates source manifest semantics to `pkg/bundle`. Output records require clean paths, `source` or `generated` kind, valid source paths, lowercase SHA-256 digests, non-negative sizes, created and updated timestamps, and transform ids for generated outputs. Stored URLs must pass `internal/link` validation. -The package also provides helpers for finding output records by path, projecting planned publish outputs into timestamped state outputs, merging retained and newly planned output records, computing managed output paths from single-owner state, and removing missing managed output records from single-owner state. +The package also provides helpers for finding output records by path, projecting planned publish outputs into timestamped state outputs, merging retained and newly planned output records, computing managed output paths from single-owner state, removing missing managed output records from single-owner state, and building prune candidates from managed outputs. -For shared-root state, helpers parse either state shape, identify the current owner scope, return an owner's latest source manifest, list managed paths for one owner or all owners, detect path ownership conflicts, project planned owner outputs, merge one owner's planned outputs while preserving unrelated owners, replace one owner's outputs by removing that owner's omitted outputs, and remove missing managed output records for either the current owner or every owner. +For shared-root state, helpers parse either state shape, identify the current owner scope, return an owner's latest source manifest, list managed paths for one owner or all owners, detect path ownership conflicts, project planned owner outputs, merge one owner's planned outputs while preserving unrelated owners, replace one owner's outputs by removing that owner's omitted outputs, remove missing managed output records for either the current owner or every owner, and build owner-scoped prune candidates. Shared-root helper projections preserve output `created_at` for existing managed paths and use the current publication time for rewritten `updated_at`. Root-level `created_at` preservation is owned by publish execution. @@ -50,6 +50,8 @@ Shared-root comparison is owner-scoped. It compares only the owner keyed by the Missing-output removal helpers are pure state transformations used by app-level state repair. They remove matching output records only and leave storage inspection, timestamp updates, validation, and state rewrites to callers. +Prune planning helpers are pure. They select managed output candidates, sort deterministically by `updated_at` and path, preserve the newest `keep_latest` candidates before evaluating `older_than`, and return planned prune/preserve lists without mutating state. + ## Failure Behavior Parsing rejects invalid JSON, trailing data, missing required fields, invalid timestamps, invalid state mode, invalid reconciliation mode, invalid embedded manifests, duplicate owners, duplicate outputs, invalid output paths, unsupported output kinds, missing generated transforms, invalid URLs, invalid digests, negative sizes, and shared-root outputs whose owner is not registered. @@ -58,6 +60,7 @@ Parsing rejects invalid JSON, trailing data, missing required fields, invalid ti - `internal/state/distributor_test.go` - `internal/state/shared_root_test.go` +- `internal/state/prune_test.go` - `internal/state/compare_test.go` - `internal/app/reconcile_state_test.go` - `internal/cli/reconcile_state_test.go` @@ -73,6 +76,7 @@ Parsing rejects invalid JSON, trailing data, missing required fields, invalid ti - Schema version `3` shared-root state is parsed and validated without converting unrelated single-owner state. - Shared-root owner updates preserve unrelated owners and reject planned path collisions with other owners. - Missing-output repair helpers preserve unrelated owner records and outputs. +- Prune planning uses output `updated_at` and preserves unrelated shared-root owners. - Generated outputs always record a transform id. - Output records always carry created and updated timestamps after parsing. - Stored URLs are optional and must be absolute HTTP or HTTPS URLs when present. diff --git a/internal/app/prune.go b/internal/app/prune.go new file mode 100644 index 0000000..377ff06 --- /dev/null +++ b/internal/app/prune.go @@ -0,0 +1,106 @@ +package app + +import ( + "fmt" + "time" + + "gitea.maximumdirect.net/eric/distributor/internal/config" + "gitea.maximumdirect.net/eric/distributor/internal/state" +) + +type PrunePlanOptions struct { + PipelineID string + DestinationID string + Now time.Time +} + +type PrunePlanReport struct { + PipelineID string `json:"pipeline_id"` + DestinationID string `json:"destination_id"` + OwnerScope PruneOwnerScope `json:"owner_scope"` + Enabled bool `json:"enabled"` + CheckedCount int `json:"checked_count"` + PrunedOutputs []PruneOutputRecord `json:"pruned_outputs"` + PreservedOutputs []PruneOutputRecord `json:"preserved_outputs"` +} + +type PruneOwnerScope struct { + PipelineID string `json:"pipeline_id"` + DestinationID string `json:"destination_id"` +} + +type PruneOutputRecord struct { + Path string `json:"path"` + UpdatedAt string `json:"updated_at"` + Owner *PruneOwnerScope `json:"owner,omitempty"` +} + +func PlanPrune(document state.StateDocument, policy config.PrunePolicy, options PrunePlanOptions) (PrunePlanReport, error) { + scope := state.CurrentOwnerScope(options.PipelineID, options.DestinationID) + report := PrunePlanReport{ + PipelineID: options.PipelineID, + DestinationID: options.DestinationID, + OwnerScope: PruneOwnerScope{PipelineID: scope.PipelineID, DestinationID: scope.DestinationID}, + Enabled: policy.Enabled, + PrunedOutputs: []PruneOutputRecord{}, + PreservedOutputs: []PruneOutputRecord{}, + } + if !policy.Enabled { + return report, nil + } + + candidates, err := pruneCandidatesForDocument(document, scope) + if err != nil { + return PrunePlanReport{}, err + } + report.CheckedCount = len(candidates) + plan := state.PlanPrune(candidates, state.PrunePlanOptions{ + Now: options.Now, + OlderThan: pruneOlderThan(policy), + KeepLatest: policy.KeepLatest, + }) + report.PrunedOutputs = pruneOutputRecords(plan.Pruned) + report.PreservedOutputs = pruneOutputRecords(plan.Preserved) + return report, nil +} + +func pruneCandidatesForDocument(document state.StateDocument, scope state.OwnerScope) ([]state.PruneCandidate, error) { + if document.SingleOwner != nil { + singleOwner := *document.SingleOwner + if singleOwner.PipelineID != scope.PipelineID || singleOwner.DestinationID != scope.DestinationID { + return nil, fmt.Errorf("state owner is %s/%s, not %s/%s", singleOwner.PipelineID, singleOwner.DestinationID, scope.PipelineID, scope.DestinationID) + } + return state.SingleOwnerPruneCandidates(singleOwner), nil + } + if document.SharedRoot != nil { + return state.SharedRootPruneCandidates(*document.SharedRoot, scope), nil + } + return nil, fmt.Errorf("destination state document is empty") +} + +func pruneOlderThan(policy config.PrunePolicy) *time.Duration { + if policy.OlderThan == nil { + return nil + } + duration := policy.OlderThan.AsDuration() + return &duration +} + +func pruneOutputRecords(candidates []state.PruneCandidate) []PruneOutputRecord { + records := make([]PruneOutputRecord, 0, len(candidates)) + for _, candidate := range candidates { + var owner *PruneOwnerScope + if candidate.Owner != nil { + owner = &PruneOwnerScope{ + PipelineID: candidate.Owner.PipelineID, + DestinationID: candidate.Owner.DestinationID, + } + } + records = append(records, PruneOutputRecord{ + Path: candidate.Path, + UpdatedAt: candidate.UpdatedAt.UTC().Format(time.RFC3339), + Owner: owner, + }) + } + return records +} diff --git a/internal/app/prune_test.go b/internal/app/prune_test.go new file mode 100644 index 0000000..b910be9 --- /dev/null +++ b/internal/app/prune_test.go @@ -0,0 +1,162 @@ +package app + +import ( + "strings" + "testing" + "time" + + "gitea.maximumdirect.net/eric/distributor/internal/config" + "gitea.maximumdirect.net/eric/distributor/internal/state" + "gitea.maximumdirect.net/eric/distributor/internal/testutil" +) + +func TestPlanPruneDisabledPolicy(t *testing.T) { + document := state.StateDocument{SingleOwner: &state.DistributorState{}} + report, err := PlanPrune(document, config.PrunePolicy{}, PrunePlanOptions{ + PipelineID: "reports", + DestinationID: "archive", + }) + if err != nil { + t.Fatalf("PlanPrune() error = %v", err) + } + if report.Enabled || report.CheckedCount != 0 || len(report.PrunedOutputs) != 0 { + t.Fatalf("report = %#v, want disabled empty plan", report) + } +} + +func TestPlanPruneSingleOwnerOutputs(t *testing.T) { + now := time.Date(2026, 6, 8, 12, 0, 0, 0, time.UTC) + olderThan := config.Duration(48 * time.Hour) + destinationState := pruneSingleOwnerState(now) + + report, err := PlanPrune(state.StateDocument{SingleOwner: &destinationState}, config.PrunePolicy{ + Enabled: true, + OlderThan: &olderThan, + }, PrunePlanOptions{ + PipelineID: "reports", + DestinationID: "archive", + Now: now, + }) + if err != nil { + t.Fatalf("PlanPrune() error = %v", err) + } + if got, want := pruneRecordPaths(report.PrunedOutputs), "old.txt"; got != want { + t.Fatalf("pruned = %q, want %q", got, want) + } + if got, want := pruneRecordPaths(report.PreservedOutputs), "fresh.txt"; got != want { + t.Fatalf("preserved = %q, want %q", got, want) + } +} + +func TestPlanPruneSharedRootCurrentOwnerOnly(t *testing.T) { + now := time.Date(2026, 6, 8, 12, 0, 0, 0, time.UTC) + keepLatest := 0 + sharedRoot := pruneSharedRootState(now) + + report, err := PlanPrune(state.StateDocument{SharedRoot: &sharedRoot}, config.PrunePolicy{ + Enabled: true, + KeepLatest: &keepLatest, + }, PrunePlanOptions{ + PipelineID: "reports", + DestinationID: "archive", + Now: now, + }) + if err != nil { + t.Fatalf("PlanPrune() error = %v", err) + } + if got, want := report.CheckedCount, 1; got != want { + t.Fatalf("checked count = %d, want %d", got, want) + } + if got, want := pruneRecordPaths(report.PrunedOutputs), "archive.txt"; got != want { + t.Fatalf("pruned = %q, want %q", got, want) + } +} + +func pruneSingleOwnerState(now time.Time) state.DistributorState { + manifest := testutil.ValidManifest(testutil.BundleOptions{}) + publishedAt := now.Add(-96 * time.Hour) + return state.DistributorState{ + SchemaVersion: state.SchemaVersion, + PipelineID: "reports", + DestinationID: "archive", + PublishedAt: publishedAt, + CreatedAt: publishedAt, + UpdatedAt: publishedAt, + State: state.StatePolicy{Mode: state.StateModeSingleOwner}, + Reconciliation: state.ReconciliationPolicy{Mode: config.ReconciliationModeReplace}, + Source: state.SourceState{Manifest: manifest}, + DistributorVersion: "test", + Outputs: []state.OutputFile{{ + Path: "old.txt", + Kind: state.OutputKindSource, + SourcePath: "report.md", + SHA256: manifest.Files[0].SHA256, + Size: manifest.Files[0].Size, + CreatedAt: now.Add(-96 * time.Hour), + UpdatedAt: now.Add(-72 * time.Hour), + }, { + Path: "fresh.txt", + Kind: state.OutputKindSource, + SourcePath: "summary.txt", + SHA256: manifest.Files[1].SHA256, + Size: manifest.Files[1].Size, + CreatedAt: now.Add(-24 * time.Hour), + UpdatedAt: now.Add(-24 * time.Hour), + }}, + } +} + +func pruneSharedRootState(now time.Time) state.SharedRootState { + manifest := testutil.ValidManifest(testutil.BundleOptions{}) + archive := state.CurrentOwnerScope("reports", "archive") + html := state.CurrentOwnerScope("reports", "html") + return state.SharedRootState{ + SchemaVersion: state.SharedRootSchemaVersion, + DistributorVersion: "test", + CreatedAt: now.Add(-96 * time.Hour), + UpdatedAt: now.Add(-24 * time.Hour), + State: state.StatePolicy{Mode: state.StateModeSharedRoot}, + Owners: []state.OwnerRecord{{ + Scope: archive, + Reconciliation: state.ReconciliationPolicy{Mode: config.ReconciliationModeReplace}, + Source: state.SourceState{Manifest: manifest}, + }, { + Scope: html, + Reconciliation: state.ReconciliationPolicy{Mode: config.ReconciliationModeReplace}, + Source: state.SourceState{Manifest: manifest}, + }}, + Outputs: []state.SharedRootOutputFile{{ + Path: "archive.txt", + Kind: state.OutputKindSource, + SourcePath: "report.md", + SHA256: manifest.Files[0].SHA256, + Size: manifest.Files[0].Size, + Owner: archive, + SourceID: manifest.ID, + SourceDigest: manifest.Digest, + SourceCreated: manifest.Created, + CreatedAt: now.Add(-96 * time.Hour), + UpdatedAt: now.Add(-72 * time.Hour), + }, { + Path: "html.txt", + Kind: state.OutputKindSource, + SourcePath: "summary.txt", + SHA256: manifest.Files[1].SHA256, + Size: manifest.Files[1].Size, + Owner: html, + SourceID: manifest.ID, + SourceDigest: manifest.Digest, + SourceCreated: manifest.Created, + CreatedAt: now.Add(-96 * time.Hour), + UpdatedAt: now.Add(-72 * time.Hour), + }}, + } +} + +func pruneRecordPaths(records []PruneOutputRecord) string { + paths := make([]string, 0, len(records)) + for _, record := range records { + paths = append(paths, record.Path) + } + return strings.Join(paths, ",") +} diff --git a/internal/config/config.go b/internal/config/config.go index 3b38dcd..7f9e075 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -57,6 +57,7 @@ type Destination struct { Links *Links `yaml:"links"` State StatePolicy `yaml:"state"` Reconciliation ReconciliationPolicy `yaml:"reconciliation"` + Retention RetentionPolicy `yaml:"retention"` Transfer TransferPolicy `yaml:"transfer"` } @@ -128,6 +129,16 @@ type StatePolicy struct { Mode string `yaml:"mode"` } +type RetentionPolicy struct { + Prune PrunePolicy `yaml:"prune"` +} + +type PrunePolicy struct { + Enabled bool `yaml:"enabled"` + OlderThan *Duration `yaml:"older_than"` + KeepLatest *int `yaml:"keep_latest"` +} + type TransferPolicy struct { OnDestinationSame string `yaml:"on_destination_same"` OnDestinationOlder string `yaml:"on_destination_older"` diff --git a/internal/config/load_test.go b/internal/config/load_test.go index ded5c55..ddf6b03 100644 --- a/internal/config/load_test.go +++ b/internal/config/load_test.go @@ -39,6 +39,9 @@ pipelines: if got, want := destination.State.Mode, StateModeSingleOwner; got != want { t.Fatalf("state mode default = %q, want %q", got, want) } + if destination.Retention.Prune.Enabled { + t.Fatal("retention.prune.enabled default = true, want false") + } if cfg.Secrets.Directory != "" { t.Fatalf("secrets.directory = %q, want empty", cfg.Secrets.Directory) } @@ -228,6 +231,36 @@ pipelines: } } +func TestLoadFileAcceptsRetentionPruneConfig(t *testing.T) { + cfg := loadConfig(t, ` +pipelines: + - id: reports + source: + backend: local + path: /source + destinations: + - id: archive + backend: local + path: /archive + retention: + prune: + enabled: true + older_than: 168h + keep_latest: 3 +`) + + prune := cfg.Pipelines[0].Destinations[0].Retention.Prune + if !prune.Enabled { + t.Fatal("retention.prune.enabled = false, want true") + } + if prune.OlderThan == nil || prune.OlderThan.String() != "168h0m0s" { + t.Fatalf("retention.prune.older_than = %v, want 168h", prune.OlderThan) + } + if prune.KeepLatest == nil || *prune.KeepLatest != 3 { + t.Fatalf("retention.prune.keep_latest = %v, want 3", prune.KeepLatest) + } +} + func TestLoadFileAcceptsFixedPathMapping(t *testing.T) { cfg := loadConfig(t, ` pipelines: @@ -705,6 +738,7 @@ pipelines: [{id: reports, source: {backend: http_upload, max_upload_size: 20XB}, pipelines: [{id: reports, source: {backend: http_upload, max_upload_size: 0B}, destinations: [{id: archive, backend: local, path: /archive}]}]`, "server duration": `server: {http: {retention: forever}}`, "zero server duration": `server: {http: {retention: 0s}}`, + "prune duration": `pipelines: [{id: reports, source: {backend: local, path: /source}, destinations: [{id: archive, backend: local, path: /archive, retention: {prune: {enabled: true, older_than: forever}}}]}]`, "missing upload tokens": `pipelines: [{id: reports, source: {backend: http_upload}, destinations: [{id: archive, backend: local, path: /archive}]}]`, "destination http upload": `pipelines: [{id: reports, source: {backend: local, path: /source}, destinations: [{id: ingest, backend: http_upload}]}]`, "literal token": `upload_tokens: [{id: reporter, token: secret, token_env: UPLOAD_TOKEN, allow_pipelines: [reports]}] diff --git a/internal/config/validate.go b/internal/config/validate.go index 223b7d8..002983f 100644 --- a/internal/config/validate.go +++ b/internal/config/validate.go @@ -74,6 +74,7 @@ func Validate(cfg Config) error { errs = validateLinks(errs, destinationContext+".links", destination.Links) errs = validateStatePolicy(errs, destinationContext+".state", destination.State) errs = validateReconciliationPolicy(errs, destinationContext+".reconciliation", destination.Reconciliation) + errs = validateRetentionPolicy(errs, destinationContext+".retention", destination.Retention) errs = validateTransferPolicy(errs, destinationContext+".transfer", destination.Transfer) } } @@ -323,6 +324,23 @@ func validateReconciliationPolicy(errs ValidationErrors, context string, policy return errs } +func validateRetentionPolicy(errs ValidationErrors, context string, policy RetentionPolicy) ValidationErrors { + prune := policy.Prune + if !prune.Enabled { + return errs + } + if prune.OlderThan == nil && prune.KeepLatest == nil { + errs = append(errs, context+".prune must set older_than or keep_latest when enabled is true") + } + if prune.OlderThan != nil && *prune.OlderThan <= 0 { + errs = append(errs, context+".prune.older_than must be greater than zero") + } + if prune.KeepLatest != nil && *prune.KeepLatest < 0 { + errs = append(errs, context+".prune.keep_latest must be zero or greater") + } + return errs +} + func validateStatePolicy(errs ValidationErrors, context string, policy StatePolicy) ValidationErrors { if policy.Mode != StateModeSingleOwner && policy.Mode != StateModeSharedRoot { errs = append(errs, context+".mode must be "+StateModeSingleOwner+" or "+StateModeSharedRoot) diff --git a/internal/config/validate_test.go b/internal/config/validate_test.go index 6e9fbd8..8d71906 100644 --- a/internal/config/validate_test.go +++ b/internal/config/validate_test.go @@ -3,6 +3,7 @@ package config import ( "strings" "testing" + "time" ) func TestValidatePublishTransformPolicy(t *testing.T) { @@ -175,6 +176,50 @@ func TestValidateStatePolicy(t *testing.T) { } } +func TestValidateRetentionPolicy(t *testing.T) { + olderThan := Duration(24 * time.Hour) + zeroDuration := Duration(0) + keepZero := 0 + keepThree := 3 + keepNegative := -1 + tests := []struct { + name string + retention RetentionPolicy + wantErr bool + }{ + {name: "disabled"}, + {name: "older than", retention: RetentionPolicy{Prune: PrunePolicy{Enabled: true, OlderThan: &olderThan}}}, + {name: "keep zero", retention: RetentionPolicy{Prune: PrunePolicy{Enabled: true, KeepLatest: &keepZero}}}, + {name: "keep latest", retention: RetentionPolicy{Prune: PrunePolicy{Enabled: true, KeepLatest: &keepThree}}}, + {name: "combined", retention: RetentionPolicy{Prune: PrunePolicy{Enabled: true, OlderThan: &olderThan, KeepLatest: &keepThree}}}, + {name: "missing policy", retention: RetentionPolicy{Prune: PrunePolicy{Enabled: true}}, wantErr: true}, + {name: "zero older than", retention: RetentionPolicy{Prune: PrunePolicy{Enabled: true, OlderThan: &zeroDuration}}, wantErr: true}, + {name: "negative keep latest", retention: RetentionPolicy{Prune: PrunePolicy{Enabled: true, KeepLatest: &keepNegative}}, wantErr: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cfg := Config{Pipelines: []Pipeline{{ + ID: "reports", + Source: Backend{Backend: BackendLocal, Path: "/source"}, + Destinations: []Destination{{ + ID: "archive", + Backend: BackendLocal, + Path: "/destination", + Retention: tt.retention, + }}, + }}} + ApplyDefaults(&cfg) + err := Validate(cfg) + if tt.wantErr && err == nil { + t.Fatal("Validate() error = nil, want error") + } + if !tt.wantErr && err != nil { + t.Fatalf("Validate() error = %v", err) + } + }) + } +} + func TestValidateLinks(t *testing.T) { tests := []struct { name string diff --git a/internal/state/prune.go b/internal/state/prune.go new file mode 100644 index 0000000..527424f --- /dev/null +++ b/internal/state/prune.go @@ -0,0 +1,121 @@ +package state + +import ( + "sort" + "time" +) + +type PruneCandidate struct { + Path string + UpdatedAt time.Time + Owner *OwnerScope +} + +type PrunePlanOptions struct { + Now time.Time + OlderThan *time.Duration + KeepLatest *int +} + +type PrunePlan struct { + Pruned []PruneCandidate + Preserved []PruneCandidate +} + +func SingleOwnerPruneCandidates(s DistributorState) []PruneCandidate { + candidates := make([]PruneCandidate, 0, len(s.Outputs)) + for _, output := range s.Outputs { + candidates = append(candidates, PruneCandidate{ + Path: output.Path, + UpdatedAt: output.UpdatedAt, + }) + } + return candidates +} + +func SharedRootPruneCandidates(s SharedRootState, scope OwnerScope) []PruneCandidate { + candidates := make([]PruneCandidate, 0, len(s.Outputs)) + for _, output := range s.Outputs { + if output.Owner != scope { + continue + } + owner := output.Owner + candidates = append(candidates, PruneCandidate{ + Path: output.Path, + UpdatedAt: output.UpdatedAt, + Owner: &owner, + }) + } + return candidates +} + +func PlanPrune(candidates []PruneCandidate, options PrunePlanOptions) PrunePlan { + ordered := append([]PruneCandidate(nil), candidates...) + sortPruneCandidatesNewestFirst(ordered) + if options.OlderThan == nil && options.KeepLatest == nil { + return PrunePlan{ + Pruned: []PruneCandidate{}, + Preserved: ordered, + } + } + + preservedByPath := make(map[string]struct{}) + if options.KeepLatest != nil { + keep := *options.KeepLatest + if keep < 0 { + keep = 0 + } + if keep > len(ordered) { + keep = len(ordered) + } + for _, candidate := range ordered[:keep] { + preservedByPath[candidate.Path] = struct{}{} + } + } + + plan := PrunePlan{ + Pruned: []PruneCandidate{}, + Preserved: []PruneCandidate{}, + } + cutoff := time.Time{} + if options.OlderThan != nil { + now := options.Now.UTC() + if now.IsZero() { + now = time.Now().UTC() + } + cutoff = now.Add(-*options.OlderThan) + } + + for _, candidate := range ordered { + if _, preserved := preservedByPath[candidate.Path]; preserved { + plan.Preserved = append(plan.Preserved, candidate) + continue + } + if options.OlderThan == nil || candidate.UpdatedAt.Before(cutoff) { + plan.Pruned = append(plan.Pruned, candidate) + continue + } + plan.Preserved = append(plan.Preserved, candidate) + } + sortPruneCandidatesOldestFirst(plan.Pruned) + sortPruneCandidatesNewestFirst(plan.Preserved) + return plan +} + +func sortPruneCandidatesNewestFirst(candidates []PruneCandidate) { + sort.Slice(candidates, func(i, j int) bool { + if !candidates[i].UpdatedAt.Equal(candidates[j].UpdatedAt) { + return candidates[i].UpdatedAt.After(candidates[j].UpdatedAt) + } + return candidates[i].Path < candidates[j].Path + }) +} + +func sortPruneCandidatesOldestFirst(candidates []PruneCandidate) { + sort.Slice(candidates, func(i, j int) bool { + if !candidates[i].UpdatedAt.Equal(candidates[j].UpdatedAt) { + return candidates[i].UpdatedAt.Before(candidates[j].UpdatedAt) + } + return candidates[i].Path < candidates[j].Path + }) +} diff --git a/internal/state/prune_test.go b/internal/state/prune_test.go new file mode 100644 index 0000000..2d89a8f --- /dev/null +++ b/internal/state/prune_test.go @@ -0,0 +1,96 @@ +package state + +import ( + "strings" + "testing" + "time" +) + +func TestPlanPruneOlderThan(t *testing.T) { + now := time.Date(2026, 6, 8, 12, 0, 0, 0, time.UTC) + olderThan := 48 * time.Hour + plan := PlanPrune([]PruneCandidate{ + {Path: "old.txt", UpdatedAt: now.Add(-72 * time.Hour)}, + {Path: "fresh.txt", UpdatedAt: now.Add(-24 * time.Hour)}, + }, PrunePlanOptions{Now: now, OlderThan: &olderThan}) + + if got, want := pruneCandidatePaths(plan.Pruned), "old.txt"; got != want { + t.Fatalf("pruned = %q, want %q", got, want) + } + if got, want := pruneCandidatePaths(plan.Preserved), "fresh.txt"; got != want { + t.Fatalf("preserved = %q, want %q", got, want) + } +} + +func TestPlanPruneKeepLatest(t *testing.T) { + now := time.Date(2026, 6, 8, 12, 0, 0, 0, time.UTC) + keepLatest := 2 + plan := PlanPrune([]PruneCandidate{ + {Path: "old.txt", UpdatedAt: now.Add(-72 * time.Hour)}, + {Path: "new.txt", UpdatedAt: now.Add(-1 * time.Hour)}, + {Path: "middle.txt", UpdatedAt: now.Add(-24 * time.Hour)}, + }, PrunePlanOptions{KeepLatest: &keepLatest}) + + if got, want := pruneCandidatePaths(plan.Pruned), "old.txt"; got != want { + t.Fatalf("pruned = %q, want %q", got, want) + } + if got, want := pruneCandidatePaths(plan.Preserved), "new.txt,middle.txt"; got != want { + t.Fatalf("preserved = %q, want %q", got, want) + } +} + +func TestPlanPruneCombinedPolicyPreservesLatestBeforeAgeCheck(t *testing.T) { + now := time.Date(2026, 6, 8, 12, 0, 0, 0, time.UTC) + olderThan := 48 * time.Hour + keepLatest := 1 + plan := PlanPrune([]PruneCandidate{ + {Path: "oldest.txt", UpdatedAt: now.Add(-96 * time.Hour)}, + {Path: "old.txt", UpdatedAt: now.Add(-72 * time.Hour)}, + {Path: "fresh.txt", UpdatedAt: now.Add(-24 * time.Hour)}, + }, PrunePlanOptions{Now: now, OlderThan: &olderThan, KeepLatest: &keepLatest}) + + if got, want := pruneCandidatePaths(plan.Pruned), "oldest.txt,old.txt"; got != want { + t.Fatalf("pruned = %q, want %q", got, want) + } + if got, want := pruneCandidatePaths(plan.Preserved), "fresh.txt"; got != want { + t.Fatalf("preserved = %q, want %q", got, want) + } +} + +func TestPlanPruneDeterministicTieBreaking(t *testing.T) { + now := time.Date(2026, 6, 8, 12, 0, 0, 0, time.UTC) + keepLatest := 1 + plan := PlanPrune([]PruneCandidate{ + {Path: "b.txt", UpdatedAt: now}, + {Path: "a.txt", UpdatedAt: now}, + {Path: "c.txt", UpdatedAt: now.Add(-time.Hour)}, + }, PrunePlanOptions{KeepLatest: &keepLatest}) + + if got, want := pruneCandidatePaths(plan.Preserved), "a.txt"; got != want { + t.Fatalf("preserved = %q, want %q", got, want) + } + if got, want := pruneCandidatePaths(plan.Pruned), "c.txt,b.txt"; got != want { + t.Fatalf("pruned = %q, want %q", got, want) + } +} + +func TestSharedRootPruneCandidatesPreserveOtherOwners(t *testing.T) { + sharedRoot := validSharedRootState(t) + scope := CurrentOwnerScope("reports", "archive") + candidates := SharedRootPruneCandidates(sharedRoot, scope) + + if got, want := pruneCandidatePaths(candidates), "report.md"; got != want { + t.Fatalf("candidates = %q, want %q", got, want) + } + if candidates[0].Owner == nil || *candidates[0].Owner != scope { + t.Fatalf("candidate owner = %#v, want current owner", candidates[0].Owner) + } +} + +func pruneCandidatePaths(candidates []PruneCandidate) string { + paths := make([]string, 0, len(candidates)) + for _, candidate := range candidates { + paths = append(paths, candidate.Path) + } + return strings.Join(paths, ",") +}