diff --git a/modules/brother/cmd/brother/deviceproc.go b/modules/brother/cmd/brother/deviceproc.go index c615ca30..bf13f742 100644 --- a/modules/brother/cmd/brother/deviceproc.go +++ b/modules/brother/cmd/brother/deviceproc.go @@ -62,8 +62,12 @@ type deviceWorker struct { Emit func(body map[string]any, id string) error Answers Answers Node string - Now func() time.Time - Logf func(format string, a ...any) + // Capture scans; nil is the device's own (a test puts a stand-in). + Capture func(ctx context.Context) (Capture, error) + // answerPart is the raw size of a document part in an answer; zero is AnswerPartSize (a test makes it larger). + answerPart int + Now func() time.Time + Logf func(format string, a ...any) mu sync.Mutex pending map[string]*pendingUpload @@ -105,10 +109,16 @@ func (w *deviceWorker) Handle(body []byte, early func()) (action, time.Duration) } if _, err := w.Answers.Create(ask.ID, answer); err != nil && !errors.Is(err, stdio.ErrChanged) { w.logf("[brother] device: the answer to %s could not be written: %v", ask.ID, err) + // The ask ends here: a refused answer would be refused again, and a redelivery would do the work again. + // Say the failure as a small error answer, so the tool's wait ends with the reason (issue 1535). + msg := fmt.Sprintf("the answer to this ask could not be written in the state %s: %v", AnswersState, err) + if _, err2 := w.Answers.Create(ask.ID, map[string]any{"error": msg}); err2 != nil && !errors.Is(err2, stdio.ErrChanged) { + w.logf("[brother] device: the failure of %s could not be written either: %v", ask.ID, err2) + } if settled { return actionSettled, 0 } - return actionRetry, 5 * time.Second + return actionTerm, 0 } if settled { return actionSettled, 0 @@ -150,7 +160,11 @@ func (w *deviceWorker) scan(ctx context.Context) (map[string]any, error) { return nil, err } svc := &Service{Settings: set, Node: w.Node, Now: w.Now} - c, err := svc.Capture(ctx) + capture := svc.Capture + if w.Capture != nil { + capture = w.Capture + } + c, err := capture(ctx) if err != nil { return nil, err } @@ -162,7 +176,7 @@ func (w *deviceWorker) scan(ctx context.Context) (map[string]any, error) { if err := w.Emit(body, c.SHA256); err != nil { return nil, fmt.Errorf("the scan is kept as %s, but %s could not be said on the bus: %w", key, EventAdded, err) } - answer := map[string]any{"key": key, "size": len(c.PDF), "parts": partCount(len(c.PDF))} + answer := map[string]any{"key": key, "size": len(c.PDF), "parts": answerPartCount(len(c.PDF))} for k, v := range body { answer[k] = v } @@ -182,7 +196,11 @@ func (w *deviceWorker) document(ctx context.Context, ask Ask) (map[string]any, e if err != nil { return nil, err } - part, n, err := cutPart(data, ask.Part) + each := w.answerPart + if each <= 0 { + each = AnswerPartSize + } + part, n, err := cutPartOf(data, ask.Part, each) if err != nil { return nil, err } diff --git a/modules/brother/cmd/brother/documents_test.go b/modules/brother/cmd/brother/documents_test.go index ae111a76..9046bd33 100644 --- a/modules/brother/cmd/brother/documents_test.go +++ b/modules/brother/cmd/brother/documents_test.go @@ -2,12 +2,15 @@ package main import ( "bytes" + "context" "encoding/base64" "encoding/binary" "encoding/json" "io" "net/http" "net/http/httptest" + "os" + "path/filepath" "strings" "testing" "time" @@ -28,8 +31,8 @@ func pdfOf(size int) []byte { // order are the document, with the SHA-256 checked; a wrong SHA-256 or a part missing is refused; a part out of // range is refused naming the range. func TestADocumentIsCutIntoPartsAndJoinedWithItsSha256Checked(t *testing.T) { - doc := pdfOf(2*PartSize + 1234) - n := partCount(len(doc)) + doc := pdfOf(2*AnswerPartSize + 1234) + n := answerPartCount(len(doc)) if n != 3 { t.Fatalf("%d parts, want 3", n) } @@ -39,8 +42,8 @@ func TestADocumentIsCutIntoPartsAndJoinedWithItsSha256Checked(t *testing.T) { if err != nil || total != n { t.Fatalf("part %d: %v (of %d)", i, err, total) } - if len(p) > MaxPart { - t.Errorf("part %d is %d bytes, over %d", i, len(p), MaxPart) + if len(p) > AnswerPartSize { + t.Errorf("part %d is %d bytes, over %d", i, len(p), AnswerPartSize) } parts = append(parts, p) } @@ -69,7 +72,7 @@ func TestADocumentIsCutIntoPartsAndJoinedWithItsSha256Checked(t *testing.T) { // refused, the first before anything is queued. func TestScannerDocumentGivesEachPartOfAKeptScan(t *testing.T) { r := newRig(t, `{}`, "pw") - doc := pdfOf(PartSize + 5000) + doc := pdfOf(AnswerPartSize + 5000) key := sha256Hex(doc) + ".pdf" r.objects.Put(t.Context(), key, "application/pdf", doc) var joined []byte @@ -117,7 +120,7 @@ func TestScanNowFetchesAScanOfSeveralParts(t *testing.T) { r, _ := scanRig(t, f, Settings{Crop: false}) // Everything a scan is made of but its size is the real one; the stored object is what the parts carry, so a // bigger object stands for a bigger scan: replace what the device keeps with a three-part document. - big := pdfOf(2*PartSize + 99) + big := pdfOf(2*AnswerPartSize + 99) bigKey := sha256Hex(big) + ".pdf" r.objects.Put(t.Context(), bigKey, "application/pdf", big) got := make([][]byte, 0, 3) @@ -153,8 +156,8 @@ func TestAnUploadIsStoredOnceAllItsPartsCameAndTheSha256Matches(t *testing.T) { r := newRig(t, `{}`, "pw") doc := pdfOf(PartSize + 700) sum := sha256Hex(doc) - p1, _, _ := cutPart(doc, 1) - p2, _, _ := cutPart(doc, 2) + p1, _, _ := cutPartOf(doc, 1, PartSize) + p2, _, _ := cutPartOf(doc, 2, PartSize) got, err := upload(t, r, "up-1", doc, sum, 2, 2, p2) if err != nil { t.Fatal(err) @@ -224,7 +227,7 @@ func TestAnUploadWithAWrongSha256OrAnOutOfRangePartStoresNothing(t *testing.T) { } // Parts that disagree about the whole. p := pdfOf(PartSize + 10) - p1, _, _ := cutPart(p, 1) + p1, _, _ := cutPartOf(p, 1, PartSize) if _, err := upload(t, r, "mix", p, sha256Hex(p), 1, 2, p1); err != nil { t.Fatal(err) } @@ -240,8 +243,8 @@ func TestAnUploadWithAWrongSha256OrAnOutOfRangePartStoresNothing(t *testing.T) { func TestAnIncompleteUploadIsDroppedAfterAWhile(t *testing.T) { r := newRig(t, `{}`, "pw") doc := pdfOf(PartSize + 10) - p1, _, _ := cutPart(doc, 1) - p2, _, _ := cutPart(doc, 2) + p1, _, _ := cutPartOf(doc, 1, PartSize) + p2, _, _ := cutPartOf(doc, 2, PartSize) now := fixedNow r.worker.Now = func() time.Time { return now } if _, err := upload(t, r, "late", doc, sha256Hex(doc), 1, 2, p1); err != nil { @@ -459,3 +462,59 @@ func TestAPartOverSevenHundredKiBIsRefusedByTheModulesOwnMessage(t *testing.T) { t.Errorf("a part of exactly 700 KiB: %v", err) } } + +// 1535: a scan larger than the state's value limit (256 KiB) is filed whole, its SHA-256 equal, and the answer to +// the scan itself is small: it carries the event's body and never the document. Every part answer fits the state. +func TestAScanLargerThanTheStatesValueLimitIsFiledWholeAndItsAnswerIsSmall(t *testing.T) { + f := newFakeScanner(t, fixture(t, "scan.jpg")) + r, folder := scanRig(t, f, Settings{Crop: false}) + // The real capture is small; the worker keeps a stand-in of 3 parts' size under the key the scan names. + big := pdfOf(222388 + 2*AnswerPartSize) + if len(big) <= StateMaxValue { + t.Fatal("the stand-in is not larger than the state's value limit") + } + r.worker.Capture = func(ctx context.Context) (Capture, error) { + return Capture{PDF: big, SHA256: sha256Hex(big), Resolution: 300, At: fixedNow}, nil + } + body, err := r.scan(t) + if err != nil { + t.Fatal(err) + } + path, _ := body["path"].(string) + got, err := os.ReadFile(path) + if err != nil || sha256Hex(got) != sha256Hex(big) || filepath.Dir(path) != folder { + t.Fatalf("filed %s: %v, SHA-256 equal %v", path, err, err == nil && sha256Hex(got) == sha256Hex(big)) + } + for id, raw := range r.answers.m { + if len(raw) > StateMaxValue { + t.Errorf("answer %s is %d bytes, over the state's %d", id, len(raw), StateMaxValue) + } + } + for _, a := range r.bus.published { + if a.Op == OpScan { + if raw := r.answers.get(a.ID); len(raw) > 4<<10 || bytes.Contains(raw, []byte(`"data"`)) { + t.Errorf("the scan's answer is %d bytes and may carry the document: %s", len(raw), raw) + } + } + } +} + +// 1535: an answer the state refuses ends the ask: a small error answer says why, and the queued message is +// terminated, never redelivered into the same failure; a scan's message, acknowledged early, stays settled. +func TestARefusedAnswerEndsTheAskWithAnErrorAndNoRedelivery(t *testing.T) { + r := newRig(t, `{}`, "pw") + big := pdfOf(300 << 10) + key := sha256Hex(big) + ".pdf" + r.objects.Put(t.Context(), key, "application/pdf", big) + // A part answer of 700 KiB, as the old cut made it, is over the state's limit; the worker must not loop on it. + r.worker.answerPart = PartSize + ask, _ := json.Marshal(Ask{ID: "big1", Op: OpDocument, Object: key, Part: 1, At: fixedNow}) + act, _ := r.worker.Handle(ask, nil) + if act != actionTerm { + t.Errorf("action = %v, want term", act) + } + raw := r.answers.get("big1") + if raw == nil || !bytes.Contains(raw, []byte(`"error"`)) || len(raw) > 4<<10 { + t.Errorf("no small error answer: %s", raw) + } +} diff --git a/modules/brother/cmd/brother/parts.go b/modules/brother/cmd/brother/parts.go index df357ded..34571b9a 100644 --- a/modules/brother/cmd/brother/parts.go +++ b/modules/brother/cmd/brother/parts.go @@ -14,6 +14,15 @@ const ( // it stay under the bus's 1 MiB. (ADR 0468 clause 3.) PartSize = 700 << 10 MaxPart = PartSize + // AnswerPartSize is the raw size of a part the device process answers in the state device-answers + // (scanner.document). The state refuses a value over 256 KiB (the mesh's cap on every bucket's value), and a + // part travels base64-encoded in JSON, so 128 KiB (about 175 KB written) fits with room to spare; a part of + // 700 KiB (about 933 KB) can only cross the bus as a message, never as a state value (issue 1535). + AnswerPartSize = 128 << 10 + // StateMaxValue is the most a state value may hold. + StateMaxValue = 256 << 10 + // MostAnswerParts bounds how many answer parts a document has (a 100 MiB document is the most). + MostAnswerParts = 800 // MostParts bounds how many parts one document has (a 100 MiB document is the most). MostParts = 150 ) @@ -24,21 +33,30 @@ func sha256Hex(b []byte) string { } // partCount is how many parts of PartSize hold size bytes: at least one. -func partCount(size int) int { +func partCount(size int) int { return partsOf(size, PartSize) } + +// answerPartCount is how many parts of AnswerPartSize hold size bytes: at least one. +func answerPartCount(size int) int { return partsOf(size, AnswerPartSize) } + +func partsOf(size, each int) int { if size <= 0 { return 1 } - return (size + PartSize - 1) / PartSize + return (size + each - 1) / each } -// cutPart is part number part (counted from 1) of data, and how many parts there are. +// cutPart is part number part (counted from 1) of data, in parts of AnswerPartSize, and how many parts there are. func cutPart(data []byte, part int) ([]byte, int, error) { - n := partCount(len(data)) + return cutPartOf(data, part, AnswerPartSize) +} + +func cutPartOf(data []byte, part, each int) ([]byte, int, error) { + n := partsOf(len(data), each) if part < 1 || part > n { return nil, n, fmt.Errorf("part %d is out of range: the document has parts 1 to %d", part, n) } - lo := (part - 1) * PartSize - hi := min(lo+PartSize, len(data)) + lo := (part - 1) * each + hi := min(lo+each, len(data)) return data[lo:hi], n, nil } diff --git a/modules/brother/cmd/brother/printer_test.go b/modules/brother/cmd/brother/printer_test.go index 2ef711a3..1619ee6c 100644 --- a/modules/brother/cmd/brother/printer_test.go +++ b/modules/brother/cmd/brother/printer_test.go @@ -207,8 +207,8 @@ func TestTheAdminWorkerSettlesEachAsk(t *testing.T) { _ = conns r.answers.fail = fmt.Errorf("the bus is away") again, _ := json.Marshal(Ask{ID: "a2", Op: OpSettings, Page: DefaultPage, At: fixedNow}) - if act, delay := w.Handle(again, nil); act != actionRetry || delay <= 0 { - t.Errorf("an answer that cannot be written: action %v delay %v, want retry", act, delay) + if act, delay := w.Handle(again, nil); act != actionTerm || delay != 0 { + t.Errorf("an answer that cannot be written: action %v delay %v, want term (issue 1535: no redelivery)", act, delay) } } diff --git a/modules/brother/cmd/brother/rig_test.go b/modules/brother/cmd/brother/rig_test.go index 0d2f0d23..a7f98757 100644 --- a/modules/brother/cmd/brother/rig_test.go +++ b/modules/brother/cmd/brother/rig_test.go @@ -38,8 +38,8 @@ func (a *memAnswers) Create(key string, v any) (uint64, error) { a.m = map[string]json.RawMessage{} } raw, _ := json.Marshal(v) - if len(raw) > 1<<20 { - return 0, fmt.Errorf("an answer of %d bytes is over the bus's 1 MiB", len(raw)) + if len(raw) > StateMaxValue { + return 0, fmt.Errorf("nats: API error: code=400 err_code=10054 description=message size exceeds maximum allowed (%d bytes)", len(raw)) } a.m[key] = raw return 1, nil diff --git a/modules/brother/cmd/brother/scan.go b/modules/brother/cmd/brother/scan.go index 97b07928..0c939669 100644 --- a/modules/brother/cmd/brother/scan.go +++ b/modules/brother/cmd/brother/scan.go @@ -40,7 +40,7 @@ func (s *Service) ScanNow(ctx context.Context) (map[string]any, error) { key, _ := ans["key"].(string) sum, _ := ans["sha256"].(string) parts, _ := ans["parts"].(float64) - if key == "" || sum == "" || parts < 1 || parts > MostParts { + if key == "" || sum == "" || parts < 1 || parts > MostAnswerParts { return nil, fmt.Errorf("the device process's answer to the scan names no stored scan: %v", ans) } got := make([][]byte, 0, int(parts))