Merge pull request 'Send the vault's export compact and measure every declaration against the bus's limit, raising a condition from 90%, so novox's declaration fits again and cannot fail silently (issue 1689)' (#418) from fix/1689-declaration-room into main

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,

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

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

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)

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 {

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

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

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

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)