From 6c7a57d53576860f1be8fa53a46ac15e16adc754 Mon Sep 17 00:00:00 2001 From: Felipe Fumero Date: Fri, 30 Oct 2026 15:34:31 -0600 Subject: [PATCH 1/3] test(b2): pin progress content and terminal delivery --- b2/progress_review_test.go | 25 ++++++++++++++++++------- b2/progress_test.go | 10 ++++++++++ 2 files changed, 28 insertions(+), 7 deletions(-) diff --git a/b2/progress_review_test.go b/b2/progress_review_test.go index 0fa68f52..c8a1adaa 100644 --- a/b2/progress_review_test.go +++ b/b2/progress_review_test.go @@ -93,7 +93,8 @@ func TestProgressResumePipeWithoutCallback(t *testing.T) { if err := pw.Close(); err != nil { t.Fatal(err) } - w := newProgressReviewBucket(t).Object("pipe").NewWriter(context.Background()) + object := newProgressReviewBucket(t).Object("pipe") + w := object.NewWriter(context.Background()) w.Resume = true n, err := w.ReadFrom(pr) if err != nil || n != int64(len("pipe payload")) { @@ -102,6 +103,12 @@ func TestProgressResumePipeWithoutCallback(t *testing.T) { if err := w.Close(); err != nil { t.Fatal(err) } + r := object.NewReader(context.Background()) + defer func() { _ = r.Close() }() + got, err := io.ReadAll(r) + if err != nil || string(got) != "pipe payload" { + t.Fatalf("uploaded content = %q, %v; want pipe payload, nil", got, err) + } } func TestProgressResumePastEOFReportsZero(t *testing.T) { @@ -144,8 +151,13 @@ func TestProgressCloseWaitsForInitialization(t *testing.T) { gate := blockingHeadBucket{bucket.b, make(chan struct{}), make(chan struct{})} bucket.b = gate r := bucket.Object("file").NewRangeReader(context.Background(), 0, 7) - terminal := make(chan struct{}, 1) - r.Progress = func(_, _ int64) { terminal <- struct{}{} } + var mu sync.Mutex + var samples []progressSample + r.Progress = func(complete, total int64) { + mu.Lock() + samples = append(samples, progressSample{complete, total}) + mu.Unlock() + } readDone := make(chan struct{}) go func() { _, _ = io.ReadAll(r); close(readDone) }() <-gate.entered @@ -166,10 +178,9 @@ func TestProgressCloseWaitsForInitialization(t *testing.T) { t.Fatal("Close did not finish after initialization") } <-readDone - select { - case <-terminal: - default: - t.Fatal("no terminal callback after Close") + got := progressSnapshot(&mu, &samples) + if len(got) == 0 || got[len(got)-1] != (progressSample{0, -1}) { + t.Fatalf("progress after Close = %+v, want final 0/-1 after cancelled initialization", got) } } diff --git a/b2/progress_test.go b/b2/progress_test.go index 8b51e0f7..1241e584 100644 --- a/b2/progress_test.go +++ b/b2/progress_test.go @@ -656,7 +656,17 @@ func TestProgressCancellationHasNoCallbackAfterTerminal(t *testing.T) { } <-readDone before := progressSnapshot(&mu, &samples) + select { + case <-reader.reporter.wake: // finish may leave a pre-terminal wake queued. + default: + } reader.reporter.add(1) // A late metered read must not produce another callback. + select { + case <-reader.reporter.wake: + t.Fatal("late progress update queued after terminal callback") + case <-time.After(100 * time.Millisecond): + // Give an incorrectly queued update time to reach the progress owner. + } after := progressSnapshot(&mu, &samples) if len(after) != len(before) { t.Fatalf("callback after terminal call: before=%+v after=%+v", before, after) From e0a1a61618c94506d5f67d7b07966b0e5c091af5 Mon Sep 17 00:00:00 2001 From: Felipe Fumero Date: Fri, 30 Oct 2026 15:34:31 -0600 Subject: [PATCH 2/3] test(b2): pin writer and reader edge assertions --- b2/writer_reader_edges_test.go | 230 +++++++++++++++++++++++++++++++++ 1 file changed, 230 insertions(+) create mode 100644 b2/writer_reader_edges_test.go diff --git a/b2/writer_reader_edges_test.go b/b2/writer_reader_edges_test.go new file mode 100644 index 00000000..c247f3ca --- /dev/null +++ b/b2/writer_reader_edges_test.go @@ -0,0 +1,230 @@ +package b2 + +import ( + "bytes" + "context" + "errors" + "io" + "net/http" + "reflect" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/Backblaze/blazer/base" + "github.com/Backblaze/blazer/internal/b2wiretest" +) + +func TestReaderChunkSizeOneAndDefaults(t *testing.T) { + for _, tc := range []struct { + name string + size int + chunk int + want []string + }{ + {"one byte", 1, 1, []string{"bytes=0-0", "bytes=1-1", "bytes=2-2"}}, + {"zero uses default", 0, 10_000_000, []string{"bytes=0-2"}}, + {"negative uses default", -1, 10_000_000, []string{"bytes=0-2"}}, + } { + t.Run(tc.name, func(t *testing.T) { + srv, bucket := newSHA1WireBucket(t) + srv.Handle("one-byte", func(req b2wiretest.Request) b2wiretest.Response { + body := map[string]string{"bytes=0-0": "a", "bytes=1-1": "b", "bytes=2-2": "c", "bytes=0-2": "abc"}[req.Header.Get("Range")] + return b2wiretest.Response{Status: http.StatusPartialContent, Body: body} + }) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + r := bucket.Object("one-byte").NewRangeReader(ctx, 0, 3) + r.ChunkSize = tc.size + defer func() { _ = r.Close() }() + r.init.Do(r.initFunc) + if r.csize != tc.chunk { + t.Fatalf("chunk size = %d, want %d", r.csize, tc.chunk) + } + got, err := io.ReadAll(r) + if err != nil || string(got) != "abc" { + t.Fatalf("ReadAll = %q, %v; want abc, nil", got, err) + } + if err := r.Close(); err != nil { + t.Fatal(err) + } + var ranges []string + for _, req := range srv.Calls("one-byte") { + ranges = append(ranges, req.Header.Get("Range")) + } + if !reflect.DeepEqual(ranges, tc.want) { + t.Fatalf("Range requests = %q, want %q", ranges, tc.want) + } + }) + } +} + +type edgeWriteBuffer struct { + writeBuffer + write func([]byte) (int, error) +} + +func (b edgeWriteBuffer) Write(p []byte) (int, error) { return b.write(p) } + +func TestWriterBufferWriteErrorIsSticky(t *testing.T) { + for _, tc := range []struct { + name string + payload string + }{ + {"within chunk", "ab"}, + {"crosses chunk", "abcd"}, + } { + t.Run(tc.name, func(t *testing.T) { + w := newProgressReviewBucket(t).Object("sticky").NewWriter(context.Background()) + w.ChunkSize = 3 + defer func() { _ = w.Close() }() + want := errors.New("buffer write failed") + calls := 0 + w.newBuffer = func() (writeBuffer, error) { + buffer := newMemoryBuffer() + return edgeWriteBuffer{buffer, func(p []byte) (int, error) { + calls++ + if calls == 1 { + return 0, want + } + return buffer.Write(p) + }}, nil + } + if n, err := w.Write([]byte(tc.payload)); n != 0 || err != want { + t.Fatalf("first Write = %d, %v; want 0, buffer write failed", n, err) + } + if n, err := w.Write([]byte("x")); n != 0 || err != want { + t.Fatalf("second Write = %d, %v; want 0, original buffer error", n, err) + } + if calls != 1 { + t.Fatalf("buffer Write calls = %d, want 1", calls) + } + }) + } +} + +func TestWriterStatusIncludesAllThreeConcurrentParts(t *testing.T) { + srv, bucket := newSHA1WireBucket(t) + srv.Reply("b2_start_large_file", http.StatusOK, `{"fileId":"status-large"}`) + srv.Reply("b2_finish_large_file", http.StatusOK, `{"fileId":"status-large","fileName":"status","action":"upload"}`) + entered := make(chan struct{}, 3) + release := make(chan struct{}) + releaseParts := sync.OnceFunc(func() { close(release) }) + srv.Handle("upload_part", func(b2wiretest.Request) b2wiretest.Response { + entered <- struct{}{} + <-release + return b2wiretest.Response{Body: `{"contentSha1":"part-sha1"}`} + }) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + w := bucket.Object("status").NewWriter(ctx) + w.ChunkSize = 3 + w.ConcurrentUploads = 3 + defer func() { _ = w.Close() }() + defer releaseParts() + if n, err := w.ReadFrom(bytes.NewReader([]byte("abcdefghi"))); n != 9 || err != nil { + t.Fatalf("ReadFrom = %d, %v; want 9, nil", n, err) + } + for range 3 { + select { + case <-entered: + case <-ctx.Done(): + t.Fatal("three concurrent parts did not enter upload") + } + } + if got := w.status().Progress; !reflect.DeepEqual(got, []float64{1, 1, 1}) { + t.Fatalf("in-flight Progress = %v, want [1 1 1]", got) + } + // Closing release lets all three uploads complete before the final snapshot. + releaseParts() + if err := w.Close(); err != nil { + t.Fatal(err) + } + if got := w.status().Progress; !reflect.DeepEqual(got, []float64{1, 1, 1}) { + t.Fatalf("final Progress = %v, want [1 1 1]", got) + } +} + +// Err snapshots the context before pausing the producer at its second handoff. +// Cancellation while paused must be observed by sendChunk's select. +type edgeHandoffContext struct { + context.Context + armed atomic.Bool + checked, release chan struct{} +} + +func (c *edgeHandoffContext) Err() error { + err := c.Context.Err() + if err == nil && c.armed.CompareAndSwap(true, false) { + close(c.checked) + <-c.release + } + return err +} + +func TestWriterDirectWriteUnblocksOnPermanentPartFailure(t *testing.T) { + srv, bucket := newSHA1WireBucket(t) + srv.Reply("b2_start_large_file", http.StatusOK, `{"fileId":"failed-large"}`) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + w := bucket.Object("direct-failure").NewWriter(ctx) + w.ChunkSize = 3 + w.ConcurrentUploads = 1 + gate := &edgeHandoffContext{Context: w.ctx, checked: make(chan struct{}), release: make(chan struct{})} + releaseGate := sync.OnceFunc(func() { close(gate.release) }) + w.ctx = gate + partEntered := make(chan struct{}) + srv.Handle("upload_part", func(b2wiretest.Request) b2wiretest.Response { + close(partEntered) + <-gate.checked + return b2wiretest.Response{Status: http.StatusBadRequest, Body: `{"status":400,"code":"bad_request","message":"permanent part failure"}`} + }) + buffers := 0 + w.newBuffer = func() (writeBuffer, error) { + buffers++ + buffer := newMemoryBuffer() + if buffers == 2 { + return edgeWriteBuffer{buffer, func(p []byte) (int, error) { + n, err := buffer.Write(p) + <-partEntered + gate.armed.Store(true) + return n, err + }}, nil + } + return buffer, nil + } + done := make(chan struct{}) + var n int + var writeErr error + go func() { n, writeErr = w.Write([]byte("abcdefghi")); close(done) }() + defer func() { + cancel() + releaseGate() + if w.cdone != nil { + close(w.cdone) // also releases the deliberately broken sendChunk select + } + <-done + _ = w.Close() + }() + select { + case <-gate.checked: + case <-time.After(2 * time.Second): + t.Fatal("producer did not reach its second chunk handoff") + } + select { + case <-w.ctx.Done(): + case <-time.After(2 * time.Second): + t.Fatal("permanent part failure did not cancel writer") + } + releaseGate() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("direct Write blocked after permanent part failure") + } + code, reason := base.Code(writeErr) + if n != 6 || code != 400 || reason != "permanent part failure" { + t.Fatalf("Write = %d, %v (status %d, message %q); want 6, 400 permanent part failure", n, writeErr, code, reason) + } +} From 2b04a3e8f8e55d6eb7fac4a211e08241a3f51ba0 Mon Sep 17 00:00:00 2001 From: Felipe Fumero Date: Fri, 30 Oct 2026 16:08:34 -0600 Subject: [PATCH 3/3] test(b2): inline the folder-delete test setup removed by #247 Same 2-line fix as in #252, so this branch's CI can run before #252 merges; the identical change merges cleanly. --- 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) }