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,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
+// under the ask's id and the message acknowledged; an answer the state refuses as too large is
-// bus's limit ends that). early acknowledges the message; it is called before a scan or a print, which are not
+// 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) {

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)
+	}
+}

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 {
+	if act, delay := w.Handle(again, nil); act != actionRetry || delay <= 0 {
-		t.Errorf("an answer that cannot be written: action %v delay %v, want term (issue 1535: no redelivery)", 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,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