novox/mesh-catalog
17
0
Fork 0
Code Issues Pull Requests 3 Actions Packages Projects Releases Wiki Activity

Merge pull request 'File a scan of any size: answer its parts in 128 KiB under the state's 256 KiB value limit, and end only a too-large answer at once (issue 1535)' (#394) from fix/1535-scan-answer-size into main

Browse Source
This commit was merged in pull request #394.
This commit is contained in:
mesh-tools
2026-10-20 22:47:14 +00:00
parent 3811ac69ed db581abb05
commit a28398c16b
6 changed files with 159 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/brother/cmd/brother/deviceproc.go
+39 -8
View File
@@ -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"
)
@@ -62,8 +63,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
@@ -75,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
@@ -105,10 +114,24 @@ 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)
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) {
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 +173,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 +189,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 +209,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
}
modules/brother/cmd/brother/documents_test.go
+91 -11
View File
@@ -2,12 +2,16 @@ package main
import (
"bytes"
"context"
"encoding/base64"
"encoding/binary"
"encoding/json"
"fmt"
"io"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"strings"
"testing"
"time"
@@ -28,8 +32,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 +43,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 +73,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 +121,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 +157,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 +228,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 +244,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 +463,79 @@ 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)
}
}
// 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)
}
}
modules/brother/cmd/brother/parts.go
+24 -6
View File
@@ -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
}
modules/brother/cmd/brother/printer_test.go
+1 -1
View File
@@ -207,7 +207,7 @@ func TestTheAdminWorkerSettlesEachAsk(t *testing.T) {
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)
t.Errorf("an answer that cannot be written for want of the bus: action %v delay %v, want retry", act, delay)
}
}
modules/brother/cmd/brother/rig_test.go
+3 -2
View File
@@ -3,6 +3,7 @@ package main
import (
"encoding/json"
"fmt"
"github.com/nats-io/nats.go/jetstream"
"os"
"path/filepath"
"sync"
@@ -38,8 +39,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("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
modules/brother/cmd/brother/scan.go
+1 -1
View File
@@ -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))