From 973d9991017ae49d83a36cd58cee5121e96f88f2 Mon Sep 17 00:00:00 2001 From: Felipe Fumero Date: Thu, 29 Oct 2026 14:50:04 -0600 Subject: [PATCH 01/12] 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/12] 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/12] 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/12] 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/12] 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/12] 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/12] 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/12] 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/12] 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 dee1563b7b077b1e51f2877c6757b92c7e0076d6 Mon Sep 17 00:00:00 2001 From: Felipe Fumero Date: Fri, 30 Oct 2026 14:47:52 -0600 Subject: [PATCH 10/12] 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 11/12] 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 12/12] 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():