Repository: lima-vm/lima
Pull request #2728 (2024-10-13 10:43:15)

Test parallel downloads
Add sub tests for parallel downloads with "with cache" and "caching-only
mode". The test run 20 concurrent downloads with the current test http
server and ensure that downloads succeeded.

To make it possible to test using the same process, we include now a
counter in the perProcessTempfile. The file name is now:

    {filename}.pid.{count}

The test revealed an issue with clonefile, not seen when testing 3
parallel `limactl create`. When the data file is replaced during a
clonefile syscall, it may fail with:

    clonefile failed: no such file or directory

This smells like a darwin bug, but we can avoid this issue by using the 
first download result. When a download finishes, we check if data file
exists, and return success if it does. We take a lock for the very
short time needed to check and rename the temporary data file to the 
target file.

The parallel tests failed on windows - it seems that os.Rename() does
not work on windows if the target exist. Replacing atomicWrite with a
simpler version that takes a lock and write the file fixes the issue.

Tested using:

    % go test ./pkg/downloader -run 'TestDownloadRemote/with_cache/parallel' -count=1000
    ok          github.com/lima-vm/lima/pkg/downloader  116.025s

    % go test ./pkg/downloader -run 'TestDownloadRemote/caching-only_mode/parallel' -count=1000
    ok          github.com/lima-vm/lima/pkg/downloader  98.535s


nirs: The windows tests fail, probably already broken since #2723:

```
downloader_test.go:110: Download failed: rename
C:\Users\RUNNER~1\AppData\Local\Temp\TestDownloadRemotewith_cacheparallel3648796552\001\cache\download\by-url-sha256\234be6a0193368c635111cf397120fb9724641a6641a1adb8b16ad5b74e951f3\url.tmp.1624.11
C:\Users\RUNNER~1\AppData\Local\Temp\TestDownloadRemotewith_cacheparallel3648796552\001\cache\download\by-url-sha256\234be6a0193368c635111cf397120fb9724641a6641a1adb8b16ad5b74e951f3\url:
Access is denied.
```

`os.Rename()` calls `MoveFileEx(from, to, MOVEFILE_REPLACE_EXISTING)`
https://github.com/golang/go/blob/b521ebb55a9b26c8824b219376c7f91f7cda6ec2/src/internal/syscall/windows/syscall_windows.go#L360 

Maybe it does not work when another process is renaming the same file?
nirs: The fix works for windows for the metadata files, but renaming the data file fails:

```
=== RUN   TestDownloadRemote/with_cache/parallel
    downloader_test.go:110: Download failed: rename C:\Users\RUNNER~1\AppData\Local\Temp\TestDownloadRemotewith_cacheparallel26671273\001\cache\download\by-url-sha256\6c11daee298801cf583ce308783f6d28f0ddfce3214064d55cc45217174035fc\data.tmp.6508.6 C:\Users\RUNNER~1\AppData\Local\Temp\TestDownloadRemotewith_cacheparallel26671273\001\cache\download\by-url-sha256\6c11daee298801cf583ce308783f6d28f0ddfce3214064d55cc45217174035fc\data: Access is denied.
=== RUN   TestDownloadRemote/caching-only_mode
=== RUN   TestDownloadRemote/caching-only_mode/serial
=== RUN   TestDownloadRemote/caching-only_mode/parallel
    downloader_test.go:153: Download failed: rename C:\Users\RUNNER~1\AppData\Local\Temp\TestDownloadRemotecaching-only_modeparallel2628695855\001\cache\download\by-url-sha256\6c11daee298801cf583ce308783f6d28f0ddfce3214064d55cc45217174035fc\data.tmp.6508.27 C:\Users\RUNNER~1\AppData\Local\Temp\TestDownloadRemotecaching-only_modeparallel2628695855\001\cache\download\by-url-sha256\6c11daee298801cf583ce308783f6d28f0ddfce3214064d55cc45217174035fc\data: Access is denied.
```

We know that the "data" does not exist, not sure why os.Rename() fails.
nirs: Adding logs:

