eric/distributor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 7 Wiki Activity

Implement single-owner managed takeover

Browse Source
This commit is contained in:
Eric Rakestraw
2026-06-18 15:18:38 +00:00
parent 598b665307
commit c02106987f
14 changed files with 329 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
internal/app/run_destination.go
View File

@@ -71,6 +71,7 @@ func processDestinationSelection(ctx context.Context, request runDestinationRequ
Links: request.destination.Links,
State: request.destination.State,
Reconciliation: request.destination.Reconciliation,
Takeover: request.destination.Takeover,
Transformers: request.transforms,
Transfer: request.destination.Transfer,
DistributorVersion: Version,

2
internal/app/run_notify.go
View File

@@ -6,7 +6,7 @@ import (
)
func shouldNotify(action publish.Action) bool {
return action == publish.ActionPublishNew || action == publish.ActionReplaceOlder || action == publish.ActionForceReplace
return action == publish.ActionPublishNew || action == publish.ActionReplaceOlder || action == publish.ActionReplaceTakeover || action == publish.ActionForceReplace
}
func notifyEvent(plan publish.Plan) notify.Event {

2
internal/app/run_selection.go
View File

@@ -63,7 +63,7 @@ func fixedPathSelectionWarning(pipelineID, destinationID string, selections []de
}
func isDestructiveFixedPathAction(action publish.Action) bool {
return action == publish.ActionReplaceOlder || action == publish.ActionForceReplace
return action == publish.ActionReplaceOlder || action == publish.ActionReplaceTakeover || action == publish.ActionForceReplace
}
func fixedPathReplacementWarning(plan publish.Plan) OutputWarning {

25
internal/app/run_test.go
View File

@@ -592,8 +592,8 @@ func TestRunFixedPathDryRunWarnsForReplacement(t *testing.T) {
}
output := stdout.String()
for _, want := range []string{
"Warning: pipeline=reports destination=archive path_mapping=fixed action=replace_older replaces destination root for selected_bundle=new",
"bundle=new destination=archive backend=local path_mapping=fixed target=. action=replace_older",
"Warning: pipeline=reports destination=archive path_mapping=fixed action=replace_takeover replaces destination root for selected_bundle=new",
"bundle=new destination=archive backend=local path_mapping=fixed target=. action=replace_takeover",
} {
if !strings.Contains(output, want) {
t.Fatalf("stdout = %q, want substring %q", output, want)
@@ -642,7 +642,7 @@ func TestRunFixedPathSkipsWhenDestinationStateIsNewer(t *testing.T) {
sourceRoot := t.TempDir()
destinationRoot := t.TempDir()
newer := testutil.ValidManifest(testutil.BundleOptions{
ID: "reports.newer",
ID: "reports.same",
Created: testutil.DefaultCreated.Add(time.Hour),
})
writeDestinationState(t, destinationRoot, "", newer)
@@ -650,7 +650,7 @@ func TestRunFixedPathSkipsWhenDestinationStateIsNewer(t *testing.T) {
t.Fatalf("write existing report: %v", err)
}
writeSourceBundle(t, sourceRoot, "older", testBundleOptions{
ID: "reports.older",
ID: "reports.same",
Created: testutil.DefaultCreated,
})
@@ -1526,14 +1526,27 @@ func TestRunSkipsNewerDestination(t *testing.T) {
testutil.AssertFile(t, filepath.Join(destinationRoot, "report.md"), "newer\n")
}
func TestRunFailsOnConflict(t *testing.T) {
func TestRunTakeoverNeverFailsOnConflict(t *testing.T) {
sourceRoot := t.TempDir()
destinationRoot := t.TempDir()
manifest := writeSourceBundle(t, sourceRoot, "", testBundleOptions{})
manifest.ID = "other.source"
writeDestinationState(t, destinationRoot, "", manifest)
configPath := writeConfigFile(t, `
pipelines:
- id: reports
source:
backend: local
path: `+sourceRoot+`
destinations:
- id: archive
backend: local
path: `+destinationRoot+`
takeover:
mode: never
`)
err := Run(context.Background(), RunOptions{ConfigPath: writeLocalConfig(t, sourceRoot, destinationRoot)})
err := Run(context.Background(), RunOptions{ConfigPath: configPath})
if err == nil || !strings.Contains(err.Error(), "fail_conflict") {
t.Fatalf("Run() error = %v, want fail_conflict", err)
}

6
internal/publish/execute.go
View File

@@ -16,7 +16,7 @@ func Execute(ctx context.Context, req Request, plan Plan) error {
switch plan.Action {
case ActionSkipSame, ActionSkipDestinationNewer:
return nil
case ActionPublishNew, ActionReplaceOlder, ActionForceReplace:
case ActionPublishNew, ActionReplaceOlder, ActionReplaceTakeover, ActionForceReplace:
if usesSharedRootState(req, plan) {
return executeSharedRoot(ctx, req, plan)
}
@@ -24,11 +24,11 @@ func Execute(ctx context.Context, req Request, plan Plan) error {
return fmt.Errorf("cannot execute action %s: %s", plan.Action, plan.Reason)
}
if plan.Action == ActionReplaceOlder {
if plan.Action == ActionReplaceOlder || plan.Action == ActionReplaceTakeover {
if plan.ExistingState == nil {
return fmt.Errorf("replace requires existing destination state")
}
if plan.Reconciliation.Mode == config.ReconciliationModeReplace {
if plan.Reconciliation.Mode == config.ReconciliationModeReplace || plan.Action == ActionReplaceTakeover {
if err := req.DestinationBackend.DeleteManagedBundle(ctx, req.DestinationBundlePath, state.ManagedOutputPaths(*plan.ExistingState), storage.DeleteOptions{IgnoreMissing: true, PruneEmptyDirs: true}); err != nil {
return err
}

1
internal/publish/execute_test.go
View File

@@ -189,7 +189,6 @@ func TestExecuteFixedPathSupportsReconciliationModes(t *testing.T) {
})
destinationBackend := fake.New()
older := sourceBundle.Manifest
older.ID = "older.source"
older.Created = older.Created.Add(-time.Hour)
testutil.WriteFakeDestinationState(t, destinationBackend, "", older, testutil.DestinationStateOptions{})

1
internal/publish/force_test.go
View File

@@ -166,6 +166,7 @@ func forceRequest(sourceBackend, destinationBackend *fake.Backend, sourceBundle
DestinationBackend: destinationBackend,
DestinationBundlePath: sourceBundle.RootRelativePath,
Publish: config.PublishPolicy{Source: true},
Takeover: config.TakeoverPolicy{Mode: config.TakeoverModeNever},
Transfer: transfer,
DistributorVersion: "test",
}

71
internal/publish/plan.go
View File

@@ -21,6 +21,7 @@ const (
ActionFailConflict Action = "fail_conflict"
ActionFailUnmanaged Action = "fail_unmanaged"
ActionForceReplace Action = "force_replace"
ActionReplaceTakeover Action = "replace_takeover"
)
type Request struct {
@@ -36,6 +37,7 @@ type Request struct {
Links *config.Links
State config.StatePolicy
Reconciliation config.ReconciliationPolicy
Takeover config.TakeoverPolicy
Transformers TransformerResolver
Transfer config.TransferPolicy
DistributorVersion string
@@ -96,10 +98,14 @@ func Build(ctx context.Context, req Request) (Plan, error) {
if err != nil {
return Plan{}, err
}
comparison := compareDestination(req, status)
action, reason := actionForComparison(comparison, req.Transfer, req.Force)
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
}
plan := Plan{
PipelineID: req.PipelineID,
DestinationID: req.DestinationID,
@@ -162,6 +168,11 @@ func validateRequest(req Request) error {
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 nil
}
@@ -179,6 +190,13 @@ func normalizeState(policy config.StatePolicy) config.StatePolicy {
return policy
}
func normalizeTakeover(policy config.TakeoverPolicy) config.TakeoverPolicy {
if policy.Mode == "" {
policy.Mode = config.TakeoverModeSamePipeline
}
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)
@@ -207,24 +225,6 @@ func compareDestination(req Request, status state.DestinationStatus) state.Compa
return comparison
}
comparison := state.Compare(req.SourceBundle.Manifest, req.PipelineID, req.DestinationID, status)
if req.PathMapping != config.PathMappingFixed || comparison.Outcome != state.OutcomeDifferentSourceConflict || status.State == nil {
return comparison
}
destinationManifest := status.State.Source.Manifest
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
}
@@ -296,7 +296,7 @@ func planSharedRootOwner(ctx context.Context, req Request, status state.Destinat
func isWriteAction(action Action) bool {
switch action {
case ActionPublishNew, ActionReplaceOlder, ActionForceReplace:
case ActionPublishNew, ActionReplaceOlder, ActionReplaceTakeover, ActionForceReplace:
return true
default:
return false
@@ -439,3 +439,32 @@ func actionForComparison(comparison state.Comparison, transfer config.TransferPo
return ActionFailConflict, "unsupported comparison outcome"
}
}
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
}
}

15
internal/publish/plan_test.go
View File

@@ -2,7 +2,6 @@ package publish
import (
"testing"
"time"
"gitea.maximumdirect.net/eric/distributor/internal/config"
"gitea.maximumdirect.net/eric/distributor/internal/state"
@@ -10,12 +9,11 @@ import (
"gitea.maximumdirect.net/eric/distributor/internal/testutil"
)
func TestCompareDestinationFixedPathReportsDestinationNewerDetail(t *testing.T) {
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"
destinationState.Source.Manifest.Created = sourceBundle.Manifest.Created.Add(time.Hour)
comparison := compareDestination(Request{
PipelineID: "reports",
@@ -26,14 +24,11 @@ func TestCompareDestinationFixedPathReportsDestinationNewerDetail(t *testing.T)
State: config.StatePolicy{Mode: config.StateModeSingleOwner},
}, state.DestinationStatus{State: &destinationState, HasContents: true})
if comparison.Outcome != state.OutcomeDestinationNewer {
t.Fatalf("comparison outcome = %s, want %s", comparison.Outcome, state.OutcomeDestinationNewer)
if comparison.Outcome != state.OutcomeDifferentSourceConflict {
t.Fatalf("comparison outcome = %s, want %s", comparison.Outcome, state.OutcomeDifferentSourceConflict)
}
if comparison.Reason != "fixed destination source is newer than selected source" {
t.Fatalf("comparison reason = %q, want fixed-path newer reason", comparison.Reason)
}
if comparison.Detail.Kind != state.ComparisonDetailDestinationNewer {
t.Fatalf("detail kind = %q, want %q", comparison.Detail.Kind, state.ComparisonDetailDestinationNewer)
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)

215
internal/publish/takeover_test.go Normal file
View File

@@ -0,0 +1,215 @@
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",
}
}