Add Restic read concurrency setting

internal/model/model.go

@@ -47,24 +47,25 @@ type Settings struct {
 }
 
 type Job struct {
-	SchemaVersion int          `json:"schemaVersion"`
-	ID            string       `json:"id"`
-	Name          string       `json:"name"`
-	Type          JobType      `json:"type"`
-	Enabled       bool         `json:"enabled"`
-	RepositoryID  string       `json:"repositoryId"`
-	Sources       []Source     `json:"sources"`
-	Schedule      Schedule     `json:"schedule"`
-	Consistency   Consistency  `json:"consistency"`
-	Compression   string       `json:"compression"`
-	CPUCores      int          `json:"cpuCores,omitempty"`
-	Excludes      []string     `json:"excludes,omitempty"`
-	Tags          []string     `json:"tags,omitempty"`
-	FlashImage    bool         `json:"flashImage,omitempty"`
-	Retention     Retention    `json:"retention"`
-	NotifyOn      []string     `json:"notifyOn,omitempty"`
-	ShutdownSecs  int          `json:"shutdownTimeoutSeconds"`
-	Rsync         RsyncOptions `json:"rsync,omitempty"`
+	SchemaVersion   int          `json:"schemaVersion"`
+	ID              string       `json:"id"`
+	Name            string       `json:"name"`
+	Type            JobType      `json:"type"`
+	Enabled         bool         `json:"enabled"`
+	RepositoryID    string       `json:"repositoryId"`
+	Sources         []Source     `json:"sources"`
+	Schedule        Schedule     `json:"schedule"`
+	Consistency     Consistency  `json:"consistency"`
+	Compression     string       `json:"compression"`
+	CPUCores        int          `json:"cpuCores,omitempty"`
+	ReadConcurrency int          `json:"readConcurrency,omitempty"`
+	Excludes        []string     `json:"excludes,omitempty"`
+	Tags            []string     `json:"tags,omitempty"`
+	FlashImage      bool         `json:"flashImage,omitempty"`
+	Retention       Retention    `json:"retention"`
+	NotifyOn        []string     `json:"notifyOn,omitempty"`
+	ShutdownSecs    int          `json:"shutdownTimeoutSeconds"`
+	Rsync           RsyncOptions `json:"rsync,omitempty"`
 }
 
 type RsyncOptions struct {

internal/model/validate.go

@@ -138,6 +138,9 @@ func ValidateJob(j Job) error {
 	if j.CPUCores < 0 || j.CPUCores > 256 {
 		return errors.New("cpuCores must be between 0 and 256")
 	}
+	if j.ReadConcurrency < 0 || j.ReadConcurrency > 64 {
+		return errors.New("readConcurrency must be between 0 and 64")
+	}
 	if j.Consistency.Mode != "live" && j.Consistency.Mode != "stop" {
 		return errors.New("consistency mode must be live or stop")
 	}

internal/model/validate_test.go

@@ -92,6 +92,18 @@ func TestBackupCPUCoresRange(t *testing.T) {
 	}
 }
 
+func TestBackupReadConcurrencyRange(t *testing.T) {
+	job := Job{SchemaVersion: 1, ID: "job", Name: "Share", Type: JobShare, RepositoryID: "repo", Sources: []Source{{Path: "/mnt/user/data"}}, Consistency: Consistency{Mode: "live"}, Compression: "auto", Retention: Retention{KeepWithinDays: 30}}
+	job.ReadConcurrency = 8
+	if err := ValidateJob(job); err != nil {
+		t.Fatalf("valid read concurrency rejected: %v", err)
+	}
+	job.ReadConcurrency = 65
+	if err := ValidateJob(job); err == nil {
+		t.Fatal("excessive read concurrency accepted")
+	}
+}
+
 func TestRsyncJobDoesNotRequireRepository(t *testing.T) {
 	c := DefaultConfig()
 	c.Jobs = []Job{{SchemaVersion: 1, ID: "rsync-job", Name: "Mirror", Type: JobRsync, Enabled: true, Sources: []Source{{Path: "/mnt/user/data"}}, Rsync: RsyncOptions{Target: "/mnt/remotes/backup", Overwrite: true}}}

internal/restic/restic.go

@@ -121,6 +121,9 @@ func (r *Runner) TestPassword(ctx context.Context, repo model.Repository, passwo
 
 func (r *Runner) Backup(ctx context.Context, repo model.Repository, job model.Job, sources []string, progress ProgressCallback) (Summary, error) {
 	args := []string{"backup", "--json", "--compression", job.Compression}
+	if job.ReadConcurrency > 0 {
+		args = append(args, "--read-concurrency", fmt.Sprint(job.ReadConcurrency))
+	}
 	for _, tag := range append([]string{"urbm", "job:" + job.ID}, job.Tags...) {
 		args = append(args, "--tag", tag)
 	}

internal/restic/restic_test.go

@@ -110,6 +110,29 @@ func TestBackupLimitsCPUForThisJob(t *testing.T) {
 	}
 }
 
+func TestBackupSetsReadConcurrencyForThisJob(t *testing.T) {
+	dir := t.TempDir()
+	argsPath := filepath.Join(dir, "args")
+	script := filepath.Join(dir, "restic")
+	body := fmt.Sprintf("#!/bin/sh\nprintf '%%s\\n' \"$@\" > '%s'\nprintf '%%s\\n' '{\"message_type\":\"summary\",\"snapshot_id\":\"read123\"}'\n", argsPath)
+	if err := os.WriteFile(script, []byte(body), 0700); err != nil {
+		t.Fatal(err)
+	}
+	runner := &Runner{Binary: script, RuntimeDir: dir, Secrets: fakeSecrets{"password": "secret"}}
+	repo := model.Repository{Type: model.RepositoryLocal, Location: "/repo", PasswordRef: "password"}
+	job := model.Job{ID: "job", Compression: "auto", ReadConcurrency: 8}
+	if _, err := runner.Backup(context.Background(), repo, job, []string{"/source"}, nil); err != nil {
+		t.Fatal(err)
+	}
+	args, err := os.ReadFile(argsPath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !strings.Contains(string(args), "--read-concurrency\n8\n") {
+		t.Fatalf("read concurrency missing from arguments: %s", args)
+	}
+}
+
 func TestBackupImageStreamsRawDeviceWithStableFilename(t *testing.T) {
 	dir := t.TempDir()
 	argsPath := filepath.Join(dir, "args")