From 973d9991017ae49d83a36cd58cee5121e96f88f2 Mon Sep 17 00:00:00 2001 From: Felipe Fumero Date: Thu, 29 Oct 2026 14:50:04 -0600 Subject: [PATCH 01/21] test(b2): reproduce writes after copy cancellation Pin delayed source reads that resume after copyContext returns and Writer.ReadFrom has been cancelled and closed. Part of #229 --- b2/copy_context_test.go | 147 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 147 insertions(+) create mode 100644 b2/copy_context_test.go diff --git a/b2/copy_context_test.go b/b2/copy_context_test.go new file mode 100644 index 00000000..c826bfa7 --- /dev/null +++ b/b2/copy_context_test.go @@ -0,0 +1,147 @@ +package b2 + +import ( + "context" + "errors" + "io" + "sync" + "testing" + "time" +) + +// WriteTo reports completion even when the destination rejects the delayed +// Write, in which case io.Copy never calls Read a second time. +type blockedCopyReader struct { + entered chan struct{} + release chan struct{} + done chan struct{} + read bool + n int64 + err error +} + +func (r *blockedCopyReader) Read(p []byte) (int, error) { + if r.read { + return 0, io.EOF + } + r.read = true + close(r.entered) + <-r.release + return copy(p, "payload"), nil +} + +func (r *blockedCopyReader) WriteTo(w io.Writer) (int64, error) { + defer close(r.done) + r.n, r.err = io.Copy(w, struct{ io.Reader }{r}) + return r.n, r.err +} + +func newBlockedCopyReader(t *testing.T) (*blockedCopyReader, func()) { + t.Helper() + r := &blockedCopyReader{ + entered: make(chan struct{}), + release: make(chan struct{}), + done: make(chan struct{}), + } + var once sync.Once + release := func() { once.Do(func() { close(r.release) }) } + t.Cleanup(release) + return r, release +} + +type copyRecordingWriter struct{ calls int } + +func (w *copyRecordingWriter) Write(p []byte) (int, error) { + w.calls++ + return len(p), nil +} + +func TestCopyContextCancellationStopsDelayedWrite(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + wait, stop := context.WithTimeout(context.Background(), 5*time.Second) + defer stop() + source, release := newBlockedCopyReader(t) + destination := ©RecordingWriter{} + done := make(chan error, 1) + go func() { + _, err := copyContext(ctx, destination, source) + done <- err + }() + select { + case <-source.entered: + case <-wait.Done(): + t.Fatal("copy did not enter the blocked Read") + } + cancel() + select { + case err := <-done: + if !errors.Is(err, context.Canceled) { + t.Fatalf("copyContext error = %v, want context.Canceled", err) + } + case <-wait.Done(): + t.Fatal("copyContext waited for the blocked Read after cancellation") + } + release() + select { + case <-source.done: + case <-wait.Done(): + t.Fatal("copy goroutine did not finish after releasing the Read") + } + if destination.calls != 0 { + t.Errorf("destination Write calls after copyContext returned = %d, want 0", destination.calls) + } +} + +func TestWriterReadFromCancellationStopsWriteAfterClose(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + wait, stop := context.WithTimeout(context.Background(), 5*time.Second) + defer stop() + client := &Client{backend: &beRoot{b2i: &testRoot{ + bucketMap: make(map[string]map[string]string), + errs: &errCont{}, + }}} + bucket, err := client.NewBucket(ctx, "copy-context", &BucketAttrs{Type: Private}) + if err != nil { + t.Fatal(err) + } + writer := bucket.Object("cancelled").NewWriter(ctx) + // Initialize the Writer so Close releases the same buffer a late Write + // would otherwise touch. No upload is needed to reach the cancelled path. + writer.init() + source, release := newBlockedCopyReader(t) + done := make(chan error, 1) + go func() { + _, err := writer.ReadFrom(source) + done <- err + }() + select { + case <-source.entered: + case <-wait.Done(): + t.Fatal("ReadFrom did not enter the non-seekable reader") + } + cancel() + select { + case err := <-done: + if !errors.Is(err, context.Canceled) { + t.Fatalf("ReadFrom error = %v, want context.Canceled", err) + } + case <-wait.Done(): + t.Fatal("ReadFrom waited for the blocked Read after cancellation") + } + if err := writer.Close(); !errors.Is(err, context.Canceled) { + t.Fatalf("Close error = %v, want context.Canceled", err) + } + release() + select { + case <-source.done: + case <-wait.Done(): + t.Fatal("copy goroutine did not finish after Writer.Close") + } + // ErrClosed or context.Canceled here would come from Writer.Write itself: + // the abandoned copy must reject the call before it can reach that method. + if source.n != 0 || source.err == nil || source.err.Error() != "b2: copy abandoned" { + t.Errorf("delayed Write after Close = (%d, %v), want (0, b2: copy abandoned)", source.n, source.err) + } +} From 809a1fba9fdc221eef2a8ebe33e9df6626037277 Mon Sep 17 00:00:00 2001 From: Felipe Fumero Date: Thu, 29 Oct 2026 14:57:10 -0600 Subject: [PATCH 02/21] fix(b2): stop a cancelled copy from writing after copyContext returns A cancelled copy could return while its source Read was still blocked. When that Read resumed, the copy goroutine could write to a Writer that the caller had already closed, causing the slice-bounds panic seen in CI run 40925001975. Guard destination writes and mark the guard stopped before returning on cancellation. Waiting for the copy goroutine is not an option because the source Read may block forever. Preserve the closer path that closes the body and waits for its copy to finish. Closes #229 --- .../unreleased/Fixed-20261029-145247.yaml | 3 ++ b2/reader.go | 34 +++++++++++++++++-- 2 files changed, 34 insertions(+), 3 deletions(-) create mode 100644 .changes/unreleased/Fixed-20261029-145247.yaml diff --git a/.changes/unreleased/Fixed-20261029-145247.yaml b/.changes/unreleased/Fixed-20261029-145247.yaml new file mode 100644 index 00000000..f0e6d3b0 --- /dev/null +++ b/.changes/unreleased/Fixed-20261029-145247.yaml @@ -0,0 +1,3 @@ +kind: Fixed +body: Writer.ReadFrom no longer writes to the Writer after a cancelled copy returns (#229). +time: 2026-10-29T14:52:47.457984-06:00 diff --git a/b2/reader.go b/b2/reader.go index e637ae2f..94340b61 100644 --- a/b2/reader.go +++ b/b2/reader.go @@ -375,20 +375,44 @@ type onlyWriter struct{ w io.Writer } func (ow onlyWriter) Write(p []byte) (int, error) { return ow.w.Write(p) } +var errCopyAbandoned = errors.New("b2: copy abandoned") + +type guardedCopyWriter struct { + mu sync.Mutex + w io.Writer + stopped bool +} + +func (w *guardedCopyWriter) Write(p []byte) (int, error) { + w.mu.Lock() + defer w.mu.Unlock() + if w.stopped { + return 0, errCopyAbandoned + } + return w.w.Write(p) +} + func copyContext(ctx context.Context, w io.Writer, r io.Reader) (int64, error) { return copyContextWithCloser(ctx, w, r, nil) } // copyContextWithCloser closes a download body on cancellation before waiting // for its copy goroutine. A blocked body Read must not outlive Reader.Close. +// Without a closer, it stops destination writes before returning instead of +// waiting for a source Read that may never finish. func copyContextWithCloser(ctx context.Context, w io.Writer, r io.Reader, closer io.Closer) (int64, error) { + if _, ok := w.(*Writer); ok { + w = onlyWriter{w} + } + var guarded *guardedCopyWriter + if closer == nil { + guarded = &guardedCopyWriter{w: w} + w = guarded + } var n int64 var err error done := make(chan struct{}) go func() { - if _, ok := w.(*Writer); ok { - w = onlyWriter{w} - } n, err = io.Copy(w, r) close(done) }() @@ -399,6 +423,10 @@ func copyContextWithCloser(ctx context.Context, w io.Writer, r io.Reader, closer if closer != nil { _ = closer.Close() <-done + } else { + guarded.mu.Lock() + guarded.stopped = true + guarded.mu.Unlock() } return 0, ctx.Err() } From 73a3ae8e09bb733bfe74fe980a0fe4dcaea45999 Mon Sep 17 00:00:00 2001 From: Felipe Fumero Date: Thu, 29 Oct 2026 15:40:55 -0600 Subject: [PATCH 03/21] test(b2): match the abandoned-copy error with errors.Is and document ReadFrom cancellation Address the panel's nits on #229: the regression test compared the error text, so wrapping errCopyAbandoned would break it while the guard still works; and ReadFrom's godoc did not say that a Read already in progress on a non-seekable source can outlive a cancelled call. --- b2/copy_context_test.go | 2 +- b2/writer.go | 6 ++++++ 2 files changed, 7 insertions(+), 1 deletion(-) diff --git a/b2/copy_context_test.go b/b2/copy_context_test.go index c826bfa7..86d9cf59 100644 --- a/b2/copy_context_test.go +++ b/b2/copy_context_test.go @@ -141,7 +141,7 @@ func TestWriterReadFromCancellationStopsWriteAfterClose(t *testing.T) { } // ErrClosed or context.Canceled here would come from Writer.Write itself: // the abandoned copy must reject the call before it can reach that method. - if source.n != 0 || source.err == nil || source.err.Error() != "b2: copy abandoned" { + if source.n != 0 || !errors.Is(source.err, errCopyAbandoned) { t.Errorf("delayed Write after Close = (%d, %v), want (0, b2: copy abandoned)", source.n, source.err) } } diff --git a/b2/writer.go b/b2/writer.go index 653e0049..4825a246 100644 --- a/b2/writer.go +++ b/b2/writer.go @@ -644,6 +644,12 @@ 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 waiting on a +// non-seekable r, ReadFrom returns the context error at once, but a Read on r +// that is already in progress may still finish later. Nothing it returns is +// written to w. Do not reuse r until that Read 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) { From cd459b61f5e96e34202bb8986305dc21cff9c816 Mon Sep 17 00:00:00 2001 From: Felipe Fumero Date: Thu, 29 Oct 2026 16:26:36 -0600 Subject: [PATCH 04/21] docs(b2): state exactly what ReadFrom guarantees after a cancelled copy The comment said nothing a pending Read returns is written, but the guard only stops writes once copyContext returns: a Write already in progress, with data read before the cancellation, can still complete. Say what the code guarantees: after ReadFrom returns, nothing more from r reaches the Writer. --- b2/writer.go | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/b2/writer.go b/b2/writer.go index 4825a246..4f56d04c 100644 --- a/b2/writer.go +++ b/b2/writer.go @@ -644,11 +644,12 @@ 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 waiting on a -// non-seekable r, ReadFrom returns the context error at once, but a Read on r -// that is already in progress may still finish later. Nothing it returns is -// written to w. Do not reuse r until that Read has returned (for example, close -// r first if it supports it). +// 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). // // With Resume, ReadFrom uses the ordinary buffered path; if r is seekable, // its remaining length supplies the progress total. From a37b5faf04001f80d9c0d29efaa9c9612f58596c Mon Sep 17 00:00:00 2001 From: Felipe Fumero Date: Fri, 30 Oct 2026 10:38:58 -0600 Subject: [PATCH 05/21] test(b2): pin ReadFrom cancellation during blocked error cleanup Hold cancel_large_file or WithCancelOnError callbacks behind channels after producer and worker upload failures. ReadFrom must return on cancellation before those channels are released, and the callback must receive the cancellation request result (#229). --- b2/copy_context_cleanup_test.go | 132 ++++++++++++++++++++++++++++++++ 1 file changed, 132 insertions(+) create mode 100644 b2/copy_context_cleanup_test.go diff --git a/b2/copy_context_cleanup_test.go b/b2/copy_context_cleanup_test.go new file mode 100644 index 00000000..47561705 --- /dev/null +++ b/b2/copy_context_cleanup_test.go @@ -0,0 +1,132 @@ +package b2 + +import ( + "context" + "errors" + "net/http" + "strings" + "sync" + "testing" + "time" + + "github.com/Backblaze/blazer/internal/b2wiretest" +) + +func TestWriterReadFromCancellationDoesNotWaitForCleanup(t *testing.T) { + for _, blocked := range []string{"cancel_large_file", "errf"} { + for _, failure := range []string{"sendChunk", "upload_part"} { + t.Run(blocked+"/"+failure, func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + partEntered := make(chan struct{}) + enterPart := sync.OnceFunc(func() { close(partEntered) }) + cancelEntered := make(chan struct{}) + callbackEntered := make(chan struct{}) + partRelease := make(chan struct{}) + cancelRelease := make(chan struct{}) + callbackRelease := make(chan struct{}) + releasePart := sync.OnceFunc(func() { close(partRelease) }) + releaseCancel := sync.OnceFunc(func() { close(cancelRelease) }) + releaseCallback := sync.OnceFunc(func() { close(callbackRelease) }) + if blocked != "cancel_large_file" { + releaseCancel() + } + if blocked != "errf" { + releaseCallback() + } + await := func(ch <-chan struct{}, what string) { + t.Helper() + select { + case <-ch: + case <-time.After(5 * time.Second): + t.Fatalf("timed out waiting for %s", what) + } + } + + srv := b2wiretest.New(t) + srv.Reply("b2_list_buckets", http.StatusOK, `{"buckets":[]}`) + srv.Reply("b2_create_bucket", http.StatusOK, noReplicationBucketJSON) + 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"}`) + srv.Handle("b2_cancel_large_file", func(r b2wiretest.Request) b2wiretest.Response { + b2wiretest.AssertJSON(t, r, "fileId", "large-id") + close(cancelEntered) + <-cancelRelease + return b2wiretest.Response{Status: http.StatusBadRequest, Body: `{"status":400,"code":"bad_request","message":"cleanup failed"}`} + }) + transport := bucketInfoRoundTripper(func(r *http.Request) (*http.Response, error) { + if r.URL.Path == "/upload_part" { + enterPart() + // Hold the worker so cancellation makes the producer's sendChunk fail. + <-partRelease + } + return http.DefaultTransport.RoundTrip(r) + }) + client, err := NewClient(ctx, "account", "key", APIBase(srv.URL()), Transport(transport)) + if err != nil { + t.Fatal(err) + } + bucket, err := client.NewBucket(ctx, "copy-cleanup", &BucketAttrs{Type: Private}) + if err != nil { + t.Fatal(err) + } + callbackResult := make(chan error, 1) + writer := bucket.Object("cancelled").NewWriter(ctx, WithCancelOnError(context.Background, func(err error) { + callbackResult <- err + close(callbackEntered) + <-callbackRelease + })) + // The recording server does not enforce B2's minimum part size. + writer.ChunkSize = 3 + source, releaseSource := newBlockedCopyReader(t) + releaseSource() + returned := make(chan struct{}) + done := make(chan error, 1) + t.Cleanup(func() { + releaseCancel() + releaseCallback() + releasePart() + await(source.done, "copy goroutine cleanup") + await(returned, "ReadFrom goroutine cleanup") + workersDone := make(chan struct{}) + go func() { writer.wg.Wait(); close(workersDone) }() + await(workersDone, "upload worker cleanup") + _ = writer.Close() + }) + go func() { + defer close(returned) + _, err := writer.ReadFrom(source) + done <- err + }() + await(partEntered, "first part upload") + if failure == "upload_part" { + releasePart() + } else { + cancel() + } + entered := cancelEntered + if blocked == "errf" { + entered = callbackEntered + } + await(entered, "blocked "+blocked) + cancel() + select { + case err := <-done: + // A worker's permanent error may win the race with ctx.Done. + partFailed := failure == "upload_part" && err != nil && strings.Contains(err.Error(), "b2_upload_part: 400: part failed") + if !errors.Is(err, context.Canceled) && !partFailed { + t.Fatalf("ReadFrom error = %v, want context.Canceled or the injected part failure", err) + } + case <-time.After(5 * time.Second): + t.Fatalf("ReadFrom waited for blocked %s after cancellation", blocked) + } + releaseCancel() + releaseCallback() + await(callbackEntered, "cancel callback") + if err := <-callbackResult; err == nil || !strings.Contains(err.Error(), "cleanup failed") { + t.Fatalf("errf result = %v, want cancel_large_file's cleanup failed error", err) + } + }) + } + } +} From 42b69e2421ce2be5375f711002dfa515c958804d Mon Sep 17 00:00:00 2001 From: Felipe Fumero Date: Fri, 30 Oct 2026 10:52:50 -0600 Subject: [PATCH 06/21] fix(b2): release the copy guard before upload error cleanup Keep the existing Writer error path and release the copy guard only on terminal write errors. No buffer writes follow that hook, and the stopped check still gates every later write, so no source bytes can reach the destination after a cancelled copy returns. Release the error-state mutex before cancel_large_file and errf as well, so a producer observing a worker failure does not wait behind cleanup. Close joins its existing worker wait group on error exits to preserve delivery of the cleanup result to WithCancelOnError callers. Reject a separate deferred error-recording and cleanup mechanism: reusing setErr preserves first-error ownership and callback semantics without adding parallel state. The public API and ReadFrom godoc remain unchanged. Closes #229 --- b2/reader.go | 15 +++++++++++---- b2/writer.go | 17 +++++++++++++---- 2 files changed, 24 insertions(+), 8 deletions(-) diff --git a/b2/reader.go b/b2/reader.go index 94340b61..0e41e381 100644 --- a/b2/reader.go +++ b/b2/reader.go @@ -385,10 +385,17 @@ type guardedCopyWriter struct { func (w *guardedCopyWriter) Write(p []byte) (int, error) { w.mu.Lock() - defer w.mu.Unlock() + unlock := sync.OnceFunc(w.mu.Unlock) + defer unlock() if w.stopped { return 0, errCopyAbandoned } + if writer, ok := w.w.(*Writer); ok { + return writer.write(p, func(err error) { + unlock() + writer.setErr(err) + }) + } return w.w.Write(p) } @@ -401,14 +408,14 @@ func copyContext(ctx context.Context, w io.Writer, r io.Reader) (int64, error) { // Without a closer, it stops destination writes before returning instead of // waiting for a source Read that may never finish. func copyContextWithCloser(ctx context.Context, w io.Writer, r io.Reader, closer io.Closer) (int64, error) { - if _, ok := w.(*Writer); ok { - w = onlyWriter{w} - } var guarded *guardedCopyWriter if closer == nil { guarded = &guardedCopyWriter{w: w} w = guarded } + if _, ok := w.(*Writer); ok { + w = onlyWriter{w} + } var n int64 var err error done := make(chan struct{}) diff --git a/b2/writer.go b/b2/writer.go index 37b14101..90ecbbfc 100644 --- a/b2/writer.go +++ b/b2/writer.go @@ -145,13 +145,14 @@ func (w *Writer) setErr(err error) { return } w.emux.Lock() - defer w.emux.Unlock() if w.err != nil { + w.emux.Unlock() return } blog.V(1).Infof("error writing object: %T", err) w.err = err w.cancel() + w.emux.Unlock() if w.ctxf == nil || w.file == nil { return } @@ -282,6 +283,12 @@ func (w *Writer) checkObjectLock() error { // Write satisfies the io.Writer interface. func (w *Writer) Write(p []byte) (int, error) { + return w.write(p, w.setErr) +} + +// write calls setErr only on terminal error paths, after its last buffer write. +// A guarded copy can release its write lock there before error cleanup blocks. +func (w *Writer) write(p []byte, setErr func(error)) (int, error) { w.closeWrite.RLock() defer w.closeWrite.RUnlock() if w.closed { @@ -302,18 +309,18 @@ func (w *Writer) Write(p []byte) (int, error) { if len(p) <= left { k, err := w.w.Write(p) if err != nil { - w.setErr(err) + setErr(err) } return n + k, err } i, err := w.w.Write(p[:left]) n += i if err != nil { - w.setErr(err) + setErr(err) return n, err } if err := w.sendChunk(); err != nil { - w.setErr(err) + setErr(err) return n, w.getErr() } p = p[left:] @@ -729,6 +736,8 @@ func (w *Writer) Close() error { w.done.Do(func() { w.closeWrite.Lock() defer w.closeWrite.Unlock() + // Error exits must also wait for worker-owned cancellation callbacks. + defer w.wg.Wait() if !w.everStarted { w.init() w.setErr(w.simpleWriteFile()) From ed73f2477f29b62e64b78b54bbb8a9dd3e362d45 Mon Sep 17 00:00:00 2001 From: Felipe Fumero Date: Fri, 30 Oct 2026 11:18:05 -0600 Subject: [PATCH 07/21] test(b2): pin error mutex release and Close cleanup waiting Hold the producer inside buffer allocation until a worker enters blocked cancellation cleanup, then release it into getErr while the copy guard is still held. This makes moving cleanup back under emux fail deterministically. Gate the worker cancellation request separately and verify that its nonblocking error callback has completed when Close returns. Removing the deferred worker wait now fails without making the callback depend on Close. Keep the existing cancellation cases and production code unchanged. --- b2/copy_context_cleanup_test.go | 120 +++++++++++++++++++++++++++++++- 1 file changed, 117 insertions(+), 3 deletions(-) diff --git a/b2/copy_context_cleanup_test.go b/b2/copy_context_cleanup_test.go index 47561705..6a2dfaff 100644 --- a/b2/copy_context_cleanup_test.go +++ b/b2/copy_context_cleanup_test.go @@ -14,7 +14,7 @@ import ( func TestWriterReadFromCancellationDoesNotWaitForCleanup(t *testing.T) { for _, blocked := range []string{"cancel_large_file", "errf"} { - for _, failure := range []string{"sendChunk", "upload_part"} { + for _, failure := range []string{"sendChunk", "upload_part", "getErr"} { t.Run(blocked+"/"+failure, func(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() @@ -25,9 +25,12 @@ func TestWriterReadFromCancellationDoesNotWaitForCleanup(t *testing.T) { partRelease := make(chan struct{}) cancelRelease := make(chan struct{}) callbackRelease := make(chan struct{}) + bufferEntered := make(chan struct{}) + bufferRelease := make(chan struct{}) releasePart := sync.OnceFunc(func() { close(partRelease) }) releaseCancel := sync.OnceFunc(func() { close(cancelRelease) }) releaseCallback := sync.OnceFunc(func() { close(callbackRelease) }) + releaseBuffer := sync.OnceFunc(func() { close(bufferRelease) }) if blocked != "cancel_large_file" { releaseCancel() } @@ -78,6 +81,18 @@ func TestWriterReadFromCancellationDoesNotWaitForCleanup(t *testing.T) { })) // The recording server does not enforce B2's minimum part size. writer.ChunkSize = 3 + if failure == "getErr" { + buffers := 0 + writer.newBuffer = func() (writeBuffer, error) { + buffers++ + if buffers == 2 { + // The producer still owns the copy guard after handing off part 1. + close(bufferEntered) + <-bufferRelease + } + return newMemoryBuffer(), nil + } + } source, releaseSource := newBlockedCopyReader(t) releaseSource() returned := make(chan struct{}) @@ -86,6 +101,7 @@ func TestWriterReadFromCancellationDoesNotWaitForCleanup(t *testing.T) { releaseCancel() releaseCallback() releasePart() + releaseBuffer() await(source.done, "copy goroutine cleanup") await(returned, "ReadFrom goroutine cleanup") workersDone := make(chan struct{}) @@ -99,7 +115,10 @@ func TestWriterReadFromCancellationDoesNotWaitForCleanup(t *testing.T) { done <- err }() await(partEntered, "first part upload") - if failure == "upload_part" { + if failure == "getErr" { + await(bufferEntered, "producer buffer allocation") + } + if failure != "sendChunk" { releasePart() } else { cancel() @@ -110,10 +129,11 @@ func TestWriterReadFromCancellationDoesNotWaitForCleanup(t *testing.T) { } await(entered, "blocked "+blocked) cancel() + releaseBuffer() select { case err := <-done: // A worker's permanent error may win the race with ctx.Done. - partFailed := failure == "upload_part" && err != nil && strings.Contains(err.Error(), "b2_upload_part: 400: part failed") + partFailed := failure != "sendChunk" && err != nil && strings.Contains(err.Error(), "b2_upload_part: 400: part failed") if !errors.Is(err, context.Canceled) && !partFailed { t.Fatalf("ReadFrom error = %v, want context.Canceled or the injected part failure", err) } @@ -130,3 +150,97 @@ func TestWriterReadFromCancellationDoesNotWaitForCleanup(t *testing.T) { } } } + +func TestWriterCloseWaitsForErrorCallback(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + cancelEntered := make(chan struct{}) + cancelRelease := make(chan struct{}) + releaseCancel := sync.OnceFunc(func() { close(cancelRelease) }) + callbackDone := make(chan struct{}) + callbackResult := make(chan error, 1) + srv := b2wiretest.New(t) + srv.Reply("b2_list_buckets", http.StatusOK, `{"buckets":[]}`) + srv.Reply("b2_create_bucket", http.StatusOK, noReplicationBucketJSON) + 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"}`) + srv.Handle("b2_cancel_large_file", func(r b2wiretest.Request) b2wiretest.Response { + b2wiretest.AssertJSON(t, r, "fileId", "large-id") + close(cancelEntered) + <-cancelRelease + return b2wiretest.Response{Status: http.StatusBadRequest, Body: `{"status":400,"code":"bad_request","message":"cleanup failed"}`} + }) + client, err := NewClient(ctx, "account", "key", APIBase(srv.URL())) + if err != nil { + t.Fatal(err) + } + bucket, err := client.NewBucket(ctx, "copy-cleanup", &BucketAttrs{Type: Private}) + if err != nil { + t.Fatal(err) + } + writer := bucket.Object("cancelled").NewWriter(ctx, WithCancelOnError(context.Background, func(err error) { + callbackResult <- err + close(callbackDone) // No wait on Close or on any other channel. + })) + // The recording server does not enforce B2's minimum part size. + writer.ChunkSize = 3 + t.Cleanup(func() { + releaseCancel() + workersDone := make(chan struct{}) + go func() { writer.wg.Wait(); close(workersDone) }() + select { + case <-workersDone: + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for upload worker cleanup") + } + _ = writer.Close() + }) + // The worker fails while the producer is handing off the next full chunk. + if _, err := writer.Write([]byte("payload")); err == nil || !strings.Contains(err.Error(), "part failed") { + t.Fatalf("Write error = %v, want the worker's part failed error", err) + } + select { + case <-cancelEntered: + case <-time.After(5 * time.Second): + t.Fatal("worker did not enter cancel_large_file") + } + type closeResult struct { + err error + callbackCompleted bool + } + closeStarted := make(chan struct{}) + closed := make(chan closeResult, 1) + go func() { + close(closeStarted) + err := writer.Close() + completed := false + select { + case <-callbackDone: + completed = true + default: + } + closed <- closeResult{err: err, callbackCompleted: completed} + }() + <-closeStarted + select { + case got := <-closed: + t.Fatalf("errf completed when Close returned = %t, want true", got.callbackCompleted) + case <-time.After(time.Second): + // Release cleanup independently of Close, so even a broken Close cannot deadlock the test. + } + releaseCancel() + select { + case got := <-closed: + if !got.callbackCompleted { + t.Error("errf completed when Close returned = false, want true") + } + if got.err == nil || !strings.Contains(got.err.Error(), "part failed") { + t.Errorf("Close error = %v, want the worker's part failed error", got.err) + } + case <-time.After(5 * time.Second): + t.Fatal("Close did not return after worker cleanup completed") + } + if err := <-callbackResult; err == nil || !strings.Contains(err.Error(), "cleanup failed") { + t.Fatalf("errf result = %v, want cancel_large_file's cleanup failed error", err) + } +} From af786d62a2cb08869a9c900998c7760700c15259 Mon Sep 17 00:00:00 2001 From: Felipe Fumero Date: Fri, 30 Oct 2026 11:40:27 -0600 Subject: [PATCH 08/21] docs(b2): document Writer.Close callback wait --- .../unreleased/Changed-20261030-113335.yaml | 3 ++ b2/copy_context_cleanup_test.go | 37 ++++++------------- b2/writer.go | 10 +++-- 3 files changed, 22 insertions(+), 28 deletions(-) create mode 100644 .changes/unreleased/Changed-20261030-113335.yaml diff --git a/.changes/unreleased/Changed-20261030-113335.yaml b/.changes/unreleased/Changed-20261030-113335.yaml new file mode 100644 index 00000000..548f22bf --- /dev/null +++ b/.changes/unreleased/Changed-20261030-113335.yaml @@ -0,0 +1,3 @@ +kind: Changed +body: Writer.Close now waits for the cancel-on-error callback before returning. +time: 2026-10-30T11:33:35.000000-06:00 diff --git a/b2/copy_context_cleanup_test.go b/b2/copy_context_cleanup_test.go index 6a2dfaff..d65a7986 100644 --- a/b2/copy_context_cleanup_test.go +++ b/b2/copy_context_cleanup_test.go @@ -204,38 +204,25 @@ func TestWriterCloseWaitsForErrorCallback(t *testing.T) { case <-time.After(5 * time.Second): t.Fatal("worker did not enter cancel_large_file") } - type closeResult struct { - err error - callbackCompleted bool - } - closeStarted := make(chan struct{}) - closed := make(chan closeResult, 1) + closed := make(chan error, 1) go func() { - close(closeStarted) - err := writer.Close() - completed := false - select { - case <-callbackDone: - completed = true - default: - } - closed <- closeResult{err: err, callbackCompleted: completed} + closed <- writer.Close() }() - <-closeStarted select { - case got := <-closed: - t.Fatalf("errf completed when Close returned = %t, want true", got.callbackCompleted) - case <-time.After(time.Second): - // Release cleanup independently of Close, so even a broken Close cannot deadlock the test. + case err := <-closed: + t.Fatalf("Close returned before cancel_large_file was released: %v", err) + default: } releaseCancel() select { - case got := <-closed: - if !got.callbackCompleted { - t.Error("errf completed when Close returned = false, want true") + case err := <-closed: + select { + case <-callbackDone: + default: + t.Error("errf had not completed when Close returned") } - if got.err == nil || !strings.Contains(got.err.Error(), "part failed") { - t.Errorf("Close error = %v, want the worker's part failed error", got.err) + if err == nil || !strings.Contains(err.Error(), "part failed") { + t.Errorf("Close error = %v, want the worker's part failed error", err) } case <-time.After(5 * time.Second): t.Fatal("Close did not return after worker cleanup completed") diff --git a/b2/writer.go b/b2/writer.go index 90ecbbfc..7b323627 100644 --- a/b2/writer.go +++ b/b2/writer.go @@ -730,8 +730,10 @@ func (w *Writer) ReadFrom(r io.Reader) (int64, error) { } } -// Close satisfies the io.Closer interface. It is critical to check the return -// value of Close for all writers. +// Close satisfies the io.Closer interface. It waits for in-flight part uploads +// and, when WithCancelOnError is set, for the b2_cancel_large_file call and errf +// to finish before it returns. It is critical to check the return value of Close +// for all writers. func (w *Writer) Close() error { w.done.Do(func() { w.closeWrite.Lock() @@ -891,7 +893,9 @@ func WithSSEB2(algorithm string) WriterOption { // upload, to call b2_cancel_large_file on any permanent error. It calls ctxf // to obtain a context with which to cancel the file; this is to allow callers // to set specific timeouts. If errf is non-nil, then it is called with the -// (possibly nil) output of b2_cancel_large_file. +// (possibly nil) output of b2_cancel_large_file. The errf callback may run on a +// background goroutine, and Close waits for it to finish. errf must not call Close +// or wait on the goroutine that calls Close, or they will deadlock. func WithCancelOnError(ctxf func() context.Context, errf func(error)) WriterOption { return func(w *Writer) { w.ctxf = ctxf From 6803b3d1e9f846931e4c65d547afa771a750b6d6 Mon Sep 17 00:00:00 2001 From: Felipe Fumero Date: Fri, 30 Oct 2026 11:49:22 -0600 Subject: [PATCH 09/21] test(b2): make Writer.Close callback proof deterministic --- b2/copy_context_cleanup_test.go | 20 ++++++++++++++++---- 1 file changed, 16 insertions(+), 4 deletions(-) diff --git a/b2/copy_context_cleanup_test.go b/b2/copy_context_cleanup_test.go index d65a7986..48e82951 100644 --- a/b2/copy_context_cleanup_test.go +++ b/b2/copy_context_cleanup_test.go @@ -157,6 +157,9 @@ func TestWriterCloseWaitsForErrorCallback(t *testing.T) { cancelEntered := make(chan struct{}) cancelRelease := make(chan struct{}) releaseCancel := sync.OnceFunc(func() { close(cancelRelease) }) + callbackEntered := make(chan struct{}) + callbackRelease := make(chan struct{}) + releaseCallback := sync.OnceFunc(func() { close(callbackRelease) }) callbackDone := make(chan struct{}) callbackResult := make(chan error, 1) srv := b2wiretest.New(t) @@ -180,12 +183,15 @@ func TestWriterCloseWaitsForErrorCallback(t *testing.T) { } writer := bucket.Object("cancelled").NewWriter(ctx, WithCancelOnError(context.Background, func(err error) { callbackResult <- err - close(callbackDone) // No wait on Close or on any other channel. + close(callbackEntered) + <-callbackRelease // Test-controlled gate; never waits on Close. + close(callbackDone) })) // The recording server does not enforce B2's minimum part size. writer.ChunkSize = 3 t.Cleanup(func() { releaseCancel() + releaseCallback() workersDone := make(chan struct{}) go func() { writer.wg.Wait(); close(workersDone) }() select { @@ -208,12 +214,18 @@ func TestWriterCloseWaitsForErrorCallback(t *testing.T) { go func() { closed <- writer.Close() }() + releaseCancel() + select { + case <-callbackEntered: + case <-time.After(5 * time.Second): + t.Fatal("errf did not start after cancel_large_file completed") + } select { case err := <-closed: - t.Fatalf("Close returned before cancel_large_file was released: %v", err) - default: + t.Fatalf("Close returned while errf was blocked: %v", err) + case <-time.After(200 * time.Millisecond): } - releaseCancel() + releaseCallback() select { case err := <-closed: select { From 80716770553ab8e9c43abf71d3dc973b08d558d2 Mon Sep 17 00:00:00 2001 From: Felipe Fumero Date: Fri, 30 Oct 2026 13:03:26 -0600 Subject: [PATCH 10/21] test(b2): pin Writer.ReadFrom upload error precedence --- b2/b2_test.go | 20 +++--- b2/writer_readfrom_error_test.go | 110 +++++++++++++++++++++++++++++++ 2 files changed, 121 insertions(+), 9 deletions(-) create mode 100644 b2/writer_readfrom_error_test.go diff --git a/b2/b2_test.go b/b2/b2_test.go index 793fd226..d2dd4acb 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" @@ -1677,6 +1678,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{ @@ -1685,13 +1687,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, }, }, }, @@ -1707,8 +1709,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/writer_readfrom_error_test.go b/b2/writer_readfrom_error_test.go new file mode 100644 index 00000000..d32ad7fa --- /dev/null +++ b/b2/writer_readfrom_error_test.go @@ -0,0 +1,110 @@ +package b2 + +import ( + "bytes" + "context" + "errors" + "io" + "net/http" + "testing" + "time" +) + +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 = 3 + 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) + } + }) + } +} From 93ef1079cc9e6d643e89f61b0ad3087f8289e813 Mon Sep 17 00:00:00 2001 From: Felipe Fumero Date: Fri, 30 Oct 2026 13:10:25 -0600 Subject: [PATCH 11/21] fix(b2): preserve upload errors from Writer.ReadFrom --- .../unreleased/Fixed-20261030-130448.yaml | 3 +++ b2/writer.go | 20 +++++++++++++------ 2 files changed, 17 insertions(+), 6 deletions(-) create mode 100644 .changes/unreleased/Fixed-20261030-130448.yaml diff --git a/.changes/unreleased/Fixed-20261030-130448.yaml b/.changes/unreleased/Fixed-20261030-130448.yaml new file mode 100644 index 00000000..a0f5b507 --- /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/writer.go b/b2/writer.go index 7b323627..09b40bad 100644 --- a/b2/writer.go +++ b/b2/writer.go @@ -654,15 +654,23 @@ 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). +// non-seekable r, ReadFrom returns the recorded Writer error, if any, or 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 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) { + if recorded := w.getErr(); recorded != nil && !errors.Is(recorded, err) { + err = recorded + } + } + }() rs, ok := r.(io.ReadSeeker) if !ok { return copyContext(w.ctx, w, r) From 3a2d7fa8f1ef9048f919945ad9d8b5d0cf79c817 Mon Sep 17 00:00:00 2001 From: Felipe Fumero Date: Fri, 30 Oct 2026 13:35:27 -0600 Subject: [PATCH 12/21] test(b2): pin ReadFrom caller cancellation with recorded transport errors --- b2/writer_readfrom_error_test.go | 189 +++++++++++++++++++++++++++++++ 1 file changed, 189 insertions(+) diff --git a/b2/writer_readfrom_error_test.go b/b2/writer_readfrom_error_test.go index d32ad7fa..27c0287e 100644 --- a/b2/writer_readfrom_error_test.go +++ b/b2/writer_readfrom_error_test.go @@ -6,6 +6,7 @@ import ( "errors" "io" "net/http" + "sync" "testing" "time" ) @@ -108,3 +109,191 @@ func TestWriterReadFromCallerContext(t *testing.T) { }) } } + +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 = 3 + 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) { + 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(), 3*time.Second) + } 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 = 3 + 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 { + 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) + } + 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 +} From 906f795c4402e84b566d6e3d0c7e3e2d6f047af5 Mon Sep 17 00:00:00 2001 From: Felipe Fumero Date: Fri, 30 Oct 2026 13:45:46 -0600 Subject: [PATCH 13/21] fix(b2): preserve caller cancellation cause in Writer.ReadFrom --- b2/b2.go | 2 +- b2/progress_test.go | 4 ++-- b2/writer.go | 15 ++++++++------- b2/writer_log_security_test.go | 2 +- 4 files changed, 12 insertions(+), 11 deletions(-) diff --git a/b2/b2.go b/b2/b2.go index 1d9b9d7d..db908fb0 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/progress_test.go b/b2/progress_test.go index 8b51e0f7..8cf3b87f 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 09b40bad..a122a959 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 @@ -151,7 +151,7 @@ func (w *Writer) setErr(err error) { } blog.V(1).Infof("error writing object: %T", err) w.err = err - w.cancel() + w.cancel(err) w.emux.Unlock() if w.ctxf == nil || w.file == nil { return @@ -653,9 +653,10 @@ 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 recorded Writer error, if any, or the -// context error without waiting for a Read on r that is already in progress. +// On every path, ReadFrom returns the Writer's error when an upload fails, or +// the caller's context error when the caller cancels. 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. // 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, @@ -666,8 +667,8 @@ func (w *Writer) sendChunk() 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) { - if recorded := w.getErr(); recorded != nil && !errors.Is(recorded, err) { - err = recorded + if cause := context.Cause(w.ctx); cause != nil { + err = cause } } }() diff --git a/b2/writer_log_security_test.go b/b2/writer_log_security_test.go index fe890f1e..d4782fe3 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") From cad0e6d58877f3b7dadc17dab2fce5d215e7705f Mon Sep 17 00:00:00 2001 From: Felipe Fumero Date: Fri, 30 Oct 2026 13:58:33 -0600 Subject: [PATCH 14/21] test(b2): pin caller custom causes in Writer.ReadFrom --- b2/writer_readfrom_error_test.go | 38 +++++++++++++++++++++++++------- 1 file changed, 30 insertions(+), 8 deletions(-) diff --git a/b2/writer_readfrom_error_test.go b/b2/writer_readfrom_error_test.go index 27c0287e..2dbf6263 100644 --- a/b2/writer_readfrom_error_test.go +++ b/b2/writer_readfrom_error_test.go @@ -11,6 +11,9 @@ import ( "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 @@ -28,7 +31,7 @@ func TestWriterReadFromReturnsUploadError(t *testing.T) { srv.Reply("upload_part", http.StatusBadRequest, `{"status":400,"code":"bad_request","message":"part failed"}`) writer := bucket.Object("failed").NewWriter(context.Background()) - writer.ChunkSize = 3 + writer.ChunkSize = readFromTestChunkSize writer.ConcurrentUploads = 1 writer.Resume = tc.resume payload := bytes.Repeat([]byte("x"), 128) @@ -125,7 +128,7 @@ func TestWriterReadFromReturnsInjectedUploadError(t *testing.T) { bucket := newProgressReviewBucket(t) bucket.b = &readFromFailureBucket{beBucketInterface: bucket.b, failure: injected} writer := bucket.Object("injected-error").NewWriter(context.Background()) - writer.ChunkSize = 3 + writer.ChunkSize = readFromTestChunkSize writer.ConcurrentUploads = 1 writer.Resume = tc.resume payload := bytes.Repeat([]byte("x"), 128) @@ -150,19 +153,31 @@ func TestWriterReadFromReturnsInjectedUploadError(t *testing.T) { // 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 - want error + 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 { - ctx, cancel = context.WithTimeout(context.Background(), 3*time.Second) + if tc.customCause { + ctx, cancel = context.WithTimeoutCause(context.Background(), 3*time.Second, customCause) + } else { + ctx, cancel = context.WithTimeout(context.Background(), 3*time.Second) + } + } else if tc.customCause { + ctx, cancelWithCause = context.WithCancelCause(context.Background()) + cancel = func() { cancelWithCause(nil) } } else { ctx, cancel = context.WithCancel(context.Background()) } @@ -185,7 +200,7 @@ func TestWriterReadFromStreamingCallerContextWithRecordedTransportError(t *testi } writer := bucket.Object("caller-context-streaming").NewWriter(ctx, WithCancelOnError(func() context.Context { return context.Background() }, func(error) { close(recorded) })) - writer.ChunkSize = 3 + writer.ChunkSize = readFromTestChunkSize writer.ConcurrentUploads = 1 done := make(chan error, 1) finished := make(chan struct{}) @@ -220,7 +235,11 @@ func TestWriterReadFromStreamingCallerContextWithRecordedTransportError(t *testi waitReadFromEvent(t, partEntered, "part upload") waitReadFromEvent(t, bufferEntered, "next buffer") if !tc.deadline { - cancel() + if tc.customCause { + cancelWithCause(customCause) + } else { + cancel() + } } waitReadFromEvent(t, recorded, "recorded transport error") if err := writer.getErr(); !errors.Is(err, flattened) { @@ -232,6 +251,9 @@ func TestWriterReadFromStreamingCallerContextWithRecordedTransportError(t *testi 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") } From 9652b15c1380f8267c9f6606658612e1904f9c4e Mon Sep 17 00:00:00 2001 From: Felipe Fumero Date: Fri, 30 Oct 2026 14:10:16 -0600 Subject: [PATCH 15/21] fix(b2): distinguish Writer cancellation from caller causes --- b2/writer.go | 19 +++++++++++++------ 1 file changed, 13 insertions(+), 6 deletions(-) diff --git a/b2/writer.go b/b2/writer.go index a122a959..3069e2b0 100644 --- a/b2/writer.go +++ b/b2/writer.go @@ -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(err) + w.cancel(writerCause{err}) w.emux.Unlock() if w.ctxf == nil || w.file == nil { return @@ -653,9 +658,10 @@ func (w *Writer) sendChunk() error { // // Note that io.Copy will automatically choose to use ReadFrom. // -// On every path, ReadFrom returns the Writer's error when an upload fails, or -// the caller's context error when the caller cancels. If the Writer's context -// is cancelled while copying from a non-seekable r, ReadFrom returns without +// 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. // 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 @@ -667,8 +673,9 @@ func (w *Writer) sendChunk() 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) { - if cause := context.Cause(w.ctx); cause != nil { - err = cause + var cause writerCause + if errors.As(context.Cause(w.ctx), &cause) { + err = cause.err } } }() From 8c13b6d0161a326f5518981cb52474826410b0e1 Mon Sep 17 00:00:00 2001 From: Felipe Fumero Date: Fri, 30 Oct 2026 14:33:45 -0600 Subject: [PATCH 16/21] test(b2): start ReadFrom deadline after upload setup --- b2/writer_readfrom_error_test.go | 91 ++++++++++++++++++++++++++++++-- 1 file changed, 87 insertions(+), 4 deletions(-) diff --git a/b2/writer_readfrom_error_test.go b/b2/writer_readfrom_error_test.go index 2dbf6263..0e3f50ba 100644 --- a/b2/writer_readfrom_error_test.go +++ b/b2/writer_readfrom_error_test.go @@ -170,11 +170,12 @@ func TestWriterReadFromStreamingCallerContextWithRecordedTransportError(t *testi var cancel context.CancelFunc var cancelWithCause context.CancelCauseFunc if tc.deadline { + var cause error if tc.customCause { - ctx, cancel = context.WithTimeoutCause(context.Background(), 3*time.Second, customCause) - } else { - ctx, cancel = context.WithTimeout(context.Background(), 3*time.Second) + 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) } @@ -234,7 +235,10 @@ func TestWriterReadFromStreamingCallerContextWithRecordedTransportError(t *testi releaseStartOnce.Do(func() { close(releaseStart) }) waitReadFromEvent(t, partEntered, "part upload") waitReadFromEvent(t, bufferEntered, "next buffer") - if !tc.deadline { + if tc.deadline { + delayed := ctx.(*delayedTimeoutCauseContext) + delayed.start() + } else { if tc.customCause { cancelWithCause(customCause) } else { @@ -319,3 +323,82 @@ func (c *readFromFailureChunk) uploadPart(ctx context.Context, _ readResetter, _ } 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) +} From 34bb7ac9354e913c250d73e4752f3a1b6ccdc104 Mon Sep 17 00:00:00 2001 From: Felipe Fumero Date: Fri, 30 Oct 2026 14:34:13 -0600 Subject: [PATCH 17/21] docs(b2): clarify Writer ReadFrom error identity --- b2/writer.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/b2/writer.go b/b2/writer.go index 3069e2b0..1e00e64a 100644 --- a/b2/writer.go +++ b/b2/writer.go @@ -663,6 +663,9 @@ func (w *Writer) sendChunk() error { // 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, From dee1563b7b077b1e51f2877c6757b92c7e0076d6 Mon Sep 17 00:00:00 2001 From: Felipe Fumero Date: Fri, 30 Oct 2026 14:47:52 -0600 Subject: [PATCH 18/21] test(b2): bound copy cancellation and Close cleanup waits --- b2/copy_context_cleanup_test.go | 184 ++++++++++++++++++++++++++++++++ 1 file changed, 184 insertions(+) diff --git a/b2/copy_context_cleanup_test.go b/b2/copy_context_cleanup_test.go index 48e82951..d57185fa 100644 --- a/b2/copy_context_cleanup_test.go +++ b/b2/copy_context_cleanup_test.go @@ -3,6 +3,7 @@ package b2 import ( "context" "errors" + "io" "net/http" "strings" "sync" @@ -12,6 +13,189 @@ import ( "github.com/Backblaze/blazer/internal/b2wiretest" ) +func TestWriterCloseWaitsForCancelDeadline(t *testing.T) { + ctx := context.Background() + cancelEntered := make(chan struct{}) + cancelRelease := make(chan struct{}) + srv := b2wiretest.New(t) + srv.Reply("b2_list_buckets", http.StatusOK, `{"buckets":[]}`) + srv.Reply("b2_create_bucket", http.StatusOK, noReplicationBucketJSON) + 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"}`) + transport := bucketInfoRoundTripper(func(r *http.Request) (*http.Response, error) { + if strings.HasSuffix(r.URL.Path, "/b2_cancel_large_file") { + defer r.Body.Close() + close(cancelEntered) + select { + case <-r.Context().Done(): + case <-cancelRelease: + } + return nil, r.Context().Err() + } + return http.DefaultTransport.RoundTrip(r) + }) + client, err := NewClient(ctx, "account", "key", APIBase(srv.URL()), Transport(transport)) + if err != nil { + t.Fatal(err) + } + bucket, err := client.NewBucket(ctx, "copy-deadline", &BucketAttrs{Type: Private}) + if err != nil { + t.Fatal(err) + } + callbackResult := make(chan error, 1) + writer := bucket.Object("cancelled").NewWriter(ctx, WithCancelOnError(func() context.Context { + cleanupCtx, cancel := context.WithTimeout(context.Background(), 250*time.Millisecond) + t.Cleanup(cancel) + return cleanupCtx + }, func(err error) { callbackResult <- err })) + // The recording server does not enforce B2's minimum part size. + writer.ChunkSize = 3 + closed := make(chan error, 1) + t.Cleanup(func() { + close(cancelRelease) + select { + case <-closed: + case <-time.After(5 * time.Second): + t.Error("Close goroutine did not finish during cleanup") + } + }) + // Leave the worker in the cancel RPC before asking Close to wait for it. + if _, err := writer.Write([]byte("payload")); err == nil || !strings.Contains(err.Error(), "part failed") { + t.Fatalf("Write error = %v, want the worker's part failed error", err) + } + select { + case <-cancelEntered: + case <-time.After(5 * time.Second): + t.Fatal("worker did not enter cancel_large_file") + } + go func() { + closed <- writer.Close() + close(closed) + }() + select { + case err := <-closed: + if err == nil || !strings.Contains(err.Error(), "part failed") { + t.Fatalf("Close error = %v, want the recorded part failed error", err) + } + case <-time.After(5 * time.Second): + t.Fatal("Close waited past the ctxf deadline for cancel_large_file") + } + select { + case err := <-callbackResult: + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("errf result = %v, want context.DeadlineExceeded", err) + } + default: + t.Fatal("Close returned before errf recorded the cancel RPC's deadline error") + } +} + +// Retain the destination until after ReadFrom returns, then attempt a late Write. +type lateCopyWriteReader struct { + release chan struct{} + done chan struct{} + n int + err error +} + +func (*lateCopyWriteReader) Read([]byte) (int, error) { return 0, io.EOF } + +func (r *lateCopyWriteReader) WriteTo(w io.Writer) (int64, error) { + defer close(r.done) + n, err := w.Write([]byte("payload")) + <-r.release + r.n, r.err = w.Write([]byte("late")) + return int64(n), err +} + +func TestWriterReadFromCancellationCancelsStartLargeFile(t *testing.T) { + for _, fileBuffer := range []bool{false, true} { + name := "memory" + if fileBuffer { + name = "file" + } + t.Run(name, func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + startEntered := make(chan struct{}) + startRelease := make(chan struct{}) + srv := b2wiretest.New(t) + srv.Reply("b2_list_buckets", http.StatusOK, `{"buckets":[]}`) + srv.Reply("b2_create_bucket", http.StatusOK, noReplicationBucketJSON) + transport := bucketInfoRoundTripper(func(r *http.Request) (*http.Response, error) { + if strings.HasSuffix(r.URL.Path, "/b2_start_large_file") { + defer r.Body.Close() + close(startEntered) + select { + case <-r.Context().Done(): + case <-startRelease: + } + return nil, r.Context().Err() + } + return http.DefaultTransport.RoundTrip(r) + }) + client, err := NewClient(ctx, "account", "key", APIBase(srv.URL()), Transport(transport)) + if err != nil { + t.Fatal(err) + } + bucket, err := client.NewBucket(ctx, "copy-start", &BucketAttrs{Type: Private}) + if err != nil { + t.Fatal(err) + } + writer := bucket.Object("cancelled").NewWriter(ctx) + writer.ChunkSize = 3 + writer.UseFileBuffer = fileBuffer + writer.FileBufferDir = t.TempDir() + source := &lateCopyWriteReader{release: make(chan struct{}), done: make(chan struct{})} + releaseSource := sync.OnceFunc(func() { close(source.release) }) + returned := make(chan struct{}) + done := make(chan error, 1) + t.Cleanup(func() { + cancel() + close(startRelease) + releaseSource() + for _, ch := range []<-chan struct{}{source.done, returned} { + select { + case <-ch: + case <-time.After(5 * time.Second): + t.Error("copy goroutine did not finish during cleanup") + return + } + } + _ = writer.Close() + }) + go func() { + defer close(returned) + _, err := writer.ReadFrom(source) + done <- err + }() + select { + case <-startEntered: + case <-time.After(5 * time.Second): + t.Fatal("ReadFrom did not enter start_large_file") + } + cancel() + select { + case err := <-done: + if !errors.Is(err, context.Canceled) { + t.Fatalf("ReadFrom error = %v, want context.Canceled", err) + } + case <-time.After(5 * time.Second): + t.Fatal("ReadFrom waited for start_large_file after cancellation") + } + releaseSource() + select { + case <-source.done: + case <-time.After(5 * time.Second): + t.Fatal("copy goroutine did not finish after releasing the source") + } + if source.n != 0 || !errors.Is(source.err, errCopyAbandoned) { + t.Errorf("late Write = (%d, %v), want (0, b2: copy abandoned)", source.n, source.err) + } + }) + } +} + func TestWriterReadFromCancellationDoesNotWaitForCleanup(t *testing.T) { for _, blocked := range []string{"cancel_large_file", "errf"} { for _, failure := range []string{"sendChunk", "upload_part", "getErr"} { From 84b17e26d9f8e54b44019e9ae748ee3794a2329a Mon Sep 17 00:00:00 2001 From: Felipe Fumero Date: Fri, 30 Oct 2026 14:50:49 -0600 Subject: [PATCH 19/21] docs(b2): clarify cancellation and breaking Close wait contract --- .../unreleased/Changed-20261030-113335.yaml | 7 ++++++- .changes/unreleased/Fixed-20261029-145247.yaml | 5 ++++- b2/reader.go | 3 +++ b2/writer.go | 17 +++++++++++------ 4 files changed, 24 insertions(+), 8 deletions(-) diff --git a/.changes/unreleased/Changed-20261030-113335.yaml b/.changes/unreleased/Changed-20261030-113335.yaml index 548f22bf..6d6ad867 100644 --- a/.changes/unreleased/Changed-20261030-113335.yaml +++ b/.changes/unreleased/Changed-20261030-113335.yaml @@ -1,3 +1,8 @@ kind: Changed -body: Writer.Close now waits for the cancel-on-error callback before returning. +body: >- + **Breaking:** Writer.Close now waits for in-flight part uploads and, with + WithCancelOnError, for b2_cancel_large_file and errf to finish. An errf that calls + Close or waits on the goroutine calling Close deadlocks. Bound the cancellation + RPC's wait with a deadline on the context returned by ctxf; errf must return + promptly, as that deadline does not bound the callback. time: 2026-10-30T11:33:35.000000-06:00 diff --git a/.changes/unreleased/Fixed-20261029-145247.yaml b/.changes/unreleased/Fixed-20261029-145247.yaml index f0e6d3b0..d7ba669f 100644 --- a/.changes/unreleased/Fixed-20261029-145247.yaml +++ b/.changes/unreleased/Fixed-20261029-145247.yaml @@ -1,3 +1,6 @@ kind: Fixed -body: Writer.ReadFrom no longer writes to the Writer after a cancelled copy returns (#229). +body: >- + With a non-seekable reader, Writer.ReadFrom now returns the context error on + cancellation without waiting for a Read in progress, and nothing more from the + reader is written after ReadFrom returns (#229). time: 2026-10-29T14:52:47.457984-06:00 diff --git a/b2/reader.go b/b2/reader.go index 0e41e381..979365e4 100644 --- a/b2/reader.go +++ b/b2/reader.go @@ -391,6 +391,9 @@ func (w *guardedCopyWriter) Write(p []byte) (int, error) { return 0, errCopyAbandoned } if writer, ok := w.w.(*Writer); ok { + // Cancellation waits for an in-flight Write. Its network calls use the same + // context as copyContext, so their I/O is cancelled too. Only a RoundTripper + // ignoring the request context can prolong that network wait, as for any API call. return writer.write(p, func(err error) { unlock() writer.setErr(err) diff --git a/b2/writer.go b/b2/writer.go index 7b323627..9674ff1d 100644 --- a/b2/writer.go +++ b/b2/writer.go @@ -290,6 +290,8 @@ func (w *Writer) Write(p []byte) (int, error) { // A guarded copy can release its write lock there before error cleanup blocks. func (w *Writer) write(p []byte, setErr func(error)) (int, error) { w.closeWrite.RLock() + // Keep this lock through setErr cleanup: Close takes closeWrite.Lock and + // waits for cleanup anyway (TestWriterCloseWaitsForErrorCallback). defer w.closeWrite.RUnlock() if w.closed { return 0, ErrClosed @@ -732,8 +734,10 @@ func (w *Writer) ReadFrom(r io.Reader) (int64, error) { // Close satisfies the io.Closer interface. It waits for in-flight part uploads // and, when WithCancelOnError is set, for the b2_cancel_large_file call and errf -// to finish before it returns. It is critical to check the return value of Close -// for all writers. +// to finish before it returns. Give the context returned by ctxf a deadline to +// bound the cancellation RPC's wait. errf must return promptly and must not call +// Close or wait on the goroutine calling Close, or they will deadlock. The context +// deadline does not bound errf. It is critical to check Close's return value. func (w *Writer) Close() error { w.done.Do(func() { w.closeWrite.Lock() @@ -891,11 +895,12 @@ func WithSSEB2(algorithm string) WriterOption { // WithCancelOnError requests the writer, if it has started a large file // upload, to call b2_cancel_large_file on any permanent error. It calls ctxf -// to obtain a context with which to cancel the file; this is to allow callers -// to set specific timeouts. If errf is non-nil, then it is called with the +// to obtain a fresh context with which to cancel the file; give that context a +// deadline to bound the cancellation RPC's wait. If errf is non-nil, it receives the // (possibly nil) output of b2_cancel_large_file. The errf callback may run on a -// background goroutine, and Close waits for it to finish. errf must not call Close -// or wait on the goroutine that calls Close, or they will deadlock. +// background goroutine, and Close waits for it to finish. The context deadline +// does not bound errf: it must return promptly and must not call Close or wait on +// the goroutine that calls Close, or they will deadlock. func WithCancelOnError(ctxf func() context.Context, errf func(error)) WriterOption { return func(w *Writer) { w.ctxf = ctxf From 5f1c8769905af7a8c486efbb5da059364160b72b Mon Sep 17 00:00:00 2001 From: Felipe Fumero Date: Fri, 30 Oct 2026 14:53:38 -0600 Subject: [PATCH 20/21] test(b2): acknowledge cancellation stub body close results --- b2/copy_context_cleanup_test.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/b2/copy_context_cleanup_test.go b/b2/copy_context_cleanup_test.go index d57185fa..faf69455 100644 --- a/b2/copy_context_cleanup_test.go +++ b/b2/copy_context_cleanup_test.go @@ -24,7 +24,7 @@ func TestWriterCloseWaitsForCancelDeadline(t *testing.T) { srv.Reply("upload_part", http.StatusBadRequest, `{"status":400,"code":"bad_request","message":"part failed"}`) transport := bucketInfoRoundTripper(func(r *http.Request) (*http.Response, error) { if strings.HasSuffix(r.URL.Path, "/b2_cancel_large_file") { - defer r.Body.Close() + defer func() { _ = r.Body.Close() }() close(cancelEntered) select { case <-r.Context().Done(): @@ -124,7 +124,7 @@ func TestWriterReadFromCancellationCancelsStartLargeFile(t *testing.T) { srv.Reply("b2_create_bucket", http.StatusOK, noReplicationBucketJSON) transport := bucketInfoRoundTripper(func(r *http.Request) (*http.Response, error) { if strings.HasSuffix(r.URL.Path, "/b2_start_large_file") { - defer r.Body.Close() + defer func() { _ = r.Body.Close() }() close(startEntered) select { case <-r.Context().Done(): From 9d869bb713e42972f4786c871ccb0120e5191b26 Mon Sep 17 00:00:00 2001 From: Felipe Fumero Date: Fri, 30 Oct 2026 16:07:10 -0600 Subject: [PATCH 21/21] test(b2): inline the folder-delete test setup removed by #247 #247 removed objectForDeleteOptionTest from b2_test.go while #249's TestObjectDeleteRejectsFolderStatus still used it, so the b2 tests no longer compile on main. Use the same setup as TestObjectDeleteAllowsCopyStatus. --- b2/core_edges_test.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/b2/core_edges_test.go b/b2/core_edges_test.go index 481d415a..402f32f3 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) }