```
diff --git a/pkg/downloader/downloader.go b/pkg/downloader/downloader.go
index 2611740..ea7e648 100644
--- a/pkg/downloader/downloader.go
+++ b/pkg/downloader/downloader.go
@@ -679,7 +679,9 @@ func downloadHTTP(ctx context.Context, localPath, lastModified, contentType, url
        // data, and when we return data file exists.
 
        return lockutil.WithDirLock(filepath.Dir(localPath), func() error {
-               if _, err := os.Stat(localPath); err == nil {
+               fi, err := os.Stat(localPath)
+               logrus.Infof("os.Stat('%s') -> %+v, %+v", localPath, fi, err)
+               if err == nil {
                        return nil
                } else if !errors.Is(err, os.ErrNotExist) {
                        return err
```

1. First thread did not find the file - expected

```
=== RUN   TestDownloadRemote/with_cache/parallel
time="2024-10-13T13:41:07Z" level=info msg="os.Stat('C:\\Users\\RUNNER~1\\AppData\\Local\\Temp\\TestDownloadRemotewith_cacheparallel1978374964\\001\\cache\\download\\by-url-sha256\\28be12e26c23dd7b394a4970c0da2ae02769dd834c338a550d7a6264536c0440\\data') -> <nil>, CreateFile C:\\Users\\RUNNER~1\\AppData\\Local\\Temp\\TestDownloadRemotewith_cacheparallel1978374964\\001\\cache\\download\\by-url-sha256\\28be12e26c23dd7b394a4970c0da2ae02769dd834c338a550d7a6264536c0440\\data: The system cannot find the file specified."
```

2. Next 4 thread did not find the file?!
   - should be impossible: the lock should prevent these threads from checking the file until the first thread rename the temporary file and released the lock.

```
time="2024-10-13T13:41:07Z" level=info msg="os.Stat('C:\\Users\\RUNNER~1\\AppData\\Local\\Temp\\TestDownloadRemotewith_cacheparallel1978374964\\001\\cache\\download\\by-url-sha256\\28be12e26c23dd7b394a4970c0da2ae02769dd834c338a550d7a6264536c0440\\data') -> <nil>, CreateFile C:\\Users\\RUNNER~1\\AppData\\Local\\Temp\\TestDownloadRemotewith_cacheparallel1978374964\\001\\cache\\download\\by-url-sha256\\28be12e26c23dd7b394a4970c0da2ae02769dd834c338a550d7a6264536c0440\\data: The system cannot find the file specified."
```

```
time="2024-10-13T13:41:07Z" level=info msg="os.Stat('C:\\Users\\RUNNER~1\\AppData\\Local\\Temp\\TestDownloadRemotewith_cacheparallel1978374964\\001\\cache\\download\\by-url-sha256\\28be12e26c23dd7b394a4970c0da2ae02769dd834c338a550d7a6264536c0440\\data') -> <nil>, CreateFile C:\\Users\\RUNNER~1\\AppData\\Local\\Temp\\TestDownloadRemotewith_cacheparallel1978374964\\001\\cache\\download\\by-url-sha256\\28be12e26c23dd7b394a4970c0da2ae02769dd834c338a550d7a6264536c0440\\data: The system cannot find the file specified."
```

```
time="2024-10-13T13:41:07Z" level=info msg="os.Stat('C:\\Users\\RUNNER~1\\AppData\\Local\\Temp\\TestDownloadRemotewith_cacheparallel1978374964\\001\\cache\\download\\by-url-sha256\\28be12e26c23dd7b394a4970c0da2ae02769dd834c338a550d7a6264536c0440\\data') -> <nil>, CreateFile C:\\Users\\RUNNER~1\\AppData\\Local\\Temp\\TestDownloadRemotewith_cacheparallel1978374964\\001\\cache\\download\\by-url-sha256\\28be12e26c23dd7b394a4970c0da2ae02769dd834c338a550d7a6264536c0440\\data: The system cannot find the file specified."
```

```
time="2024-10-13T13:41:07Z" level=info msg="os.Stat('C:\\Users\\RUNNER~1\\AppData\\Local\\Temp\\TestDownloadRemotewith_cacheparallel1978374964\\001\\cache\\download\\by-url-sha256\\28be12e26c23dd7b394a4970c0da2ae02769dd834c338a550d7a6264536c0440\\data') -> <nil>, CreateFile C:\\Users\\RUNNER~1\\AppData\\Local\\Temp\\TestDownloadRemotewith_cacheparallel1978374964\\001\\cache\\download\\by-url-sha256\\28be12e26c23dd7b394a4970c0da2ae02769dd834c338a550d7a6264536c0440\\data: The system cannot find the file specified."
```

5. From this point, all threads find the file and return - expected

