diff --git a/internal/app/run_destination.go b/internal/app/run_destination.go
index 7bebad9..6de6b0c 100644
--- a/internal/app/run_destination.go
+++ b/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,
diff --git a/internal/app/run_test.go b/internal/app/run_test.go
index fec7241..f8de694 100644
--- a/internal/app/run_test.go
+++ b/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()
diff --git a/internal/publish/execute.go b/internal/publish/execute.go
index 21403fa..04f2095 100644
--- a/internal/publish/execute.go
+++ b/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,11 +25,13 @@ 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 {
- return err
- }
- if err := ensureDestinationEmpty(ctx, req.DestinationBackend, req.DestinationBundlePath); err != nil {
- return err
+ 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 {
@@ -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
+}
diff --git a/internal/publish/execute_test.go b/internal/publish/execute_test.go
index 3cf1458..4e00c39 100644
--- a/internal/publish/execute_test.go
+++ b/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("Report
\n"),
+ SHA256: bundle.FileDigest([]byte("Report
\n")),
+ Size: int64(len("Report
\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("Report
\n"),
+ SHA256: bundle.FileDigest([]byte("Report
\n")),
+ Size: int64(len("Report
\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
+}
diff --git a/internal/publish/output.go b/internal/publish/output.go
index b7853a2..2be281a 100644
--- a/internal/publish/output.go
+++ b/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
-}
diff --git a/internal/publish/plan.go b/internal/publish/plan.go
index 200a485..203a3b1 100644
--- a/internal/publish/plan.go
+++ b/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 {