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))