From 57063cca2eccc571900eb624f9cb9ebaceaa9ad0 Mon Sep 17 00:00:00 2001 From: Terry Rankine Date: Thu, 17 Sep 2026 21:04:03 +0200 Subject: [PATCH] fix: replace os.Exit and panic in cp and sync with error handling When a worker hit "too many open files", cp and sync printed the fd limit hint and called os.Exit(1). That skipped every deferred cleanup: the progress bar was never finished, the log writer was never flushed, and the --stat summary was never printed. cp also panicked on an "unexpected src-dst pair", which the command validator already rejects, so a normal error is enough. sync additionally created its cancel func without ever calling it, and its planRun goroutines ranged over the object channels with no way to stop on cancellation. cp: on "too many open files", print the hint, record the error, cancel the context so in-flight and queued objects stop, and let Run return through the normal path. Cancellation errors from the remaining tasks are already filtered out by printError, and the process still exits non-zero because the error is returned. The unexpected src-dst pair is now reported like any other per-object error. sync: add the missing defer cancel(). Drop the waiter goroutine that held the other os.Exit; no task was ever submitted to that waiter, so the branch was unreachable (the generated cp commands handle the fd limit themselves through Copy.Run). planRun goroutines now select on ctx.Done() so they return promptly on cancel, and Run closes its side of the command pipe on return so a goroutine blocked writing a command is unblocked and can exit too. Extsort error channels are drained in the listing goroutines instead of in separate goroutines; extsort closes them together with the output channel, so this cannot block. extsort sends a nil error when the context is cancelled mid-merge, which previously made printError dereference nil, so nil is skipped. shouldStopSync now also stops on RequestError and SerializationError, since continuing to list after a transport failure only produces a misleading partial sync. Co-Authored-By: Claude Opus 5 (1M context) --- command/cp.go | 15 +++- command/sync.go | 139 +++++++++++++++++------------- command/sync_test.go | 198 +++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 288 insertions(+), 64 deletions(-) create mode 100644 command/sync_test.go diff --git a/command/cp.go b/command/cp.go index 7ed06bfee..d2e5c59a8 100644 --- a/command/cp.go +++ b/command/cp.go @@ -419,6 +419,9 @@ increase the open file limit or try to decrease the number of workers with // Run starts copying given source objects to destination. func (c Copy) Run(ctx context.Context) error { + ctx, cancel := context.WithCancel(ctx) + defer cancel() + // override source region if set if c.srcRegion != "" { c.storageOpts.SetRegion(c.srcRegion) @@ -451,9 +454,10 @@ func (c Copy) Run(ctx context.Context) error { for err := range waiter.Err() { if strings.Contains(err.Error(), "too many open files") { fmt.Println(strings.TrimSpace(fdlimitWarning)) - fmt.Printf("ERROR %v\n", err) - - os.Exit(1) + printError(c.fullCommand, c.op, err) + merrorWaiter = multierror.Append(merrorWaiter, err) + cancel() + continue } printError(c.fullCommand, c.op, err) merrorWaiter = multierror.Append(merrorWaiter, err) @@ -558,7 +562,10 @@ func (c Copy) Run(ctx context.Context) error { } task = c.prepareUploadTask(ctx, srcurl, c.dst, isBatch, c.metadata) default: - panic("unexpected src-dst pair") + err := fmt.Errorf("unexpected src-dst pair: %v -> %v", srcurl, c.dst) + merrorObjects = multierror.Append(merrorObjects, err) + printError(c.fullCommand, c.op, err) + continue } parallel.Run(task, waiter) } diff --git a/command/sync.go b/command/sync.go index 6dede5a66..37bbff737 100644 --- a/command/sync.go +++ b/command/sync.go @@ -4,20 +4,17 @@ import ( "context" "fmt" "io" - "os" "path/filepath" "strings" "sync" "github.com/aws/aws-sdk-go/aws/awserr" - "github.com/hashicorp/go-multierror" "github.com/lanrat/extsort" "github.com/urfave/cli/v2" errorpkg "github.com/peak/s5cmd/v2/error" "github.com/peak/s5cmd/v2/log" "github.com/peak/s5cmd/v2/log/stat" - "github.com/peak/s5cmd/v2/parallel" "github.com/peak/s5cmd/v2/storage" "github.com/peak/s5cmd/v2/storage/url" ) @@ -181,6 +178,7 @@ func (s Sync) Run(c *cli.Context) error { } ctx, cancel := context.WithCancel(c.Context) + defer cancel() sourceObjects, destObjects, err := s.getSourceAndDestinationObjects(ctx, cancel, srcurl, dsturl) if err != nil { @@ -208,34 +206,18 @@ func (s Sync) Run(c *cli.Context) error { sourceObjects = nil destObjects = nil - waiter := parallel.NewWaiter() - var ( - merrorWaiter error - errDoneCh = make(chan struct{}) - ) - - go func() { - defer close(errDoneCh) - for err := range waiter.Err() { - if strings.Contains(err.Error(), "too many open files") { - fmt.Println(strings.TrimSpace(fdlimitWarning)) - fmt.Printf("ERROR %v\n", err) - - os.Exit(1) - } - printError(s.fullCommand, s.op, err) - merrorWaiter = multierror.Append(merrorWaiter, err) - } - }() - strategy := NewStrategy(s.sizeOnly) // create comparison strategy. pipeReader, pipeWriter := io.Pipe() // create a reader, writer pipe to pass commands to run + // If Run stops consuming early (e.g. the context is cancelled), closing the + // reader unblocks any planRun goroutine stuck writing to the pipe so it can + // observe the cancellation and return. + defer pipeReader.Close() + // Create commands in background. - go s.planRun(c, onlySource, onlyDest, commonObjects, dsturl, strategy, pipeWriter, isBatch) + go s.planRun(ctx, c, onlySource, onlyDest, commonObjects, dsturl, strategy, pipeWriter, isBatch) - err = NewRun(c, pipeReader).Run(ctx) - return multierror.Append(err, merrorWaiter).ErrorOrNil() + return NewRun(c, pipeReader).Run(ctx) } // compareObjects compares source and destination objects. It assumes that @@ -377,12 +359,14 @@ func (s Sync) getSourceAndDestinationObjects(ctx context.Context, cancel context sourceObjects <- &o } - // read and print the external sort errors - go func() { - for err := range srcErrCh { + // read and print the external sort errors. extsort closes the error + // channel together with the output channel, so this does not block. + // It may send a nil error when the context is cancelled mid-merge. + for err := range srcErrCh { + if err != nil { printError(s.fullCommand, s.op, err) } - }() + } }() // get destination objects. @@ -424,12 +408,14 @@ func (s Sync) getSourceAndDestinationObjects(ctx context.Context, cancel context destObjects <- &o } - // read and print the external sort errors - go func() { - for err := range dstErrCh { + // read and print the external sort errors. extsort closes the error + // channel together with the output channel, so this does not block. + // It may send a nil error when the context is cancelled mid-merge. + for err := range dstErrCh { + if err != nil { printError(s.fullCommand, s.op, err) } - }() + } }() return sourceObjects, destObjects, nil @@ -437,6 +423,7 @@ func (s Sync) getSourceAndDestinationObjects(ctx context.Context, cancel context // planRun prepares the commands and writes them to writer 'w'. func (s Sync) planRun( + ctx context.Context, c *cli.Context, onlySource, onlyDest chan *url.URL, common chan *ObjectPair, @@ -462,14 +449,22 @@ func (s Sync) planRun( wg.Add(1) go func() { defer wg.Done() - for srcurl := range onlySource { - curDestURL := generateDestinationURL(srcurl, dsturl, isBatch) - command, err := generateCommand(c, "cp", defaultFlags, srcurl, curDestURL) - if err != nil { - printDebug(s.op, err, srcurl, curDestURL) - continue + for { + select { + case <-ctx.Done(): + return + case srcurl, ok := <-onlySource: + if !ok { + return + } + curDestURL := generateDestinationURL(srcurl, dsturl, isBatch) + command, err := generateCommand(c, "cp", defaultFlags, srcurl, curDestURL) + if err != nil { + printDebug(s.op, err, srcurl, curDestURL) + continue + } + fmt.Fprintln(w, command) } - fmt.Fprintln(w, command) } }() @@ -477,21 +472,29 @@ func (s Sync) planRun( wg.Add(1) go func() { defer wg.Done() - for commonObject := range common { - sourceObject, destObject := commonObject.src, commonObject.dst - curSourceURL, curDestURL := sourceObject.URL, destObject.URL - err := strategy.ShouldSync(sourceObject, destObject) // check if object should be copied. - if err != nil { - printDebug(s.op, err, curSourceURL, curDestURL) - continue - } + for { + select { + case <-ctx.Done(): + return + case commonObject, ok := <-common: + if !ok { + return + } + sourceObject, destObject := commonObject.src, commonObject.dst + curSourceURL, curDestURL := sourceObject.URL, destObject.URL + err := strategy.ShouldSync(sourceObject, destObject) // check if object should be copied. + if err != nil { + printDebug(s.op, err, curSourceURL, curDestURL) + continue + } - command, err := generateCommand(c, "cp", defaultFlags, curSourceURL, curDestURL) - if err != nil { - printDebug(s.op, err, curSourceURL, curDestURL) - continue + command, err := generateCommand(c, "cp", defaultFlags, curSourceURL, curDestURL) + if err != nil { + printDebug(s.op, err, curSourceURL, curDestURL) + continue + } + fmt.Fprintln(w, command) } - fmt.Fprintln(w, command) } }() @@ -504,8 +507,17 @@ func (s Sync) planRun( // or rewrite generateCommand function? dstURLs := make([]*url.URL, 0, extsortChunkSize) - for d := range onlyDest { - dstURLs = append(dstURLs, d) + collect: + for { + select { + case <-ctx.Done(): + return + case d, ok := <-onlyDest: + if !ok { + break collect + } + dstURLs = append(dstURLs, d) + } } if len(dstURLs) == 0 { @@ -519,10 +531,17 @@ func (s Sync) planRun( } fmt.Fprintln(w, command) } else { - // we only need to consume them from the channel so that rest of the objects + // we only need to consume them from the channel so that rest of the objects // can be sent to channel. - for d := range onlyDest { - _ = d + for { + select { + case <-ctx.Done(): + return + case _, ok := <-onlyDest: + if !ok { + return + } + } } } }() @@ -594,7 +613,7 @@ func (s Sync) shouldStopSync(err error) bool { } if awsErr, ok := err.(awserr.Error); ok { switch awsErr.Code() { - case "AccessDenied", "NoSuchBucket": + case "AccessDenied", "NoSuchBucket", "RequestError", "SerializationError": return true } } diff --git a/command/sync_test.go b/command/sync_test.go new file mode 100644 index 000000000..df7ac1baa --- /dev/null +++ b/command/sync_test.go @@ -0,0 +1,198 @@ +package command + +import ( + "context" + "fmt" + "io" + "sort" + "strings" + "testing" + "time" + + "github.com/aws/aws-sdk-go/aws/awserr" + "github.com/google/go-cmp/cmp" + "github.com/urfave/cli/v2" + + "github.com/peak/s5cmd/v2/log" + "github.com/peak/s5cmd/v2/storage" + "github.com/peak/s5cmd/v2/storage/url" +) + +func TestShouldStopSync(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + err error + exitOnError bool + want bool + }{ + { + name: "AccessDenied", + err: awserr.New("AccessDenied", "access denied", nil), + want: true, + }, + { + name: "NoSuchBucket", + err: awserr.New("NoSuchBucket", "no such bucket", nil), + want: true, + }, + { + name: "RequestError", + err: awserr.New("RequestError", "request error", nil), + want: true, + }, + { + name: "SerializationError", + err: awserr.New("SerializationError", "serialization error", nil), + want: true, + }, + { + name: "SlowDown should not stop", + err: awserr.New("SlowDown", "slow down", nil), + want: false, + }, + { + name: "non-AWS error with exitOnError true", + err: fmt.Errorf("some random error"), + exitOnError: true, + want: true, + }, + { + name: "non-AWS error with exitOnError false", + err: fmt.Errorf("some random error"), + exitOnError: false, + want: false, + }, + { + name: "nil error", + err: nil, + want: false, + }, + } + + for _, tc := range tests { + tc := tc + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + s := Sync{ + exitOnError: tc.exitOnError, + } + + got := s.shouldStopSync(tc.err) + if got != tc.want { + t.Errorf("shouldStopSync(%v) = %v, want %v", tc.err, got, tc.want) + } + }) + } +} + +// readPlan runs planRun in the background and returns everything it wrote +// before closing the writer. It fails the test if the writer is never closed. +func readPlan(t *testing.T, ctx context.Context, s Sync, c *cli.Context, onlySource, onlyDest chan *url.URL, common chan *ObjectPair, isBatch bool) []string { + t.Helper() + + dsturl := mustNewURL(t, "s3://bucket/dst/") + pr, pw := io.Pipe() + + go s.planRun(ctx, c, onlySource, onlyDest, common, dsturl, NewStrategy(false), pw, isBatch) + + type result struct { + out []byte + err error + } + resultCh := make(chan result, 1) + go func() { + out, err := io.ReadAll(pr) + resultCh <- result{out, err} + }() + + select { + case r := <-resultCh: + if r.err != nil { + t.Fatalf("unexpected read error: %v", r.err) + } + lines := strings.Split(strings.TrimSpace(string(r.out)), "\n") + if len(lines) == 1 && lines[0] == "" { + return nil + } + sort.Strings(lines) + return lines + case <-time.After(10 * time.Second): + t.Fatal("planRun did not close the writer") + return nil + } +} + +func TestPlanRunReturnsOnCancelledContext(t *testing.T) { + t.Parallel() + + for _, del := range []bool{false, true} { + del := del + t.Run(fmt.Sprintf("delete=%v", del), func(t *testing.T) { + t.Parallel() + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + // Nothing is ever sent on (or closes) these channels. Without the + // cancellation checks planRun would block on them forever. + onlySource := make(chan *url.URL) + onlyDest := make(chan *url.URL) + common := make(chan *ObjectPair) + + got := readPlan(t, ctx, Sync{delete: del}, nil, onlySource, onlyDest, common, true) + if len(got) != 0 { + t.Errorf("expected no commands, got %v", got) + } + }) + } +} + +func TestPlanRunGeneratesCommands(t *testing.T) { + t.Parallel() + + // planRun logs skipped objects through the global logger. + log.Init("error", false) + + app := cli.NewApp() + c := cli.NewContext(app, flagSet(t, "sync", nil), nil) + + now := time.Now() + older := now.Add(-time.Hour) + + onlySource := make(chan *url.URL, 1) + onlyDest := make(chan *url.URL, 2) + common := make(chan *ObjectPair, 2) + + onlySource <- mustNewURL(t, "s3://bucket/src/new") + close(onlySource) + + onlyDest <- mustNewURL(t, "s3://bucket/dst/stale1") + onlyDest <- mustNewURL(t, "s3://bucket/dst/stale2") + close(onlyDest) + + // source is newer and differs in size: should be copied. + common <- &ObjectPair{ + src: &storage.Object{URL: mustNewURL(t, "s3://bucket/src/changed"), Size: 2, ModTime: &now}, + dst: &storage.Object{URL: mustNewURL(t, "s3://bucket/dst/changed"), Size: 1, ModTime: &older}, + } + // identical: should be skipped. + common <- &ObjectPair{ + src: &storage.Object{URL: mustNewURL(t, "s3://bucket/src/same"), Size: 1, ModTime: &now}, + dst: &storage.Object{URL: mustNewURL(t, "s3://bucket/dst/same"), Size: 1, ModTime: &now}, + } + close(common) + + got := readPlan(t, context.Background(), Sync{delete: true}, c, onlySource, onlyDest, common, false) + + want := []string{ + `cp --raw='true' "s3://bucket/src/changed" "s3://bucket/dst/changed"`, + `cp --raw='true' "s3://bucket/src/new" "s3://bucket/dst/new"`, + `rm --raw='true' "s3://bucket/dst/stale1" "s3://bucket/dst/stale2"`, + } + if diff := cmp.Diff(want, got); diff != "" { + t.Errorf("(-want +got):\n%v", diff) + } +}