From 61796cd3ddf14a7597340dd7670967de2d26f28d Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 21 Oct 2026 00:33:00 +0200 Subject: [PATCH 1/2] Answer a scan's parts in 128 KiB, under the state's 256 KiB value limit, and end an ask whose answer is refused instead of redelivering it (issue 1535) --- modules/brother/cmd/brother/deviceproc.go | 30 +++++-- modules/brother/cmd/brother/documents_test.go | 81 ++++++++++++++++--- modules/brother/cmd/brother/parts.go | 30 +++++-- modules/brother/cmd/brother/printer_test.go | 4 +- modules/brother/cmd/brother/rig_test.go | 4 +- modules/brother/cmd/brother/scan.go | 2 +- 6 files changed, 123 insertions(+), 28 deletions(-) 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)) From db581abb0570a45382a5ab059326dcfe25facb9d Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 21 Oct 2026 00:35:26 +0200 Subject: [PATCH 2/2] Terminate an ask only for an answer the state refuses as too large; retry any other failed answer write (issue 1535) --- modules/brother/cmd/brother/deviceproc.go | 19 ++++++++++++++--- modules/brother/cmd/brother/documents_test.go | 21 +++++++++++++++++++ modules/brother/cmd/brother/printer_test.go | 4 ++-- modules/brother/cmd/brother/rig_test.go | 3 ++- 4 files changed, 41 insertions(+), 6 deletions(-) diff --git a/modules/brother/cmd/brother/deviceproc.go b/modules/brother/cmd/brother/deviceproc.go index bf13f742..2327869e 100644 --- a/modules/brother/cmd/brother/deviceproc.go +++ b/modules/brother/cmd/brother/deviceproc.go @@ -19,6 +19,7 @@ import ( stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/process" + "github.com/nats-io/nats.go/jetstream" "brother/cmd/brother/internal/secret" ) @@ -79,10 +80,14 @@ func (w *deviceWorker) logf(format string, a ...any) { } } +// tooLargeCode is the bus's error code for a message over the maximum allowed (the state refuses a value so). +const tooLargeCode = 10054 + // Handle answers one queued ask. An ask that cannot be read is terminated; an ask too old for anyone to wait for // is acknowledged and the device not touched; every other outcome, a failure included, is written as the answer -// under the ask's id and the message acknowledged; only an answer that could not be written is redelivered (the -// bus's limit ends that). early acknowledges the message; it is called before a scan or a print, which are not +// under the ask's id and the message acknowledged; an answer the state refuses as too large is +// replaced by a small error answer and the message terminated; an answer that could not be written for another +// reason (the store away, a timeout) is redelivered (the bus's limit ends that). early acknowledges the message; it is called before a scan or a print, which are not // done twice. func (w *deviceWorker) Handle(body []byte, early func()) (action, time.Duration) { var ask Ask @@ -109,7 +114,15 @@ 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. + var api *jetstream.APIError + if !errors.As(err, &api) || api.ErrorCode != tooLargeCode { + // The store is away, timed out or cut: the answer may be written next time (a scan is settled already). + if settled { + return actionSettled, 0 + } + return actionRetry, 5 * time.Second + } + // The ask ends here: a refused value would be refused 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) { diff --git a/modules/brother/cmd/brother/documents_test.go b/modules/brother/cmd/brother/documents_test.go index 9046bd33..9e96c1ef 100644 --- a/modules/brother/cmd/brother/documents_test.go +++ b/modules/brother/cmd/brother/documents_test.go @@ -6,6 +6,7 @@ import ( "encoding/base64" "encoding/binary" "encoding/json" + "fmt" "io" "net/http" "net/http/httptest" @@ -518,3 +519,23 @@ func TestARefusedAnswerEndsTheAskWithAnErrorAndNoRedelivery(t *testing.T) { t.Errorf("no small error answer: %s", raw) } } + +// 1535: only a refused value (error code 10054, message too large) ends the ask; a store that is away, a timeout +// or a cut connection is redelivered, a document fetch being safe to repeat. +func TestOnlyATooLargeAnswerEndsTheAskATransientFailureIsRetried(t *testing.T) { + r := newRig(t, `{}`, "pw") + big := pdfOf(300 << 10) + key := sha256Hex(big) + ".pdf" + r.objects.Put(t.Context(), key, "application/pdf", big) + ask, _ := json.Marshal(Ask{ID: "t1", Op: OpDocument, Object: key, Part: 1, At: fixedNow}) + r.answers.fail = fmt.Errorf("nats: timeout") + if act, d := r.worker.Handle(ask, nil); act != actionRetry || d <= 0 { + t.Errorf("a timeout: action %v delay %v, want retry", act, d) + } + r.answers.fail = nil + r.worker.answerPart = PartSize + ask2, _ := json.Marshal(Ask{ID: "t2", Op: OpDocument, Object: key, Part: 1, At: fixedNow}) + if act, _ := r.worker.Handle(ask2, nil); act != actionTerm { + t.Errorf("a too large answer: action %v, want term", act) + } +} diff --git a/modules/brother/cmd/brother/printer_test.go b/modules/brother/cmd/brother/printer_test.go index 1619ee6c..7e217755 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 != actionTerm || delay != 0 { - t.Errorf("an answer that cannot be written: action %v delay %v, want term (issue 1535: no redelivery)", act, delay) + if act, delay := w.Handle(again, nil); act != actionRetry || delay <= 0 { + t.Errorf("an answer that cannot be written for want of the bus: action %v delay %v, want retry", act, delay) } } diff --git a/modules/brother/cmd/brother/rig_test.go b/modules/brother/cmd/brother/rig_test.go index a7f98757..75c6b7d9 100644 --- a/modules/brother/cmd/brother/rig_test.go +++ b/modules/brother/cmd/brother/rig_test.go @@ -3,6 +3,7 @@ package main import ( "encoding/json" "fmt" + "github.com/nats-io/nats.go/jetstream" "os" "path/filepath" "sync" @@ -39,7 +40,7 @@ func (a *memAnswers) Create(key string, v any) (uint64, error) { } raw, _ := json.Marshal(v) 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)) + return 0, fmt.Errorf("brother's state device-answers, key %s: nats: %w", key, &jetstream.APIError{Code: 400, ErrorCode: jetstream.ErrorCode(10054), Description: "message size exceeds maximum allowed"}) } a.m[key] = raw return 1, nil