```
time="2024-10-13T13:41:07Z" level=info msg="os.Stat('C:\\Users\\RUNNER~1\\AppData\\Local\\Temp\\TestDownloadRemotewith_cacheparallel1978374964\\001\\cache\\download\\by-url-sha256\\28be12e26c23dd7b394a4970c0da2ae02769dd834c338a550d7a6264536c0440\\data') -> &{name:data FileAttributes:32 CreationTime:{LowDateTime:2384380421 HighDateTime:31137141} LastAccessTime:{LowDateTime:2384386325 HighDateTime:31137141} LastWriteTime:{LowDateTime:2384386325 HighDateTime:31137141} FileSizeHigh:0 FileSizeLow:49 ReparseTag:0 filetype:0 Mutex:{state:0 sema:0} path:C:\\Users\\RUNNER~1\\AppData\\Local\\Temp\\TestDownloadRemotewith_cacheparallel1978374964\\001\\cache\\download\\by-url-sha256\\28be12e26c23dd7b394a4970c0da2ae02769dd834c338a550d7a6264536c0440\\data vol:0 idxhi:0 idxlo:0 appendNameToPath:false}, <nil>"
```

6. 3 thread failed to rename - this means the lock is broken - only one thread should be able to rename.

```
    downloader_test.go:110: Download failed: rename C:\Users\RUNNER~1\AppData\Local\Temp\TestDownloadRemotewith_cacheparallel1978374964\001\cache\download\by-url-sha256\28be12e26c23dd7b394a4970c0da2ae02769dd834c338a550d7a6264536c0440\data.tmp.908.6 C:\Users\RUNNER~1\AppData\Local\Temp\TestDownloadRemotewith_cacheparallel1978374964\001\cache\download\by-url-sha256\28be12e26c23dd7b394a4970c0da2ae02769dd834c338a550d7a6264536c0440\data: Access is denied.
    downloader_test.go:110: Download failed: rename C:\Users\RUNNER~1\AppData\Local\Temp\TestDownloadRemotewith_cacheparallel1978374964\001\cache\download\by-url-sha256\28be12e26c23dd7b394a4970c0da2ae02769dd834c338a550d7a6264536c0440\data.tmp.908.7 C:\Users\RUNNER~1\AppData\Local\Temp\TestDownloadRemotewith_cacheparallel1978374964\001\cache\download\by-url-sha256\28be12e26c23dd7b394a4970c0da2ae02769dd834c338a550d7a6264536c0440\data: Access is denied.
    downloader_test.go:110: Download failed: rename C:\Users\RUNNER~1\AppData\Local\Temp\TestDownloadRemotewith_cacheparallel1978374964\001\cache\download\by-url-sha256\28be12e26c23dd7b394a4970c0da2ae02769dd834c338a550d7a6264536c0440\data.tmp.908.8 C:\Users\RUNNER~1\AppData\Local\Temp\TestDownloadRemotewith_cacheparallel1978374964\001\cache\download\by-url-sha256\28be12e26c23dd7b394a4970c0da2ae02769dd834c338a550d7a6264536c0440\data: Access is denied.
```

@AkihiroSuda looks like the windows lock is broken. We don't have tests for the lockutil so this is expected :-)
nirs: Or maybe on windows the lock works only for different processes?

We used the same process to simplify the test, but maybe we should test separate processes. We can create a minimal executable that run fileutils.Download(), or maybe create a hidden `limactl download` command and test it?
nirs: The windows tests pass now. I will split the lockutil fix to another pr, it is much worse bug than the rare failure for parallel downloads.
nirs: Added separate PR for lockutil: #2729.

I'll rebase and squash the commit when it is merge.
nirs: Squashed the test commit and rebase on #2729 since the tests are broken without it. I'll rebase on master after #2729 it is merged.

