Move code to different places, add comments here and there, no actual changes

internal/backer/archive.go

@@ -0,0 +1,175 @@
+// Package backer implements HTTP/HTTPS backup server.
+package backer
+
+import (
+	"archive/tar"
+	"context"
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+	"strings"
+
+	"backer/internal/log"
+)
+
+// createArchiveStream creates a pipe and streams a tar archive compressed by the provided writer.
+// The writerFactory creates the compression writer wrapping the pipe writer.
+// This is the core of all CreateTar*Stream functions.
+func createArchiveStream(ctx context.Context, filepaths []string, writerFactory func(io.Writer) (io.WriteCloser, error)) io.ReadCloser {
+	pr, pw := io.Pipe()
+
+	go func() {
+		var pipeErr error
+
+		closePipeWithError := func(err error) {
+			if pipeErr != nil {
+				return
+			}
+
+			pipeErr = err
+			pw.CloseWithError(err)
+		}
+
+		defer func() {
+			if pipeErr == nil {
+				pw.Close()
+			}
+		}()
+
+		cw, err := writerFactory(pw)
+		if err != nil {
+			closePipeWithError(fmt.Errorf("compression writer error: %w", err))
+
+			return
+		}
+
+		defer func() {
+			if pipeErr == nil {
+				if err := cw.Close(); err != nil {
+					if !isPipeClosedError(err) {
+						log.Errorf("Failed to close compression writer: %v", err)
+					}
+				}
+			}
+		}()
+
+		tw := tar.NewWriter(cw)
+
+		defer func() {
+			if pipeErr == nil {
+				if err := tw.Close(); err != nil {
+					if !isPipeClosedError(err) {
+						log.Errorf("Failed to close tar writer: %v", err)
+					}
+				}
+			}
+		}()
+
+		writeFilesToTar(ctx, filepaths, tw, closePipeWithError)
+	}()
+
+	return pr
+}
+
+// writeFilesToTar iterates over filepaths and writes each file to the tar archive.
+func writeFilesToTar(ctx context.Context, filepaths []string, tw *tar.Writer, closePipeWithError func(error)) {
+	for _, fpath := range filepaths {
+		var (
+			err        error
+			linkTarget string
+		)
+
+		select {
+		case <-ctx.Done():
+			closePipeWithError(ctx.Err())
+
+			return
+
+		default:
+		}
+
+		log.Debugf("Adding file: %s", fpath)
+
+		st, err := os.Lstat(fpath)
+
+		if err != nil {
+			log.Warnf("Skipping %s: %v", fpath, err)
+
+			continue
+		}
+
+		if st.Mode()&os.ModeSymlink != 0 {
+			linkTarget, err = os.Readlink(fpath)
+
+			if err != nil {
+				log.Warnf("Skipping %s: readlink failed: %v", fpath, err)
+
+				continue
+			}
+		}
+
+		// Skip sockets explicitly (they cannot be backed up).
+		if st.Mode()&os.ModeSocket != 0 {
+			log.Debugf("Skipping socket: %s", fpath)
+
+			continue
+		}
+
+		header, err := tar.FileInfoHeader(st, linkTarget)
+
+		if err != nil {
+			log.Warnf("Skipping %s: tar header failed: %v", fpath, err)
+
+			continue
+		}
+
+		header.Format = tar.FormatGNU
+		header.Name = strings.TrimPrefix(filepath.ToSlash(filepath.Clean(fpath)), "/")
+
+		mode := st.Mode()
+
+		if err := writeWithContext(ctx, func() error {
+			return tw.WriteHeader(header)
+		}); err != nil {
+			closePipeWithError(err)
+
+			return
+		}
+
+		// Only stream file contents for regular files.
+		if mode.IsRegular() {
+			f, err := os.Open(fpath)
+			if err != nil {
+				log.Warnf("Skipping %s: open failed: %v", fpath, err)
+
+				continue
+			}
+
+			err = copyWithContext(ctx, tw, f)
+
+			if e := f.Close(); e != nil {
+				log.Warnf("Failed to close file %s: %v", fpath, e)
+			}
+
+			if err != nil {
+				log.Warnf("Skipping %s: copy failed: %v", fpath, err)
+
+				continue
+			}
+		}
+	}
+}
+
+// isPipeClosedError checks if the error is due to a closed pipe.
+func isPipeClosedError(err error) bool {
+	if err == nil {
+		return false
+	}
+
+	// Check for the specific pipe closed error message.
+	return err.Error() == "io: read/write on closed pipe" ||
+		err.Error() == "io: read on closed pipe"
+}
+
+/* vim: setlocal ft=go noet ai ts=4 sw=4 sts=4: */

internal/backer/bzip2.go

@@ -0,0 +1,17 @@
+package backer
+
+import (
+	"context"
+	"io"
+
+	"github.com/dsnet/compress/bzip2"
+)
+
+// CreateTarBzip2Stream takes a list of files and returns a reader that reads from a bzip2-compressed tar archive.
+func CreateTarBzip2Stream(ctx context.Context, filepaths []string) io.ReadCloser {
+	return createArchiveStream(ctx, filepaths, func(w io.Writer) (io.WriteCloser, error) {
+		return bzip2.NewWriter(w, nil)
+	})
+}
+
+/* vim: setlocal ft=go noet ai ts=4 sw=4 sts=4: */

