Merge pull request #1837 from restic/fix-1833

cache: Ensure failed downloads are retried

changelog/unreleased/issue-1833

@@ -0,0 +1,9 @@
+Bugfix: Fix caching files on error
+
+During `check` it may happen that different threads access the same file in the
+backend, which is then downloaded into the cache only once. When that fails,
+only the thread which is responsible for downloading the file signals the
+correct error. The other threads just assume that the file has been downloaded
+successfully and then get an error when they try to access the cached file.
+
+https://github.com/restic/restic/issues/1833

internal/cache/backend.go

@@ -91,14 +91,6 @@ var autoCacheFiles = map[restic.FileType]bool{
 
 func (b *Backend) cacheFile(ctx context.Context, h restic.Handle) error {
 	finish := make(chan struct{})
-	defer func() {
-		close(finish)
-
-		// remove the finish channel from the map
-		b.inProgressMutex.Lock()
-		delete(b.inProgress, h)
-		b.inProgressMutex.Unlock()
-	}()
 
 	b.inProgressMutex.Lock()
 	other, alreadyDownloading := b.inProgress[h]
@@ -120,10 +112,17 @@ func (b *Backend) cacheFile(ctx context.Context, h restic.Handle) error {
 	if err != nil {
 		// try to remove from the cache, ignore errors
 		_ = b.Cache.Remove(h)
-		return err
 	}
 
-	return nil
+	// signal other waiting goroutines that the file may now be cached
+	close(finish)
+
+	// remove the finish channel from the map
+	b.inProgressMutex.Lock()
+	delete(b.inProgress, h)
+	b.inProgressMutex.Unlock()
+
+	return err
 }
 
 // loadFromCacheOrDelegate will try to load the file from the cache, and fall
@@ -131,12 +130,13 @@ func (b *Backend) cacheFile(ctx context.Context, h restic.Handle) error {
 func (b *Backend) loadFromCacheOrDelegate(ctx context.Context, h restic.Handle, length int, offset int64, consumer func(rd io.Reader) error) error {
 	rd, err := b.Cache.Load(h, length, offset)
 	if err != nil {
+		debug.Log("error caching %v: %v, falling back to backend", h, err)
 		return b.Backend.Load(ctx, h, length, offset, consumer)
 	}
 
 	err = consumer(rd)
 	if err != nil {
-		rd.Close() // ignore secondary errors
+		_ = rd.Close() // ignore secondary errors
 		return err
 	}
 	return rd.Close()
@@ -193,19 +193,8 @@ func (b *Backend) Load(ctx context.Context, h restic.Handle, length int, offset
 
 	debug.Log("auto-store %v in the cache", h)
 	err := b.cacheFile(ctx, h)
-
 	if err == nil {
-		// load the cached version
-		rd, err := b.Cache.Load(h, 0, 0)
-		if err != nil {
-			return err
-		}
-		err = consumer(rd)
-		if err != nil {
-			rd.Close() // ignore secondary errors
-			return err
-		}
-		return rd.Close()
+		return b.loadFromCacheOrDelegate(ctx, h, length, offset, consumer)
 	}
 
 	debug.Log("error caching %v: %v, falling back to backend", h, err)

internal/cache/backend_test.go

@@ -3,9 +3,13 @@ package cache
 import (
 	"bytes"
 	"context"
+	"io"
 	"math/rand"
+	"sync"
 	"testing"
+	"time"
 
+	"github.com/pkg/errors"
 	"github.com/restic/restic/internal/backend"
 	"github.com/restic/restic/internal/backend/mem"
 	"github.com/restic/restic/internal/restic"
@@ -112,3 +116,59 @@ func TestBackend(t *testing.T) {
 		t.Errorf("removed file still in cache after stat")
 	}
 }
+
+type loadErrorBackend struct {
+	restic.Backend
+	loadError error
+}
+
+func (be loadErrorBackend) Load(ctx context.Context, h restic.Handle, length int, offset int64, fn func(rd io.Reader) error) error {
+	time.Sleep(10 * time.Millisecond)
+	return be.loadError
+}
+
+func TestErrorBackend(t *testing.T) {
+	be := mem.New()
+
+	c, cleanup := TestNewCache(t)
+	defer cleanup()
+
+	h, data := randomData(5234142)
+
+	// save directly in backend
+	save(t, be, h, data)
+
+	testErr := errors.New("test error")
+	errBackend := loadErrorBackend{
+		Backend:   be,
+		loadError: testErr,
+	}
+
+	loadTest := func(wg *sync.WaitGroup, be restic.Backend) {
+		defer wg.Done()
+
+		buf, err := backend.LoadAll(context.TODO(), be, h)
+		if err == testErr {
+			return
+		}
+
+		if err != nil {
+			t.Error(err)
+			return
+		}
+
+		if !bytes.Equal(buf, data) {
+			t.Errorf("data does not match")
+		}
+		time.Sleep(time.Millisecond)
+	}
+
+	wrappedBE := c.Wrap(errBackend)
+	var wg sync.WaitGroup
+	for i := 0; i < 5; i++ {
+		wg.Add(1)
+		go loadTest(&wg, wrappedBE)
+	}
+
+	wg.Wait()
+}