Merge pull request #3207 from aawsome/filerestorer-fix-errorhandling

Fix error handling in filerestorer and fix #3166
This commit is contained in:
MichaelEischer 2021-01-31 17:47:46 +01:00 committed by GitHub
commit b3dc127af5
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
5 changed files with 104 additions and 60 deletions

View file

@ -0,0 +1,9 @@
Bugfix: Improve error handling in restore
Restic restore used to not print errors while downloading file contents from
the repository. restore also incorrectly exited with a zero error code even
when there were errors during the restore process. Now, a non-zero code is
returned.
https://github.com/restic/restic/issues/3166
https://github.com/restic/restic/pull/3207

View file

@ -191,14 +191,26 @@ func runRestore(opts RestoreOptions, gopts GlobalOptions, args []string) error {
Verbosef("restoring %s to %s\n", res.Snapshot(), opts.Target)
err = res.RestoreTo(ctx, opts.Target)
if err == nil && opts.Verify {
if err != nil {
return err
}
if totalErrors > 0 {
return errors.Fatalf("There were %d errors\n", totalErrors)
}
if opts.Verify {
Verbosef("verifying files in %s\n", opts.Target)
var count int
count, err = res.VerifyFiles(ctx, opts.Target)
if err != nil {
return err
}
if totalErrors > 0 {
return errors.Fatalf("There were %d errors\n", totalErrors)
}
Verbosef("finished verifying %d files in %s\n", count, opts.Target)
}
if totalErrors > 0 {
Printf("There were %d errors\n", totalErrors)
}
return err
return nil
}

View file

@ -9,6 +9,8 @@ import (
"sort"
"sync"
"golang.org/x/sync/errgroup"
"github.com/restic/restic/internal/crypto"
"github.com/restic/restic/internal/debug"
"github.com/restic/restic/internal/errors"
@ -23,20 +25,16 @@ import (
const (
workerCount = 8
// fileInfo flags
fileProgress = 1
fileError = 2
largeFileBlobCount = 25
)
// information about regular file being restored
type fileInfo struct {
lock sync.Mutex
flags int
size int64
location string // file on local filesystem relative to restorer basedir
blobs interface{} // blobs of the file
lock sync.Mutex
inProgress bool
size int64
location string // file on local filesystem relative to restorer basedir
blobs interface{} // blobs of the file
}
type fileBlobInfo struct {
@ -60,6 +58,7 @@ type fileRestorer struct {
dst string
files []*fileInfo
Error func(string, error) error
}
func newFileRestorer(dst string,
@ -73,6 +72,7 @@ func newFileRestorer(dst string,
packLoader: packLoader,
filesWriter: newFilesWriter(workerCount),
dst: dst,
Error: restorerAbortOnAllErrors,
}
}
@ -142,47 +142,42 @@ func (r *fileRestorer) restoreFiles(ctx context.Context) error {
}
}
var wg sync.WaitGroup
wg, ctx := errgroup.WithContext(ctx)
downloadCh := make(chan *packInfo)
worker := func() {
defer wg.Done()
for {
select {
case <-ctx.Done():
return // context cancelled
case pack, ok := <-downloadCh:
if !ok {
return // channel closed
}
r.downloadPack(ctx, pack)
worker := func() error {
for pack := range downloadCh {
if err := r.downloadPack(ctx, pack); err != nil {
return err
}
}
return nil
}
for i := 0; i < workerCount; i++ {
wg.Add(1)
go worker()
wg.Go(worker)
}
// the main restore loop
for _, id := range packOrder {
pack := packs[id]
select {
case <-ctx.Done():
return ctx.Err()
case downloadCh <- pack:
debug.Log("Scheduled download pack %s", pack.id.Str())
wg.Go(func() error {
for _, id := range packOrder {
pack := packs[id]
select {
case <-ctx.Done():
return ctx.Err()
case downloadCh <- pack:
debug.Log("Scheduled download pack %s", pack.id.Str())
}
}
}
close(downloadCh)
return nil
})
close(downloadCh)
wg.Wait()
return nil
return wg.Wait()
}
const maxBufferSize = 4 * 1024 * 1024
func (r *fileRestorer) downloadPack(ctx context.Context, pack *packInfo) {
func (r *fileRestorer) downloadPack(ctx context.Context, pack *packInfo) error {
// calculate pack byte range and blob->[]files->[]offsets mappings
start, end := int64(math.MaxInt64), int64(0)
@ -237,12 +232,11 @@ func (r *fileRestorer) downloadPack(ctx context.Context, pack *packInfo) {
return blobs[sortedBlobs[i]].offset < blobs[sortedBlobs[j]].offset
})
markFileError := func(file *fileInfo, err error) {
file.lock.Lock()
defer file.lock.Unlock()
if file.flags&fileError == 0 {
file.flags |= fileError
sanitizeError := func(file *fileInfo, err error) error {
if err != nil {
err = r.Error(file.location, err)
}
return err
}
h := restic.Handle{Type: restic.PackFile, Name: pack.id.String()}
@ -263,7 +257,9 @@ func (r *fileRestorer) downloadPack(ctx context.Context, pack *packInfo) {
blobData, buf, err = r.loadBlob(bufRd, blobID, blob.length, buf)
if err != nil {
for file := range blob.files {
markFileError(file, err)
if errFile := sanitizeError(file, err); errFile != nil {
return errFile
}
}
continue
}
@ -277,37 +273,36 @@ func (r *fileRestorer) downloadPack(ctx context.Context, pack *packInfo) {
// - should allow concurrent writes to the file
// so write the first blob while holding file lock
// write other blobs after releasing the lock
file.lock.Lock()
create := file.flags&fileProgress == 0
createSize := int64(-1)
if create {
defer file.lock.Unlock()
file.flags |= fileProgress
createSize = file.size
} else {
file.lock.Lock()
if file.inProgress {
file.lock.Unlock()
} else {
defer file.lock.Unlock()
file.inProgress = true
createSize = file.size
}
return r.filesWriter.writeToFile(r.targetPath(file.location), blobData, offset, createSize)
}
err := writeToFile()
err := sanitizeError(file, writeToFile())
if err != nil {
markFileError(file, err)
break
return err
}
}
}
}
return nil
})
if err != nil {
for file := range pack.files {
markFileError(file, err)
if errFile := sanitizeError(file, err); errFile != nil {
return errFile
}
}
return
}
return nil
}
func (r *fileRestorer) loadBlob(rd io.Reader, blobID restic.ID, length int, buf []byte) ([]byte, []byte, error) {

View file

@ -8,6 +8,7 @@ import (
"testing"
"github.com/restic/restic/internal/crypto"
"github.com/restic/restic/internal/errors"
"github.com/restic/restic/internal/restic"
rtest "github.com/restic/restic/internal/test"
)
@ -237,3 +238,29 @@ func TestFileRestorerPackSkip(t *testing.T) {
},
}, files)
}
func TestErrorRestoreFiles(t *testing.T) {
tempdir, cleanup := rtest.TempDir(t)
defer cleanup()
content := []TestFile{
{
name: "file1",
blobs: []TestBlob{
{"data1-1", "pack1-1"},
},
}}
repo := newTestRepo(content)
loadError := errors.New("load error")
// loader always returns an error
repo.loader = func(ctx context.Context, h restic.Handle, length int, offset int64, fn func(rd io.Reader) error) error {
return loadError
}
r := newFileRestorer(tempdir, repo.loader, repo.key, repo.Lookup)
r.files = repo.files
err := r.restoreFiles(context.TODO())
rtest.Equals(t, loadError, err)
}

View file

@ -216,6 +216,7 @@ func (res *Restorer) RestoreTo(ctx context.Context, dst string) error {
idx := restic.NewHardlinkIndex()
filerestorer := newFileRestorer(dst, res.repo.Backend().Load, res.repo.Key(), res.repo.Index().Lookup)
filerestorer.Error = res.Error
debug.Log("first pass for %q", dst)