diff --git a/internal/app/prune.go b/internal/app/prune.go index d69e16b..a099c86 100644 --- a/internal/app/prune.go +++ b/internal/app/prune.go @@ -211,24 +211,13 @@ func removePrunedStateRecords(ctx context.Context, backend storage.Backend, stat if len(paths) == 0 { return false, nil } - if document.SingleOwner != nil { - next, changed := state.RemoveMissingOutputs(*document.SingleOwner, paths) + if document.Catalog != nil { + next, changed := state.RemoveMissingCatalogOwnerOutputs(*document.Catalog, scope, 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 { + if err := state.ValidateCatalog(next); err != nil { return false, err } return true, writeRepairedState(ctx, backend, statePath, next) @@ -266,15 +255,8 @@ func PlanPrune(document state.StateDocument, policy config.PrunePolicy, options } 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 + if document.Catalog != nil { + return state.CatalogPruneCandidates(*document.Catalog, scope), nil } return nil, unsupportedStateDocumentError(document) } @@ -283,8 +265,8 @@ func unsupportedStateDocumentError(document state.StateDocument) error { if document.SupersededLegacy != nil { return fmt.Errorf("destination state schema_version %d is superseded legacy state", document.SupersededLegacy.SchemaVersion) } - if document.Catalog != nil { - return fmt.Errorf("catalog destination state is not supported by this command") + if document.SingleOwner != nil || document.SharedRoot != nil { + return fmt.Errorf("legacy destination state is not supported by this command") } return fmt.Errorf("destination state document is empty") } diff --git a/internal/app/prune_test.go b/internal/app/prune_test.go index dad454f..6dedb87 100644 --- a/internal/app/prune_test.go +++ b/internal/app/prune_test.go @@ -15,7 +15,7 @@ import ( ) func TestPlanPruneDisabledPolicy(t *testing.T) { - document := state.StateDocument{SingleOwner: &state.DistributorState{}} + document := state.StateDocument{Catalog: &state.CatalogState{}} report, err := PlanPrune(document, config.PrunePolicy{}, PrunePlanOptions{ PipelineID: "reports", DestinationID: "archive", @@ -28,12 +28,12 @@ func TestPlanPruneDisabledPolicy(t *testing.T) { } } -func TestPlanPruneSingleOwnerOutputs(t *testing.T) { +func TestPlanPruneCatalogOutputs(t *testing.T) { now := time.Date(2026, 6, 8, 12, 0, 0, 0, time.UTC) olderThan := config.Duration(48 * time.Hour) - destinationState := pruneSingleOwnerState(now) + catalog := pruneCatalogState(now) - report, err := PlanPrune(state.StateDocument{SingleOwner: &destinationState}, config.PrunePolicy{ + report, err := PlanPrune(state.StateDocument{Catalog: &catalog}, config.PrunePolicy{ Enabled: true, OlderThan: &olderThan, }, PrunePlanOptions{ @@ -52,12 +52,12 @@ func TestPlanPruneSingleOwnerOutputs(t *testing.T) { } } -func TestPlanPruneSharedRootCurrentOwnerOnly(t *testing.T) { +func TestPlanPruneCatalogCurrentOwnerOnly(t *testing.T) { now := time.Date(2026, 6, 8, 12, 0, 0, 0, time.UTC) keepLatest := 0 - sharedRoot := pruneSharedRootState(now) + catalog := pruneCatalogState(now) - report, err := PlanPrune(state.StateDocument{SharedRoot: &sharedRoot}, config.PrunePolicy{ + report, err := PlanPrune(state.StateDocument{Catalog: &catalog}, config.PrunePolicy{ Enabled: true, KeepLatest: &keepLatest, }, PrunePlanOptions{ @@ -68,21 +68,20 @@ func TestPlanPruneSharedRootCurrentOwnerOnly(t *testing.T) { if err != nil { t.Fatalf("PlanPrune() error = %v", err) } - if got, want := report.CheckedCount, 1; got != want { + if got, want := report.CheckedCount, 2; got != want { t.Fatalf("checked count = %d, want %d", got, want) } - if got, want := pruneRecordPaths(report.PrunedOutputs), "archive.txt"; got != want { + if got, want := pruneRecordPaths(report.PrunedOutputs), "old.txt,fresh.txt"; got != want { t.Fatalf("pruned = %q, want %q", got, want) } } func TestPruneDryRunReportsPlannedDeletesWithoutDeletingOrRewritingState(t *testing.T) { - t.Skip("catalog prune execution is covered by the catalog maintenance work") 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) + original := pruneCatalogState(now) + writeFakeCatalogState(t, backend, original) testutil.WriteFakeFile(t, backend, "unmanaged.txt", "keep") report, err := pruneConfigWithBackendFactory(context.Background(), cfg, PruneOptions{ @@ -103,21 +102,20 @@ func TestPruneDryRunReportsPlannedDeletesWithoutDeletingOrRewritingState(t *test 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" { + catalog := readFakeCatalogState(t, backend) + if got := strings.Join(state.CatalogManagedOutputPaths(catalog), ","); got != "old.txt,fresh.txt,html.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) + if !catalog.UpdatedAt.Equal(original.UpdatedAt) { + t.Fatalf("state updated_at = %s, want original %s", catalog.UpdatedAt, original.UpdatedAt) } } func TestPruneApplyDeletesOnlyManagedOutputsAndUpdatesState(t *testing.T) { - t.Skip("catalog prune execution is covered by the catalog maintenance work") 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)) + writeFakeCatalogState(t, backend, pruneCatalogState(now)) testutil.WriteFakeFile(t, backend, "unmanaged.txt", "keep") report, err := pruneConfigWithBackendFactory(context.Background(), cfg, PruneOptions{ @@ -136,24 +134,27 @@ func TestPruneApplyDeletesOnlyManagedOutputsAndUpdatesState(t *testing.T) { } testutil.AssertFakeMissing(t, backend, "old.txt") testutil.AssertFakeFile(t, backend, "fresh.txt", "managed") + testutil.AssertFakeFile(t, backend, "html.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" { + catalog := readFakeCatalogState(t, backend) + if got := strings.Join(state.CatalogManagedOutputPaths(catalog), ","); got != "fresh.txt,html.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) + if catalog.SchemaVersion != state.CatalogSchemaVersion { + t.Fatalf("state schema_version = %d, want %d", catalog.SchemaVersion, state.CatalogSchemaVersion) + } + if !catalog.UpdatedAt.Equal(now) { + t.Fatalf("state updated_at = %s, want %s", catalog.UpdatedAt, now) } } func TestPruneApplyPreservesStateForFailedDeletes(t *testing.T) { - t.Skip("catalog prune execution is covered by the catalog maintenance work") 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)) + writeFakeCatalogState(t, backend, pruneCatalogState(now)) failingBackend := failingDeleteBackend{Backend: backend, failPath: "fresh.txt"} report, err := pruneConfigWithBackendFactory(context.Background(), cfg, PruneOptions{ @@ -173,19 +174,17 @@ func TestPruneApplyPreservesStateForFailedDeletes(t *testing.T) { 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" { + catalog := readFakeCatalogState(t, backend) + if got := strings.Join(state.CatalogManagedOutputPaths(catalog), ","); got != "fresh.txt,html.txt" { t.Fatalf("state outputs = %q, want only failed output preserved", got) } } -func TestPruneSharedRootPreservesOtherOwnersWhenScopedToCurrentOwner(t *testing.T) { - t.Skip("catalog prune execution is covered by the catalog maintenance work") +func TestPrunePreservesOtherOwnersWhenScopedToCurrentOwner(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)) + cfg := pruneS3Config(t, pruneOlderThanPolicy(48*time.Hour)) + writeFakeCatalogState(t, backend, pruneCatalogState(now)) testutil.WriteFakeFile(t, backend, "unmanaged.txt", "keep") report, err := pruneConfigWithBackendFactory(context.Background(), cfg, PruneOptions{ @@ -196,93 +195,56 @@ func TestPruneSharedRootPreservesOtherOwnersWhenScopedToCurrentOwner(t *testing. if err != nil { t.Fatalf("pruneConfigWithBackendFactory() error = %v", err) } - if got, want := pruneRecordPaths(report.DeletedOutputs), "archive.txt"; got != want { + if got, want := pruneRecordPaths(report.DeletedOutputs), "old.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.AssertFakeMissing(t, backend, "old.txt") + testutil.AssertFakeFile(t, backend, "html.txt", "managed") 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) + catalog := readFakeCatalogState(t, backend) + if got := strings.Join(state.CatalogManagedOutputPaths(catalog), ","); got != "fresh.txt,html.txt" { + t.Fatalf("catalog outputs = %q, want other owner output preserved", got) } } -func pruneSingleOwnerState(now time.Time) state.DistributorState { +func pruneCatalogState(now time.Time) state.CatalogState { 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}, + createdAt := now.Add(-96 * time.Hour) + source := state.CatalogSourceIdentity{ID: manifest.ID, Digest: manifest.Digest, Created: manifest.Created} + return state.CatalogState{ + SchemaVersion: state.CatalogSchemaVersion, 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", + CreatedAt: createdAt, + UpdatedAt: createdAt, + State: state.StatePolicy{Mode: state.StateModeCatalog}, + Outputs: []state.CatalogOutputFile{{ + Path: "old.txt", + PipelineID: "reports", + DestinationID: "archive", + Source: source, 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", + Path: "fresh.txt", + PipelineID: "reports", + DestinationID: "archive", + Source: source, + Kind: state.OutputKindSource, + SHA256: manifest.Files[0].SHA256, + Size: manifest.Files[0].Size, + CreatedAt: now.Add(-24 * time.Hour), + UpdatedAt: now.Add(-24 * time.Hour), + }, { + Path: "html.txt", + PipelineID: "reports", + DestinationID: "html", + Source: source, 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), }}, @@ -323,18 +285,31 @@ func pruneOlderThanPolicy(duration time.Duration) config.PrunePolicy { } } -func writeFakeSingleOwnerStateForPrune(t *testing.T, backend *fake.Backend, destinationState state.DistributorState) { +func writeFakeCatalogState(t *testing.T, backend *fake.Backend, catalog state.CatalogState) { t.Helper() - data, err := json.MarshalIndent(destinationState, "", " ") + data, err := json.MarshalIndent(catalog, "", " ") if err != nil { - t.Fatalf("marshal single-owner state: %v", err) + t.Fatalf("marshal catalog state: %v", err) } testutil.WriteFakeFile(t, backend, storage.StateFileName, string(append(data, '\n'))) - for _, output := range destinationState.Outputs { + for _, output := range catalog.Outputs { testutil.WriteFakeFile(t, backend, output.Path, "managed") } } +func readFakeCatalogState(t *testing.T, backend *fake.Backend) state.CatalogState { + t.Helper() + data, err := backend.ReadFile(context.Background(), storage.StateFileName) + if err != nil { + t.Fatalf("read catalog state: %v", err) + } + catalog, err := state.ParseCatalog(data) + if err != nil { + t.Fatalf("parse catalog state: %v", err) + } + return catalog +} + func assertFakeStateExists(t *testing.T, backend *fake.Backend) { t.Helper() if _, err := backend.Stat(context.Background(), storage.StateFileName); err != nil { diff --git a/internal/app/reconcile_state.go b/internal/app/reconcile_state.go index f57d008..7fe0b98 100644 --- a/internal/app/reconcile_state.go +++ b/internal/app/reconcile_state.go @@ -158,65 +158,25 @@ func buildReconcileStateReport(ctx context.Context, backend storage.Backend, pip DryRun: options.DryRun, } scope := state.CurrentOwnerScope(pipeline.ID, destination.ID) - if document.SingleOwner != nil { - return reconcileSingleOwnerState(ctx, backend, statePath, *document.SingleOwner, scope, report, options) - } - if document.SharedRoot != nil { - return reconcileSharedRootState(ctx, backend, statePath, *document.SharedRoot, scope, report, options) + if document.Catalog != nil { + return reconcileCatalogState(ctx, backend, statePath, *document.Catalog, scope, report, options) } return ReconcileStateReport{}, unsupportedStateDocumentError(document) } -func reconcileSingleOwnerState(ctx context.Context, backend storage.Backend, statePath string, destinationState state.DistributorState, scope state.OwnerScope, report ReconcileStateReport, options ReconcileStateOptions) (ReconcileStateReport, error) { - if destinationState.PipelineID != scope.PipelineID || destinationState.DestinationID != scope.DestinationID { - return ReconcileStateReport{}, fmt.Errorf("state owner is %s/%s, not %s/%s", destinationState.PipelineID, destinationState.DestinationID, scope.PipelineID, scope.DestinationID) - } - report.StateSchema = destinationState.SchemaVersion - report.OwnerScope = &ReconcileStateOwnerScope{PipelineID: scope.PipelineID, DestinationID: scope.DestinationID} - managed := state.ManagedOutputPaths(destinationState) - missing, err := missingSingleOwnerOutputs(ctx, backend, destinationState.Outputs) - if err != nil { - return ReconcileStateReport{}, err - } - report.CheckedCount = len(managed) - report.MissingManagedOutputs = missing - unmanaged, err := unmanagedEntries(ctx, backend, managed) - if err != nil { - return ReconcileStateReport{}, err - } - report.UnmanagedEntries = unmanaged - report.WouldChange = options.DryRun && len(missing) > 0 - - if !options.DryRun && len(missing) > 0 { - missingPaths := missingReportPaths(missing) - next, changed := state.RemoveMissingOutputs(destinationState, missingPaths) - report.Changed = changed - if changed { - next.UpdatedAt = time.Now().UTC() - if err := state.Validate(next); err != nil { - return ReconcileStateReport{}, err - } - if err := writeRepairedState(ctx, backend, statePath, next); err != nil { - return ReconcileStateReport{}, err - } - } - } - return report, nil -} - -func reconcileSharedRootState(ctx context.Context, backend storage.Backend, statePath string, sharedRoot state.SharedRootState, scope state.OwnerScope, report ReconcileStateReport, options ReconcileStateOptions) (ReconcileStateReport, error) { - report.StateSchema = sharedRoot.SchemaVersion +func reconcileCatalogState(ctx context.Context, backend storage.Backend, statePath string, catalog state.CatalogState, scope state.OwnerScope, report ReconcileStateReport, options ReconcileStateOptions) (ReconcileStateReport, error) { + report.StateSchema = catalog.SchemaVersion report.OwnerScope = &ReconcileStateOwnerScope{ PipelineID: scope.PipelineID, DestinationID: scope.DestinationID, AllOwners: options.AllOwners, } - managed := sharedRoot.AllManagedOutputPaths() - outputs := sharedRoot.Outputs + managed := state.CatalogManagedOutputPaths(catalog) + outputs := catalog.Outputs if !options.AllOwners { - outputs = sharedRootOutputsForOwner(sharedRoot.Outputs, scope) + outputs = state.CatalogOutputsForOwner(catalog.Outputs, scope) } - missing, err := missingSharedRootOutputs(ctx, backend, outputs) + missing, err := missingCatalogOutputs(ctx, backend, outputs) if err != nil { return ReconcileStateReport{}, err } @@ -231,17 +191,17 @@ func reconcileSharedRootState(ctx context.Context, backend storage.Backend, stat if !options.DryRun && len(missing) > 0 { missingPaths := missingReportPaths(missing) - var next state.SharedRootState + var next state.CatalogState var changed bool if options.AllOwners { - next, changed = state.RemoveMissingSharedRootOutputs(sharedRoot, missingPaths) + next, changed = state.RemoveMissingCatalogOutputs(catalog, missingPaths) } else { - next, changed = state.RemoveMissingSharedRootOwnerOutputs(sharedRoot, scope, missingPaths) + next, changed = state.RemoveMissingCatalogOwnerOutputs(catalog, scope, missingPaths) } report.Changed = changed if changed { next.UpdatedAt = time.Now().UTC() - if err := state.ValidateSharedRoot(next); err != nil { + if err := state.ValidateCatalog(next); err != nil { return ReconcileStateReport{}, err } if err := writeRepairedState(ctx, backend, statePath, next); err != nil { @@ -252,21 +212,7 @@ func reconcileSharedRootState(ctx context.Context, backend storage.Backend, stat return report, nil } -func missingSingleOwnerOutputs(ctx context.Context, backend storage.Backend, outputs []state.OutputFile) ([]ReconcileStatePath, error) { - missing := make([]ReconcileStatePath, 0) - for _, output := range outputs { - if err := checkManagedOutput(ctx, backend, output.Path); err != nil { - if storage.IsNotFound(err) { - missing = append(missing, ReconcileStatePath{Path: output.Path, StorageStatus: "missing"}) - continue - } - return nil, err - } - } - return missing, nil -} - -func missingSharedRootOutputs(ctx context.Context, backend storage.Backend, outputs []state.SharedRootOutputFile) ([]ReconcileStatePath, error) { +func missingCatalogOutputs(ctx context.Context, backend storage.Backend, outputs []state.CatalogOutputFile) ([]ReconcileStatePath, error) { missing := make([]ReconcileStatePath, 0) for _, output := range outputs { if err := checkManagedOutput(ctx, backend, output.Path); err != nil { @@ -274,8 +220,8 @@ func missingSharedRootOutputs(ctx context.Context, backend storage.Backend, outp missing = append(missing, ReconcileStatePath{ Path: output.Path, OwnerScope: &ReconcileStateOwnerScope{ - PipelineID: output.Owner.PipelineID, - DestinationID: output.Owner.DestinationID, + PipelineID: output.PipelineID, + DestinationID: output.DestinationID, }, StorageStatus: "missing", }) @@ -323,16 +269,6 @@ func unmanagedEntries(ctx context.Context, backend storage.Backend, managedPaths return entries, nil } -func sharedRootOutputsForOwner(outputs []state.SharedRootOutputFile, scope state.OwnerScope) []state.SharedRootOutputFile { - selected := make([]state.SharedRootOutputFile, 0, len(outputs)) - for _, output := range outputs { - if output.Owner == scope { - selected = append(selected, output) - } - } - return selected -} - func missingReportPaths(missing []ReconcileStatePath) []string { paths := make([]string, 0, len(missing)) for _, item := range missing { diff --git a/internal/app/reconcile_state_test.go b/internal/app/reconcile_state_test.go index 17cc76e..b2ba9b7 100644 --- a/internal/app/reconcile_state_test.go +++ b/internal/app/reconcile_state_test.go @@ -2,7 +2,6 @@ package app import ( "context" - "encoding/json" "strings" "testing" "time" @@ -15,12 +14,11 @@ import ( ) func TestReconcileStateDryRunReportsMissingManagedOutputsWithoutRewrite(t *testing.T) { - t.Skip("catalog reconcile-state execution is covered by the catalog maintenance work") backend := fake.New() cfg := reconcileStateS3Config(t) - manifest := testutil.ValidManifest(testutil.BundleOptions{}) - testutil.WriteFakeDestinationState(t, backend, "", manifest, testutil.DestinationStateOptions{}) - if err := backend.DeleteManagedOutputs(context.Background(), "", []string{"summary.txt"}, storage.DeleteOptions{}); err != nil { + catalog := pruneCatalogState(time.Date(2026, 6, 8, 12, 0, 0, 0, time.UTC)) + writeFakeCatalogState(t, backend, catalog) + if err := backend.DeleteManagedOutputs(context.Background(), "", []string{"fresh.txt"}, storage.DeleteOptions{}); err != nil { t.Fatalf("delete managed output: %v", err) } testutil.WriteFakeFile(t, backend, "extra.txt", "unmanaged") @@ -36,25 +34,23 @@ func TestReconcileStateDryRunReportsMissingManagedOutputsWithoutRewrite(t *testi if !report.WouldChange || report.Changed { t.Fatalf("report changed=%t would_change=%t, want dry-run pending change", report.Changed, report.WouldChange) } - if got := reportPathList(report.MissingManagedOutputs); got != "summary.txt" { - t.Fatalf("missing outputs = %q, want summary.txt", got) + if got := reportPathList(report.MissingManagedOutputs); got != "fresh.txt" { + t.Fatalf("missing outputs = %q, want fresh.txt", got) } if got := entryPathList(report.UnmanagedEntries); got != "extra.txt" { t.Fatalf("unmanaged entries = %q, want extra.txt", got) } - destinationState := readFakeSingleOwnerState(t, backend) - if got := strings.Join(state.ManagedOutputPaths(destinationState), ","); got != "report.md,summary.txt" { + repaired := readFakeCatalogState(t, backend) + if got := strings.Join(state.CatalogManagedOutputPaths(repaired), ","); got != "old.txt,fresh.txt,html.txt" { t.Fatalf("state outputs = %q, want original outputs", got) } } func TestReconcileStateApplyRemovesMissingRecordsAndPreservesUnmanagedFiles(t *testing.T) { - t.Skip("catalog reconcile-state execution is covered by the catalog maintenance work") backend := fake.New() cfg := reconcileStateS3Config(t) - manifest := testutil.ValidManifest(testutil.BundleOptions{}) - testutil.WriteFakeDestinationState(t, backend, "", manifest, testutil.DestinationStateOptions{}) - if err := backend.DeleteManagedOutputs(context.Background(), "", []string{"summary.txt"}, storage.DeleteOptions{}); err != nil { + writeFakeCatalogState(t, backend, pruneCatalogState(time.Date(2026, 6, 8, 12, 0, 0, 0, time.UTC))) + if err := backend.DeleteManagedOutputs(context.Background(), "", []string{"fresh.txt"}, storage.DeleteOptions{}); err != nil { t.Fatalf("delete managed output: %v", err) } testutil.WriteFakeFile(t, backend, "extra.txt", "unmanaged") @@ -69,12 +65,15 @@ func TestReconcileStateApplyRemovesMissingRecordsAndPreservesUnmanagedFiles(t *t if !report.Changed || report.WouldChange { t.Fatalf("report changed=%t would_change=%t, want applied change", report.Changed, report.WouldChange) } - destinationState := readFakeSingleOwnerState(t, backend) - if err := state.Validate(destinationState); err != nil { - t.Fatalf("Validate() repaired state error = %v", err) + repaired := readFakeCatalogState(t, backend) + if err := state.ValidateCatalog(repaired); err != nil { + t.Fatalf("ValidateCatalog() repaired state error = %v", err) } - if got := strings.Join(state.ManagedOutputPaths(destinationState), ","); got != "report.md" { - t.Fatalf("state outputs = %q, want report.md", got) + if got := strings.Join(state.CatalogManagedOutputPaths(repaired), ","); got != "old.txt,html.txt" { + t.Fatalf("state outputs = %q, want old.txt,html.txt", got) + } + if repaired.SchemaVersion != state.CatalogSchemaVersion { + t.Fatalf("state schema_version = %d, want %d", repaired.SchemaVersion, state.CatalogSchemaVersion) } testutil.AssertFakeFile(t, backend, "extra.txt", "unmanaged") } @@ -101,13 +100,11 @@ func TestReconcileStateInvalidStateFailsWithoutRewrite(t *testing.T) { } } -func TestReconcileStateSharedRootOwnerScopeRepairsCurrentOwnerOnly(t *testing.T) { - t.Skip("catalog reconcile-state execution is covered by the catalog maintenance work") +func TestReconcileStateOwnerScopeRepairsCurrentOwnerOnly(t *testing.T) { backend := fake.New() cfg := reconcileStateS3Config(t) - sharedRoot := reconcileSharedRootFixture(t) - writeFakeSharedRootStateForApp(t, backend, sharedRoot) - if err := backend.DeleteManagedOutputs(context.Background(), "", []string{"report.md", "report.html"}, storage.DeleteOptions{}); err != nil { + writeFakeCatalogState(t, backend, pruneCatalogState(time.Date(2026, 6, 8, 12, 0, 0, 0, time.UTC))) + if err := backend.DeleteManagedOutputs(context.Background(), "", []string{"old.txt", "html.txt"}, storage.DeleteOptions{}); err != nil { t.Fatalf("delete managed outputs: %v", err) } @@ -121,19 +118,17 @@ func TestReconcileStateSharedRootOwnerScopeRepairsCurrentOwnerOnly(t *testing.T) if !report.Changed { t.Fatal("report changed = false, want true") } - repaired := readFakeSharedRootStateForApp(t, backend) - if got := strings.Join(repaired.AllManagedOutputPaths(), ","); got != "report.html" { - t.Fatalf("shared-root outputs = %q, want other owner output preserved", got) + repaired := readFakeCatalogState(t, backend) + if got := strings.Join(state.CatalogManagedOutputPaths(repaired), ","); got != "fresh.txt,html.txt" { + t.Fatalf("catalog outputs = %q, want other owner output preserved", got) } } -func TestReconcileStateSharedRootAllOwnersRepairsEveryOwner(t *testing.T) { - t.Skip("catalog reconcile-state execution is covered by the catalog maintenance work") +func TestReconcileStateAllOwnersRepairsEveryOwner(t *testing.T) { backend := fake.New() cfg := reconcileStateS3Config(t) - sharedRoot := reconcileSharedRootFixture(t) - writeFakeSharedRootStateForApp(t, backend, sharedRoot) - if err := backend.DeleteManagedOutputs(context.Background(), "", []string{"report.md", "report.html"}, storage.DeleteOptions{}); err != nil { + writeFakeCatalogState(t, backend, pruneCatalogState(time.Date(2026, 6, 8, 12, 0, 0, 0, time.UTC))) + if err := backend.DeleteManagedOutputs(context.Background(), "", []string{"old.txt", "html.txt"}, storage.DeleteOptions{}); err != nil { t.Fatalf("delete managed outputs: %v", err) } @@ -145,12 +140,12 @@ func TestReconcileStateSharedRootAllOwnersRepairsEveryOwner(t *testing.T) { if err != nil { t.Fatalf("reconcileStateConfigWithBackendFactory() error = %v", err) } - if !report.Changed || report.CheckedCount != 2 { + if !report.Changed || report.CheckedCount != 3 { t.Fatalf("report changed=%t checked=%d, want all-owner repair", report.Changed, report.CheckedCount) } - repaired := readFakeSharedRootStateForApp(t, backend) - if got := repaired.AllManagedOutputPaths(); len(got) != 0 { - t.Fatalf("shared-root outputs = %#v, want none", got) + repaired := readFakeCatalogState(t, backend) + if got := strings.Join(state.CatalogManagedOutputPaths(repaired), ","); got != "fresh.txt" { + t.Fatalf("catalog outputs = %q, want fresh.txt", got) } } @@ -169,93 +164,6 @@ func reconcileStateS3Config(t *testing.T) config.Config { return cfg } -func readFakeSingleOwnerState(t *testing.T, backend *fake.Backend) state.DistributorState { - t.Helper() - data, err := backend.ReadFile(context.Background(), storage.StateFileName) - if err != nil { - t.Fatalf("read state: %v", err) - } - destinationState, err := state.Parse(data) - if err != nil { - t.Fatalf("parse state: %v", err) - } - return destinationState -} - -func writeFakeSharedRootStateForApp(t *testing.T, backend *fake.Backend, sharedRoot state.SharedRootState) { - t.Helper() - data, err := json.MarshalIndent(sharedRoot, "", " ") - if err != nil { - t.Fatalf("marshal shared-root state: %v", err) - } - testutil.WriteFakeFile(t, backend, storage.StateFileName, string(append(data, '\n'))) - for _, output := range sharedRoot.Outputs { - testutil.WriteFakeFile(t, backend, output.Path, "old") - } -} - -func readFakeSharedRootStateForApp(t *testing.T, backend *fake.Backend) state.SharedRootState { - t.Helper() - data, err := backend.ReadFile(context.Background(), storage.StateFileName) - if err != nil { - t.Fatalf("read shared-root state: %v", err) - } - sharedRoot, err := state.ParseSharedRoot(data) - if err != nil { - t.Fatalf("parse shared-root state: %v", err) - } - return sharedRoot -} - -func reconcileSharedRootFixture(t *testing.T) state.SharedRootState { - t.Helper() - source := testutil.ValidManifest(testutil.BundleOptions{}) - htmlSource := source - createdAt := time.Date(2026, 5, 30, 11, 12, 0, 0, time.UTC) - return state.SharedRootState{ - SchemaVersion: state.SharedRootSchemaVersion, - DistributorVersion: "test", - CreatedAt: createdAt, - UpdatedAt: createdAt, - State: state.StatePolicy{Mode: state.StateModeSharedRoot}, - Owners: []state.OwnerRecord{{ - Scope: state.CurrentOwnerScope("reports", "archive"), - Reconciliation: state.ReconciliationPolicy{Mode: config.ReconciliationModeReplace}, - Source: state.SourceState{Manifest: source}, - }, { - Scope: state.CurrentOwnerScope("reports", "html"), - Reconciliation: state.ReconciliationPolicy{Mode: config.ReconciliationModeMerge}, - Source: state.SourceState{Manifest: htmlSource}, - }}, - Outputs: []state.SharedRootOutputFile{{ - Path: "report.md", - Kind: state.OutputKindSource, - SourcePath: "report.md", - SHA256: source.Files[0].SHA256, - Size: source.Files[0].Size, - Owner: state.CurrentOwnerScope("reports", "archive"), - SourceID: source.ID, - SourceDigest: source.Digest, - SourceCreated: source.Created, - CreatedAt: createdAt, - UpdatedAt: createdAt, - }, { - Path: "report.html", - Kind: state.OutputKindGenerated, - SourcePath: "report.md", - Transform: "markdown_to_html", - SHA256: "sha256:" + strings.Repeat("a", 64), - Size: 128, - Owner: state.CurrentOwnerScope("reports", "html"), - SourceID: htmlSource.ID, - SourceDigest: htmlSource.Digest, - SourceCreated: htmlSource.Created, - CreatedAt: createdAt, - UpdatedAt: createdAt, - }}, - } -} - func reportPathList(paths []ReconcileStatePath) string { values := make([]string, 0, len(paths)) for _, path := range paths { diff --git a/internal/cli/prune_test.go b/internal/cli/prune_test.go index 49790c9..9b4bc03 100644 --- a/internal/cli/prune_test.go +++ b/internal/cli/prune_test.go @@ -3,11 +3,14 @@ package cli import ( "bytes" "context" + "encoding/json" "os" "path/filepath" "strings" "testing" + "time" + "gitea.maximumdirect.net/eric/distributor/internal/bundle" "gitea.maximumdirect.net/eric/distributor/internal/state" "gitea.maximumdirect.net/eric/distributor/internal/storage" "gitea.maximumdirect.net/eric/distributor/internal/testutil" @@ -73,7 +76,6 @@ func TestExecutePruneRejectsInvalidFlags(t *testing.T) { } func TestExecutePruneDryRunReportsWithoutWriting(t *testing.T) { - t.Skip("catalog prune execution is covered by the catalog maintenance work") destinationRoot, configPath := writePruneLocalFixture(t) var stdout, stderr bytes.Buffer @@ -93,9 +95,10 @@ func TestExecutePruneDryRunReportsWithoutWriting(t *testing.T) { } assertLocalFile(t, filepath.Join(destinationRoot, "report.md"), "# Report\nSunny.\n") assertLocalFile(t, filepath.Join(destinationRoot, "summary.txt"), "Summary\n") + assertLocalFile(t, filepath.Join(destinationRoot, "html.txt"), "other") assertLocalFile(t, filepath.Join(destinationRoot, "extra.txt"), "unmanaged") - destinationState := testutil.ReadDestinationState(t, filepath.Join(destinationRoot, storage.StateFileName)) - if got := strings.Join(state.ManagedOutputPaths(destinationState), ","); got != "report.md,summary.txt" { + catalog := readLocalCatalogState(t, filepath.Join(destinationRoot, storage.StateFileName)) + if got := strings.Join(state.CatalogManagedOutputPaths(catalog), ","); got != "report.md,summary.txt,html.txt" { t.Fatalf("state outputs = %q, want original outputs", got) } if stderr.Len() != 0 { @@ -104,7 +107,6 @@ func TestExecutePruneDryRunReportsWithoutWriting(t *testing.T) { } func TestExecutePruneJSONReport(t *testing.T) { - t.Skip("catalog prune execution is covered by the catalog maintenance work") _, configPath := writePruneLocalFixture(t) var stdout, stderr bytes.Buffer @@ -138,7 +140,6 @@ func TestExecutePruneJSONReport(t *testing.T) { } func TestExecutePruneApplyDeletesManagedOutputs(t *testing.T) { - t.Skip("catalog prune execution is covered by the catalog maintenance work") destinationRoot, configPath := writePruneLocalFixture(t) var stdout, stderr bytes.Buffer @@ -162,13 +163,14 @@ func TestExecutePruneApplyDeletesManagedOutputs(t *testing.T) { if _, err := os.Stat(filepath.Join(destinationRoot, "summary.txt")); !os.IsNotExist(err) { t.Fatalf("summary.txt stat error = %v, want not exist", err) } + assertLocalFile(t, filepath.Join(destinationRoot, "html.txt"), "other") assertLocalFile(t, filepath.Join(destinationRoot, "extra.txt"), "unmanaged") if _, err := os.Stat(filepath.Join(destinationRoot, storage.StateFileName)); err != nil { t.Fatalf("state file stat error = %v", err) } - destinationState := testutil.ReadDestinationState(t, filepath.Join(destinationRoot, storage.StateFileName)) - if got := state.ManagedOutputPaths(destinationState); len(got) != 0 { - t.Fatalf("state outputs = %#v, want none", got) + catalog := readLocalCatalogState(t, filepath.Join(destinationRoot, storage.StateFileName)) + if got := strings.Join(state.CatalogManagedOutputPaths(catalog), ","); got != "html.txt" { + t.Fatalf("state outputs = %q, want html.txt", got) } if stderr.Len() != 0 { t.Fatalf("stderr = %q, want empty", stderr.String()) @@ -180,13 +182,16 @@ func writePruneLocalFixture(t *testing.T) (string, string) { sourceRoot := t.TempDir() destinationRoot := t.TempDir() manifest := testutil.WriteSourceBundle(t, sourceRoot, "", testutil.BundleOptions{}) - testutil.WriteDestinationState(t, destinationRoot, "", manifest, testutil.DestinationStateOptions{}) + writeCatalogDestinationState(t, destinationRoot, manifest, true) for _, file := range testutil.DefaultSourceFiles() { path := filepath.Join(destinationRoot, filepath.FromSlash(file.Path)) if err := os.WriteFile(path, []byte(file.Data), 0o600); err != nil { t.Fatalf("write destination output: %v", err) } } + if err := os.WriteFile(filepath.Join(destinationRoot, "html.txt"), []byte("other"), 0o600); err != nil { + t.Fatalf("write other owner output: %v", err) + } if err := os.WriteFile(filepath.Join(destinationRoot, "extra.txt"), []byte("unmanaged"), 0o600); err != nil { t.Fatalf("write unmanaged output: %v", err) } @@ -211,3 +216,71 @@ pipelines: } return destinationRoot, configPath } + +func writeCatalogDestinationState(t *testing.T, root string, manifest bundle.Manifest, includeOtherOwner bool) { + t.Helper() + createdAt := time.Date(2026, 6, 8, 12, 0, 0, 0, time.UTC) + source := state.CatalogSourceIdentity{ID: manifest.ID, Digest: manifest.Digest, Created: manifest.Created} + outputs := []state.CatalogOutputFile{{ + Path: "report.md", + PipelineID: "reports", + DestinationID: "archive", + Source: source, + Kind: state.OutputKindSource, + SHA256: manifest.Files[0].SHA256, + Size: manifest.Files[0].Size, + CreatedAt: createdAt, + UpdatedAt: createdAt, + }, { + Path: "summary.txt", + PipelineID: "reports", + DestinationID: "archive", + Source: source, + Kind: state.OutputKindSource, + SHA256: manifest.Files[1].SHA256, + Size: manifest.Files[1].Size, + CreatedAt: createdAt, + UpdatedAt: createdAt, + }} + if includeOtherOwner { + outputs = append(outputs, state.CatalogOutputFile{ + Path: "html.txt", + PipelineID: "reports", + DestinationID: "html", + Source: source, + Kind: state.OutputKindSource, + SHA256: manifest.Files[0].SHA256, + Size: manifest.Files[0].Size, + CreatedAt: createdAt, + UpdatedAt: createdAt, + }) + } + catalog := state.CatalogState{ + SchemaVersion: state.CatalogSchemaVersion, + DistributorVersion: "test", + CreatedAt: createdAt, + UpdatedAt: createdAt, + State: state.StatePolicy{Mode: state.StateModeCatalog}, + Outputs: outputs, + } + data, err := json.MarshalIndent(catalog, "", " ") + if err != nil { + t.Fatalf("marshal catalog state: %v", err) + } + if err := os.WriteFile(filepath.Join(root, storage.StateFileName), append(data, '\n'), 0o600); err != nil { + t.Fatalf("write catalog state: %v", err) + } +} + +func readLocalCatalogState(t *testing.T, path string) state.CatalogState { + t.Helper() + data, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read catalog state: %v", err) + } + catalog, err := state.ParseCatalog(data) + if err != nil { + t.Fatalf("parse catalog state: %v", err) + } + return catalog +} diff --git a/internal/cli/reconcile_state_test.go b/internal/cli/reconcile_state_test.go index 0797579..af466ad 100644 --- a/internal/cli/reconcile_state_test.go +++ b/internal/cli/reconcile_state_test.go @@ -14,7 +14,6 @@ import ( ) func TestExecuteReconcileStateAppliesByDefault(t *testing.T) { - t.Skip("catalog reconcile-state execution is covered by the catalog maintenance work") _, destinationRoot, configPath := writeReconcileStateLocalFixture(t) if err := os.WriteFile(filepath.Join(destinationRoot, "report.md"), []byte("# Report\nSunny.\n"), 0o600); err != nil { t.Fatalf("write managed output: %v", err) @@ -37,9 +36,9 @@ func TestExecuteReconcileStateAppliesByDefault(t *testing.T) { if !strings.Contains(stdout.String(), "status=changed") { t.Fatalf("stdout = %q, want changed status", stdout.String()) } - destinationState := testutil.ReadDestinationState(t, filepath.Join(destinationRoot, storage.StateFileName)) - if got := strings.Join(state.ManagedOutputPaths(destinationState), ","); got != "report.md" { - t.Fatalf("state outputs = %q, want report.md", got) + catalog := readLocalCatalogState(t, filepath.Join(destinationRoot, storage.StateFileName)) + if got := strings.Join(state.CatalogManagedOutputPaths(catalog), ","); got != "report.md,html.txt" { + t.Fatalf("state outputs = %q, want report.md,html.txt", got) } assertLocalFile(t, filepath.Join(destinationRoot, "extra.txt"), "unmanaged") if stderr.Len() != 0 { @@ -48,7 +47,6 @@ func TestExecuteReconcileStateAppliesByDefault(t *testing.T) { } func TestExecuteReconcileStateDryRunReportsWithoutWriting(t *testing.T) { - t.Skip("catalog reconcile-state execution is covered by the catalog maintenance work") _, destinationRoot, configPath := writeReconcileStateLocalFixture(t) if err := os.WriteFile(filepath.Join(destinationRoot, "report.md"), []byte("# Report\nSunny.\n"), 0o600); err != nil { t.Fatalf("write managed output: %v", err) @@ -69,8 +67,8 @@ func TestExecuteReconcileStateDryRunReportsWithoutWriting(t *testing.T) { if !strings.Contains(stdout.String(), "status=would_change") { t.Fatalf("stdout = %q, want would_change status", stdout.String()) } - destinationState := testutil.ReadDestinationState(t, filepath.Join(destinationRoot, storage.StateFileName)) - if got := strings.Join(state.ManagedOutputPaths(destinationState), ","); got != "report.md,summary.txt" { + catalog := readLocalCatalogState(t, filepath.Join(destinationRoot, storage.StateFileName)) + if got := strings.Join(state.CatalogManagedOutputPaths(catalog), ","); got != "report.md,summary.txt,html.txt" { t.Fatalf("state outputs = %q, want original outputs", got) } if stderr.Len() != 0 { @@ -79,7 +77,6 @@ func TestExecuteReconcileStateDryRunReportsWithoutWriting(t *testing.T) { } func TestExecuteReconcileStateJSONReport(t *testing.T) { - t.Skip("catalog reconcile-state execution is covered by the catalog maintenance work") _, destinationRoot, configPath := writeReconcileStateLocalFixture(t) if err := os.WriteFile(filepath.Join(destinationRoot, "report.md"), []byte("# Report\nSunny.\n"), 0o600); err != nil { t.Fatalf("write managed output: %v", err) @@ -176,7 +173,10 @@ func writeReconcileStateLocalFixture(t *testing.T) (string, string, string) { sourceRoot := t.TempDir() destinationRoot := t.TempDir() manifest := testutil.WriteSourceBundle(t, sourceRoot, "", testutil.BundleOptions{}) - testutil.WriteDestinationState(t, destinationRoot, "", manifest, testutil.DestinationStateOptions{}) + writeCatalogDestinationState(t, destinationRoot, manifest, true) + if err := os.WriteFile(filepath.Join(destinationRoot, "html.txt"), []byte("other"), 0o600); err != nil { + t.Fatalf("write other owner output: %v", err) + } configPath := testutil.WriteMinimalLocalConfig(t, sourceRoot, destinationRoot) return sourceRoot, destinationRoot, configPath } diff --git a/internal/state/outputs.go b/internal/state/outputs.go index 2887c45..9d5fd48 100644 --- a/internal/state/outputs.go +++ b/internal/state/outputs.go @@ -187,6 +187,62 @@ func RemoveMissingSharedRootOutputs(s SharedRootState, missingPaths []string) (S return next, changed } +func CatalogOutputsForOwner(outputs []CatalogOutputFile, scope OwnerScope) []CatalogOutputFile { + selected := make([]CatalogOutputFile, 0, len(outputs)) + for _, output := range outputs { + if output.PipelineID == scope.PipelineID && output.DestinationID == scope.DestinationID { + selected = append(selected, output) + } + } + return selected +} + +func CatalogManagedOutputPaths(s CatalogState) []string { + paths := make([]string, 0, len(s.Outputs)) + for _, output := range s.Outputs { + paths = append(paths, output.Path) + } + return paths +} + +func RemoveMissingCatalogOwnerOutputs(s CatalogState, scope OwnerScope, missingPaths []string) (CatalogState, bool) { + if len(missingPaths) == 0 { + return s, false + } + missing := pathSet(missingPaths) + next := s + next.Outputs = make([]CatalogOutputFile, 0, len(s.Outputs)) + changed := false + for _, output := range s.Outputs { + if output.PipelineID == scope.PipelineID && output.DestinationID == scope.DestinationID { + if _, remove := missing[output.Path]; remove { + changed = true + continue + } + } + next.Outputs = append(next.Outputs, output) + } + return next, changed +} + +func RemoveMissingCatalogOutputs(s CatalogState, missingPaths []string) (CatalogState, bool) { + if len(missingPaths) == 0 { + return s, false + } + missing := pathSet(missingPaths) + next := s + next.Outputs = make([]CatalogOutputFile, 0, len(s.Outputs)) + changed := false + for _, output := range s.Outputs { + if _, remove := missing[output.Path]; remove { + changed = true + continue + } + next.Outputs = append(next.Outputs, output) + } + return next, changed +} + func (s SharedRootState) OutputOwner(path string) (OwnerScope, bool) { for _, output := range s.Outputs { if output.Path == path { diff --git a/internal/state/prune.go b/internal/state/prune.go index 527424f..3d68d35 100644 --- a/internal/state/prune.go +++ b/internal/state/prune.go @@ -49,6 +49,22 @@ func SharedRootPruneCandidates(s SharedRootState, scope OwnerScope) []PruneCandi return candidates } +func CatalogPruneCandidates(s CatalogState, scope OwnerScope) []PruneCandidate { + candidates := make([]PruneCandidate, 0, len(s.Outputs)) + for _, output := range s.Outputs { + if output.PipelineID != scope.PipelineID || output.DestinationID != scope.DestinationID { + continue + } + owner := scope + 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)