83 lines
3.1 KiB
Go
83 lines
3.1 KiB
Go
//go:build integration
|
|
|
|
package link
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
|
|
"github.com/novox/mesh-controller/internal/testbus"
|
|
)
|
|
|
|
// novox/hq issue 314, replayed with only what the link had before its fix, so it can be laid over the
|
|
// older commit. On 2026-10-08 the controller's `conditions` answered 689 stalled deliveries, each an
|
|
// open condition with its evidence, and `status` led with them: more than the bus carries in one message
|
|
// (its max_payload, a mebibyte). The client library refused the answer before it left the controller —
|
|
// `nats: maximum payload exceeded`, in the controller's journal only — and every caller, the console and
|
|
// the operator's channel among them, waited out its timeout and read that the controller did not answer,
|
|
// while `calls` said each call answered in 130 ms. An answer larger than one message arrives whole to a
|
|
// caller that pages, and is said, in one message that fits, to one that does not — never silence.
|
|
func TestReplay314(t *testing.T) {
|
|
conn, err := nats.Connect(testbus.URL(t))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer conn.Close()
|
|
limit := conn.MaxPayload()
|
|
|
|
// Three times what one message carries: the size of the answers of that morning.
|
|
type finding struct {
|
|
Key, Summary string
|
|
}
|
|
var large []finding
|
|
for len(large)*200 < int(3*limit) {
|
|
large = append(large, finding{Key: "delivery.d-" + strings.Repeat("7", 8) + ".stalled",
|
|
Summary: strings.Repeat("held past its bound ", 9)})
|
|
}
|
|
stop, err := OverNATS{Conn: conn}.ServeSeatTools("replay-314", map[string]ToolHandler{
|
|
"conditions": func(context.Context, json.RawMessage) (any, error) { return large, nil },
|
|
}, nil)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer stop()
|
|
|
|
t.Run("a caller of the controller's own reads it whole", func(t *testing.T) {
|
|
answer, err := AskMeshSeatTool(context.Background(), conn, "replay-314", "conditions", map[string]any{},
|
|
5*time.Second)
|
|
if err != nil {
|
|
t.Fatalf("an answer of about %d bytes did not arrive: %v", 3*limit, err)
|
|
}
|
|
if answer.Error != "" {
|
|
t.Fatalf("answered an error: %.300s", answer.Error)
|
|
}
|
|
var got []finding
|
|
if err := json.Unmarshal(answer.Result, &got); err != nil || len(got) != len(large) {
|
|
t.Fatalf("answered %d findings of %d (%v)", len(got), len(large), err)
|
|
}
|
|
})
|
|
|
|
t.Run("a caller that does not page is told, not left waiting", func(t *testing.T) {
|
|
reply, err := conn.Request(SeatToolSubject("replay-314", "conditions"), []byte(`{}`), 5*time.Second)
|
|
if err != nil {
|
|
t.Fatalf("no answer at all: %v — the caller waits out its timeout and reads silence", err)
|
|
}
|
|
if int64(len(reply.Data)) > limit {
|
|
t.Fatalf("an answer of %d bytes, more than the bus carries", len(reply.Data))
|
|
}
|
|
var r struct {
|
|
Error string `json:"error"`
|
|
}
|
|
if json.Unmarshal(reply.Data, &r) != nil || !strings.Contains(r.Error, "more than the bus carries") ||
|
|
!bytes.Contains(reply.Data, []byte("ask for less")) && !bytes.Contains(reply.Data, []byte("Ask for less")) {
|
|
t.Fatalf("the answer does not say it was too large and what to do: %.400s", reply.Data)
|
|
}
|
|
})
|
|
}
|