Implement single-owner reconciliation modes

internal/app/run_destination.go

@@ -69,6 +69,7 @@ func processDestinationSelection(ctx context.Context, request runDestinationRequ
 		Publish:               *request.destination.Publish,
 		Transform:             request.destination.Transform,
 		Links:                 request.destination.Links,
+		Reconciliation:        request.destination.Reconciliation,
 		Transformers:          request.transforms,
 		Transfer:              request.destination.Transfer,
 		DistributorVersion:    Version,

internal/app/run_test.go

@@ -790,6 +790,53 @@ func TestRunNotifiesAfterReplacement(t *testing.T) {
 	}
 }
 
+func TestRunMergeReconciliationRetainsManagedOutput(t *testing.T) {
+	sourceRoot := t.TempDir()
+	destinationRoot := t.TempDir()
+	manifest := testutil.WriteSourceBundle(t, sourceRoot, "", testutil.BundleOptions{
+		Files: []testutil.SourceFile{{Path: "report.md", Data: "# Report\nNew.\n"}},
+	})
+	older := manifest
+	older.Created = older.Created.Add(-time.Hour)
+	defaultManifest := testutil.ValidManifest(testutil.BundleOptions{})
+	older.Files = append([]bundle.ManifestFile(nil), defaultManifest.Files...)
+	older.Digest = bundle.BundleDigest(older.Files)
+	writeDestinationState(t, destinationRoot, "", older)
+	if err := os.WriteFile(filepath.Join(destinationRoot, "report.md"), []byte("old\n"), 0o600); err != nil {
+		t.Fatalf("write old report: %v", err)
+	}
+	if err := os.WriteFile(filepath.Join(destinationRoot, "summary.txt"), []byte("old summary\n"), 0o600); err != nil {
+		t.Fatalf("write old summary: %v", err)
+	}
+	configPath := writeConfigFile(t, `
+pipelines:
+  - id: reports
+    source:
+      backend: local
+      path: `+sourceRoot+`
+    destinations:
+      - id: archive
+        backend: local
+        path: `+destinationRoot+`
+        reconciliation:
+          mode: merge
+`)
+
+	err := Run(context.Background(), RunOptions{ConfigPath: configPath})
+	if err != nil {
+		t.Fatalf("Run() error = %v", err)
+	}
+	testutil.AssertFile(t, filepath.Join(destinationRoot, "report.md"), "# Report\nNew.\n")
+	testutil.AssertFile(t, filepath.Join(destinationRoot, "summary.txt"), "old summary\n")
+	destinationState := readStateFile(t, filepath.Join(destinationRoot, storage.StateFileName))
+	if got, want := destinationState.Reconciliation.Mode, config.ReconciliationModeMerge; got != want {
+		t.Fatalf("reconciliation mode = %q, want %q", got, want)
+	}
+	if got, want := len(destinationState.Outputs), 2; got != want {
+		t.Fatalf("state output count = %d, want %d", got, want)
+	}
+}
+
 func TestRunJSONIncludesGeneratedOutputMetadata(t *testing.T) {
 	sourceRoot := t.TempDir()
 	destinationRoot := t.TempDir()

internal/publish/execute.go

@@ -12,6 +12,7 @@ import (
 )
 
 func Execute(ctx context.Context, req Request, plan Plan) error {
+	plan.Reconciliation = normalizeReconciliation(plan.Reconciliation)
 	switch plan.Action {
 	case ActionSkipSame, ActionSkipDestinationNewer:
 		return nil
@@ -24,13 +25,15 @@ func Execute(ctx context.Context, req Request, plan Plan) error {
 		if plan.ExistingState == nil {
 			return fmt.Errorf("replace requires existing destination state")
 		}
-		if err := req.DestinationBackend.DeleteManagedBundle(ctx, req.DestinationBundlePath, stateOutputManagedPaths(plan.ExistingState.Outputs), storage.DeleteOptions{IgnoreMissing: true, PruneEmptyDirs: true}); err != nil {
+		if plan.Reconciliation.Mode == config.ReconciliationModeReplace {
+			if err := req.DestinationBackend.DeleteManagedBundle(ctx, req.DestinationBundlePath, state.ManagedOutputPaths(*plan.ExistingState), storage.DeleteOptions{IgnoreMissing: true, PruneEmptyDirs: true}); err != nil {
 				return err
 			}
 			if err := ensureDestinationEmpty(ctx, req.DestinationBackend, req.DestinationBundlePath); err != nil {
 				return err
 			}
 		}
+	}
 	if plan.Action == ActionForceReplace {
 		if err := req.DestinationBackend.DeletePrefix(ctx, req.DestinationBundlePath, storage.DeleteOptions{IgnoreMissing: true, PruneEmptyDirs: true}); err != nil {
 			return err
@@ -39,10 +42,20 @@ func Execute(ctx context.Context, req Request, plan Plan) error {
 			return err
 		}
 	}
+	if usesMergeRetention(plan) {
+		if err := ensureMergeOutputPaths(ctx, req.DestinationBackend, req.DestinationBundlePath, plan); err != nil {
+			return err
+		}
+	}
 
 	writtenOutputs := make([]Output, 0, len(plan.Outputs))
+	newOutputs := make([]Output, 0, len(plan.Outputs))
 	cleanup := func() {
-		_ = req.DestinationBackend.DeleteManagedBundle(ctx, req.DestinationBundlePath, ManagedOutputPaths(writtenOutputs), storage.DeleteOptions{IgnoreMissing: true, PruneEmptyDirs: true})
+		outputs := writtenOutputs
+		if usesMergeRetention(plan) {
+			outputs = newOutputs
+		}
+		_ = req.DestinationBackend.DeleteManagedBundle(ctx, req.DestinationBundlePath, ManagedOutputPaths(outputs), storage.DeleteOptions{IgnoreMissing: true, PruneEmptyDirs: true})
 	}
 	for _, output := range plan.Outputs {
 		destinationPath, err := storage.Join(req.DestinationBundlePath, output.DestinationPath)
@@ -63,19 +76,26 @@ func Execute(ctx context.Context, req Request, plan Plan) error {
 				return err
 			}
 		}
-		if _, err := req.DestinationBackend.WriteFile(ctx, destinationPath, data, storage.WriteOptions{Overwrite: false, PreferAtomic: true}); err != nil {
+		managed := outputManagedByExistingState(output, plan.ExistingState)
+		if _, err := req.DestinationBackend.WriteFile(ctx, destinationPath, data, storage.WriteOptions{Overwrite: managed && usesMergeRetention(plan), PreferAtomic: true}); err != nil {
 			cleanup()
 			return err
 		}
 		writtenOutputs = append(writtenOutputs, output)
+		if !managed {
+			newOutputs = append(newOutputs, output)
+		}
 	}
 
 	now := time.Now().UTC()
 	createdAt := now
-	existingOutputs := []state.OutputFile(nil)
 	if plan.ExistingState != nil {
 		createdAt = plan.ExistingState.CreatedAt
-		existingOutputs = plan.ExistingState.Outputs
+	}
+	stateOutputs, err := stateOutputsForPlan(plan, now)
+	if err != nil {
+		cleanup()
+		return err
 	}
 	destinationState := state.DistributorState{
 		SchemaVersion:      state.SchemaVersion,
@@ -86,9 +106,9 @@ func Execute(ctx context.Context, req Request, plan Plan) error {
 		CreatedAt:          createdAt,
 		UpdatedAt:          now,
 		State:              state.StatePolicy{Mode: state.StateModeSingleOwner},
-		Reconciliation:     state.ReconciliationPolicy{Mode: config.ReconciliationModeReplace},
+		Reconciliation:     state.ReconciliationPolicy{Mode: plan.Reconciliation.Mode},
 		Source:             state.SourceState{Manifest: req.SourceBundle.Manifest},
-		Outputs:            state.ProjectOutputs(StateOutputProjections(plan.Outputs), existingOutputs, now),
+		Outputs:            stateOutputs,
 	}
 	if plan.PrimaryURL != "" {
 		destinationState.Links = &state.LinkState{PrimaryURL: plan.PrimaryURL}
@@ -108,9 +128,51 @@ func Execute(ctx context.Context, req Request, plan Plan) error {
 		cleanup()
 		return err
 	}
-	if _, err := req.DestinationBackend.WriteFile(ctx, statePath, data, storage.WriteOptions{Overwrite: false, PreferAtomic: true}); err != nil {
+	if _, err := req.DestinationBackend.WriteFile(ctx, statePath, data, storage.WriteOptions{Overwrite: plan.ExistingState != nil, PreferAtomic: true}); err != nil {
 		cleanup()
 		return err
 	}
 	return nil
 }
+
+func ensureMergeOutputPaths(ctx context.Context, backend storage.Backend, bundlePath string, plan Plan) error {
+	for _, output := range plan.Outputs {
+		if outputManagedByExistingState(output, plan.ExistingState) {
+			continue
+		}
+		destinationPath, err := storage.Join(bundlePath, output.DestinationPath)
+		if err != nil {
+			return err
+		}
+		if _, err := backend.Stat(ctx, destinationPath); err == nil {
+			return fmt.Errorf("merge output path %s exists but is not managed by destination state", storage.DisplayPath(output.DestinationPath))
+		} else if !storage.IsNotFound(err) {
+			return err
+		}
+	}
+	return nil
+}
+
+func outputManagedByExistingState(output Output, existing *state.DistributorState) bool {
+	if existing == nil {
+		return false
+	}
+	_, ok := state.FindOutputByPath(existing.Outputs, output.DestinationPath)
+	return ok
+}
+
+func stateOutputsForPlan(plan Plan, now time.Time) ([]state.OutputFile, error) {
+	existingOutputs := []state.OutputFile(nil)
+	if plan.ExistingState != nil {
+		existingOutputs = plan.ExistingState.Outputs
+	}
+	planned := state.ProjectOutputs(StateOutputProjections(plan.Outputs), existingOutputs, now)
+	if !usesMergeRetention(plan) || plan.ExistingState == nil {
+		return planned, nil
+	}
+	return state.MergeOutputFiles(plan.ExistingState.Outputs, planned)
+}
+
+func usesMergeRetention(plan Plan) bool {
+	return plan.Reconciliation.Mode == config.ReconciliationModeMerge && plan.Action == ActionReplaceOlder
+}

internal/publish/execute_test.go

@@ -5,11 +5,15 @@ import (
 	"fmt"
 	"io"
 	"testing"
+	"time"
 
+	"gitea.maximumdirect.net/eric/distributor/internal/bundle"
 	"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"
+	"gitea.maximumdirect.net/eric/distributor/internal/transform"
 )
 
 func TestExecuteCleansUpAfterWriteFailure(t *testing.T) {
@@ -44,6 +48,250 @@ func TestExecuteCleansUpAfterWriteFailure(t *testing.T) {
 	}
 }
 
+func TestExecuteReplaceDeletesOmittedManagedOutputs(t *testing.T) {
+	sourceBackend := fake.New()
+	sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "", testutil.BundleOptions{
+		Files: []testutil.SourceFile{{Path: "report.md", Data: "# Report\nNew.\n"}},
+	})
+	destinationBackend := fake.New()
+	older := sourceBundle.Manifest
+	older.Created = older.Created.Add(-time.Hour)
+	older.Files = append([]bundle.ManifestFile(nil), testutil.ValidManifest(testutil.BundleOptions{}).Files...)
+	older.Digest = bundle.BundleDigest(older.Files)
+	testutil.WriteFakeDestinationState(t, destinationBackend, "", older, testutil.DestinationStateOptions{})
+
+	req := testRequest(sourceBackend, destinationBackend, sourceBundle, config.ReconciliationModeReplace)
+	plan, err := Build(context.Background(), req)
+	if err != nil {
+		t.Fatalf("Build() error = %v", err)
+	}
+	if plan.Action != ActionReplaceOlder {
+		t.Fatalf("plan action = %s, want replace_older", plan.Action)
+	}
+	if err := Execute(context.Background(), req, plan); err != nil {
+		t.Fatalf("Execute() error = %v", err)
+	}
+	testutil.AssertFakeFile(t, destinationBackend, "report.md", "# Report\nNew.\n")
+	testutil.AssertFakeMissing(t, destinationBackend, "summary.txt")
+	destinationState := readFakeState(t, destinationBackend, "")
+	if got, want := len(destinationState.Outputs), 1; got != want {
+		t.Fatalf("state output count = %d, want %d", got, want)
+	}
+	if got, want := destinationState.Reconciliation.Mode, config.ReconciliationModeReplace; got != want {
+		t.Fatalf("reconciliation mode = %q, want %q", got, want)
+	}
+}
+
+func TestExecuteMergeRetainsOmittedAndOverwritesManagedOutputs(t *testing.T) {
+	sourceBackend := fake.New()
+	sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "", testutil.BundleOptions{
+		Files: []testutil.SourceFile{{Path: "report.md", Data: "# Report\nNew.\n"}},
+	})
+	destinationBackend := fake.New()
+	older := sourceBundle.Manifest
+	older.Created = older.Created.Add(-time.Hour)
+	older.Files = append([]bundle.ManifestFile(nil), testutil.ValidManifest(testutil.BundleOptions{}).Files...)
+	older.Digest = bundle.BundleDigest(older.Files)
+	existingState := testutil.WriteFakeDestinationState(t, destinationBackend, "", older, testutil.DestinationStateOptions{})
+
+	req := testRequest(sourceBackend, destinationBackend, sourceBundle, config.ReconciliationModeMerge)
+	plan, err := Build(context.Background(), req)
+	if err != nil {
+		t.Fatalf("Build() error = %v", err)
+	}
+	if err := Execute(context.Background(), req, plan); err != nil {
+		t.Fatalf("Execute() error = %v", err)
+	}
+	testutil.AssertFakeFile(t, destinationBackend, "report.md", "# Report\nNew.\n")
+	testutil.AssertFakeFile(t, destinationBackend, "summary.txt", "old")
+	destinationState := readFakeState(t, destinationBackend, "")
+	outputs := outputsByPath(destinationState.Outputs)
+	if got, want := len(outputs), 2; got != want {
+		t.Fatalf("state output count = %d, want %d", got, want)
+	}
+	if !outputs["summary.txt"].CreatedAt.Equal(existingState.Outputs[1].CreatedAt) || !outputs["summary.txt"].UpdatedAt.Equal(existingState.Outputs[1].UpdatedAt) {
+		t.Fatalf("retained output timestamps = %#v, want existing %#v", outputs["summary.txt"], existingState.Outputs[1])
+	}
+	if !outputs["report.md"].CreatedAt.Equal(existingState.Outputs[0].CreatedAt) || !outputs["report.md"].UpdatedAt.After(existingState.Outputs[0].UpdatedAt) {
+		t.Fatalf("updated output timestamps = %#v, want preserved created_at and newer updated_at", outputs["report.md"])
+	}
+	if got, want := destinationState.Reconciliation.Mode, config.ReconciliationModeMerge; got != want {
+		t.Fatalf("reconciliation mode = %q, want %q", got, want)
+	}
+}
+
+func TestExecuteMergeFailsOnUnmanagedDestinationPathCollision(t *testing.T) {
+	sourceBackend := fake.New()
+	sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "", testutil.BundleOptions{
+		Files: []testutil.SourceFile{{Path: "report.md", Data: "# Report\nNew.\n"}},
+	})
+	destinationBackend := fake.New()
+	older := sourceBundle.Manifest
+	older.Created = older.Created.Add(-time.Hour)
+	older.Files = []bundle.ManifestFile{{
+		Path:   "summary.txt",
+		SHA256: bundle.FileDigest([]byte("Summary\n")),
+		Size:   int64(len("Summary\n")),
+	}}
+	older.Digest = bundle.BundleDigest(older.Files)
+	testutil.WriteFakeDestinationState(t, destinationBackend, "", older, testutil.DestinationStateOptions{})
+	testutil.WriteFakeFile(t, destinationBackend, "report.md", "unmanaged")
+
+	req := testRequest(sourceBackend, destinationBackend, sourceBundle, config.ReconciliationModeMerge)
+	plan, err := Build(context.Background(), req)
+	if err != nil {
+		t.Fatalf("Build() error = %v", err)
+	}
+	err = Execute(context.Background(), req, plan)
+	if err == nil {
+		t.Fatal("Execute() error = nil, want unmanaged path collision")
+	}
+	testutil.AssertFakeFile(t, destinationBackend, "report.md", "unmanaged")
+	testutil.AssertFakeFile(t, destinationBackend, "summary.txt", "old")
+}
+
+func TestExecuteMergeFailureCleansUpOnlyNewOutputs(t *testing.T) {
+	sourceBackend := fake.New()
+	sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "", testutil.BundleOptions{
+		Files: []testutil.SourceFile{
+			{Path: "report.md", Data: "# Report\nNew.\n"},
+			{Path: "new.md", Data: "new\n"},
+			{Path: "fail.md", Data: "fail\n"},
+		},
+	})
+	destinationBackend := &failingBackend{Backend: fake.New(), failPath: "fail.md"}
+	older := sourceBundle.Manifest
+	older.Created = older.Created.Add(-time.Hour)
+	older.Files = []bundle.ManifestFile{sourceBundle.Manifest.Files[0]}
+	older.Digest = bundle.BundleDigest(older.Files)
+	testutil.WriteFakeDestinationState(t, destinationBackend.Backend, "", older, testutil.DestinationStateOptions{})
+
+	req := testRequest(sourceBackend, destinationBackend, sourceBundle, config.ReconciliationModeMerge)
+	plan, err := Build(context.Background(), req)
+	if err != nil {
+		t.Fatalf("Build() error = %v", err)
+	}
+	err = Execute(context.Background(), req, plan)
+	if err == nil {
+		t.Fatal("Execute() error = nil, want injected failure")
+	}
+	testutil.AssertFakeFile(t, destinationBackend.Backend, "report.md", "# Report\nNew.\n")
+	testutil.AssertFakeMissing(t, destinationBackend.Backend, "new.md")
+	testutil.AssertFakeMissing(t, destinationBackend.Backend, "fail.md")
+}
+
+func TestExecuteFixedPathSupportsReconciliationModes(t *testing.T) {
+	for _, mode := range []string{config.ReconciliationModeReplace, config.ReconciliationModeMerge} {
+		t.Run(mode, func(t *testing.T) {
+			sourceBackend := fake.New()
+			sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "new", testutil.BundleOptions{
+				Files: []testutil.SourceFile{{Path: "report.md", Data: "# Report\nNew.\n"}},
+			})
+			destinationBackend := fake.New()
+			older := sourceBundle.Manifest
+			older.ID = "older.source"
+			older.Created = older.Created.Add(-time.Hour)
+			testutil.WriteFakeDestinationState(t, destinationBackend, "", older, testutil.DestinationStateOptions{})
+
+			req := testRequest(sourceBackend, destinationBackend, sourceBundle, mode)
+			req.DestinationBundlePath = ""
+			req.PathMapping = config.PathMappingFixed
+			plan, err := Build(context.Background(), req)
+			if err != nil {
+				t.Fatalf("Build() error = %v", err)
+			}
+			if plan.Action != ActionReplaceOlder {
+				t.Fatalf("plan action = %s, want replace_older", plan.Action)
+			}
+			if err := Execute(context.Background(), req, plan); err != nil {
+				t.Fatalf("Execute() error = %v", err)
+			}
+			testutil.AssertFakeFile(t, destinationBackend, "report.md", "# Report\nNew.\n")
+			destinationState := readFakeState(t, destinationBackend, "")
+			if got, want := destinationState.Reconciliation.Mode, mode; got != want {
+				t.Fatalf("state reconciliation mode = %q, want %q", got, want)
+			}
+		})
+	}
+}
+
+func TestExecuteReconciliationModesHonorPublishPolicies(t *testing.T) {
+	tests := []struct {
+		name        string
+		publish     config.PublishPolicy
+		transform   config.Transform
+		transformer TransformerResolver
+		wantPaths   []string
+	}{
+		{
+			name:      "source only",
+			publish:   config.PublishPolicy{Source: true},
+			wantPaths: []string{"report.md"},
+		},
+		{
+			name:      "generated only",
+			publish:   config.PublishPolicy{HTML: true},
+			transform: config.Transform{MarkdownToHTML: &config.MarkdownToHTML{Enabled: true, Mode: config.TransformModeSidecar}},
+			transformer: testResolver{transform.MarkdownToHTML: testTransformer{outputs: []transform.Output{{
+				Path:       "report.html",
+				SourcePath: "report.md",
+				Transform:  transform.MarkdownToHTML,
+				Data:       []byte("<h1>Report</h1>\n"),
+				SHA256:     bundle.FileDigest([]byte("<h1>Report</h1>\n")),
+				Size:       int64(len("<h1>Report</h1>\n")),
+			}}}},
+			wantPaths: []string{"report.html"},
+		},
+		{
+			name:      "source and generated",
+			publish:   config.PublishPolicy{Source: true, HTML: true},
+			transform: config.Transform{MarkdownToHTML: &config.MarkdownToHTML{Enabled: true, Mode: config.TransformModeSidecar}},
+			transformer: testResolver{transform.MarkdownToHTML: testTransformer{outputs: []transform.Output{{
+				Path:       "report.html",
+				SourcePath: "report.md",
+				Transform:  transform.MarkdownToHTML,
+				Data:       []byte("<h1>Report</h1>\n"),
+				SHA256:     bundle.FileDigest([]byte("<h1>Report</h1>\n")),
+				Size:       int64(len("<h1>Report</h1>\n")),
+			}}}},
+			wantPaths: []string{"report.md", "report.html"},
+		},
+	}
+	for _, mode := range []string{config.ReconciliationModeReplace, config.ReconciliationModeMerge} {
+		for _, tt := range tests {
+			t.Run(mode+" "+tt.name, func(t *testing.T) {
+				sourceBackend := fake.New()
+				sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "", testutil.BundleOptions{
+					Files: []testutil.SourceFile{{Path: "report.md", Data: "# Report\nNew.\n"}},
+				})
+				destinationBackend := fake.New()
+				older := sourceBundle.Manifest
+				older.Created = older.Created.Add(-time.Hour)
+				testutil.WriteFakeDestinationState(t, destinationBackend, "", older, testutil.DestinationStateOptions{})
+
+				req := testRequest(sourceBackend, destinationBackend, sourceBundle, mode)
+				req.Publish = tt.publish
+				req.Transform = tt.transform
+				req.Transformers = tt.transformer
+				plan, err := Build(context.Background(), req)
+				if err != nil {
+					t.Fatalf("Build() error = %v", err)
+				}
+				if err := Execute(context.Background(), req, plan); err != nil {
+					t.Fatalf("Execute() error = %v", err)
+				}
+				destinationState := readFakeState(t, destinationBackend, "")
+				outputs := outputsByPath(destinationState.Outputs)
+				for _, path := range tt.wantPaths {
+					if _, ok := outputs[path]; !ok {
+						t.Fatalf("state outputs = %#v, missing %s", destinationState.Outputs, path)
+					}
+				}
+			})
+		}
+	}
+}
+
 type failingBackend struct {
 	*fake.Backend
 	failPath string
@@ -62,3 +310,48 @@ func (b *failingBackend) WriteFrom(ctx context.Context, path string, r io.Reader
 	}
 	return b.Backend.WriteFrom(ctx, path, r, opts)
 }
+
+func testRequest(sourceBackend storage.Backend, destinationBackend storage.Backend, sourceBundle bundle.Bundle, reconciliationMode string) Request {
+	return Request{
+		PipelineID:            "reports",
+		DestinationID:         "archive",
+		SourceBundle:          sourceBundle,
+		SourceBackend:         sourceBackend,
+		DestinationBackend:    destinationBackend,
+		DestinationBundlePath: sourceBundle.RootRelativePath,
+		Publish:               config.PublishPolicy{Source: true},
+		Reconciliation:        config.ReconciliationPolicy{Mode: reconciliationMode},
+		Transfer: config.TransferPolicy{
+			OnDestinationSame:  config.TransferActionSkip,
+			OnDestinationOlder: config.TransferActionReplace,
+			OnDestinationNewer: config.TransferActionSkip,
+			OnConflict:         config.TransferActionFail,
+		},
+		DistributorVersion: "test",
+	}
+}
+
+func readFakeState(t *testing.T, backend storage.Backend, bundlePath string) state.DistributorState {
+	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 state: %v", err)
+	}
+	destinationState, err := state.Parse(data)
+	if err != nil {
+		t.Fatalf("parse state: %v", err)
+	}
+	return destinationState
+}
+
+func outputsByPath(outputs []state.OutputFile) map[string]state.OutputFile {
+	byPath := make(map[string]state.OutputFile, len(outputs))
+	for _, output := range outputs {
+		byPath[output.Path] = output
+	}
+	return byPath
+}