internal/backer/consts.go

@@ -0,0 +1,57 @@
+package backer
+
+import (
+	"time"
+)
+
+const (
+	// readHeaderTimeout is the maximum duration for reading request headers.
+	readHeaderTimeout = 5 * time.Second
+
+	// copyBufferSize is the buffer size used for streaming file contents.
+	copyBufferSize = 32 * 1024
+
+	// portMin is the minimum valid port number.
+	portMin = 1
+
+	// portMax is the maximum valid port number.
+	portMax = 65535
+
+	// backupTimeoutMin is the minimum backup timeout in minutes.
+	backupTimeoutMin = 1
+
+	// backupTimeoutMax is the maximum backup timeout in minutes (24 hours).
+	backupTimeoutMax = 1440
+
+	// compressionLevelMin is the minimum compression level.
+	compressionLevelMin = 1
+
+	// compressionLevelMax is the maximum compression level.
+	compressionLevelMax = 9
+
+	// defaultPort is the default listen port.
+	defaultPort = 8086
+
+	// defaultBackupTimeout is the default backup timeout in minutes.
+	defaultBackupTimeout = 60
+
+	// defaultCompressionLevel is the default compression level.
+	defaultCompressionLevel = 9
+
+	// defaultLocation is the default API endpoint path.
+	defaultLocation = "/archive"
+
+	// defaultAddress is the default listen address.
+	defaultAddress = "0.0.0.0"
+
+	// defaultLogLevel is the default log verbosity.
+	defaultLogLevel = "info"
+
+	// defaultFilenamePrefix is the default prefix for backup filenames.
+	defaultFilenamePrefix = "backup"
+
+	// defaultCompressionAlgorithm is the default compression algorithm.
+	defaultCompressionAlgorithm = "gzip"
+)
+
+/* vim: setlocal ft=go noet ai ts=4 sw=4 sts=4: */

internal/backer/getFilesFromDirectories.go

