From 9a2eaf8e5e0dd72b5eedd5ed2ce5f19ceafab783 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Fri, 19 Jun 2026 16:39:01 +0000 Subject: [PATCH] Implement catalog force replacement --- docs/cli.md | 6 +- docs/integrations/destination-state.md | 2 +- docs/integrations/s3.md | 2 +- docs/integrations/ssh-sftp.md | 2 +- docs/internal/publish.md | 6 +- docs/operations.md | 12 +-- docs/troubleshooting.md | 14 ++-- internal/app/run_test.go | 85 ++++++++++++++++++-- internal/cli/root_test.go | 4 +- internal/publish/execute.go | 13 +++- internal/publish/execute_test.go | 59 ++++++++++++++ internal/publish/plan.go | 33 +++++++- internal/publish/plan_test.go | 104 +++++++++++++++++++++++++ 13 files changed, 307 insertions(+), 35 deletions(-) diff --git a/docs/cli.md b/docs/cli.md index b48f99f..6c2c701 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -72,7 +72,7 @@ distributor run [--config ] [--dry-run] [--force] [--format text|json] - `--config ` loads the pipeline configuration. If omitted, the application uses `/usr/local/etc/distributor/config.yml`. - `--dry-run` validates inputs and reports destination actions without applying changes. -- `--force` permits a run when destination state indicates a conservative safety check would otherwise block it. +- `--force` permits exceptional catalog replacement when a dry run reports `force_replace` for unmanaged content, a planned unmanaged path collision, invalid state, or unsupported future state. - `--format text|json` selects human-readable or machine-readable output. `run` accepts no positional arguments. @@ -206,7 +206,7 @@ go run ./cmd/distributor run --config examples/local-publish.yml --dry-run go run ./cmd/distributor run --config examples/local-publish.yml ``` -Use `--format json` when automation needs structured run results. Use `--force` only when the operator has reviewed the destination state conflict and intentionally wants to continue. +Use `--format json` when automation needs structured run results. Use `--force` only after `--dry-run --force` reports the intended bounded `force_replace` action. ### Repair Destination State Records @@ -292,7 +292,7 @@ Text output is optimized for direct operator use. JSON output is optimized for a - Usage errors and fatal setup errors exit non-zero and do not emit a JSON result document. - `run --format json` emits a JSON result for partial destination failures, sets `ok` to `false`, includes result details and errors, and exits non-zero. - Warnings are included in JSON output and are printed in text output when relevant. -- `run` summaries include separate `replace_older`, `replace_conflict`, `replace_newer`, `replace_takeover`, and `force_replace` counters. Takeover destination actions include `takeover_mode` in JSON and text output. +- `run` summaries include separate `force_replace` counters. In JSON output, forced replacement actions use the stable action value `force_replace` and include the resolved `destination_path`. ## Diagnostics And Recovery diff --git a/docs/integrations/destination-state.md b/docs/integrations/destination-state.md index 4306b85..e392c4b 100644 --- a/docs/integrations/destination-state.md +++ b/docs/integrations/destination-state.md @@ -117,7 +117,7 @@ If a same-source merge publication fails after writing outputs, cleanup removes - Shared-root output path owned by a different owner: conflict unless `takeover.mode` permits managed ownership transfer or `transfer.on_conflict: replace` is configured. - Invalid state JSON or invalid state fields: conflict. -Normal single-owner replacement deletes only managed output paths recorded in `outputs` plus `.distributor.json`. Shared-root replacement deletes only omitted outputs for the current owner. Merge publication retains omitted managed outputs for same-source replacement. Forced replacement deletes the bounded destination bundle path. +Normal managed replacement deletes only managed output paths recorded in `.distributor.json`. Forced replacement deletes the bounded destination bundle path, then writes planned outputs and schema version `4` catalog state. ## State Repair Semantics diff --git a/docs/integrations/s3.md b/docs/integrations/s3.md index 46f935c..574e73c 100644 --- a/docs/integrations/s3.md +++ b/docs/integrations/s3.md @@ -52,7 +52,7 @@ The adapter uses these S3 operations: Writes buffer the input and set `ContentLength`. If no content type is supplied by the caller, the adapter infers a content type from the logical path. -Normal replacement and failed-write cleanup delete only managed output objects plus `.distributor.json`. Same-source merge publication retains omitted managed objects and may overwrite existing managed objects. Takeover replacement does not retain omitted outputs through merge reconciliation. Forced replacement deletes objects under the bounded destination bundle prefix. The backend does not manage bucket versioning, lifecycle rules, object lock, or delete markers. +Normal managed replacement and failed-write cleanup delete only managed output objects plus `.distributor.json`. Forced replacement deletes objects under the bounded destination bundle prefix, then writes planned outputs and schema version `4` catalog state. For fixed-path destinations, that bounded prefix is the configured backend root. The backend does not manage bucket versioning, lifecycle rules, object lock, or delete markers. ## Error Mapping diff --git a/docs/integrations/ssh-sftp.md b/docs/integrations/ssh-sftp.md index 311e1a4..9044dbe 100644 --- a/docs/integrations/ssh-sftp.md +++ b/docs/integrations/ssh-sftp.md @@ -54,7 +54,7 @@ The configured `path` is the backend root. All source discovery, destination pat The adapter rejects symlink ancestors for reads and writes. Reads require regular files. Writes create parent directories and prefer atomic temp-file-plus-rename writes when requested. Walk output is sorted through the shared storage walker. -Managed cleanup and normal replacement delete only managed output paths plus `.distributor.json`. Same-source merge publication retains omitted managed paths and may overwrite existing managed paths. Takeover replacement does not retain omitted outputs through merge reconciliation. Forced replacement deletes the bounded destination bundle path. +Managed cleanup and normal replacement delete only managed output paths plus `.distributor.json`. Forced replacement deletes the bounded destination bundle path, then writes planned outputs and schema version `4` catalog state. For fixed-path destinations, that bounded path is the configured backend root. ## Boundaries diff --git a/docs/internal/publish.md b/docs/internal/publish.md index 9f9e409..bfb1627 100644 --- a/docs/internal/publish.md +++ b/docs/internal/publish.md @@ -28,7 +28,7 @@ The package depends on `internal/storage.Backend` for source and destination IO, ## State And Manifest Behavior -Planning inspects destination state through `internal/state`, compares it with the source manifest, and maps comparison outcomes plus transfer and takeover policy into actions: `publish_new`, `replace_older`, `replace_conflict`, `replace_newer`, `replace_takeover`, `force_replace`, `skip_same`, `skip_destination_newer`, `fail_conflict`, or `fail_unmanaged`. +Planning inspects destination state through `internal/state`, compares it with the source manifest, and maps comparison outcomes plus transfer and takeover policy into actions. Catalog planning uses `publish_new`, `upsert_additive`, `replace_catalog`, `force_replace`, `fail_conflict`, or `fail_unmanaged`; legacy action labels remain only for older planning and removal work scheduled separately. Single-owner destinations compare the whole destination state against the configured pipeline and destination ids. Valid managed identity and source conflicts can become `replace_takeover` when `takeover.mode` allows them. Shared-root destinations compare only the current owner scope, keyed by pipeline id and destination id. An absent shared-root owner is publishable for that owner unless a planned output collides with unmanaged storage content. Planned writes to a path owned by another shared-root owner become `replace_takeover` when `takeover.mode` allows that managed output path to move to the current owner. @@ -40,7 +40,7 @@ Execution writes destination state after selected outputs are written. Destinati Shared-root execution writes schema version `3` state. It preserves unrelated owner records and outputs, updates only the publishing owner metadata, preserves root `created_at`, and updates root `updated_at` after successful state writes. Compatible single-owner state for the same pipeline and destination is converted to shared-root state on successful publish. -Forced replacement is explicit per request and deletes the bounded destination bundle path before writing new outputs and state. Valid managed conflict and newer-state replacements are normal managed replacement actions when transfer policy allows them; they are not forced replacement actions. +Forced replacement is explicit per request and deletes the bounded destination bundle path before writing planned outputs and schema version `4` catalog state. Catalog planning selects `force_replace` only when `Force` is true and normal planning would otherwise fail for a non-empty no-state destination, a planned path collision with unmanaged storage content, invalid destination state, or unsupported future destination state. Valid catalog-managed additive upserts and replacement workflow publishes are normal managed actions; they are not forced replacement actions. Retention pruning is not part of publish execution and does not run automatically after a successful publish. The app-level prune workflow uses destination state after publication to select managed outputs for deletion. @@ -67,7 +67,7 @@ Execution fails on delete, read, transform output, unmanaged merge path collisio - Merge state output records are cumulative for the single owner. - Shared-root planning is owner-scoped, preserves unrelated owner outputs, and records taken-over managed output paths separately from unrelated owners. - Shared-root execution writes owner-scoped changes without deleting unrelated owners. -- Forced replacement deletes only within the supplied destination bundle path. +- Forced replacement deletes only within the supplied destination bundle path and then writes catalog state using the same output projection as normal catalog publish planning. - Destination state is written after selected outputs are written. - Transform resolution stays behind a caller-supplied interface. - Unmanaged content is claimed only by explicit force. diff --git a/docs/operations.md b/docs/operations.md index 74a52cd..9cc8aa9 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -166,12 +166,12 @@ Review these action labels before publishing: - `replace_takeover`: destination state is valid managed state and `takeover.mode` allows replacement across an identity, source, or shared-root output-owner mismatch. - `skip_same`: destination state already matches the source. - `skip_destination_newer`: destination state is newer than the source and is skipped. -- `force_replace`: destructive replacement selected because `--force` is present and policy permits it. +- `force_replace`: destructive catalog replacement selected because `--force` is present for unmanaged content, a planned unmanaged path collision, invalid state, or unsupported future state. - `error`: planning or execution failed for that destination. Fixed destinations add fixed-path warnings during dry runs, including the selected source bundle and replacement warnings when the destination root would be replaced. -Text and JSON summaries count `publish_new`, `replace_older`, `replace_conflict`, `replace_newer`, `replace_takeover`, `force_replace`, skipped, and failed destinations separately. Takeover action records include the configured takeover mode and the conflict reason. JSON output includes warnings, pipeline summaries, destination action records, output records, URLs when configured, final counters, and partial failure details. Fatal setup failures such as unreadable config or invalid secrets do not produce a JSON result document. +Text and JSON summaries count `force_replace` separately from normal publish and catalog workflow actions. JSON output includes warnings, pipeline summaries, destination action records, output records, URLs when configured, final counters, and partial failure details. Fatal setup failures such as unreadable config or invalid secrets do not produce a JSON result document. ## Forced Replacement Workflow @@ -182,11 +182,11 @@ go run ./cmd/distributor run --config --dry-run --force go run ./cmd/distributor run --config --force ``` -Forced replacement can claim unmanaged non-empty destination paths and is reserved for exceptional destructive replacement. Valid managed newer state and valid managed conflict state use `replace_newer` and `replace_conflict` when the corresponding transfer policy allows replacement; they do not require `--force`. +Forced replacement can claim a non-empty destination path with no valid `.distributor.json`, replace planned output paths that collide with storage content not recorded in valid catalog state, and recover from invalid or unsupported future destination state. It is reserved for exceptional destructive replacement. Valid catalog-managed additive upserts and replacement workflow publishes do not require `--force`. -Forced replacement deletes the current destination bundle path before writing outputs and state. It does not delete parent paths, sibling paths, or storage outside the destination bundle path. For fixed destinations, the destination bundle path is the backend root, so a forced replacement can clear that configured root. +Forced replacement deletes the current destination bundle path before writing planned outputs and schema version `4` catalog state. It does not delete parent paths, sibling paths, or storage outside the destination bundle path. For fixed destinations, the destination bundle path is the configured backend root, so a forced replacement can clear that configured root. Dry-run text and JSON output report that root as `target=.` or `destination_path: "."`. -For shared-root destinations, forced replacement also deletes the configured destination bundle path before writing new shared-root state. This removes unrelated owners inside that destination root. Preview with `--dry-run --force` and confirm the destination path before applying. +Preview with `--dry-run --force` and confirm the destination path before applying. `--force` applies only to the current invocation. There is no config field that enables forced replacement by default. @@ -272,7 +272,7 @@ S3 execution uses the AWS SDK for Go v2. See [S3-Compatible Storage Integration] When explicit S3 credential variable names are configured, both must resolve to non-empty values through the process environment or `secrets.directory`. When omitted, the AWS SDK default credential chain is used as-is. -Normal single-owner replacement and failed-write cleanup delete only managed output objects recorded in `.distributor.json` plus the state object. Shared-root replacement deletes only current-owner omitted output objects and rewrites the shared state object. Same-source merge publication retains omitted managed objects and may overwrite existing managed objects. Takeover replacement does not retain omitted outputs through merge reconciliation. Forced replacement deletes objects under the bounded destination bundle prefix. Distributor does not manage bucket versioning or delete markers. +Normal managed replacement and failed-write cleanup delete only managed output objects recorded in `.distributor.json` plus the state object. Forced replacement deletes objects under the bounded destination bundle prefix and then writes schema version `4` catalog state. Distributor does not manage bucket versioning or delete markers. ## Secrets Operation diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index afddb17..5260a5d 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -187,9 +187,9 @@ Reference: [Operations](operations.md#cleanup-and-recovery). ## Destination Has Unmanaged Content -Symptom: `destination has content but no distributor state` or a plan reason containing `fail_unmanaged`. +Symptom: `destination has content but no distributor state`, `destination output path ... exists but is not managed by catalog state`, or a plan reason containing `fail_unmanaged`. -Likely cause: the destination bundle path contains files but no valid `.distributor.json`, so `distributor` will not claim it by default. +Likely cause: the destination bundle path contains files but no valid `.distributor.json`, or a planned output path collides with storage content that valid catalog state does not record. `distributor` will not claim unmanaged content by default. Diagnostic: @@ -197,7 +197,7 @@ Diagnostic: find -maxdepth 2 -print ``` -Safe fix: choose an empty destination path, move unrelated files aside, or preview `run --dry-run --force` only after confirming the reported bundle path is safe to replace. +Safe fix: choose an empty destination path, move unrelated files aside, or preview `run --dry-run --force` only after confirming the reported destination bundle path is safe to replace. Reference: [Operations](operations.md#forced-replacement-workflow). @@ -214,7 +214,7 @@ cat /.distributor.json go run ./cmd/distributor inspect ``` -Safe fix: verify the source and destination are intended to match. Use a separate destination path for unrelated content. For normal managed replacement, configure destination `takeover.mode` to match the intended ownership boundary or configure `transfer.on_conflict: replace`, then preview with `--dry-run`. Use `--force` only for exceptional replacement of unmanaged content or other force-only cases reported as `force_replace`. +Safe fix: verify the source and destination are intended to match. Use a separate destination path for unrelated content. Use `--force` only for exceptional replacement of unmanaged content, invalid state, unsupported future state, or other cases that `run --dry-run --force` reports as `force_replace`. Reference: [Operations](operations.md#destination-state-and-retry-behavior). @@ -280,7 +280,7 @@ Reference: [Operations](operations.md#forced-replacement-workflow). Symptom: dry-run output includes `force_replace`. -Likely cause: the run used `--force`, and planning selected a supported destructive replacement. +Likely cause: the run used `--force`, and catalog planning selected a supported destructive replacement for unmanaged destination content, a planned unmanaged path collision, invalid destination state, or unsupported future destination state. Diagnostic: @@ -288,7 +288,7 @@ Diagnostic: go run ./cmd/distributor run --config --dry-run --force ``` -Safe fix: inspect the pipeline id, destination id, backend, and bundle path. Proceed only if deleting everything inside that destination bundle path is intended. +Safe fix: inspect the pipeline id, destination id, backend, and bundle path. Confirm the `destination_path` in JSON output, or the fixed-path `target=.` text output, before applying. Proceed only if deleting everything inside that destination bundle path is intended; `force_replace` then writes planned outputs and schema version `4` catalog state. Reference: [Operations](operations.md#forced-replacement-workflow). @@ -327,7 +327,7 @@ Reference: [Operations](operations.md#destination-state-and-retry-behavior). ## Shared-Root Unmanaged Path Collision -Symptom: `fail_unmanaged` with a reason like `destination output path ... exists but is not managed by destination state`. +Symptom: `fail_unmanaged` with a reason like `destination output path ... exists but is not managed by catalog state`. Likely cause: a `state.mode: shared_root` destination planned a new output path that already exists in storage but is not recorded as managed in `.distributor.json`. diff --git a/internal/app/run_test.go b/internal/app/run_test.go index 7fe2c49..8dcb756 100644 --- a/internal/app/run_test.go +++ b/internal/app/run_test.go @@ -597,7 +597,6 @@ func TestRunJSONIncludesWorkflowActionAndSummary(t *testing.T) { } func TestRunFixedPathFailsUnmanagedWithoutForce(t *testing.T) { - t.Skip("catalog workflow protects planned path collisions rather than unrelated unplanned content") sourceRoot := t.TempDir() destinationRoot := t.TempDir() writeSourceBundle(t, sourceRoot, "bundle", testBundleOptions{}) @@ -612,7 +611,6 @@ func TestRunFixedPathFailsUnmanagedWithoutForce(t *testing.T) { } func TestRunFixedPathForceReplacementStaysWithinDestinationRoot(t *testing.T) { - t.Skip("force reporting and execution behavior is covered by the catalog reporting work") sourceRoot := t.TempDir() parent := t.TempDir() destinationRoot := filepath.Join(parent, "latest") @@ -639,6 +637,10 @@ func TestRunFixedPathForceReplacementStaysWithinDestinationRoot(t *testing.T) { if _, err := os.Stat(filepath.Join(destinationRoot, "unmanaged.txt")); !os.IsNotExist(err) { t.Fatalf("unmanaged stat error = %v, want removed", err) } + catalog := readLocalCatalogState(t, destinationRoot) + if catalog.SchemaVersion != state.CatalogSchemaVersion || catalog.State.Mode != state.StateModeCatalog { + t.Fatalf("catalog identity = schema %d mode %s", catalog.SchemaVersion, catalog.State.Mode) + } } func TestRunFixedPathRemoteBackendsUseBackendRoots(t *testing.T) { @@ -1384,7 +1386,6 @@ func TestRunSkipsWhenDestinationStateMatches(t *testing.T) { } func TestRunFailsOnUnmanagedDestination(t *testing.T) { - t.Skip("catalog workflow protects planned path collisions rather than unrelated unplanned content") sourceRoot := t.TempDir() destinationRoot := t.TempDir() writeSourceBundle(t, sourceRoot, "", testBundleOptions{}) @@ -1399,7 +1400,6 @@ func TestRunFailsOnUnmanagedDestination(t *testing.T) { } func TestRunForceReplacesUnmanagedDestination(t *testing.T) { - t.Skip("force reporting and execution behavior is covered by the catalog reporting work") sourceRoot := t.TempDir() destinationRoot := t.TempDir() writeSourceBundle(t, sourceRoot, "", testBundleOptions{}) @@ -1419,10 +1419,50 @@ func TestRunForceReplacesUnmanagedDestination(t *testing.T) { if !strings.Contains(stdout.String(), "action=force_replace") { t.Fatalf("stdout = %q, want force_replace", stdout.String()) } + if !strings.Contains(stdout.String(), "Final status: ok planned=1 publish_new=0 upsert_additive=0 replace_catalog=0 skip_same=0 force_replace=1 fail_unmanaged=0 fail_conflict=0 failed=0 dry_run=false") { + t.Fatalf("stdout = %q, want force_replace counter only", stdout.String()) + } if _, err := os.Stat(filepath.Join(destinationRoot, "unmanaged.txt")); !os.IsNotExist(err) { t.Fatalf("unmanaged file stat error = %v, want not exist", err) } testutil.AssertFile(t, filepath.Join(destinationRoot, "report.md"), "# Report\nSunny.\n") + catalog := readLocalCatalogState(t, destinationRoot) + if catalog.SchemaVersion != state.CatalogSchemaVersion || catalog.State.Mode != state.StateModeCatalog { + t.Fatalf("catalog identity = schema %d mode %s", catalog.SchemaVersion, catalog.State.Mode) + } +} + +func TestRunForceReplacementJSONOutput(t *testing.T) { + sourceRoot := t.TempDir() + destinationRoot := t.TempDir() + writeSourceBundle(t, sourceRoot, "", testBundleOptions{}) + if err := os.WriteFile(filepath.Join(destinationRoot, "unmanaged.txt"), []byte("old"), 0o600); err != nil { + t.Fatalf("write unmanaged file: %v", err) + } + + var stdout bytes.Buffer + err := Run(context.Background(), RunOptions{ + ConfigPath: writeLocalConfig(t, sourceRoot, destinationRoot), + Force: true, + Stdout: &stdout, + OutputFormat: OutputFormatJSON, + }) + if err != nil { + t.Fatalf("Run() error = %v", err) + } + result := decodeAppResult(t, stdout.String()) + actions, ok := result["actions"].([]any) + if !ok || len(actions) != 1 { + t.Fatalf("actions = %#v, want one action", result["actions"]) + } + action, ok := actions[0].(map[string]any) + if !ok || action["action"] != "force_replace" || action["workflow"] != "additive" { + t.Fatalf("action = %#v, want force_replace additive", actions[0]) + } + summary, ok := result["summary"].(map[string]any) + if !ok || summary["force_replace"] != float64(1) || summary["publish_new"] != float64(0) { + t.Fatalf("summary = %#v, want force_replace only", result["summary"]) + } } func TestRunFansOutToLocalDestinations(t *testing.T) { @@ -1527,7 +1567,6 @@ func TestRunExercisesRemoteBackendShapesThroughCommonPath(t *testing.T) { } func TestRunForceReplacementStaysWithinRemoteBundlePaths(t *testing.T) { - t.Skip("force reporting and execution behavior is covered by the catalog reporting work") localSourceRoot := t.TempDir() writeSourceBundle(t, localSourceRoot, "bundle", testBundleOptions{}) s3Destination := fake.New() @@ -1569,6 +1608,12 @@ func TestRunForceReplacementStaysWithinRemoteBundlePaths(t *testing.T) { testutil.AssertFakeFile(t, sshDestination, "bundle/report.md", "# Report\nSunny.\n") testutil.AssertFakeMissing(t, sshDestination, "bundle/old.txt") testutil.AssertFakeFile(t, sshDestination, "bundle-sibling/keep.txt", "keep") + for name, backend := range map[string]*fake.Backend{"s3": s3Destination, "ssh": sshDestination} { + catalog := readFakeCatalogStateAt(t, backend, "bundle") + if catalog.SchemaVersion != state.CatalogSchemaVersion || catalog.State.Mode != state.StateModeCatalog { + t.Fatalf("%s catalog identity = schema %d mode %s", name, catalog.SchemaVersion, catalog.State.Mode) + } + } } func TestRunDryRunDoesNotWrite(t *testing.T) { @@ -1772,6 +1817,36 @@ func readStateFile(t *testing.T, path string) testDestinationState { return view } +func readLocalCatalogState(t *testing.T, destinationRoot string) state.CatalogState { + t.Helper() + data, err := os.ReadFile(filepath.Join(destinationRoot, 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 readFakeCatalogStateAt(t *testing.T, backend *fake.Backend, bundlePath string) state.CatalogState { + t.Helper() + statePath, err := storage.StatePath(bundlePath) + if err != nil { + t.Fatalf("state path: %v", err) + } + data, err := backend.ReadFile(context.Background(), statePath) + 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 outputsByPath(outputs []testStateOutput) map[string]testStateOutput { byPath := make(map[string]testStateOutput, len(outputs)) for _, output := range outputs { diff --git a/internal/cli/root_test.go b/internal/cli/root_test.go index 3fe5be4..1f7a900 100644 --- a/internal/cli/root_test.go +++ b/internal/cli/root_test.go @@ -938,8 +938,8 @@ func TestExecuteRunForceDryRunReportsWithoutWriting(t *testing.T) { if code != exitOK { t.Fatalf("exit code = %d, want %d; stderr = %q", code, exitOK, stderr.String()) } - if !strings.Contains(stdout.String(), "action=publish_new workflow=additive") { - t.Fatalf("stdout = %q, want additive publish", stdout.String()) + if !strings.Contains(stdout.String(), "action=force_replace workflow=additive") || !strings.Contains(stdout.String(), "force_replace=1") { + t.Fatalf("stdout = %q, want forced replacement", stdout.String()) } if _, err := os.Stat(filepath.Join(destinationRoot, "unmanaged.txt")); err != nil { t.Fatalf("unmanaged file stat error = %v", err) diff --git a/internal/publish/execute.go b/internal/publish/execute.go index 2635d61..fa3a7de 100644 --- a/internal/publish/execute.go +++ b/internal/publish/execute.go @@ -17,9 +17,9 @@ func Execute(ctx context.Context, req Request, plan Plan) error { switch plan.Action { case ActionSkipSame, ActionSkipDestinationNewer: return nil - case ActionPublishNew, ActionUpsertAdditive, ActionReplaceCatalog: + case ActionPublishNew, ActionUpsertAdditive, ActionReplaceCatalog, ActionForceReplace: return executeCatalog(ctx, req, plan) - case ActionReplaceOlder, ActionReplaceConflict, ActionReplaceNewer, ActionReplaceTakeover, ActionForceReplace: + case ActionReplaceOlder, ActionReplaceConflict, ActionReplaceNewer, ActionReplaceTakeover: if usesSharedRootState(req, plan) { return executeSharedRoot(ctx, req, plan) } @@ -142,7 +142,14 @@ func Execute(ctx context.Context, req Request, plan Plan) error { } func executeCatalog(ctx context.Context, req Request, plan Plan) error { - if plan.Action == ActionReplaceCatalog { + if plan.Action == ActionForceReplace { + if err := req.DestinationBackend.DeletePrefix(ctx, req.DestinationBundlePath, storage.DeleteOptions{IgnoreMissing: true, PruneEmptyDirs: true}); err != nil { + return err + } + if err := ensureDestinationEmpty(ctx, req.DestinationBackend, req.DestinationBundlePath); err != nil { + return err + } + } else if plan.Action == ActionReplaceCatalog { if plan.ClearDestinationRoot { if err := req.DestinationBackend.DeletePrefix(ctx, req.DestinationBundlePath, storage.DeleteOptions{IgnoreMissing: true, PruneEmptyDirs: true}); err != nil { return err diff --git a/internal/publish/execute_test.go b/internal/publish/execute_test.go index 797e149..bf3b26a 100644 --- a/internal/publish/execute_test.go +++ b/internal/publish/execute_test.go @@ -131,6 +131,65 @@ func TestExecuteSupersededAdditiveLeavesUnplannedFilesUnmanaged(t *testing.T) { } } +func TestExecuteForceReplaceClearsDestinationBundlePathOnly(t *testing.T) { + _, destinationBackend, req := catalogPlanRequest(t, config.WorkflowAdditive) + req.DestinationBundlePath = "bundle" + req.Force = true + testutil.WriteFakeFile(t, destinationBackend, "bundle/report.md", "old report") + testutil.WriteFakeFile(t, destinationBackend, "bundle/unplanned.txt", "remove") + testutil.WriteFakeFile(t, destinationBackend, "bundle-child/keep.txt", "keep") + testutil.WriteFakeFile(t, destinationBackend, "outside.txt", "keep") + + plan, err := Build(context.Background(), req) + if err != nil { + t.Fatalf("Build() error = %v", err) + } + if plan.Action != ActionForceReplace { + t.Fatalf("plan action = %s, want %s", plan.Action, ActionForceReplace) + } + if err := Execute(context.Background(), req, plan); err != nil { + t.Fatalf("Execute() error = %v", err) + } + + testutil.AssertFakeFile(t, destinationBackend, "bundle/report.md", "# Report\nSunny.\n") + testutil.AssertFakeFile(t, destinationBackend, "bundle/summary.txt", "Summary\n") + testutil.AssertFakeMissing(t, destinationBackend, "bundle/unplanned.txt") + testutil.AssertFakeFile(t, destinationBackend, "bundle-child/keep.txt", "keep") + testutil.AssertFakeFile(t, destinationBackend, "outside.txt", "keep") + catalog := readCatalogState(t, destinationBackend, "bundle") + if catalog.SchemaVersion != state.CatalogSchemaVersion || catalog.State.Mode != state.StateModeCatalog { + t.Fatalf("catalog identity = schema %d mode %s", catalog.SchemaVersion, catalog.State.Mode) + } + if len(catalog.Outputs) != 2 { + t.Fatalf("catalog outputs = %#v, want planned outputs only", catalog.Outputs) + } +} + +func TestExecuteForceReplaceClearsFixedDestinationRoot(t *testing.T) { + _, destinationBackend, req := catalogPlanRequest(t, config.WorkflowAdditive) + req.DestinationBundlePath = "" + req.PathMapping = config.PathMappingFixed + req.Force = true + testutil.WriteFakeFile(t, destinationBackend, "report.md", "old report") + testutil.WriteFakeFile(t, destinationBackend, "unplanned.txt", "remove") + + plan, err := Build(context.Background(), req) + if err != nil { + t.Fatalf("Build() error = %v", err) + } + if plan.Action != ActionForceReplace || storage.DisplayPath(plan.DestinationBundlePath) != "." { + t.Fatalf("plan action=%s destination=%s, want force_replace at root", plan.Action, storage.DisplayPath(plan.DestinationBundlePath)) + } + if err := Execute(context.Background(), req, plan); err != nil { + t.Fatalf("Execute() error = %v", err) + } + + testutil.AssertFakeFile(t, destinationBackend, "report.md", "# Report\nSunny.\n") + testutil.AssertFakeFile(t, destinationBackend, "summary.txt", "Summary\n") + testutil.AssertFakeMissing(t, destinationBackend, "unplanned.txt") + readCatalogState(t, destinationBackend, "") +} + func TestExecuteFailedWriteDoesNotWriteCatalogState(t *testing.T) { _, destinationBackend, req := catalogPlanRequest(t, config.WorkflowAdditive) plan, err := Build(context.Background(), req) diff --git a/internal/publish/plan.go b/internal/publish/plan.go index 4adbfa6..28267d6 100644 --- a/internal/publish/plan.go +++ b/internal/publish/plan.go @@ -139,8 +139,15 @@ func Build(ctx context.Context, req Request) (Plan, error) { SupersededLegacy: status.SupersededLegacy, } if status.StateErr != nil { - plan.Action = ActionFailConflict plan.Reason = status.StateErr.Error() + if req.Force { + plan.Action = ActionForceReplace + plan.Force = true + plan.ClearDestinationRoot = true + plan.CatalogOutputsToWrite = catalogOutputsForPlan(req, outputs, nil, scope, now) + return plan, nil + } + plan.Action = ActionFailConflict return plan, fmt.Errorf("%s: %s", plan.Action, plan.Reason) } @@ -151,7 +158,7 @@ func Build(ctx context.Context, req Request) (Plan, error) { case status.SupersededLegacy != nil: details = planSupersededLegacy(req, outputs, workflow, scope, now) default: - details, err = planWithoutCatalog(ctx, req, outputs, workflow, scope, now) + details, err = planWithoutCatalog(ctx, req, outputs, workflow, scope, now, status.HasContents) } plan.Action = details.Action plan.Reason = details.Reason @@ -159,6 +166,15 @@ func Build(ctx context.Context, req Request) (Plan, error) { plan.CatalogOutputsToRetain = details.CatalogOutputsToRetain plan.CatalogOutputsToDelete = details.CatalogOutputsToDelete plan.ClearDestinationRoot = details.ClearDestinationRoot + if err != nil && req.Force && forceCanReplace(details.Action) { + plan.Action = ActionForceReplace + plan.Force = true + plan.ClearDestinationRoot = true + plan.CatalogOutputsToWrite = catalogOutputsForPlan(req, outputs, nil, scope, now) + plan.CatalogOutputsToRetain = nil + plan.CatalogOutputsToDelete = nil + return plan, nil + } if err != nil { return plan, err } @@ -239,7 +255,14 @@ func planSupersededLegacy(req Request, outputs []Output, workflow string, scope return details } -func planWithoutCatalog(ctx context.Context, req Request, outputs []Output, workflow string, scope state.OwnerScope, now time.Time) (catalogPlanDetails, error) { +func planWithoutCatalog(ctx context.Context, req Request, outputs []Output, workflow string, scope state.OwnerScope, now time.Time, hasContents bool) (catalogPlanDetails, error) { + if hasContents { + err := fmt.Errorf("destination has content but no distributor state") + return catalogPlanDetails{ + Action: ActionFailUnmanaged, + Reason: err.Error(), + }, fmt.Errorf("%s: %s", ActionFailUnmanaged, err) + } if err := rejectCatalogUnmanagedCollisions(ctx, req.DestinationBackend, req.DestinationBundlePath, nil, outputs); err != nil { return catalogPlanDetails{ Action: ActionFailUnmanaged, @@ -252,6 +275,10 @@ func planWithoutCatalog(ctx context.Context, req Request, outputs []Output, work }, nil } +func forceCanReplace(action Action) bool { + return action == ActionFailUnmanaged || action == ActionFailConflict +} + func actionForWorkflow(workflow string) Action { if workflow == config.WorkflowReplacement { return ActionReplaceCatalog diff --git a/internal/publish/plan_test.go b/internal/publish/plan_test.go index 9a24c2e..6ae270b 100644 --- a/internal/publish/plan_test.go +++ b/internal/publish/plan_test.go @@ -137,6 +137,7 @@ func TestBuildTransfersManagedPathOwnership(t *testing.T) { func TestBuildRejectsUnmanagedPlannedPathCollision(t *testing.T) { _, destinationBackend, req := catalogPlanRequest(t, config.WorkflowAdditive) + writeCatalogState(t, destinationBackend, "", baseCatalog(req)) testutil.WriteFakeFile(t, destinationBackend, "report.md", "unmanaged") plan, err := Build(context.Background(), req) @@ -148,6 +149,55 @@ func TestBuildRejectsUnmanagedPlannedPathCollision(t *testing.T) { } } +func TestBuildForceReplacesUnmanagedPlannedPathCollision(t *testing.T) { + _, destinationBackend, req := catalogPlanRequest(t, config.WorkflowAdditive) + req.Force = true + existing := baseCatalog(req) + writeCatalogState(t, destinationBackend, "", existing) + testutil.WriteFakeFile(t, destinationBackend, "report.md", "unmanaged") + + plan, err := Build(context.Background(), req) + if err != nil { + t.Fatalf("Build() error = %v", err) + } + if plan.Action != ActionForceReplace || !plan.Force || !plan.ClearDestinationRoot { + t.Fatalf("plan action=%s force=%t clear=%t, want forced clear", plan.Action, plan.Force, plan.ClearDestinationRoot) + } + if len(plan.CatalogOutputsToWrite) != 2 || len(plan.CatalogOutputsToRetain) != 0 || len(plan.CatalogOutputsToDelete) != 0 { + t.Fatalf("catalog write=%d retain=%d delete=%d", len(plan.CatalogOutputsToWrite), len(plan.CatalogOutputsToRetain), len(plan.CatalogOutputsToDelete)) + } +} + +func TestBuildRejectsNoStateNonEmptyDestinationWithoutForce(t *testing.T) { + _, destinationBackend, req := catalogPlanRequest(t, config.WorkflowAdditive) + testutil.WriteFakeFile(t, destinationBackend, "unplanned.txt", "unmanaged") + + plan, err := Build(context.Background(), req) + if err == nil { + t.Fatal("Build() error = nil, want unmanaged destination") + } + if plan.Action != ActionFailUnmanaged || !strings.Contains(err.Error(), "destination has content but no distributor state") { + t.Fatalf("plan action=%s error=%v, want unmanaged destination", plan.Action, err) + } +} + +func TestBuildForceReplacesNoStateNonEmptyDestination(t *testing.T) { + _, destinationBackend, req := catalogPlanRequest(t, config.WorkflowAdditive) + req.Force = true + testutil.WriteFakeFile(t, destinationBackend, "unplanned.txt", "unmanaged") + + plan, err := Build(context.Background(), req) + if err != nil { + t.Fatalf("Build() error = %v", err) + } + if plan.Action != ActionForceReplace || !plan.Force || !plan.ClearDestinationRoot { + t.Fatalf("plan action=%s force=%t clear=%t, want forced clear", plan.Action, plan.Force, plan.ClearDestinationRoot) + } + if len(plan.CatalogOutputsToWrite) != 2 || len(plan.CatalogOutputsToRetain) != 0 || len(plan.CatalogOutputsToDelete) != 0 { + t.Fatalf("catalog write=%d retain=%d delete=%d", len(plan.CatalogOutputsToWrite), len(plan.CatalogOutputsToRetain), len(plan.CatalogOutputsToDelete)) + } +} + func TestBuildPlansSupersededLegacyAdditive(t *testing.T) { _, destinationBackend, req := catalogPlanRequest(t, config.WorkflowAdditive) legacyState := testutil.DestinationState(req.SourceBundle.Manifest, testutil.DestinationStateOptions{}) @@ -204,6 +254,60 @@ func TestBuildRejectsInvalidOrFutureState(t *testing.T) { } } +func TestBuildForceReplacesInvalidOrFutureState(t *testing.T) { + tests := []struct { + name string + data string + }{ + {name: "invalid json", data: `{"schema_version":`}, + {name: "future schema", data: `{"schema_version":99}`}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + _, destinationBackend, req := catalogPlanRequest(t, config.WorkflowAdditive) + req.Force = true + testutil.WriteFakeFile(t, destinationBackend, storage.StateFileName, tt.data) + + plan, err := Build(context.Background(), req) + if err != nil { + t.Fatalf("Build() error = %v", err) + } + if plan.Action != ActionForceReplace || !plan.Force || !plan.ClearDestinationRoot { + t.Fatalf("plan action=%s force=%t clear=%t, want forced clear", plan.Action, plan.Force, plan.ClearDestinationRoot) + } + if len(plan.CatalogOutputsToWrite) != 2 || len(plan.CatalogOutputsToRetain) != 0 || len(plan.CatalogOutputsToDelete) != 0 { + t.Fatalf("catalog write=%d retain=%d delete=%d", len(plan.CatalogOutputsToWrite), len(plan.CatalogOutputsToRetain), len(plan.CatalogOutputsToDelete)) + } + }) + } +} + +func TestBuildForceDoesNotChangeValidCatalogActions(t *testing.T) { + tests := []struct { + name string + workflow string + want Action + }{ + {name: "additive", workflow: config.WorkflowAdditive, want: ActionUpsertAdditive}, + {name: "replacement", workflow: config.WorkflowReplacement, want: ActionReplaceCatalog}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + _, destinationBackend, req := catalogPlanRequest(t, tt.workflow) + req.Force = true + writeCatalogState(t, destinationBackend, "", baseCatalog(req)) + + plan, err := Build(context.Background(), req) + if err != nil { + t.Fatalf("Build() error = %v", err) + } + if plan.Action != tt.want || plan.Force || plan.ClearDestinationRoot { + t.Fatalf("plan action=%s force=%t clear=%t, want %s without forced clear", plan.Action, plan.Force, plan.ClearDestinationRoot, tt.want) + } + }) + } +} + func TestValidateRequestRejectsInvalidWorkflow(t *testing.T) { sourceBackend, destinationBackend, req := catalogPlanRequest(t, "append") req.SourceBackend = sourceBackend