diff --git a/.changes/unreleased/Changed-20261030-113335.yaml b/.changes/unreleased/Changed-20261030-113335.yaml
new file mode 100644
index 0000000..6d6ad86
--- /dev/null
+++ b/.changes/unreleased/Changed-20261030-113335.yaml
@@ -0,0 +1,8 @@
+kind: Changed
+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
new file mode 100644
index 0000000..d7ba669
--- /dev/null
+++ b/.changes/unreleased/Fixed-20261029-145247.yaml
@@ -0,0 +1,6 @@
+kind: Fixed
+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/copy_context_cleanup_test.go b/b2/copy_context_cleanup_test.go
new file mode 100644
index 0000000..faf6945
--- /dev/null
+++ b/b2/copy_context_cleanup_test.go
@@ -0,0 +1,429 @@
+package b2
+
+import (
+	"context"
+	"errors"
+	"io"
+	"net/http"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"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 func() { _ = 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 func() { _ = 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"} {
+			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{})
+				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()
+				}
+				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
+				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{})
+				done := make(chan error, 1)
+				t.Cleanup(func() {
+					releaseCancel()
+					releaseCallback()
+					releasePart()
+					releaseBuffer()
+					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 == "getErr" {
+					await(bufferEntered, "producer buffer allocation")
+				}
+				if failure != "sendChunk" {
+					releasePart()
+				} else {
+					cancel()
+				}
+				entered := cancelEntered
+				if blocked == "errf" {
+					entered = callbackEntered
+				}
+				await(entered, "blocked "+blocked)
+				cancel()
+				releaseBuffer()
+				select {
+				case err := <-done:
+					// A worker's permanent error may win the race with ctx.Done.
+					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)
+					}
+				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)
+				}
+			})
+		}
+	}
+}
+
+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) })
+	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)
+	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(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 {
+		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")
+	}
+	closed := make(chan error, 1)
+	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 while errf was blocked: %v", err)
+	case <-time.After(200 * time.Millisecond):
+	}
+	releaseCallback()
+	select {
+	case err := <-closed:
+		select {
+		case <-callbackDone:
+		default:
+			t.Error("errf had not completed when Close returned")
+		}
+		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")
+	}
+	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)
+	}
+}
diff --git a/b2/copy_context_test.go b/b2/copy_context_test.go
new file mode 100644
index 0000000..86d9cf5
--- /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 := &copyRecordingWriter{}
+	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 || !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/reader.go b/b2/reader.go
index e637ae2..979365e 100644
--- a/b2/reader.go
+++ b/b2/reader.go
@@ -375,20 +375,54 @@ 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()
+	unlock := sync.OnceFunc(w.mu.Unlock)
+	defer unlock()
+	if w.stopped {
+		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)
+		})
+	}
+	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) {
+	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{})
 	go func() {
-		if _, ok := w.(*Writer); ok {
-			w = onlyWriter{w}
-		}
 		n, err = io.Copy(w, r)
 		close(done)
 	}()
@@ -399,6 +433,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()
 	}
diff --git a/b2/writer.go b/b2/writer.go
index 7d12c53..9674ff1 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,7 +283,15 @@ 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()
+	// 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
@@ -302,18 +311,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:]
@@ -646,6 +655,13 @@ 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).
+//
 // 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) {
@@ -716,12 +732,18 @@ 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. 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()
 		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())
@@ -873,9 +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
-// (possibly nil) output of b2_cancel_large_file.
+// 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. 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