Compare commits
6
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
4771ee53d9 | ||
|
|
f10fbda1f6 | ||
|
|
09aa063a13 | ||
|
|
5be1cfaed1 | ||
|
|
35829d5a9a | ||
|
|
57939d6dd4 |
@@ -232,6 +232,8 @@ func (f following) hearMerge(ctx context.Context, m link.SourceMoved, now time.T
|
|||||||
fmt.Print(nothingReadsIt(repository, m))
|
fmt.Print(nothingReadsIt(repository, m))
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
// **A module the merge adds is asked for now** (novox/hq issue 1617), not when the batch is cut.
|
||||||
|
askNewModulesAtHearing(ctx, inv, m, entries, read, facts.from)
|
||||||
// **Its delivery group's word, before the plans are held** (ADR 0306 decision 2): a slow answer holds up no walk.
|
// **Its delivery group's word, before the plans are held** (ADR 0306 decision 2): a slow answer holds up no walk.
|
||||||
own := ownPath(touches)
|
own := ownPath(touches)
|
||||||
waits, holding, err := groupWaitHolding(ctx, inv, m, deliverySeatHeld(entries))
|
waits, holding, err := groupWaitHolding(ctx, inv, m, deliverySeatHeld(entries))
|
||||||
|
|||||||
@@ -366,7 +366,7 @@ func sayFound(ctx context.Context, d compose.Declared) {
|
|||||||
|
|
||||||
// sendableOf is a composed declaration as the controller's sends know it.
|
// sendableOf is a composed declaration as the controller's sends know it.
|
||||||
func sendableOf(d compose.Declared) sendable {
|
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,
|
ReadsModules: d.ReadsModules, moduleOf: d.ModuleOf, buildIDs: d.BuildIDs, modules: d.Modules,
|
||||||
localNetwork: d.LocalNetwork, Received: d.Received, Mesh: d.Mesh, BusUsers: d.BusUsers,
|
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,
|
LeftOut: d.LeftOut, leftOutWhy: d.LeftOutWhy, withheld: d.Withheld, unbound: d.Unbound, foreseen: d.Foreseen,
|
||||||
|
|||||||
@@ -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
|
||||||
|
}
|
||||||
@@ -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%")
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,37 @@
|
|||||||
|
package main
|
||||||
|
|
||||||
|
import (
|
||||||
|
"reflect"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/novox/mesh-controller/internal/inventory"
|
||||||
|
)
|
||||||
|
|
||||||
|
// 1617/AC2 (unit): a new module whose first build a merge's hearing asked is not asked again when its batch is cut,
|
||||||
|
// and one whose ask failed, was never handed over or is long lost is.
|
||||||
|
func TestAModuleWhoseBuildIsComingIsNotAskedAgain(t *testing.T) {
|
||||||
|
now := time.Date(2026, 10, 21, 12, 0, 0, 0, time.UTC)
|
||||||
|
source := buildSource{Repository: "http://forge/novox/mesh-catalog.git", Seat: "git"}
|
||||||
|
asked := func(path string, mod func(*inventory.RequestOutcome)) inventory.RequestOutcome {
|
||||||
|
r := inventory.RequestOutcome{BuildRequest: inventory.BuildRequest{ID: "b-" + path, Repository: source.Repository,
|
||||||
|
Seat: "git", Path: path, For: "merge", At: now.Add(-3 * time.Minute)}}
|
||||||
|
if mod != nil {
|
||||||
|
mod(&r)
|
||||||
|
}
|
||||||
|
return r
|
||||||
|
}
|
||||||
|
requests := []inventory.RequestOutcome{
|
||||||
|
asked("modules/go", nil),
|
||||||
|
asked("modules/failed", func(r *inventory.RequestOutcome) { r.Heard, r.Failed = true, "no module.json" }),
|
||||||
|
asked("modules/notasked", func(r *inventory.RequestOutcome) { r.NotAsked = "the bus is silent" }),
|
||||||
|
asked("modules/lost", func(r *inventory.RequestOutcome) { r.At = now.Add(-2 * time.Hour) }),
|
||||||
|
asked("modules/byhand", func(r *inventory.RequestOutcome) { r.For = "build" }),
|
||||||
|
}
|
||||||
|
got := unaskedDirs([]string{"modules/go", "modules/failed", "modules/notasked", "modules/lost", "modules/byhand",
|
||||||
|
"modules/nodejs"}, requests, source, now)
|
||||||
|
want := []string{"modules/failed", "modules/notasked", "modules/lost", "modules/byhand", "modules/nodejs"}
|
||||||
|
if !reflect.DeepEqual(got, want) {
|
||||||
|
t.Fatalf("asked again %v, want %v", got, want)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -6,6 +6,7 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"sync"
|
"sync"
|
||||||
"testing"
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||||
"github.com/novox/mesh-controller/internal/inventory"
|
"github.com/novox/mesh-controller/internal/inventory"
|
||||||
@@ -99,3 +100,38 @@ func TestAMergeBuildsItsNewModuleBesideItsPlan(t *testing.T) {
|
|||||||
t.Fatalf("a merge adding no module asked for %v", *asked)
|
t.Fatalf("a merge adding no module asked for %v", *asked)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// 1617/AC1 (integration): a merge adding a module to a repository the mesh builds modules from, heard while a batch of
|
||||||
|
// that repository is still assembling (its window running), asks the module's first build when it is heard, so the
|
||||||
|
// module is registered while the batch waits; cutting the batch asks it no second time (1617/AC2).
|
||||||
|
func TestAMergeAddingAModuleAsksItsBuildBeforeItsBatchIsCut(t *testing.T) {
|
||||||
|
open := windowed(t)
|
||||||
|
groupsWaiting(t, "pr/c1")
|
||||||
|
asked := asksWithPaths(t)
|
||||||
|
catalog := func(commit string, made time.Time, paths ...string) link.SourceMoved {
|
||||||
|
return link.SourceMoved{Owner: "novox", Repo: "mesh-catalog", Base: "main", Commit: commit,
|
||||||
|
Head: "pr/" + commit, MergedAt: made.Format(time.RFC3339Nano), Paths: paths}
|
||||||
|
}
|
||||||
|
hear(t, open, catalog("c1", t0, "modules/app/index.ts"), t0)
|
||||||
|
if len(*asked) != 0 {
|
||||||
|
t.Fatalf("a merge adding no module asked %v", *asked)
|
||||||
|
}
|
||||||
|
hear(t, open, lone(catalog("c2", t0.Add(10*time.Second), "modules/go/packages.json", "modules/go/module.json")), t0.Add(10*time.Second))
|
||||||
|
if ws, bs := walks(t, open); len(ws) != 0 || len(bs) != 1 {
|
||||||
|
t.Fatalf("the batch should still be assembling: %d walk(s), %d batch(es)", len(ws), len(bs))
|
||||||
|
}
|
||||||
|
want := [3]string{"novox/mesh-catalog", "modules/go", "main"}
|
||||||
|
if len(*asked) != 1 || (*asked)[0] != want {
|
||||||
|
t.Fatalf("heard with its batch assembling, the merge asked %v, not %v", *asked, want)
|
||||||
|
}
|
||||||
|
cutAt(t, open, t0.Add(10*time.Second+90*time.Second))
|
||||||
|
n := 0
|
||||||
|
for _, a := range *asked {
|
||||||
|
if a[1] == "modules/go" {
|
||||||
|
n++
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if n != 1 {
|
||||||
|
t.Fatalf("modules/go was asked %d times: %v", n, *asked)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -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),
|
"there yet is sealed by the controller until the vault holds it.", m, m),
|
||||||
Resolved: "Resolved: the vault gave " + m + " the core's credentials"}
|
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 {
|
kindCoreDowngrade: worded(func(o conditions.Observation) words {
|
||||||
m := machineOr(o, "a machine")
|
m := machineOr(o, "a machine")
|
||||||
component := idPart(o, 0)
|
component := idPart(o, 0)
|
||||||
|
|||||||
@@ -415,7 +415,59 @@ func askNewModules(ctx context.Context, inv *inventory.Inventory, m link.SourceM
|
|||||||
if len(added) == 0 || len(from) == 0 {
|
if len(added) == 0 || len(from) == 0 {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
return askBuilds(ctx, inv, m, buildSource{Repository: from[0].Source.Repository, Seat: from[0].Source.Seat}, added)
|
source := buildSource{Repository: from[0].Source.Repository, Seat: from[0].Source.Seat}
|
||||||
|
// Asked once (novox/hq issue 1617): a module is asked when its merge is heard, and again here when the batch is cut.
|
||||||
|
var requests []inventory.RequestOutcome
|
||||||
|
for _, dir := range added {
|
||||||
|
if got, err := inv.RequestsNamed(ctx, inventory.BuildRequest{Repository: source.Repository, Path: dir}.Name()); err == nil {
|
||||||
|
requests = append(requests, got...)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return askBuilds(ctx, inv, m, source, unaskedDirs(added, requests, source, time.Now()))
|
||||||
|
}
|
||||||
|
|
||||||
|
// askedInFlightFor is how long a merge's build request stands as the build coming for its module: past it, a lost one
|
||||||
|
// is asked again.
|
||||||
|
const askedInFlightFor = time.Hour
|
||||||
|
|
||||||
|
// unaskedDirs are the directories of a new module no build of a merge is on its way for: a kept request of the same
|
||||||
|
// repository and directory, asked for a merge and handed over, whose outcome has not come and is not given up on,
|
||||||
|
// is the build coming (novox/hq issue 1617).
|
||||||
|
func unaskedDirs(added []string, requests []inventory.RequestOutcome, source buildSource, now time.Time) []string {
|
||||||
|
var out []string
|
||||||
|
for _, dir := range added {
|
||||||
|
path := dir
|
||||||
|
if path == "." {
|
||||||
|
path = ""
|
||||||
|
}
|
||||||
|
coming := false
|
||||||
|
for _, r := range requests {
|
||||||
|
if r.For == "merge" && r.Repository == source.Repository && strings.Trim(r.Path, "/") == path &&
|
||||||
|
r.NotAsked == "" && r.OutcomeUnknown == "" && !r.Heard && now.Sub(r.At) < askedInFlightFor {
|
||||||
|
coming = true
|
||||||
|
break
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if !coming {
|
||||||
|
out = append(out, dir)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return out
|
||||||
|
}
|
||||||
|
|
||||||
|
// askNewModulesAtHearing asks the first build of the modules a merge adds to a repository the mesh builds modules from,
|
||||||
|
// when the merge is heard, not when its batch is cut (novox/hq issue 1617): a batch waits out its window and the walk
|
||||||
|
// before it, and a module added by a merge delivered meanwhile could not be assigned. Outside the batch and its plan.
|
||||||
|
func askNewModulesAtHearing(ctx context.Context, inv *inventory.Inventory, m link.SourceMoved, entries []inventory.Entry,
|
||||||
|
read map[string][]inventory.ReadRepository, from []inventory.Entry) {
|
||||||
|
if len(from) == 0 {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
_, added, _ := touchedBy(from, entries, m, read)
|
||||||
|
if asked := askNewModules(ctx, inv, m, from, added); len(asked) > 0 {
|
||||||
|
fmt.Printf("%s/%s merged into %s (%.8s) adds %s: the first build is asked now, before its batch is cut\n",
|
||||||
|
m.Owner, m.Repo, m.Base, m.Commit, strings.Join(asked, ", "))
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// askBuilds asks the first build of each directory of a merge's repository, as askNewModules does, from the source given.
|
// askBuilds asks the first build of each directory of a merge's repository, as askNewModules does, from the source given.
|
||||||
|
|||||||
@@ -1773,9 +1773,24 @@ func whatFailed(s string) string {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
// The controller's module check refuses with a count line and one line per problem, then prints its
|
||||||
|
// warning counts: the refusal is what failed, not the count printed last (novox/hq issue 1310).
|
||||||
|
for i, line := range lines {
|
||||||
|
if line = strings.TrimSpace(line); problemsLine.MatchString(line) {
|
||||||
|
if i+1 < len(lines) {
|
||||||
|
if first := strings.TrimSpace(lines[i+1]); first != "" {
|
||||||
|
return line + ": " + first
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return line
|
||||||
|
}
|
||||||
|
}
|
||||||
return lastLine(s)
|
return lastLine(s)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// problemsLine is the module check's refusal: "mesh-controller: 3 problem(s) in 139 manifest(s)".
|
||||||
|
var problemsLine = regexp.MustCompile(`^\S+: [1-9]\d* problem\(s\) in \d+ manifest\(s\)$`)
|
||||||
|
|
||||||
func lastLine(s string) string {
|
func lastLine(s string) string {
|
||||||
lines := strings.Split(strings.TrimSpace(s), "\n")
|
lines := strings.Split(strings.TrimSpace(s), "\n")
|
||||||
return strings.TrimSpace(lines[len(lines)-1])
|
return strings.TrimSpace(lines[len(lines)-1])
|
||||||
|
|||||||
@@ -199,3 +199,29 @@ func TestOnlyATrysSelectionReachesTheScript(t *testing.T) {
|
|||||||
t.Fatalf("a try's selection: %v", env)
|
t.Fatalf("a try's selection: %v", env)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// **Issue 1310**: the controller's module check prints its warning counts after its refusal; a failed check
|
||||||
|
// is said by the refusal, never by the count printed last.
|
||||||
|
func TestAFailedModuleCheckIsSaidByItsRefusalNotItsWarningCount(t *testing.T) {
|
||||||
|
log := "mesh-controller: 3 problem(s) in 139 manifest(s)\n" +
|
||||||
|
"mesh-files claims node-files and says it serves upload, download, which that seat's protocol does not promise\n" +
|
||||||
|
"minio claims the seat \"object-store\", which no module declares and the mesh does not define\n" +
|
||||||
|
"only-office claims the seat \"document-converter\", which no module declares and the mesh does not define\n" +
|
||||||
|
"adwaita: ok, 4 tool(s)\nlong-running resources without health: 59\n" +
|
||||||
|
"files asking for a setting without saying whether it is trusted: 36\n"
|
||||||
|
// 1310/AC1
|
||||||
|
want := "mesh-controller: 3 problem(s) in 139 manifest(s): mesh-files claims node-files and says it serves upload, " +
|
||||||
|
"download, which that seat's protocol does not promise"
|
||||||
|
if got := whatFailed(log); got != want {
|
||||||
|
t.Errorf("1310/AC1: the captured log is said as %q, not %q", got, want)
|
||||||
|
}
|
||||||
|
// 1310/AC2
|
||||||
|
if got, want := whatFailed("ok \tx/a\t1s\n--- FAIL: TestX (0.01s)\n x_test.go:12: got 1\nFAIL\nlong-running resources without health: 59\n"),
|
||||||
|
"--- FAIL: TestX (0.01s)"; got != want {
|
||||||
|
t.Errorf("1310/AC2: a failing test is said as %q, not %q", got, want)
|
||||||
|
}
|
||||||
|
// 1310/AC3
|
||||||
|
if got, want := whatFailed("starting\nsomething odd happened\n"), "something odd happened"; got != want {
|
||||||
|
t.Errorf("1310/AC3: a log with no failure line is said as %q, not %q", got, want)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -184,6 +184,14 @@ func (b OverNATS) PublishMembershipAfter(ctx context.Context, node, module strin
|
|||||||
return nil
|
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 {
|
func (b OverNATS) PublishDeclaration(ctx context.Context, node string, body []byte) error {
|
||||||
_, err := b.JS.Publish(DeclareSubject(node), body, nats.Context(ctx))
|
_, err := b.JS.Publish(DeclareSubject(node), body, nats.Context(ctx))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
@@ -46,8 +46,41 @@ func Declare(ctx context.Context, bus Bus, signer Signer, node string,
|
|||||||
return err
|
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)
|
publish, cancel := context.WithTimeout(ctx, timeout)
|
||||||
defer cancel()
|
defer cancel()
|
||||||
|
|
||||||
return bus.PublishDeclaration(publish, node, body)
|
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)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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
|
// 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.
|
// 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()
|
t.Helper()
|
||||||
opts := &server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, JetStream: true, StoreDir: t.TempDir(),
|
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)
|
s, err := server.NewServer(opts)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("a bus for this test could not be made: %v", err)
|
t.Fatalf("a bus for this test could not be made: %v", err)
|
||||||
|
|||||||
Reference in New Issue
Block a user