Ask a failed part fetch once more before a scan fails, and name the stored object (issue 1536)

modules/brother/cmd/brother/scan.go

@@ -49,6 +49,10 @@ func (s *Service) ScanNow(ctx context.Context, source string, duplex bool) (map[
 	got := make([][]byte, 0, int(parts))
 	for i := 1; i <= int(parts); i++ {
 		p, err := s.fetchPart(ctx, key, i)
+		if err != nil && ctx.Err() == nil {
+			// A part fetch is a read, safe to repeat: one retry rides out a short bus outage (issue 1536).
+			p, err = s.fetchPart(ctx, key, i)
+		}
 		if err != nil {
 			return nil, fmt.Errorf("the scan is kept as %s, but its part %d could not be fetched: %w", key, i, err)
 		}

modules/brother/cmd/brother/service_test.go

@@ -9,6 +9,7 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -334,3 +335,63 @@ func TestLoadSettingsLaysTheDefaultsUnder(t *testing.T) {
 		t.Fatalf("settings = %+v", s)
 	}
 }
+
+// flakyBus fails the first n document asks with an error answer, as a bus outage would end one wait, and passes
+// everything else on.
+type flakyBus struct {
+	*fakeAskBus
+	mu    sync.Mutex
+	fails int
+}
+
+func (b *flakyBus) Publish(id string, body any) error {
+	raw, _ := json.Marshal(body)
+	var ask Ask
+	json.Unmarshal(raw, &ask)
+	b.mu.Lock()
+	fail := ask.Op == OpDocument && b.fails > 0
+	if fail {
+		b.fails--
+	}
+	b.mu.Unlock()
+	if fail {
+		b.answers.Create(id, map[string]any{"error": "nats: timeout"})
+		return nil
+	}
+	return b.fakeAskBus.Publish(id, body)
+}
+
+// 1536: a part fetch that fails once is asked again, and the scan is filed; one that fails twice is a failure
+// naming the stored object, so it can be filed later.
+func TestAFailedPartFetchIsAskedOnceMoreBeforeTheScanFails(t *testing.T) {
+	f := newFakeScanner(t, fixture(t, "scan.jpg"))
+	r, _ := scanRig(t, f, Settings{Crop: false})
+	r.bus = nil
+	flaky := &flakyBus{fakeAskBus: &fakeAskBus{worker: r.worker, answers: r.answers}, fails: 1}
+	askBus = flaky
+	body, err := r.scan(t)
+	if err != nil {
+		t.Fatalf("one failed part fetch ended the scan: %v", err)
+	}
+	if _, err := os.Stat(body["path"].(string)); err != nil {
+		t.Errorf("the scan is not filed: %v", err)
+	}
+
+	f2 := newFakeScanner(t, fixture(t, "scan.jpg"))
+	r2, folder := scanRig(t, f2, Settings{Crop: false})
+	askBus = &flakyBus{fakeAskBus: &fakeAskBus{worker: r2.worker, answers: r2.answers}, fails: 2}
+	_, err = r2.scan(t)
+	if err == nil {
+		t.Fatal("two failed part fetches did not fail the scan")
+	}
+	key := ""
+	if got := *r2.said; len(got) == 1 {
+		key, _ = got[0].body["object"].(string)
+	}
+	if key == "" || !strings.Contains(err.Error(), key) {
+		t.Errorf("the failure %q does not name the stored object %q", err, key)
+	}
+	if entries, _ := os.ReadDir(folder); len(entries) != 0 {
+		t.Errorf("a scan whose parts failed was filed: %v", entries)
+	}
+}