Compare commits
14
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
69a67876c2 | ||
|
|
12752911b6 | ||
|
|
98ffcb667f | ||
|
|
2aa291f3b3 | ||
|
|
6ef78580ce | ||
|
|
0e0c775bfa | ||
|
|
61c8b56107 | ||
|
|
c24139a0fd | ||
|
|
d8d6d196a4 | ||
|
|
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))
|
||||
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.
|
||||
own := ownPath(touches)
|
||||
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.
|
||||
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,
|
||||
|
||||
@@ -0,0 +1,65 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"io"
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
)
|
||||
|
||||
// Issue 1638 (novox/hq ADR 0480 clause 7), taken in with mesh-sdk's container `groups`: the controller sends a container's
|
||||
// groups only to a node-engine that reports the capability container-groups, and holds the module on any other, naming it.
|
||||
|
||||
const n8nWithAGroup = `{"module":"n8n","version":"1",
|
||||
"config":{"media-gid":{"type":"number","minimum":1,"required":true,"description":"the group that owns the media library"}},
|
||||
"resources":[{"id":"server","type":"container","name":"n8n",
|
||||
"image":"n8nio/n8n@sha256:4846eb2f4b874ab04cde7fc1e249d2ddaec66e9aea64439beb2972cfea88e3c0",
|
||||
"groups":["${config:media-gid}"]}]}`
|
||||
|
||||
// printedBy runs reportUnhostable and returns what it printed.
|
||||
func printedBy(t *testing.T, node string, plan catalogue.Resolution) string {
|
||||
t.Helper()
|
||||
stdout := os.Stdout
|
||||
r, w, err := os.Pipe()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
os.Stdout = w
|
||||
reportUnhostable(node, plan)
|
||||
w.Close()
|
||||
os.Stdout = stdout
|
||||
out, _ := io.ReadAll(r)
|
||||
return string(out)
|
||||
}
|
||||
|
||||
// 1638/AC1: a node that reports the capability gets the module; one that does not has it held, the capability named.
|
||||
func TestAModuleWithGroupsIsHeldOnANodeThatDoesNotReportTheCapability(t *testing.T) {
|
||||
m, err := catalogue.ParseManifest([]byte(n8nWithAGroup))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
shelf := catalogue.Shelf{"n8n": m}
|
||||
reads := catalogue.Node{Name: "ace", Capabilities: map[string]bool{"container-runtime": true, "container-groups": true}}
|
||||
old := catalogue.Node{Name: "g14", Capabilities: map[string]bool{"container-runtime": true}}
|
||||
|
||||
got, err := catalogue.Resolve(shelf, []string{"n8n"}, reads, catalogue.World{})
|
||||
if err != nil || len(got.Unhostable) != 0 || len(got.Modules) != 1 {
|
||||
t.Fatalf("a node that reads groups: %v, held %+v", err, got.Unhostable)
|
||||
}
|
||||
if out := printedBy(t, "ace", got); out != "" {
|
||||
t.Errorf("a node that reads groups was told: %q", out)
|
||||
}
|
||||
|
||||
got, err = catalogue.Resolve(shelf, []string{"n8n"}, old, catalogue.World{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(got.Modules) != 0 || len(got.Unhostable) != 1 {
|
||||
t.Fatalf("n8n was sent to a node that does not report container-groups: hosted %d, held %+v", len(got.Modules), got.Unhostable)
|
||||
}
|
||||
if out := printedBy(t, "g14", got); !strings.Contains(out, "n8n") || !strings.Contains(out, "container-groups") {
|
||||
t.Errorf("the hold does not name the module and the capability: %q", out)
|
||||
}
|
||||
}
|
||||
@@ -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"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"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)
|
||||
}
|
||||
}
|
||||
|
||||
// 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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,198 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/inventory"
|
||||
)
|
||||
|
||||
// heldElsewhere is the plans held by another act for longer than any test runs: waiting for them ends only with the
|
||||
// context, and a look that does not wait is told at once that they are busy.
|
||||
func heldElsewhere(ctx context.Context, wait bool) (func(), error) {
|
||||
if !wait {
|
||||
return nil, inventory.ErrPlansBusy
|
||||
}
|
||||
<-ctx.Done()
|
||||
return nil, inventory.ErrPlansBusy
|
||||
}
|
||||
|
||||
// 1708/AC1: given the plans held by another act, when the event loop takes a build's outcome, then it returns at
|
||||
// once, without waiting the plans' two minutes (on 2026-10-21 at 14:20:45 nodered's outcome held the loop until
|
||||
// 14:22:45, and the operator was told the controller stopped listening).
|
||||
func TestAnOutcomeDoesNotWaitForThePlansAnotherActHolds(t *testing.T) {
|
||||
ctx, cancel := context.WithTimeout(t.Context(), 3*time.Second)
|
||||
defer cancel()
|
||||
start := time.Now()
|
||||
release, err := holdPlansForOutcome(ctx, heldElsewhere)
|
||||
if took := time.Since(start); took > time.Second {
|
||||
t.Fatalf("taking an outcome waited %s for plans another act holds; the event loop takes nothing meanwhile", took)
|
||||
}
|
||||
if err == nil || release != nil {
|
||||
t.Fatalf("plans another act holds were said to be held: release %v, err %v", release != nil, err)
|
||||
}
|
||||
}
|
||||
|
||||
// 1708/AC4: given the plans free, when the event loop takes a build's outcome, then it holds them and marks the plan
|
||||
// from the outcome directly, as before.
|
||||
func TestAnOutcomeHoldsThePlansWhenTheyAreFree(t *testing.T) {
|
||||
let := false
|
||||
release, err := holdPlansForOutcome(t.Context(), func(context.Context, bool) (func(), error) {
|
||||
return func() { let = true }, nil
|
||||
})
|
||||
if err != nil || release == nil {
|
||||
t.Fatalf("free plans were not held: %v", err)
|
||||
}
|
||||
release()
|
||||
if !let {
|
||||
t.Fatal("the hold given was not the one taken")
|
||||
}
|
||||
}
|
||||
|
||||
// withLooker installs a looker that is never run, and answers it, so a test can read whether a look was asked.
|
||||
func withLooker(t *testing.T) *looker {
|
||||
t.Helper()
|
||||
l := newLooker(func(context.Context) bool { return false })
|
||||
lookerMu.Lock()
|
||||
before := theLooker
|
||||
theLooker = l
|
||||
lookerMu.Unlock()
|
||||
t.Cleanup(func() {
|
||||
lookerMu.Lock()
|
||||
theLooker = before
|
||||
lookerMu.Unlock()
|
||||
})
|
||||
return l
|
||||
}
|
||||
|
||||
// 1708/AC2: given an outcome left to the build records because the plans were held, when it is left, then the look
|
||||
// at the plans is asked for, so the records settle it once the holder lets go rather than at the next event.
|
||||
func TestAnOutcomeLeftToTheRecordsAsksForTheLook(t *testing.T) {
|
||||
l := withLooker(t)
|
||||
leaveToRecords("nodered", "build-1708-ac2", "", "", inventory.ErrPlansBusy)
|
||||
select {
|
||||
case <-l.signal:
|
||||
default:
|
||||
t.Fatal("an outcome left to the build records asked for no look: the plan waits for whatever event comes next")
|
||||
}
|
||||
}
|
||||
|
||||
// 1708/AC3: given an outcome left to the records that said what the build did (ADR 0350 §4), when the look settles
|
||||
// the plan from the build records by that build's id, then the module keeps those words, once.
|
||||
func TestTheRecordsSettleAnOutcomeWithWhatTheBuildDid(t *testing.T) {
|
||||
withLooker(t)
|
||||
asked := time.Date(2026, 10, 21, 12, 15, 0, 0, time.UTC)
|
||||
plan := func() inventory.Plan {
|
||||
return inventory.Plan{ID: "plan-1708", Tiers: [][]string{{"nodered"}},
|
||||
Modules: map[string]*inventory.PlanModule{"nodered": {State: "asked", AskedAt: &asked, Build: "build-1708-ac3"}}}
|
||||
}
|
||||
byID := map[string]inventory.Build{"build-1708-ac3": {ID: "build-1708-ac3", Module: "nodered", Commit: "4efac366",
|
||||
At: asked.Add(5 * time.Minute)}}
|
||||
leaveToRecords("nodered", "build-1708-ac3", "reused from build-1708-check", "", inventory.ErrPlansBusy)
|
||||
|
||||
p := plan()
|
||||
if !settleFromRecords(&p, p.Tiers[0], nil, byID) || p.Modules["nodered"].State != "built" {
|
||||
t.Fatalf("the records did not settle the outcome: %+v", p.Modules["nodered"])
|
||||
}
|
||||
if got := p.Modules["nodered"].Made; got != "reused from build-1708-check" {
|
||||
t.Fatalf("settled from the records, the module says %q of what its build did, not the outcome's words", got)
|
||||
}
|
||||
again := plan()
|
||||
settleFromRecords(&again, again.Tiers[0], nil, byID)
|
||||
if got := again.Modules["nodered"].Made; got != "" {
|
||||
t.Fatalf("an outcome's words were kept for a second settle: %q", got)
|
||||
}
|
||||
}
|
||||
|
||||
// 1708/AC5: given a serving controller (its look at the plans running), when an outcome was taken into the plans,
|
||||
// then moving them (a release's sends, which may retry for minutes) is left to the look, off the event loop
|
||||
// (2026-10-21 14:28:40 to 14:32:10: the loop sent okioki's release to g14 after mesh-delivery-builder's outcome).
|
||||
func TestAnOutcomeLeavesMovingThePlansToTheLook(t *testing.T) {
|
||||
l := withLooker(t)
|
||||
moved := false
|
||||
advanceAfterOutcome(func() { moved = true })
|
||||
if moved {
|
||||
t.Fatal("the plans were moved on the event loop after an outcome, where a send that retries holds every message")
|
||||
}
|
||||
select {
|
||||
case <-l.signal:
|
||||
default:
|
||||
t.Fatal("the look was not asked to move the plans after an outcome")
|
||||
}
|
||||
}
|
||||
|
||||
// 1708/AC5: given no look running (a command, which has no event loop), when an outcome was taken, then the plans are
|
||||
// moved at once, as before.
|
||||
func TestAnOutcomeMovesThePlansWhereNoLookRuns(t *testing.T) {
|
||||
lookerMu.Lock()
|
||||
before := theLooker
|
||||
theLooker = nil
|
||||
lookerMu.Unlock()
|
||||
t.Cleanup(func() {
|
||||
lookerMu.Lock()
|
||||
theLooker = before
|
||||
lookerMu.Unlock()
|
||||
})
|
||||
moved := false
|
||||
advanceAfterOutcome(func() { moved = true })
|
||||
if !moved {
|
||||
t.Fatal("with no look running, an outcome left the plans unmoved")
|
||||
}
|
||||
}
|
||||
|
||||
// 1708/AC6: given a registration the mesh refused (its build recorded as worked, the outcome failed), left to the
|
||||
// records because the plans were held, when the look settles the plan from the records, then the module and the plan
|
||||
// are failed, never built: settled as built, the tier would be sent the version from before (reviewer-1708@shanks).
|
||||
func TestARefusedRegistrationLeftToTheRecordsFailsThePlan(t *testing.T) {
|
||||
withLooker(t)
|
||||
asked := time.Date(2026, 10, 21, 12, 15, 0, 0, time.UTC)
|
||||
p := inventory.Plan{ID: "plan-1708-ac6", State: inventory.PlanBuilding, Tiers: [][]string{{"nodered"}},
|
||||
Modules: map[string]*inventory.PlanModule{"nodered": {State: "asked", AskedAt: &asked, Build: "build-1708-ac6"}}}
|
||||
byID := map[string]inventory.Build{"build-1708-ac6": {ID: "build-1708-ac6", Module: "nodered", Commit: "4efac366",
|
||||
At: asked.Add(5 * time.Minute)}}
|
||||
leaveToRecords("nodered", "build-1708-ac6", "", "the mesh does not register it: a port is taken", inventory.ErrPlansBusy)
|
||||
|
||||
overlayRefusals(nil, byID, refusalsKeptFor([]string{"build-1708-ac6"}))
|
||||
settleFromRecords(&p, p.Tiers[0], nil, byID)
|
||||
if s := p.Modules["nodered"]; s.State != "failed" || p.State != inventory.PlanFailed {
|
||||
t.Fatalf("a refused registration left to the records settled as %q, the plan %q", s.State, p.State)
|
||||
}
|
||||
if !strings.Contains(p.Modules["nodered"].Why, "a port is taken") {
|
||||
t.Fatalf("the refusal's words were lost: %q", p.Modules["nodered"].Why)
|
||||
}
|
||||
}
|
||||
|
||||
// 1708/AC6: a refusal kept in the store (a controller replaced before the settle) fails the record as one kept in
|
||||
// the process does.
|
||||
func TestARefusalFromTheStoreFailsTheRecord(t *testing.T) {
|
||||
recorded := map[string][]inventory.Build{"nodered": {{ID: "b-store", Module: "nodered"}, {ID: "b-other", Module: "nodered"}}}
|
||||
byID := map[string]inventory.Build{"b-store": {ID: "b-store", Module: "nodered"}}
|
||||
overlayRefusals(recorded, byID, map[string]string{"b-store": "not registered"})
|
||||
if recorded["nodered"][0].Worked() || byID["b-store"].Worked() {
|
||||
t.Fatal("a build whose registration was refused still reads as worked")
|
||||
}
|
||||
if !recorded["nodered"][1].Worked() {
|
||||
t.Fatal("a build nobody refused was failed")
|
||||
}
|
||||
}
|
||||
|
||||
// 1708/AC7: given the records settle a module as deleted at its source, or fail the plan, when the look settles, then
|
||||
// it does what taking the outcome directly does: the deleted module is named to be forgotten, and the plan's failure
|
||||
// is said (reviewer-1708@shanks).
|
||||
func TestSettlingFromTheRecordsDoesWhatTakingTheOutcomeDoes(t *testing.T) {
|
||||
p := inventory.Plan{State: inventory.PlanFailed, Modules: map[string]*inventory.PlanModule{
|
||||
"gone": {State: planDeleted}, "x": {State: "failed"}, "y": {State: "built"}}}
|
||||
before := map[string]string{"gone": "asked", "x": "asked", "y": "built"}
|
||||
deleted, failed := settledEffects(before, &p, inventory.PlanBuilding)
|
||||
if len(deleted) != 1 || deleted[0] != "gone" {
|
||||
t.Fatalf("the module the records found deleted is not forgotten: %v", deleted)
|
||||
}
|
||||
if !failed {
|
||||
t.Fatal("the plan the records failed is not said as failed")
|
||||
}
|
||||
if d, f := settledEffects(map[string]string{"gone": planDeleted}, &p, inventory.PlanFailed); len(d) != 0 || f {
|
||||
t.Fatalf("what was deleted or failed before the settle is acted on again: %v %v", d, f)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
@@ -136,6 +136,18 @@ func askToLook() {
|
||||
}
|
||||
}
|
||||
|
||||
// askedToLook asks the serving controller's look at the plans, and says whether there was one to ask.
|
||||
func askedToLook() bool {
|
||||
lookerMu.RLock()
|
||||
l := theLooker
|
||||
lookerMu.RUnlock()
|
||||
if l == nil {
|
||||
return false
|
||||
}
|
||||
l.ask()
|
||||
return true
|
||||
}
|
||||
|
||||
// startPlanLooks starts the look at the open plans, and asks for the first: the controller took the lease, and a
|
||||
// plan left by a replaced controller resumes from there (ADR 0406 §6). It replaces the rollout's 30 second ticker.
|
||||
// It answers a channel that is closed once the look has returned after the context ended, for a test that restores what
|
||||
|
||||
@@ -625,16 +625,124 @@ func planBuilt(ctx context.Context, open *stores, module, commit, failed string,
|
||||
planBuiltSaying(ctx, open, module, commit, failed, asked, id, "")
|
||||
}
|
||||
|
||||
// holdPlansForOutcome is how an outcome taken on the event loop holds the plans: **without waiting for them**
|
||||
// (novox/hq issue 1708). The loop takes one message at a time, and the plans' holder may keep them for minutes while
|
||||
// it sends machines their declarations; an outcome that waited its turn (up to inventory.HoldWaitFor, two minutes)
|
||||
// held every report, build and merge behind it for as long, and the watchdog said the controller stopped listening
|
||||
// (2026-10-21 14:20:45 to 14:22:45). Busy, the outcome is left to the build records, where it already is, and the
|
||||
// look settles the plan from them (leaveToRecords).
|
||||
func holdPlansForOutcome(ctx context.Context, hold func(context.Context, bool) (func(), error)) (func(), error) {
|
||||
return hold(ctx, false)
|
||||
}
|
||||
|
||||
// leaveToRecords says an outcome was not taken into the plans, which settle it from the build records (novox/hq issue
|
||||
// 214). When another act holds the plans it asks for the look, which waits for them off the event loop and settles
|
||||
// the plan from the records once they are let go (issue 1708); what the build did is kept for that settle by the
|
||||
// build's id, so the walk's module says it as it would have (ADR 0350 §4).
|
||||
func leaveToRecords(module, id, made, failed string, err error) {
|
||||
fmt.Printf("plans: %s's outcome is left to the build records: %v\n", module, err)
|
||||
if !errors.Is(err, inventory.ErrPlansBusy) {
|
||||
return
|
||||
}
|
||||
keepMadeFor(id, made, time.Now())
|
||||
// A refused outcome (failed, its build recorded as worked) is kept as refused, or the records settle it built.
|
||||
keepRefusalFor(id, failed, time.Now())
|
||||
askToLook()
|
||||
}
|
||||
|
||||
// refusalsKeptFor is the refusals kept in this process for these builds (keepRefusalFor), by id.
|
||||
func refusalsKeptFor(ids []string) map[string]string {
|
||||
out := map[string]string{}
|
||||
refusedLeftMu.Lock()
|
||||
defer refusedLeftMu.Unlock()
|
||||
for _, id := range ids {
|
||||
if k, ok := refusedLeft[id]; ok {
|
||||
out[id] = k.made
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// refusedLeft is why each outcome left to the records was refused, by the build's id: kept in the process beside the
|
||||
// store's record (RecordRefused), so a store that did not take it still fails the plan while this controller serves.
|
||||
var (
|
||||
refusedLeftMu sync.Mutex
|
||||
refusedLeft = map[string]madeKept{}
|
||||
)
|
||||
|
||||
// keepRefusalFor keeps why an outcome left to the records was refused, forgetting refusals older than madeLeftFor.
|
||||
func keepRefusalFor(id, why string, now time.Time) {
|
||||
if id == "" || why == "" {
|
||||
return
|
||||
}
|
||||
refusedLeftMu.Lock()
|
||||
defer refusedLeftMu.Unlock()
|
||||
for k, v := range refusedLeft {
|
||||
if now.Sub(v.at) > madeLeftFor {
|
||||
delete(refusedLeft, k)
|
||||
}
|
||||
}
|
||||
refusedLeft[id] = madeKept{made: why, at: now}
|
||||
}
|
||||
|
||||
// madeLeft is what each build left to the records did, by the build's id, until a settle takes it: kept in the
|
||||
// process, so a controller replaced before the settle loses only these words, never the outcome.
|
||||
var (
|
||||
madeLeftMu sync.Mutex
|
||||
madeLeft = map[string]madeKept{}
|
||||
)
|
||||
|
||||
type madeKept struct {
|
||||
made string
|
||||
at time.Time
|
||||
}
|
||||
|
||||
// madeLeftFor is how long words nobody settled are kept: a plan that ended or moved on never asks for them.
|
||||
const madeLeftFor = time.Hour
|
||||
|
||||
// keepMadeFor keeps what a build left to the records did for the settle that takes it, forgetting words older than
|
||||
// madeLeftFor.
|
||||
func keepMadeFor(id, made string, now time.Time) {
|
||||
if id == "" || made == "" {
|
||||
return
|
||||
}
|
||||
madeLeftMu.Lock()
|
||||
defer madeLeftMu.Unlock()
|
||||
for k, v := range madeLeft {
|
||||
if now.Sub(v.at) > madeLeftFor {
|
||||
delete(madeLeft, k)
|
||||
}
|
||||
}
|
||||
madeLeft[id] = madeKept{made: made, at: now}
|
||||
}
|
||||
|
||||
// takeMadeOf is what the build of this id did, if its outcome was left to the records with words; taken once.
|
||||
func takeMadeOf(id string) string {
|
||||
madeLeftMu.Lock()
|
||||
defer madeLeftMu.Unlock()
|
||||
k, ok := madeLeft[id]
|
||||
if !ok {
|
||||
return ""
|
||||
}
|
||||
delete(madeLeft, id)
|
||||
return k.made
|
||||
}
|
||||
|
||||
// planBuiltSaying is planBuilt keeping what the build did — "reused from <check build>" or "built: <why>" (novox/hq
|
||||
// ADR 0350 §4) — on the module of the walk it answers.
|
||||
func planBuiltSaying(ctx context.Context, open *stores, module, commit, failed string, asked time.Time, id, made string) {
|
||||
inv := open.inventory
|
||||
// One controller works the plans at a time (novox/hq issue 213); an outcome waits its turn rather
|
||||
// than write over what the holder is about to save. Not taken, it is still in the build records,
|
||||
// which the holder settles the plan from (issue 214).
|
||||
release, err := inv.HoldPlans(ctx, true)
|
||||
// One controller works the plans at a time (novox/hq issue 213). Not taken, the outcome is still in the build
|
||||
// records, which the holder settles the plan from (issue 214).
|
||||
release, err := holdPlansForOutcome(ctx, inv.HoldPlans)
|
||||
if err != nil {
|
||||
fmt.Printf("plans: %s's outcome is left to the build records: %v\n", module, err)
|
||||
if failed != "" && id != "" && errors.Is(err, inventory.ErrPlansBusy) {
|
||||
// In the store too, so a controller replaced before the look settles still fails the plan (issue 1708).
|
||||
if err := inv.RecordRefused(ctx, id, module, failed); err != nil {
|
||||
fmt.Printf("plans: %s's refusal could not be kept in the store, only in this controller: %v\n", module, err)
|
||||
}
|
||||
}
|
||||
leaveToRecords(module, id, made, failed, err)
|
||||
return
|
||||
}
|
||||
defer release()
|
||||
@@ -696,7 +804,19 @@ func planBuiltSaying(ctx context.Context, open *stores, module, commit, failed s
|
||||
fmt.Printf("%s: %s; the tiers after it are not asked\n", p.ID, p.Note)
|
||||
}
|
||||
}
|
||||
advanceHeld(ctx, open)
|
||||
advanceAfterOutcome(func() { advanceHeld(ctx, open) })
|
||||
}
|
||||
|
||||
// advanceAfterOutcome moves the plans after an outcome was taken into them: **in the serving controller, by asking
|
||||
// the look** (novox/hq issue 1708). The outcome is taken on the event loop, and moving the plans may send machines
|
||||
// their declarations, a release's send retried for minutes among them; done here it held every report, build and
|
||||
// merge behind it (2026-10-21 14:28:40 to 14:32:10). The look runs off the loop, holding the plans itself. Where no
|
||||
// look runs (a command, which has no event loop) the plans are moved at once, as before.
|
||||
func advanceAfterOutcome(advance func()) {
|
||||
if askedToLook() {
|
||||
return
|
||||
}
|
||||
advance()
|
||||
}
|
||||
|
||||
// startedBeside is the id of a started walk open beside this one that it is related to — a merge's walk past
|
||||
@@ -921,7 +1041,17 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan,
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
before, was := statesOf(p), p.State
|
||||
if settleFromRecords(p, tier, recorded, byID) {
|
||||
// As taking the outcome directly does (planBuiltSaying): a module deleted at its source is forgotten, and a
|
||||
// failed plan says what it built and never sent (issue 1708).
|
||||
deleted, failed := settledEffects(before, p, was)
|
||||
for _, m := range deleted {
|
||||
forgetDeleted(ctx, inv, m, p)
|
||||
}
|
||||
if failed {
|
||||
sayUnsent(p, rollsOut)
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
// Asked: wait for every build.
|
||||
@@ -2479,9 +2609,81 @@ func recordsOfAsked(ctx context.Context, inv *inventory.Inventory, p *inventory.
|
||||
}
|
||||
}
|
||||
}
|
||||
refused, err := refusalsOf(ctx, inv, recorded, byID)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
overlayRefusals(recorded, byID, refused)
|
||||
return recorded, byID, nil
|
||||
}
|
||||
|
||||
// overlayRefusals marks failed every build record whose registration the mesh refused (novox/hq issue 1708): the record
|
||||
// keeps the builder's words, which worked, and settled from it as it is the refused build would be built and sent.
|
||||
func overlayRefusals(recorded map[string][]inventory.Build, byID map[string]inventory.Build, refused map[string]string) {
|
||||
if len(refused) == 0 {
|
||||
return
|
||||
}
|
||||
fail := func(b *inventory.Build) {
|
||||
if why, ok := refused[b.ID]; ok && b.Worked() {
|
||||
b.Failed = why
|
||||
}
|
||||
}
|
||||
for m := range recorded {
|
||||
for i := range recorded[m] {
|
||||
fail(&recorded[m][i])
|
||||
}
|
||||
}
|
||||
for id, b := range byID {
|
||||
fail(&b)
|
||||
byID[id] = b
|
||||
}
|
||||
}
|
||||
|
||||
// refusalsOf is every refusal kept for the builds these records name, in the store and in this process.
|
||||
func refusalsOf(ctx context.Context, inv *inventory.Inventory, recorded map[string][]inventory.Build,
|
||||
byID map[string]inventory.Build) (map[string]string, error) {
|
||||
var ids []string
|
||||
for _, bs := range recorded {
|
||||
for _, b := range bs {
|
||||
ids = append(ids, b.ID)
|
||||
}
|
||||
}
|
||||
for id := range byID {
|
||||
ids = append(ids, id)
|
||||
}
|
||||
refused, err := inv.RefusedOf(ctx, ids)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the refused registrations cannot be read: %w", err)
|
||||
}
|
||||
for id, why := range refusalsKeptFor(ids) {
|
||||
refused[id] = why
|
||||
}
|
||||
return refused, nil
|
||||
}
|
||||
|
||||
// settledEffects is what settling from the records did that needs more than the plan: the modules it found deleted
|
||||
// at their source, and whether it failed the plan.
|
||||
func settledEffects(before map[string]string, p *inventory.Plan, was string) (deleted []string, failed bool) {
|
||||
for m, s := range p.Modules {
|
||||
if s != nil && s.State == planDeleted && before[m] != planDeleted {
|
||||
deleted = append(deleted, m)
|
||||
}
|
||||
}
|
||||
sort.Strings(deleted)
|
||||
return deleted, p.State == inventory.PlanFailed && was != inventory.PlanFailed
|
||||
}
|
||||
|
||||
// statesOf is each module's state in a plan, for settledEffects.
|
||||
func statesOf(p *inventory.Plan) map[string]string {
|
||||
out := map[string]string{}
|
||||
for m, s := range p.Modules {
|
||||
if s != nil {
|
||||
out[m] = s.State
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// settleFromRecords marks every module of the tier still `asked` built — or failed — from a build
|
||||
// recorded after it was asked, and says whether it changed anything (novox/hq 04-ISSUES/214).
|
||||
// Newest first, as Builds answers: the first record after the ask is the outcome of that ask.
|
||||
@@ -2531,6 +2733,10 @@ func settleFromRecords(p *inventory.Plan, tier []string, recorded map[string][]i
|
||||
s.State = "built"
|
||||
s.BuiltAt = &at
|
||||
s.Commit = outcome.Commit
|
||||
// What the build did, when its outcome was left to the records with words (issue 1708).
|
||||
if made := takeMadeOf(outcome.ID); made != "" {
|
||||
s.Made = made
|
||||
}
|
||||
} else {
|
||||
s.State = "failed"
|
||||
s.Why = outcome.Failed
|
||||
|
||||
@@ -415,7 +415,59 @@ func askNewModules(ctx context.Context, inv *inventory.Inventory, m link.SourceM
|
||||
if len(added) == 0 || len(from) == 0 {
|
||||
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.
|
||||
|
||||
@@ -3,7 +3,7 @@ module github.com/novox/mesh-controller
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021105408-c8e35097e861
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021150202-6f6327587d19
|
||||
github.com/jackc/pgx/v5 v5.10.0
|
||||
github.com/nats-io/nats-server/v2 v2.11.17
|
||||
github.com/nats-io/nats.go v1.54.0
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
git.novox.be/novox/mesh-host v0.0.0-20261018171808-9d9e3a46c4f5 h1:E5stfIeCN6uYgA9TCv+4K8JgaURg3zQppvUgJLaDMP4=
|
||||
git.novox.be/novox/mesh-host v0.0.0-20261018171808-9d9e3a46c4f5/go.mod h1:4wyoCr6IIIIarEUuROThWDj7DAvJvIidqMjWcpSnOwE=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021105408-c8e35097e861 h1:BsqolZMMsC5jLP8X88BwShqFssLgtREcgBIy6OfPSI8=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021105408-c8e35097e861/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021150202-6f6327587d19 h1:OZ5igcMOiwx+uhsOHXF58pdF3LGt/HWhMhAEARO8LaM=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021150202-6f6327587d19/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
|
||||
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
|
||||
@@ -100,8 +100,9 @@ var ErrPlansBusy = errors.New("another controller is working the plans")
|
||||
// returned function and by the session ending, so a controller that dies holding it holds nothing.
|
||||
//
|
||||
// wait false gives ErrPlansBusy at once when another holds them — the timer's way: the holder is
|
||||
// moving the plans already. wait true looks again every HoldPoll for up to HoldWaitFor — an
|
||||
// outcome's or a merge's way, which must be written.
|
||||
// moving the plans already; and a build outcome's on the event loop, which the build records keep
|
||||
// for the look to settle (novox/hq issue 1708). wait true looks again every HoldPoll for up to
|
||||
// HoldWaitFor — a merge's way, which must be written.
|
||||
func (i *Inventory) HoldPlans(ctx context.Context, wait bool) (func(), error) {
|
||||
deadline := time.Now().Add(HoldWaitFor)
|
||||
for {
|
||||
|
||||
@@ -0,0 +1,10 @@
|
||||
-- A build's outcome whose registration the mesh refused, by the build's id (novox/hq issue 1708). The build record says
|
||||
-- what the builder said, which worked; an outcome the event loop could not take into the plans (another act held them)
|
||||
-- is settled from the records by the look, and without this it would read the refused build as built. Kept in the
|
||||
-- store so a controller replaced before the settle still fails the plan.
|
||||
create table build_refused (
|
||||
id text primary key,
|
||||
module text not null,
|
||||
why text not null,
|
||||
at timestamptz not null default now()
|
||||
);
|
||||
@@ -0,0 +1,33 @@
|
||||
package inventory
|
||||
|
||||
import "context"
|
||||
|
||||
// RecordRefused keeps that the mesh refused to register a build's outcome, and why (novox/hq issue 1708): the build
|
||||
// record says the builder's words, which worked, and the look that settles a plan from the records reads this to fail
|
||||
// it instead. The first refusal of an id stands.
|
||||
func (i *Inventory) RecordRefused(ctx context.Context, id, module, why string) error {
|
||||
_, err := i.store.Pool().Exec(ctx,
|
||||
`insert into build_refused (id, module, why) values ($1, $2, $3) on conflict (id) do nothing`, id, module, why)
|
||||
return err
|
||||
}
|
||||
|
||||
// RefusedOf is why each of these builds' registration was refused, by id; a build not refused is absent.
|
||||
func (i *Inventory) RefusedOf(ctx context.Context, ids []string) (map[string]string, error) {
|
||||
out := map[string]string{}
|
||||
if len(ids) == 0 {
|
||||
return out, nil
|
||||
}
|
||||
rows, err := i.store.Pool().Query(ctx, `select id, why from build_refused where id = any($1)`, ids)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
for rows.Next() {
|
||||
var id, why string
|
||||
if err := rows.Scan(&id, &why); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out[id] = why
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
@@ -0,0 +1,28 @@
|
||||
//go:build integration
|
||||
|
||||
package inventory
|
||||
|
||||
import "testing"
|
||||
|
||||
// novox/hq issue 1708 (1708/AC6): a refused registration is kept by its build's id, the first refusal standing, and
|
||||
// read back for the builds asked about only.
|
||||
func TestARefusedRegistrationIsKeptByItsBuild(t *testing.T) {
|
||||
inv := ForTest(t)
|
||||
ctx := t.Context()
|
||||
if err := inv.RecordRefused(ctx, "build-1708-a", "nodered", "a port is taken"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := inv.RecordRefused(ctx, "build-1708-a", "nodered", "said again"); err != nil {
|
||||
t.Fatalf("a refusal said twice was refused: %v", err)
|
||||
}
|
||||
got, err := inv.RefusedOf(ctx, []string{"build-1708-a", "build-1708-b"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(got) != 1 || got["build-1708-a"] != "a port is taken" {
|
||||
t.Fatalf("the refusals read back are %v", got)
|
||||
}
|
||||
if none, err := inv.RefusedOf(ctx, nil); err != nil || len(none) != 0 {
|
||||
t.Fatalf("no builds asked about answered %v %v", none, err)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
Vendored
+1
-1
@@ -9,7 +9,7 @@ a second place to clean besides the catalogue itself.
|
||||
|
||||
mesh-catalog e0bc919d4945bb5f13f971f74643a0b714cf45da modules/*/module.json, modules/nats/Dockerfile
|
||||
mesh-host 9d9e3a46c4f5943cfffb413d962ce48b75e2fd39 examples/foundation-first-node.lock
|
||||
mesh-sdk c8e35097e8611c174c7937d9ade86ed292314e9a conformance/events/module-event.json
|
||||
mesh-sdk 6f6327587d193bdca17ffab9008e2a7a6e9155f2 conformance/events/module-event.json
|
||||
|
||||
To move them, from this repository's root, with the two repositories checked out beside it:
|
||||
|
||||
|
||||
Vendored
+1
@@ -167,3 +167,4 @@
|
||||
./cmd/mesh-controller TestAMergeOnTheApproveIntoANewRepositoryAsksTheFirstBuildAndItsOutcomeRegistersTheModule
|
||||
./cmd/mesh-controller TestTheApprovedFirstBuildPassesTheGuardOfAVerbAndIsQueued
|
||||
./cmd/mesh-controller TestAPlanGoesOnWithoutAMachineAwayWhenItsTierIsSent
|
||||
./internal/inventory TestARefusedRegistrationIsKeptByItsBuild
|
||||
|
||||
@@ -74,6 +74,22 @@ func filesVerbs() []Verb {
|
||||
map[string]string{"path": "an absolute path", "owner": "an account's name or number",
|
||||
"group": "a group's name or number (optional)", "recursive": "\"true\": also everything under a directory"},
|
||||
[]string{"path", "owner"}, []string{"group", "recursive"}, "chown", "chgrp"),
|
||||
Verb{Name: "upload", Description: "Send one file of this node into a bucket of the object store, over a short signed link " +
|
||||
"and never over the bus; answers the key (the file's SHA-256 when none is given), the size and the SHA-256. " +
|
||||
"At most 100 MiB. A place a secret is kept, a known credential file, or a link to one is refused by name, " +
|
||||
"before anything is sent. Read " + filesAsAccount + ".",
|
||||
Input: schema(map[string]string{"path": "an absolute path to a file", "bucket": "the bucket the file goes into",
|
||||
"key": "the object's key (optional; the file's SHA-256 when not given)"}, []string{"path", "bucket"})},
|
||||
// download is a plain Verb, not through write(): write() types its optional arguments as true/false
|
||||
// switches, which a hash and a reason are not (issue 1719).
|
||||
Verb{Name: "download", Description: "Write one object of a bucket to a file of this node, over a short signed link and " +
|
||||
"never over the bus; the old file goes to the node's trash. At most 100 MiB. The bytes must hash to sha256, or to " +
|
||||
"the key when the key is a SHA-256; otherwise no file is written. Refuses a path a module owns " +
|
||||
"unless override names a reason; done as the node's operator account, through sudo -n where its rights do not reach.",
|
||||
Input: schema(map[string]string{"bucket": "the bucket", "key": "the object's key", "path": "an absolute path",
|
||||
"sha256": "the SHA-256 the bytes must have (optional; the key when that is a SHA-256)",
|
||||
"reason": "why the file is changed, for the audit trail",
|
||||
"override": overrideWord}, []string{"bucket", "key", "path", "reason"})},
|
||||
Verb{Name: "trash", Description: "What the node's trash holds: each deleted or replaced item with its id, " +
|
||||
"original path and when it went.",
|
||||
Input: schema(map[string]string{}, nil), Replaces: []string{"ls ~/.local/share/Trash"}},
|
||||
|
||||
@@ -2214,6 +2214,7 @@ func parseManifest(raw []byte, predefined PredefinedList) (Manifest, error) {
|
||||
problems = append(problems, replacesProblems(m)...)
|
||||
problems = append(problems, UnitSettingProblems(m)...)
|
||||
problems = append(problems, OwnerSettingProblems(m)...)
|
||||
problems = append(problems, GroupSettingProblems(m)...)
|
||||
problems = append(problems, endpointNameProblems(m)...)
|
||||
problems = append(problems, RouteProblems(m)...)
|
||||
for _, port := range m.Guards {
|
||||
|
||||
+18
-1
@@ -904,7 +904,7 @@ func isModule(catalogue map[string]Manifest, want string) bool {
|
||||
// machine can host the module as far as its capabilities go.
|
||||
func missingCapabilities(m Manifest, node Node) []string {
|
||||
var missing []string
|
||||
for _, c := range m.Capabilities {
|
||||
for _, c := range append(append([]string(nil), m.Capabilities...), impliedCapabilities(m)...) {
|
||||
if !node.Capabilities[c] {
|
||||
missing = append(missing, c)
|
||||
}
|
||||
@@ -1457,3 +1457,20 @@ func packageNames(m map[string][]string) []string {
|
||||
sort.Strings(names)
|
||||
return names
|
||||
}
|
||||
|
||||
// ContainerGroups is the capability a node-engine reports when it reads a container's `groups` and passes each as a
|
||||
// supplementary group (novox/hq ADR 0480 clause 7, issue 1638). An older engine refuses the key (strict decoding).
|
||||
const ContainerGroups = "container-groups"
|
||||
|
||||
// impliedCapabilities are the capabilities a module needs for what its resources say, beyond those it declares: a
|
||||
// container that names groups needs an engine that reads them, so the module is left out of a machine whose report
|
||||
// lacks the capability, naming it, and the field is sent to no engine that would refuse it. None for a module whose
|
||||
// containers name no groups.
|
||||
func impliedCapabilities(m Manifest) []string {
|
||||
for _, r := range m.Resources {
|
||||
if fmt.Sprint(r["type"]) == "container" && len(groupsOf(r)) > 0 {
|
||||
return []string{ContainerGroups}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -66,7 +66,8 @@ var capabilityName = regexp.MustCompile(`^[a-z0-9][a-z0-9-]*$`)
|
||||
// names an `if-capability` may give (novox/hq ADR 0255). A name nothing detects would leave its
|
||||
// contribution out on every machine, silently, so it is refused. A detector added there is added here.
|
||||
var KnownCapabilities = []string{
|
||||
"battery", "container-runtime", "firewall", "graphical-session", "overlay", "package-manager",
|
||||
"battery", "container-groups", "container-runtime", "file-references", "firewall", "graphical-session", "overlay",
|
||||
"package-manager",
|
||||
"power-meter", "privileged", "seat", "service-manager", "uplink-dhcpcd", "uplink-networkmanager",
|
||||
"uplink-systemd-networkd", "user-namespaces", "virtualisation",
|
||||
}
|
||||
|
||||
@@ -85,6 +85,9 @@ func settingInto(resource map[string]any, layers []Layer, module string) error {
|
||||
if fmt.Sprint(resource["type"]) == "directory" {
|
||||
return settingIntoOwner(resource, layers, module)
|
||||
}
|
||||
if fmt.Sprint(resource["type"]) == "container" {
|
||||
return settingIntoGroups(resource, layers, module)
|
||||
}
|
||||
if fmt.Sprint(resource["type"]) != "file" {
|
||||
return nil
|
||||
}
|
||||
@@ -343,6 +346,12 @@ func settingKeysAskedOutsideOwners(m Manifest) map[string]bool {
|
||||
if unit, ok := r["unit"].(string); ok {
|
||||
note(unit)
|
||||
}
|
||||
case "container":
|
||||
for _, g := range groupsOf(r) {
|
||||
if s, ok := g.(string); ok {
|
||||
note(s)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
// An item's text, filled from its template (WithTemplatesFrom), places a key as a file's does (issue 1658).
|
||||
@@ -369,3 +378,100 @@ func settingKeysAskedOutsideOwners(m Manifest) map[string]bool {
|
||||
}
|
||||
return used
|
||||
}
|
||||
|
||||
// A container's supplementary groups may be an operator's value (novox/hq ADR 0480 clause 7, ADR 0348 §11, issue 1636):
|
||||
// n8n joins the media library's group, whose number is the machine's, never the definition's (ADR 0112 §1). Each
|
||||
// element of `groups` is a whole placeholder or a whole number; a placeholder is filled with the number the
|
||||
// config value holds, and the list reaches the node-engine as numbers.
|
||||
|
||||
// groupSide is one element of a container's groups: a whole placeholder or a whole number.
|
||||
var groupSide = regexp.MustCompile(`^(?:\$\{setting:[a-z0-9][a-z0-9_.-]*\}|[0-9]+)$`)
|
||||
|
||||
// groupsOf is a container resource's groups as the manifest wrote them (text or numbers), none when it names none.
|
||||
func groupsOf(r map[string]any) []any {
|
||||
groups, _ := r["groups"].([]any)
|
||||
return groups
|
||||
}
|
||||
|
||||
// GroupSettingProblems refuses, where a manifest is read, a container whose groups hold anything but whole
|
||||
// placeholders and whole numbers, or a placeholder whose key the module neither declares nor asks for elsewhere.
|
||||
func GroupSettingProblems(m Manifest) []string {
|
||||
var problems []string
|
||||
for _, r := range m.Resources {
|
||||
if fmt.Sprint(r["type"]) != "container" {
|
||||
continue
|
||||
}
|
||||
for _, g := range groupsOf(r) {
|
||||
s, isText := g.(string)
|
||||
if !isText {
|
||||
continue // a number: the declaration judges its value
|
||||
}
|
||||
if !groupSide.MatchString(s) {
|
||||
problems = append(problems, fmt.Sprintf("%s: the container %v's group %q is not a whole "+
|
||||
"${config:key} or a whole number: a supplementary group is a number, and nothing else is "+
|
||||
"written into it (novox/hq ADR 0480 clause 7)", m.Module, r["id"], s))
|
||||
continue
|
||||
}
|
||||
if m.Templated() {
|
||||
continue
|
||||
}
|
||||
for _, key := range settingsUsed(s) {
|
||||
if _, declared := m.Settings[key]; !declared && !m.DeclaresConfig() {
|
||||
continue // the settings form judges its own keys
|
||||
}
|
||||
if _, declared := m.Settings[key]; !declared {
|
||||
problems = append(problems, fmt.Sprintf("%s: the container %v's group says ${config:%s}, and the module "+
|
||||
"declares no such value (novox/hq ADR 0348 §4)", m.Module, r["id"], key))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return problems
|
||||
}
|
||||
|
||||
// settingIntoGroups fills ${setting:…} in a container's groups, refusing — with the key and why, the resource left as
|
||||
// it was — a key nothing sets and a value that is not a whole number above zero. A number is kept; the list that
|
||||
// results is numbers alone.
|
||||
func settingIntoGroups(resource map[string]any, layers []Layer, module string) error {
|
||||
groups := groupsOf(resource)
|
||||
if len(groups) == 0 {
|
||||
return nil
|
||||
}
|
||||
filled := make([]any, 0, len(groups))
|
||||
for _, g := range groups {
|
||||
s, isText := g.(string)
|
||||
if !isText {
|
||||
if f, isNumber := g.(float64); isNumber && f == float64(int(f)) {
|
||||
g = int(f) // a number the manifest wrote, as the numbers a placeholder gives
|
||||
}
|
||||
filled = append(filled, g)
|
||||
continue
|
||||
}
|
||||
keys := settingsUsed(s)
|
||||
if len(keys) != 1 || settingRef.ReplaceAllString(s, "") != "" {
|
||||
filled = append(filled, g) // not a whole placeholder: the manifest check refuses it
|
||||
continue
|
||||
}
|
||||
key := keys[0]
|
||||
value, set := settingValue(layers, key)
|
||||
if !set {
|
||||
return &UnsetSettingError{Module: module, Setting: key,
|
||||
where: fmt.Sprintf("a container %v whose groups", resource["id"]), setToday: orNoSettings(layers),
|
||||
said: fmt.Sprintf(
|
||||
"%s has a container %v whose groups say ${setting:%s}, and nothing sets %q for it — an operator's "+
|
||||
"value is the assignment's, never the definition's (novox/hq ADR 0112): "+
|
||||
"`settings set %s <file>` with {%q: …}%s",
|
||||
module, resource["id"], key, key, module, key, orNoSettings(layers))}
|
||||
}
|
||||
v := plainly(value)
|
||||
id, whole := wholeID(v)
|
||||
if !whole || id == 0 {
|
||||
return fmt.Errorf("%s: the setting %q is %q, which cannot be a group of the container %v: a supplementary "+
|
||||
"group is a whole number above zero, never a name or other text (novox/hq ADR 0480 clause 7, ADR 0328 §15)",
|
||||
module, key, v, resource["id"])
|
||||
}
|
||||
filled = append(filled, int(id))
|
||||
}
|
||||
resource["groups"] = filled
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -56,7 +56,7 @@ type filler struct {
|
||||
// seatInto, machineInto and contributionsInto. A pass that comes to read another field adds it here,
|
||||
// or the sweep refuses the placeholder it would have filled.
|
||||
var fillers = []filler{
|
||||
{namespace: "setting", patterns: []*regexp.Regexp{settingRef}, fields: map[string][]string{"file": {"content"}, "service": {"unit"}, "directory": {"owner"}}},
|
||||
{namespace: "setting", patterns: []*regexp.Regexp{settingRef}, fields: map[string][]string{"file": {"content"}, "service": {"unit"}, "directory": {"owner"}, "container": {"groups"}}},
|
||||
{namespace: "dir", patterns: []*regexp.Regexp{dirRef}, fields: map[string][]string{"*": {"path", "content", "volumes", "env", "env-file"}}},
|
||||
{namespace: "access", patterns: []*regexp.Regexp{accessRef}, fields: map[string][]string{"*": {"path", "content", "volumes", "env", "env-file"}}},
|
||||
{namespace: "secret", patterns: []*regexp.Regexp{placeholder}, fields: map[string][]string{"file": {"content"}},
|
||||
|
||||
@@ -324,6 +324,15 @@ type File struct {
|
||||
// depend on a field somewhere else.
|
||||
Bytes string `json:"bytes,omitempty"`
|
||||
|
||||
// Digest and Size say the content is not carried but fetched: a file over FileReferenceThreshold travels as
|
||||
// the sha256 digest of its content, "sha256:<64 hex>", and its size in bytes, in place of Content or Bytes
|
||||
// (novox/hq ADR 0486 §1). The node-engine fetches the object of that name from FileBucket, checks what it
|
||||
// received against Digest and Size, and only then writes the file. Exclusive with Content, Sealed, Bytes,
|
||||
// VaultSealed and Secrets; sent only to an engine that reports FileReferencesCapability, as an older engine
|
||||
// refuses a field it does not know, whole.
|
||||
Digest string `json:"digest,omitempty"`
|
||||
Size int64 `json:"size,omitempty"`
|
||||
|
||||
// Owner is the user this belongs to, by name. Absent means root.
|
||||
Owner string `json:"owner,omitempty"`
|
||||
|
||||
@@ -465,6 +474,10 @@ func (f *File) validate(where string, _ bool) []string {
|
||||
problems = append(problems, where+": a file the vault sealed whole has no holes for secrets")
|
||||
}
|
||||
}
|
||||
problems = append(problems, f.checkReference(where)...)
|
||||
if f.Digest != "" || f.Size != 0 {
|
||||
said = append(said, "digest")
|
||||
}
|
||||
if len(said) > 1 {
|
||||
sort.Strings(said)
|
||||
problems = append(problems, where+
|
||||
@@ -1454,6 +1467,14 @@ type Container struct {
|
||||
// container runs as its image says. Its key is `identity`, the manifest's word; the field is not, because
|
||||
// a resource's Identity is its id.
|
||||
RunAs *ContainerIdentity `json:"identity,omitempty"`
|
||||
|
||||
// Groups are the supplementary groups the container's process joins, as numbers, each passed to the runtime
|
||||
// as a group to add (novox/hq ADR 0480 clause 7, issue 1636): n8n writes the media library through the
|
||||
// library's group, which a stock image knows nothing of. Each is above zero and below the kernel's "no group"
|
||||
// (ADR 0328 §15: never root). Part of the spec, so a change recreates the container. Absent: none are added.
|
||||
// An older node-engine refuses the key (strict decoding), so the controller sends it only to a node that
|
||||
// reports it reads it.
|
||||
Groups []int64 `json:"groups,omitempty"`
|
||||
}
|
||||
|
||||
// The ways an image is told who to run as (novox/hq ADR 0325 §2).
|
||||
@@ -1521,6 +1542,30 @@ func (i *ContainerIdentity) problems(where string, env map[string]string) []stri
|
||||
return problems
|
||||
}
|
||||
|
||||
// groupProblems refuses a supplementary group that is root's or below it, the kernel's "no group" number, or a
|
||||
// number named twice (novox/hq ADR 0480 clause 7, ADR 0328 §15).
|
||||
func (c *Container) groupProblems(where string) []string {
|
||||
var problems []string
|
||||
seen := map[int64]bool{}
|
||||
for _, g := range c.Groups {
|
||||
switch {
|
||||
case g == 0:
|
||||
problems = append(problems, fmt.Sprintf("%s: container %s adds group 0, and the mesh never makes or keeps anything "+
|
||||
"as root: a supplementary group is never root's (novox/hq ADR 0328 §15)", where, c.Name))
|
||||
case g < 0:
|
||||
problems = append(problems, fmt.Sprintf("%s: container %s adds group %d, below zero: a group is a whole number "+
|
||||
"above zero", where, c.Name, g))
|
||||
case g >= 1<<32-1:
|
||||
problems = append(problems, fmt.Sprintf("%s: container %s adds group %d, which is the kernel's \"no group\" "+
|
||||
"number or beyond it", where, c.Name, g))
|
||||
case seen[g]:
|
||||
problems = append(problems, fmt.Sprintf("%s: container %s adds group %d twice", where, c.Name, g))
|
||||
}
|
||||
seen[g] = true
|
||||
}
|
||||
return problems
|
||||
}
|
||||
|
||||
func (c *Container) Identity() string { return c.ID }
|
||||
func (c *Container) Kind() Type { return TypeContainer }
|
||||
func (c *Container) Target() string { return c.Name }
|
||||
@@ -1603,6 +1648,7 @@ func (c *Container) validate(where string, _ bool) []string {
|
||||
"runs and exits, and joins nothing afterwards")
|
||||
}
|
||||
problems = append(problems, c.RunAs.problems(where+": container "+c.Name, c.Env)...)
|
||||
problems = append(problems, c.groupProblems(where)...)
|
||||
return append(problems, checkImage(where, c.Image)...)
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,125 @@
|
||||
package declaration
|
||||
|
||||
import (
|
||||
"crypto/sha256"
|
||||
"encoding/base64"
|
||||
"encoding/hex"
|
||||
"fmt"
|
||||
"regexp"
|
||||
)
|
||||
|
||||
// The three names of a file travelling by digest (novox/hq ADR 0486 §1, §2, §5), shared by the node-engine that
|
||||
// reads them and the controller that writes them.
|
||||
const (
|
||||
// FileReferenceThreshold is the size above which a file travels by digest: a file of this many bytes or fewer
|
||||
// stays in the declaration as it is.
|
||||
FileReferenceThreshold = 64 * 1024
|
||||
// FileBucket is the object bucket of the bus that holds the contents, each under the hex of its sha256 digest
|
||||
// (ObjectName). The controller alone writes it; a node-engine reads an object by name.
|
||||
FileBucket = "declaration-files"
|
||||
// FileReferencesCapability is what a node-engine reports when it reads a file given as a digest and size.
|
||||
FileReferencesCapability = "file-references"
|
||||
)
|
||||
|
||||
var digestForm = regexp.MustCompile(`^sha256:[0-9a-f]{64}$`)
|
||||
|
||||
// DigestOf is the digest a reference names for a content: "sha256:<64 lower-case hex>".
|
||||
func DigestOf(content []byte) string {
|
||||
sum := sha256.Sum256(content)
|
||||
return "sha256:" + hex.EncodeToString(sum[:])
|
||||
}
|
||||
|
||||
// ObjectName is the name of the object holding the content a digest names: the hex without its "sha256:" prefix.
|
||||
func ObjectName(digest string) (string, error) {
|
||||
if !digestForm.MatchString(digest) {
|
||||
return "", fmt.Errorf("digest %q is not sha256:<64 lower-case hex characters>", digest)
|
||||
}
|
||||
return digest[len("sha256:"):], nil
|
||||
}
|
||||
|
||||
// IsReference reports whether the file gives its content as a digest and size instead of carrying it.
|
||||
func (f *File) IsReference() bool { return f.Digest != "" }
|
||||
|
||||
// checkReference judges the digest and size, and what a reference may not be given with.
|
||||
func (f *File) checkReference(where string) []string {
|
||||
if f.Digest == "" && f.Size == 0 {
|
||||
return nil
|
||||
}
|
||||
var problems []string
|
||||
if !digestForm.MatchString(f.Digest) {
|
||||
problems = append(problems, fmt.Sprintf("%s: digest %q; a file by reference names its content as "+
|
||||
"sha256:<64 lower-case hex characters>", where, f.Digest))
|
||||
}
|
||||
if f.Size <= 0 {
|
||||
problems = append(problems, fmt.Sprintf("%s: size %d; a file by reference states its size in bytes, "+
|
||||
"more than zero", where, f.Size))
|
||||
}
|
||||
if f.Content != "" || f.Bytes != "" || f.Sealed != "" || f.VaultSealed != nil || len(f.Secrets) > 0 {
|
||||
problems = append(problems, where+": a file by reference carries its content in the object its digest "+
|
||||
"names, so it says no content, bytes, sealed, vault-sealed or secrets")
|
||||
}
|
||||
if f.Into != "" {
|
||||
problems = append(problems, where+": a file by reference is written whole; a file written into says its "+
|
||||
"part in content")
|
||||
}
|
||||
return problems
|
||||
}
|
||||
|
||||
// Body is the bytes the file's content stands for: its content, or its bytes decoded.
|
||||
func (f *File) Body() ([]byte, error) {
|
||||
if f.Bytes == "" {
|
||||
return []byte(f.Content), nil
|
||||
}
|
||||
b, err := base64.StdEncoding.DecodeString(f.Bytes)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("file %s: bytes are not base64: %w", f.ID, err)
|
||||
}
|
||||
return b, nil
|
||||
}
|
||||
|
||||
// TravelsByDigest reports whether the controller may put this file's content in the bucket: it is plain content
|
||||
// or bytes (not sealed, not filled with secrets, not written into another file) of more than FileReferenceThreshold
|
||||
// bytes.
|
||||
func (f *File) TravelsByDigest() bool {
|
||||
if f.IsReference() || f.Sealed != "" || f.VaultSealed != nil || len(f.Secrets) > 0 || f.Into != "" {
|
||||
return false
|
||||
}
|
||||
if f.Bytes != "" {
|
||||
// DecodedLen counts padding, so the exact length is taken off the decoded bytes.
|
||||
body, err := f.Body()
|
||||
return err == nil && len(body) > FileReferenceThreshold
|
||||
}
|
||||
return len(f.Content) > FileReferenceThreshold
|
||||
}
|
||||
|
||||
// AsReference is the file as it is sent to an engine that reports FileReferencesCapability: a copy with the digest
|
||||
// and size in place of the content, and the content the controller puts into the bucket under ObjectName(digest).
|
||||
// A file that does not travel by digest is refused, not passed through, so a caller cannot send a small file empty.
|
||||
func (f *File) AsReference() (File, []byte, error) {
|
||||
if !f.TravelsByDigest() {
|
||||
return File{}, nil, fmt.Errorf("file %s does not travel by digest", f.ID)
|
||||
}
|
||||
body, err := f.Body()
|
||||
if err != nil {
|
||||
return File{}, nil, err
|
||||
}
|
||||
ref := *f
|
||||
ref.Content, ref.Bytes = "", ""
|
||||
ref.Digest, ref.Size = DigestOf(body), int64(len(body))
|
||||
return ref, body, nil
|
||||
}
|
||||
|
||||
// Verify checks what was fetched against the digest and size the file names; the node-engine writes the file only
|
||||
// after it returns nil. A mismatch is the apply's failure, never a partial apply (ADR 0486 §1).
|
||||
func (f *File) Verify(body []byte) error {
|
||||
if !f.IsReference() {
|
||||
return fmt.Errorf("file %s is not given by reference", f.ID)
|
||||
}
|
||||
if int64(len(body)) != f.Size {
|
||||
return fmt.Errorf("file %s: the object is %d bytes and the declaration says %d", f.ID, len(body), f.Size)
|
||||
}
|
||||
if got := DigestOf(body); got != f.Digest {
|
||||
return fmt.Errorf("file %s: the object hashes to %s and the declaration names %s", f.ID, got, f.Digest)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -25,7 +25,7 @@ type BuildRecord struct {
|
||||
}
|
||||
|
||||
// Artifact is one thing a build made: mesh-controller internal/chain/records.go Artifact, json keys name, kind and
|
||||
// reference, nothing else (novox/hq ADR 0478 clause 7).
|
||||
// reference, nothing else (novox/hq ADR 0478 clause 8).
|
||||
type Artifact struct {
|
||||
Name string `json:"name"`
|
||||
Kind string `json:"kind"`
|
||||
|
||||
@@ -390,7 +390,7 @@ type Module struct {
|
||||
Repository string `json:"repository"`
|
||||
Commit string `json:"commit"`
|
||||
// Path is the module's directory in its repository, trimmed of slashes, and RootPath (".") when its manifest is the
|
||||
// repository's root (novox/hq ADR 0478 clause 8, issue 1615): the build tools need it. Empty means not known: a value
|
||||
// repository's root (novox/hq ADR 0478 clause 9, issue 1615): the build tools need it. Empty means not known: a value
|
||||
// written before the indexer learned paths has none.
|
||||
Path string `json:"path,omitempty"`
|
||||
// Toolchains are the toolchains FP2 was computed with, by language (go, typescript): the version installed and the
|
||||
|
||||
Vendored
+1
-1
@@ -1,4 +1,4 @@
|
||||
# git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021105408-c8e35097e861
|
||||
# git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021150202-6f6327587d19
|
||||
## explicit; go 1.26.0
|
||||
git.novox.be/novox/mesh-sdk/go/asks
|
||||
git.novox.be/novox/mesh-sdk/go/catalogue
|
||||
|
||||
Reference in New Issue
Block a user