Add scheduled rsync copy jobs

2026-06-15 17:15:27 +02:00
parent 2abfc8eb26
commit 6c795fd97d
15 changed files with 524 additions and 59 deletions
@@ -0,0 +1,228 @@
+package rsync
+
+import (
+	"bufio"
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"os"
+	"os/exec"
+	"regexp"
+	"strconv"
+	"strings"
+	"sync"
+	"syscall"
+
+	"git.casaderoll.de/michael/urbm/internal/model"
+)
+
+type Runner struct {
+	Binary    string
+	mu        sync.Mutex
+	processes map[string]*os.Process
+}
+
+type Progress struct {
+	Percent     float64
+	Bytes       int64
+	Files       int64
+	CurrentFile string
+}
+
+type Summary struct {
+	Bytes int64
+	Files int64
+}
+
+var (
+	progressPattern = regexp.MustCompile(`^\s*([0-9,]+)\s+([0-9]+)%`)
+	filesPattern    = regexp.MustCompile(`^Number of regular files transferred:\s*([0-9,]+)`)
+	bytesPattern    = regexp.MustCompile(`^Total transferred file size:\s*([0-9,]+) bytes`)
+)
+
+func (r *Runner) Run(ctx context.Context, runID string, job model.Job, callback func(Progress)) (Summary, error) {
+	args := Arguments(job)
+	cmd := exec.CommandContext(ctx, r.Binary, args...)
+	cmd.Env = append(os.Environ(), "LC_ALL=C", "LANG=C")
+	cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
+	cmd.Cancel = func() error {
+		if cmd.Process == nil {
+			return os.ErrProcessDone
+		}
+		if err := syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL); err != nil && !errors.Is(err, syscall.ESRCH) {
+			return err
+		}
+		return nil
+	}
+	stdout, err := cmd.StdoutPipe()
+	if err != nil {
+		return Summary{}, err
+	}
+	stderr, err := cmd.StderrPipe()
+	if err != nil {
+		return Summary{}, err
+	}
+	if err := cmd.Start(); err != nil {
+		if errors.Is(err, exec.ErrNotFound) {
+			return Summary{}, fmt.Errorf("environment: rsync binary not found: %w", err)
+		}
+		return Summary{}, err
+	}
+	unregister := r.register(runID, cmd.Process)
+	defer unregister()
+
+	var summary Summary
+	done := make(chan struct{})
+	go func() {
+		scanOutput(stdout, func(line string) {
+			if value, ok := parseCount(filesPattern, line); ok {
+				summary.Files = value
+			}
+			if value, ok := parseCount(bytesPattern, line); ok {
+				summary.Bytes = value
+			}
+			if callback != nil {
+				if progress, ok := parseProgress(line); ok {
+					callback(progress)
+				} else if strings.HasPrefix(line, "FILE|") {
+					callback(Progress{CurrentFile: strings.TrimPrefix(line, "FILE|")})
+				}
+			}
+		})
+		close(done)
+	}()
+	errBytes, _ := io.ReadAll(io.LimitReader(stderr, 1024*1024))
+	err = cmd.Wait()
+	<-done
+	if err != nil {
+		if errors.Is(ctx.Err(), context.Canceled) {
+			return Summary{}, fmt.Errorf("cancelled: %w", ctx.Err())
+		}
+		message := strings.TrimSpace(string(errBytes))
+		if message == "" {
+			message = err.Error()
+		}
+		return Summary{}, fmt.Errorf("rsync: %s", message)
+	}
+	return summary, nil
+}
+
+func Arguments(job model.Job) []string {
+	options := job.Rsync
+	args := []string{"--recursive", "--human-readable", "--info=progress2", "--stats", "--partial", "--out-format=FILE|%n"}
+	if !options.Overwrite {
+		args = append(args, "--ignore-existing")
+	}
+	if options.Delete {
+		args = append(args, "--delete-delay")
+	}
+	if options.PreservePermissions {
+		args = append(args, "--perms")
+	}
+	if options.PreserveOwner {
+		args = append(args, "--owner")
+	}
+	if options.PreserveGroup {
+		args = append(args, "--group")
+	}
+	if options.PreserveTimes {
+		args = append(args, "--times")
+	}
+	if options.PreserveLinks {
+		args = append(args, "--links")
+	}
+	if options.PreserveACLs {
+		args = append(args, "--acls")
+	}
+	if options.PreserveXattrs {
+		args = append(args, "--xattrs")
+	}
+	if options.Checksum {
+		args = append(args, "--checksum")
+	}
+	if options.DryRun {
+		args = append(args, "--dry-run")
+	}
+	for _, exclude := range job.Excludes {
+		args = append(args, "--exclude", exclude)
+	}
+	args = append(args, "--")
+	for _, source := range job.Sources {
+		args = append(args, source.Path)
+	}
+	args = append(args, strings.TrimRight(options.Target, "/")+"/")
+	return args
+}
+
+func (r *Runner) Pause(runID string) error  { return r.signal(runID, syscall.SIGSTOP) }
+func (r *Runner) Resume(runID string) error { return r.signal(runID, syscall.SIGCONT) }
+
+func (r *Runner) signal(runID string, signal syscall.Signal) error {
+	r.mu.Lock()
+	process := r.processes[runID]
+	r.mu.Unlock()
+	if process == nil {
+		return errors.New("rsync process is not active yet")
+	}
+	if err := syscall.Kill(-process.Pid, signal); err != nil {
+		return fmt.Errorf("signal rsync process group: %w", err)
+	}
+	return nil
+}
+
+func (r *Runner) register(runID string, process *os.Process) func() {
+	r.mu.Lock()
+	if r.processes == nil {
+		r.processes = make(map[string]*os.Process)
+	}
+	r.processes[runID] = process
+	r.mu.Unlock()
+	return func() {
+		r.mu.Lock()
+		delete(r.processes, runID)
+		r.mu.Unlock()
+	}
+}
+
+func scanOutput(reader io.Reader, callback func(string)) {
+	scanner := bufio.NewScanner(reader)
+	scanner.Split(splitLinesAndCarriageReturns)
+	scanner.Buffer(make([]byte, 64*1024), 2*1024*1024)
+	for scanner.Scan() {
+		if line := strings.TrimSpace(scanner.Text()); line != "" {
+			callback(line)
+		}
+	}
+}
+
+func splitLinesAndCarriageReturns(data []byte, atEOF bool) (advance int, token []byte, err error) {
+	for i, value := range data {
+		if value == '\n' || value == '\r' {
+			return i + 1, data[:i], nil
+		}
+	}
+	if atEOF && len(data) > 0 {
+		return len(data), data, nil
+	}
+	return 0, nil, nil
+}
+
+func parseProgress(line string) (Progress, bool) {
+	match := progressPattern.FindStringSubmatch(line)
+	if len(match) != 3 {
+		return Progress{}, false
+	}
+	bytes, _ := strconv.ParseInt(strings.ReplaceAll(match[1], ",", ""), 10, 64)
+	percent, _ := strconv.ParseFloat(match[2], 64)
+	return Progress{Bytes: bytes, Percent: percent}, true
+}
+
+func parseCount(pattern *regexp.Regexp, line string) (int64, bool) {
+	match := pattern.FindStringSubmatch(line)
+	if len(match) != 2 {
+		return 0, false
+	}
+	value, err := strconv.ParseInt(strings.ReplaceAll(match[1], ",", ""), 10, 64)
+	return value, err == nil
+}