From db581abb0570a45382a5ab059326dcfe25facb9d Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 21 Oct 2026 00:35:26 +0200 Subject: [PATCH] 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