From 32231385ed20c0dccda12361eaac7cc088ec15a0 Mon Sep 17 00:00:00 2001 From: garethgeorge Date: Thu, 11 Apr 2024 20:31:49 -0700 Subject: [PATCH] fix: address minor data race in command output handling and enable --race in coverage --- .github/workflows/build-and-test.yml | 2 +- internal/ioutil/ioutil.go | 36 ++++++++++++++++++++++++---- internal/orchestrator/task.go | 4 ++-- pkg/restic/logging.go | 4 ++++ pkg/restic/restic.go | 18 +++----------- 5 files changed, 41 insertions(+), 23 deletions(-) diff --git a/.github/workflows/build-and-test.yml b/.github/workflows/build-and-test.yml index 0291e555..2040e3cd 100644 --- a/.github/workflows/build-and-test.yml +++ b/.github/workflows/build-and-test.yml @@ -37,7 +37,7 @@ jobs: args: release --snapshot --clean - name: Test - run: PATH=$(pwd):$PATH go test ./... + run: PATH=$(pwd):$PATH go test ./... --race build-win: runs-on: windows-latest diff --git a/internal/ioutil/ioutil.go b/internal/ioutil/ioutil.go index 75167cb9..939441e2 100644 --- a/internal/ioutil/ioutil.go +++ b/internal/ioutil/ioutil.go @@ -1,12 +1,20 @@ package ioutil import ( + "bytes" "fmt" "io" + "slices" + "sync" ) +type Capturer interface { + Bytes() []byte +} + // HeadWriter keeps the first 'Limit' bytes in memory. type HeadWriter struct { + mu sync.Mutex Buf []byte Limit int } @@ -25,11 +33,14 @@ func (w *HeadWriter) Write(p []byte) (n int, err error) { } func (w *HeadWriter) Bytes() []byte { - return w.Buf + w.mu.Lock() + defer w.mu.Unlock() + return slices.Clone(w.Buf) } // tailWriter keeps the last 'Limit' bytes in memory. type TailWriter struct { + mu sync.Mutex Buf []byte Limit int } @@ -37,6 +48,8 @@ type TailWriter struct { var _ io.Writer = &TailWriter{} func (w *TailWriter) Write(p []byte) (n int, err error) { + w.mu.Lock() + defer w.mu.Unlock() w.Buf = append(w.Buf, p...) if len(w.Buf) > w.Limit { w.Buf = w.Buf[len(w.Buf)-w.Limit:] @@ -45,10 +58,13 @@ func (w *TailWriter) Write(p []byte) (n int, err error) { } func (w *TailWriter) Bytes() []byte { - return w.Buf + w.mu.Lock() + defer w.mu.Unlock() + return slices.Clone(w.Buf) } type OutputCapturer struct { + mu sync.Mutex HeadWriter TailWriter Limit int @@ -66,21 +82,31 @@ func NewOutputCapturer(limit int) *OutputCapturer { } func (w *OutputCapturer) Write(p []byte) (n int, err error) { + w.mu.Lock() + defer w.mu.Unlock() w.HeadWriter.Write(p) w.TailWriter.Write(p) w.totalBytes += len(p) return len(p), nil } -func (w *OutputCapturer) String() string { +func (w *OutputCapturer) Bytes() []byte { + w.mu.Lock() + defer w.mu.Unlock() head := w.HeadWriter.Bytes() tail := w.TailWriter.Bytes() if w.totalBytes <= w.Limit { - return string(head) + return head } head = head[:w.Limit/2] tail = tail[len(tail)-w.Limit/2:] - return fmt.Sprintf("%s...[%v bytes dropped]...%s", string(head), w.totalBytes-len(head)-len(tail), string(tail)) + buf := bytes.NewBuffer(make([]byte, 0, len(head)+len(tail)+100)) + + buf.Write(head) + buf.WriteString(fmt.Sprintf("...[%v bytes dropped]...", w.totalBytes-len(head)-len(tail))) + buf.Write(tail) + + return buf.Bytes() } diff --git a/internal/orchestrator/task.go b/internal/orchestrator/task.go index ac6d19fd..a9f76939 100644 --- a/internal/orchestrator/task.go +++ b/internal/orchestrator/task.go @@ -67,8 +67,8 @@ func (t *TaskWithOperation) runWithOpAndContext(ctx context.Context, do func(ctx err := do(ctx, t.op) - if str := capture.String(); len(str) > 0 { - ref, e := t.orch.logStore.Write([]byte(str)) + if bytes := capture.Bytes(); len(bytes) > 0 { + ref, e := t.orch.logStore.Write(bytes) if e != nil { errors.Join(err, fmt.Errorf("failed to write log to logstore: %w", e)) } diff --git a/pkg/restic/logging.go b/pkg/restic/logging.go index c3e367b4..329d89ba 100644 --- a/pkg/restic/logging.go +++ b/pkg/restic/logging.go @@ -24,8 +24,12 @@ func addLoggingToCommand(ctx context.Context, cmd *exec.Cmd) { } if cmd.Stdout != nil { cmd.Stdout = io.MultiWriter(cmd.Stdout, logger) + } else { + cmd.Stdout = logger } if cmd.Stderr != nil { cmd.Stderr = io.MultiWriter(cmd.Stderr, logger) + } else { + cmd.Stderr = logger } } diff --git a/pkg/restic/restic.go b/pkg/restic/restic.go index e2223b59..c680e492 100644 --- a/pkg/restic/restic.go +++ b/pkg/restic/restic.go @@ -81,12 +81,6 @@ func (r *Repo) pipeCmdOutputToWriter(cmd *exec.Cmd, handlers ...io.Writer) { cmd.Stderr = io.MultiWriter(handlers...) } -func (r *Repo) pipeCmdOutputToLogger(ctx context.Context, cmd *exec.Cmd) { - if logger := LoggerFromContext(ctx); logger != nil { - r.pipeCmdOutputToWriter(cmd, logger) - } -} - // init initializes the repo, the command will be cancelled with the context. func (r *Repo) init(ctx context.Context, opts ...GenericOption) error { if r.initialized { @@ -175,7 +169,7 @@ func (r *Repo) Backup(ctx context.Context, paths []string, progressCallback func } if cmdErr != nil || readErr != nil { - return summary, newCmdErrorPreformatted(ctx, cmd, capture.String(), errors.Join(cmdErr, readErr)) + return summary, newCmdErrorPreformatted(ctx, cmd, string(capture.Bytes()), errors.Join(cmdErr, readErr)) } return summary, nil @@ -185,7 +179,6 @@ func (r *Repo) Snapshots(ctx context.Context, opts ...GenericOption) ([]*Snapsho cmd := r.commandWithContext(ctx, []string{"snapshots", "--json"}, opts...) output := bytes.NewBuffer(nil) r.pipeCmdOutputToWriter(cmd, output) - r.pipeCmdOutputToLogger(ctx, cmd) if err := cmd.Run(); err != nil { return nil, newCmdError(ctx, cmd, output.String(), err) @@ -247,12 +240,11 @@ func (r *Repo) Prune(ctx context.Context, pruneOutput io.Writer, opts ...Generic cmd := r.commandWithContext(ctx, args, opts...) output := bytes.NewBuffer(nil) r.pipeCmdOutputToWriter(cmd, output) - r.pipeCmdOutputToLogger(ctx, cmd) if pruneOutput != nil { r.pipeCmdOutputToWriter(cmd, pruneOutput) } if err := cmd.Run(); err != nil { - return newCmdErrorPreformatted(ctx, cmd, output.String(), err) + return newCmdError(ctx, cmd, output.String(), err) } return nil } @@ -262,7 +254,6 @@ func (r *Repo) Restore(ctx context.Context, snapshot string, callback func(*Rest output := ioutil.NewOutputCapturer(outputBufferLimit) reader, writer := io.Pipe() r.pipeCmdOutputToWriter(cmd, output, writer) - r.pipeCmdOutputToLogger(ctx, cmd) if err := cmd.Start(); err != nil { return nil, newCmdError(ctx, cmd, "", err) @@ -295,7 +286,7 @@ func (r *Repo) Restore(ctx context.Context, snapshot string, callback func(*Rest wg.Wait() if cmdErr != nil || readErr != nil { - return nil, newCmdErrorPreformatted(ctx, cmd, output.String(), errors.Join(cmdErr, readErr)) + return nil, newCmdErrorPreformatted(ctx, cmd, string(output.Bytes()), errors.Join(cmdErr, readErr)) } return summary, nil @@ -310,7 +301,6 @@ 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) - r.pipeCmdOutputToLogger(ctx, cmd) if err := cmd.Run(); err != nil { return nil, nil, newCmdError(ctx, cmd, output.String(), err) @@ -328,7 +318,6 @@ func (r *Repo) Unlock(ctx context.Context, opts ...GenericOption) error { cmd := r.commandWithContext(ctx, []string{"unlock"}, opts...) output := bytes.NewBuffer(nil) r.pipeCmdOutputToWriter(cmd, output) - r.pipeCmdOutputToLogger(ctx, cmd) if err := cmd.Run(); err != nil { return newCmdError(ctx, cmd, output.String(), err) } @@ -339,7 +328,6 @@ func (r *Repo) Stats(ctx context.Context, opts ...GenericOption) (*RepoStats, er cmd := r.commandWithContext(ctx, []string{"stats", "--json", "--mode=raw-data"}, opts...) output := bytes.NewBuffer(nil) r.pipeCmdOutputToWriter(cmd, output) - r.pipeCmdOutputToLogger(ctx, cmd) if err := cmd.Run(); err != nil { return nil, newCmdError(ctx, cmd, output.String(), err)