diff --git a/.changes/unreleased/Fixed-20261030-130448.yaml b/.changes/unreleased/Fixed-20261030-130448.yaml new file mode 100644 index 0000000..a0f5b50 --- /dev/null +++ b/.changes/unreleased/Fixed-20261030-130448.yaml @@ -0,0 +1,3 @@ +kind: Fixed +body: Writer.ReadFrom and io.Copy now return the recorded large-file upload error instead of masking it with context cancellation (#245). +time: 2026-10-30T13:04:48.509659-06:00 diff --git a/b2/b2.go b/b2/b2.go index 1d9b9d7..db908fb 100644 --- a/b2/b2.go +++ b/b2/b2.go @@ -735,7 +735,7 @@ func (o *Object) URL() string { // // Callers must close the writer when finished and check the error status. func (o *Object) NewWriter(ctx context.Context, opts ...WriterOption) *Writer { - ctx, cancel := context.WithCancel(ctx) + ctx, cancel := context.WithCancelCause(ctx) w := &Writer{ o: o, name: o.name, diff --git a/b2/b2_test.go b/b2/b2_test.go index ea64314..22d963f 100644 --- a/b2/b2_test.go +++ b/b2/b2_test.go @@ -18,6 +18,7 @@ import ( "bytes" "context" "crypto/sha1" + "errors" "fmt" "io" "io/ioutil" @@ -1726,6 +1727,7 @@ func TestWriterReturnsError(t *testing.T) { ctx := context.Background() ctx, cancel := context.WithTimeout(ctx, 10*time.Second) defer cancel() + want := testError{} client := &Client{ backend: &beRoot{ @@ -1734,13 +1736,13 @@ func TestWriterReturnsError(t *testing.T) { errs: &errCont{ errMap: map[string]map[int]error{ "uploadPart": { - 0: testError{}, - 1: testError{}, - 2: testError{}, - 3: testError{}, - 4: testError{}, - 5: testError{}, - 6: testError{}, + 0: want, + 1: want, + 2: want, + 3: want, + 4: want, + 5: want, + 6: want, }, }, }, @@ -1756,8 +1758,8 @@ func TestWriterReturnsError(t *testing.T) { r := io.LimitReader(zReader{}, 1e7) w.ChunkSize = 1e4 w.ConcurrentUploads = 4 - if _, err := io.Copy(w, r); err == nil { - t.Fatalf("io.Copy: should have returned an error") + if _, err := io.Copy(w, r); !errors.Is(err, want) { + t.Fatalf("io.Copy error = %v, want %v", err, want) } } diff --git a/b2/core_edges_test.go b/b2/core_edges_test.go index 481d415..402f32f 100644 --- a/b2/core_edges_test.go +++ b/b2/core_edges_test.go @@ -49,7 +49,8 @@ func TestObjectLockUpdateInvalidatesCachedInfo(t *testing.T) { func TestObjectDeleteRejectsFolderStatus(t *testing.T) { file := &testFile{n: "folder", a: "folder", files: map[string]string{"folder": "present"}} - object := objectForDeleteOptionTest(file) + root := &beRoot{b2i: &testRoot{errs: &errCont{}, bucketMap: make(map[string]map[string]string)}} + object := &Object{f: &beFile{b2file: file, ri: root}} if err := object.Delete(context.Background()); err == nil || !strings.Contains(err.Error(), "not a regular file or hide marker") { t.Fatalf("Delete(folder) error = %v, want non-regular-file rejection", err) } diff --git a/b2/progress_test.go b/b2/progress_test.go index 8b51e0f..8cf3b87 100644 --- a/b2/progress_test.go +++ b/b2/progress_test.go @@ -300,8 +300,8 @@ func (progressSkipChunk) uploadPart(context.Context, readResetter, string, int, } func TestProgressResumeSkipDoesNotReadClosedBuffer(t *testing.T) { - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() + ctx, cancel := context.WithCancelCause(context.Background()) + defer cancel(nil) buf := newMemoryBuffer() if _, err := buf.Write([]byte("already uploaded")); err != nil { t.Fatal(err) diff --git a/b2/writer.go b/b2/writer.go index 9674ff1..d6fa0f6 100644 --- a/b2/writer.go +++ b/b2/writer.go @@ -83,7 +83,7 @@ type Writer struct { csize int ctx context.Context - cancel context.CancelFunc // cancels ctx + cancel context.CancelCauseFunc // cancels ctx with the first Writer error ctxf func() context.Context errf func(error) ready chan chunk @@ -140,6 +140,11 @@ type chunk struct { buf writeBuffer } +type writerCause struct{ err error } + +func (c writerCause) Error() string { return c.err.Error() } +func (c writerCause) Unwrap() error { return c.err } + func (w *Writer) setErr(err error) { if err == nil || err == io.EOF { return @@ -151,7 +156,7 @@ func (w *Writer) setErr(err error) { } blog.V(1).Infof("error writing object: %T", err) w.err = err - w.cancel() + w.cancel(writerCause{err}) w.emux.Unlock() if w.ctxf == nil || w.file == nil { return @@ -655,16 +660,30 @@ func (w *Writer) sendChunk() error { // // Note that io.Copy will automatically choose to use ReadFrom. // -// If the Writer's context is cancelled while ReadFrom is copying from a -// non-seekable r, ReadFrom returns the context error without waiting for a -// Read on r that is already in progress. Once ReadFrom has returned, nothing -// more from r is written to w; data read before the cancellation may already -// have been written. That pending Read may still finish later, so do not reuse -// r until it has returned (for example, close r first if it supports it). +// On every path, ReadFrom returns the Writer's error when it cancels itself; +// caller cancellation returns context.Canceled or context.DeadlineExceeded, +// even if the caller supplied a custom cause. If the Writer's context is +// cancelled while copying from a non-seekable r, ReadFrom returns without +// waiting for a Read on r that is already in progress. +// When the Writer cancels itself, ReadFrom returns the same first recorded +// error as Close, so errors.Is and errors.As work and io.Copy sees it rather +// than context.Canceled. +// Once ReadFrom has returned, nothing more from r is written to w; data read +// before cancellation may already have been written. That pending Read may +// still finish later, so do not reuse r until it has returned (for example, +// close r first if it supports it). // // With Resume, ReadFrom uses the ordinary buffered path; if r is seekable, // its remaining length supplies the progress total. -func (w *Writer) ReadFrom(r io.Reader) (int64, error) { +func (w *Writer) ReadFrom(r io.Reader) (n int64, err error) { + defer func() { + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + var cause writerCause + if errors.As(context.Cause(w.ctx), &cause) { + err = cause.err + } + } + }() rs, ok := r.(io.ReadSeeker) if !ok { return copyContext(w.ctx, w, r) diff --git a/b2/writer_log_security_test.go b/b2/writer_log_security_test.go index fe890f1..d4782fe 100644 --- a/b2/writer_log_security_test.go +++ b/b2/writer_log_security_test.go @@ -65,7 +65,7 @@ func TestWriterErrorLoggingOmitsErrorText(t *testing.T) { } var output bytes.Buffer log.SetOutput(&output) - w := &Writer{name: "object", cancel: func() {}} + w := &Writer{name: "object", cancel: func(error) {}} w.setErr(errors.New("server-error-secret")) if strings.Contains(output.String(), "server-error-secret") { t.Fatal("writer logged untrusted error text") diff --git a/b2/writer_readfrom_error_test.go b/b2/writer_readfrom_error_test.go new file mode 100644 index 0000000..0e3f50b --- /dev/null +++ b/b2/writer_readfrom_error_test.go @@ -0,0 +1,404 @@ +package b2 + +import ( + "bytes" + "context" + "errors" + "io" + "net/http" + "sync" + "testing" + "time" +) + +// Credential-free fakes accept this below-minimum part size to force multipart uploads. +const readFromTestChunkSize = 3 + +func TestWriterReadFromReturnsUploadError(t *testing.T) { + for _, tc := range []struct { + name string + seekable bool + resume bool + }{ + {name: "non-seekable"}, + {name: "seekable resume", seekable: true, resume: true}, + {name: "seekable streaming", seekable: true}, + } { + t.Run(tc.name, func(t *testing.T) { + bucket, srv := newMappingBucket(t, updateBucketJSON) + srv.Reply("b2_list_unfinished_large_files", http.StatusOK, `{"files":[],"nextFileId":null}`) + srv.Reply("b2_start_large_file", http.StatusOK, `{"fileId":"large-id"}`) + srv.Reply("upload_part", http.StatusBadRequest, `{"status":400,"code":"bad_request","message":"part failed"}`) + + writer := bucket.Object("failed").NewWriter(context.Background()) + writer.ChunkSize = readFromTestChunkSize + writer.ConcurrentUploads = 1 + writer.Resume = tc.resume + payload := bytes.Repeat([]byte("x"), 128) + var source io.Reader + if tc.seekable { + source = struct{ io.ReadSeeker }{bytes.NewReader(payload)} + } else { + source = io.LimitReader(bytes.NewReader(payload), int64(len(payload))) + } + _, err := io.Copy(writer, source) + if err == nil || err.Error() != "b2_upload_part: 400: part failed" { + t.Errorf("io.Copy error = %v, want the permanent upload-part failure", err) + } + if got := len(srv.Calls("upload_part")); got == 0 { + t.Error("no upload_part request reached the server") + } + if closeErr := writer.Close(); closeErr == nil || closeErr.Error() != "b2_upload_part: 400: part failed" { + t.Errorf("Close error = %v, want the permanent upload-part failure", closeErr) + } + }) + } +} + +func TestWriterReadFromCallerContext(t *testing.T) { + for _, tc := range []struct { + name string + deadline bool + want error + }{ + {name: "cancel", want: context.Canceled}, + {name: "deadline", deadline: true, want: context.DeadlineExceeded}, + } { + t.Run(tc.name, func(t *testing.T) { + var ctx context.Context + var cancel context.CancelFunc + if tc.deadline { + ctx, cancel = context.WithTimeout(context.Background(), 500*time.Millisecond) + } else { + ctx, cancel = context.WithCancel(context.Background()) + } + defer cancel() + bucket := newProgressReviewBucket(t) + writer := bucket.Object("caller-context").NewWriter(ctx) + source, release := newBlockedCopyReader(t) + t.Cleanup(func() { + release() + select { + case <-source.done: + case <-time.After(5 * time.Second): + t.Error("copy goroutine did not finish") + } + _ = writer.Close() + }) + done := make(chan error, 1) + go func() { + _, err := io.Copy(writer, source) + done <- err + }() + select { + case <-source.entered: + case <-time.After(5 * time.Second): + t.Fatal("copy did not enter the blocked Read") + } + if !tc.deadline { + cancel() + } + select { + case err := <-done: + if !errors.Is(err, tc.want) { + t.Fatalf("io.Copy error = %v, want %v", err, tc.want) + } + case <-time.After(5 * time.Second): + t.Fatal("copy did not return after caller context ended") + } + if err := writer.getErr(); err != nil { + t.Fatalf("Writer recorded error = %v, want nil", err) + } + }) + } +} + +func TestWriterReadFromReturnsInjectedUploadError(t *testing.T) { + for _, tc := range []struct { + name string + seekable bool + resume bool + }{ + {name: "non-seekable"}, + {name: "seekable resume", seekable: true, resume: true}, + {name: "seekable streaming", seekable: true}, + } { + t.Run(tc.name, func(t *testing.T) { + injected := errors.New("permanent part upload failure") + bucket := newProgressReviewBucket(t) + bucket.b = &readFromFailureBucket{beBucketInterface: bucket.b, failure: injected} + writer := bucket.Object("injected-error").NewWriter(context.Background()) + writer.ChunkSize = readFromTestChunkSize + writer.ConcurrentUploads = 1 + writer.Resume = tc.resume + payload := bytes.Repeat([]byte("x"), 128) + var source io.Reader + if tc.seekable { + source = struct{ io.ReadSeeker }{bytes.NewReader(payload)} + } else { + source = io.LimitReader(bytes.NewReader(payload), int64(len(payload))) + } + _, err := io.Copy(writer, source) + if !errors.Is(err, injected) { + t.Errorf("io.Copy error = %v, want injected upload error", err) + } + if err := writer.Close(); !errors.Is(err, injected) { + t.Errorf("Close error = %v, want injected upload error", err) + } + }) + } +} + +// A transport can flatten a caller-cancelled request into a non-context error. +// Hold the streaming producer until that error is recorded, then let it observe +// the cancelled context. The caller's cancellation must still win. +func TestWriterReadFromStreamingCallerContextWithRecordedTransportError(t *testing.T) { + customCause := errors.New("caller-provided cancellation cause") + for _, tc := range []struct { + name string + deadline bool + customCause bool + want error + }{ + {name: "cancel", want: context.Canceled}, + {name: "deadline", deadline: true, want: context.DeadlineExceeded}, + {name: "cancel with cause", customCause: true, want: context.Canceled}, + {name: "deadline with cause", deadline: true, customCause: true, want: context.DeadlineExceeded}, + } { + t.Run(tc.name, func(t *testing.T) { + var ctx context.Context + var cancel context.CancelFunc + var cancelWithCause context.CancelCauseFunc + if tc.deadline { + var cause error + if tc.customCause { + cause = customCause + } + delayed := newDelayedTimeoutCauseContext(100*time.Millisecond, cause) + ctx, cancel = delayed, delayed.cancel + } else if tc.customCause { + ctx, cancelWithCause = context.WithCancelCause(context.Background()) + cancel = func() { cancelWithCause(nil) } + } else { + ctx, cancel = context.WithCancel(context.Background()) + } + defer cancel() + + startEntered := make(chan struct{}) + releaseStart := make(chan struct{}) + partEntered := make(chan struct{}) + bufferEntered := make(chan struct{}) + releaseBuffer := make(chan struct{}) + recorded := make(chan struct{}) + flattened := errors.New("flattened transport error") + bucket := newProgressReviewBucket(t) + bucket.b = &readFromFailureBucket{ + beBucketInterface: bucket.b, + startEntered: startEntered, + releaseStart: releaseStart, + partEntered: partEntered, + failure: flattened, + } + writer := bucket.Object("caller-context-streaming").NewWriter(ctx, + WithCancelOnError(func() context.Context { return context.Background() }, func(error) { close(recorded) })) + writer.ChunkSize = readFromTestChunkSize + writer.ConcurrentUploads = 1 + done := make(chan error, 1) + finished := make(chan struct{}) + go func() { + defer close(finished) + _, err := io.Copy(writer, struct{ io.ReadSeeker }{bytes.NewReader(bytes.Repeat([]byte("x"), 128))}) + done <- err + }() + var releaseStartOnce, releaseBufferOnce sync.Once + unblock := func() { + releaseStartOnce.Do(func() { close(releaseStart) }) + releaseBufferOnce.Do(func() { close(releaseBuffer) }) + } + t.Cleanup(func() { + cancel() + unblock() + select { + case <-finished: + case <-time.After(5 * time.Second): + t.Error("streaming copy did not finish") + } + _ = writer.Close() + }) + waitReadFromEvent(t, startEntered, "large-file start") + originalBuffer := writer.newBuffer + writer.newBuffer = func() (writeBuffer, error) { + close(bufferEntered) + <-releaseBuffer + return originalBuffer() + } + releaseStartOnce.Do(func() { close(releaseStart) }) + waitReadFromEvent(t, partEntered, "part upload") + waitReadFromEvent(t, bufferEntered, "next buffer") + if tc.deadline { + delayed := ctx.(*delayedTimeoutCauseContext) + delayed.start() + } else { + if tc.customCause { + cancelWithCause(customCause) + } else { + cancel() + } + } + waitReadFromEvent(t, recorded, "recorded transport error") + if err := writer.getErr(); !errors.Is(err, flattened) { + t.Fatalf("recorded error = %v, want flattened transport error", err) + } + releaseBufferOnce.Do(func() { close(releaseBuffer) }) + select { + case err := <-done: + if !errors.Is(err, tc.want) { + t.Fatalf("io.Copy error = %v, want caller's %v", err, tc.want) + } + if tc.customCause && errors.Is(err, customCause) { + t.Fatalf("io.Copy error = %v, unexpectedly matches caller's custom cause", err) + } + case <-time.After(5 * time.Second): + t.Fatal("streaming copy did not return") + } + }) + } +} + +func waitReadFromEvent(t *testing.T, event <-chan struct{}, name string) { + t.Helper() + select { + case <-event: + case <-time.After(5 * time.Second): + t.Fatalf("timed out waiting for %s", name) + } +} + +type readFromFailureBucket struct { + beBucketInterface + startEntered, releaseStart, partEntered chan struct{} + failure error +} + +func (b *readFromFailureBucket) startLargeFile(ctx context.Context, name, contentType string, info map[string]string, retention *FileRetention, legalHold LegalHold, encryption *uploadEncryption) (beLargeFileInterface, error) { + file, err := b.beBucketInterface.startLargeFile(ctx, name, contentType, info, retention, legalHold, encryption) + if err != nil { + return nil, err + } + if b.startEntered != nil { + close(b.startEntered) + <-b.releaseStart + } + return &readFromFailureFile{beLargeFileInterface: file, partEntered: b.partEntered, failure: b.failure}, nil +} + +func (b *readFromFailureBucket) listUnfinishedLargeFiles(context.Context, int, string) ([]beFileInterface, string, error) { + return nil, "", nil +} + +type readFromFailureFile struct { + beLargeFileInterface + partEntered chan struct{} + failure error +} + +func (f *readFromFailureFile) getUploadPartURL(ctx context.Context) (beFileChunkInterface, error) { + chunk, err := f.beLargeFileInterface.getUploadPartURL(ctx) + if err != nil { + return nil, err + } + return &readFromFailureChunk{beFileChunkInterface: chunk, partEntered: f.partEntered, failure: f.failure}, nil +} + +type readFromFailureChunk struct { + beFileChunkInterface + partEntered chan struct{} + failure error +} + +func (c *readFromFailureChunk) uploadPart(ctx context.Context, _ readResetter, _ string, _, _ int) (int, error) { + if c.partEntered != nil { + close(c.partEntered) + <-ctx.Done() + } + return 0, c.failure +} + +// delayedTimeoutCauseContext starts its timeout only after a test has reached +// the controlled upload state. Value forwards the timed context's cause to +// Writer's derived context. +type delayedTimeoutCauseContext struct { + mu sync.Mutex + ctx context.Context + timeout time.Duration + cause error + done chan struct{} + err error + started bool + cancelFn context.CancelFunc +} + +func newDelayedTimeoutCauseContext(timeout time.Duration, cause error) *delayedTimeoutCauseContext { + return &delayedTimeoutCauseContext{ + ctx: context.Background(), + timeout: timeout, + cause: cause, + done: make(chan struct{}), + } +} + +func (c *delayedTimeoutCauseContext) Deadline() (time.Time, bool) { + c.mu.Lock() + defer c.mu.Unlock() + return c.ctx.Deadline() +} + +func (c *delayedTimeoutCauseContext) Done() <-chan struct{} { return c.done } + +func (c *delayedTimeoutCauseContext) Err() error { + c.mu.Lock() + defer c.mu.Unlock() + return c.err +} + +func (c *delayedTimeoutCauseContext) Value(key any) any { + c.mu.Lock() + defer c.mu.Unlock() + return c.ctx.Value(key) +} + +func (c *delayedTimeoutCauseContext) start() { + c.mu.Lock() + if c.started || c.err != nil { + c.mu.Unlock() + return + } + ctx, cancel := context.WithTimeoutCause(context.Background(), c.timeout, c.cause) + c.ctx, c.cancelFn, c.started = ctx, cancel, true + c.mu.Unlock() + go func() { + <-ctx.Done() + c.finish(ctx.Err()) + }() +} + +func (c *delayedTimeoutCauseContext) cancel() { + c.mu.Lock() + cancel := c.cancelFn + c.mu.Unlock() + if cancel != nil { + cancel() + return + } + c.finish(context.Canceled) +} + +func (c *delayedTimeoutCauseContext) finish(err error) { + c.mu.Lock() + defer c.mu.Unlock() + if c.err != nil { + return + } + c.err = err + close(c.done) +}