Patch:
diff --git a/pkg/downloader/downloader.go b/pkg/downloader/downloader.go
index c64495d7..26117401 100644
--- a/pkg/downloader/downloader.go
+++ b/pkg/downloader/downloader.go
@@ -12,14 +12,15 @@ import (
 	"os/exec"
 	"path"
 	"path/filepath"
-	"strconv"
 	"strings"
+	"sync/atomic"
 	"time"
 
 	"github.com/cheggaaa/pb/v3"
 	"github.com/containerd/continuity/fs"
 	"github.com/lima-vm/lima/pkg/httpclientutil"
 	"github.com/lima-vm/lima/pkg/localpathutil"
+	"github.com/lima-vm/lima/pkg/lockutil"
 	"github.com/lima-vm/lima/pkg/progressbar"
 	"github.com/opencontainers/go-digest"
 	"github.com/sirupsen/logrus"
@@ -267,21 +268,21 @@ func Download(ctx context.Context, local, remote string, opts ...Opt) (*Result,
 		return nil, err
 	}
 	shadURL := filepath.Join(shad, "url")
-	if err := atomicWrite(shadURL, []byte(remote), 0o644); err != nil {
+	if err := writeFirst(shadURL, []byte(remote), 0o644); err != nil {
 		return nil, err
 	}
 	if err := downloadHTTP(ctx, shadData, shadTime, shadType, remote, o.description, o.expectedDigest); err != nil {
 		return nil, err
 	}
-	// no need to pass the digest to copyLocal(), as we already verified the digest
-	if err := copyLocal(ctx, localPath, shadData, ext, o.decompress, "", ""); err != nil {
-		return nil, err
-	}
 	if shadDigest != "" && o.expectedDigest != "" {
-		if err := atomicWrite(shadDigest, []byte(o.expectedDigest.String()), 0o644); err != nil {
+		if err := writeFirst(shadDigest, []byte(o.expectedDigest.String()), 0o644); err != nil {
 			return nil, err
 		}
 	}
+	// no need to pass the digest to copyLocal(), as we already verified the digest
+	if err := copyLocal(ctx, localPath, shadData, ext, o.decompress, "", ""); err != nil {
+		return nil, err
+	}
 	res := &Result{
 		Status:          StatusDownloaded,
 		CachePath:       shadData,
@@ -605,13 +606,13 @@ func downloadHTTP(ctx context.Context, localPath, lastModified, contentType, url
 	}
 	if lastModified != "" {
 		lm := resp.Header.Get("Last-Modified")
-		if err := atomicWrite(lastModified, []byte(lm), 0o644); err != nil {
+		if err := writeFirst(lastModified, []byte(lm), 0o644); err != nil {
 			return err
 		}
 	}
 	if contentType != "" {
 		ct := resp.Header.Get("Content-Type")
-		if err := atomicWrite(contentType, []byte(ct), 0o644); err != nil {
+		if err := writeFirst(contentType, []byte(ct), 0o644); err != nil {
 			return err
 		}
 	}
@@ -672,43 +673,43 @@ func downloadHTTP(ctx context.Context, localPath, lastModified, contentType, url
 		return err
 	}
 
-	return os.Rename(localPathTmp, localPath)
+	// If localPath was created by a parallel download keep it. Replacing it
+	// while another process is copying it to the destination may fail the
+	// clonefile syscall. We use a lock to ensure that only one process updates
+	// data, and when we return data file exists.
+
+	return lockutil.WithDirLock(filepath.Dir(localPath), func() error {
+		if _, err := os.Stat(localPath); err == nil {
+			return nil
+		} else if !errors.Is(err, os.ErrNotExist) {
+			return err
+		}
+		return os.Rename(localPathTmp, localPath)
+	})
 }
 
+var tempfileCount atomic.Uint64
+
 // To allow parallel download we use a per-process unique suffix for tempoary
 // files. Renaming the temporary file to the final file is safe without
 // synchronization on posix.
+// To make it easy to test we also include a counter ensuring that each
+// temporary file is unique in the same process.
 // https://github.com/lima-vm/lima/issues/2722
 func perProcessTempfile(path string) string {
-	return path + ".tmp." + strconv.FormatInt(int64(os.Getpid()), 10)
+	return fmt.Sprintf("%s.tmp.%d.%d", path, os.Getpid(), tempfileCount.Add(1))
 }
 
-// atomicWrite writes data to path, creating a new file or replacing existing
-// one. Multiple processess can write to the same path safely. Safe on posix and
-// likely safe on windows when using NTFS.
-func atomicWrite(path string, data []byte, perm os.FileMode) error {
-	tmpPath := perProcessTempfile(path)
-	tmp, err := os.OpenFile(tmpPath, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, perm)
-	if err != nil {
-		return err
-	}
-	defer func() {
-		if err != nil {
-			tmp.Close()
-			os.RemoveAll(tmpPath)
+// writeFirst writes data to path unless path already exists.
+func writeFirst(path string, data []byte, perm os.FileMode) error {
+	return lockutil.WithDirLock(filepath.Dir(path), func() error {
+		if _, err := os.Stat(path); err == nil {
+			return nil
+		} else if !errors.Is(err, os.ErrNotExist) {
+			return err
 		}
-	}()
-	if _, err = tmp.Write(data); err != nil {
-		return err
-	}
-	if err = tmp.Sync(); err != nil {
-		return err
-	}
-	if err = tmp.Close(); err != nil {
-		return err
-	}
-	err = os.Rename(tmpPath, path)
-	return err
+		return os.WriteFile(path, data, perm)
+	})
 }
 
 // CacheEntries returns a map of cache entries.


Test patch:
diff --git a/pkg/downloader/downloader_test.go b/pkg/downloader/downloader_test.go
index 136de343..00c42c6f 100644
--- a/pkg/downloader/downloader_test.go
+++ b/pkg/downloader/downloader_test.go
@@ -8,6 +8,7 @@ import (
 	"os/exec"
 	"path/filepath"
 	"runtime"
+	"slices"
 	"strings"
 	"testing"
 	"time"
@@ -21,6 +22,20 @@ func TestMain(m *testing.M) {
 	os.Exit(m.Run())
 }
 
+type downloadResult struct {
+	r   *Result
+	err error
+}
+
+// We expect only few parallel downloads. Testing with larger number to find
+// races quicker. 20 parallel downloads take about 120 milliseocnds on M1 Pro.
+const parallelDownloads = 20
+
+// When downloading in parallel usually all downloads completed with
+// StatusDownload, but some may be delayed and find the data file when they
+// start. Can be reproduced locally using 100 parallel downloads.
+var parallelStatus = []Status{StatusDownloaded, StatusUsedCache}
+
 func TestDownloadRemote(t *testing.T) {
 	ts := httptest.NewServer(http.FileServer(http.Dir("testdata")))
 	t.Cleanup(ts.Close)
@@ -57,38 +72,90 @@ func TestDownloadRemote(t *testing.T) {
 		})
 	})
 	t.Run("with cache", func(t *testing.T) {
-		cacheDir := filepath.Join(t.TempDir(), "cache")
-		localPath := filepath.Join(t.TempDir(), t.Name())
-		r, err := Download(context.Background(), localPath, dummyRemoteFileURL, WithExpectedDigest(dummyRemoteFileDigest), WithCacheDir(cacheDir))
-		assert.NilError(t, err)
-		assert.Equal(t, StatusDownloaded, r.Status)
+		t.Run("serial", func(t *testing.T) {
+			cacheDir := filepath.Join(t.TempDir(), "cache")
+			localPath := filepath.Join(t.TempDir(), t.Name())
+			r, err := Download(context.Background(), localPath, dummyRemoteFileURL,
+				WithExpectedDigest(dummyRemoteFileDigest), WithCacheDir(cacheDir))
+			assert.NilError(t, err)
+			assert.Equal(t, StatusDownloaded, r.Status)
 
-		r, err = Download(context.Background(), localPath, dummyRemoteFileURL, WithExpectedDigest(dummyRemoteFileDigest), WithCacheDir(cacheDir))
-		assert.NilError(t, err)
-		assert.Equal(t, StatusSkipped, r.Status)
+			r, err = Download(context.Background(), localPath, dummyRemoteFileURL,
+				WithExpectedDigest(dummyRemoteFileDigest), WithCacheDir(cacheDir))
+			assert.NilError(t, err)
+			assert.Equal(t, StatusSkipped, r.Status)
 
-		localPath2 := localPath + "-2"
-		r, err = Download(context.Background(), localPath2, dummyRemoteFileURL, WithExpectedDigest(dummyRemoteFileDigest), WithCacheDir(cacheDir))
-		assert.NilError(t, err)
-		assert.Equal(t, StatusUsedCache, r.Status)
+			localPath2 := localPath + "-2"
+			r, err = Download(context.Background(), localPath2, dummyRemoteFileURL,
+				WithExpectedDigest(dummyRemoteFileDigest), WithCacheDir(cacheDir))
+			assert.NilError(t, err)
+			assert.Equal(t, StatusUsedCache, r.Status)
+		})
+		t.Run("parallel", func(t *testing.T) {
+			cacheDir := filepath.Join(t.TempDir(), "cache")
+			results := make(chan downloadResult, parallelDownloads)
+			for i := 0; i < parallelDownloads; i++ {
+				go func() {
+					// Parallel download is supported only for different instances with unique localPath.
+					localPath := filepath.Join(t.TempDir(), t.Name())
+					r, err := Download(context.Background(), localPath, dummyRemoteFileURL,
+						WithExpectedDigest(dummyRemoteFileDigest), WithCacheDir(cacheDir))
+					results <- downloadResult{r, err}
+				}()
+			}
+			// We must process all results before cleanup.
+			for i := 0; i < parallelDownloads; i++ {
+				result := <-results
+				if result.err != nil {
+					t.Errorf("Download failed: %s", result.err)
+				} else if !slices.Contains(parallelStatus, result.r.Status) {
+					t.Errorf("Expected download status %s, got %s", parallelStatus, result.r.Status)
+				}
+			}
+		})
 	})
 	t.Run("caching-only mode", func(t *testing.T) {
-		_, err := Download(context.Background(), "", dummyRemoteFileURL, WithExpectedDigest(dummyRemoteFileDigest))
-		assert.ErrorContains(t, err, "cache directory to be specified")
+		t.Run("serial", func(t *testing.T) {
+			_, err := Download(context.Background(), "", dummyRemoteFileURL, WithExpectedDigest(dummyRemoteFileDigest))
+			assert.ErrorContains(t, err, "cache directory to be specified")
 
-		cacheDir := filepath.Join(t.TempDir(), "cache")
-		r, err := Download(context.Background(), "", dummyRemoteFileURL, WithExpectedDigest(dummyRemoteFileDigest), WithCacheDir(cacheDir))
-		assert.NilError(t, err)
-		assert.Equal(t, StatusDownloaded, r.Status)
+			cacheDir := filepath.Join(t.TempDir(), "cache")
+			r, err := Download(context.Background(), "", dummyRemoteFileURL, WithExpectedDigest(dummyRemoteFileDigest),
+				WithCacheDir(cacheDir))
+			assert.NilError(t, err)
+			assert.Equal(t, StatusDownloaded, r.Status)
 
-		r, err = Download(context.Background(), "", dummyRemoteFileURL, WithExpectedDigest(dummyRemoteFileDigest), WithCacheDir(cacheDir))
-		assert.NilError(t, err)
-		assert.Equal(t, StatusUsedCache, r.Status)
+			r, err = Download(context.Background(), "", dummyRemoteFileURL, WithExpectedDigest(dummyRemoteFileDigest),
+				WithCacheDir(cacheDir))
+			assert.NilError(t, err)
+			assert.Equal(t, StatusUsedCache, r.Status)
 
-		localPath := filepath.Join(t.TempDir(), t.Name())
-		r, err = Download(context.Background(), localPath, dummyRemoteFileURL, WithExpectedDigest(dummyRemoteFileDigest), WithCacheDir(cacheDir))
-		assert.NilError(t, err)
-		assert.Equal(t, StatusUsedCache, r.Status)
+			localPath := filepath.Join(t.TempDir(), t.Name())
+			r, err = Download(context.Background(), localPath, dummyRemoteFileURL,
+				WithExpectedDigest(dummyRemoteFileDigest), WithCacheDir(cacheDir))
+			assert.NilError(t, err)
+			assert.Equal(t, StatusUsedCache, r.Status)
+		})
+		t.Run("parallel", func(t *testing.T) {
+			cacheDir := filepath.Join(t.TempDir(), "cache")
+			results := make(chan downloadResult, parallelDownloads)
+			for i := 0; i < parallelDownloads; i++ {
+				go func() {
+					r, err := Download(context.Background(), "", dummyRemoteFileURL,
+						WithExpectedDigest(dummyRemoteFileDigest), WithCacheDir(cacheDir))
+					results <- downloadResult{r, err}
+				}()
+			}
+			// We must process all results before cleanup.
+			for i := 0; i < parallelDownloads; i++ {
+				result := <-results
+				if result.err != nil {
+					t.Errorf("Download failed: %s", result.err)
+				} else if !slices.Contains(parallelStatus, result.r.Status) {
+					t.Errorf("Expected download status %s, got %s", parallelStatus, result.r.Status)
+				}
+			}
+		})
 	})
 	t.Run("cached", func(t *testing.T) {
 		_, err := Cached(dummyRemoteFileURL, WithExpectedDigest(dummyRemoteFileDigest))