diff --git a/docs/internal/app.md b/docs/internal/app.md index 36466ee..f111625 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, 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. +`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/execution, and HTTP upload serving. It coordinates config loading, secret resolution, backend construction, source discovery, destination selection, publish planning/execution, state repair reporting, retention prune 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`, `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. +Outputs include `RunReport`, `ReconcileStateReport`, `PrunePlanReport`, `PruneReport`, 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 @@ -38,13 +38,15 @@ Reconcile-state workflows load one configured pipeline/destination selector, ope 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. +Prune execution loads one configured pipeline/destination selector, opens that destination root, parses the root `.distributor.json`, and builds a plan from the destination retention policy. Dry-run returns the same planned and preserved managed output records without deleting files or rewriting state. Apply mode deletes only planned managed output paths, never unmanaged files or `.distributor.json`, then removes confirmed deleted records from state and updates the state timestamp. If a delete fails after earlier deletes succeeded, it rewrites state only for the confirmed deletions and preserves records for the failed and unattempted outputs so a retry remains accurate. + 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. ## Skip And Resume Behavior -Fan-out destinations are independent. A destination failure is recorded and does not prevent later destinations from being attempted. Run dry-run builds plans and reports without destination writes, destination state writes, notifier calls, or SSH known-host persistence. Reconcile-state dry-run reports missing managed records and unmanaged entries without rewriting state. +Fan-out destinations are independent. A destination failure is recorded and does not prevent later destinations from being attempted. Run dry-run builds plans and reports without destination writes, destination state writes, notifier calls, or SSH known-host persistence. Reconcile-state dry-run reports missing managed records and unmanaged entries without rewriting state. Prune dry-run reports planned managed output deletes without deleting outputs or rewriting state. HTTP upload status is in memory. Accepted jobs move through accepted, queued, running, succeeded, or failed states and expire after configured retention. Upload idempotency records are also memory-only, expire with the completed status record for their accepted run, and are cleared by process restart. @@ -54,6 +56,8 @@ Runtime setup fails for config load, config validation, secret loading, or crede Reconcile-state setup fails unless the caller supplies a pipeline id and destination id that select one configured destination root. Single-owner state must match that pipeline/destination owner. Shared-root all-owner repair still uses the selected destination to identify the root, then applies repair across owners inside that root. Invalid, unreadable, or ambiguous state fails before any rewrite. +Prune setup fails unless the caller supplies a pipeline id and destination id that select one configured destination root. Single-owner state must match that pipeline/destination owner. Shared-root pruning is scoped to the selected owner and preserves unrelated owners. Invalid, unreadable, or ambiguous state fails before deletes or rewrites. Delete failures return a report with confirmed deletions and the failed output. + HTTP upload startup fails if upload tokens are missing, empty, or duplicated. Upload requests can fail during authentication, idempotency-key validation, content-type validation, idempotency conflict checks, queue admission, archive staging, source validation, or later publish execution. ## Tests To Inspect @@ -76,4 +80,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. +- Prune execution deletes managed output paths only and preserves failed records for retry. diff --git a/docs/internal/state.md b/docs/internal/state.md index 8befb76..4c2d236 100644 --- a/docs/internal/state.md +++ b/docs/internal/state.md @@ -50,7 +50,7 @@ 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. +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. App-level prune execution uses the missing-output removal helpers to remove only confirmed deleted records after storage deletion succeeds. ## Failure Behavior diff --git a/docs/internal/storage.md b/docs/internal/storage.md index 9ab5b83..f19019e 100644 --- a/docs/internal/storage.md +++ b/docs/internal/storage.md @@ -26,15 +26,15 @@ Local, SSH/SFTP, and S3-compatible adapters implement `storage.Backend`. `intern ## State And Manifest Behavior -Storage owns `.distributor.json` path helpers through `StateFileName`, `StatePath`, and `ManagedBundleTargets`. It does not parse source manifests or destination state. +Storage owns `.distributor.json` path helpers through `StateFileName`, `StatePath`, `ManagedOutputTargets`, and `ManagedBundleTargets`. It does not parse source manifests or destination state. Logical paths are slash-separated and relative to a backend root. Prefix validation allows an empty prefix to mean the backend root; file path validation requires a non-empty path. -Reconcile-state callers use `Stat` to check whether managed output paths still exist and bounded recursive `Walk` to report unmanaged entries under a selected destination root. Storage does not decide whether entries are managed; callers compare entries against destination state. +Reconcile-state callers use `Stat` to check whether managed output paths still exist and bounded recursive `Walk` to report unmanaged entries under a selected destination root. Prune execution callers use `DeleteManagedOutputs` for selected managed output records so `.distributor.json` is not part of the deletion target set. Storage does not decide whether entries are managed; callers compare entries against destination state. ## Skip And Resume Behavior -Storage has no publication skip policy. It supplies `HasAny` for unmanaged-content checks, `Stat` and `Walk` for state repair inspection, `DeleteManagedBundle` target construction for normal replacement cleanup, and `DeletePrefix` semantics for explicit forced replacement. +Storage has no publication skip policy. It supplies `HasAny` for unmanaged-content checks, `Stat` and `Walk` for state repair inspection, `DeleteManagedOutputs` target construction for managed output cleanup, `DeleteManagedBundle` target construction for normal replacement cleanup, and `DeletePrefix` semantics for explicit forced replacement. ## Failure Behavior @@ -54,6 +54,6 @@ Storage errors use typed categories: not found, already exists, not empty, inval - Logical paths are clean relative slash-separated paths confined to the backend root. - Core packages never import concrete adapters. - `storage.List` returns deterministic sorted entries. -- Managed deletion targets are recorded outputs plus `.distributor.json`. +- Managed output deletion targets are recorded outputs only; managed bundle deletion targets are recorded outputs plus `.distributor.json`. - Prefix deletion is bounded to the requested logical prefix. - Runtime registration remains app-owned. diff --git a/internal/app/prune.go b/internal/app/prune.go index 377ff06..114dbba 100644 --- a/internal/app/prune.go +++ b/internal/app/prune.go @@ -1,13 +1,23 @@ package app import ( + "context" "fmt" "time" "gitea.maximumdirect.net/eric/distributor/internal/config" "gitea.maximumdirect.net/eric/distributor/internal/state" + "gitea.maximumdirect.net/eric/distributor/internal/storage" ) +type PruneOptions struct { + ConfigPath string + PipelineID string + DestinationID string + DryRun bool + Now time.Time +} + type PrunePlanOptions struct { PipelineID string DestinationID string @@ -24,6 +34,23 @@ type PrunePlanReport struct { PreservedOutputs []PruneOutputRecord `json:"preserved_outputs"` } +type PruneReport struct { + PipelineID string `json:"pipeline_id"` + DestinationID string `json:"destination_id"` + Backend string `json:"backend"` + RootPath string `json:"root_path"` + OwnerScope PruneOwnerScope `json:"owner_scope"` + Enabled bool `json:"enabled"` + CheckedCount int `json:"checked_count"` + PlannedOutputs []PruneOutputRecord `json:"planned_outputs"` + DeletedOutputs []PruneOutputRecord `json:"deleted_outputs"` + PreservedOutputs []PruneOutputRecord `json:"preserved_outputs"` + FailedOutput *PruneOutputRecord `json:"failed_output,omitempty"` + StateChanged bool `json:"state_changed"` + WouldChange bool `json:"would_change"` + DryRun bool `json:"dry_run"` +} + type PruneOwnerScope struct { PipelineID string `json:"pipeline_id"` DestinationID string `json:"destination_id"` @@ -35,6 +62,167 @@ type PruneOutputRecord struct { Owner *PruneOwnerScope `json:"owner,omitempty"` } +func Prune(ctx context.Context, options PruneOptions) (PruneReport, error) { + if err := ctx.Err(); err != nil { + return PruneReport{}, err + } + + setup, err := loadRuntimeSetup(options.ConfigPath) + if err != nil { + return PruneReport{}, err + } + return pruneSetup(ctx, setup, options) +} + +func pruneConfigWithBackendFactory(ctx context.Context, cfg config.Config, options PruneOptions, provider backendFactoryProvider) (PruneReport, error) { + setup, err := runtimeSetupFromConfig("", cfg) + if err != nil { + return PruneReport{}, err + } + return pruneSetupWithBackendFactory(ctx, setup, options, provider) +} + +func pruneSetup(ctx context.Context, setup runtimeSetup, options PruneOptions) (PruneReport, error) { + return pruneSetupWithBackendFactory(ctx, setup, options, newBackendFactoryWithEnvironment) +} + +func pruneSetupWithBackendFactory(ctx context.Context, setup runtimeSetup, options PruneOptions, provider backendFactoryProvider) (PruneReport, error) { + if err := requirePruneScope(options); err != nil { + return PruneReport{}, err + } + pipeline, ok := findPipeline(setup.Config, options.PipelineID) + if !ok { + return PruneReport{}, PipelineNotFoundError{ID: options.PipelineID} + } + destination, ok := findDestination(pipeline, options.DestinationID) + if !ok { + return PruneReport{}, fmt.Errorf("pipeline %s destination %s not found", options.PipelineID, options.DestinationID) + } + + backends := provider(setup.Environment) + destinationBackend, err := backends.openDestination(ctx, destination) + if err != nil { + return PruneReport{}, err + } + defer closeBackend(destinationBackend) + + return executePrune(ctx, destinationBackend, pipeline, destination, options) +} + +func requirePruneScope(options PruneOptions) error { + if options.PipelineID == "" { + return fmt.Errorf("pipeline id is required") + } + if options.DestinationID == "" { + return fmt.Errorf("destination id is required") + } + return nil +} + +func executePrune(ctx context.Context, backend storage.Backend, pipeline config.Pipeline, destination config.Destination, options PruneOptions) (PruneReport, error) { + now := options.Now + if now.IsZero() { + now = time.Now().UTC() + } else { + now = now.UTC() + } + + statePath, err := storage.StatePath("") + if err != nil { + return PruneReport{}, err + } + data, err := backend.ReadFile(ctx, statePath) + if err != nil { + return PruneReport{}, err + } + document, err := state.ParseDocument(data) + if err != nil { + return PruneReport{}, err + } + + plan, err := PlanPrune(document, destination.Retention.Prune, PrunePlanOptions{ + PipelineID: pipeline.ID, + DestinationID: destination.ID, + Now: now, + }) + if err != nil { + return PruneReport{}, err + } + report := PruneReport{ + PipelineID: pipeline.ID, + DestinationID: destination.ID, + Backend: destination.Backend, + RootPath: destinationRootPath(destination), + OwnerScope: plan.OwnerScope, + Enabled: plan.Enabled, + CheckedCount: plan.CheckedCount, + PlannedOutputs: plan.PrunedOutputs, + DeletedOutputs: []PruneOutputRecord{}, + PreservedOutputs: plan.PreservedOutputs, + DryRun: options.DryRun, + } + report.WouldChange = options.DryRun && len(report.PlannedOutputs) > 0 + if options.DryRun || len(report.PlannedOutputs) == 0 { + return report, nil + } + + deletedPaths := make([]string, 0, len(report.PlannedOutputs)) + for _, output := range report.PlannedOutputs { + err := backend.DeleteManagedOutputs(ctx, "", []string{output.Path}, storage.DeleteOptions{ + IgnoreMissing: true, + PruneEmptyDirs: true, + }) + if err != nil { + failed := output + report.FailedOutput = &failed + if len(deletedPaths) > 0 { + changed, writeErr := removePrunedStateRecords(ctx, backend, statePath, document, state.CurrentOwnerScope(pipeline.ID, destination.ID), deletedPaths, now) + report.StateChanged = changed + report.DeletedOutputs = report.PlannedOutputs[:len(deletedPaths)] + if writeErr != nil { + return report, writeErr + } + } + return report, err + } + deletedPaths = append(deletedPaths, output.Path) + } + + changed, err := removePrunedStateRecords(ctx, backend, statePath, document, state.CurrentOwnerScope(pipeline.ID, destination.ID), deletedPaths, now) + report.StateChanged = changed + report.DeletedOutputs = report.PlannedOutputs + return report, err +} + +func removePrunedStateRecords(ctx context.Context, backend storage.Backend, statePath string, document state.StateDocument, scope state.OwnerScope, paths []string, now time.Time) (bool, error) { + if len(paths) == 0 { + return false, nil + } + if document.SingleOwner != nil { + next, changed := state.RemoveMissingOutputs(*document.SingleOwner, paths) + if !changed { + return false, nil + } + next.UpdatedAt = now + if err := state.Validate(next); err != nil { + return false, err + } + return true, writeRepairedState(ctx, backend, statePath, next) + } + if document.SharedRoot != nil { + next, changed := state.RemoveMissingSharedRootOwnerOutputs(*document.SharedRoot, scope, paths) + if !changed { + return false, nil + } + next.UpdatedAt = now + if err := state.ValidateSharedRoot(next); err != nil { + return false, err + } + return true, writeRepairedState(ctx, backend, statePath, next) + } + return false, fmt.Errorf("destination state document is empty") +} + func PlanPrune(document state.StateDocument, policy config.PrunePolicy, options PrunePlanOptions) (PrunePlanReport, error) { scope := state.CurrentOwnerScope(options.PipelineID, options.DestinationID) report := PrunePlanReport{ diff --git a/internal/app/prune_test.go b/internal/app/prune_test.go index b910be9..35a4d02 100644 --- a/internal/app/prune_test.go +++ b/internal/app/prune_test.go @@ -1,12 +1,16 @@ package app import ( + "context" + "encoding/json" "strings" "testing" "time" "gitea.maximumdirect.net/eric/distributor/internal/config" "gitea.maximumdirect.net/eric/distributor/internal/state" + "gitea.maximumdirect.net/eric/distributor/internal/storage" + "gitea.maximumdirect.net/eric/distributor/internal/storage/fake" "gitea.maximumdirect.net/eric/distributor/internal/testutil" ) @@ -72,6 +76,134 @@ func TestPlanPruneSharedRootCurrentOwnerOnly(t *testing.T) { } } +func TestPruneDryRunReportsPlannedDeletesWithoutDeletingOrRewritingState(t *testing.T) { + now := time.Date(2026, 6, 8, 12, 0, 0, 0, time.UTC) + backend := fake.New() + cfg := pruneS3Config(t, pruneOlderThanPolicy(48*time.Hour)) + original := pruneSingleOwnerState(now) + writeFakeSingleOwnerStateForPrune(t, backend, original) + testutil.WriteFakeFile(t, backend, "unmanaged.txt", "keep") + + report, err := pruneConfigWithBackendFactory(context.Background(), cfg, PruneOptions{ + PipelineID: "reports", + DestinationID: "archive", + DryRun: true, + Now: now, + }, fakeBackendFactoryProvider(t, map[string]storage.Backend{"s3:reports": backend})) + if err != nil { + t.Fatalf("pruneConfigWithBackendFactory() error = %v", err) + } + if !report.WouldChange || report.StateChanged || len(report.DeletedOutputs) != 0 { + t.Fatalf("report would_change=%t state_changed=%t deleted=%d, want dry-run only", report.WouldChange, report.StateChanged, len(report.DeletedOutputs)) + } + if got, want := pruneRecordPaths(report.PlannedOutputs), "old.txt"; got != want { + t.Fatalf("planned outputs = %q, want %q", got, want) + } + testutil.AssertFakeFile(t, backend, "old.txt", "managed") + testutil.AssertFakeFile(t, backend, "fresh.txt", "managed") + testutil.AssertFakeFile(t, backend, "unmanaged.txt", "keep") + destinationState := readFakeSingleOwnerState(t, backend) + if got := strings.Join(state.ManagedOutputPaths(destinationState), ","); got != "old.txt,fresh.txt" { + t.Fatalf("state outputs = %q, want original outputs", got) + } + if !destinationState.UpdatedAt.Equal(original.UpdatedAt) { + t.Fatalf("state updated_at = %s, want original %s", destinationState.UpdatedAt, original.UpdatedAt) + } +} + +func TestPruneApplyDeletesOnlyManagedOutputsAndUpdatesState(t *testing.T) { + now := time.Date(2026, 6, 8, 12, 0, 0, 0, time.UTC) + backend := fake.New() + cfg := pruneS3Config(t, pruneOlderThanPolicy(48*time.Hour)) + writeFakeSingleOwnerStateForPrune(t, backend, pruneSingleOwnerState(now)) + testutil.WriteFakeFile(t, backend, "unmanaged.txt", "keep") + + report, err := pruneConfigWithBackendFactory(context.Background(), cfg, PruneOptions{ + PipelineID: "reports", + DestinationID: "archive", + Now: now, + }, fakeBackendFactoryProvider(t, map[string]storage.Backend{"s3:reports": backend})) + if err != nil { + t.Fatalf("pruneConfigWithBackendFactory() error = %v", err) + } + if !report.StateChanged || report.WouldChange { + t.Fatalf("report state_changed=%t would_change=%t, want applied change", report.StateChanged, report.WouldChange) + } + if got, want := pruneRecordPaths(report.DeletedOutputs), "old.txt"; got != want { + t.Fatalf("deleted outputs = %q, want %q", got, want) + } + testutil.AssertFakeMissing(t, backend, "old.txt") + testutil.AssertFakeFile(t, backend, "fresh.txt", "managed") + testutil.AssertFakeFile(t, backend, "unmanaged.txt", "keep") + assertFakeStateExists(t, backend) + destinationState := readFakeSingleOwnerState(t, backend) + if got := strings.Join(state.ManagedOutputPaths(destinationState), ","); got != "fresh.txt" { + t.Fatalf("state outputs = %q, want fresh.txt", got) + } + if !destinationState.UpdatedAt.Equal(now) { + t.Fatalf("state updated_at = %s, want %s", destinationState.UpdatedAt, now) + } +} + +func TestPruneApplyPreservesStateForFailedDeletes(t *testing.T) { + now := time.Date(2026, 6, 8, 12, 0, 0, 0, time.UTC) + backend := fake.New() + keepLatest := 0 + cfg := pruneS3Config(t, config.PrunePolicy{Enabled: true, KeepLatest: &keepLatest}) + writeFakeSingleOwnerStateForPrune(t, backend, pruneSingleOwnerState(now)) + failingBackend := failingDeleteBackend{Backend: backend, failPath: "fresh.txt"} + + report, err := pruneConfigWithBackendFactory(context.Background(), cfg, PruneOptions{ + PipelineID: "reports", + DestinationID: "archive", + Now: now, + }, fakeBackendFactoryProvider(t, map[string]storage.Backend{"s3:reports": failingBackend})) + if err == nil { + t.Fatal("pruneConfigWithBackendFactory() error = nil, want delete failure") + } + if report.FailedOutput == nil || report.FailedOutput.Path != "fresh.txt" { + t.Fatalf("failed output = %#v, want fresh.txt", report.FailedOutput) + } + if got, want := pruneRecordPaths(report.DeletedOutputs), "old.txt"; got != want { + t.Fatalf("deleted outputs = %q, want %q", got, want) + } + testutil.AssertFakeMissing(t, backend, "old.txt") + testutil.AssertFakeFile(t, backend, "fresh.txt", "managed") + assertFakeStateExists(t, backend) + destinationState := readFakeSingleOwnerState(t, backend) + if got := strings.Join(state.ManagedOutputPaths(destinationState), ","); got != "fresh.txt" { + t.Fatalf("state outputs = %q, want only failed output preserved", got) + } +} + +func TestPruneSharedRootPreservesOtherOwnersWhenScopedToCurrentOwner(t *testing.T) { + now := time.Date(2026, 6, 8, 12, 0, 0, 0, time.UTC) + backend := fake.New() + keepLatest := 0 + cfg := pruneS3Config(t, config.PrunePolicy{Enabled: true, KeepLatest: &keepLatest}) + writeFakeSharedRootStateForApp(t, backend, pruneSharedRootState(now)) + testutil.WriteFakeFile(t, backend, "unmanaged.txt", "keep") + + report, err := pruneConfigWithBackendFactory(context.Background(), cfg, PruneOptions{ + PipelineID: "reports", + DestinationID: "archive", + Now: now, + }, fakeBackendFactoryProvider(t, map[string]storage.Backend{"s3:reports": backend})) + if err != nil { + t.Fatalf("pruneConfigWithBackendFactory() error = %v", err) + } + if got, want := pruneRecordPaths(report.DeletedOutputs), "archive.txt"; got != want { + t.Fatalf("deleted outputs = %q, want %q", got, want) + } + testutil.AssertFakeMissing(t, backend, "archive.txt") + testutil.AssertFakeFile(t, backend, "html.txt", "old") + testutil.AssertFakeFile(t, backend, "unmanaged.txt", "keep") + sharedRoot := readFakeSharedRootStateForApp(t, backend) + if got := strings.Join(sharedRoot.AllManagedOutputPaths(), ","); got != "html.txt" { + t.Fatalf("shared-root outputs = %q, want other owner output preserved", got) + } +} + func pruneSingleOwnerState(now time.Time) state.DistributorState { manifest := testutil.ValidManifest(testutil.BundleOptions{}) publishedAt := now.Add(-96 * time.Hour) @@ -160,3 +292,62 @@ func pruneRecordPaths(records []PruneOutputRecord) string { } return strings.Join(paths, ",") } + +func pruneS3Config(t *testing.T, policy config.PrunePolicy) config.Config { + t.Helper() + cfg := config.Config{Pipelines: []config.Pipeline{{ + ID: "reports", + Source: config.Backend{Backend: config.BackendLocal, Path: t.TempDir()}, + Destinations: []config.Destination{{ + ID: "archive", + Backend: config.BackendS3, + Bucket: "reports", + Retention: config.RetentionPolicy{ + Prune: policy, + }, + }}, + }}} + config.ApplyDefaults(&cfg) + return cfg +} + +func pruneOlderThanPolicy(duration time.Duration) config.PrunePolicy { + value := config.Duration(duration) + return config.PrunePolicy{ + Enabled: true, + OlderThan: &value, + } +} + +func writeFakeSingleOwnerStateForPrune(t *testing.T, backend *fake.Backend, destinationState state.DistributorState) { + t.Helper() + data, err := json.MarshalIndent(destinationState, "", " ") + if err != nil { + t.Fatalf("marshal single-owner state: %v", err) + } + testutil.WriteFakeFile(t, backend, storage.StateFileName, string(append(data, '\n'))) + for _, output := range destinationState.Outputs { + testutil.WriteFakeFile(t, backend, output.Path, "managed") + } +} + +func assertFakeStateExists(t *testing.T, backend *fake.Backend) { + t.Helper() + if _, err := backend.Stat(context.Background(), storage.StateFileName); err != nil { + t.Fatalf("state file stat error = %v", err) + } +} + +type failingDeleteBackend struct { + storage.Backend + failPath string +} + +func (b failingDeleteBackend) DeleteManagedOutputs(ctx context.Context, bundlePath string, managedOutputPaths []string, opts storage.DeleteOptions) error { + for _, path := range managedOutputPaths { + if path == b.failPath { + return storage.NewError(storage.OpDeleteManagedOutputs, "fake", path, storage.ErrPermission, nil) + } + } + return b.Backend.DeleteManagedOutputs(ctx, bundlePath, managedOutputPaths, opts) +}