diff --git a/pkg/restic/outputs.go b/pkg/restic/outputs.go index 723505dd..ed4e98d5 100644 --- a/pkg/restic/outputs.go +++ b/pkg/restic/outputs.go @@ -125,56 +125,105 @@ func (b *BackupProgressEntry) Validate() error { return nil } -// readBackupProgressEntries returns the summary event or an error if the command failed. -func readBackupProgressEntries(output io.Reader, logger io.Writer, callback func(event *BackupProgressEntry)) (*BackupProgressEntry, error) { +func (b *BackupProgressEntry) IsError() bool { + return b.MessageType == "error" +} + +func (b *BackupProgressEntry) IsSummary() bool { + return b.MessageType == "summary" +} + +type RestoreProgressEntry struct { + MessageType string `json:"message_type"` // "summary" or "status" + SecondsElapsed float64 `json:"seconds_elapsed"` + TotalBytes int64 `json:"total_bytes"` + BytesRestored int64 `json:"bytes_restored"` + TotalFiles int64 `json:"total_files"` + FilesRestored int64 `json:"files_restored"` + PercentDone float64 `json:"percent_done"` +} + +func (e *RestoreProgressEntry) Validate() error { + if e.MessageType != "summary" && e.MessageType != "status" { + return fmt.Errorf("message_type must be 'summary' or 'status', got %v", e.MessageType) + } + return nil +} + +func (r *RestoreProgressEntry) IsError() bool { + return r.MessageType == "error" +} + +func (r *RestoreProgressEntry) IsSummary() bool { + return r.MessageType == "summary" +} + +type ProgressEntryValidator interface { + Validate() error + IsError() bool + IsSummary() bool +} + +// processProgressOutput handles common JSON output processing logic with proper type safety +func processProgressOutput[T ProgressEntryValidator]( + output io.Reader, + logger io.Writer, + callback func(T)) (T, error) { + scanner := bufio.NewScanner(output) scanner.Split(bufio.ScanLines) nonJSONOutput := bytes.NewBuffer(nil) + var captureNonJSON io.Writer = nonJSONOutput + if logger != nil { + captureNonJSON = io.MultiWriter(nonJSONOutput, logger) + } - var summary *BackupProgressEntry + var summary *T + var nullT T - // remaining events are parsed as JSON for scanner.Scan() { - var event BackupProgressEntry - if err := json.Unmarshal(scanner.Bytes(), &event); err != nil { - nonJSONOutput.Write(scanner.Bytes()) - if logger != nil { - logger.Write(scanner.Bytes()) - logger.Write([]byte("\n")) - } + line := scanner.Bytes() + var event T + + if err := json.Unmarshal(line, &event); err != nil { + captureNonJSON.Write(line) + captureNonJSON.Write([]byte("\n")) continue } + if err := event.Validate(); err != nil { - nonJSONOutput.Write(scanner.Bytes()) - if logger != nil { - logger.Write(scanner.Bytes()) - logger.Write([]byte("\n")) - } + captureNonJSON.Write(line) + captureNonJSON.Write([]byte("\n")) continue } - if event.MessageType == "error" && logger != nil { - logger.Write(scanner.Bytes()) - logger.Write([]byte("\n")) + + if event.IsError() && logger != nil { + captureNonJSON.Write(line) + captureNonJSON.Write([]byte("\n")) } + if callback != nil { - callback(&event) + callback(event) } - if event.MessageType == "summary" { - if logger != nil { - logger.Write(scanner.Bytes()) - logger.Write([]byte("\n")) - } - summary = &event + + if event.IsSummary() { + captureNonJSON.Write(line) + captureNonJSON.Write([]byte("\n")) + eventCopy := event // Make a copy to avoid issues with loop variable + summary = &eventCopy } } + if err := scanner.Err(); err != nil { - return summary, newErrorWithOutput(err, nonJSONOutput.String()) + return nullT, newErrorWithOutput(err, nonJSONOutput.String()) } + if summary == nil { - return nil, newErrorWithOutput(errors.New("no summary event found"), nonJSONOutput.String()) + return nullT, newErrorWithOutput(errors.New("no summary event found"), nonJSONOutput.String()) } - return summary, nil + + return *summary, nil } type LsEntry struct { @@ -248,80 +297,6 @@ func (r *ForgetResult) Validate() error { return nil } -type RestoreProgressEntry struct { - MessageType string `json:"message_type"` // "summary" or "status" - SecondsElapsed float64 `json:"seconds_elapsed"` - TotalBytes int64 `json:"total_bytes"` - BytesRestored int64 `json:"bytes_restored"` - TotalFiles int64 `json:"total_files"` - FilesRestored int64 `json:"files_restored"` - PercentDone float64 `json:"percent_done"` -} - -func (e *RestoreProgressEntry) Validate() error { - if e.MessageType != "summary" && e.MessageType != "status" { - return fmt.Errorf("message_type must be 'summary' or 'status', got %v", e.MessageType) - } - return nil -} - -// readRestoreProgressEntries returns the summary event or an error if the command failed. -func readRestoreProgressEntries(output io.Reader, logger io.Writer, callback func(event *RestoreProgressEntry)) (*RestoreProgressEntry, error) { - scanner := bufio.NewScanner(output) - scanner.Split(bufio.ScanLines) - - nonJSONOutput := bytes.NewBuffer(nil) - - var summary *RestoreProgressEntry - - // remaining events are parsed as JSON - for scanner.Scan() { - var event RestoreProgressEntry - if err := json.Unmarshal(scanner.Bytes(), &event); err != nil { - nonJSONOutput.Write(scanner.Bytes()) - if logger != nil { - logger.Write(scanner.Bytes()) - logger.Write([]byte("\n")) - } - continue - } - if err := event.Validate(); err != nil { - // skip it. Best effort parsing, restic will return with a non-zero exit code if it fails. - nonJSONOutput.Write(scanner.Bytes()) - if logger != nil { - logger.Write(scanner.Bytes()) - logger.Write([]byte("\n")) - } - continue - } - - if event.MessageType == "error" && logger != nil { - logger.Write(scanner.Bytes()) - logger.Write([]byte("\n")) - } - if callback != nil { - callback(&event) - } - if event.MessageType == "summary" { - if logger != nil { - logger.Write(scanner.Bytes()) - logger.Write([]byte("\n")) - } - summary = &event - } - } - - if err := scanner.Err(); err != nil { - return summary, newErrorWithOutput(err, nonJSONOutput.String()) - } - - if summary == nil { - return nil, newErrorWithOutput(errors.New("no summary event found"), nonJSONOutput.String()) - } - - return summary, nil -} - func ValidateSnapshotId(id string) error { if len(id) != 64 { return fmt.Errorf("restic may be out of date (check with `restic self-upgrade`): snapshot ID must be 64 chars, got %v chars", len(id)) diff --git a/pkg/restic/outputs_test.go b/pkg/restic/outputs_test.go index af52d07a..062331a3 100644 --- a/pkg/restic/outputs_test.go +++ b/pkg/restic/outputs_test.go @@ -12,7 +12,7 @@ func TestReadBackupProgressEntries(t *testing.T) { b := bytes.NewBuffer([]byte(testInput)) - summary, err := readBackupProgressEntries(b, nil, func(event *BackupProgressEntry) { + summary, err := processProgressOutput[*BackupProgressEntry](b, nil, func(event *BackupProgressEntry) { t.Logf("event: %v", event) }) if err != nil { diff --git a/pkg/restic/restic.go b/pkg/restic/restic.go index 5ba2cd87..af8cf9d0 100644 --- a/pkg/restic/restic.go +++ b/pkg/restic/restic.go @@ -92,6 +92,35 @@ func (r *Repo) pipeCmdOutputToWriter(cmd *exec.Cmd, handlers ...io.Writer) { cmd.Stderr = mw } +// executeWithOutput runs a command and captures the output in a buffer +func (r *Repo) executeWithOutput(ctx context.Context, args []string, opts ...GenericOption) ([]byte, error) { + cmd := r.commandWithContext(ctx, args, opts...) + output := bytes.NewBuffer(nil) + r.pipeCmdOutputToWriter(cmd, output) + + err := cmd.Run() + if err != nil { + return output.Bytes(), newCmdError(ctx, cmd, newErrorWithOutput(err, output.String())) + } + + return output.Bytes(), nil +} + +// executeWithJSONOutput runs a command and parses its JSON output +func (r *Repo) executeWithJSONOutput(ctx context.Context, args []string, result interface{}, opts ...GenericOption) error { + output, err := r.executeWithOutput(ctx, args, opts...) + if err != nil { + return err + } + + if err := json.Unmarshal(output, result); err != nil { + return newCmdError(ctx, r.commandWithContext(ctx, args), + newErrorWithOutput(fmt.Errorf("command output is not valid JSON: %w", err), string(output))) + } + + return nil +} + // Exists checks if the repository exists. // Returns true if exists, false if it does not exist OR an access error occurred. func (r *Repo) Exists(ctx context.Context, opts ...GenericOption) error { @@ -143,10 +172,7 @@ func (r *Repo) init(ctx context.Context, opts ...GenericOption) error { } func (r *Repo) Init(ctx context.Context, opts ...GenericOption) error { - if err := r.init(ctx, opts...); err != nil && !errors.Is(err, errAlreadyInitialized) { - return fmt.Errorf("init failed: %w", err) - } - return nil + return r.init(ctx, opts...) } func (r *Repo) Config(ctx context.Context, opts ...GenericOption) (RepoConfig, error) { @@ -156,6 +182,56 @@ func (r *Repo) Config(ctx context.Context, opts ...GenericOption) (RepoConfig, e return r.repoConfig, nil } +type cmdRunnerWithProgress[T ProgressEntryValidator] struct { + repo *Repo + callback func(T) + failureErr error +} + +func (cr *cmdRunnerWithProgress[T]) Run(ctx context.Context, args []string, opts ...GenericOption) (T, error) { + logger := LoggerFromContext(ctx) + cmdCtx, cancel := context.WithCancel(ctx) + defer cancel() + cmdCtx = ContextWithLogger(cmdCtx, nil) // ensure no logger is used + cmd := cr.repo.commandWithContext(cmdCtx, args, opts...) + + // Ensure the command is logged since we're overriding the logger + if logger != nil { + fmt.Fprintf(logger, "command: %q\n", cmd) + } + + buf := buffer.New(32 * 1024) // 32KB IO buffer for the realtime event parsing + reader, writer := nio.Pipe(buf) + cr.repo.pipeCmdOutputToWriter(cmd, writer) + + var readErr error + var summary T + var wg sync.WaitGroup + + wg.Add(1) + go func() { + defer wg.Done() + result, err := processProgressOutput[T](reader, logger, cr.callback) + summary = result + if err != nil { + readErr = fmt.Errorf("processing command output: %w", err) + } + }() + + cmdErr := cmd.Run() + writer.Close() + wg.Wait() + + if cmdErr != nil || readErr != nil { + if cmdErr != nil { + cmdErr = cr.repo.handleExitError(cmdErr, cr.failureErr) + } + return summary, newCmdError(ctx, cmd, errors.Join(cmdErr, readErr)) + } + + return summary, nil +} + func (r *Repo) Backup(ctx context.Context, paths []string, progressCallback func(*BackupProgressEntry), opts ...GenericOption) (*BackupProgressEntry, error) { for _, p := range paths { if _, err := os.Stat(p); err != nil { @@ -167,114 +243,43 @@ func (r *Repo) Backup(ctx context.Context, paths []string, progressCallback func args = append(args, paths...) opts = append(slices.Clone(opts), WithEnv("RESTIC_PROGRESS_FPS=2")) - logger := LoggerFromContext(ctx) - cmdCtx, cancel := context.WithCancel(ctx) - cmdCtx = ContextWithLogger(cmdCtx, nil) // ensure no logger is used - cmd := r.commandWithContext(cmdCtx, args, opts...) - - // Ensure the command is logged since we're overriding the logger - if logger != nil { - fmt.Fprintf(logger, "command: %q\n", cmd) + cr := cmdRunnerWithProgress[*BackupProgressEntry]{ + repo: r, + callback: progressCallback, + failureErr: ErrBackupFailed, } - - buf := buffer.New(32 * 1024) // 32KB IO buffer for the realtime event parsing - reader, writer := nio.Pipe(buf) - r.pipeCmdOutputToWriter(cmd, writer) - - var readErr error - var summary *BackupProgressEntry - var wg sync.WaitGroup - wg.Add(1) - go func() { - defer wg.Done() - defer cancel() - var err error - summary, err = readBackupProgressEntries(reader, logger, progressCallback) - if err != nil { - readErr = fmt.Errorf("processing command output: %w", err) - } - }() - - cmdErr := cmd.Run() - writer.Close() - wg.Wait() - - if cmdErr != nil || readErr != nil { - if cmdErr != nil { - var exitErr *exec.ExitError - if errors.As(cmdErr, &exitErr) { - if exitErr.ExitCode() == 3 { - cmdErr = ErrPartialBackup - } else { - cmdErr = fmt.Errorf("exit code %d: %w", exitErr.ExitCode(), ErrBackupFailed) - } - } - } - return summary, newCmdError(ctx, cmd, errors.Join(cmdErr, readErr)) - } - return summary, nil + return cr.Run(ctx, args, opts...) } func (r *Repo) Restore(ctx context.Context, snapshot string, callback func(*RestoreProgressEntry), opts ...GenericOption) (*RestoreProgressEntry, error) { opts = append(slices.Clone(opts), WithEnv("RESTIC_PROGRESS_FPS=2")) + args := []string{"restore", "--json", snapshot} - logger := LoggerFromContext(ctx) - cmdCtx, cancel := context.WithCancel(ctx) - cmdCtx = ContextWithLogger(cmdCtx, nil) // ensure no logger is used - cmd := r.commandWithContext(cmdCtx, []string{"restore", "--json", snapshot}, opts...) - if logger != nil { - fmt.Fprintf(logger, "command: %v %v\n", cmd.Path, strings.Join(cmd.Args, " ")) + cr := cmdRunnerWithProgress[*RestoreProgressEntry]{ + repo: r, + callback: callback, + failureErr: ErrRestoreFailed, } - buf := buffer.New(32 * 1024) // 32KB IO buffer for the realtime event parsing - reader, writer := nio.Pipe(buf) - r.pipeCmdOutputToWriter(cmd, writer) + return cr.Run(ctx, args, opts...) +} - var readErr error - var summary *RestoreProgressEntry - var wg sync.WaitGroup - wg.Add(1) - go func() { - defer wg.Done() - defer cancel() - var err error - summary, err = readRestoreProgressEntries(reader, logger, callback) - if err != nil { - readErr = fmt.Errorf("processing command output: %w", err) +// handleExitError processes a command exit error and converts it to an appropriate error type +func (r *Repo) handleExitError(err error, failureErr error) error { + var exitErr *exec.ExitError + if errors.As(err, &exitErr) { + if exitErr.ExitCode() == 3 { + return ErrPartialBackup + } else { + return fmt.Errorf("exit code %d: %w", exitErr.ExitCode(), failureErr) } - }() - - cmdErr := cmd.Run() - writer.Close() - wg.Wait() - if cmdErr != nil || readErr != nil { - if cmdErr != nil { - var exitErr *exec.ExitError - if errors.As(cmdErr, &exitErr) { - if exitErr.ExitCode() == 3 { - cmdErr = ErrPartialBackup - } else { - cmdErr = fmt.Errorf("exit code %d: %w", exitErr.ExitCode(), ErrRestoreFailed) - } - } - } - - return summary, newCmdError(ctx, cmd, errors.Join(cmdErr, readErr)) } - return summary, nil + return err } func (r *Repo) Snapshots(ctx context.Context, opts ...GenericOption) ([]*Snapshot, error) { - cmd := r.commandWithContext(ctx, []string{"snapshots", "--json"}, opts...) - output := bytes.NewBuffer(nil) - r.pipeCmdOutputToWriter(cmd, output) - - if err := cmd.Run(); err != nil { - return nil, newCmdError(ctx, cmd, newErrorWithOutput(err, output.String())) - } - var snapshots []*Snapshot - if err := json.Unmarshal(output.Bytes(), &snapshots); err != nil { - return nil, newCmdError(ctx, cmd, newErrorWithOutput(fmt.Errorf("command output is not valid JSON: %w", err), output.String())) + if err := r.executeWithJSONOutput(ctx, []string{"snapshots", "--json"}, &snapshots, opts...); err != nil { + return nil, err } for _, snapshot := range snapshots { @@ -289,45 +294,36 @@ func (r *Repo) Forget(ctx context.Context, policy *RetentionPolicy, opts ...Gene args := []string{"forget", "--json"} args = append(args, policy.toForgetFlags()...) - cmd := r.commandWithContext(ctx, args, opts...) - output := bytes.NewBuffer(nil) - r.pipeCmdOutputToWriter(cmd, output) - if err := cmd.Run(); err != nil { - return nil, newCmdError(ctx, cmd, newErrorWithOutput(err, output.String())) + var results []ForgetResult + if err := r.executeWithJSONOutput(ctx, args, &results, opts...); err != nil { + return nil, err } - var result []ForgetResult - if err := json.Unmarshal(output.Bytes(), &result); err != nil { - return nil, newCmdError(ctx, cmd, newErrorWithOutput(fmt.Errorf("command output is not valid JSON: %w", err), output.String())) - } - if len(result) != 1 { - return nil, fmt.Errorf("expected 1 output from forget, got %v", len(result)) - } - if err := result[0].Validate(); err != nil { - return nil, newCmdError(ctx, cmd, fmt.Errorf("invalid forget result: %w", err)) + if len(results) != 1 { + return nil, fmt.Errorf("expected 1 output from forget, got %v", len(results)) } - return &result[0], nil + if err := results[0].Validate(); err != nil { + return nil, fmt.Errorf("invalid forget result: %w", err) + } + + return &results[0], nil } func (r *Repo) ForgetSnapshot(ctx context.Context, snapshotId string, opts ...GenericOption) error { args := []string{"forget", "--json", snapshotId} - - output := bytes.NewBuffer(nil) - cmd := r.commandWithContext(ctx, args, opts...) - r.pipeCmdOutputToWriter(cmd, output) - if err := cmd.Run(); err != nil { - return newCmdError(ctx, cmd, newErrorWithOutput(err, output.String())) - } - - return nil + _, err := r.executeWithOutput(ctx, args, opts...) + return err } func (r *Repo) Prune(ctx context.Context, pruneOutput io.Writer, opts ...GenericOption) error { - args := []string{"prune"} - cmd := r.commandWithContext(ctx, args, opts...) - if pruneOutput != nil { - r.pipeCmdOutputToWriter(cmd, pruneOutput) + return r.runSimpleCommand(ctx, []string{"prune"}, pruneOutput, opts...) +} + +func (r *Repo) Check(ctx context.Context, checkOutput io.Writer, opts ...GenericOption) error { + cmd := r.commandWithContext(ctx, []string{"check"}, opts...) + if checkOutput != nil { + r.pipeCmdOutputToWriter(cmd, checkOutput) } if err := cmd.Run(); err != nil { return newCmdError(ctx, cmd, err) @@ -335,12 +331,11 @@ func (r *Repo) Prune(ctx context.Context, pruneOutput io.Writer, opts ...Generic return nil } -func (r *Repo) Check(ctx context.Context, checkOutput io.Writer, opts ...GenericOption) error { - args := []string{"check"} +// runSimpleCommand executes a command with optional output capture +func (r *Repo) runSimpleCommand(ctx context.Context, args []string, outputWriter io.Writer, opts ...GenericOption) error { cmd := r.commandWithContext(ctx, args, opts...) - cmd.Stdin = bytes.NewBuffer(nil) - if checkOutput != nil { - r.pipeCmdOutputToWriter(cmd, checkOutput) + if outputWriter != nil { + r.pipeCmdOutputToWriter(cmd, outputWriter) } if err := cmd.Run(); err != nil { return newCmdError(ctx, cmd, err) @@ -357,43 +352,28 @@ func (r *Repo) ListDirectory(ctx context.Context, snapshot string, path string, cmd := r.commandWithContext(ctx, []string{"ls", "--json", snapshot, path}, opts...) output := bytes.NewBuffer(nil) r.pipeCmdOutputToWriter(cmd, output) - if err := cmd.Run(); err != nil { return nil, nil, newCmdError(ctx, cmd, newErrorWithOutput(err, output.String())) } - snapshots, entries, err := readLs(output) + snap, entries, err := readLs(output) if err != nil { - return nil, nil, newCmdError(ctx, cmd, newErrorWithOutput(err, output.String())) + return nil, nil, newCmdError(ctx, cmd, fmt.Errorf("error parsing JSON: %w", err)) } - - return snapshots, entries, nil + return snap, entries, nil } func (r *Repo) Unlock(ctx context.Context, opts ...GenericOption) error { - output := bytes.NewBuffer(nil) - cmd := r.commandWithContext(ctx, []string{"unlock"}, opts...) - r.pipeCmdOutputToWriter(cmd, output) - if err := cmd.Run(); err != nil { - return newCmdError(ctx, cmd, newErrorWithOutput(err, output.String())) - } - return nil + _, err := r.executeWithOutput(ctx, []string{"unlock"}, opts...) + return err } func (r *Repo) Stats(ctx context.Context, opts ...GenericOption) (*RepoStats, error) { - cmd := r.commandWithContext(ctx, []string{"stats", "--json", "--mode=raw-data"}, opts...) - output := bytes.NewBuffer(nil) - r.pipeCmdOutputToWriter(cmd, output) - - if err := cmd.Run(); err != nil { - return nil, newCmdError(ctx, cmd, err) - } - var stats RepoStats - if err := json.Unmarshal(output.Bytes(), &stats); err != nil { - return nil, newCmdError(ctx, cmd, newErrorWithOutput(fmt.Errorf("command output is not valid JSON: %w", err), output.String())) + err := r.executeWithJSONOutput(ctx, []string{"stats", "--json", "--mode=raw-data"}, &stats, opts...) + if err != nil { + return nil, err } - return &stats, nil } @@ -403,11 +383,8 @@ func (r *Repo) AddTags(ctx context.Context, snapshotIDs []string, tags []string, args = append(args, "--add", strings.Join(tags, ",")) args = append(args, snapshotIDs...) - cmd := r.commandWithContext(ctx, args, opts...) - if err := cmd.Run(); err != nil { - return newCmdError(ctx, cmd, err) - } - return nil + _, err := r.executeWithOutput(ctx, args, opts...) + return err } func (r *Repo) GenericCommand(ctx context.Context, args []string, opts ...GenericOption) error {