From 77cde4029627594dafc350d4e48da4a0f973cdd6 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Fri, 19 Jun 2026 15:36:26 +0000 Subject: [PATCH] Add catalog workflow planning --- internal/app/run_destination.go | 5 +- internal/publish/execute.go | 2 +- internal/publish/execute_test.go | 421 ------------------ internal/publish/force_test.go | 222 ---------- internal/publish/output_test.go | 6 - internal/publish/plan.go | 616 +++++++++----------------- internal/publish/plan_test.go | 296 ++++++++++++- internal/publish/shared_root_test.go | 632 --------------------------- internal/publish/takeover_test.go | 215 --------- internal/state/outputs.go | 9 + 10 files changed, 488 insertions(+), 1936 deletions(-) delete mode 100644 internal/publish/execute_test.go delete mode 100644 internal/publish/force_test.go delete mode 100644 internal/publish/shared_root_test.go delete mode 100644 internal/publish/takeover_test.go diff --git a/internal/app/run_destination.go b/internal/app/run_destination.go index 23a6990..57bef8b 100644 --- a/internal/app/run_destination.go +++ b/internal/app/run_destination.go @@ -69,11 +69,8 @@ func processDestinationSelection(ctx context.Context, request runDestinationRequ Publish: *request.destination.Publish, Transform: request.destination.Transform, Links: request.destination.Links, - State: request.destination.State, - Reconciliation: request.destination.Reconciliation, - Takeover: request.destination.Takeover, + Workflow: request.destination.Workflow, Transformers: request.transforms, - Transfer: request.destination.Transfer, DistributorVersion: Version, Force: request.options.Force, } diff --git a/internal/publish/execute.go b/internal/publish/execute.go index 131ab0b..f0bf4d1 100644 --- a/internal/publish/execute.go +++ b/internal/publish/execute.go @@ -240,7 +240,7 @@ func ensureMergeOutputPaths(ctx context.Context, backend storage.Backend, bundle } func usesSharedRootState(req Request, plan Plan) bool { - return normalizeState(req.State).Mode == config.StateModeSharedRoot || plan.StateMode == config.StateModeSharedRoot + return plan.StateMode == config.StateModeSharedRoot } func outputManagedByExistingState(output Output, existing *state.DistributorState) bool { diff --git a/internal/publish/execute_test.go b/internal/publish/execute_test.go deleted file mode 100644 index 62e3536..0000000 --- a/internal/publish/execute_test.go +++ /dev/null @@ -1,421 +0,0 @@ -package publish - -import ( - "context" - "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) { - sourceBackend := fake.New() - destinationBackend := &failingBackend{Backend: fake.New(), failPath: "summary.txt"} - sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "", testutil.BundleOptions{}) - req := Request{ - PipelineID: "reports", - DestinationID: "archive", - SourceBundle: sourceBundle, - SourceBackend: sourceBackend, - DestinationBackend: destinationBackend, - DestinationBundlePath: "", - Publish: config.PublishPolicy{Source: true}, - Transfer: config.TransferPolicy{OnDestinationSame: config.TransferActionSkip, OnDestinationOlder: config.TransferActionReplace, OnDestinationNewer: config.TransferActionSkip, OnConflict: config.TransferActionFail}, - DistributorVersion: "test", - } - 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 error") - } - found, err := destinationBackend.HasAny(context.Background(), "") - if err != nil { - t.Fatalf("HasAny() error = %v", err) - } - if found { - t.Fatal("destination has content after failed execution") - } -} - -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 TestExecuteReplaceConflictDeletesOmittedManagedOutputs(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() - conflict := sourceBundle.Manifest - conflict.ID = "other.source" - conflict.Files = append([]bundle.ManifestFile(nil), testutil.ValidManifest(testutil.BundleOptions{}).Files...) - conflict.Digest = bundle.BundleDigest(conflict.Files) - testutil.WriteFakeDestinationState(t, destinationBackend, "", conflict, testutil.DestinationStateOptions{}) - - req := testRequest(sourceBackend, destinationBackend, sourceBundle, config.ReconciliationModeMerge) - req.Takeover = config.TakeoverPolicy{Mode: config.TakeoverModeNever} - req.Transfer.OnConflict = config.TransferActionReplace - plan, err := Build(context.Background(), req) - if err != nil { - t.Fatalf("Build() error = %v", err) - } - if plan.Action != ActionReplaceConflict { - t.Fatalf("plan action = %s, want replace_conflict", 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, "") - outputs := outputsByPath(destinationState.Outputs) - if got, want := len(outputs), 1; got != want { - t.Fatalf("state output count = %d, want %d", got, want) - } - if _, ok := outputs["summary.txt"]; ok { - t.Fatalf("state retained summary.txt after replace_conflict: %#v", destinationState.Outputs) - } -} - -func TestExecuteReplaceNewerUsesManagedReplacement(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() - newer := sourceBundle.Manifest - newer.Created = newer.Created.Add(time.Hour) - newer.Files = append([]bundle.ManifestFile(nil), testutil.ValidManifest(testutil.BundleOptions{}).Files...) - newer.Digest = bundle.BundleDigest(newer.Files) - testutil.WriteFakeDestinationState(t, destinationBackend, "", newer, testutil.DestinationStateOptions{}) - - req := testRequest(sourceBackend, destinationBackend, sourceBundle, config.ReconciliationModeReplace) - req.Transfer.OnDestinationNewer = config.TransferActionReplace - plan, err := Build(context.Background(), req) - if err != nil { - t.Fatalf("Build() error = %v", err) - } - if plan.Action != ActionReplaceNewer { - t.Fatalf("plan action = %s, want replace_newer", 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") -} - -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.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 -} - -func (b *failingBackend) WriteFile(ctx context.Context, path string, data []byte, opts storage.WriteOptions) (storage.Entry, error) { - if path == b.failPath { - return storage.Entry{}, fmt.Errorf("injected write failure") - } - return b.Backend.WriteFile(ctx, path, data, opts) -} - -func (b *failingBackend) WriteFrom(ctx context.Context, path string, r io.Reader, opts storage.WriteOptions) (storage.Entry, error) { - if path == b.failPath { - return storage.Entry{}, fmt.Errorf("injected write failure") - } - 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/force_test.go b/internal/publish/force_test.go deleted file mode 100644 index fd4f418..0000000 --- a/internal/publish/force_test.go +++ /dev/null @@ -1,222 +0,0 @@ -package publish - -import ( - "context" - "strings" - "testing" - - "gitea.maximumdirect.net/eric/distributor/internal/bundle" - "gitea.maximumdirect.net/eric/distributor/internal/config" - "gitea.maximumdirect.net/eric/distributor/internal/storage/fake" - "gitea.maximumdirect.net/eric/distributor/internal/testutil" -) - -func TestBuildPlansForcedReplacementOnlyWhenExplicit(t *testing.T) { - tests := []struct { - name string - prepare func(t *testing.T, backend *fake.Backend, source bundle.Manifest) - transfer config.TransferPolicy - wantReason string - forceAction bool - }{ - { - name: "unmanaged content", - prepare: func(t *testing.T, backend *fake.Backend, source bundle.Manifest) { - t.Helper() - testutil.WriteFakeFile(t, backend, "bundle/old.txt", "old") - }, - transfer: defaultTransfer(), - wantReason: "fail_unmanaged", - forceAction: true, - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - sourceBackend := fake.New() - sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "bundle", testutil.BundleOptions{}) - destinationBackend := fake.New() - tt.prepare(t, destinationBackend, sourceBundle.Manifest) - - req := forceRequest(sourceBackend, destinationBackend, sourceBundle, tt.transfer) - _, err := Build(context.Background(), req) - if err == nil || !strings.Contains(err.Error(), tt.wantReason) { - t.Fatalf("Build() error = %v, want %q", err, tt.wantReason) - } - - req.Force = true - plan, err := Build(context.Background(), req) - if tt.forceAction { - if err != nil { - t.Fatalf("Build() with force error = %v", err) - } - if plan.Action != ActionForceReplace || !plan.Force { - t.Fatalf("forced plan action = %s force=%t", plan.Action, plan.Force) - } - } - }) - } -} - -func TestBuildPlansConflictReplacementWithoutForce(t *testing.T) { - tests := []struct { - name string - prepare func(t *testing.T, backend *fake.Backend, source bundle.Manifest) - wantReason string - }{ - { - name: "different source id", - prepare: func(t *testing.T, backend *fake.Backend, source bundle.Manifest) { - t.Helper() - conflict := source - conflict.ID = "other.source" - testutil.WriteFakeDestinationState(t, backend, "bundle", conflict, testutil.DestinationStateOptions{}) - }, - wantReason: "destination source id differs", - }, - { - name: "same created digest conflict", - prepare: func(t *testing.T, backend *fake.Backend, source bundle.Manifest) { - t.Helper() - conflict := testutil.ValidManifest(testutil.BundleOptions{Files: []testutil.SourceFile{{Path: "report.md", Data: "# Different\n"}}}) - testutil.WriteFakeDestinationState(t, backend, "bundle", conflict, testutil.DestinationStateOptions{}) - }, - wantReason: "same id and created time but different digest", - }, - { - name: "pipeline mismatch", - prepare: func(t *testing.T, backend *fake.Backend, source bundle.Manifest) { - t.Helper() - testutil.WriteFakeDestinationState(t, backend, "bundle", source, testutil.DestinationStateOptions{PipelineID: "other-pipeline"}) - }, - wantReason: "pipeline id", - }, - { - name: "destination mismatch", - prepare: func(t *testing.T, backend *fake.Backend, source bundle.Manifest) { - t.Helper() - testutil.WriteFakeDestinationState(t, backend, "bundle", source, testutil.DestinationStateOptions{DestinationID: "other-destination"}) - }, - wantReason: "destination id", - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - sourceBackend := fake.New() - sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "bundle", testutil.BundleOptions{}) - destinationBackend := fake.New() - tt.prepare(t, destinationBackend, sourceBundle.Manifest) - - req := forceRequest(sourceBackend, destinationBackend, sourceBundle, conflictReplaceTransfer()) - plan, err := Build(context.Background(), req) - if err != nil { - t.Fatalf("Build() error = %v", err) - } - if plan.Action != ActionReplaceConflict || plan.Force { - t.Fatalf("plan action = %s force=%t, want replace_conflict without force", plan.Action, plan.Force) - } - if !strings.Contains(plan.Reason, tt.wantReason) { - t.Fatalf("plan reason = %q, want %q", plan.Reason, tt.wantReason) - } - }) - } -} - -func TestBuildPlansNewerReplacementWithoutForce(t *testing.T) { - sourceBackend := fake.New() - sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "bundle", testutil.BundleOptions{}) - destinationBackend := fake.New() - newer := sourceBundle.Manifest - newer.Created = newer.Created.AddDate(0, 0, 1) - testutil.WriteFakeDestinationState(t, destinationBackend, "bundle", newer, testutil.DestinationStateOptions{}) - - req := forceRequest(sourceBackend, destinationBackend, sourceBundle, newerReplaceTransfer()) - plan, err := Build(context.Background(), req) - if err != nil { - t.Fatalf("Build() error = %v", err) - } - if plan.Action != ActionReplaceNewer || plan.Force { - t.Fatalf("plan action = %s force=%t, want replace_newer without force", plan.Action, plan.Force) - } -} - -func TestBuildRequiresConflictPolicyForStateConflicts(t *testing.T) { - sourceBackend := fake.New() - sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "bundle", testutil.BundleOptions{}) - destinationBackend := fake.New() - conflict := sourceBundle.Manifest - conflict.ID = "other.source" - testutil.WriteFakeDestinationState(t, destinationBackend, "bundle", conflict, testutil.DestinationStateOptions{}) - - req := forceRequest(sourceBackend, destinationBackend, sourceBundle, defaultTransfer()) - req.Force = true - _, err := Build(context.Background(), req) - if err == nil || !strings.Contains(err.Error(), "destination source id differs") { - t.Fatalf("Build() error = %v, want conservative conflict", err) - } -} - -func TestExecuteForcedReplacementDeletesOnlyBundlePath(t *testing.T) { - sourceBackend := fake.New() - sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "bundle", testutil.BundleOptions{}) - destinationBackend := fake.New() - testutil.WriteFakeFile(t, destinationBackend, "bundle/old.txt", "old") - testutil.WriteFakeFile(t, destinationBackend, "bundle/nested/old.txt", "old") - testutil.WriteFakeFile(t, destinationBackend, "bundle-sibling/keep.txt", "keep") - testutil.WriteFakeFile(t, destinationBackend, "outside.txt", "outside") - - req := forceRequest(sourceBackend, destinationBackend, sourceBundle, defaultTransfer()) - req.Force = true - 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 force_replace", plan.Action) - } - 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.AssertFakeMissing(t, destinationBackend, "bundle/old.txt") - testutil.AssertFakeMissing(t, destinationBackend, "bundle/nested/old.txt") - testutil.AssertFakeFile(t, destinationBackend, "bundle-sibling/keep.txt", "keep") - testutil.AssertFakeFile(t, destinationBackend, "outside.txt", "outside") -} - -func forceRequest(sourceBackend, destinationBackend *fake.Backend, sourceBundle bundle.Bundle, transfer config.TransferPolicy) Request { - return Request{ - PipelineID: "reports", - DestinationID: "archive", - SourceBundle: sourceBundle, - SourceBackend: sourceBackend, - DestinationBackend: destinationBackend, - DestinationBundlePath: sourceBundle.RootRelativePath, - Publish: config.PublishPolicy{Source: true}, - Takeover: config.TakeoverPolicy{Mode: config.TakeoverModeNever}, - Transfer: transfer, - DistributorVersion: "test", - } -} - -func defaultTransfer() config.TransferPolicy { - return config.TransferPolicy{ - OnDestinationSame: config.TransferActionSkip, - OnDestinationOlder: config.TransferActionReplace, - OnDestinationNewer: config.TransferActionSkip, - OnConflict: config.TransferActionFail, - } -} - -func conflictReplaceTransfer() config.TransferPolicy { - transfer := defaultTransfer() - transfer.OnConflict = config.TransferActionReplace - return transfer -} - -func newerReplaceTransfer() config.TransferPolicy { - transfer := defaultTransfer() - transfer.OnDestinationNewer = config.TransferActionReplace - return transfer -} diff --git a/internal/publish/output_test.go b/internal/publish/output_test.go index 96be01d..d977903 100644 --- a/internal/publish/output_test.go +++ b/internal/publish/output_test.go @@ -194,12 +194,6 @@ func TestBuildRejectsHTMLWithoutTransform(t *testing.T) { DestinationBundlePath: "", SourceBundle: sourceBundle, Publish: config.PublishPolicy{HTML: true}, - Transfer: config.TransferPolicy{ - OnDestinationSame: config.TransferActionSkip, - OnDestinationOlder: config.TransferActionReplace, - OnDestinationNewer: config.TransferActionSkip, - OnConflict: config.TransferActionFail, - }, }) if err == nil { t.Fatal("Build() error = nil, want missing transform error") diff --git a/internal/publish/plan.go b/internal/publish/plan.go index 3bb8bac..4adbfa6 100644 --- a/internal/publish/plan.go +++ b/internal/publish/plan.go @@ -3,6 +3,7 @@ package publish import ( "context" "fmt" + "time" "gitea.maximumdirect.net/eric/distributor/internal/bundle" "gitea.maximumdirect.net/eric/distributor/internal/config" @@ -24,6 +25,8 @@ const ( ActionFailUnmanaged Action = "fail_unmanaged" ActionForceReplace Action = "force_replace" ActionReplaceTakeover Action = "replace_takeover" + ActionUpsertAdditive Action = "upsert_additive" + ActionReplaceCatalog Action = "replace_catalog" ) type Request struct { @@ -37,44 +40,17 @@ type Request struct { Publish config.PublishPolicy Transform config.Transform Links *config.Links - State config.StatePolicy - Reconciliation config.ReconciliationPolicy - Takeover config.TakeoverPolicy + Workflow string Transformers TransformerResolver - Transfer config.TransferPolicy DistributorVersion string Force bool + Now time.Time } type TransformerResolver interface { Get(name string) (transform.Transformer, bool) } -type Plan struct { - PipelineID string - DestinationID string - BundleID string - BundlePath string - DestinationBundlePath string - PathMapping string - Action Action - Reason string - Force bool - PrimaryURL string - StateMode string - OwnerScope state.OwnerScope - Reconciliation config.ReconciliationPolicy - TakeoverMode string - Outputs []Output - ExistingState *state.DistributorState - ExistingSharedRoot *state.SharedRootState - OtherOwnerOutputs []state.SharedRootOutputFile - TakenOverOwnerOutputs []state.SharedRootOutputFile - RetainedOwnerOutputs []state.SharedRootOutputFile - OwnerOutputsToDelete []state.SharedRootOutputFile - OwnerOutputsToWrite []Output -} - type Output struct { SourcePath string DestinationPath string @@ -86,6 +62,49 @@ type Output struct { Size int64 } +type Plan struct { + PipelineID string + DestinationID string + BundleID string + BundlePath string + DestinationBundlePath string + PathMapping string + Action Action + Reason string + Force bool + PrimaryURL string + Workflow string + OwnerScope state.OwnerScope + Outputs []Output + ExistingCatalog *state.CatalogState + SupersededLegacy *state.SupersededLegacyState + CatalogOutputsToWrite []state.CatalogOutputFile + CatalogOutputsToRetain []state.CatalogOutputFile + CatalogOutputsToDelete []state.CatalogOutputFile + ClearDestinationRoot bool + + // Retained while the executor is migrated to catalog state. + StateMode string + Reconciliation config.ReconciliationPolicy + TakeoverMode string + ExistingState *state.DistributorState + ExistingSharedRoot *state.SharedRootState + OtherOwnerOutputs []state.SharedRootOutputFile + TakenOverOwnerOutputs []state.SharedRootOutputFile + RetainedOwnerOutputs []state.SharedRootOutputFile + OwnerOutputsToDelete []state.SharedRootOutputFile + OwnerOutputsToWrite []Output +} + +type catalogPlanDetails struct { + Action Action + Reason string + CatalogOutputsToWrite []state.CatalogOutputFile + CatalogOutputsToRetain []state.CatalogOutputFile + CatalogOutputsToDelete []state.CatalogOutputFile + ClearDestinationRoot bool +} + func Build(ctx context.Context, req Request) (Plan, error) { if err := validateRequest(req); err != nil { return Plan{}, err @@ -102,14 +121,9 @@ func Build(ctx context.Context, req Request) (Plan, error) { if err != nil { return Plan{}, err } - reconciliation := normalizeReconciliation(req.Reconciliation) - stateMode := normalizeState(req.State).Mode - comparison := compareDestination(req, status) - action, reason := actionForComparison(comparison, req.Transfer, req.Force) - if takeoverActionAllowed(req, status, comparison, stateMode, action) { - action = ActionReplaceTakeover - reason = comparison.Reason - } + workflow := normalizeWorkflow(req.Workflow) + scope := state.CurrentOwnerScope(req.PipelineID, req.DestinationID) + now := requestTime(req) plan := Plan{ PipelineID: req.PipelineID, DestinationID: req.DestinationID, @@ -117,36 +131,37 @@ func Build(ctx context.Context, req Request) (Plan, error) { BundlePath: req.SourceBundle.RootRelativePath, DestinationBundlePath: req.DestinationBundlePath, PathMapping: req.PathMapping, - Action: action, - Reason: reason, - Force: action == ActionForceReplace, PrimaryURL: primaryURL, - StateMode: stateMode, - OwnerScope: state.CurrentOwnerScope(req.PipelineID, req.DestinationID), - Reconciliation: reconciliation, - TakeoverMode: normalizeTakeover(req.Takeover).Mode, + Workflow: workflow, + OwnerScope: scope, Outputs: outputs, - ExistingState: status.State, - ExistingSharedRoot: status.SharedRoot, + ExistingCatalog: status.Catalog, + SupersededLegacy: status.SupersededLegacy, } - if stateMode == config.StateModeSharedRoot { - sharedDetails, err := planSharedRootOwner(ctx, req, status, action, reconciliation, outputs) - plan.Action = sharedDetails.Action - if sharedDetails.Reason != "" { - plan.Reason = sharedDetails.Reason - } - plan.OtherOwnerOutputs = sharedDetails.OtherOwnerOutputs - plan.TakenOverOwnerOutputs = sharedDetails.TakenOverOwnerOutputs - plan.RetainedOwnerOutputs = sharedDetails.RetainedOwnerOutputs - plan.OwnerOutputsToDelete = sharedDetails.OwnerOutputsToDelete - plan.OwnerOutputsToWrite = sharedDetails.OwnerOutputsToWrite - if err != nil { - return plan, err - } - } - if plan.Action == ActionFailConflict || plan.Action == ActionFailUnmanaged { + if status.StateErr != nil { + plan.Action = ActionFailConflict + plan.Reason = status.StateErr.Error() return plan, fmt.Errorf("%s: %s", plan.Action, plan.Reason) } + + var details catalogPlanDetails + switch { + case status.Catalog != nil: + details, err = planExistingCatalog(ctx, req, *status.Catalog, outputs, workflow, scope, now) + case status.SupersededLegacy != nil: + details = planSupersededLegacy(req, outputs, workflow, scope, now) + default: + details, err = planWithoutCatalog(ctx, req, outputs, workflow, scope, now) + } + plan.Action = details.Action + plan.Reason = details.Reason + plan.CatalogOutputsToWrite = details.CatalogOutputsToWrite + plan.CatalogOutputsToRetain = details.CatalogOutputsToRetain + plan.CatalogOutputsToDelete = details.CatalogOutputsToDelete + plan.ClearDestinationRoot = details.ClearDestinationRoot + if err != nil { + return plan, err + } return plan, nil } @@ -166,165 +181,134 @@ 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: + switch normalizeWorkflow(req.Workflow) { + case config.WorkflowAdditive, config.WorkflowReplacement: default: - return fmt.Errorf("reconciliation.mode must be %s or %s", config.ReconciliationModeReplace, config.ReconciliationModeMerge) - } - switch normalizeState(req.State).Mode { - case config.StateModeSingleOwner, config.StateModeSharedRoot: - default: - return fmt.Errorf("state.mode must be %s or %s", config.StateModeSingleOwner, config.StateModeSharedRoot) - } - switch normalizeTakeover(req.Takeover).Mode { - case config.TakeoverModeSamePipeline, config.TakeoverModeSameSource, config.TakeoverModeAnyManaged, config.TakeoverModeNever: - default: - return fmt.Errorf("takeover.mode must be %s, %s, %s, or %s", config.TakeoverModeSamePipeline, config.TakeoverModeSameSource, config.TakeoverModeAnyManaged, config.TakeoverModeNever) + return fmt.Errorf("destination.workflow must be %s or %s", config.WorkflowAdditive, config.WorkflowReplacement) } return nil } -func normalizeReconciliation(policy config.ReconciliationPolicy) config.ReconciliationPolicy { - if policy.Mode == "" { - policy.Mode = config.ReconciliationModeReplace +func normalizeWorkflow(workflow string) string { + if workflow == "" { + return config.WorkflowAdditive } - return policy + return workflow } -func normalizeState(policy config.StatePolicy) config.StatePolicy { - if policy.Mode == "" { - policy.Mode = config.StateModeSingleOwner +func requestTime(req Request) time.Time { + if req.Now.IsZero() { + return time.Now().UTC() } - return policy + return req.Now.UTC() } -func normalizeTakeover(policy config.TakeoverPolicy) config.TakeoverPolicy { - if policy.Mode == "" { - policy.Mode = config.TakeoverModeSamePipeline +func planExistingCatalog(ctx context.Context, req Request, catalog state.CatalogState, outputs []Output, workflow string, scope state.OwnerScope, now time.Time) (catalogPlanDetails, error) { + if err := rejectCatalogUnmanagedCollisions(ctx, req.DestinationBackend, req.DestinationBundlePath, catalog.Outputs, outputs); err != nil { + return catalogPlanDetails{ + Action: ActionFailUnmanaged, + Reason: err.Error(), + }, fmt.Errorf("%s: %s", ActionFailUnmanaged, err) } - return policy -} - -func compareDestination(req Request, status state.DestinationStatus) state.Comparison { - if normalizeState(req.State).Mode == config.StateModeSharedRoot { - scope := state.CurrentOwnerScope(req.PipelineID, req.DestinationID) - comparison := state.CompareSharedRootOwner(req.SourceBundle.Manifest, scope, status) - if req.PathMapping != config.PathMappingFixed || comparison.Outcome != state.OutcomeDifferentSourceConflict { - return comparison - } - destinationManifest, ok := sharedRootComparisonManifest(status, scope) - if !ok { - return comparison - } - if destinationManifest.Created.Before(req.SourceBundle.Manifest.Created) { - return state.Comparison{Outcome: state.OutcomeDestinationOlder, Reason: "fixed destination source is older than selected source"} - } - if destinationManifest.Created.After(req.SourceBundle.Manifest.Created) { - return state.Comparison{ - Outcome: state.OutcomeDestinationNewer, - Reason: "fixed destination source is newer than selected source", - Detail: state.ComparisonDetail{ - Kind: state.ComparisonDetailDestinationNewer, - CurrentSourceID: req.SourceBundle.Manifest.ID, - DestinationSourceID: destinationManifest.ID, - }, - } - } - return comparison + planned := outputPathSet(outputs) + details := catalogPlanDetails{ + Action: actionForWorkflow(workflow), + CatalogOutputsToWrite: catalogOutputsForPlan(req, outputs, catalog.Outputs, scope, now), } - comparison := state.Compare(req.SourceBundle.Manifest, req.PipelineID, req.DestinationID, status) - return comparison -} - -func sharedRootComparisonManifest(status state.DestinationStatus, scope state.OwnerScope) (bundle.Manifest, bool) { - if status.SharedRoot != nil { - return status.SharedRoot.SourceManifest(scope) - } - if status.State != nil && status.State.PipelineID == scope.PipelineID && status.State.DestinationID == scope.DestinationID { - return status.State.Source.Manifest, true - } - return bundle.Manifest{}, false -} - -type sharedRootPlanDetails struct { - Action Action - Reason string - OtherOwnerOutputs []state.SharedRootOutputFile - TakenOverOwnerOutputs []state.SharedRootOutputFile - RetainedOwnerOutputs []state.SharedRootOutputFile - OwnerOutputsToDelete []state.SharedRootOutputFile - OwnerOutputsToWrite []Output -} - -func planSharedRootOwner(ctx context.Context, req Request, status state.DestinationStatus, action Action, reconciliation config.ReconciliationPolicy, outputs []Output) (sharedRootPlanDetails, error) { - scope := state.CurrentOwnerScope(req.PipelineID, req.DestinationID) - details := sharedRootPlanDetails{Action: action} - if !isWriteAction(action) { - details.OtherOwnerOutputs = otherOwnerOutputs(status, scope) - return details, nil - } - - plannedPaths := outputPaths(outputs) - if action == ActionForceReplace { - details.OwnerOutputsToWrite = append([]Output(nil), outputs...) - return details, nil - } - conflicts := sharedRootPathOwnershipConflicts(status, scope, plannedPaths) - if len(conflicts) > 0 { - conflictAction := ActionReplaceTakeover - for _, conflict := range conflicts { - if sharedRootTakeoverAllowed(req, status, conflict) { - continue - } - if req.Transfer.OnConflict == config.TransferActionReplace { - conflictAction = ActionReplaceConflict - continue - } - reason := sharedRootOwnershipConflictReason(conflict) - details.Action = ActionFailConflict - details.Reason = reason - return details, fmt.Errorf("%s: %s", ActionFailConflict, reason) - } - details.Action = conflictAction - details.Reason = sharedRootOwnershipConflictReason(conflicts[0]) - details.TakenOverOwnerOutputs = sharedRootConflictOutputs(status.SharedRoot, conflicts) - } - if err := rejectSharedRootUnmanagedCollisions(ctx, req.DestinationBackend, req.DestinationBundlePath, status, scope, plannedPaths); err != nil { - details.Action = ActionFailUnmanaged - details.Reason = err.Error() - return details, fmt.Errorf("%s: %s", ActionFailUnmanaged, err) - } - - takenOverPaths := sharedRootOutputPathSet(details.TakenOverOwnerOutputs) - details.OtherOwnerOutputs = otherOwnerOutputsExcept(status, scope, takenOverPaths) - ownerOutputs := currentOwnerOutputs(status, scope) - planned := make(map[string]struct{}, len(plannedPaths)) - for _, path := range plannedPaths { - planned[path] = struct{}{} - } - for _, output := range ownerOutputs { + for _, output := range catalog.Outputs { if _, exists := planned[output.Path]; exists { continue } - if details.Action == ActionReplaceTakeover || details.Action == ActionReplaceConflict || (isReconciliationReplacementAction(details.Action) && reconciliation.Mode == config.ReconciliationModeReplace) { - details.OwnerOutputsToDelete = append(details.OwnerOutputsToDelete, output) + if workflow == config.WorkflowReplacement && output.PipelineID == scope.PipelineID && output.DestinationID == scope.DestinationID { + details.CatalogOutputsToDelete = append(details.CatalogOutputsToDelete, output) continue } - if isReconciliationReplacementAction(details.Action) && reconciliation.Mode == config.ReconciliationModeMerge { - details.RetainedOwnerOutputs = append(details.RetainedOwnerOutputs, output) - } + details.CatalogOutputsToRetain = append(details.CatalogOutputsToRetain, output) } - details.OwnerOutputsToWrite = append([]Output(nil), outputs...) return details, nil } -func isWriteAction(action Action) bool { - switch action { - case ActionPublishNew, ActionReplaceOlder, ActionReplaceConflict, ActionReplaceNewer, ActionReplaceTakeover, ActionForceReplace: - return true - default: - return false +func planSupersededLegacy(req Request, outputs []Output, workflow string, scope state.OwnerScope, now time.Time) catalogPlanDetails { + details := catalogPlanDetails{ + Action: actionForWorkflow(workflow), + CatalogOutputsToWrite: catalogOutputsForPlan(req, outputs, nil, scope, now), } + if workflow == config.WorkflowReplacement { + details.ClearDestinationRoot = true + } + return details +} + +func planWithoutCatalog(ctx context.Context, req Request, outputs []Output, workflow string, scope state.OwnerScope, now time.Time) (catalogPlanDetails, error) { + if err := rejectCatalogUnmanagedCollisions(ctx, req.DestinationBackend, req.DestinationBundlePath, nil, outputs); err != nil { + return catalogPlanDetails{ + Action: ActionFailUnmanaged, + Reason: err.Error(), + }, fmt.Errorf("%s: %s", ActionFailUnmanaged, err) + } + return catalogPlanDetails{ + Action: ActionPublishNew, + CatalogOutputsToWrite: catalogOutputsForPlan(req, outputs, nil, scope, now), + }, nil +} + +func actionForWorkflow(workflow string) Action { + if workflow == config.WorkflowReplacement { + return ActionReplaceCatalog + } + return ActionUpsertAdditive +} + +func catalogOutputsForPlan(req Request, outputs []Output, existing []state.CatalogOutputFile, scope state.OwnerScope, now time.Time) []state.CatalogOutputFile { + files := make([]state.CatalogOutputFile, 0, len(outputs)) + source := state.CatalogSourceIdentity{ + ID: req.SourceBundle.Manifest.ID, + Digest: req.SourceBundle.Manifest.Digest, + Created: req.SourceBundle.Manifest.Created, + } + for _, output := range outputs { + createdAt := now + if existingOutput, ok := state.FindCatalogOutputByPath(existing, output.DestinationPath); ok { + createdAt = existingOutput.CreatedAt + } + file := state.CatalogOutputFile{ + Path: output.DestinationPath, + PipelineID: scope.PipelineID, + DestinationID: scope.DestinationID, + Source: source, + Kind: output.Kind, + URL: output.URL, + SHA256: output.SHA256, + Size: output.Size, + CreatedAt: createdAt, + UpdatedAt: now, + } + if output.Kind == state.OutputKindGenerated { + file.SourcePath = output.SourcePath + file.Transform = output.Transform + } + files = append(files, file) + } + return files +} + +func rejectCatalogUnmanagedCollisions(ctx context.Context, backend storage.Backend, bundlePath string, existing []state.CatalogOutputFile, outputs []Output) error { + managed := catalogOutputPathSet(existing) + for _, output := range outputs { + if _, exists := managed[output.DestinationPath]; exists { + continue + } + destinationPath, err := storage.Join(bundlePath, output.DestinationPath) + if err != nil { + return err + } + if _, err := backend.Stat(ctx, destinationPath); err == nil { + return fmt.Errorf("destination output path %s exists but is not managed by catalog state", storage.DisplayPath(output.DestinationPath)) + } else if !storage.IsNotFound(err) { + return err + } + } + return nil } func outputPaths(outputs []Output) []string { @@ -335,127 +319,31 @@ func outputPaths(outputs []Output) []string { return paths } -func sharedRootOwnershipConflictReason(conflict state.PathOwnershipConflict) string { - return fmt.Sprintf("destination output path %s is owned by %s/%s", conflict.Path, conflict.Owner.PipelineID, conflict.Owner.DestinationID) +func outputPathSet(outputs []Output) map[string]struct{} { + paths := make(map[string]struct{}, len(outputs)) + for _, output := range outputs { + paths[output.DestinationPath] = struct{}{} + } + return paths } -func sharedRootPathOwnershipConflicts(status state.DestinationStatus, scope state.OwnerScope, paths []string) []state.PathOwnershipConflict { - if status.SharedRoot == nil { - return nil +func catalogOutputPathSet(outputs []state.CatalogOutputFile) map[string]struct{} { + paths := make(map[string]struct{}, len(outputs)) + for _, output := range outputs { + paths[output.Path] = struct{}{} } - conflicts := make([]state.PathOwnershipConflict, 0) - seen := make(map[string]struct{}, len(paths)) - for _, path := range paths { - if _, exists := seen[path]; exists { - continue - } - seen[path] = struct{}{} - owner, exists := status.SharedRoot.OutputOwner(path) - if !exists || owner == scope { - continue - } - conflicts = append(conflicts, state.PathOwnershipConflict{ - Path: path, - Owner: owner, - CurrentOwner: scope, - Detail: state.ComparisonDetail{ - Kind: state.ComparisonDetailSharedRootOutputOwner, - Path: path, - CurrentOwner: scope, - ConflictingOwner: owner, - }, - }) - } - return conflicts + return paths } -func sharedRootConflictOutputs(sharedRoot *state.SharedRootState, conflicts []state.PathOwnershipConflict) []state.SharedRootOutputFile { - if sharedRoot == nil || len(conflicts) == 0 { - return nil +func normalizeReconciliation(policy config.ReconciliationPolicy) config.ReconciliationPolicy { + if policy.Mode == "" { + policy.Mode = config.ReconciliationModeReplace } - paths := make(map[string]struct{}, len(conflicts)) - for _, conflict := range conflicts { - paths[conflict.Path] = struct{}{} - } - outputs := make([]state.SharedRootOutputFile, 0, len(conflicts)) - for _, output := range sharedRoot.Outputs { - if _, exists := paths[output.Path]; exists { - outputs = append(outputs, output) - } - } - return outputs + return policy } -func sharedRootTakeoverAllowed(req Request, status state.DestinationStatus, conflict state.PathOwnershipConflict) bool { - if status.SharedRoot == nil { - return false - } - takeover := normalizeTakeover(req.Takeover) - switch takeover.Mode { - case config.TakeoverModeSamePipeline: - return conflict.Owner.PipelineID == req.PipelineID - case config.TakeoverModeSameSource: - owner, ok := status.SharedRoot.Owner(conflict.Owner) - return ok && owner.Source.Manifest.ID == req.SourceBundle.Manifest.ID - case config.TakeoverModeAnyManaged: - _, ok := status.SharedRoot.Owner(conflict.Owner) - return ok - case config.TakeoverModeNever: - return false - default: - return false - } -} - -func rejectSharedRootUnmanagedCollisions(ctx context.Context, backend storage.Backend, bundlePath string, status state.DestinationStatus, scope state.OwnerScope, paths []string) error { - for _, path := range paths { - if pathManagedBySharedRootStatus(status, scope, path) { - continue - } - destinationPath, err := storage.Join(bundlePath, path) - if err != nil { - return err - } - if _, err := backend.Stat(ctx, destinationPath); err == nil { - return fmt.Errorf("destination output path %s exists but is not managed by destination state", storage.DisplayPath(path)) - } else if !storage.IsNotFound(err) { - return err - } - } - return nil -} - -func pathManagedBySharedRootStatus(status state.DestinationStatus, scope state.OwnerScope, path string) bool { - if status.SharedRoot != nil { - _, exists := status.SharedRoot.OutputOwner(path) - return exists - } - if status.State != nil && status.State.PipelineID == scope.PipelineID && status.State.DestinationID == scope.DestinationID { - _, exists := state.FindOutputByPath(status.State.Outputs, path) - return exists - } - return false -} - -func otherOwnerOutputs(status state.DestinationStatus, scope state.OwnerScope) []state.SharedRootOutputFile { - return otherOwnerOutputsExcept(status, scope, nil) -} - -func otherOwnerOutputsExcept(status state.DestinationStatus, scope state.OwnerScope, exclude map[string]struct{}) []state.SharedRootOutputFile { - if status.SharedRoot == nil { - return nil - } - outputs := make([]state.SharedRootOutputFile, 0, len(status.SharedRoot.Outputs)) - for _, output := range status.SharedRoot.Outputs { - if output.Owner == scope { - continue - } - if _, skip := exclude[output.Path]; skip { - continue - } - outputs = append(outputs, output) - } - return outputs +func isReconciliationReplacementAction(action Action) bool { + return action == ActionReplaceOlder || action == ActionReplaceNewer } func sharedRootOutputPathSet(outputs []state.SharedRootOutputFile) map[string]struct{} { @@ -465,109 +353,3 @@ func sharedRootOutputPathSet(outputs []state.SharedRootOutputFile) map[string]st } return paths } - -func currentOwnerOutputs(status state.DestinationStatus, scope state.OwnerScope) []state.SharedRootOutputFile { - if status.SharedRoot != nil { - outputs := make([]state.SharedRootOutputFile, 0, len(status.SharedRoot.Outputs)) - for _, output := range status.SharedRoot.Outputs { - if output.Owner == scope { - outputs = append(outputs, output) - } - } - return outputs - } - if status.State != nil && status.State.PipelineID == scope.PipelineID && status.State.DestinationID == scope.DestinationID { - outputs := make([]state.SharedRootOutputFile, 0, len(status.State.Outputs)) - for _, output := range status.State.Outputs { - outputs = append(outputs, state.SharedRootOutputFile{ - Path: output.Path, - Kind: output.Kind, - SourcePath: output.SourcePath, - Transform: output.Transform, - URL: output.URL, - SHA256: output.SHA256, - Size: output.Size, - Owner: scope, - SourceID: status.State.Source.Manifest.ID, - SourceDigest: status.State.Source.Manifest.Digest, - SourceCreated: status.State.Source.Manifest.Created, - CreatedAt: output.CreatedAt, - UpdatedAt: output.UpdatedAt, - }) - } - return outputs - } - return nil -} - -func actionForComparison(comparison state.Comparison, transfer config.TransferPolicy, force bool) (Action, string) { - switch comparison.Outcome { - case state.OutcomeDestinationAbsent: - return ActionPublishNew, comparison.Reason - case state.OutcomeDestinationUnmanaged: - if force { - return ActionForceReplace, "forced replacement of unmanaged destination content" - } - return ActionFailUnmanaged, comparison.Reason - case state.OutcomeInvalidState: - return ActionFailConflict, comparison.Reason - case state.OutcomeIdentityMismatch, state.OutcomeSameCreatedConflict, state.OutcomeDifferentSourceConflict: - if transfer.OnConflict == config.TransferActionReplace { - return ActionReplaceConflict, comparison.Reason - } - return ActionFailConflict, comparison.Reason - case state.OutcomeSameSource: - if transfer.OnDestinationSame == config.TransferActionFail { - return ActionFailConflict, "destination matches source and transfer policy requires failure" - } - return ActionSkipSame, comparison.Reason - case state.OutcomeDestinationOlder: - if transfer.OnDestinationOlder == config.TransferActionFail { - return ActionFailConflict, "destination is older and transfer policy requires failure" - } - return ActionReplaceOlder, comparison.Reason - case state.OutcomeDestinationNewer: - if transfer.OnDestinationNewer == config.TransferActionReplace { - return ActionReplaceNewer, comparison.Reason - } - if transfer.OnDestinationNewer == config.TransferActionFail { - return ActionFailConflict, "destination is newer and transfer policy requires failure" - } - return ActionSkipDestinationNewer, comparison.Reason - default: - return ActionFailConflict, "unsupported comparison outcome" - } -} - -func isReconciliationReplacementAction(action Action) bool { - return action == ActionReplaceOlder || action == ActionReplaceNewer -} - -func takeoverActionAllowed(req Request, status state.DestinationStatus, comparison state.Comparison, stateMode string, action Action) bool { - if stateMode != config.StateModeSingleOwner || status.State == nil { - return false - } - if action == ActionForceReplace { - return false - } - switch comparison.Detail.Kind { - case state.ComparisonDetailPipelineIDMismatch, - state.ComparisonDetailDestinationIDMismatch, - state.ComparisonDetailDifferentSourceID: - default: - return false - } - takeover := normalizeTakeover(req.Takeover) - switch takeover.Mode { - case config.TakeoverModeSamePipeline: - return status.State.PipelineID == req.PipelineID - case config.TakeoverModeSameSource: - return status.State.Source.Manifest.ID == req.SourceBundle.Manifest.ID - case config.TakeoverModeAnyManaged: - return true - case config.TakeoverModeNever: - return false - default: - return false - } -} diff --git a/internal/publish/plan_test.go b/internal/publish/plan_test.go index 7a31d4c..9a24c2e 100644 --- a/internal/publish/plan_test.go +++ b/internal/publish/plan_test.go @@ -1,36 +1,296 @@ package publish import ( + "context" + "encoding/json" + "strings" "testing" + "time" "gitea.maximumdirect.net/eric/distributor/internal/config" "gitea.maximumdirect.net/eric/distributor/internal/state" + "gitea.maximumdirect.net/eric/distributor/internal/storage" "gitea.maximumdirect.net/eric/distributor/internal/storage/fake" "gitea.maximumdirect.net/eric/distributor/internal/testutil" ) -func TestCompareDestinationFixedPathPreservesDifferentSourceConflict(t *testing.T) { - sourceBackend := fake.New() - sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "bundle", testutil.BundleOptions{}) - destinationState := testutil.DestinationState(sourceBundle.Manifest, testutil.DestinationStateOptions{}) - destinationState.Source.Manifest.ID = "latest.previous" +var ( + planCreatedAt = time.Date(2026, 5, 30, 11, 12, 0, 0, time.UTC) + planUpdatedAt = time.Date(2026, 6, 1, 9, 30, 0, 0, time.UTC) +) - comparison := compareDestination(Request{ +func TestBuildAdditivePublishesCatalogOutputs(t *testing.T) { + _, destinationBackend, req := catalogPlanRequest(t, config.WorkflowAdditive) + + plan, err := Build(context.Background(), req) + if err != nil { + t.Fatalf("Build() error = %v", err) + } + if plan.Action != ActionPublishNew || plan.Workflow != config.WorkflowAdditive { + t.Fatalf("plan action=%s workflow=%s, want publish_new additive", plan.Action, plan.Workflow) + } + 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)) + } + for _, output := range plan.CatalogOutputsToWrite { + if output.PipelineID != "reports" || output.DestinationID != "archive" { + t.Fatalf("catalog output owner = %s/%s", output.PipelineID, output.DestinationID) + } + if output.Source.ID != req.SourceBundle.Manifest.ID || output.Source.Digest != req.SourceBundle.Manifest.Digest || !output.Source.Created.Equal(req.SourceBundle.Manifest.Created) { + t.Fatalf("catalog output source = %#v", output.Source) + } + if output.Kind == state.OutputKindSource && output.SourcePath != "" { + t.Fatalf("source catalog output source_path = %q, want empty", output.SourcePath) + } + if !output.CreatedAt.Equal(planUpdatedAt) || !output.UpdatedAt.Equal(planUpdatedAt) { + t.Fatalf("catalog output times = %s/%s", output.CreatedAt, output.UpdatedAt) + } + } + if _, err := destinationBackend.Stat(context.Background(), storage.StateFileName); !storage.IsNotFound(err) { + t.Fatalf("destination state stat error = %v, want missing", err) + } +} + +func TestBuildAdditiveOverwritesManagedAndRetainsOthers(t *testing.T) { + _, destinationBackend, req := catalogPlanRequest(t, config.WorkflowAdditive) + existing := baseCatalog(req) + existing.Outputs = []state.CatalogOutputFile{ + catalogOutput(req, "reports", "archive", "report.md", state.OutputKindSource, planCreatedAt), + catalogOutput(req, "reports", "web", "old.txt", state.OutputKindSource, planCreatedAt), + } + writeCatalogState(t, destinationBackend, "", existing) + testutil.WriteFakeFile(t, destinationBackend, "report.md", "old") + + plan, err := Build(context.Background(), req) + if err != nil { + t.Fatalf("Build() error = %v", err) + } + if plan.Action != ActionUpsertAdditive { + t.Fatalf("plan action = %s, want %s", plan.Action, ActionUpsertAdditive) + } + if len(plan.CatalogOutputsToDelete) != 0 { + t.Fatalf("delete outputs = %#v, want none", plan.CatalogOutputsToDelete) + } + if len(plan.CatalogOutputsToRetain) != 1 || plan.CatalogOutputsToRetain[0].Path != "old.txt" { + t.Fatalf("retained outputs = %#v, want old.txt", plan.CatalogOutputsToRetain) + } + written, ok := state.FindCatalogOutputByPath(plan.CatalogOutputsToWrite, "report.md") + if !ok { + t.Fatalf("written outputs = %#v, want report.md", plan.CatalogOutputsToWrite) + } + if !written.CreatedAt.Equal(planCreatedAt) || !written.UpdatedAt.Equal(planUpdatedAt) { + t.Fatalf("report.md times = %s/%s, want created preserved and updated now", written.CreatedAt, written.UpdatedAt) + } +} + +func TestBuildReplacementDeletesCurrentOwnerAndRetainsOtherOwners(t *testing.T) { + _, destinationBackend, req := catalogPlanRequest(t, config.WorkflowReplacement) + existing := baseCatalog(req) + existing.Outputs = []state.CatalogOutputFile{ + catalogOutput(req, "reports", "archive", "report.md", state.OutputKindSource, planCreatedAt), + catalogOutput(req, "reports", "archive", "stale.txt", state.OutputKindSource, planCreatedAt), + catalogOutput(req, "reports", "web", "shared.txt", state.OutputKindSource, planCreatedAt), + } + writeCatalogState(t, destinationBackend, "", existing) + + plan, err := Build(context.Background(), req) + if err != nil { + t.Fatalf("Build() error = %v", err) + } + if plan.Action != ActionReplaceCatalog { + t.Fatalf("plan action = %s, want %s", plan.Action, ActionReplaceCatalog) + } + if len(plan.CatalogOutputsToDelete) != 1 || plan.CatalogOutputsToDelete[0].Path != "stale.txt" { + t.Fatalf("delete outputs = %#v, want stale.txt", plan.CatalogOutputsToDelete) + } + if len(plan.CatalogOutputsToRetain) != 1 || plan.CatalogOutputsToRetain[0].Path != "shared.txt" { + t.Fatalf("retained outputs = %#v, want shared.txt", plan.CatalogOutputsToRetain) + } +} + +func TestBuildTransfersManagedPathOwnership(t *testing.T) { + _, destinationBackend, req := catalogPlanRequest(t, config.WorkflowReplacement) + existing := baseCatalog(req) + existing.Outputs = []state.CatalogOutputFile{ + catalogOutput(req, "reports", "web", "report.md", state.OutputKindSource, planCreatedAt), + } + writeCatalogState(t, destinationBackend, "", existing) + + plan, err := Build(context.Background(), req) + if err != nil { + t.Fatalf("Build() error = %v", err) + } + written, ok := state.FindCatalogOutputByPath(plan.CatalogOutputsToWrite, "report.md") + if !ok { + t.Fatalf("written outputs = %#v, want report.md", plan.CatalogOutputsToWrite) + } + if written.PipelineID != "reports" || written.DestinationID != "archive" { + t.Fatalf("written owner = %s/%s, want reports/archive", written.PipelineID, written.DestinationID) + } + if !written.CreatedAt.Equal(planCreatedAt) || !written.UpdatedAt.Equal(planUpdatedAt) { + t.Fatalf("written times = %s/%s", written.CreatedAt, written.UpdatedAt) + } + if len(plan.CatalogOutputsToRetain) != 0 || len(plan.CatalogOutputsToDelete) != 0 { + t.Fatalf("retain=%#v delete=%#v, want no old record for overwritten path", plan.CatalogOutputsToRetain, plan.CatalogOutputsToDelete) + } +} + +func TestBuildRejectsUnmanagedPlannedPathCollision(t *testing.T) { + _, destinationBackend, req := catalogPlanRequest(t, config.WorkflowAdditive) + testutil.WriteFakeFile(t, destinationBackend, "report.md", "unmanaged") + + plan, err := Build(context.Background(), req) + if err == nil { + t.Fatal("Build() error = nil, want unmanaged collision") + } + if plan.Action != ActionFailUnmanaged || !strings.Contains(err.Error(), "not managed by catalog state") { + t.Fatalf("plan action=%s error=%v, want unmanaged catalog collision", plan.Action, err) + } +} + +func TestBuildPlansSupersededLegacyAdditive(t *testing.T) { + _, destinationBackend, req := catalogPlanRequest(t, config.WorkflowAdditive) + legacyState := testutil.DestinationState(req.SourceBundle.Manifest, testutil.DestinationStateOptions{}) + writeJSONState(t, destinationBackend, "", legacyState) + testutil.WriteFakeFile(t, destinationBackend, "report.md", "legacy") + + plan, err := Build(context.Background(), req) + if err != nil { + t.Fatalf("Build() error = %v", err) + } + if plan.SupersededLegacy == nil || plan.SupersededLegacy.SchemaVersion != state.SchemaVersion { + t.Fatalf("superseded legacy = %#v", plan.SupersededLegacy) + } + if plan.Action != ActionUpsertAdditive || plan.ClearDestinationRoot { + t.Fatalf("plan action=%s clear=%t, want additive overwrite without clear", plan.Action, plan.ClearDestinationRoot) + } +} + +func TestBuildPlansSupersededLegacyReplacementClear(t *testing.T) { + _, destinationBackend, req := catalogPlanRequest(t, config.WorkflowReplacement) + legacyState := testutil.DestinationState(req.SourceBundle.Manifest, testutil.DestinationStateOptions{}) + writeJSONState(t, destinationBackend, "", legacyState) + + plan, err := Build(context.Background(), req) + if err != nil { + t.Fatalf("Build() error = %v", err) + } + if plan.Action != ActionReplaceCatalog || !plan.ClearDestinationRoot { + t.Fatalf("plan action=%s clear=%t, want replacement clear", plan.Action, plan.ClearDestinationRoot) + } +} + +func TestBuildRejectsInvalidOrFutureState(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) + testutil.WriteFakeFile(t, destinationBackend, storage.StateFileName, tt.data) + + plan, err := Build(context.Background(), req) + if err == nil { + t.Fatal("Build() error = nil, want conflict") + } + if plan.Action != ActionFailConflict { + t.Fatalf("plan action = %s, want %s", plan.Action, ActionFailConflict) + } + }) + } +} + +func TestValidateRequestRejectsInvalidWorkflow(t *testing.T) { + sourceBackend, destinationBackend, req := catalogPlanRequest(t, "append") + req.SourceBackend = sourceBackend + req.DestinationBackend = destinationBackend + + err := validateRequest(req) + if err == nil { + t.Fatal("validateRequest() error = nil, want invalid workflow") + } + if !strings.Contains(err.Error(), "destination.workflow") { + t.Fatalf("validateRequest() error = %v, want workflow context", err) + } +} + +func catalogPlanRequest(t *testing.T, workflow string) (*fake.Backend, *fake.Backend, Request) { + t.Helper() + sourceBackend := fake.New() + destinationBackend := fake.New() + sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "", testutil.BundleOptions{}) + return sourceBackend, destinationBackend, Request{ PipelineID: "reports", DestinationID: "archive", SourceBundle: sourceBundle, + SourceBackend: sourceBackend, + DestinationBackend: destinationBackend, DestinationBundlePath: "", - PathMapping: config.PathMappingFixed, - State: config.StatePolicy{Mode: config.StateModeSingleOwner}, - }, state.DestinationStatus{State: &destinationState, HasContents: true}) - - if comparison.Outcome != state.OutcomeDifferentSourceConflict { - t.Fatalf("comparison outcome = %s, want %s", comparison.Outcome, state.OutcomeDifferentSourceConflict) - } - if comparison.Detail.Kind != state.ComparisonDetailDifferentSourceID { - t.Fatalf("detail kind = %q, want %q", comparison.Detail.Kind, state.ComparisonDetailDifferentSourceID) - } - if comparison.Detail.CurrentSourceID != sourceBundle.Manifest.ID || comparison.Detail.DestinationSourceID != "latest.previous" { - t.Fatalf("detail = %#v, want source ids", comparison.Detail) + PathMapping: config.PathMappingPreserveRelative, + Publish: config.PublishPolicy{Source: true}, + Workflow: workflow, + DistributorVersion: "test", + Now: planUpdatedAt, } } + +func baseCatalog(req Request) state.CatalogState { + return state.CatalogState{ + SchemaVersion: state.CatalogSchemaVersion, + DistributorVersion: "previous", + CreatedAt: planCreatedAt, + UpdatedAt: planCreatedAt, + State: state.StatePolicy{Mode: state.StateModeCatalog}, + Outputs: []state.CatalogOutputFile{}, + } +} + +func catalogOutput(req Request, pipelineID, destinationID, path, kind string, createdAt time.Time) state.CatalogOutputFile { + sourceSHA := req.SourceBundle.Manifest.Files[0].SHA256 + sourceSize := req.SourceBundle.Manifest.Files[0].Size + for _, file := range req.SourceBundle.Manifest.Files { + if file.Path == path { + sourceSHA = file.SHA256 + sourceSize = file.Size + break + } + } + output := state.CatalogOutputFile{ + Path: path, + PipelineID: pipelineID, + DestinationID: destinationID, + Source: state.CatalogSourceIdentity{ + ID: req.SourceBundle.Manifest.ID, + Digest: req.SourceBundle.Manifest.Digest, + Created: req.SourceBundle.Manifest.Created, + }, + Kind: kind, + SHA256: sourceSHA, + Size: sourceSize, + CreatedAt: createdAt, + UpdatedAt: createdAt, + } + return output +} + +func writeCatalogState(t *testing.T, backend *fake.Backend, relative string, catalog state.CatalogState) { + t.Helper() + writeJSONState(t, backend, relative, catalog) +} + +func writeJSONState(t *testing.T, backend *fake.Backend, relative string, value any) { + t.Helper() + data, err := json.MarshalIndent(value, "", " ") + if err != nil { + t.Fatalf("marshal state: %v", err) + } + statePath, err := storage.StatePath(relative) + if err != nil { + t.Fatalf("state path: %v", err) + } + testutil.WriteFakeFile(t, backend, statePath, string(append(data, '\n'))) +} diff --git a/internal/publish/shared_root_test.go b/internal/publish/shared_root_test.go deleted file mode 100644 index c9fbe5c..0000000 --- a/internal/publish/shared_root_test.go +++ /dev/null @@ -1,632 +0,0 @@ -package publish - -import ( - "context" - "encoding/json" - "strings" - "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" -) - -func TestBuildSharedRootTreatsAbsentOwnerAsPublishable(t *testing.T) { - sourceBackend := fake.New() - sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "bundle", testutil.BundleOptions{ - Files: []testutil.SourceFile{{Path: "report.md", Data: "# Report\n"}}, - }) - destinationBackend := fake.New() - writeFakeSharedRootState(t, destinationBackend, "bundle", sharedRootStateWithOwners(t, sourceBundle.Manifest, false)) - - req := sharedRootRequest(sourceBackend, destinationBackend, sourceBundle, config.ReconciliationModeReplace) - plan, err := Build(context.Background(), req) - if err != nil { - t.Fatalf("Build() error = %v", err) - } - if plan.Action != ActionPublishNew { - t.Fatalf("plan action = %s, want publish_new", plan.Action) - } - if plan.OwnerScope != state.CurrentOwnerScope("reports", "archive") { - t.Fatalf("owner scope = %#v, want reports/archive", plan.OwnerScope) - } - if got, want := len(plan.OtherOwnerOutputs), 1; got != want { - t.Fatalf("other owner output count = %d, want %d", got, want) - } - if got, want := len(plan.OwnerOutputsToWrite), 1; got != want { - t.Fatalf("owner output write count = %d, want %d", got, want) - } - if len(plan.OwnerOutputsToDelete) != 0 || len(plan.RetainedOwnerOutputs) != 0 { - t.Fatalf("delete=%#v retained=%#v, want none", plan.OwnerOutputsToDelete, plan.RetainedOwnerOutputs) - } -} - -func TestBuildSharedRootReplaceDeletesOnlyCurrentOwnerOmittedOutputs(t *testing.T) { - sourceBackend := fake.New() - sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "bundle", testutil.BundleOptions{ - Files: []testutil.SourceFile{{Path: "report.md", Data: "# Report\nNew.\n"}}, - }) - destinationBackend := fake.New() - writeFakeSharedRootState(t, destinationBackend, "bundle", sharedRootStateWithOwners(t, sourceBundle.Manifest, true)) - - req := sharedRootRequest(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 got, want := sharedRootOutputPathList(plan.OwnerOutputsToDelete), "old.md"; got != want { - t.Fatalf("owner outputs to delete = %q, want %q", got, want) - } - if got, want := sharedRootOutputPathList(plan.OtherOwnerOutputs), "other/report.md"; got != want { - t.Fatalf("other owner outputs = %q, want %q", got, want) - } - if len(plan.RetainedOwnerOutputs) != 0 { - t.Fatalf("retained owner outputs = %#v, want none", plan.RetainedOwnerOutputs) - } -} - -func TestBuildSharedRootMergeRetainsCurrentOwnerOmittedOutputs(t *testing.T) { - sourceBackend := fake.New() - sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "bundle", testutil.BundleOptions{ - Files: []testutil.SourceFile{{Path: "report.md", Data: "# Report\nNew.\n"}}, - }) - destinationBackend := fake.New() - writeFakeSharedRootState(t, destinationBackend, "bundle", sharedRootStateWithOwners(t, sourceBundle.Manifest, true)) - - req := sharedRootRequest(sourceBackend, destinationBackend, sourceBundle, config.ReconciliationModeMerge) - plan, err := Build(context.Background(), req) - if err != nil { - t.Fatalf("Build() error = %v", err) - } - if got, want := sharedRootOutputPathList(plan.RetainedOwnerOutputs), "old.md"; got != want { - t.Fatalf("retained owner outputs = %q, want %q", got, want) - } - if len(plan.OwnerOutputsToDelete) != 0 { - t.Fatalf("owner outputs to delete = %#v, want none", plan.OwnerOutputsToDelete) - } -} - -func TestBuildSharedRootRejectsOtherOwnerPathConflict(t *testing.T) { - sourceBackend := fake.New() - sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "bundle", testutil.BundleOptions{ - Files: []testutil.SourceFile{{Path: "report.md", Data: "# Report\n"}}, - }) - destinationBackend := fake.New() - sharedRoot := sharedRootStateWithOwners(t, sourceBundle.Manifest, false) - sharedRoot.Outputs[0].Path = "report.md" - writeFakeSharedRootState(t, destinationBackend, "bundle", sharedRoot) - - req := sharedRootRequest(sourceBackend, destinationBackend, sourceBundle, config.ReconciliationModeReplace) - plan, err := Build(context.Background(), req) - if err == nil || !strings.Contains(err.Error(), "fail_conflict") { - t.Fatalf("Build() error = %v, want fail_conflict", err) - } - if plan.Action != ActionFailConflict { - t.Fatalf("plan action = %s, want fail_conflict", plan.Action) - } -} - -func TestBuildSharedRootPlansOutputTakeoverByPolicy(t *testing.T) { - tests := []struct { - name string - takeover config.TakeoverPolicy - transfer config.TransferPolicy - ownerScope state.OwnerScope - sameSource bool - wantAction Action - wantErr string - }{ - { - name: "default same pipeline allows different destination", - takeover: config.TakeoverPolicy{}, - ownerScope: state.CurrentOwnerScope("reports", "web"), - wantAction: ActionReplaceTakeover, - }, - { - name: "same pipeline refuses different pipeline", - takeover: config.TakeoverPolicy{Mode: config.TakeoverModeSamePipeline}, - ownerScope: state.CurrentOwnerScope("other", "archive"), - wantErr: "fail_conflict", - }, - { - name: "same source allows different pipeline", - takeover: config.TakeoverPolicy{Mode: config.TakeoverModeSameSource}, - ownerScope: state.CurrentOwnerScope("other", "archive"), - sameSource: true, - wantAction: ActionReplaceTakeover, - }, - { - name: "same source refuses different source", - takeover: config.TakeoverPolicy{Mode: config.TakeoverModeSameSource}, - ownerScope: state.CurrentOwnerScope("reports", "web"), - wantErr: "fail_conflict", - }, - { - name: "any managed allows different pipeline", - takeover: config.TakeoverPolicy{Mode: config.TakeoverModeAnyManaged}, - ownerScope: state.CurrentOwnerScope("other", "archive"), - wantAction: ActionReplaceTakeover, - }, - { - name: "never refuses same pipeline", - takeover: config.TakeoverPolicy{Mode: config.TakeoverModeNever}, - ownerScope: state.CurrentOwnerScope("reports", "web"), - wantErr: "fail_conflict", - }, - { - name: "transfer conflict replacement allows managed owner conflict", - takeover: config.TakeoverPolicy{Mode: config.TakeoverModeNever}, - transfer: config.TransferPolicy{OnConflict: config.TransferActionReplace}, - ownerScope: state.CurrentOwnerScope("other", "archive"), - wantAction: ActionReplaceConflict, - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - sourceBackend := fake.New() - sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "bundle", testutil.BundleOptions{ - Files: []testutil.SourceFile{{Path: "report.md", Data: "# Report\n"}}, - }) - destinationBackend := fake.New() - sharedRoot := sharedRootStateWithOwners(t, sourceBundle.Manifest, false) - ownerManifest := sharedRoot.Owners[0].Source.Manifest - if tt.sameSource { - ownerManifest = sourceBundle.Manifest - } - setSharedRootOwnerOutput(t, &sharedRoot, 0, tt.ownerScope, ownerManifest, "report.md") - writeFakeSharedRootState(t, destinationBackend, "bundle", sharedRoot) - - req := sharedRootRequest(sourceBackend, destinationBackend, sourceBundle, config.ReconciliationModeReplace) - req.Takeover = tt.takeover - if tt.transfer.OnConflict != "" { - req.Transfer.OnConflict = tt.transfer.OnConflict - } - plan, err := Build(context.Background(), req) - if tt.wantErr != "" { - if err == nil || !strings.Contains(err.Error(), tt.wantErr) { - t.Fatalf("Build() error = %v, want %q", err, tt.wantErr) - } - if plan.Action != ActionFailConflict { - t.Fatalf("plan action = %s, want fail_conflict", plan.Action) - } - return - } - if err != nil { - t.Fatalf("Build() error = %v", err) - } - if plan.Action != tt.wantAction { - t.Fatalf("plan action = %s, want %s", plan.Action, tt.wantAction) - } - if got, want := sharedRootOutputPathList(plan.TakenOverOwnerOutputs), "report.md"; got != want { - t.Fatalf("taken over outputs = %q, want %q", got, want) - } - }) - } -} - -func TestBuildSharedRootRejectsUnmanagedPathCollision(t *testing.T) { - sourceBackend := fake.New() - sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "bundle", testutil.BundleOptions{ - Files: []testutil.SourceFile{{Path: "report.md", Data: "# Report\n"}}, - }) - destinationBackend := fake.New() - writeFakeSharedRootState(t, destinationBackend, "bundle", sharedRootStateWithOwners(t, sourceBundle.Manifest, false)) - testutil.WriteFakeFile(t, destinationBackend, "bundle/report.md", "unmanaged") - - req := sharedRootRequest(sourceBackend, destinationBackend, sourceBundle, config.ReconciliationModeReplace) - plan, err := Build(context.Background(), req) - if err == nil || !strings.Contains(err.Error(), "fail_unmanaged") { - t.Fatalf("Build() error = %v, want fail_unmanaged", err) - } - if plan.Action != ActionFailUnmanaged { - t.Fatalf("plan action = %s, want fail_unmanaged", plan.Action) - } -} - -func TestExecuteSharedRootPublishesOwnerAndPreservesOtherOwners(t *testing.T) { - sourceBackend := fake.New() - sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "bundle", testutil.BundleOptions{ - Files: []testutil.SourceFile{{Path: "report.md", Data: "# Report\n"}}, - }) - destinationBackend := fake.New() - writeFakeSharedRootState(t, destinationBackend, "bundle", sharedRootStateWithOwners(t, sourceBundle.Manifest, false)) - req := sharedRootRequest(sourceBackend, destinationBackend, sourceBundle, config.ReconciliationModeReplace) - 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, "bundle/report.md", "# Report\n") - testutil.AssertFakeFile(t, destinationBackend, "bundle/other/report.md", "old") - destinationState := readFakeSharedRootState(t, destinationBackend, "bundle") - if got, want := len(destinationState.Owners), 2; got != want { - t.Fatalf("owner count = %d, want %d", got, want) - } - if got, want := destinationState.State.Mode, state.StateModeSharedRoot; got != want { - t.Fatalf("state mode = %q, want %q", got, want) - } - if _, ok := destinationState.Owner(state.CurrentOwnerScope("other", "archive")); !ok { - t.Fatal("other owner missing from shared-root state") - } - if _, ok := destinationState.Owner(state.CurrentOwnerScope("reports", "archive")); !ok { - t.Fatal("current owner missing from shared-root state") - } - if got, want := strings.Join(destinationState.AllManagedOutputPaths(), ","), "other/report.md,report.md"; got != want { - t.Fatalf("managed paths = %q, want %q", got, want) - } -} - -func TestExecuteSharedRootReplaceDeletesOnlyCurrentOwnerOmittedOutputs(t *testing.T) { - sourceBackend := fake.New() - sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "bundle", testutil.BundleOptions{ - Files: []testutil.SourceFile{{Path: "report.md", Data: "# Report\nNew.\n"}}, - }) - destinationBackend := fake.New() - existing := sharedRootStateWithOwners(t, sourceBundle.Manifest, true) - writeFakeSharedRootState(t, destinationBackend, "bundle", existing) - - req := sharedRootRequest(sourceBackend, destinationBackend, sourceBundle, config.ReconciliationModeReplace) - 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, "bundle/report.md", "# Report\nNew.\n") - testutil.AssertFakeMissing(t, destinationBackend, "bundle/old.md") - testutil.AssertFakeFile(t, destinationBackend, "bundle/other/report.md", "old") - destinationState := readFakeSharedRootState(t, destinationBackend, "bundle") - if got, want := strings.Join(destinationState.AllManagedOutputPaths(), ","), "other/report.md,report.md"; got != want { - t.Fatalf("managed paths = %q, want %q", got, want) - } - if !destinationState.CreatedAt.Equal(existing.CreatedAt) { - t.Fatalf("created_at = %s, want %s", destinationState.CreatedAt, existing.CreatedAt) - } - output, ok := findSharedRootOutputForTest(destinationState.Outputs, "report.md") - if !ok { - t.Fatal("report.md missing from shared-root outputs") - } - if !output.CreatedAt.Equal(existing.Outputs[1].CreatedAt) || !output.UpdatedAt.After(existing.Outputs[1].UpdatedAt) { - t.Fatalf("report.md timestamps = created:%s updated:%s, want preserved created and newer updated", output.CreatedAt, output.UpdatedAt) - } -} - -func TestExecuteSharedRootTakeoverReassignsPathAndPreservesUnrelatedOutputs(t *testing.T) { - sourceBackend := fake.New() - sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "bundle", testutil.BundleOptions{ - Files: []testutil.SourceFile{{Path: "report.md", Data: "# Report\nNew.\n"}}, - }) - destinationBackend := fake.New() - sharedRoot := sharedRootStateWithOwners(t, sourceBundle.Manifest, false) - previousScope := state.CurrentOwnerScope("reports", "web") - setSharedRootOwnerOutput(t, &sharedRoot, 0, previousScope, sharedRoot.Owners[0].Source.Manifest, "report.md") - keepOutput := sharedRoot.Outputs[0] - keepOutput.Path = "web/keep.md" - keepOutput.SourcePath = "web/keep.md" - sharedRoot.Outputs = append(sharedRoot.Outputs, keepOutput) - writeFakeSharedRootState(t, destinationBackend, "bundle", sharedRoot) - - req := sharedRootRequest(sourceBackend, destinationBackend, sourceBundle, config.ReconciliationModeReplace) - plan, err := Build(context.Background(), req) - if err != nil { - t.Fatalf("Build() error = %v", err) - } - if plan.Action != ActionReplaceTakeover { - t.Fatalf("plan action = %s, want %s", plan.Action, ActionReplaceTakeover) - } - if err := Execute(context.Background(), req, plan); err != nil { - t.Fatalf("Execute() error = %v", err) - } - - testutil.AssertFakeFile(t, destinationBackend, "bundle/report.md", "# Report\nNew.\n") - testutil.AssertFakeFile(t, destinationBackend, "bundle/web/keep.md", "old") - destinationState := readFakeSharedRootState(t, destinationBackend, "bundle") - reportOutput, ok := findSharedRootOutputForTest(destinationState.Outputs, "report.md") - if !ok { - t.Fatal("report.md missing from shared-root outputs") - } - if reportOutput.Owner != state.CurrentOwnerScope("reports", "archive") { - t.Fatalf("report.md owner = %#v, want reports/archive", reportOutput.Owner) - } - keep, ok := findSharedRootOutputForTest(destinationState.Outputs, "web/keep.md") - if !ok { - t.Fatal("web/keep.md missing from shared-root outputs") - } - if keep.Owner != previousScope { - t.Fatalf("web/keep.md owner = %#v, want reports/web", keep.Owner) - } -} - -func TestExecuteSharedRootTakeoverMergeDoesNotRetainOldSourceOutputs(t *testing.T) { - sourceBackend := fake.New() - sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "bundle", testutil.BundleOptions{ - Files: []testutil.SourceFile{{Path: "report.md", Data: "# Report\nNew.\n"}}, - }) - destinationBackend := fake.New() - sharedRoot := sharedRootStateWithOwners(t, sourceBundle.Manifest, false) - previousScope := state.CurrentOwnerScope("reports", "web") - setSharedRootOwnerOutput(t, &sharedRoot, 0, previousScope, sharedRoot.Owners[0].Source.Manifest, "report.md") - - createdAt := sharedRoot.CreatedAt - oldManifest := sourceBundle.Manifest - oldManifest.ID = "old.source" - oldManifest.Created = oldManifest.Created.Add(-time.Hour) - oldManifest.Files = []bundle.ManifestFile{{ - Path: "old.md", - SHA256: bundle.FileDigest([]byte("old\n")), - Size: int64(len("old\n")), - }} - oldManifest.Digest = bundle.BundleDigest(oldManifest.Files) - currentScope := state.CurrentOwnerScope("reports", "archive") - sharedRoot.Owners = append(sharedRoot.Owners, state.OwnerRecord{ - Scope: currentScope, - Reconciliation: state.ReconciliationPolicy{Mode: config.ReconciliationModeMerge}, - Source: state.SourceState{Manifest: oldManifest}, - }) - sharedRoot.Outputs = append(sharedRoot.Outputs, state.SharedRootOutputFile{ - Path: "old.md", - Kind: state.OutputKindSource, - SourcePath: "old.md", - SHA256: oldManifest.Files[0].SHA256, - Size: oldManifest.Files[0].Size, - Owner: currentScope, - SourceID: oldManifest.ID, - SourceDigest: oldManifest.Digest, - SourceCreated: oldManifest.Created, - CreatedAt: createdAt, - UpdatedAt: createdAt, - }) - writeFakeSharedRootState(t, destinationBackend, "bundle", sharedRoot) - - req := sharedRootRequest(sourceBackend, destinationBackend, sourceBundle, config.ReconciliationModeMerge) - req.PathMapping = config.PathMappingFixed - plan, err := Build(context.Background(), req) - if err != nil { - t.Fatalf("Build() error = %v", err) - } - if plan.Action != ActionReplaceTakeover { - t.Fatalf("plan action = %s, want %s", plan.Action, ActionReplaceTakeover) - } - if got, want := sharedRootOutputPathList(plan.OwnerOutputsToDelete), "old.md"; got != want { - t.Fatalf("owner outputs to delete = %q, want %q", got, want) - } - if len(plan.RetainedOwnerOutputs) != 0 { - t.Fatalf("retained owner outputs = %#v, want none", plan.RetainedOwnerOutputs) - } - if err := Execute(context.Background(), req, plan); err != nil { - t.Fatalf("Execute() error = %v", err) - } - - testutil.AssertFakeFile(t, destinationBackend, "bundle/report.md", "# Report\nNew.\n") - testutil.AssertFakeMissing(t, destinationBackend, "bundle/old.md") - destinationState := readFakeSharedRootState(t, destinationBackend, "bundle") - if got, want := strings.Join(destinationState.AllManagedOutputPaths(), ","), "report.md"; got != want { - t.Fatalf("managed paths = %q, want %q", got, want) - } -} - -func TestExecuteSharedRootMergeRetainsCurrentOwnerOmittedOutputs(t *testing.T) { - sourceBackend := fake.New() - sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "bundle", testutil.BundleOptions{ - Files: []testutil.SourceFile{{Path: "report.md", Data: "# Report\nNew.\n"}}, - }) - destinationBackend := fake.New() - writeFakeSharedRootState(t, destinationBackend, "bundle", sharedRootStateWithOwners(t, sourceBundle.Manifest, true)) - - req := sharedRootRequest(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, "bundle/report.md", "# Report\nNew.\n") - testutil.AssertFakeFile(t, destinationBackend, "bundle/old.md", "old") - testutil.AssertFakeFile(t, destinationBackend, "bundle/other/report.md", "old") - destinationState := readFakeSharedRootState(t, destinationBackend, "bundle") - if got, want := strings.Join(destinationState.AllManagedOutputPaths(), ","), "other/report.md,report.md,old.md"; got != want { - t.Fatalf("managed paths = %q, want %q", got, want) - } -} - -func TestExecuteSharedRootForceReplaceDeletesOnlyBundlePath(t *testing.T) { - sourceBackend := fake.New() - sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "bundle", testutil.BundleOptions{ - Files: []testutil.SourceFile{{Path: "report.md", Data: "# Report\n"}}, - }) - destinationBackend := fake.New() - testutil.WriteFakeFile(t, destinationBackend, "bundle/unmanaged.txt", "unmanaged") - testutil.WriteFakeFile(t, destinationBackend, "bundle/nested/old.txt", "old") - testutil.WriteFakeFile(t, destinationBackend, "bundle-sibling/keep.txt", "keep") - - req := sharedRootRequest(sourceBackend, destinationBackend, sourceBundle, config.ReconciliationModeReplace) - req.Force = true - 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 force_replace", plan.Action) - } - if err := Execute(context.Background(), req, plan); err != nil { - t.Fatalf("Execute() error = %v", err) - } - testutil.AssertFakeFile(t, destinationBackend, "bundle/report.md", "# Report\n") - testutil.AssertFakeMissing(t, destinationBackend, "bundle/unmanaged.txt") - testutil.AssertFakeMissing(t, destinationBackend, "bundle/nested/old.txt") - testutil.AssertFakeFile(t, destinationBackend, "bundle-sibling/keep.txt", "keep") - destinationState := readFakeSharedRootState(t, destinationBackend, "bundle") - if got, want := len(destinationState.Owners), 1; got != want { - t.Fatalf("owner count = %d, want %d", got, want) - } -} - -func sharedRootRequest(sourceBackend, destinationBackend *fake.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}, - State: config.StatePolicy{Mode: config.StateModeSharedRoot}, - Reconciliation: config.ReconciliationPolicy{Mode: reconciliationMode}, - Transfer: defaultTransfer(), - DistributorVersion: "test", - } -} - -func writeFakeSharedRootState(t *testing.T, backend *fake.Backend, relative string, sharedRoot state.SharedRootState) { - t.Helper() - data, err := json.MarshalIndent(sharedRoot, "", " ") - if err != nil { - t.Fatalf("marshal shared-root state: %v", err) - } - statePath, err := storage.StatePath(relative) - if err != nil { - t.Fatalf("state path: %v", err) - } - testutil.WriteFakeFile(t, backend, statePath, string(append(data, '\n'))) - for _, output := range sharedRoot.Outputs { - path, err := storage.Join(relative, output.Path) - if err != nil { - t.Fatalf("join output path: %v", err) - } - testutil.WriteFakeFile(t, backend, path, "old") - } -} - -func readFakeSharedRootState(t *testing.T, backend *fake.Backend, relative string) state.SharedRootState { - t.Helper() - statePath, err := storage.StatePath(relative) - if err != nil { - t.Fatalf("state path: %v", err) - } - data, err := backend.ReadFile(context.Background(), statePath) - if err != nil { - t.Fatalf("read shared-root state: %v", err) - } - destinationState, err := state.ParseSharedRoot(data) - if err != nil { - t.Fatalf("parse shared-root state: %v", err) - } - return destinationState -} - -func findSharedRootOutputForTest(outputs []state.SharedRootOutputFile, path string) (state.SharedRootOutputFile, bool) { - for _, output := range outputs { - if output.Path == path { - return output, true - } - } - return state.SharedRootOutputFile{}, false -} - -func setSharedRootOwnerOutput(t *testing.T, sharedRoot *state.SharedRootState, index int, scope state.OwnerScope, manifest bundle.Manifest, path string) { - t.Helper() - sharedRoot.Owners[index].Scope = scope - sharedRoot.Owners[index].Source = state.SourceState{Manifest: manifest} - sourcePath := path - if len(manifest.Files) > 0 { - sourcePath = manifest.Files[0].Path - } - sharedRoot.Outputs[index].Path = path - sharedRoot.Outputs[index].SourcePath = sourcePath - sharedRoot.Outputs[index].Owner = scope - sharedRoot.Outputs[index].SourceID = manifest.ID - sharedRoot.Outputs[index].SourceDigest = manifest.Digest - sharedRoot.Outputs[index].SourceCreated = manifest.Created - if len(manifest.Files) > 0 { - sharedRoot.Outputs[index].SHA256 = manifest.Files[0].SHA256 - sharedRoot.Outputs[index].Size = manifest.Files[0].Size - } -} - -func sharedRootStateWithOwners(t *testing.T, current bundle.Manifest, includeCurrent bool) state.SharedRootState { - t.Helper() - createdAt := time.Date(2026, 5, 30, 11, 12, 0, 0, time.UTC) - otherManifest := testutil.ValidManifest(testutil.BundleOptions{ - ID: "other.source", - Files: []testutil.SourceFile{{Path: "other/report.md", Data: "# Other\n"}}, - }) - sharedRoot := state.SharedRootState{ - SchemaVersion: state.SharedRootSchemaVersion, - DistributorVersion: "test", - CreatedAt: createdAt, - UpdatedAt: createdAt, - State: state.StatePolicy{Mode: state.StateModeSharedRoot}, - Owners: []state.OwnerRecord{{ - Scope: state.CurrentOwnerScope("other", "archive"), - Reconciliation: state.ReconciliationPolicy{Mode: config.ReconciliationModeReplace}, - Source: state.SourceState{Manifest: otherManifest}, - }}, - Outputs: []state.SharedRootOutputFile{{ - Path: "other/report.md", - Kind: state.OutputKindSource, - SourcePath: "other/report.md", - SHA256: otherManifest.Files[0].SHA256, - Size: otherManifest.Files[0].Size, - Owner: state.CurrentOwnerScope("other", "archive"), - SourceID: otherManifest.ID, - SourceDigest: otherManifest.Digest, - SourceCreated: otherManifest.Created, - CreatedAt: createdAt, - UpdatedAt: createdAt, - }}, - } - if !includeCurrent { - return sharedRoot - } - older := current - older.Created = older.Created.Add(-time.Hour) - older.Files = append([]bundle.ManifestFile(nil), current.Files...) - older.Files = append(older.Files, bundle.ManifestFile{ - Path: "old.md", - SHA256: bundle.FileDigest([]byte("old\n")), - Size: int64(len("old\n")), - }) - older.Digest = bundle.BundleDigest(older.Files) - scope := state.CurrentOwnerScope("reports", "archive") - sharedRoot.Owners = append(sharedRoot.Owners, state.OwnerRecord{ - Scope: scope, - Reconciliation: state.ReconciliationPolicy{Mode: config.ReconciliationModeReplace}, - Source: state.SourceState{Manifest: older}, - }) - for _, file := range older.Files { - sharedRoot.Outputs = append(sharedRoot.Outputs, state.SharedRootOutputFile{ - Path: file.Path, - Kind: state.OutputKindSource, - SourcePath: file.Path, - SHA256: file.SHA256, - Size: file.Size, - Owner: scope, - SourceID: older.ID, - SourceDigest: older.Digest, - SourceCreated: older.Created, - CreatedAt: createdAt, - UpdatedAt: createdAt, - }) - } - return sharedRoot -} - -func sharedRootOutputPathList(outputs []state.SharedRootOutputFile) string { - paths := make([]string, 0, len(outputs)) - for _, output := range outputs { - paths = append(paths, output.Path) - } - return strings.Join(paths, ",") -} diff --git a/internal/publish/takeover_test.go b/internal/publish/takeover_test.go deleted file mode 100644 index d8d3f7b..0000000 --- a/internal/publish/takeover_test.go +++ /dev/null @@ -1,215 +0,0 @@ -package publish - -import ( - "context" - "strings" - "testing" - - "gitea.maximumdirect.net/eric/distributor/internal/bundle" - "gitea.maximumdirect.net/eric/distributor/internal/config" - "gitea.maximumdirect.net/eric/distributor/internal/storage" - "gitea.maximumdirect.net/eric/distributor/internal/storage/fake" - "gitea.maximumdirect.net/eric/distributor/internal/testutil" -) - -func TestBuildPlansSingleOwnerTakeoverByPolicy(t *testing.T) { - tests := []struct { - name string - takeover config.TakeoverPolicy - mutateState func(*bundle.Manifest, *testutil.DestinationStateOptions) - wantAction Action - wantErr string - }{ - { - name: "default same pipeline different source", - takeover: config.TakeoverPolicy{}, - mutateState: func(manifest *bundle.Manifest, opts *testutil.DestinationStateOptions) { - manifest.ID = "other.source" - }, - wantAction: ActionReplaceTakeover, - }, - { - name: "same pipeline different newer source", - takeover: config.TakeoverPolicy{Mode: config.TakeoverModeSamePipeline}, - mutateState: func(manifest *bundle.Manifest, opts *testutil.DestinationStateOptions) { - manifest.ID = "other.source" - manifest.Created = manifest.Created.AddDate(0, 0, 1) - }, - wantAction: ActionReplaceTakeover, - }, - { - name: "default same pipeline different destination", - takeover: config.TakeoverPolicy{}, - mutateState: func(manifest *bundle.Manifest, opts *testutil.DestinationStateOptions) { - opts.DestinationID = "web" - }, - wantAction: ActionReplaceTakeover, - }, - { - name: "same pipeline refuses different pipeline", - takeover: config.TakeoverPolicy{Mode: config.TakeoverModeSamePipeline}, - mutateState: func(manifest *bundle.Manifest, opts *testutil.DestinationStateOptions) { - opts.PipelineID = "other" - }, - wantErr: "fail_conflict", - }, - { - name: "same source allows different pipeline", - takeover: config.TakeoverPolicy{Mode: config.TakeoverModeSameSource}, - mutateState: func(manifest *bundle.Manifest, opts *testutil.DestinationStateOptions) { - opts.PipelineID = "other" - }, - wantAction: ActionReplaceTakeover, - }, - { - name: "same source refuses different source", - takeover: config.TakeoverPolicy{Mode: config.TakeoverModeSameSource}, - mutateState: func(manifest *bundle.Manifest, opts *testutil.DestinationStateOptions) { - manifest.ID = "other.source" - }, - wantErr: "fail_conflict", - }, - { - name: "any managed allows different pipeline", - takeover: config.TakeoverPolicy{Mode: config.TakeoverModeAnyManaged}, - mutateState: func(manifest *bundle.Manifest, opts *testutil.DestinationStateOptions) { - manifest.ID = "other.source" - opts.PipelineID = "other" - }, - wantAction: ActionReplaceTakeover, - }, - { - name: "never refuses different source", - takeover: config.TakeoverPolicy{Mode: config.TakeoverModeNever}, - mutateState: func(manifest *bundle.Manifest, opts *testutil.DestinationStateOptions) { - manifest.ID = "other.source" - }, - wantErr: "fail_conflict", - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - sourceBackend := fake.New() - sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "bundle", testutil.BundleOptions{}) - destinationBackend := fake.New() - destinationManifest := sourceBundle.Manifest - destinationManifest.Files = append([]bundle.ManifestFile(nil), sourceBundle.Manifest.Files...) - opts := testutil.DestinationStateOptions{} - tt.mutateState(&destinationManifest, &opts) - testutil.WriteFakeDestinationState(t, destinationBackend, "bundle", destinationManifest, opts) - - req := takeoverRequest(sourceBackend, destinationBackend, sourceBundle, tt.takeover, config.ReconciliationModeReplace) - plan, err := Build(context.Background(), req) - if tt.wantErr != "" { - if err == nil || !strings.Contains(err.Error(), tt.wantErr) { - t.Fatalf("Build() error = %v, want %q", err, tt.wantErr) - } - return - } - if err != nil { - t.Fatalf("Build() error = %v", err) - } - if plan.Action != tt.wantAction { - t.Fatalf("plan action = %s, want %s", plan.Action, tt.wantAction) - } - }) - } -} - -func TestBuildDoesNotTakeOverInvalidOrUnmanagedDestination(t *testing.T) { - tests := []struct { - name string - prepare func(t *testing.T, backend *fake.Backend) - wantErr string - }{ - { - name: "invalid state", - prepare: func(t *testing.T, backend *fake.Backend) { - t.Helper() - statePath, err := storage.StatePath("bundle") - if err != nil { - t.Fatalf("state path: %v", err) - } - testutil.WriteFakeFile(t, backend, statePath, "{invalid") - }, - wantErr: "fail_conflict", - }, - { - name: "unmanaged content", - prepare: func(t *testing.T, backend *fake.Backend) { - t.Helper() - testutil.WriteFakeFile(t, backend, "bundle/old.txt", "old") - }, - wantErr: "fail_unmanaged", - }, - } - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - sourceBackend := fake.New() - sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "bundle", testutil.BundleOptions{}) - destinationBackend := fake.New() - tt.prepare(t, destinationBackend) - - req := takeoverRequest(sourceBackend, destinationBackend, sourceBundle, config.TakeoverPolicy{Mode: config.TakeoverModeAnyManaged}, config.ReconciliationModeReplace) - _, err := Build(context.Background(), req) - if err == nil || !strings.Contains(err.Error(), tt.wantErr) { - t.Fatalf("Build() error = %v, want %q", err, tt.wantErr) - } - }) - } -} - -func TestExecuteTakeoverMergeDoesNotRetainOldSourceOutputs(t *testing.T) { - sourceBackend := fake.New() - sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "bundle", testutil.BundleOptions{ - Files: []testutil.SourceFile{{Path: "report.md", Data: "# Report\nNew.\n"}}, - }) - destinationBackend := fake.New() - oldManifest := sourceBundle.Manifest - oldManifest.ID = "old.source" - oldManifest.Files = []bundle.ManifestFile{ - {Path: "report.md", SHA256: bundle.FileDigest([]byte("old\n")), Size: int64(len("old\n"))}, - {Path: "summary.txt", SHA256: bundle.FileDigest([]byte("old summary\n")), Size: int64(len("old summary\n"))}, - } - oldManifest.Digest = bundle.BundleDigest(oldManifest.Files) - testutil.WriteFakeDestinationState(t, destinationBackend, "bundle", oldManifest, testutil.DestinationStateOptions{}) - - req := takeoverRequest(sourceBackend, destinationBackend, sourceBundle, config.TakeoverPolicy{Mode: config.TakeoverModeSamePipeline}, config.ReconciliationModeMerge) - plan, err := Build(context.Background(), req) - if err != nil { - t.Fatalf("Build() error = %v", err) - } - if plan.Action != ActionReplaceTakeover { - t.Fatalf("plan action = %s, want %s", plan.Action, ActionReplaceTakeover) - } - if err := Execute(context.Background(), req, plan); err != nil { - t.Fatalf("Execute() error = %v", err) - } - - testutil.AssertFakeFile(t, destinationBackend, "bundle/report.md", "# Report\nNew.\n") - testutil.AssertFakeMissing(t, destinationBackend, "bundle/summary.txt") - destinationState := readFakeState(t, destinationBackend, "bundle") - if got, want := len(destinationState.Outputs), 1; got != want { - t.Fatalf("state output count = %d, want %d", got, want) - } - if got, want := destinationState.Source.Manifest.ID, sourceBundle.Manifest.ID; got != want { - t.Fatalf("state source id = %q, want %q", got, want) - } -} - -func takeoverRequest(sourceBackend, destinationBackend *fake.Backend, sourceBundle bundle.Bundle, takeover config.TakeoverPolicy, 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}, - Takeover: takeover, - Transfer: defaultTransfer(), - DistributorVersion: "test", - } -} diff --git a/internal/state/outputs.go b/internal/state/outputs.go index 50980cb..2887c45 100644 --- a/internal/state/outputs.go +++ b/internal/state/outputs.go @@ -26,6 +26,15 @@ func FindOutputByPath(outputs []OutputFile, path string) (OutputFile, bool) { return OutputFile{}, false } +func FindCatalogOutputByPath(outputs []CatalogOutputFile, path string) (CatalogOutputFile, bool) { + for _, output := range outputs { + if output.Path == path { + return output, true + } + } + return CatalogOutputFile{}, false +} + func MergeOutputFiles(retained, planned []OutputFile) ([]OutputFile, error) { outputs := make([]OutputFile, 0, len(retained)+len(planned)) indexByPath := make(map[string]int, len(retained)+len(planned))