issue 9460: flush buffer when uploader completes

Signed-off-by: Lyndon-Li <lyonghui@vmware.com>

changelogs/unreleased/9561-Lyndon-Li‎‎

@@ -0,0 +1 @@
+Fix issue #9460, flush buffer before data mover completes

go.mod

@@ -43,6 +43,7 @@ require (
 	go.uber.org/zap v1.27.1
 	golang.org/x/mod v0.30.0
 	golang.org/x/oauth2 v0.33.0
+	golang.org/x/sys v0.38.0
 	golang.org/x/text v0.31.0
 	google.golang.org/api v0.256.0
 	google.golang.org/grpc v1.77.0
@@ -183,7 +184,6 @@ require (
 	golang.org/x/exp v0.0.0-20240719175910-8a7402abbf56 // indirect
 	golang.org/x/net v0.47.0 // indirect
 	golang.org/x/sync v0.18.0 // indirect
-	golang.org/x/sys v0.38.0 // indirect
 	golang.org/x/term v0.37.0 // indirect
 	golang.org/x/time v0.14.0 // indirect
 	golang.org/x/tools v0.38.0 // indirect

pkg/uploader/kopia/block_restore.go

@@ -35,6 +35,7 @@ type BlockOutput struct {
 	*restore.FilesystemOutput
 
 	targetFileName string
+	targetFile     *os.File
 }
 
 var _ restore.Output = &BlockOutput{}
@@ -52,7 +53,7 @@ func (o *BlockOutput) WriteFile(ctx context.Context, relativePath string, remote
 	if err != nil {
 		return errors.Wrapf(err, "failed to open file %s", o.targetFileName)
 	}
-	defer targetFile.Close()
+	o.targetFile = targetFile
 
 	buffer := make([]byte, bufferSize)
 
@@ -103,5 +104,21 @@ func (o *BlockOutput) BeginDirectory(ctx context.Context, relativePath string, e
 }
 
 func (o *BlockOutput) Flush() error {
-	return flushVolume(o.targetFileName)
+	if o.targetFile != nil {
+		if err := o.targetFile.Sync(); err != nil {
+			return errors.Wrapf(err, "error syncing block dev %v", o.targetFileName)
+		}
+	}
+
+	return nil
+}
+
+func (o *BlockOutput) Terminate() error {
+	if o.targetFile != nil {
+		if err := o.targetFile.Close(); err != nil {
+			return errors.Wrapf(err, "error closing block dev %v", o.targetFileName)
+		}
+	}
+
+	return nil
 }

pkg/uploader/kopia/flush_volume_linux.go

@@ -37,11 +37,14 @@ func flushVolume(dirPath string) error {
 		return errors.Wrapf(err, "error getting handle of dir %v", dirPath)
 	}
 
-	raw.Control(func(fd uintptr) {
+	var syncErr error
+	if err := raw.Control(func(fd uintptr) {
 		if e := unix.Syncfs(int(fd)); e != nil {
-			err = e
+			syncErr = e
 		}
-	})
+	}); err != nil {
+		return errors.Wrapf(err, "error calling fs sync from %v", dirPath)
+	}
 
-	return errors.Wrapf(err, "error syncing fs from %v", dirPath)
+	return errors.Wrapf(syncErr, "error syncing fs from %v", dirPath)
 }

pkg/uploader/kopia/restore_output.go

@@ -16,10 +16,15 @@ limitations under the License.
 
 package kopia
 
-import "github.com/pkg/errors"
+import (
+	"github.com/kopia/kopia/snapshot/restore"
+	"github.com/pkg/errors"
+)
 
 var errFlushUnsupported = errors.New("flush is not supported")
 
-type Flusher interface {
+type RestoreOutput interface {
+	restore.Output
 	Flush() error
+	Terminate() error
 }

pkg/uploader/kopia/snapshot.go

@@ -384,6 +384,10 @@ func (o *fileSystemRestoreOutput) Flush() error {
 	return flushVolumeFunc(o.TargetPath)
 }
 
+func (o *fileSystemRestoreOutput) Terminate() error {
+	return nil
+}
+
 // Restore restore specific sourcePath with given snapshotID and update progress
 func Restore(ctx context.Context, rep repo.RepositoryWriter, progress *Progress, snapshotID, dest string, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string,
 	log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, error) {
@@ -443,24 +447,23 @@ func Restore(ctx context.Context, rep repo.RepositoryWriter, progress *Progress,
 		return 0, 0, errors.Wrap(err, "error to init output")
 	}
 
-	var output restore.Output
-	var flusher Flusher
+	var output RestoreOutput
 	if volMode == uploader.PersistentVolumeBlock {
-		o := &BlockOutput{
+		output = &BlockOutput{
 			FilesystemOutput: fsOutput,
 		}
-
-		output = o
-		flusher = o
 	} else {
-		o := &fileSystemRestoreOutput{
+		output = &fileSystemRestoreOutput{
 			FilesystemOutput: fsOutput,
 		}
-
-		output = o
-		flusher = o
 	}
 
+	defer func() {
+		if err := output.Terminate(); err != nil {
+			log.Warnf("error terminating restore output for %v", path)
+		}
+	}()
+
 	stat, err := restoreEntryFunc(kopiaCtx, rep, output, rootEntry, restore.Options{
 		Parallel:               restoreConcurrency,
 		RestoreDirEntryAtDepth: math.MaxInt32,
@@ -474,14 +477,14 @@ func Restore(ctx context.Context, rep repo.RepositoryWriter, progress *Progress,
 		return 0, 0, errors.Wrapf(err, "Failed to copy snapshot data to the target")
 	}
 
-	if err := flusher.Flush(); err != nil {
+	if err := output.Flush(); err != nil {
 		if err == errFlushUnsupported {
 			log.Warnf("Skip flushing data for %v under the current OS %v", path, runtime.GOOS)
 		} else {
 			return 0, 0, errors.Wrapf(err, "Failed to flush data to target")
 		}
 	} else {
-		log.Warnf("Flush done for volume dir %v", path)
+		log.Infof("Flush done for volume dir %v", path)
 	}
 
 	return stat.RestoredTotalFileSize, stat.RestoredFileCount, nil