download: dont cleanup directory after download failed

DownloadDir has the option to ignore paths that we have already on disk.
In theory that makes downloads resumable, but in practice this doesnt
work because we delete everything if one download fails. This means if
one chunk in 1TiB block fails to download for whatever reason we start
all over.

Signed-off-by: Michael Hoffmann <[email protected]>

Author: MichaHoffmann

objstore.go

@@ -14,7 +14,6 @@ import (
 	"path/filepath"
 	"slices"
 	"strings"
-	"sync"
 	"time"
 
 	"github.com/efficientgo/core/logerrcapture"
@@ -442,7 +441,11 @@ func DownloadFile(ctx context.Context, logger log.Logger, bkt BucketReader, src,
 }
 
 // DownloadDir downloads all object found in the directory into the local directory.
-func DownloadDir(ctx context.Context, logger log.Logger, bkt BucketReader, originalSrc, src, dst string, options ...DownloadOption) error {
+func DownloadDir(ctx context.Context, logger log.Logger, bkt BucketReader, src, dst string, options ...DownloadOption) error {
+	return downloadDir(ctx, logger, bkt, src, src, dst, options...)
+}
+
+func downloadDir(ctx context.Context, logger log.Logger, bkt BucketReader, originalSrc, src, dst string, options ...DownloadOption) error {
 	if err := os.MkdirAll(dst, 0750); err != nil {
 		return errors.Wrap(err, "create dir")
 	}
@@ -453,19 +456,13 @@ func DownloadDir(ctx context.Context, logger log.Logger, bkt BucketReader, origi
 	g, ctx := errgroup.WithContext(ctx)
 	g.SetLimit(opts.concurrency)
 
-	var downloadedFiles []string
-	var m sync.Mutex
-
 	err := bkt.Iter(ctx, src, func(name string) error {
 		g.Go(func() error {
 			dst := filepath.Join(dst, filepath.Base(name))
 			if strings.HasSuffix(name, DirDelim) {
-				if err := DownloadDir(ctx, logger, bkt, originalSrc, name, dst, options...); err != nil {
+				if err := downloadDir(ctx, logger, bkt, originalSrc, name, dst, options...); err != nil {
 					return err
 				}
-				m.Lock()
-				downloadedFiles = append(downloadedFiles, dst)
-				m.Unlock()
 				return nil
 			}
 			for _, ignoredPath := range opts.ignoredPaths {
@@ -474,34 +471,14 @@ func DownloadDir(ctx context.Context, logger log.Logger, bkt BucketReader, origi
 					return nil
 				}
 			}
-			if err := DownloadFile(ctx, logger, bkt, name, dst); err != nil {
-				return err
-			}
-
-			m.Lock()
-			downloadedFiles = append(downloadedFiles, dst)
-			m.Unlock()
-			return nil
+			return DownloadFile(ctx, logger, bkt, name, dst)
 		})
 		return nil
 	})
-
-	if err == nil {
-		err = g.Wait()
-	}
-
 	if err != nil {
-		downloadedFiles = append(downloadedFiles, dst) // Last, clean up the root dst directory.
-		// Best-effort cleanup if the download failed.
-		for _, f := range downloadedFiles {
-			if rerr := os.RemoveAll(f); rerr != nil {
-				level.Warn(logger).Log("msg", "failed to remove file on partial dir download error", "file", f, "err", rerr)
-			}
-		}
 		return err
 	}
-
-	return nil
+	return g.Wait()
 }
 
 // IsOpFailureExpectedFunc allows to mark certain errors as expected, so they will not increment objstore_bucket_operation_failures_total metric.

objstore_test.go

@@ -305,7 +305,7 @@ func TestDownloadUploadDirConcurrency(t *testing.T) {
         objstore_bucket_operations_total{bucket="",operation="upload"} 3
 		`), `objstore_bucket_operations_total`))
 
-	testutil.Ok(t, DownloadDir(context.Background(), log.NewNopLogger(), m, "dir/", "dir/", tempDir, WithFetchConcurrency(10)))
+	testutil.Ok(t, DownloadDir(context.Background(), log.NewNopLogger(), m, "dir/", tempDir, WithFetchConcurrency(10)))
 	i, err := os.ReadDir(tempDir)
 	testutil.Ok(t, err)
 	testutil.Assert(t, len(i) == 3)
@@ -519,39 +519,6 @@ func TestTimingReader_ShouldCorrectlyWrapFile(t *testing.T) {
 	testutil.Assert(t, isReaderAt)
 }
 
-func TestDownloadDir_CleanUp(t *testing.T) {
-	b := unreliableBucket{
-		Bucket:  NewInMemBucket(),
-		n:       3,
-		current: atomic.NewInt32(0),
-	}
-	tempDir := t.TempDir()
-
-	testutil.Ok(t, b.Upload(context.Background(), "dir/obj1", bytes.NewReader([]byte("1"))))
-	testutil.Ok(t, b.Upload(context.Background(), "dir/obj2", bytes.NewReader([]byte("2"))))
-	testutil.Ok(t, b.Upload(context.Background(), "dir/obj3", bytes.NewReader([]byte("3"))))
-
-	// We exapect the third Get to fail
-	testutil.NotOk(t, DownloadDir(context.Background(), log.NewNopLogger(), b, "dir/", "dir/", tempDir))
-	_, err := os.Stat(tempDir)
-	testutil.Assert(t, os.IsNotExist(err))
-}
-
-// unreliableBucket implements Bucket and returns an error on every n-th Get.
-type unreliableBucket struct {
-	Bucket
-
-	n       int32
-	current *atomic.Int32
-}
-
-func (b unreliableBucket) Get(ctx context.Context, name string) (io.ReadCloser, error) {
-	if b.current.Inc()%b.n == 0 {
-		return nil, errors.Errorf("some error message")
-	}
-	return b.Bucket.Get(ctx, name)
-}
-
 // mockReader implements io.ReadCloser and allows to mock the functions.
 type mockReader struct {
 	io.Reader