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

Add explicit force replacement workflow

Browse Source
This commit is contained in:
Eric Rakestraw
2026-05-31 17:29:20 +00:00
parent 7a174ce5f1
commit 48169dc8b4
28 changed files with 811 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
internal/adapters/local/backend.go
View File

@@ -267,6 +267,51 @@ func (b *Backend) DeleteManagedBundle(ctx context.Context, bundlePath string, ma
return nil
}
func (b *Backend) DeletePrefix(ctx context.Context, prefix string, opts storage.DeleteOptions) error {
if err := ctx.Err(); err != nil {
return err
}
nativePrefix, err := b.nativePath(prefix, true)
if err != nil {
return err
}
if err := b.rejectSymlinkAncestors(nativePrefix, false); err != nil {
return err
}
if prefix == "" {
entries, err := os.ReadDir(nativePrefix)
if err != nil {
if opts.IgnoreMissing && errors.Is(err, fs.ErrNotExist) {
return nil
}
return b.translateError(storage.OpDeletePrefix, prefix, err)
}
for _, entry := range entries {
if err := ctx.Err(); err != nil {
return err
}
child := filepath.Join(nativePrefix, entry.Name())
if err := os.RemoveAll(child); err != nil {
return b.translateError(storage.OpDeletePrefix, entry.Name(), err)
}
}
return nil
}
if _, err := os.Lstat(nativePrefix); err != nil {
if opts.IgnoreMissing && errors.Is(err, fs.ErrNotExist) {
return nil
}
return b.translateError(storage.OpDeletePrefix, prefix, err)
}
if err := os.RemoveAll(nativePrefix); err != nil {
return b.translateError(storage.OpDeletePrefix, prefix, err)
}
if opts.PruneEmptyDirs {
b.pruneEmptyParents(filepath.Dir(nativePrefix))
}
return nil
}
func (b *Backend) nativePath(logicalPath string, allowEmpty bool) (string, error) {
if logicalPath == "" {
if !allowEmpty {

24
internal/adapters/local/backend_test.go
View File

@@ -169,6 +169,30 @@ func TestBackendManagedDeletion(t *testing.T) {
}
}
func TestBackendDeletePrefixStaysWithinPrefix(t *testing.T) {
backend := newBackend(t)
mustWrite(t, backend, "bundle/report.md", "report")
mustWrite(t, backend, "bundle/nested/old.txt", "old")
mustWrite(t, backend, "bundle-sibling/keep.txt", "keep")
mustWrite(t, backend, "outside.txt", "outside")
if err := backend.DeletePrefix(context.Background(), "bundle", storage.DeleteOptions{IgnoreMissing: true, PruneEmptyDirs: true}); err != nil {
t.Fatalf("DeletePrefix() error = %v", err)
}
if _, err := backend.Stat(context.Background(), "bundle/report.md"); !storage.IsNotFound(err) {
t.Fatalf("deleted file stat error = %v, want not found", err)
}
if _, err := backend.Stat(context.Background(), "bundle/nested/old.txt"); !storage.IsNotFound(err) {
t.Fatalf("deleted nested file stat error = %v, want not found", err)
}
if _, err := backend.Stat(context.Background(), "bundle-sibling/keep.txt"); err != nil {
t.Fatalf("sibling stat error = %v", err)
}
if _, err := backend.Stat(context.Background(), "outside.txt"); err != nil {
t.Fatalf("outside stat error = %v", err)
}
}
func TestBackendHasAny(t *testing.T) {
backend := newBackend(t)
found, err := backend.HasAny(context.Background(), "missing")

86
internal/adapters/s3/backend.go
View File

@@ -242,33 +242,87 @@ func (b *Backend) DeleteManagedBundle(ctx context.Context, bundlePath string, ma
return err
}
for _, target := range targets {
key, err := b.objectKey(target, false)
if err := b.deleteObject(ctx, storage.OpDeleteManagedBundle, target, opts); err != nil {
return err
}
}
return nil
}
func (b *Backend) DeletePrefix(ctx context.Context, logicalPrefix string, opts storage.DeleteOptions) error {
if err := ctx.Err(); err != nil {
return err
}
if err := storage.ValidatePrefix(logicalPrefix); err != nil {
return err
}
found := false
if logicalPrefix != "" {
key, err := b.objectKey(logicalPrefix, false)
if err != nil {
return err
}
if key == b.prefix {
return storage.NewError(storage.OpDeleteManagedBundle, BackendName, target, storage.ErrInvalidPath, nil)
}
if !opts.IgnoreMissing {
_, err := b.client.HeadObject(ctx, &awss3.HeadObjectInput{
Bucket: aws.String(b.bucket),
Key: aws.String(key),
})
if err != nil {
return b.translateError(storage.OpDeleteManagedBundle, target, err)
_, err = b.client.HeadObject(ctx, &awss3.HeadObjectInput{
Bucket: aws.String(b.bucket),
Key: aws.String(key),
})
if err == nil {
found = true
if err := b.deleteObject(ctx, storage.OpDeletePrefix, logicalPrefix, storage.DeleteOptions{IgnoreMissing: false}); err != nil {
return err
}
} else if !isNotFound(err) {
return b.translateError(storage.OpDeletePrefix, logicalPrefix, err)
}
_, err = b.client.DeleteObject(ctx, &awss3.DeleteObjectInput{
}
var entries []storage.Entry
if err := b.walkObjects(ctx, logicalPrefix, storage.WalkOptions{Recursive: true}, func(entry storage.Entry) error {
if entry.Type == storage.EntryTypeFile {
entries = append(entries, entry)
}
return nil
}); err != nil {
return err
}
for _, entry := range entries {
found = true
if err := b.deleteObject(ctx, storage.OpDeletePrefix, entry.Path, storage.DeleteOptions{IgnoreMissing: true}); err != nil {
return err
}
}
if !found && !opts.IgnoreMissing {
return storage.NewError(storage.OpDeletePrefix, BackendName, logicalPrefix, storage.ErrNotFound, nil)
}
return nil
}
func (b *Backend) deleteObject(ctx context.Context, op, logicalPath string, opts storage.DeleteOptions) error {
key, err := b.objectKey(logicalPath, false)
if err != nil {
return err
}
if key == b.prefix {
return storage.NewError(op, BackendName, logicalPath, storage.ErrInvalidPath, nil)
}
if !opts.IgnoreMissing {
_, err := b.client.HeadObject(ctx, &awss3.HeadObjectInput{
Bucket: aws.String(b.bucket),
Key: aws.String(key),
})
if err != nil {
if opts.IgnoreMissing && isNotFound(err) {
continue
}
return b.translateError(storage.OpDeleteManagedBundle, target, err)
return b.translateError(op, logicalPath, err)
}
}
_, err = b.client.DeleteObject(ctx, &awss3.DeleteObjectInput{
Bucket: aws.String(b.bucket),
Key: aws.String(key),
})
if err != nil {
if opts.IgnoreMissing && isNotFound(err) {
return nil
}
return b.translateError(op, logicalPath, err)
}
return nil
}

28
internal/adapters/s3/backend_test.go
View File

@@ -191,6 +191,34 @@ func TestDeleteManagedBundleDeletesOnlyManagedTargets(t *testing.T) {
}
}
func TestDeletePrefixStaysWithinPrefix(t *testing.T) {
client := newFakeClient(map[string]string{
"root/bundle/report.md": "report",
"root/bundle/nested/old.txt": "old",
"root/bundle-sibling/keep.txt": "keep",
"root/outside.txt": "outside",
"other-root/bundle/report.md": "other",
"root/.distributor-prefix-marker": "marker",
})
backend := newTestBackend(t, "root", client)
if err := backend.DeletePrefix(context.Background(), "bundle", storage.DeleteOptions{IgnoreMissing: true}); err != nil {
t.Fatalf("DeletePrefix() error = %v", err)
}
for _, deleted := range []string{"root/bundle/report.md", "root/bundle/nested/old.txt"} {
if _, ok := client.objects[deleted]; ok {
t.Fatalf("%s still exists", deleted)
}
}
for _, kept := range []string{"root/bundle-sibling/keep.txt", "root/outside.txt", "other-root/bundle/report.md", "root/.distributor-prefix-marker"} {
if _, ok := client.objects[kept]; !ok {
t.Fatalf("%s was deleted", kept)
}
}
if got, want := sortedStrings(client.deleteKeys), []string{"root/bundle/nested/old.txt", "root/bundle/report.md"}; !equalStrings(got, want) {
t.Fatalf("deleted keys = %v, want %v", got, want)
}
}
func newTestBackend(t *testing.T, prefix string, client *fakeClient) *Backend {
t.Helper()
if client == nil {

76
internal/adapters/ssh/backend.go
View File

@@ -292,6 +292,82 @@ func (b *Backend) DeleteManagedBundle(ctx context.Context, bundlePath string, ma
return nil
}
func (b *Backend) DeletePrefix(ctx context.Context, prefix string, opts storage.DeleteOptions) error {
if err := ctx.Err(); err != nil {
return err
}
if err := storage.ValidatePrefix(prefix); err != nil {
return err
}
var entries []storage.Entry
if prefix != "" {
entry, err := b.Stat(ctx, prefix)
if err != nil {
if opts.IgnoreMissing && storage.IsNotFound(err) {
return nil
}
return err
}
if entry.Type != storage.EntryTypeDirectory {
return b.deleteEntry(ctx, entry, opts)
}
entries = append(entries, entry)
}
if err := b.Walk(ctx, prefix, storage.WalkOptions{Recursive: true}, func(entry storage.Entry) error {
entries = append(entries, entry)
return nil
}); err != nil {
return err
}
if prefix != "" && len(entries) == 1 {
if err := b.deleteEntry(ctx, entries[0], opts); err != nil {
return err
}
if opts.PruneEmptyDirs {
b.pruneEmptyParents(parentOf(prefix))
}
return nil
}
sort.Slice(entries, func(i, j int) bool {
return strings.Count(entries[i].Path, "/") > strings.Count(entries[j].Path, "/")
})
for _, entry := range entries {
if entry.Path == "" {
continue
}
if err := b.deleteEntry(ctx, entry, storage.DeleteOptions{IgnoreMissing: true}); err != nil {
return err
}
}
if opts.PruneEmptyDirs {
b.pruneEmptyParents(parentOf(prefix))
}
return nil
}
func (b *Backend) deleteEntry(ctx context.Context, entry storage.Entry, opts storage.DeleteOptions) error {
if err := ctx.Err(); err != nil {
return err
}
nativePath, err := b.nativePath(entry.Path, false)
if err != nil {
return err
}
var removeErr error
if entry.Type == storage.EntryTypeDirectory {
removeErr = b.client.RemoveDirectory(nativePath)
} else {
removeErr = b.client.Remove(nativePath)
}
if removeErr != nil {
if opts.IgnoreMissing && isNotExist(removeErr) {
return nil
}
return b.translateError(storage.OpDeletePrefix, entry.Path, removeErr)
}
return nil
}
func (b *Backend) walkDirectory(ctx context.Context, logicalPrefix, nativePrefix string, opts storage.WalkOptions, emit func(storage.Entry) error) error {
entries, err := b.client.ReadDir(nativePrefix)
if err != nil {

9
internal/app/run.go
View File

@@ -17,6 +17,7 @@ import (
type RunOptions struct {
ConfigPath string
DryRun bool
Force bool
Stdout io.Writer
Notifier notify.Notifier
}
@@ -117,6 +118,7 @@ func runConfigWithBackendFactory(ctx context.Context, cfg config.Config, options
Transformers: transforms,
Transfer: destination.Transfer,
DistributorVersion: Version,
Force: options.Force,
}
plan, err := publish.Build(ctx, req)
if err != nil && plan.DestinationID == "" {
@@ -246,7 +248,7 @@ func writeSSHWarnings(w io.Writer, pipeline config.Pipeline) error {
}
func shouldNotify(action publish.Action) bool {
return action == publish.ActionPublishNew || action == publish.ActionReplaceOlder
return action == publish.ActionPublishNew || action == publish.ActionReplaceOlder || action == publish.ActionForceReplace
}
func notifyEvent(plan publish.Plan) notify.Event {
@@ -276,6 +278,7 @@ type runSummary struct {
planned int
publishNew int
replaceOlder int
forceReplace int
skipped int
failures int
}
@@ -287,6 +290,8 @@ func (s *runSummary) recordPlan(action publish.Action) {
s.publishNew++
case publish.ActionReplaceOlder:
s.replaceOlder++
case publish.ActionForceReplace:
s.forceReplace++
case publish.ActionSkipSame, publish.ActionSkipDestinationNewer:
s.skipped++
}
@@ -301,7 +306,7 @@ func (s runSummary) Line() string {
if s.failures > 0 {
status = "failed"
}
return fmt.Sprintf("Final status: %s planned=%d publish_new=%d replace_older=%d skipped=%d failed=%d dry_run=%t", status, s.planned, s.publishNew, s.replaceOlder, s.skipped, s.failures, s.dryRun)
return fmt.Sprintf("Final status: %s planned=%d publish_new=%d replace_older=%d force_replace=%d skipped=%d failed=%d dry_run=%t", status, s.planned, s.publishNew, s.replaceOlder, s.forceReplace, s.skipped, s.failures, s.dryRun)
}
type runFailure struct {

90
internal/app/run_test.go
View File

@@ -41,7 +41,7 @@ func TestRunDryRunPrintsConfigSummary(t *testing.T) {
"Configured pipelines: 1",
"- pipeline=reports source=local bundles=1 destinations=archive",
"bundle=. destination=archive backend=local action=publish_new outputs=report.md,summary.txt",
"Final status: ok planned=1 publish_new=1 replace_older=0 skipped=0 failed=0 dry_run=true",
"Final status: ok planned=1 publish_new=1 replace_older=0 force_replace=0 skipped=0 failed=0 dry_run=true",
} {
if !strings.Contains(output, want) {
t.Fatalf("Run() output = %q, want substring %q", output, want)
@@ -301,7 +301,7 @@ func TestRunContinuesAfterDestinationFailure(t *testing.T) {
for _, want := range []string{
"destination=archive-one backend=local action=error",
"destination=archive-two backend=local action=publish_new",
"Final status: failed planned=1 publish_new=1 replace_older=0 skipped=0 failed=1 dry_run=false",
"Final status: failed planned=1 publish_new=1 replace_older=0 force_replace=0 skipped=0 failed=1 dry_run=false",
} {
if !strings.Contains(output, want) {
t.Fatalf("stdout = %q, want substring %q", output, want)
@@ -497,6 +497,32 @@ func TestRunFailsOnUnmanagedDestination(t *testing.T) {
}
}
func TestRunForceReplacesUnmanagedDestination(t *testing.T) {
sourceRoot := t.TempDir()
destinationRoot := t.TempDir()
writeSourceBundle(t, sourceRoot, "", testBundleOptions{})
if err := os.WriteFile(filepath.Join(destinationRoot, "unmanaged.txt"), []byte("old"), 0o600); err != nil {
t.Fatalf("write unmanaged file: %v", err)
}
var stdout bytes.Buffer
err := Run(context.Background(), RunOptions{
ConfigPath: writeLocalConfig(t, sourceRoot, destinationRoot),
Force: true,
Stdout: &stdout,
})
if err != nil {
t.Fatalf("Run() error = %v", err)
}
if !strings.Contains(stdout.String(), "action=force_replace") {
t.Fatalf("stdout = %q, want force_replace", stdout.String())
}
if _, err := os.Stat(filepath.Join(destinationRoot, "unmanaged.txt")); !os.IsNotExist(err) {
t.Fatalf("unmanaged file stat error = %v, want not exist", err)
}
assertFile(t, filepath.Join(destinationRoot, "report.md"), "# Report\nSunny.\n")
}
func TestRunFansOutToLocalDestinations(t *testing.T) {
sourceRoot := t.TempDir()
firstDestination := t.TempDir()
@@ -567,7 +593,7 @@ func TestRunExercisesRemoteBackendShapesThroughCommonPath(t *testing.T) {
"pipeline=local-to-ssh source=local",
"destination=ssh-archive backend=ssh action=publish_new",
"pipeline=ssh-to-local source=ssh",
"Final status: ok planned=4 publish_new=4 replace_older=0 skipped=0 failed=0 dry_run=true",
"Final status: ok planned=4 publish_new=4 replace_older=0 force_replace=0 skipped=0 failed=0 dry_run=true",
} {
if !strings.Contains(dryRunOutput.String(), want) {
t.Fatalf("dry-run output = %q, want substring %q", dryRunOutput.String(), want)
@@ -598,6 +624,50 @@ func TestRunExercisesRemoteBackendShapesThroughCommonPath(t *testing.T) {
}
}
func TestRunForceReplacementStaysWithinRemoteBundlePaths(t *testing.T) {
localSourceRoot := t.TempDir()
writeSourceBundle(t, localSourceRoot, "bundle", testBundleOptions{})
s3Destination := fake.New()
sshDestination := fake.New()
mustWriteFake(t, s3Destination, "bundle/old.txt", "old")
mustWriteFake(t, s3Destination, "bundle-sibling/keep.txt", "keep")
mustWriteFake(t, sshDestination, "bundle/old.txt", "old")
mustWriteFake(t, sshDestination, "bundle-sibling/keep.txt", "keep")
cfg := config.Config{Pipelines: []config.Pipeline{{
ID: "reports",
Source: config.Backend{Backend: config.BackendLocal, Path: localSourceRoot},
Destinations: []config.Destination{
{
ID: "object-archive",
Backend: config.BackendS3,
Endpoint: "http://s3.test",
Bucket: "destination-bucket",
},
{
ID: "ssh-archive",
Backend: config.BackendSSH,
Host: "ssh.test",
Path: "/destination",
},
},
}}}
config.ApplyDefaults(&cfg)
provider := fakeBackendFactoryProvider(t, map[string]storage.Backend{
"s3:destination-bucket": s3Destination,
"ssh:/destination": sshDestination,
})
if err := runConfigWithBackendFactory(context.Background(), cfg, RunOptions{Force: true}, provider); err != nil {
t.Fatalf("Run() error = %v", err)
}
assertFakeFile(t, s3Destination, "bundle/report.md", "# Report\nSunny.\n")
assertFakeMissing(t, s3Destination, "bundle/old.txt")
assertFakeFile(t, s3Destination, "bundle-sibling/keep.txt", "keep")
assertFakeFile(t, sshDestination, "bundle/report.md", "# Report\nSunny.\n")
assertFakeMissing(t, sshDestination, "bundle/old.txt")
assertFakeFile(t, sshDestination, "bundle-sibling/keep.txt", "keep")
}
func TestRunDryRunDoesNotWrite(t *testing.T) {
sourceRoot := t.TempDir()
destinationRoot := t.TempDir()
@@ -755,6 +825,20 @@ func assertFakeFile(t *testing.T, backend *fake.Backend, path, want string) {
}
}
func assertFakeMissing(t *testing.T, backend *fake.Backend, path string) {
t.Helper()
if _, err := backend.Stat(context.Background(), path); !storage.IsNotFound(err) {
t.Fatalf("fake file %s stat error = %v, want not found", path, err)
}
}
func mustWriteFake(t *testing.T, backend *fake.Backend, path, data string) {
t.Helper()
if _, err := backend.WriteFile(context.Background(), path, []byte(data), storage.WriteOptions{}); err != nil {
t.Fatalf("write fake file %s: %v", path, err)
}
}
func crossBackendConfig(localSourceRoot, s3ToLocalDestination, sshToLocalDestination string) config.Config {
cfg := config.Config{
Pipelines: []config.Pipeline{

26
internal/cli/root_test.go
View File

@@ -180,6 +180,32 @@ func TestExecuteRunDryRun(t *testing.T) {
}
}
func TestExecuteRunForceDryRunReportsWithoutWriting(t *testing.T) {
sourceRoot := t.TempDir()
destinationRoot := t.TempDir()
testutil.WriteSourceBundle(t, sourceRoot, "", testutil.BundleOptions{})
if err := os.WriteFile(filepath.Join(destinationRoot, "unmanaged.txt"), []byte("old"), 0o600); err != nil {
t.Fatalf("write unmanaged file: %v", err)
}
configPath := testutil.WriteMinimalLocalConfig(t, sourceRoot, destinationRoot)
var stdout, stderr bytes.Buffer
code := Execute(context.Background(), []string{"run", "--config", configPath, "--force", "--dry-run"}, &stdout, &stderr)
if code != exitOK {
t.Fatalf("exit code = %d, want %d; stderr = %q", code, exitOK, stderr.String())
}
if !strings.Contains(stdout.String(), "action=force_replace") {
t.Fatalf("stdout = %q, want force_replace", stdout.String())
}
if _, err := os.Stat(filepath.Join(destinationRoot, "unmanaged.txt")); err != nil {
t.Fatalf("unmanaged file stat error = %v", err)
}
if _, err := os.Stat(filepath.Join(destinationRoot, storage.StateFileName)); !os.IsNotExist(err) {
t.Fatalf("state stat error = %v, want not exist", err)
}
}
func TestExecuteRunRejectsExtraPositionalArgs(t *testing.T) {
var stdout, stderr bytes.Buffer

7
internal/cli/run.go
View File

@@ -19,6 +19,7 @@ func runCommand(ctx context.Context, args []string, stdout, stderr io.Writer) in
flags.SetOutput(stderr)
configPath := flags.String("config", "", "path to config file")
dryRun := flags.Bool("dry-run", false, "load and validate config without publishing")
force := flags.Bool("force", false, "allow explicit destructive replacement for supported conflicts")
if err := flags.Parse(args); err != nil {
return exitUsage
}
@@ -29,6 +30,7 @@ func runCommand(ctx context.Context, args []string, stdout, stderr io.Writer) in
if err := app.Run(ctx, app.RunOptions{
ConfigPath: *configPath,
DryRun: *dryRun,
Force: *force,
Stdout: stdout,
}); err != nil {
return fail(stderr, err)
@@ -38,13 +40,14 @@ func runCommand(ctx context.Context, args []string, stdout, stderr io.Writer) in
func printRunHelp(w io.Writer) {
fmt.Fprint(w, `Usage:
distributor run --config <path> --dry-run
distributor run --config <path> [--dry-run] [--force]
Options:
--config <path> Path to config file
--dry-run Load and validate config without publishing
--force Allow explicit destructive replacement for supported conflicts
Run discovers local source bundles, plans each configured destination, publishes
Run discovers configured source bundles, plans each destination, publishes
selected outputs unless --dry-run is set, and prints a final status summary.
`)
}

8
internal/config/validate.go
View File

@@ -174,11 +174,11 @@ func validateTransferPolicy(errs ValidationErrors, context string, policy Transf
if policy.OnDestinationOlder != TransferActionReplace && policy.OnDestinationOlder != TransferActionFail {
errs = append(errs, context+".on_destination_older must be replace or fail")
}
if policy.OnDestinationNewer != TransferActionSkip && policy.OnDestinationNewer != TransferActionFail {
errs = append(errs, context+".on_destination_newer must be skip or fail")
if policy.OnDestinationNewer != TransferActionSkip && policy.OnDestinationNewer != TransferActionFail && policy.OnDestinationNewer != TransferActionReplace {
errs = append(errs, context+".on_destination_newer must be skip, replace, or fail")
}
if policy.OnConflict != TransferActionFail {
errs = append(errs, context+".on_conflict must be fail")
if policy.OnConflict != TransferActionFail && policy.OnConflict != TransferActionReplace {
errs = append(errs, context+".on_conflict must be fail or replace")
}
return errs
}

23
internal/config/validate_test.go
View File

@@ -47,6 +47,29 @@ func TestValidateChecksPublishTransformPolicy(t *testing.T) {
}
}
func TestValidateAcceptsForceReplacementTransferActions(t *testing.T) {
cfg := Config{Pipelines: []Pipeline{{
ID: "reports",
Source: Backend{
Backend: BackendLocal,
Path: "/source",
},
Destinations: []Destination{{
ID: "archive",
Backend: BackendLocal,
Path: "/destination",
Transfer: TransferPolicy{
OnDestinationNewer: TransferActionReplace,
OnConflict: TransferActionReplace,
},
}},
}}}
ApplyDefaults(&cfg)
if err := Validate(cfg); err != nil {
t.Fatalf("Validate() error = %v", err)
}
}
type publishTransformPolicyCase struct {
name string
publish PublishPolicy

10
internal/publish/execute.go
View File

@@ -14,7 +14,7 @@ func Execute(ctx context.Context, req Request, plan Plan) error {
switch plan.Action {
case ActionSkipSame, ActionSkipDestinationNewer:
return nil
case ActionPublishNew, ActionReplaceOlder:
case ActionPublishNew, ActionReplaceOlder, ActionForceReplace:
default:
return fmt.Errorf("cannot execute action %s: %s", plan.Action, plan.Reason)
}
@@ -30,6 +30,14 @@ func Execute(ctx context.Context, req Request, plan Plan) error {
return err
}
}
if plan.Action == ActionForceReplace {
if err := req.DestinationBackend.DeletePrefix(ctx, req.DestinationBundlePath, storage.DeleteOptions{IgnoreMissing: true, PruneEmptyDirs: true}); err != nil {
return err
}
if err := ensureDestinationEmpty(ctx, req.DestinationBackend, req.DestinationBundlePath); err != nil {
return err
}
}
writtenOutputs := make([]Output, 0, len(plan.Outputs))
cleanup := func() {

241
internal/publish/force_test.go Normal file
View File

@@ -0,0 +1,241 @@
package publish
import (
"context"
"encoding/json"
"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 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()
writeFakeFile(t, backend, "bundle/old.txt", "old")
},
transfer: defaultTransfer(),
wantReason: "fail_unmanaged",
forceAction: true,
},
{
name: "different source id",
prepare: func(t *testing.T, backend *fake.Backend, source bundle.Manifest) {
t.Helper()
conflict := source
conflict.ID = "other.source"
writeFakeDestinationState(t, backend, "bundle", conflict, testutil.DestinationStateOptions{})
},
transfer: conflictReplaceTransfer(),
wantReason: "requires --force",
forceAction: true,
},
{
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"}}})
writeFakeDestinationState(t, backend, "bundle", conflict, testutil.DestinationStateOptions{})
},
transfer: conflictReplaceTransfer(),
wantReason: "requires --force",
forceAction: true,
},
{
name: "pipeline mismatch",
prepare: func(t *testing.T, backend *fake.Backend, source bundle.Manifest) {
t.Helper()
writeFakeDestinationState(t, backend, "bundle", source, testutil.DestinationStateOptions{PipelineID: "other-pipeline"})
},
transfer: conflictReplaceTransfer(),
wantReason: "requires --force",
forceAction: true,
},
{
name: "destination mismatch",
prepare: func(t *testing.T, backend *fake.Backend, source bundle.Manifest) {
t.Helper()
writeFakeDestinationState(t, backend, "bundle", source, testutil.DestinationStateOptions{DestinationID: "other-destination"})
},
transfer: conflictReplaceTransfer(),
wantReason: "requires --force",
forceAction: true,
},
{
name: "newer destination",
prepare: func(t *testing.T, backend *fake.Backend, source bundle.Manifest) {
t.Helper()
newer := source
newer.Created = newer.Created.AddDate(0, 0, 1)
writeFakeDestinationState(t, backend, "bundle", newer, testutil.DestinationStateOptions{})
},
transfer: newerReplaceTransfer(),
wantReason: "requires --force",
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 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"
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()
writeFakeFile(t, destinationBackend, "bundle/old.txt", "old")
writeFakeFile(t, destinationBackend, "bundle/nested/old.txt", "old")
writeFakeFile(t, destinationBackend, "bundle-sibling/keep.txt", "keep")
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)
}
assertFakeFile(t, destinationBackend, "bundle/report.md", "# Report\nSunny.\n")
assertFakeMissing(t, destinationBackend, "bundle/old.txt")
assertFakeMissing(t, destinationBackend, "bundle/nested/old.txt")
assertFakeFile(t, destinationBackend, "bundle-sibling/keep.txt", "keep")
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},
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
}
func writeFakeDestinationState(t *testing.T, backend *fake.Backend, relative string, manifest bundle.Manifest, opts testutil.DestinationStateOptions) {
t.Helper()
destinationState := testutil.DestinationState(manifest, opts)
data, err := json.MarshalIndent(destinationState, "", " ")
if err != nil {
t.Fatalf("marshal destination state: %v", err)
}
statePath, err := storage.StatePath(relative)
if err != nil {
t.Fatalf("state path: %v", err)
}
writeFakeFile(t, backend, statePath, string(append(data, '\n')))
for _, output := range destinationState.Outputs {
path, err := storage.Join(relative, output.Path)
if err != nil {
t.Fatalf("join output path: %v", err)
}
writeFakeFile(t, backend, path, "old")
}
}
func writeFakeFile(t *testing.T, backend *fake.Backend, path, data string) {
t.Helper()
if _, err := backend.WriteFile(context.Background(), path, []byte(data), storage.WriteOptions{}); err != nil {
t.Fatalf("write fake file %s: %v", path, err)
}
}
func assertFakeFile(t *testing.T, backend *fake.Backend, path, want string) {
t.Helper()
data, err := backend.ReadFile(context.Background(), path)
if err != nil {
t.Fatalf("read fake file %s: %v", path, err)
}
if got := string(data); got != want {
t.Fatalf("fake file %s = %q, want %q", path, got, want)
}
}
func assertFakeMissing(t *testing.T, backend *fake.Backend, path string) {
t.Helper()
if _, err := backend.Stat(context.Background(), path); !storage.IsNotFound(err) {
t.Fatalf("fake file %s stat error = %v, want not found", path, err)
}
}

27
internal/publish/plan.go
View File

@@ -20,6 +20,7 @@ const (
ActionSkipDestinationNewer Action = "skip_destination_newer"
ActionFailConflict Action = "fail_conflict"
ActionFailUnmanaged Action = "fail_unmanaged"
ActionForceReplace Action = "force_replace"
)
type Request struct {
@@ -34,6 +35,7 @@ type Request struct {
Transformers TransformerResolver
Transfer config.TransferPolicy
DistributorVersion string
Force bool
}
type TransformerResolver interface {
@@ -48,6 +50,7 @@ type Plan struct {
DestinationBundlePath string
Action Action
Reason string
Force bool
Outputs []Output
ExistingState *state.DistributorState
}
@@ -75,7 +78,7 @@ func Build(ctx context.Context, req Request) (Plan, error) {
return Plan{}, err
}
comparison := state.Compare(req.SourceBundle.Manifest, req.PipelineID, req.DestinationID, status)
action, reason := actionForComparison(comparison, req.Transfer)
action, reason := actionForComparison(comparison, req.Transfer, req.Force)
plan := Plan{
PipelineID: req.PipelineID,
DestinationID: req.DestinationID,
@@ -84,6 +87,7 @@ func Build(ctx context.Context, req Request) (Plan, error) {
DestinationBundlePath: req.DestinationBundlePath,
Action: action,
Reason: reason,
Force: action == ActionForceReplace,
Outputs: outputs,
ExistingState: status.State,
}
@@ -112,13 +116,24 @@ func validateRequest(req Request) error {
return nil
}
func actionForComparison(comparison state.Comparison, transfer config.TransferPolicy) (Action, string) {
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, state.OutcomeIdentityMismatch, state.OutcomeSameCreatedConflict, state.OutcomeDifferentSourceConflict:
case state.OutcomeInvalidState:
return ActionFailConflict, comparison.Reason
case state.OutcomeIdentityMismatch, state.OutcomeSameCreatedConflict, state.OutcomeDifferentSourceConflict:
if transfer.OnConflict == config.TransferActionReplace {
if force {
return ActionForceReplace, "forced replacement of conflicting destination state: " + comparison.Reason
}
return ActionFailConflict, "destination conflict replacement requires --force"
}
return ActionFailConflict, comparison.Reason
case state.OutcomeSameSource:
if transfer.OnDestinationSame == config.TransferActionFail {
@@ -131,6 +146,12 @@ func actionForComparison(comparison state.Comparison, transfer config.TransferPo
}
return ActionReplaceOlder, comparison.Reason
case state.OutcomeDestinationNewer:
if transfer.OnDestinationNewer == config.TransferActionReplace {
if force {
return ActionForceReplace, "forced replacement of newer destination state"
}
return ActionFailConflict, "destination is newer and replacement requires --force"
}
if transfer.OnDestinationNewer == config.TransferActionFail {
return ActionFailConflict, "destination is newer and transfer policy requires failure"
}

1
internal/storage/backend.go
View File

@@ -30,6 +30,7 @@ type Backend interface {
Walk(ctx context.Context, prefix string, opts WalkOptions, fn WalkFunc) error
HasAny(ctx context.Context, prefix string) (bool, error)
DeleteManagedBundle(ctx context.Context, bundlePath string, managedOutputPaths []string, opts DeleteOptions) error
DeletePrefix(ctx context.Context, prefix string, opts DeleteOptions) error
}
type WalkOptions struct {

1
internal/storage/errors.go
View File

@@ -29,6 +29,7 @@ const (
OpWalk = "walk"
OpHasAny = "has any"
OpDeleteManagedBundle = "delete managed bundle"
OpDeletePrefix = "delete prefix"
OpRegisterBackend = "register backend"
OpOpenBackend = "open backend"
)

35
internal/storage/fake/backend.go
View File

@@ -205,6 +205,41 @@ func (b *Backend) DeleteManagedBundle(ctx context.Context, bundlePath string, ma
return nil
}
func (b *Backend) DeletePrefix(ctx context.Context, prefix string, opts storage.DeleteOptions) error {
if err := ctx.Err(); err != nil {
return err
}
if err := storage.ValidatePrefix(prefix); err != nil {
return err
}
if prefix != "" && !b.exists(prefix) && !b.hasChild(prefix) {
if opts.IgnoreMissing {
return nil
}
return storage.NewError(storage.OpDeletePrefix, backendName, prefix, storage.ErrNotFound, nil)
}
for path := range b.files {
if path == prefix || entryBelow(prefix, path) {
delete(b.files, path)
}
}
for path := range b.symlinks {
if path == prefix || entryBelow(prefix, path) {
delete(b.symlinks, path)
}
}
for path := range b.dirs {
if path != "" && (path == prefix || entryBelow(prefix, path)) {
delete(b.dirs, path)
}
}
if opts.PruneEmptyDirs {
b.pruneEmptyParents(parentOf(prefix))
}
b.dirs[""] = struct{}{}
return nil
}
func (b *Backend) ensureParents(path string) {
parent := parentOf(path)
for parent != "" {

24
internal/storage/fake/backend_test.go
View File

@@ -140,6 +140,30 @@ func TestBackendManagedDeletion(t *testing.T) {
}
}
func TestBackendDeletePrefixStaysWithinPrefix(t *testing.T) {
backend := New()
mustWrite(t, backend, "bundle/report.md", "report")
mustWrite(t, backend, "bundle/nested/old.txt", "old")
mustWrite(t, backend, "bundle-sibling/keep.txt", "keep")
mustWrite(t, backend, "outside.txt", "outside")
if err := backend.DeletePrefix(context.Background(), "bundle", storage.DeleteOptions{IgnoreMissing: true, PruneEmptyDirs: true}); err != nil {
t.Fatalf("DeletePrefix() error = %v", err)
}
if _, err := backend.Stat(context.Background(), "bundle/report.md"); !storage.IsNotFound(err) {
t.Fatalf("deleted file stat error = %v, want not found", err)
}
if _, err := backend.Stat(context.Background(), "bundle/nested/old.txt"); !storage.IsNotFound(err) {
t.Fatalf("deleted nested file stat error = %v, want not found", err)
}
if _, err := backend.Stat(context.Background(), "bundle-sibling/keep.txt"); err != nil {
t.Fatalf("sibling stat error = %v", err)
}
if _, err := backend.Stat(context.Background(), "outside.txt"); err != nil {
t.Fatalf("outside stat error = %v", err)
}
}
func TestBackendHasAnyAndWalkStop(t *testing.T) {
backend := New()
found, err := backend.HasAny(context.Background(), "missing")

4
internal/storage/path_test.go
View File

@@ -206,3 +206,7 @@ func (b walkBackend) HasAny(context.Context, string) (bool, error) {
func (b walkBackend) DeleteManagedBundle(context.Context, string, []string, DeleteOptions) error {
return nil
}
func (b walkBackend) DeletePrefix(context.Context, string, DeleteOptions) error {
return nil
}