restore: re-read block file while falling back to different decompression method

backupbackingimage/backupbackingimage.go

@@ -497,17 +497,10 @@ func restoreBlock(bsDriver backupstore.BackupStoreDriver, backingImageFile *os.F
 
 func restoreBlockToFile(bsDriver backupstore.BackupStoreDriver, backingImageFile *os.File, decompression string, blk common.BlockMapping) error {
  blkFile := getBackingImageBlockFilePath(blk.BlockChecksum)
- rc, err := bsDriver.Read(blkFile)
+ r, err := backupstore.DecompressAndVerifyWithFallback(bsDriver, blkFile, decompression, blk.BlockChecksum)
  if err != nil {
  return err
  }
- defer rc.Close()
- r, err := util.DecompressAndVerifyWithFallback(decompression, rc, blk.BlockChecksum)
- if err != nil {
- if r == nil {
- return err
- }
- }
 
  if _, err := backingImageFile.Seek(blk.Offset, 0); err != nil {
  return err

deltablock.go

@@ -775,17 +775,9 @@ func RestoreDeltaBlockBackup(ctx context.Context, config *DeltaRestoreConfig) er
 
 func restoreBlockToFile(bsDriver BackupStoreDriver, volumeName string, volDev *os.File, decompression string, blk BlockMapping) error {
  blkFile := getBlockFilePath(volumeName, blk.BlockChecksum)
- rc, err := bsDriver.Read(blkFile)
+ r, err := DecompressAndVerifyWithFallback(bsDriver, blkFile, decompression, blk.BlockChecksum)
  if err != nil {
- return errors.Wrapf(err, "failed to read block %v with checksum %v", blkFile, blk.BlockChecksum)
- }
- defer rc.Close()
-
- r, err := util.DecompressAndVerifyWithFallback(decompression, rc, blk.BlockChecksum)
- if err != nil {
- if r == nil {
- return err
- }
+ return errors.Wrapf(err, "failed to decompress and verify block %v with checksum %v", blkFile, blk.BlockChecksum)
  }
 
  if _, err := volDev.Seek(blk.Offset, 0); err != nil {

util.go

@@ -1,9 +1,16 @@
 package backupstore
 
 import (
+ "compress/gzip"
  "context"
+ "io"
  "path/filepath"
+ "strings"
  "sync"
+
+ "github.com/pkg/errors"
+
+ "github.com/longhorn/backupstore/util"
 )
 
 func getBlockPath(volumeName string) string {
@@ -50,3 +57,53 @@ func mergeErrorChannels(ctx context.Context, channels ...<-chan error) <-chan er
  }()
  return out
 }
+
+// DecompressAndVerifyWithFallback decompresses the given data and verifies the data integrity.
+// If the decompression fails, it will try to decompress with the fallback method.
+func DecompressAndVerifyWithFallback(bsDriver BackupStoreDriver, blkFile, decompression, checksum string) (io.Reader, error) {
+ // Helper function to read block from backup store
+ readBlock := func() (io.ReadCloser, error) {
+ rc, err := bsDriver.Read(blkFile)
+ if err != nil {
+ return nil, errors.Wrapf(err, "failed to read block %v", blkFile)
+ }
+ return rc, nil
+ }
+
+ // First attempt to read and decompress/verify
+ rc, err := readBlock()
+ if err != nil {
+ return nil, err
+ }
+ defer rc.Close()
+
+ r, err := util.DecompressAndVerify(decompression, rc, checksum)
+ if err == nil {
+ return r, nil
+ }
+
+ // If there's an error, determine the alternative decompression method
+ alternativeDecompression := ""
+ if strings.Contains(err.Error(), gzip.ErrHeader.Error()) {
+ alternativeDecompression = "lz4"
+ } else if strings.Contains(err.Error(), "lz4: bad magic number") {
+ alternativeDecompression = "gzip"
+ }
+
+ // Second attempt with alternative decompression, if applicable
+ if alternativeDecompression != "" {
+ retriedRc, err := readBlock()
+ if err != nil {
+ return nil, err
+ }
+ defer retriedRc.Close()
+
+ r, err = util.DecompressAndVerify(alternativeDecompression, retriedRc, checksum)
+ if err != nil {
+ return nil, errors.Wrapf(err, "fallback decompression also failed for block %v", blkFile)
+ }
+ return r, nil
+ }
+
+ return nil, errors.Wrapf(err, "decompression verification failed for block %v", blkFile)
+}

util/util.go

@@ -23,7 +23,6 @@ import (
  lz4 "github.com/pierrec/lz4/v4"
  "github.com/pkg/errors"
  "github.com/sirupsen/logrus"
- "go.uber.org/multierr"
  "golang.org/x/sys/unix"
 
  "k8s.io/apimachinery/pkg/util/wait"
@@ -133,39 +132,6 @@ func DecompressAndVerify(method string, src io.Reader, checksum string) (io.Read
  return bytes.NewReader(block), nil
 }
 
-// DecompressAndVerifyWithFallback decompresses the given data and verifies the data integrity.
-// If the decompression fails, it will try to decompress with the fallback method.
-func DecompressAndVerifyWithFallback(decompression string, rc io.ReadCloser, checksum string) (io.Reader, error) {
- r, err := DecompressAndVerify(decompression, rc, checksum)
- if err == nil {
- return r, nil
- }
- // Fall back to other decompression method if the current one fails
- // The mitigation will be removed after identifying https:/longhorn/longhorn/issues/7687
- // Seek rc to offset 0
- seeker, ok := rc.(io.Seeker)
- if !ok {
- return nil, errors.Wrapf(err, "failed to cast to io.Seeker for block %v", checksum)
- }
-
- _, errFallback := seeker.Seek(0, io.SeekStart)
- if errFallback != nil {
- // Merge the err1 and err2 and error out
- return nil, errors.Wrapf(multierr.Append(err, errFallback), "failed to seek to offset 0 for block %v", checksum)
- }
-
- if strings.Contains(err.Error(), gzip.ErrHeader.Error()) {
- r, errFallback = DecompressAndVerify("lz4", rc, checksum)
- } else if strings.Contains(err.Error(), "lz4: bad magic number") {
- r, errFallback = DecompressAndVerify("gzip", rc, checksum)
- }
- if errFallback != nil {
- return nil, errors.Wrapf(multierr.Append(err, errFallback), "failed to decompress and verify block %v with fallback", checksum)
- }
-
- return r, err
-}
-
 func newCompressionWriter(method string, buffer io.Writer) (io.WriteCloser, error) {
  switch method {
  case "gzip":