From 09aa063a130d73ed99f154199d4e7c2fb22acb7c Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 21 Oct 2026 13:54:11 +0200 Subject: [PATCH] Measure a declaration against the bus's largest message, raise a condition near it, and send the vault's export compact (issue 1689) 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. --- cmd/mesh-controller/compose_sdk.go | 2 +- cmd/mesh-controller/declaration_room.go | 96 +++++++++++++++++++ cmd/mesh-controller/declaration_room_test.go | 71 ++++++++++++++ cmd/mesh-controller/plain_words.go | 15 +++ internal/link/bus.go | 8 ++ internal/link/declare.go | 33 +++++++ .../link/declare_room_integration_test.go | 52 ++++++++++ internal/link/declare_room_test.go | 57 +++++++++++ internal/testbus/testbus.go | 8 +- 9 files changed, 339 insertions(+), 3 deletions(-) create mode 100644 cmd/mesh-controller/declaration_room.go create mode 100644 cmd/mesh-controller/declaration_room_test.go create mode 100644 internal/link/declare_room_integration_test.go create mode 100644 internal/link/declare_room_test.go diff --git a/cmd/mesh-controller/compose_sdk.go b/cmd/mesh-controller/compose_sdk.go index 56653cf5..02aae949 100644 --- a/cmd/mesh-controller/compose_sdk.go +++ b/cmd/mesh-controller/compose_sdk.go @@ -366,7 +366,7 @@ func sayFound(ctx context.Context, d compose.Declared) { // sendableOf is a composed declaration as the controller's sends know it. func sendableOf(d compose.Declared) sendable { - out := sendable{Resources: d.Resources, Sequence: d.Sequence, Epoch: d.Epoch, Generation: d.Generation, + out := sendable{Resources: compactKept(d.Resources), Sequence: d.Sequence, Epoch: d.Epoch, Generation: d.Generation, ReadsModules: d.ReadsModules, moduleOf: d.ModuleOf, buildIDs: d.BuildIDs, modules: d.Modules, localNetwork: d.LocalNetwork, Received: d.Received, Mesh: d.Mesh, BusUsers: d.BusUsers, LeftOut: d.LeftOut, leftOutWhy: d.LeftOutWhy, withheld: d.Withheld, unbound: d.Unbound, foreseen: d.Foreseen, diff --git a/cmd/mesh-controller/declaration_room.go b/cmd/mesh-controller/declaration_room.go new file mode 100644 index 00000000..827348c4 --- /dev/null +++ b/cmd/mesh-controller/declaration_room.go @@ -0,0 +1,96 @@ +package main + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "strings" + + "github.com/novox/mesh-controller/internal/conditions" + "github.com/novox/mesh-controller/internal/link" +) + +// **A declaration is one bus message, and the bus's largest message is a cliff** (novox/hq issue 1689). On 2026-10-21 +// the control node's declaration passed it by 2 KB and nothing reached that node until two assignments were undone. +// So the size is measured at every send: a warning condition from 90% of the limit, urgent over it, cleared when +// the declaration is back under 90%. The limit is the connection's own, not a constant. + +// kindDeclarationNearLimit is a declaration at or near the bus's largest message. +const kindDeclarationNearLimit = "declaration-near-limit" + +// declarationRoomSource is the condition's source: the sends. +const declarationRoomSource = "sends" + +func declarationRoomObservation(node string, size, limit int64) conditions.Observation { + o := conditions.Observation{Scope: conditions.ScopeMachine, ID: node, Kind: kindDeclarationNearLimit, + Token: kindDeclarationNearLimit, Machine: node, Severity: conditions.Warning, Source: declarationRoomSource} + if size > limit { + o.Severity = conditions.Urgent + } + o.Summary = fmt.Sprintf("%s's declaration is %d bytes on the wire, %d%% of the bus's largest message of %d bytes", + node, size, size*100/limit, limit) + if size > limit { + o.Summary += ": over it, so it cannot be sent" + } + return o +} + +func init() { + link.DeclarationMeasured = func(ctx context.Context, node string, size, limit int64) { + settleDeclarationRoom(ctx, conditionsFrom, node, size, limit) + } +} + +// settleDeclarationRoom raises or clears the condition of a declaration's size. A nil keeper (a tool, not the +// serving controller) says nothing. +func settleDeclarationRoom(ctx context.Context, keeper *conditions.Keeper, node string, size, limit int64) { + if keeper == nil || limit <= 0 { + return + } + obs := declarationRoomObservation(node, size, limit) + if float64(size) < link.NearTheLimit*float64(limit) { + if _, err := keeper.Clear(ctx, obs.Key(), fmt.Sprintf("%s's declaration is %d bytes, under the limit's margin", node, size)); err != nil { + fmt.Printf("the condition %s could not be cleared: %v\n", obs.Key(), err) + } + return + } + fmt.Printf("NEAR THE LIMIT: %s\n", obs.Summary) + if _, err := keeper.Observe(ctx, obs); err != nil { + fmt.Printf("the condition %s could not be raised: %v\n", obs.Key(), err) + } +} + +// compactKept writes the vault's export compact. The composition (the SDK's keptFile) writes it indented, and its 334 +// sealed secrets were 164,763 bytes of it against 138,474 compact: a fifth of the declaration's room, in whitespace, +// base64-encoded again on the wire. Every reader parses JSON, so the spacing is nobody's. A resource that is not +// the export, or whose content is not JSON, is left as it is. The resources are copied, never changed in place. +func compactKept(resources []map[string]any) []map[string]any { + out := make([]map[string]any, len(resources)) + for i, r := range resources { + out[i] = r + id, _ := r["id"].(string) + path, _ := r["path"].(string) + content, ok := r["content"].(string) + if !ok || (id != "kept" && !strings.HasSuffix(id, ".kept")) || !strings.HasSuffix(path, "/export.json") { + continue + } + var dst bytes.Buffer + if json.Compact(&dst, []byte(content)) != nil { + continue + } + if strings.HasSuffix(content, "\n") { + dst.WriteByte('\n') + } + if dst.String() == content { + continue + } + c := make(map[string]any, len(r)) + for k, v := range r { + c[k] = v + } + c["content"] = dst.String() + out[i] = c + } + return out +} diff --git a/cmd/mesh-controller/declaration_room_test.go b/cmd/mesh-controller/declaration_room_test.go new file mode 100644 index 00000000..c1aba998 --- /dev/null +++ b/cmd/mesh-controller/declaration_room_test.go @@ -0,0 +1,71 @@ +package main + +import ( + "strings" + "testing" + "time" + + "github.com/novox/mesh-controller/internal/conditions" +) + +// The vault's export is sent compact (issue 1689): same document, no indentation, one rewrite. +func TestTheVaultsExportIsSentCompactAndNothingElseIsTouched(t *testing.T) { + indented := "{\n \"export\": 1,\n \"kept\": [\n {\n \"name\": \"a\"\n }\n ]\n}\n" + other := "{\n \"x\": 1\n}\n" + in := []map[string]any{ + {"id": "mesh-vault.kept", "type": "file", "path": "/srv/vault/export.json", "mode": "0600", "content": indented}, + {"id": "mesh-vault.config", "type": "file", "path": "/srv/vault/config.json", "content": other}, + {"id": "other.kept", "type": "file", "path": "/srv/other/notes.json", "content": other}, + } + out := compactKept(in) + want := "{\"export\":1,\"kept\":[{\"name\":\"a\"}]}\n" + if out[0]["content"] != want || out[0]["mode"] != "0600" || out[0]["path"] != "/srv/vault/export.json" { + t.Fatalf("the export is %q (%v), expected %q with its mode and path kept", out[0]["content"], out[0], want) + } + if out[1]["content"] != other || out[2]["content"] != other { + t.Fatalf("a file that is not the export was rewritten: %v %v", out[1], out[2]) + } + if in[0]["content"] != indented { + t.Fatalf("the composed resource was changed in place") + } + if again := compactKept(out); again[0]["content"] != want { + t.Fatalf("compacting twice changed it: %q", again[0]["content"]) + } +} + +// From 90% of the bus's largest message the declaration is a warning; over it, urgent; back under 90%, cleared. +func TestADeclarationNearTheBusLimitRaisesAConditionAndClearsWhenSmaller(t *testing.T) { + store := conditions.NewInMemory() + k := conditions.NewKeeper(t.Context(), conditions.Options{Store: store, History: store, + Teller: &conditions.Told{}, Now: func() time.Time { return time.Date(2026, 10, 21, 14, 0, 0, 0, time.UTC) }}) + t.Cleanup(func() { k.Close(t.Context()) }) + const limit = 1000 + key := declarationRoomObservation("novox", 0, limit).Key() + read := func() (conditions.Condition, bool) { + t.Helper() + c, found, err := conditions.ReadOne(t.Context(), store, key) + if err != nil { + t.Fatal(err) + } + return c, found + } + + settleDeclarationRoom(t.Context(), k, "novox", 899, limit) + if _, found := read(); found { + t.Fatal("a declaration at 89.9% raised a condition") + } + settleDeclarationRoom(t.Context(), k, "novox", 900, limit) + c, found := read() + if !found || c.Severity != conditions.Warning || !strings.Contains(c.Summary, "900 bytes") || + !strings.Contains(c.Summary, "1000") || c.Subject.Machine != "novox" { + t.Fatalf("at 90%% the condition is %v %+v", found, c) + } + settleDeclarationRoom(t.Context(), k, "novox", 1001, limit) + if c, _ := read(); c.Severity != conditions.Urgent { + t.Fatalf("over the limit the condition is %q, expected urgent", c.Severity) + } + settleDeclarationRoom(t.Context(), k, "novox", 850, limit) + if _, found := read(); found { + t.Fatal("the condition stayed after the declaration fell under 90%") + } +} diff --git a/cmd/mesh-controller/plain_words.go b/cmd/mesh-controller/plain_words.go index 285d05cc..0c69739b 100644 --- a/cmd/mesh-controller/plain_words.go +++ b/cmd/mesh-controller/plain_words.go @@ -195,6 +195,21 @@ var plainWordings = map[string]func(conditions.Observation) words{ "there yet is sealed by the controller until the vault holds it.", m, m), Resolved: "Resolved: the vault gave " + m + " the core's credentials"} }), + kindDeclarationNearLimit: worded(func(o conditions.Observation) words { + m := machineOr(o, "a machine") + if o.Severity == conditions.Urgent { + return words{Headline: m + " cannot be sent its instructions", + Needs: "make the instructions for " + m + " smaller, for example by taking a module off it.", + Explanation: fmt.Sprintf("The instructions for %s have grown past the largest message the bus takes, so "+ + "nothing new reaches %s until they are smaller.", m, m), + Resolved: "Resolved: the instructions for " + m + " fit the bus again"} + } + return words{Headline: "The instructions for " + m + " are nearly too big", + Needs: "do not add more to " + m + " until this is dealt with; the details give the size.", + Explanation: fmt.Sprintf("The instructions for %s are close to the largest message the bus takes. "+ + "One more module or secret may stop anything new from reaching %s.", m, m), + Resolved: "Resolved: the instructions for " + m + " have room again"} + }), kindCoreDowngrade: worded(func(o conditions.Observation) words { m := machineOr(o, "a machine") component := idPart(o, 0) diff --git a/internal/link/bus.go b/internal/link/bus.go index fdb38c23..c4fbdf27 100644 --- a/internal/link/bus.go +++ b/internal/link/bus.go @@ -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 { diff --git a/internal/link/declare.go b/internal/link/declare.go index f8730374..63eca97d 100644 --- a/internal/link/declare.go +++ b/internal/link/declare.go @@ -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 +} diff --git a/internal/link/declare_room_integration_test.go b/internal/link/declare_room_integration_test.go new file mode 100644 index 00000000..eb607b2f --- /dev/null +++ b/internal/link/declare_room_integration_test.go @@ -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) + } +} diff --git a/internal/link/declare_room_test.go b/internal/link/declare_room_test.go new file mode 100644 index 00000000..4339277b --- /dev/null +++ b/internal/link/declare_room_test.go @@ -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) + } +} diff --git a/internal/testbus/testbus.go b/internal/testbus/testbus.go index 8755317a..d3e2dcd5 100644 --- a/internal/testbus/testbus.go +++ b/internal/testbus/testbus.go @@ -56,10 +56,14 @@ var buses sync.Map // Start is a server with JetStream, for this test alone: on loopback, on a port the system chooses, its // store in the test's own directory. -func Start(t testing.TB) *server.Server { +func Start(t testing.TB) *server.Server { return StartWithMaxPayload(t, 0) } + +// StartWithMaxPayload is Start with the largest message the server takes set to maxPayload bytes; zero keeps the +// server's default (1 MiB, as the mesh's bus has). +func StartWithMaxPayload(t testing.TB, maxPayload int32) *server.Server { t.Helper() opts := &server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, JetStream: true, StoreDir: t.TempDir(), - NoLog: true, NoSigs: true} + NoLog: true, NoSigs: true, MaxPayload: maxPayload} s, err := server.NewServer(opts) if err != nil { t.Fatalf("a bus for this test could not be made: %v", err)