Measure a declaration against the bus's largest message, raise a condition near it, and send the vault's export compact (issue 1689)
mesh/merge-gate out of date: judged against trunk 5be1cfaed19b, the trunk is now f10fbda1f619; judged again on a recheck or when it is merged
mesh/repo-check out of date: judged against trunk 5be1cfaed19b, the trunk is now f10fbda1f619; judged again on a recheck or when it is merged
mesh/delivery delivered
mesh/merge-gate out of date: judged against trunk 5be1cfaed19b, the trunk is now f10fbda1f619; judged again on a recheck or when it is merged
mesh/repo-check out of date: judged against trunk 5be1cfaed19b, the trunk is now f10fbda1f619; judged again on a recheck or when it is merged
mesh/delivery delivered
A declaration is one bus message and the control node's passed the limit by 2 KB. The size is checked at every send against the connection's own limit, with a condition from 90% and a message naming size and limit when over; the export, 138 KB compact of 165 KB indented, is sent compact.
This commit is contained in:
@@ -184,6 +184,14 @@ func (b OverNATS) PublishMembershipAfter(ctx context.Context, node, module strin
|
||||
return nil
|
||||
}
|
||||
|
||||
// MaxPayload is the largest message the connected server takes, as the server told this connection.
|
||||
func (b OverNATS) MaxPayload() int64 {
|
||||
if b.Conn == nil {
|
||||
return 0
|
||||
}
|
||||
return b.Conn.MaxPayload()
|
||||
}
|
||||
|
||||
func (b OverNATS) PublishDeclaration(ctx context.Context, node string, body []byte) error {
|
||||
_, err := b.JS.Publish(DeclareSubject(node), body, nats.Context(ctx))
|
||||
if err != nil {
|
||||
|
||||
@@ -46,8 +46,41 @@ func Declare(ctx context.Context, bus Bus, signer Signer, node string,
|
||||
return err
|
||||
}
|
||||
|
||||
// **Measured before it is sent** (novox/hq issue 1689): the bare bus error said only "maximum payload
|
||||
// exceeded", on 2026-10-21 for the control node's declaration, 2 KB over. The limit is the connection's own.
|
||||
if limit := payloadLimitOf(bus); limit > 0 {
|
||||
size := int64(len(body))
|
||||
if DeclarationMeasured != nil {
|
||||
DeclarationMeasured(ctx, node, size, limit)
|
||||
}
|
||||
if size > limit {
|
||||
return fmt.Errorf("%s's declaration is %d bytes on the wire (%d bytes of declaration, signed and encoded), "+
|
||||
"over the bus's largest message of %d bytes by %d: it cannot be sent (novox/hq issue 1689)",
|
||||
node, size, len(declaration), limit, size-limit)
|
||||
}
|
||||
}
|
||||
|
||||
publish, cancel := context.WithTimeout(ctx, timeout)
|
||||
defer cancel()
|
||||
|
||||
return bus.PublishDeclaration(publish, node, body)
|
||||
}
|
||||
|
||||
// NearTheLimit is the share of the bus's largest message at which a declaration is said to be near it.
|
||||
const NearTheLimit = 0.9
|
||||
|
||||
// DeclarationMeasured is told, at every send of a declaration, its size on the wire and the bus's largest message,
|
||||
// so the controller raises a condition when it nears the limit and clears it when it is back under (issue 1689).
|
||||
// Set by the controller; nil in a test or a tool.
|
||||
var DeclarationMeasured func(ctx context.Context, node string, size, limit int64)
|
||||
|
||||
// payloadLimiter is a bus that knows its largest message.
|
||||
type payloadLimiter interface{ MaxPayload() int64 }
|
||||
|
||||
// payloadLimitOf is the bus's largest message in bytes, or 0 when the bus does not say.
|
||||
func payloadLimitOf(bus Bus) int64 {
|
||||
if l, ok := bus.(payloadLimiter); ok {
|
||||
return l.MaxPayload()
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
@@ -0,0 +1,52 @@
|
||||
//go:build integration
|
||||
|
||||
package link
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/novox/mesh-controller/internal/testbus"
|
||||
)
|
||||
|
||||
// Against a real bus whose largest message is small: the limit is read from the connection, the one that fits goes
|
||||
// through, the one over is refused by us with its size and the limit, not by the client's bare error (issue 1689).
|
||||
func TestADeclarationIsMeasuredAgainstTheRealBusLimit(t *testing.T) {
|
||||
s := testbus.StartWithMaxPayload(t, 2048)
|
||||
conn, err := nats.Connect(s.ClientURL())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer conn.Close()
|
||||
js, err := conn.JetStream()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := js.AddStream(&nats.StreamConfig{Name: "NODES", Subjects: []string{"mesh.node.*.declare"}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
bus := OverNATS{Conn: conn, JS: js}
|
||||
if bus.MaxPayload() != 2048 {
|
||||
t.Fatalf("the limit read from the connection is %d, expected 2048", bus.MaxPayload())
|
||||
}
|
||||
var told []int64
|
||||
was := DeclarationMeasured
|
||||
DeclarationMeasured = func(_ context.Context, _ string, size, limit int64) { told = append(told, size, limit) }
|
||||
t.Cleanup(func() { DeclarationMeasured = was })
|
||||
|
||||
fits := []byte(`{"a":"` + strings.Repeat("x", 600) + `"}`)
|
||||
if err := Declare(t.Context(), bus, fixedSigner{}, "novox", fits, 5*time.Second); err != nil {
|
||||
t.Fatalf("a declaration under the limit: %v", err)
|
||||
}
|
||||
over := []byte(`{"a":"` + strings.Repeat("x", 1600) + `"}`)
|
||||
err = Declare(t.Context(), bus, fixedSigner{}, "novox", over, 5*time.Second)
|
||||
if err == nil || !strings.Contains(err.Error(), "2048 bytes") || strings.Contains(err.Error(), "maximum payload exceeded") {
|
||||
t.Fatalf("a declaration over the limit: %v", err)
|
||||
}
|
||||
if len(told) != 4 || told[1] != 2048 || told[2] <= 2048 {
|
||||
t.Fatalf("measured %v", told)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,57 @@
|
||||
package link
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
type roomBus struct {
|
||||
Bus
|
||||
limit int64
|
||||
sent [][]byte
|
||||
}
|
||||
|
||||
func (b *roomBus) MaxPayload() int64 { return b.limit }
|
||||
func (b *roomBus) PublishDeclaration(_ context.Context, _ string, body []byte) error {
|
||||
b.sent = append(b.sent, body)
|
||||
return nil
|
||||
}
|
||||
|
||||
type fixedSigner struct{}
|
||||
|
||||
func (fixedSigner) Sign(context.Context, []byte) ([]byte, error) { return []byte("sig"), nil }
|
||||
|
||||
// A declaration is measured as it goes on the wire against the bus's own limit; one over it is refused saying size and
|
||||
// limit, and what the measure sees is told to whoever raises a condition (issue 1689).
|
||||
func TestADeclarationOverTheBusLimitIsRefusedSayingSizeAndLimit(t *testing.T) {
|
||||
var seen []int64
|
||||
was := DeclarationMeasured
|
||||
DeclarationMeasured = func(_ context.Context, node string, size, limit int64) {
|
||||
if node != "novox" || limit != 400 {
|
||||
t.Errorf("measured %s against %d", node, limit)
|
||||
}
|
||||
seen = append(seen, size)
|
||||
}
|
||||
t.Cleanup(func() { DeclarationMeasured = was })
|
||||
|
||||
bus := &roomBus{limit: 400}
|
||||
small := []byte(`{"a":"` + strings.Repeat("x", 100) + `"}`)
|
||||
if err := Declare(t.Context(), bus, fixedSigner{}, "novox", small, time.Second); err != nil || len(bus.sent) != 1 {
|
||||
t.Fatalf("a small declaration: %v, %d sent", err, len(bus.sent))
|
||||
}
|
||||
if int64(len(bus.sent[0])) != seen[0] {
|
||||
t.Fatalf("measured %d bytes, sent %d", seen[0], len(bus.sent[0]))
|
||||
}
|
||||
|
||||
big := []byte(`{"a":"` + strings.Repeat("x", 300) + `"}`)
|
||||
err := Declare(t.Context(), bus, fixedSigner{}, "novox", big, time.Second)
|
||||
if err == nil || !strings.Contains(err.Error(), "400 bytes") || !strings.Contains(err.Error(), "novox") ||
|
||||
!strings.Contains(err.Error(), "over") || strings.Contains(err.Error(), "maximum payload exceeded") {
|
||||
t.Fatalf("an oversize declaration: %v", err)
|
||||
}
|
||||
if len(bus.sent) != 1 || len(seen) != 2 || seen[1] <= 400 {
|
||||
t.Fatalf("it was sent (%d) or not measured (%v)", len(bus.sent), seen)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user