internal/publish/output.go

@@ -148,11 +148,3 @@ func ManagedOutputPaths(outputs []Output) []string {
 	}
 	return paths
 }
-
-func stateOutputManagedPaths(outputs []state.OutputFile) []string {
-	paths := make([]string, 0, len(outputs))
-	for _, output := range outputs {
-		paths = append(paths, output.Path)
-	}
-	return paths
-}

internal/publish/plan.go

@@ -34,6 +34,7 @@ type Request struct {
 	Publish               config.PublishPolicy
 	Transform             config.Transform
 	Links                 *config.Links
+	Reconciliation        config.ReconciliationPolicy
 	Transformers          TransformerResolver
 	Transfer              config.TransferPolicy
 	DistributorVersion    string
@@ -55,6 +56,7 @@ type Plan struct {
 	Reason                string
 	Force                 bool
 	PrimaryURL            string
+	Reconciliation        config.ReconciliationPolicy
 	Outputs               []Output
 	ExistingState         *state.DistributorState
 }
@@ -88,6 +90,7 @@ func Build(ctx context.Context, req Request) (Plan, error) {
 	}
 	comparison := compareDestination(req, status)
 	action, reason := actionForComparison(comparison, req.Transfer, req.Force)
+	reconciliation := normalizeReconciliation(req.Reconciliation)
 	plan := Plan{
 		PipelineID:            req.PipelineID,
 		DestinationID:         req.DestinationID,
@@ -99,6 +102,7 @@ func Build(ctx context.Context, req Request) (Plan, error) {
 		Reason:                reason,
 		Force:                 action == ActionForceReplace,
 		PrimaryURL:            primaryURL,
+		Reconciliation:        reconciliation,
 		Outputs:               outputs,
 		ExistingState:         status.State,
 	}
@@ -124,9 +128,21 @@ func validateRequest(req Request) error {
 	if err := config.ValidatePublishTransformPolicy(req.Publish, req.Transform); err != nil {
 		return fmt.Errorf("publish/transform policy: %w", err)
 	}
+	switch normalizeReconciliation(req.Reconciliation).Mode {
+	case config.ReconciliationModeReplace, config.ReconciliationModeMerge:
+	default:
+		return fmt.Errorf("reconciliation.mode must be %s or %s", config.ReconciliationModeReplace, config.ReconciliationModeMerge)
+	}
 	return nil
 }
 
+func normalizeReconciliation(policy config.ReconciliationPolicy) config.ReconciliationPolicy {
+	if policy.Mode == "" {
+		policy.Mode = config.ReconciliationModeReplace
+	}
+	return policy
+}
+
 func compareDestination(req Request, status state.DestinationStatus) state.Comparison {
 	comparison := state.Compare(req.SourceBundle.Manifest, req.PipelineID, req.DestinationID, status)
 	if req.PathMapping != config.PathMappingFixed || comparison.Outcome != state.OutcomeDifferentSourceConflict || status.State == nil {