@@ -1,4 +1,3 @@
-// Package backer implements simple backup server.
 package backer
 
 import (

internal/backer/globals.go

@@ -6,8 +6,11 @@ import (
 
 var (
 	// C contains the global configuration loaded from config file.
+	// This is intentional — config is loaded once at startup and never modified concurrently.
 	C Config
+
 	// excludePatterns contains compiled regex patterns from C.ExcludePatterns.
+	// Populated by LoadConfig, used by isExcluded during directory walking.
 	excludePatterns []*regexp.Regexp
 )
 

internal/backer/gzip.go

@@ -0,0 +1,25 @@
+package backer
+
+import (
+	"compress/gzip"
+	"context"
+	"io"
+
+	"github.com/klauspost/pgzip"
+)
+
+// CreateTarGzStream takes a list of files and returns a reader that reads from a gzip-compressed tar archive.
+func CreateTarGzStream(ctx context.Context, filepaths []string) io.ReadCloser {
+	return createArchiveStream(ctx, filepaths, func(w io.Writer) (io.WriteCloser, error) {
+		return gzip.NewWriterLevel(w, C.CompressionLevel)
+	})
+}
+
+// CreateTarPgzipStream takes a list of files and returns a reader that reads from a parallel gzip-compressed tar archive.
+func CreateTarPgzipStream(ctx context.Context, filepaths []string) io.ReadCloser {
+	return createArchiveStream(ctx, filepaths, func(w io.Writer) (io.WriteCloser, error) {
+		return pgzip.NewWriterLevel(w, C.CompressionLevel)
+	})
+}
+
+/* vim: setlocal ft=go noet ai ts=4 sw=4 sts=4: */

internal/backer/loadConfig.go

@@ -38,25 +38,25 @@ func LoadConfig(path string) error {
 
 	// Address: default and no range validation needed.
 	if C.Address == "" {
-		C.Address = "0.0.0.0"
+		C.Address = defaultAddress
 
 		log.Warnf("Config option address is not set, fallback to %s", C.Address)
 	}
 
 	// Port: set default, then validate range.
 	if C.Port == 0 {
-		C.Port = 8086
+		C.Port = defaultPort
 
 		log.Warnf("Config option port is not set, fallback to %d", C.Port)
 	}
 
-	if C.Port < 1 || C.Port > 65535 {
-		return fmt.Errorf("Port must be between 1 and 65535, got %d", C.Port)
+	if C.Port < portMin || C.Port > portMax {
+		return fmt.Errorf("Config option port must be between %d and %d, got %d", portMin, portMax, C.Port)
 	}
 
 	// Location: default and no range validation needed.
 	if C.Location == "" {
-		C.Location = "/archive"
+		C.Location = defaultLocation
 
 		log.Warnf("Config option location is not set, fallback to %s", C.Location)
 	}
@@ -68,7 +68,7 @@ func LoadConfig(path string) error {
 
 	// LogLevel: default and no range validation needed.
 	if C.LogLevel == "" {
-		C.LogLevel = "info"
+		C.LogLevel = defaultLogLevel
 
 		log.Warnf("Config option loglevel is not set, fallback to %s", C.LogLevel)
 	}
@@ -86,24 +86,24 @@ func LoadConfig(path string) error {
 
 	// BackupTimeout: set default, then validate range.
 	if C.BackupTimeout == 0 {
-		C.BackupTimeout = 60
+		C.BackupTimeout = defaultBackupTimeout
 
 		log.Warnf("Config option backup_timeout is not set, fallback to %d minutes", C.BackupTimeout)
 	}
 
-	if C.BackupTimeout < 1 || C.BackupTimeout > 1440 {
-		return fmt.Errorf("Backup_timeout must be between 1 and 1440 minutes, got %d", C.BackupTimeout)
+	if C.BackupTimeout < backupTimeoutMin || C.BackupTimeout > backupTimeoutMax {
+		return fmt.Errorf("Config option backup_timeout must be between %d and %d minutes, got %d", backupTimeoutMin, backupTimeoutMax, C.BackupTimeout)
 	}
 
 	// CompressionLevel: set default, then validate range.
 	if C.CompressionLevel == 0 {
-		C.CompressionLevel = 9
+		C.CompressionLevel = defaultCompressionLevel
 
 		log.Warnf("Config option compression_level is not set, fallback to %d", C.CompressionLevel)
 	}
 
-	if C.CompressionLevel < 1 || C.CompressionLevel > 9 {
-		return fmt.Errorf("Compression_level must be between 1 and 9, got %d", C.CompressionLevel)
+	if C.CompressionLevel < compressionLevelMin || C.CompressionLevel > compressionLevelMax {
+		return fmt.Errorf("Config option compression_level must be between %d and %d, got %d", compressionLevelMin, compressionLevelMax, C.CompressionLevel)
 	}
 
 	// HTTPS: validate cert/key when HTTPS enabled.
@@ -120,30 +120,30 @@ func LoadConfig(path string) error {
 
 	// User: required.
 	if C.User == "" {
-		return errors.New("Option user is not set")
+		return errors.New("Config option user is not set")
 	}
 
 	// Password: required.
 	if C.Password == "" {
-		return errors.New("Option password is not set")
+		return errors.New("Config option password is not set")
 	}
 
 	// FilenamePrefix: default.
 	if C.FilenamePrefix == "" {
-		C.FilenamePrefix = "backup"
+		C.FilenamePrefix = defaultFilenamePrefix
 
 		log.Warnf("Config option filename_prefix is not set, fallback to %s", C.FilenamePrefix)
 	}
 
 	// CompressionAlgorithm: default and validate.
 	if C.CompressionAlgorithm == "" {
-		C.CompressionAlgorithm = "gzip"
+		C.CompressionAlgorithm = defaultCompressionAlgorithm
 
 		log.Warnf("Config option compression_algorithm is not set, fallback to %s", C.CompressionAlgorithm)
 	}
 
 	if C.CompressionAlgorithm != "gzip" && C.CompressionAlgorithm != "bzip2" && C.CompressionAlgorithm != "zstd" && C.CompressionAlgorithm != "lz4" && C.CompressionAlgorithm != "xz" && C.CompressionAlgorithm != "pgzip" {
-		return fmt.Errorf("Compression_algorithm must be gzip, bzip2, zstd, lz4, xz or pgzip, got %s", C.CompressionAlgorithm)
+		return fmt.Errorf("Config option compression_algorithm must be gzip, bzip2, zstd, lz4, xz or pgzip, got %s", C.CompressionAlgorithm)
 	}
 
 	// Compile exclude patterns.

internal/backer/lz4.go

@@ -0,0 +1,17 @@
+package backer
+
+import (
+	"context"
+	"io"
+
+	"github.com/pierrec/lz4"
+)
+
+// CreateTarLz4Stream takes a list of files and returns a reader that reads from an lz4-compressed tar archive.
+func CreateTarLz4Stream(ctx context.Context, filepaths []string) io.ReadCloser {
+	return createArchiveStream(ctx, filepaths, func(w io.Writer) (io.WriteCloser, error) {
+		return lz4.NewWriter(w), nil
+	})
+}
+
+/* vim: setlocal ft=go noet ai ts=4 sw=4 sts=4: */

internal/backer/newServer.go

@@ -134,7 +134,18 @@ func NewServer(configPath string) (*http.Server, error) {
 		return &http.Server{
 			Addr:              fmt.Sprintf("%s:%d", C.Address, C.Port),
 			Handler:           withServerHeader(mux),
-			ReadHeaderTimeout: 5 * time.Second,
+			ReadHeaderTimeout: readHeaderTimeout,
+			WriteTimeout:      time.Duration(C.BackupTimeout) * time.Minute,
+		}, nil
+	}
+
+	// Decision about which mode (http or https) to run made in cmd/backer/main.go, where we run server.ListenAndServe()
+	// or server.ListenAndServeTLS() depending on value of C.NoHTTPS.
+	if C.NoHTTPS {
+		return &http.Server{
+			Addr:              fmt.Sprintf("%s:%d", C.Address, C.Port),
+			Handler:           withServerHeader(mux),
+			ReadHeaderTimeout: readHeaderTimeout,
 			WriteTimeout:      time.Duration(C.BackupTimeout) * time.Minute,
 		}, nil
 	}
@@ -166,7 +177,7 @@ func writeWithContext(ctx context.Context, fn func() error) error {
 // context cancellation. If the context is canceled, the copy stops and returns
 // ctx.Err().
 func copyWithContext(ctx context.Context, dst io.Writer, src io.Reader) error {
-	buf := make([]byte, 32*1024)
+	buf := make([]byte, copyBufferSize)
 
 	for {
 		select {

internal/backer/types.go

@@ -1,7 +1,7 @@
 package backer
 
+// Config main config struct.
 type (
-	// Config main config struct.
 	Config struct {
 		Address              string   `json:"address"`
 		Port                 int      `json:"port"`
@@ -15,10 +15,10 @@ type (
 		LogLevel             string   `json:"loglevel"`
 		Directories          []string `json:"directories"`
 		BackupTimeout        int      `json:"backup_timeout"`        // Timeout in minutes for backup streaming.
-		CompressionLevel     int      `json:"compression_level"`     // Gzip compression level (1-9, default 9).
+		CompressionLevel     int      `json:"compression_level"`     // Compression level (1-9, default 9).
 		ExcludePatterns      []string `json:"exclude_patterns"`      // Regex patterns to exclude from backup.
 		FilenamePrefix       string   `json:"filename_prefix"`       // Prefix for backup filename (default: "backup").
-		CompressionAlgorithm string   `json:"compression_algorithm"` // Compression algorithm: gzip or bzip2 (default: "gzip").
+		CompressionAlgorithm string   `json:"compression_algorithm"` // Compression algorithm: gzip, pgzip, bzip2, zstd, lz4 or xz.
 	}
 )
 

internal/backer/utils.go

@@ -1,884 +0,0 @@
-package backer
-
-import (
-	"archive/tar"
-	"compress/gzip"
-	"context"
-	"fmt"
-	"io"
-	"os"
-	"path/filepath"
-	"strings"
-
-	"backer/internal/log"
-
-	"github.com/dsnet/compress/bzip2"
-	"github.com/klauspost/compress/zstd"
-	"github.com/klauspost/pgzip"
-	"github.com/pierrec/lz4"
-	"github.com/ulikunitz/xz"
-)
-
-// CreateTarGzStream takes a list of files and returns a reader that reads from a tar archive containing these files.
-func CreateTarGzStream(ctx context.Context, filepaths []string) io.ReadCloser {
-	pr, pw := io.Pipe()
-
-	go func() {
-		var pipeErr error // Track if we close pipe with error.
-
-		// Helper to close pipe with error, only once.
-		closePipeWithError := func(err error) {
-			if pipeErr != nil {
-				return // Already closed with error.
-			}
-
-			pipeErr = err
-			pw.CloseWithError(err)
-		}
-
-		defer func() {
-			// Only do normal close if we haven't closed with error.
-			if pipeErr == nil {
-				pw.Close()
-			}
-		}()
-
-		gw, err := gzip.NewWriterLevel(pw, C.CompressionLevel)
-		if err != nil {
-			closePipeWithError(fmt.Errorf("gzip writer error: %w", err))
-
-			return
-		}
-
-		defer func() {
-			// Only close gzip writer if pipe is still open.
-			if pipeErr == nil {
-				if err := gw.Close(); err != nil {
-					// Ignore pipe closed errors - reader may have closed early.
-					if !isPipeClosedError(err) {
-						log.Errorf("Failed to close gzip writer: %v", err)
-					}
-				}
-			}
-		}()
-
-		tw := tar.NewWriter(gw)
-
-		defer func() {
-			// Only close tar writer if pipe is still open.
-			if pipeErr == nil {
-				if err := tw.Close(); err != nil {
-					// Ignore pipe closed errors - reader may have closed early.
-					if !isPipeClosedError(err) {
-						log.Errorf("Failed to close tar writer: %v", err)
-					}
-				}
-			}
-		}()
-
-		for _, fpath := range filepaths {
-			var (
-				err        error
-				linkTarget string
-			)
-
-			// Check for cancellation early.
-			select {
-			case <-ctx.Done():
-				closePipeWithError(ctx.Err())
-
-				return
-
-			default:
-			}
-
-			log.Debugf("Adding file: %s", fpath)
-
-			st, err := os.Lstat(fpath)
-
-			if err != nil {
-				log.Warnf("Skipping %s: %v", fpath, err)
-
-				continue
-			}
-
-			if st.Mode()&os.ModeSymlink != 0 {
-				linkTarget, err = os.Readlink(fpath)
-
-				if err != nil {
-					log.Warnf("Skipping %s: readlink failed: %v", fpath, err)
-
-					continue
-				}
-			}
-
-			// Skip sockets explicitly (they cannot be backed up).
-			if st.Mode()&os.ModeSocket != 0 {
-				log.Debugf("Skipping socket: %s", fpath)
-
-				continue
-			}
-
-			header, err := tar.FileInfoHeader(st, linkTarget)
-
-			if err != nil {
-				log.Warnf("Skipping %s: tar header failed: %v", fpath, err)
-
-				continue
-			}
-
-			header.Format = tar.FormatGNU
-			header.Name = strings.TrimPrefix(filepath.ToSlash(filepath.Clean(fpath)), "/")
-
-			mode := st.Mode()
-
-			// Write header with cancellation awareness.
-			if err := writeWithContext(ctx, func() error {
-				return tw.WriteHeader(header)
-			}); err != nil {
-				closePipeWithError(err)
-
-				return
-			}
-
-			// Only stream file contents for regular files.
-			if mode.IsRegular() {
-				f, err := os.Open(fpath)
-				if err != nil {
-					log.Warnf("Skipping %s: open failed: %v", fpath, err)
-
-					continue
-				}
-
-				err = copyWithContext(ctx, tw, f)
-
-				if e := f.Close(); e != nil {
-					log.Warnf("Failed to close file %s: %v", fpath, e)
-				}
-
-				if err != nil {
-					log.Warnf("Skipping %s: copy failed: %v", fpath, err)
-
-					continue
-				}
-			}
-		}
-	}()
-
-	return pr
-}
-
-// CreateTarBzip2Stream takes a list of files and returns a reader that reads from a bzip2-compressed tar archive.
-func CreateTarBzip2Stream(ctx context.Context, filepaths []string) io.ReadCloser {
-	pr, pw := io.Pipe()
-
-	go func() {
-		var pipeErr error
-
-		closePipeWithError := func(err error) {
-			if pipeErr != nil {
-				return
-			}
-
-			pipeErr = err
-			pw.CloseWithError(err)
-		}
-
-		defer func() {
-			if pipeErr == nil {
-				pw.Close()
-			}
-		}()
-
-		bw, err := bzip2.NewWriter(pw, nil)
-		if err != nil {
-			closePipeWithError(fmt.Errorf("bzip2 writer error: %w", err))
-
-			return
-		}
-
-		defer func() {
-			if pipeErr == nil {
-				if err := bw.Close(); err != nil {
-					if !isPipeClosedError(err) {
-						log.Errorf("Failed to close bzip2 writer: %v", err)
-					}
-				}
-			}
-		}()
-
-		tw := tar.NewWriter(bw)
-
-		defer func() {
-			if pipeErr == nil {
-				if err := tw.Close(); err != nil {
-					if !isPipeClosedError(err) {
-						log.Errorf("Failed to close tar writer: %v", err)
-					}
-				}
-			}
-		}()
-
-		for _, fpath := range filepaths {
-			var (
-				err        error
-				linkTarget string
-			)
-
-			select {
-			case <-ctx.Done():
-				closePipeWithError(ctx.Err())
-
-				return
-
-			default:
-			}
-
-			log.Debugf("Adding file: %s", fpath)
-
-			st, err := os.Lstat(fpath)
-
-			if err != nil {
-				log.Warnf("Skipping %s: %v", fpath, err)
-
-				continue
-			}
-
-			if st.Mode()&os.ModeSymlink != 0 {
-				linkTarget, err = os.Readlink(fpath)
-
-				if err != nil {
-					log.Warnf("Skipping %s: readlink failed: %v", fpath, err)
-
-					continue
-				}
-			}
-
-			if st.Mode()&os.ModeSocket != 0 {
-				log.Debugf("Skipping socket: %s", fpath)
-
-				continue
-			}
-
-			header, err := tar.FileInfoHeader(st, linkTarget)
-
-			if err != nil {
-				log.Warnf("Skipping %s: tar header failed: %v", fpath, err)
-
-				continue
-			}
-
-			header.Format = tar.FormatGNU
-			header.Name = strings.TrimPrefix(filepath.ToSlash(filepath.Clean(fpath)), "/")
-
-			mode := st.Mode()
-
-			if err := writeWithContext(ctx, func() error {
-				return tw.WriteHeader(header)
-			}); err != nil {
-				closePipeWithError(err)
-
-				return
-			}
-
-			if mode.IsRegular() {
-				f, err := os.Open(fpath)
-				if err != nil {
-					log.Warnf("Skipping %s: open failed: %v", fpath, err)
-
-					continue
-				}
-
-				err = copyWithContext(ctx, tw, f)
-
-				if e := f.Close(); e != nil {
-					log.Warnf("Failed to close file %s: %v", fpath, e)
-				}
-
-				if err != nil {
-					log.Warnf("Skipping %s: copy failed: %v", fpath, err)
-
-					continue
-				}
-			}
-		}
-	}()
-
-	return pr
-}
-
-// zstdLevel returns the zstd encoder level based on the 1-9 compression level.
-func zstdLevel(level int) zstd.EncoderLevel {
-	switch {
-	case level <= 3:
-		return zstd.SpeedFastest
-	case level <= 6:
-		return zstd.SpeedDefault
-	default:
-		return zstd.SpeedBetterCompression
-	}
-}
-
-// CreateTarZstdStream takes a list of files and returns a reader that reads from a zstd-compressed tar archive.
-func CreateTarZstdStream(ctx context.Context, filepaths []string) io.ReadCloser {
-	pr, pw := io.Pipe()
-
-	go func() {
-		var pipeErr error
-
-		closePipeWithError := func(err error) {
-			if pipeErr != nil {
-				return
-			}
-
-			pipeErr = err
-			pw.CloseWithError(err)
-		}
-
-		defer func() {
-			if pipeErr == nil {
-				pw.Close()
-			}
-		}()
-
-		zw, err := zstd.NewWriter(pw, zstd.WithEncoderLevel(zstdLevel(C.CompressionLevel)))
-		if err != nil {
-			closePipeWithError(fmt.Errorf("zstd writer error: %w", err))
-
-			return
-		}
-
-		defer func() {
-			if pipeErr == nil {
-				if err := zw.Close(); err != nil {
-					if !isPipeClosedError(err) {
-						log.Errorf("Failed to close zstd writer: %v", err)
-					}
-				}
-			}
-		}()
-
-		tw := tar.NewWriter(zw)
-
-		defer func() {
-			if pipeErr == nil {
-				if err := tw.Close(); err != nil {
-					if !isPipeClosedError(err) {
-						log.Errorf("Failed to close tar writer: %v", err)
-					}
-				}
-			}
-		}()
-
-		for _, fpath := range filepaths {
-			var (
-				err        error
-				linkTarget string
-			)
-
-			select {
-			case <-ctx.Done():
-				closePipeWithError(ctx.Err())
-
-				return
-
-			default:
-			}
-
-			log.Debugf("Adding file: %s", fpath)
-
-			st, err := os.Lstat(fpath)
-
-			if err != nil {
-				log.Warnf("Skipping %s: %v", fpath, err)
-
-				continue
-			}
-
-			if st.Mode()&os.ModeSymlink != 0 {
-				linkTarget, err = os.Readlink(fpath)
-
-				if err != nil {
-					log.Warnf("Skipping %s: readlink failed: %v", fpath, err)
-
-					continue
-				}
-			}
-
-			if st.Mode()&os.ModeSocket != 0 {
-				log.Debugf("Skipping socket: %s", fpath)
-
-				continue
-			}
-
-			header, err := tar.FileInfoHeader(st, linkTarget)
-
-			if err != nil {
-				log.Warnf("Skipping %s: tar header failed: %v", fpath, err)
-
-				continue
-			}
-
-			header.Format = tar.FormatGNU
-			header.Name = strings.TrimPrefix(filepath.ToSlash(filepath.Clean(fpath)), "/")
-
-			mode := st.Mode()
-
-			if err := writeWithContext(ctx, func() error {
-				return tw.WriteHeader(header)
-			}); err != nil {
-				closePipeWithError(err)
-
-				return
-			}
-
-			if mode.IsRegular() {
-				f, err := os.Open(fpath)
-				if err != nil {
-					log.Warnf("Skipping %s: open failed: %v", fpath, err)
-
-					continue
-				}
-
-				err = copyWithContext(ctx, tw, f)
-
-				if e := f.Close(); e != nil {
-					log.Warnf("Failed to close file %s: %v", fpath, e)
-				}
-
-				if err != nil {
-					log.Warnf("Skipping %s: copy failed: %v", fpath, err)
-
-					continue
-				}
-			}
-		}
-	}()
-
-	return pr
-}
-
-// CreateTarLz4Stream takes a list of files and returns a reader that reads from an lz4-compressed tar archive.
-func CreateTarLz4Stream(ctx context.Context, filepaths []string) io.ReadCloser {
-	pr, pw := io.Pipe()
-
-	go func() {
-		var pipeErr error
-
-		closePipeWithError := func(err error) {
-			if pipeErr != nil {
-				return
-			}
-
-			pipeErr = err
-			pw.CloseWithError(err)
-		}
-
-		defer func() {
-			if pipeErr == nil {
-				pw.Close()
-			}
-		}()
-
-		lw := lz4.NewWriter(pw)
-
-		defer func() {
-			if pipeErr == nil {
-				if err := lw.Close(); err != nil {
-					if !isPipeClosedError(err) {
-						log.Errorf("Failed to close lz4 writer: %v", err)
-					}
-				}
-			}
-		}()
-
-		tw := tar.NewWriter(lw)
-
-		defer func() {
-			if pipeErr == nil {
-				if err := tw.Close(); err != nil {
-					if !isPipeClosedError(err) {
-						log.Errorf("Failed to close tar writer: %v", err)
-					}
-				}
-			}
-		}()
-
-		for _, fpath := range filepaths {
-			var (
-				err        error
-				linkTarget string
-			)
-
-			select {
-			case <-ctx.Done():
-				closePipeWithError(ctx.Err())
-
-				return
-
-			default:
-			}
-
-			log.Debugf("Adding file: %s", fpath)
-
-			st, err := os.Lstat(fpath)
-
-			if err != nil {
-				log.Warnf("Skipping %s: %v", fpath, err)
-
-				continue
-			}
-
-			if st.Mode()&os.ModeSymlink != 0 {
-				linkTarget, err = os.Readlink(fpath)
-
-				if err != nil {
-					log.Warnf("Skipping %s: readlink failed: %v", fpath, err)
-
-					continue
-				}
-			}
-
-			if st.Mode()&os.ModeSocket != 0 {
-				log.Debugf("Skipping socket: %s", fpath)
-
-				continue
-			}
-
-			header, err := tar.FileInfoHeader(st, linkTarget)
-
-			if err != nil {
-				log.Warnf("Skipping %s: tar header failed: %v", fpath, err)
-
-				continue
-			}
-
-			header.Format = tar.FormatGNU
-			header.Name = strings.TrimPrefix(filepath.ToSlash(filepath.Clean(fpath)), "/")
-
-			mode := st.Mode()
-
-			if err := writeWithContext(ctx, func() error {
-				return tw.WriteHeader(header)
-			}); err != nil {
-				closePipeWithError(err)
-
-				return
-			}
-
-			if mode.IsRegular() {
-				f, err := os.Open(fpath)
-				if err != nil {
-					log.Warnf("Skipping %s: open failed: %v", fpath, err)
-
-					continue
-				}
-
-				err = copyWithContext(ctx, tw, f)
-
-				if e := f.Close(); e != nil {
-					log.Warnf("Failed to close file %s: %v", fpath, e)
-				}
-
-				if err != nil {
-					log.Warnf("Skipping %s: copy failed: %v", fpath, err)
-
-					continue
-				}
-			}
-		}
-	}()
-
-	return pr
-}
-
-// CreateTarXzStream takes a list of files and returns a reader that reads from an xz-compressed tar archive.
-func CreateTarXzStream(ctx context.Context, filepaths []string) io.ReadCloser {
-	pr, pw := io.Pipe()
-
-	go func() {
-		var pipeErr error
-
-		closePipeWithError := func(err error) {
-			if pipeErr != nil {
-				return
-			}
-
-			pipeErr = err
-			pw.CloseWithError(err)
-		}
-
-		defer func() {
-			if pipeErr == nil {
-				pw.Close()
-			}
-		}()
-
-		xw, err := xz.NewWriter(pw)
-		if err != nil {
-			closePipeWithError(fmt.Errorf("xz writer error: %w", err))
-
-			return
-		}
-
-		defer func() {
-			if pipeErr == nil {
-				if err := xw.Close(); err != nil {
-					if !isPipeClosedError(err) {
-						log.Errorf("Failed to close xz writer: %v", err)
-					}
-				}
-			}
-		}()
-
-		tw := tar.NewWriter(xw)
-
-		defer func() {
-			if pipeErr == nil {
-				if err := tw.Close(); err != nil {
-					if !isPipeClosedError(err) {
-						log.Errorf("Failed to close tar writer: %v", err)
-					}
-				}
-			}
-		}()
-
-		for _, fpath := range filepaths {
-			var (
-				err        error
-				linkTarget string
-			)
-
-			select {
-			case <-ctx.Done():
-				closePipeWithError(ctx.Err())
-
-				return
-
-			default:
-			}
-
-			log.Debugf("Adding file: %s", fpath)
-
-			st, err := os.Lstat(fpath)
-
-			if err != nil {
-				log.Warnf("Skipping %s: %v", fpath, err)
-
-				continue
-			}
-
-			if st.Mode()&os.ModeSymlink != 0 {
-				linkTarget, err = os.Readlink(fpath)
-
-				if err != nil {
-					log.Warnf("Skipping %s: readlink failed: %v", fpath, err)
-
-					continue
-				}
-			}
-
-			if st.Mode()&os.ModeSocket != 0 {
-				log.Debugf("Skipping socket: %s", fpath)
-
-				continue
-			}
-
-			header, err := tar.FileInfoHeader(st, linkTarget)
-
-			if err != nil {
-				log.Warnf("Skipping %s: tar header failed: %v", fpath, err)
-
-				continue
-			}
-
-			header.Format = tar.FormatGNU
-			header.Name = strings.TrimPrefix(filepath.ToSlash(filepath.Clean(fpath)), "/")
-
-			mode := st.Mode()
-
-			if err := writeWithContext(ctx, func() error {
-				return tw.WriteHeader(header)
-			}); err != nil {
-				closePipeWithError(err)
-
-				return
-			}
-
-			if mode.IsRegular() {
-				f, err := os.Open(fpath)
-				if err != nil {
-					log.Warnf("Skipping %s: open failed: %v", fpath, err)
-
-					continue
-				}
-
-				err = copyWithContext(ctx, tw, f)
-
-				if e := f.Close(); e != nil {
-					log.Warnf("Failed to close file %s: %v", fpath, e)
-				}
-
-				if err != nil {
-					log.Warnf("Skipping %s: copy failed: %v", fpath, err)
-
-					continue
-				}
-			}
-		}
-	}()
-
-	return pr
-}
-
-// CreateTarPgzipStream takes a list of files and returns a reader that reads from a parallel gzip-compressed tar archive.
-func CreateTarPgzipStream(ctx context.Context, filepaths []string) io.ReadCloser {
-	pr, pw := io.Pipe()
-
-	go func() {
-		var pipeErr error
-
-		closePipeWithError := func(err error) {
-			if pipeErr != nil {
-				return
-			}
-
-			pipeErr = err
-			pw.CloseWithError(err)
-		}
-
-		defer func() {
-			if pipeErr == nil {
-				pw.Close()
-			}
-		}()
-
-		gw, err := pgzip.NewWriterLevel(pw, C.CompressionLevel)
-		if err != nil {
-			closePipeWithError(fmt.Errorf("pgzip writer error: %w", err))
-
-			return
-		}
-
-		defer func() {
-			if pipeErr == nil {
-				if err := gw.Close(); err != nil {
-					if !isPipeClosedError(err) {
-						log.Errorf("Failed to close pgzip writer: %v", err)
-					}
-				}
-			}
-		}()
-
-		tw := tar.NewWriter(gw)
-
-		defer func() {
-			if pipeErr == nil {
-				if err := tw.Close(); err != nil {
-					if !isPipeClosedError(err) {
-						log.Errorf("Failed to close tar writer: %v", err)
-					}
-				}
-			}
-		}()
-
-		for _, fpath := range filepaths {
-			var (
-				err        error
-				linkTarget string
-			)
-
-			select {
-			case <-ctx.Done():
-				closePipeWithError(ctx.Err())
-
-				return
-
-			default:
-			}
-
-			log.Debugf("Adding file: %s", fpath)
-
-			st, err := os.Lstat(fpath)
-
-			if err != nil {
-				log.Warnf("Skipping %s: %v", fpath, err)
-
-				continue
-			}
-
-			if st.Mode()&os.ModeSymlink != 0 {
-				linkTarget, err = os.Readlink(fpath)
-
-				if err != nil {
-					log.Warnf("Skipping %s: readlink failed: %v", fpath, err)
-
-					continue
-				}
-			}
-
-			if st.Mode()&os.ModeSocket != 0 {
-				log.Debugf("Skipping socket: %s", fpath)
-
-				continue
-			}
-
-			header, err := tar.FileInfoHeader(st, linkTarget)
-
-			if err != nil {
-				log.Warnf("Skipping %s: tar header failed: %v", fpath, err)
-
-				continue
-			}
-
-			header.Format = tar.FormatGNU
-			header.Name = strings.TrimPrefix(filepath.ToSlash(filepath.Clean(fpath)), "/")
-
-			mode := st.Mode()
-
-			if err := writeWithContext(ctx, func() error {
-				return tw.WriteHeader(header)
-			}); err != nil {
-				closePipeWithError(err)
-
-				return
-			}
-
-			if mode.IsRegular() {
-				f, err := os.Open(fpath)
-				if err != nil {
-					log.Warnf("Skipping %s: open failed: %v", fpath, err)
-
-					continue
-				}
-
-				err = copyWithContext(ctx, tw, f)
-
-				if e := f.Close(); e != nil {
-					log.Warnf("Failed to close file %s: %v", fpath, e)
-				}
-
-				if err != nil {
-					log.Warnf("Skipping %s: copy failed: %v", fpath, err)
-
-					continue
-				}
-			}
-		}
-	}()
-
-	return pr
-}
-
-// isPipeClosedError checks if the error is due to a closed pipe.
-func isPipeClosedError(err error) bool {
-	if err == nil {
-		return false
-	}
-
-	// Check for the specific pipe closed error message.
-	return err.Error() == "io: read/write on closed pipe" ||
-		err.Error() == "io: read on closed pipe"
-}
-
-/* vim: setlocal ft=go noet ai ts=4 sw=4 sts=4: */

internal/backer/xz.go

@@ -0,0 +1,17 @@
+package backer
+
+import (
+	"context"
+	"io"
+
+	"github.com/ulikunitz/xz"
+)
+
+// CreateTarXzStream takes a list of files and returns a reader that reads from an xz-compressed tar archive.
+func CreateTarXzStream(ctx context.Context, filepaths []string) io.ReadCloser {
+	return createArchiveStream(ctx, filepaths, func(w io.Writer) (io.WriteCloser, error) {
+		return xz.NewWriter(w)
+	})
+}
+
+/* vim: setlocal ft=go noet ai ts=4 sw=4 sts=4: */

internal/backer/zstd.go

@@ -0,0 +1,29 @@
+package backer
+
+import (
+	"context"
+	"io"
+
+	"github.com/klauspost/compress/zstd"
+)
+
+// CreateTarZstdStream takes a list of files and returns a reader that reads from a zstd-compressed tar archive.
+func CreateTarZstdStream(ctx context.Context, filepaths []string) io.ReadCloser {
+	return createArchiveStream(ctx, filepaths, func(w io.Writer) (io.WriteCloser, error) {
+		return zstd.NewWriter(w, zstd.WithEncoderLevel(zstdLevel(C.CompressionLevel)))
+	})
+}
+
+// zstdLevel returns the zstd encoder level based on the 1-9 compression level.
+func zstdLevel(level int) zstd.EncoderLevel {
+	switch {
+	case level <= 3:
+		return zstd.SpeedFastest
+	case level <= 6:
+		return zstd.SpeedDefault
+	default:
+		return zstd.SpeedBetterCompression
+	}
+}
+
+/* vim: setlocal ft=go noet ai ts=4 sw=4 sts=4: */